企业官网建设流程全解析

• 金融科技
• 科学计算

导读

本文围绕 tf-quant-finance 中随机过程路径采样的核心入口tf_quant_finance.models.euler_sampling.sample（源码见 euler_sampling.py）展开，系统讲解如何使用 Euler-Maruyama 数值格式对一般 Ito 过程进行 Monte Carlo 路径模拟。你将掌握该函数的完整参数语义（时间网格、随机数类型、批处理、自定义噪声、梯度观测等）、一维/多维与批处理场景下的真实调用方式，以及底层tf.while_loop/custom for_loop双引擎实现原理与测试验证方法，可直接用于金融模型（如利率、波动率、权益资产模型）的路径模拟与定价实践。

一、方法背景：从 Ito 过程到 Euler-Maruyama 格式

sample面向如下形式的 Ito 过程（维数为dim）：

dX = a(t, X_t) dt + b(t, X_t) dW_t X(t=0) = x0

其中a（漂移）与b（波动率）均为用户提供的函数。Euler 方法（即 Euler-Maruyama 方法）将时间离散化后生成路径序列{X_n}：

X_{n+1} = X_n + a(t_n, X_n) dt + b(t_n, X_n) (N(0, t_{n+1}) - N(0, t_n)) X_0 = x0

其中dt = t_{n+1} - t_n，N为标准正态分布采样。X_{n+1} - X_n的增量由两部分构成：确定性漂移项a(t_n, X_n) dt与扩散项b(t_n, X_n) * dW，而dW正是正态增量N(0, t_{n+1}) - N(0, t_n)，其标准差为sqrt(dt)。

从源码实现看（euler_sampling.py），算法先计算相邻时刻的步长dt = times[1:] - times[:-1]及其平方根sqrt_dt，然后在每步中按dw = normal_draws[i] * sqrt_dt[i]缩放噪声、按dt_inc = dt[i] * drift_fn(...)累加漂移，最终得到next_state = current_state + dt_inc + dw_inc（_euler_step）。

适用性与局限：Euler-Maruyama 是强收敛阶为 0.5 的显式格式，实现简单、对任意漂移/波动率函数通用；若需要更高的强收敛精度，可考虑仓库中基于 Milstein 格式的 milstein_sampling.sample。

二、函数签名与完整参数说明

sample的完整签名（与文档及源码一致）：

tf_quant_finance.models.euler_sampling.sample( dim, drift_fn, volatility_fn, times, time_step=None, num_time_steps=None, num_samples=1, initial_state=None, random_type=None, seed=None, swap_memory=True, skip=0, precompute_normal_draws=True, times_grid=None, normal_draws=None, watch_params=None, validate_args=False, tolerance=None, dtype=None, name=None )

2.1 核心必选参数

注意drift_fn/volatility_fn返回形状的差异：漂移是向量（[..., num_samples, dim]），波动率是矩阵（[..., num_samples, dim, dim]）。在 euler_sampling.py 中，扩散项通过tf.linalg.matvec(volatility_fn(...), dw)实现，因此多维过程可以包含交叉相关性（非对角波动率项）。

2.2 时间离散化参数（三选一）

三者必须且只能提供其一（times_grid除外），否则抛出ValueError——这一约束在源码中有明确校验（euler_sampling.py）：既未提供num_time_steps也未提供time_step时报错，两者同时提供时同样报错。

2.3 采样规模与初始状态

2.4 随机数类型与随机种子

2.5 性能与噪声控制

2.6 自定义噪声与梯度观测

2.7 参数校验

2.8 类型与命名

三、随机数类型 RandomType：从伪随机到拟随机

random_type对应枚举tff.math.random.RandomType（定义于 multivariate_normal.py），共七种取值：

低差异序列（HALTON/SOBOL）的使用约束：这类序列必须在算法开始前一次性生成完整的高维正态随机数，因此源码中当random_type为SOBOL、HALTON、HALTON_RANDOMIZED、STATELESS、STATELESS_ANTITHETIC，或precompute_normal_draws=True时，会通过 utils.generate_mc_normal_draws 预先抽取全部噪声；而PSEUDO/PSEUDO_ANTITHETIC可在每步_euler_step内即时采样（euler_sampling.py）。generate_mc_normal_draws会把num_time_steps * dim维的联合正态采样 reshape 为[steps_num] + batch_shape + [num_samples, dim]供逐步消费。

四、端到端示例：二维 Ito 过程路径采样

以下示例完整继承自官方文档，采样如下二维过程：

dX_1 = mu_1 * sqrt(t) dt + s11 * dW_1 + s12 * dW_2 dX_2 = mu_2 * sqrt(t) dt + s21 * dW_1 + s22 * dW_2

