问题描述
怎么计算策略回撤的最长恢复时间？
解决方案
在 QMT（Quant Trader）量化交易平台中，计算策略回撤的最长恢复时间（Maximum Drawdown Recovery Time），通常需要在回测结束时，利用记录下来的净值曲线进行分析。
最长恢复时间指的是：策略从一个权益高点跌落后，直到重新创出新高所经历的最长时间段。
以下是具体的实现逻辑和代码示例。
核心逻辑
- 数据收集：在
handlebar运行过程中，记录每个时间点（K线时间）的策略净值。 - 计算累计最大值：利用 Pandas 计算净值的累计最大值（High Water Mark）。
- 标记创新高时间：记录每一次净值创新高的时间点。
- 计算恢复期：对于每一个交易日，计算其距离“上一次创新高时间”的时间差。
- 取最大值：所有时间差中的最大值即为最长恢复时间。
QMT 策略代码实现
你可以将以下代码直接复制到 QMT 的策略编辑器中进行回测。代码会在回测结束（最后一根 K 线）时输出计算结果。
# -*- coding: gbk -*-
import pandas as pd
import numpy as np
def init(ContextInfo):
# 1. 设置回测参数（示例）
ContextInfo.set_account('account_id')
ContextInfo.set_universe(['000300.SH'])
ContextInfo.set_commission(0.0003)
ContextInfo.set_slippage(1, 0.0)
# 2. 初始化用于存储净值数据的列表
# 格式: [{'date': '2023-01-01', 'nav': 1.02}, ...]
ContextInfo.nav_records = []
def handlebar(ContextInfo):
# 简单示例策略：始终持有
# 实际使用时，这里是你的交易逻辑
if ContextInfo.is_new_bar():
passorder(23, 1101, ContextInfo.accid, '000300.SH', 5, -1, 1, ContextInfo)
# -------------------------------------------------------
# 数据记录逻辑
# -------------------------------------------------------
# 获取当前K线的时间戳并转换为日期字符串
timetag = ContextInfo.get_bar_timetag(ContextInfo.barpos)
date_str = timetag_to_datetime(timetag, '%Y-%m-%d')
# 获取当前策略净值
# 注意：get_net_value 在回测模式下有效
nav = ContextInfo.get_net_value(ContextInfo.barpos)
# 记录数据
ContextInfo.nav_records.append({
'date': date_str,
'nav': nav
})
# -------------------------------------------------------
# 在回测结束时进行计算
# -------------------------------------------------------
if ContextInfo.is_last_bar():
calculate_max_recovery_time(ContextInfo)
def calculate_max_recovery_time(ContextInfo):
"""
计算最长回撤恢复时间
"""
if not ContextInfo.nav_records:
print("没有净值记录，无法计算。")
return
# 1. 将记录转换为 Pandas DataFrame
df = pd.DataFrame(ContextInfo.nav_records)
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)
# 2. 计算累计最大净值 (High Water Mark)
df['high_water_mark'] = df['nav'].cummax()
# 3. 标记是否处于回撤状态 (当前净值 < 历史最高)
# 如果 nav < high_water_mark，说明处于回撤期
# 如果 nav == high_water_mark，说明创新高（恢复完成）
# 4. 计算恢复天数逻辑
# 我们需要找到每个时间点，距离“上一次创新高时间”的天数
# 创建一个序列记录“最近一次创新高的日期”
df['last_peak_date'] = df.index
# 只有在创新高的时候保留日期，否则设为NaN
df.loc[df['nav'] < df['high_water_mark'], 'last_peak_date'] = np.nan
# 前向填充，这样每一行都知道最近一次创新高是哪天
df['last_peak_date'] = df['last_peak_date'].fillna(method='ffill')
# 计算当前日期距离最近一次创新高日期的天数差
df['recovery_days'] = (df.index - df['last_peak_date']).dt.days
# 5. 获取最大值
max_recovery_days = df['recovery_days'].max()
# 找到发生最长恢复期的结束日期（便于排查）
if max_recovery_days > 0:
end_date_of_longest = df['recovery_days'].idxmax()
start_date_of_longest = df.loc[end_date_of_longest, 'last_peak_date']
print("=" * 50)
print(f"【回撤恢复分析结果】")
print(f"最长回撤恢复时间: {max_recovery_days} 天")
print(f"该期间范围: {start_date_of_longest.date()} 至 {end_date_of_longest.date()}")
print("=" * 50)
else:
print("策略表现完美，无回撤或始终创新高。")
代码详解
ContextInfo.nav_records: 我们在init中创建一个列表，用来在内存中暂存每一根 K 线的日期和净值。timetag_to_datetime: QMT 的时间戳是毫秒级的，使用此函数将其转换为 Pandas 可识别的日期格式（YYYY-MM-DD）。df['nav'].cummax(): 这是 Pandas 的核心函数，用于计算序列的累计最大值。如果当前净值小于这个值，说明策略处于回撤中。fillna(method='ffill'): 这是计算恢复时间的关键技巧。- 我们首先标记出所有创新高的日期。
- 对于处于回撤期的日子，我们将它的“最近创新高日期”设为 NaN。
- 使用
ffill（前向填充），将最近的一个有效日期（即峰值日期）填充给后续的回撤日子。 - 最后用
当前日期 - 最近峰值日期即可得到当前的恢复时长。
is_last_bar(): 确保计算逻辑只在回测结束时运行一次，避免在每一根 K 线都进行繁重的 DataFrame 运算，提高回测速度。
注意事项
- 周期选择：上述代码计算的是自然日（Calendar Days）。如果你使用的是日线回测，结果就是天数。如果你使用的是分钟线回测，
dt.days依然计算的是日期差。如果需要计算“交易 K 线数量”，可以将索引改为整数索引（0, 1, 2...）直接相减。 - 未恢复的回撤：如果策略在回测结束时正处于一个巨大的回撤中（尚未恢复），上述逻辑依然会计算这段时间，并将其视为潜在的最长恢复时间之一。这是符合风险控制逻辑的。
Q&A
Q: 为什么使用 ContextInfo.get_net_value 而不是自己计算资产？
A: ContextInfo.get_net_value 是 QMT 官方提供的接口，它自动处理了分红、送股、出入金等对净值的影响，比手动用 ContextInfo.capital + 持仓市值 计算更准确且方便。
Q: 这段代码可以在实盘中使用吗？
A: 理论上可以，但实盘中 is_last_bar() 的含义是“最新的一根 K 线”。在实盘模式下，通常不需要实时计算历史最长恢复时间，建议在盘后分析或回测阶段使用此逻辑。
Q: 如何计算最长恢复的 K 线根数而不是天数？
A: 在代码中，不要使用 df.index (日期) 相减。可以创建一个从 0 开始的整数列 df['bar_count'] = range(len(df))，然后用同样的逻辑计算 bar_count 的差值。