问题描述
解决方案
在PTrade中进行蒙特卡洛模拟(Monte Carlo Simulation)通常用于检验策略在不同市场路径下的表现,特别是评估策略是否存在“过度拟合”以及在极端情况下的最大回撤(Max Drawdown)风险。
由于PTrade回测引擎本身是按时间序列运行的,我们无法直接“重置”市场行情。因此,最实用的方法是基于策略产生的“日收益率”进行重抽样(Resampling/Bootstrapping)。
以下是实现思路与完整的代码示例。
核心思路
- 记录收益率:在回测过程中,记录每一个交易日的收益率。
- 重抽样(Shuffle):在回测结束时(或指定时间),将历史日收益率的顺序随机打乱。
- 生成路径:基于打乱后的收益率序列,重新计算资金曲线。
- 重复模拟:重复上述步骤 $N$ 次(例如1000次)。
- 统计分析:统计这1000次模拟中的最大回撤、年化收益等指标的分布情况(如95%置信区间下的最差表现)。
PTrade 策略代码实现
以下代码包含了一个简单的双均线策略作为“底座”,并在其中嵌入了蒙特卡洛模拟模块。请注意,由于PTrade没有专门的“回测结束事件”,我们在代码中设置了一个g.end_date,当回测运行到该日期时触发模拟计算。
import numpy as np
import pandas as pd
def initialize(context):
"""
初始化函数
"""
# 1. 设定策略参数
g.security = '600570.SS' # 示例股票:恒生电子
set_universe(g.security)
# 2. 设定回测结束日期(用于触发蒙特卡洛计算)
# 请确保此日期与您回测面板设置的结束日期一致
g.end_date = '20231229'
# 3. 初始化蒙特卡洛模拟所需的全局变量
g.daily_returns = [] # 用于存储每日收益率
g.last_total_value = context.portfolio.portfolio_value # 记录上日总资产
# 4. 设定模拟次数
g.simulations = 1000
def before_trading_start(context, data):
"""
盘前处理
"""
pass
def handle_data(context, data):
"""
策略核心逻辑:简单的双均线策略用于生成交易记录
"""
security = g.security
# 获取历史数据
# 注意:为了演示,这里count设为20,实际使用建议更大
hist = get_history(20, '1d', 'close', security, fq='pre')
if len(hist) < 20:
return
close_prices = hist['close'].values
# 计算均线
ma5 = close_prices[-5:].mean()
ma10 = close_prices[-10:].mean()
curr_price = data[security]['close']
position = get_position(security).amount
cash = context.portfolio.cash
# 交易逻辑
if ma5 > ma10 and position == 0:
# 金叉买入
order_value(security, cash)
log.info("买入 %s" % security)
elif ma5 < ma10 and position > 0:
# 死叉卖出
order_target(security, 0)
log.info("卖出 %s" % security)
def after_trading_end(context, data):
"""
盘后处理:记录收益率并在最后一天执行蒙特卡洛模拟
"""
# 1. 计算当日收益率
current_total_value = context.portfolio.portfolio_value
if g.last_total_value > 0:
daily_ret = (current_total_value - g.last_total_value) / g.last_total_value
g.daily_returns.append(daily_ret)
# 更新昨日资产
g.last_total_value = current_total_value
# 2. 检查是否是回测的最后一天
# 获取当前日期字符串 YYYYMMDD
current_date_str = context.blotter.current_dt.strftime("%Y%m%d")
if current_date_str == g.end_date:
log.info("==========================================")
log.info("回测结束,开始执行蒙特卡洛稳健性检验...")
run_monte_carlo_simulation(context)
log.info("==========================================")
def run_monte_carlo_simulation(context):
"""
执行蒙特卡洛模拟的核心函数
"""
returns = np.array(g.daily_returns)
if len(returns) < 10:
log.warning("交易天数过少,无法进行有效的蒙特卡洛模拟。")
return
n_days = len(returns)
simulations = g.simulations
# 存储每次模拟的最大回撤
sim_mdds = []
# 存储每次模拟的累计收益
sim_total_returns = []
log.info("正在进行 %d 次路径模拟..." % simulations)
for i in range(simulations):
# 核心步骤:随机打乱收益率顺序 (Bootstrap)
# replace=True 表示有放回抽样,replace=False 表示无放回重排
# 通常检验稳健性使用无放回重排(False)来测试序列风险
# 使用有放回(True)可以模拟更多未知的极端情况
shuffled_returns = np.random.choice(returns, size=n_days, replace=True)
# 计算资金曲线 (假设初始资金为1)
equity_curve = np.cumprod(1 + shuffled_returns)
# 计算该条路径的累计收益
total_ret = equity_curve[-1] - 1
sim_total_returns.append(total_ret)
# 计算该条路径的最大回撤
# 累积最大值
cum_max = np.maximum.accumulate(equity_curve)
# 回撤序列
drawdowns = (cum_max - equity_curve) / cum_max
# 最大回撤
max_dd = np.max(drawdowns)
sim_mdds.append(max_dd)
# --- 统计分析 ---
sim_mdds = np.array(sim_mdds)
sim_total_returns = np.array(sim_total_returns)
# 1. 原始策略表现
original_equity = np.cumprod(1 + returns)
orig_cum_max = np.maximum.accumulate(original_equity)
orig_dd = np.max((orig_cum_max - original_equity) / orig_cum_max)
orig_ret = original_equity[-1] - 1
# 2. 蒙特卡洛统计指标
avg_mdd = np.mean(sim_mdds)
worst_mdd_95 = np.percentile(sim_mdds, 95) # 95%置信度下的最差回撤
best_mdd_5 = np.percentile(sim_mdds, 5) # 5%置信度下的最好回撤
avg_ret = np.mean(sim_total_returns)
worst_ret_5 = np.percentile(sim_total_returns, 5) # 5%置信度下的最差收益
# --- 输出报告 ---
log.info("【蒙特卡洛模拟报告】")
log.info("模拟次数: %d, 样本天数: %d" % (simulations, n_days))
log.info("-" * 30)
log.info("原始策略最大回撤: %.2f%%" % (orig_dd * 100))
log.info("模拟平均最大回撤: %.2f%%" % (avg_mdd * 100))
log.info("95%%置信度最差回撤: %.2f%% (若超过此值,说明策略风险极高)" % (worst_mdd_95 * 100))
log.info("-" * 30)
log.info("原始策略累计收益: %.2f%%" % (orig_ret * 100))
log.info("模拟平均累计收益: %.2f%%" % (avg_ret * 100))
log.info("5%%置信度最差收益: %.2f%%" % (worst_ret_5 * 100))
# 简单的稳健性评价
if orig_dd < avg_mdd:
log.info("评价: 原始策略的回撤优于模拟平均值,可能存在一定的运气成分(序列正向)。")
else:
log.info("评价: 原始策略的回撤劣于模拟平均值,说明当前市场路径对策略较为不利。")
结果解读指南
运行上述代码后,请查看PTrade底部的“日志”窗口,在回测结束日期会输出如下信息:
-
95%置信度最差回撤:
- 这是最关键的指标。它告诉你:“在95%的概率下,这个策略的最坏回撤不会超过这个数值”。
- 如果你的原始回撤是10%,但蒙特卡洛模拟出的95%最差回撤是30%,说明你的策略在当前回测中可能只是“运气好”,没有遇到不利的交易顺序,实际风险可能远高于回测显示的值。
-
平均最大回撤 vs 原始回撤:
- 如果原始回撤 << 平均最大回撤:说明历史行情的顺序对你非常有利(运气好)。
- 如果原始回撤 >> 平均最大回撤:说明历史行情专门“针对”你的策略(运气差),未来实盘可能表现会好于回测。
-
重抽样方法说明:
- 代码中使用
replace=True(有放回抽样)。这意味着某天的极端大跌可能会在一条模拟路径中出现多次,这能模拟出比历史更极端的“黑天鹅”情形,是压力测试的常用手段。
- 代码中使用
注意事项
- Python版本兼容性:PTrade环境通常支持Python 3,上述代码使用了基础的
numpy和格式化字符串%语法,兼容性良好。 - 计算耗时:如果回测周期非常长(如10年),
g.daily_returns列表会很长。进行1000次模拟可能会导致日志输出有几秒钟的延迟,这是正常的。 - 日期设置:务必修改
initialize函数中的g.end_date为你实际回测的最后一天,否则run_monte_carlo_simulation函数不会被触发。
Q&A
Q1: 为什么要在策略代码里写蒙特卡洛,而不是导出数据在Excel里做?
A1: 在策略代码里直接实现可以实现自动化。每次调整参数或更换股票池后,回测结束即自动生成稳健性报告,无需手动导出数据处理,提高了研究效率。
Q2: 这种模拟方法有什么局限性?
A2: 这种基于“日收益率重排”的方法假设收益率是独立同分布(I.I.D)的,破坏了市场的时间序列相关性(例如波动率聚集效应)。它主要测试的是资金管理和盈亏序列的风险,而不是策略逻辑本身对市场趋势的捕捉能力。
Q3: 代码中的 replace=True 和 replace=False 有什么区别?
A3: replace=False 只是打乱顺序,总收益率在所有模拟路径中是一样的,主要看回撤的变化。replace=True 是有放回抽样(Bootstrap),每条路径的总收益率都会不同,能模拟出历史上未曾发生的极端行情组合,压力测试效果更好。