import tensorflow as tf import tf_quant_finance as tff import numpy as np mu = np.array([0.2, 0.7]) s = np.array([[0.3, 0.1], [0.1, 0.3]]) num_samples = 10000 dim = 2 dtype = tf.float64 # 定义漂移与波动率函数 def drift_fn(t, x): return mu * tf.sqrt(t) * tf.ones([num_samples, dim], dtype=dtype) def vol_fn(t, x): return s * tf.ones([num_samples, dim, dim], dtype=dtype) # 设置初始位置 x0 = np.array([0.1, -1.1]) # 在指定时刻上使用 Euler 格式采样 num_samples 条路径 times = [0.1, 1.0, 2.0] paths = tff.models.euler_sampling.sample( dim=dim, drift_fn=drift_fn, volatility_fn=vol_fn, times=times, num_samples=num_samples, initial_state=x0, time_step=0.01, seed=42, dtype=dtype) # 预期结果：paths.shape = [10000, 3, 2]

这里需要注意两点实操细节：

1. drift_fn/vol_fn返回形状必须与调用方约定一致。若想写出与num_samples解耦的通用函数，可以像 euler_sampling_test.py 那样基于tf.ones_like(x)构造返回张量，例如drift = mu * tf.sqrt(t) * tf.ones_like(x, dtype=t.dtype)。
2. times可以直接传入 Python 列表（内部会经tf.convert_to_tensor转换，且dtype未指定时以其推导）。

五、时间网格的三种构造方式与取舍

sample内部通过 utils.prepare_grid 构造完整时间网格：

• time_step模式：在[0, times[-1])上以time_step生成均匀网格，与times合并、排序并做容差去重，最后用searchsorted定位各times在网格中的索引（_grid_from_time_step）。
• num_time_steps模式：构造网格点数不少于max(len(times), num_time_steps)，确保times全部落在网格上（_grid_from_num_times）。
• times_grid模式：直接使用给定网格，并通过比较前后邻近点的距离把times映射到最近网格点（utils.py）。

无论哪种模式，最终都会生成布尔掩码keep_mask，用于标记哪些迭代步的输出需要写入结果TensorArray——这正是"只在times对应时刻记录路径、中间网格点只做递推"的实现机制（euler_sampling.py）。

实践建议：time_step最直观（固定最大步长），num_time_steps适合控制总计算量，times_grid适合复用既有的交易日历/利率曲线时间网格。网格越细，Euler 格式离散误差越小，但计算量与内存（尤其precompute_normal_draws=True时）越大。

六、返回值、批处理与形状语义

sample返回一个实值Tensor，形状为batch_shape_process + [num_samples, k, n]，其中k为times的大小（输出时刻数），n为过程维数dim。具体包括：

• 无批处理：[num_samples, k, dim]，如上述示例的[10000, 3, 2]；
• 有批处理：initial_state形状为batch_shape + [1, 1]时，返回batch_shape + [num_samples, k, dim]。测试用例中批量初值x0 = [[[0.1]], [[0.1]]]对应的返回形状为[2, num_samples, 6, 1]（见 euler_sampling_test.py）；
• 单时刻：当times只有 1 个点时，结果仍会扩展出长度为 1 的维度轴（euler_sampling.py）。

若times[0] = 0，则返回的首个时刻点即初始状态x0（测试用例验证了"包含初始状态"与"不包含初始状态"两次采样的路径一致性，见 euler_sampling_test.py）。

七、高级特性：自定义噪声、对偶变量与可控求导

7.1 外部提供噪声normal_draws

当需要复现特定随机数、或实现方差缩减（如对偶变量）时，可预先构造normal_draws。测试用例中的典型做法（euler_sampling_test.py）：

num_samples = 1 normal_draws = tf.random.stateless_normal( shape=[5000, 30, 1], seed=[1, 42], dtype=dtype) # 拼上相反数即得到对偶变量样本 normal_draws = tf.concat([normal_draws, -normal_draws], axis=0) paths = euler_sampling.sample( dim=1, drift_fn=drift_fn, volatility_fn=vol_fn, times=[0.1, 0.2, 0.3], num_samples=num_samples, normal_draws=normal_draws, random_type=random.RandomType.STATELESS_ANTITHETIC, seed=[1, 42]) # 实际有效样本数由 normal_draws 第一维决定（此处 10000）

提供normal_draws时，num_samples参数被忽略；若normal_draws.shape[-1] != dim会抛出ValueError，若其时间步数与num_time_steps不匹配则抛出tf.errors.InvalidArgumentError（源码校验见 euler_sampling.py）。

