Sim-to-Real与回测框架
概述
强化学习策略从模拟环境到真实市场的迁移 (Sim-to-Real Transfer) 是金融 RL 落地的核心瓶颈。模拟环境的保真度 (Fidelity)、回测框架的严谨性以及现实差距 (Reality Gap) 的缓解策略共同决定了 RL 策略的实盘效果。本文系统讨论模拟环境的构建、Sim-to-Real Gap 的来源与缓解方法，以及回测的最佳实践。
模拟环境 (Simulation Environments)
基于历史数据的环境
最简单的模拟方式是在历史行情数据上回放 (Replay)：
import gymnasium as gym
import numpy as np
class HistoricalTradingEnv(gym.Env):
def __init__(self, price_data, feature_data, initial_cash=1e6,
transaction_cost=0.001):
super().__init__()
self.prices = price_data # (T, N_assets)
self.features = feature_data # (T, N_assets, N_features)
self.initial_cash = initial_cash
self.tc = transaction_cost
self.n_assets = price_data.shape[1]
# 动作空间: 各资产的目标权重
self.action_space = gym.spaces.Box(
low=0, high=1, shape=(self.n_assets,), dtype=np.float32
)
# 状态空间
state_dim = self.n_assets * (feature_data.shape[2] + 1) + 2
self.observation_space = gym.spaces.Box(
low=-np.inf, high=np.inf, shape=(state_dim,), dtype=np.float32
)
def reset(self, seed=None):
self.t = 0
self.weights = np.zeros(self.n_assets)
self.portfolio_value = self.initial_cash
return self._get_obs(), {}
def step(self, action):
target_weights = action / (action.sum() + 1e-8)
# 计算交易成本
weight_diff = np.abs(target_weights - self.weights)
tc = self.tc * weight_diff.sum() * self.portfolio_value
# 更新组合价值
returns = self.prices[self.t+1] / self.prices[self.t] - 1
port_return = np.dot(target_weights, returns)
self.portfolio_value = self.portfolio_value * (1 + port_return) - tc
# 更新权重（考虑价格变动后的权重漂移）
new_values = target_weights * (1 + returns)
self.weights = new_values / new_values.sum()
self.t += 1
reward = np.log(1 + port_return) - self.tc * weight_diff.sum()
done = self.t >= len(self.prices) - 2
truncated = False
return self._get_obs(), reward, done, truncated, {
'portfolio_value': self.portfolio_value,
'turnover': weight_diff.sum()
}
def _get_obs(self):
features = self.features[self.t].flatten()
return np.concatenate([
features,
self.weights,
[self.portfolio_value / self.initial_cash],
[self.t / len(self.prices)]
])
基于模型的环境
使用统计模型或生成模型构建更逼真的市场模拟器：
| 模型 | 特点 | 适用场景 |
|---|---|---|
| GBM | 几何布朗运动，简单 | 基础测试 |
| GARCH | 波动率聚集 | 波动率建模 |
| VAR/VECM | 多资产联动 | 组合策略 |
| GAN | 数据驱动，高保真 | 复杂分布模拟 |
| Agent-Based Model | 多参与者交互 | 市场微观结构 |
class GARCHMarketSimulator:
def __init__(self, params, n_assets, correlation_matrix):
self.params = params # GARCH(1,1) 参数
self.n_assets = n_assets
self.corr_matrix = correlation_matrix
self.cholesky = np.linalg.cholesky(correlation_matrix)
def generate_path(self, n_steps, initial_prices):
prices = np.zeros((n_steps, self.n_assets))
prices[0] = initial_prices
variances = np.ones(self.n_assets) * self.params['omega'] / (
1 - self.params['alpha'] - self.params['beta']
)
for t in range(1, n_steps):
# 相关标准正态随机数
z = self.cholesky @ np.random.randn(self.n_assets)
returns = np.sqrt(variances) * z
prices[t] = prices[t-1] * np.exp(returns)
# GARCH 方差更新
variances = (self.params['omega'] +
self.params['alpha'] * returns**2 +
self.params['beta'] * variances)
return prices
环境设计的关键原则
- 交易成本必须真实反映（佣金、滑点、冲击）
- 订单成交逻辑需模拟限价单的部分成交和排队
- 不可交易状态（停牌、涨跌停）需正确处理
- 数据频率应匹配实际交易频率
Sim-to-Real Gap
Gap 的来源
\[\text{Real Performance} = \text{Sim Performance} - \text{Sim-to-Real Gap}\]
| Gap 来源 | 描述 | 量级 |
|---|---|---|
| 市场冲击缺失 | 模拟中忽略大单对价格的影响 | 高 |
| 滑点低估 | 实际成交价劣于理论价格 | 中-高 |
| 延迟与异步 | 网络延迟、数据不同步 | 中（高频场景高） |
| 流动性假设 | 假设任意时刻可交易任意量 | 中 |
| 分布偏移 | 训练期与实盘期市场状态不同 | 高 |
| 反身性忽略 | 策略自身对市场的影响 | 低-中 |
滑点建模
class RealisticSlippageModel:
def __init__(self, base_spread_bps=5, impact_coeff=0.1):
self.base_spread = base_spread_bps / 10000
self.impact_coeff = impact_coeff
def compute_slippage(self, order_size, adv, volatility, side):
"""
order_size: 订单金额
adv: 平均日成交额 (Average Daily Volume)
volatility: 当前波动率
side: 'buy' or 'sell'
"""
participation_rate = order_size / adv
# 市场冲击: 平方根模型
market_impact = self.impact_coeff * volatility * np.sqrt(
participation_rate
)
# 价差成本
spread_cost = self.base_spread / 2
# 总滑点
total_slippage = spread_cost + market_impact
return total_slippage # 占交易金额的比例
过于乐观的回测
忽略市场冲击和滑点是回测中最常见的"致命乐观"。一个在无摩擦回测中年化收益 30% 的策略，加入真实的交易成本后可能变为亏损。规则：任何回测结果都应先打折 30-50% 再进行评估。
回测最佳实践
1. 时间序列切分
def proper_train_test_split(data, train_ratio=0.6, val_ratio=0.2,
gap_days=20):
"""带间隔的时序切分"""
n = len(data)
train_end = int(n * train_ratio)
val_start = train_end + gap_days # 清洗间隔
val_end = val_start + int(n * val_ratio)
test_start = val_end + gap_days # 清洗间隔
return {
'train': data[:train_end],
'val': data[val_start:val_end],
'test': data[test_start:] # 仅最终评估使用一次
}
2. 多维度评估
def comprehensive_backtest_metrics(returns, benchmark_returns, rf_rate=0.02):
"""全面的回测评估指标"""
excess_returns = returns - rf_rate / 252
metrics = {
# 收益指标
'annual_return': returns.mean() * 252,
'cumulative_return': (1 + returns).prod() - 1,
# 风险指标
'annual_volatility': returns.std() * np.sqrt(252),
'max_drawdown': compute_max_drawdown(returns),
'var_95': np.percentile(returns, 5),
'cvar_95': returns[returns <= np.percentile(returns, 5)].mean(),
# 风险调整收益
'sharpe_ratio': excess_returns.mean() / returns.std() * np.sqrt(252),
'sortino_ratio': (excess_returns.mean() /
returns[returns < 0].std() * np.sqrt(252)),
'calmar_ratio': returns.mean() * 252 / abs(compute_max_drawdown(returns)),
# 策略特征
'win_rate': (returns > 0).mean(),
'profit_loss_ratio': abs(returns[returns > 0].mean() /
returns[returns < 0].mean()),
# 相对基准
'alpha': compute_alpha(returns, benchmark_returns),
'beta': compute_beta(returns, benchmark_returns),
'information_ratio': compute_ir(returns, benchmark_returns),
}
return metrics
3. 稳健性检验
def robustness_checks(strategy, data, n_bootstrap=1000):
"""策略稳健性检验"""
results = {}
# 1. Bootstrap 置信区间
sharpe_samples = []
for _ in range(n_bootstrap):
boot_returns = np.random.choice(
strategy.returns, size=len(strategy.returns), replace=True
)
sr = boot_returns.mean() / boot_returns.std() * np.sqrt(252)
sharpe_samples.append(sr)
results['sharpe_ci_95'] = np.percentile(sharpe_samples, [2.5, 97.5])
# 2. 子区间稳定性
sub_periods = np.array_split(strategy.returns, 4)
sub_sharpes = [s.mean() / s.std() * np.sqrt(252) for s in sub_periods]
results['sharpe_stability'] = np.std(sub_sharpes) / np.mean(sub_sharpes)
# 3. 参数敏感性
for param_name, param_range in strategy.param_ranges.items():
param_sharpes = []
for val in param_range:
s = strategy.run_with_param(param_name, val, data)
param_sharpes.append(s.sharpe)
results[f'{param_name}_sensitivity'] = np.std(param_sharpes)
return results
Deflated Sharpe Ratio
Bailey & Lopez de Prado 的 Deflated Sharpe Ratio (DSR) 校正了多重测试偏差：
\[\text{DSR} = \Phi\left(\frac{(\hat{\text{SR}} - \text{SR}_0)\sqrt{n-1}}{\sqrt{1 - \hat{\gamma}_3 \hat{\text{SR}} + \frac{\hat{\gamma}_4 - 1}{4}\hat{\text{SR}}^2}}\right)\]
其中 \(\hat{\gamma}_3, \hat{\gamma}_4\) 分别为偏度和峰度，\(\text{SR}_0 = \sqrt{\frac{2\log N}{252}}\) 为 \(N\) 次独立试验的期望最大夏普比率。
Reality Gap 缓解策略
Domain Randomization
在训练时随机化模拟环境的参数，提高策略的泛化性：
def randomized_environment(base_params):
"""域随机化: 随机化市场参数"""
randomized = base_params.copy()
randomized['volatility'] *= np.random.uniform(0.7, 1.3)
randomized['spread'] *= np.random.uniform(0.5, 2.0)
randomized['impact_coeff'] *= np.random.uniform(0.8, 1.5)
randomized['correlation_noise'] = np.random.uniform(0, 0.1)
return randomized
Conservative Estimation
对策略的预期性能做保守估计：
\[\hat{R}_{\text{conservative}} = \hat{R}_{\text{sim}} - k \cdot \hat{\sigma}_{\text{sim}}\]
取 \(k = 1 \sim 2\)，即从模拟性能中减去 1-2 倍标准差。
渐进式上线
[模拟回测] → [纸上交易] → [小资金实盘] → [逐步放量] → [全量运行]
Paper Trading 1% AUM 10% AUM 100% AUM
每个阶段需要足够的观察期和性能验证。
在线适应 (Online Adaptation)
在实盘运行中持续学习和适应：
\[\theta_{t+1} = \theta_t + \alpha_{\text{online}} \nabla_\theta J(\theta_t | \text{recent data})\]
在线学习的稳定性
在线适应可能导致灾难性遗忘 (Catastrophic Forgetting) 或过度适应近期市场。建议使用弹性权重巩固 (EWC) 或保持与离线训练模型的 KL 散度约束：
\[\text{KL}(\pi_{\text{online}} \| \pi_{\text{offline}}) \leq \epsilon\]
小结
Sim-to-Real 迁移是金融 RL 从学术研究走向实际应用的最大挑战。高保真的模拟环境、真实的交易成本建模、严谨的回测方法论以及渐进式的上线策略构成了完整的实践框架。核心原则是对回测结果保持怀疑——模拟中看起来完美的策略在实盘中往往大幅折损。Domain Randomization、保守估计和在线适应是缓解 Reality Gap 的关键技术手段。