7.2 对偶变量提升均值估计精度

采用对偶变量后，同一均值估计的误差显著下降。测试 test_antithetic_sample_paths_mean_2d 显示：对偶变量场景用rtol/atol = 5e-3的容差即可通过均值一致性检验，而普通PSEUDO场景容差为1e-2。

7.3 基于watch_params的可微采样

当需要在采样路径上对模型参数（如漂移系数、波动率矩阵元素）求梯度（典型应用：蒙特卡洛定价中的梯度回传、校准问题）时，把待求导参数放入watch_params列表。源码据此选择两条路径（euler_sampling.py）：

• watch_params is None：使用tf.while_loop（_while_loop）传播路径，适合普通采样；
• 提供watch_params：改用 custom_loops.for_loop 驱动（_for_loop），该实现能沿watch_params高效求导，同时把对其它张量的梯度置零。

测试 test_halton_sample_paths_2d 即把mu, a, b三个系数传入watch_params，在tf.function内完成带参采样。

八、错误处理与参数校验策略

sample可能抛出的异常（与文档一致，且均在源码中实现）：

• ValueError：
  • (a) 未提供times_grid时，num_time_steps与time_step同时缺失或同时给出；
  • (b) 提供了normal_draws但dim与其最后一维不匹配；
• tf.errors.InvalidArgumentError：提供了normal_draws但其时间步数与num_time_steps不匹配；或validate_args=True时times/times_grid不严格递增。

测试 test_wrong_times 与 test_sample_shape_mismatch 分别验证了上述两类运行时错误。建议在开发阶段开启validate_args=True以尽早发现times排序或维度问题，在生产性能敏感场景可关闭以省去校验开销。

九、实现原理与测试验证

9.1 双引擎实现：while_loop 与 custom for_loop

_sample完成网格与噪声准备后，按watch_params分流（euler_sampling.py）：

• _while_loop：使用tf.TensorArray记录各时刻输出，tf.while_loop的终止条件为i < steps_num 且 written_count < num_requested_times，最后stack并transpose到batch_shape + [num_samples, k, dim]（euler_sampling.py）；
• _for_loop：基于custom_loops.for_loop，仅返回最终状态并按time_indices聚合输出，避免 TensorArray 开销，利于反向传播（euler_sampling.py）。

两者的单步逻辑共用_euler_step：先取该步噪声（预计算或即时random.mv_normal_sample），再叠加漂移增量与扩散增量。

9.2 测试如何验证正确性

euler_sampling_test.py 提供了可复现的验证范式，共覆盖 Wiener 过程、一维/二维 Ito 过程、批处理、对偶变量、低差异序列、dtype、异常路径等场景：

• 统计矩校验：对dX = dW的 Wiener 过程，断言路径均值为零、协方差矩阵等于min(t_i, t_j)（test_sample_paths_wiener）；
• 解析期望校验：对dX = mu * sqrt(t) dt + (a * t + b) dW，期望E[X(t)] = x0 + 2/3 * mu * t^1.5，测试以rtol=1e-2断言（test_sample_paths_1d）；
• 形状与 dtype 校验：断言paths.shape与paths.dtype（test_sample_paths_dtypes）。

十、与其它采样方法的定位

sample位于 models/euler_sampling.py，是tff.models子包中通用 Ito 过程路径生成的基础设施。仓库还提供：

• models/milstein_sampling/sample.md：Milstein 格式，强收敛阶更高；
• models/GenericItoProcess.md 等模型类：在sample之上封装了 GBM、Heston、CIR、Hull-White 等具体金融过程。

理解euler_sampling.sample的参数与实现，是使用上述高级模型、乃至自行扩展自定义随机过程（如局部波动率模型）的前提。

十一、快速上手要点总结

1. 必选四件套：dim、drift_fn、volatility_fn、times；时间网格在time_step/num_time_steps/times_grid中三选一。
2. 形状记忆口诀：漂移输出[..., num_samples, dim]，波动率输出[..., num_samples, dim, dim]，返回值batch_shape + [num_samples, k, dim]。
3. 随机类型选型：追求可复现用STATELESS（seed=[s1, s2]）；方差缩减用PSEUDO_ANTITHETIC/STATELESS_ANTITHETIC；低维高精度可用SOBOL/HALTON并配skip跳过序列头部。
4. 需要求导时：把待求导参数放入watch_params，函数将自动切换高效求导路径。
5. 调试期开启validate_args=True，可捕获times非递增、normal_draws维度不匹配等问题。

• 金融科技
• 科学计算