最大回撤(Max Drawdown)算法
最大回撤(Max Drawdown)算法详解
📋 目录
算法概述
什么是最大回撤(Max Drawdown)?
最大回撤是指在选定周期内,投资组合从峰值到谷底的最大跌幅,是衡量投资风险的核心指标。
指标意义
- MDD = 20%: 账户从历史最高点下跌了 20%
- MDD = 50%: 账户从历史最高点损失了一半
- MDD = 80%: 账户几乎全部亏损,风险极高
为什么需要最大回撤?
最大回撤直观反映了最坏情况下的损失幅度:
| 最大回撤 | 评级 | 恢复难度 |
|---|---|---|
| < 10% | 🏆 优秀 | 需要盈利 11% 恢复 |
| 10% - 20% | ✅ 良好 | 需要盈利 25% 恢复 |
| 20% - 30% | ⚠️ 可接受 | 需要盈利 43% 恢复 |
| 30% - 50% | ❌ 高风险 | 需要盈利 100% 恢复 |
| > 50% | 🚨 极高风险 | 需要盈利 200%+ 恢复 |
恢复公式:
恢复所需收益率 = 1 / (1 - 回撤率) - 1
示例:
- 回撤 20% → 需要盈利 25% 恢复
- 回撤 50% → 需要盈利 100% 恢复
- 回撤 80% → 需要盈利 400% 恢复
核心概念
1. 传统最大回撤计算
回撤 = (峰值 - 当前值) / 峰值 × 100%
最大回撤 = max(所有回撤)
2. 基于真实本金的改进算法
传统方法的问题:
- 基于账户价值或持仓价值
- 受杠杆影响严重
- 不能真实反映风险
改进方案:
每笔交易收益率 = closedPnl / true_capital
累计收益率 = Π(1 + 每笔收益率)
回撤 = (峰值收益率 - 当前收益率) / 峰值收益率
为什么使用真实本金?
| 场景 | 10倍杠杆 | 基于持仓价值(❌) | 基于真实本金(✅) |
|---|---|---|---|
| 投入本金 | $100 | - | - |
| 持仓价值 | $1,000 | - | - |
| 亏损金额 | $50 | - | - |
| 回撤率 | - | -50/1000 = -5% | -50/100 = -50% |
计算公式
核心公式
1. 每笔交易收益率
trade_return_i = closedPnl_i / true_capital
# 限制范围防止极端值
trade_return_i = max(-0.99, min(trade_return_i, 10.0))
2. 累计收益率序列(复利计算)
cumulative_return_i = Π(1 + trade_return_j) for j in [0, i]
# 从1.0开始(代表100%本金)
# 带上限保护,防止溢出
cumulative_return_i = min(cumulative_return_i, 10000.0)
3. 回撤计算
peak = max(cumulative_return_0, ..., cumulative_return_i)
drawdown_i = (peak - cumulative_return_i) / peak × 100%
4. 最大回撤
max_drawdown = max(drawdown_0, drawdown_1, ..., drawdown_n)
# 限制不超过100%
max_drawdown = min(max_drawdown, 100.0)
公式推导
累计收益率复利计算示例
假设交易序列:
交易1: +5% → 累计 = 1.0 × 1.05 = 1.05
交易2: -3% → 累计 = 1.05 × 0.97 = 1.0185
交易3: +10% → 累计 = 1.0185 × 1.10 = 1.12035
交易4: -8% → 累计 = 1.12035 × 0.92 = 1.030722
最大回撤计算:
峰值 = 1.12035(第3笔后)
谷底 = 1.030722(第4笔后)
回撤 = (1.12035 - 1.030722) / 1.12035 = 8.00%
算法实现
核心函数:calculate_max_drawdown_on_capital()
代码位置: apex_fork.py:1067-1205
from typing import List, Dict
from datetime import datetime
def calculate_max_drawdown_on_capital(
fills: List[Dict],
true_capital: float
) -> Dict[str, float]:
"""
基于真实本金计算最大回撤(推荐方法)
关键改进:使用真实本金而非持仓价值计算收益率,完全不受杠杆影响
参数:
fills: 成交记录列表
true_capital: 真实本金(充值 - 提现 + 外部转入 - 外部转出)
返回:
字典,包含:
- max_drawdown_pct: 最大回撤百分比
- peak_return: 峰值累计收益率(百分比)
- trough_return: 谷底累计收益率(百分比)
- peak_date: 峰值发生日期
- trough_date: 谷底发生日期
- total_trades: 分析的交易数量
算法说明:
1. 每笔交易收益率 = closedPnL / true_capital(不是 position_value)
2. 构建累计收益率序列(复利计算)
3. 追踪峰值,计算每个点相对峰值的回撤
4. 记录最大回撤及对应的峰值和谷底
优势:
- ✅ 不受杠杆影响,真实反映风险
- ✅ 与 Sharpe Ratio 计算逻辑一致
- ✅ 不受出入金影响
- ✅ 反映真实的资金使用效率
为什么使用真实本金?
- 10倍杠杆:投入 $100,持仓价值 $1000
- 亏损 $50:
* 旧算法:-50/1000 = -5%(❌ 严重低估)
* 新算法:-50/100 = -50%(✅ 真实风险)
"""
# 边界条件检查
if true_capital <= 0:
return {
"max_drawdown_pct": 0,
"peak_return": 0,
"trough_return": 0,
"peak_date": "N/A",
"trough_date": "N/A",
"total_trades": 0
}
trade_returns = []
trade_times = [] # 记录每笔交易的时间戳
# ==================== 步骤1: 提取交易收益率 ====================
for fill in fills:
closed_pnl = float(fill.get('closedPnl', 0))
# 只分析平仓交易
if closed_pnl == 0:
continue
# ✅ 关键改进:使用真实本金计算收益率
trade_return = closed_pnl / true_capital
# 限制单笔收益率范围:[-0.99, 10.0](防止极端值)
# -0.99 = -99%(最多亏完)
# 10.0 = 1000%(合理的最大盈利上限)
trade_return = max(-0.99, min(trade_return, 10.0))
trade_returns.append(trade_return)
trade_times.append(fill.get('time', 0)) # 记录时间戳
# 数据不足时返回零值
if len(trade_returns) < 2:
return {
"max_drawdown_pct": 0,
"peak_return": 0,
"trough_return": 0,
"peak_date": "N/A",
"trough_date": "N/A",
"total_trades": len(trade_returns)
}
# ==================== 步骤2: 构建累计收益率序列 ====================
cumulative_returns = []
cumulative = 1.0 # 从1.0开始(代表100%本金)
MAX_CUMULATIVE = 10000.0 # 最大累计收益倍数(10000倍 = 1000000%)
for ret in trade_returns:
cumulative *= (1 + ret) # 复利累积
# 防止数值溢出:限制累计收益上限
cumulative = min(cumulative, MAX_CUMULATIVE)
cumulative_returns.append(cumulative)
# ==================== 步骤3: 计算最大回撤 ====================
peak = cumulative_returns[0] # 初始峰值
peak_index = 0 # 峰值位置
max_drawdown = 0 # 最大回撤
trough_value = peak # 谷底值
trough_index = 0 # 谷底位置
for i, value in enumerate(cumulative_returns):
# 更新峰值
if value > peak:
peak = value
peak_index = i
# 计算当前回撤 = (峰值 - 当前值) / 峰值
drawdown = (peak - value) / peak * 100 if peak > 0 else 0
# 更新最大回撤和谷底
if drawdown > max_drawdown:
max_drawdown = drawdown
trough_value = value
trough_index = i
# 限制最大回撤不超过100%
max_drawdown = min(max_drawdown, 100.0)
# ==================== 步骤4: 格式化日期 ====================
def format_date(timestamp_ms: int) -> str:
"""将毫秒时间戳转换为日期字符串"""
if timestamp_ms > 0:
try:
dt = datetime.fromtimestamp(timestamp_ms / 1000)
return dt.strftime('%Y-%m-%d')
except:
return "N/A"
return "N/A"
peak_date = format_date(trade_times[peak_index])
trough_date = format_date(trade_times[trough_index])
return {
"max_drawdown_pct": max_drawdown,
"peak_return": (peak - 1) * 100, # 转换为百分比
"trough_return": (trough_value - 1) * 100,
"peak_date": peak_date,
"trough_date": trough_date,
"total_trades": len(trade_returns)
}
算法流程图
开始
↓
检查 true_capital > 0?
↓ 否
返回零值结果
↓ 是
初始化 trade_returns = [], trade_times = []
↓
遍历成交记录(fills)
↓
对每条记录:
- 获取 closedPnl
- 如果 = 0 → 跳过(开仓交易)
- 计算收益率 = closedPnl / true_capital
- 限制范围 [-0.99, 10.0]
- 添加到 trade_returns 和 trade_times
↓
trade_returns 长度 < 2?
↓ 是
返回零值结果
↓ 否
构建累计收益率序列
↓
初始化:
- cumulative = 1.0
- cumulative_returns = []
↓
对每笔收益率:
- cumulative *= (1 + trade_return)
- 限制上限 min(cumulative, 10000.0)
- 添加到 cumulative_returns
↓
计算最大回撤
↓
初始化:
- peak = cumulative_returns[0]
- max_drawdown = 0
- trough_value = peak
↓
对每个累计收益率:
- 如果 > peak → 更新 peak
- 计算回撤 = (peak - value) / peak × 100
- 如果回撤 > max_drawdown → 更新 max_drawdown 和 trough_value
↓
限制 max_drawdown ≤ 100%
↓
格式化峰值和谷底日期
↓
返回完整结果
↓
结束
应用场景
场景 1:风险评估
# 计算最大回撤
dd_result = calculate_max_drawdown_on_capital(fills, true_capital)
print(f"最大回撤: {dd_result['max_drawdown_pct']:.2f}%")
print(f"峰值收益率: {dd_result['peak_return']:.2f}%")
print(f"谷底收益率: {dd_result['trough_return']:.2f}%")
print(f"峰值日期: {dd_result['peak_date']}")
print(f"谷底日期: {dd_result['trough_date']}")
# 评估风险
if dd_result['max_drawdown_pct'] < 10:
print("🏆 低风险策略")
elif dd_result['max_drawdown_pct'] < 20:
print("✅ 中低风险策略")
elif dd_result['max_drawdown_pct'] < 30:
print("⚠️ 中等风险策略")
else:
print("🚨 高风险策略")
场景 2:计算恢复难度
def calculate_recovery_needed(max_drawdown_pct: float) -> float:
"""计算恢复到峰值所需的收益率"""
if max_drawdown_pct >= 100:
return float('inf') # 完全亏损,无法恢复
recovery_rate = 1 / (1 - max_drawdown_pct / 100) - 1
return recovery_rate * 100
# 示例
mdd = dd_result['max_drawdown_pct']
recovery = calculate_recovery_needed(mdd)
print(f"\n恢复分析:")
print(f"最大回撤: {mdd:.2f}%")
print(f"恢复需要盈利: {recovery:.2f}%")
场景 3:风险调整后的策略选择
def calculate_calmar_ratio(annualized_return: float, max_drawdown: float) -> float:
"""
计算卡玛比率(Calmar Ratio)
= 年化收益率 / 最大回撤
越高越好,反映单位回撤的收益能力
"""
if max_drawdown == 0:
return float('inf') if annualized_return > 0 else 0
return annualized_return / max_drawdown
# 对比策略
strategies = {
"保守策略": {
"annualized_return": 15.0,
"max_drawdown": 8.0
},
"激进策略": {
"annualized_return": 45.0,
"max_drawdown": 35.0
},
"平衡策略": {
"annualized_return": 25.0,
"max_drawdown": 15.0
}
}
print("\n策略对比(卡玛比率):")
for name, metrics in strategies.items():
calmar = calculate_calmar_ratio(
metrics['annualized_return'],
metrics['max_drawdown']
)
print(f"{name}: {calmar:.2f}")
完整代码示例
主程序示例
from typing import List, Dict
from datetime import datetime
class MaxDrawdownCalculator:
"""最大回撤计算器"""
def calculate_max_drawdown_on_capital(
self,
fills: List[Dict],
true_capital: float
) -> Dict[str, float]:
"""计算基于真实本金的最大回撤"""
if true_capital <= 0:
return self._zero_result()
trade_returns = []
trade_times = []
print("\n📊 交易收益率序列:")
print("=" * 90)
print(f"{'序号':<6} {'日期':<12} {'盈亏 ($)':<15} {'收益率 (%)':<15} {'累计收益率'}")
print("=" * 90)
cumulative = 1.0
for i, fill in enumerate(fills, 1):
closed_pnl = float(fill.get('closedPnl', 0))
if closed_pnl == 0:
continue
# 计算交易收益率
trade_return = closed_pnl / true_capital
trade_return = max(-0.99, min(trade_return, 10.0))
trade_returns.append(trade_return)
# 记录时间
timestamp = fill.get('time', 0)
trade_times.append(timestamp)
date_str = self._format_date(timestamp)
# 计算累计收益率
cumulative *= (1 + trade_return)
cumulative = min(cumulative, 10000.0)
# 打印详情
return_pct = trade_return * 100
cumulative_pct = (cumulative - 1) * 100
status = "✅" if closed_pnl > 0 else "❌"
print(f"{i:<6} {date_str:<12} ${closed_pnl:<14,.2f} "
f"{return_pct:<14.4f}% {cumulative_pct:>10.2f}% {status}")
print("=" * 90)
if len(trade_returns) < 2:
print("⚠️ 数据不足,无法计算最大回撤(需要至少2笔交易)")
return self._zero_result(len(trade_returns))
# 构建累计收益率序列
cumulative_returns = []
cumulative = 1.0
for ret in trade_returns:
cumulative *= (1 + ret)
cumulative = min(cumulative, 10000.0)
cumulative_returns.append(cumulative)
# 计算最大回撤
peak = cumulative_returns[0]
peak_index = 0
max_drawdown = 0
trough_value = peak
trough_index = 0
print("\n📈 回撤分析:")
print("=" * 90)
print(f"{'时点':<6} {'累计收益率':<15} {'当前峰值':<15} {'回撤':<15} {'状态'}")
print("=" * 90)
for i, value in enumerate(cumulative_returns):
# 更新峰值
if value > peak:
peak = value
peak_index = i
# 计算回撤
drawdown = (peak - value) / peak * 100 if peak > 0 else 0
# 更新最大回撤
if drawdown > max_drawdown:
max_drawdown = drawdown
trough_value = value
trough_index = i
# 打印关键时点
if drawdown > 0 or i == peak_index:
cumulative_pct = (value - 1) * 100
peak_pct = (peak - 1) * 100
status = "📍 峰值" if i == peak_index else f"📉 -{drawdown:.2f}%"
print(f"{i+1:<6} {cumulative_pct:<14.2f}% {peak_pct:<14.2f}% "
f"{drawdown:<14.2f}% {status}")
print("=" * 90)
# 限制最大回撤
max_drawdown = min(max_drawdown, 100.0)
# 格式化日期
peak_date = self._format_date(trade_times[peak_index])
trough_date = self._format_date(trade_times[trough_index])
# 打印结果
print(f"\n🎯 最大回撤结果:")
print("=" * 90)
print(f"最大回撤: {max_drawdown:.2f}%")
print(f"峰值收益率: {(peak - 1) * 100:.2f}% (日期: {peak_date})")
print(f"谷底收益率: {(trough_value - 1) * 100:.2f}% (日期: {trough_date})")
print(f"分析交易数: {len(trade_returns)}")
# 评级
if max_drawdown < 10:
rating = "🏆 优秀"
elif max_drawdown < 20:
rating = "✅ 良好"
elif max_drawdown < 30:
rating = "⚠️ 可接受"
else:
rating = "🚨 高风险"
print(f"风险评级: {rating}")
# 恢复难度
if max_drawdown < 100:
recovery_needed = 1 / (1 - max_drawdown / 100) - 1
print(f"恢复需要盈利: {recovery_needed * 100:.2f}%")
return {
"max_drawdown_pct": max_drawdown,
"peak_return": (peak - 1) * 100,
"trough_return": (trough_value - 1) * 100,
"peak_date": peak_date,
"trough_date": trough_date,
"total_trades": len(trade_returns)
}
def _format_date(self, timestamp_ms: int) -> str:
"""格式化时间戳"""
if timestamp_ms > 0:
try:
dt = datetime.fromtimestamp(timestamp_ms / 1000)
return dt.strftime('%Y-%m-%d')
except:
return "N/A"
return "N/A"
def _zero_result(self, total_trades: int = 0) -> Dict[str, float]:
"""返回零值结果"""
return {
"max_drawdown_pct": 0,
"peak_return": 0,
"trough_return": 0,
"peak_date": "N/A",
"trough_date": "N/A",
"total_trades": total_trades
}
# 使用示例
if __name__ == "__main__":
calculator = MaxDrawdownCalculator()
# 模拟交易数据(包含一次大幅回撤)
import time
current_time = int(time.time() * 1000)
fills = [
{"closedPnl": "500", "time": current_time - 86400000 * 20},
{"closedPnl": "300", "time": current_time - 86400000 * 18},
{"closedPnl": "800", "time": current_time - 86400000 * 16},
{"closedPnl": "400", "time": current_time - 86400000 * 14}, # 峰值
{"closedPnl": "-600", "time": current_time - 86400000 * 12}, # 开始回撤
{"closedPnl": "-400", "time": current_time - 86400000 * 10}, # 继续回撤
{"closedPnl": "-300", "time": current_time - 86400000 * 8}, # 谷底
{"closedPnl": "200", "time": current_time - 86400000 * 6}, # 开始恢复
{"closedPnl": "350", "time": current_time - 86400000 * 4},
{"closedPnl": "450", "time": current_time - 86400000 * 2},
]
# 真实本金
true_capital = 10000.0
# 计算最大回撤
result = calculator.calculate_max_drawdown_on_capital(fills, true_capital)
运行结果示例
📊 交易收益率序列:
==========================================================================================
序号 日期 盈亏 ($) 收益率 (%) 累计收益率
==========================================================================================
1 2026-01-14 $500.00 5.0000% 5.00% ✅
2 2026-01-16 $300.00 3.0000% 8.15% ✅
3 2026-01-18 $800.00 8.0000% 16.81% ✅
4 2026-01-20 $400.00 4.0000% 21.48% ✅
5 2026-01-22 $-600.00 -6.0000% 14.19% ❌
6 2026-01-24 $-400.00 -4.0000% 9.62% ❌
7 2026-01-26 $-300.00 -3.0000% 6.34% ❌
8 2026-01-28 $200.00 2.0000% 8.46% ✅
9 2026-01-30 $350.00 3.5000% 12.26% ✅
10 2026-02-01 $450.00 4.5000% 17.31% ✅
==========================================================================================
📈 回撤分析:
==========================================================================================
时点 累计收益率 当前峰值 回撤 状态
==========================================================================================
4 21.48% 21.48% 0.00% 📍 峰值
5 14.19% 21.48% 6.00% 📉 -6.00%
6 9.62% 21.48% 9.76% 📉 -9.76%
7 6.34% 21.48% 12.47% 📉 -12.47%
==========================================================================================
🎯 最大回撤结果:
==========================================================================================
最大回撤: 12.47%
峰值收益率: 21.48% (日期: 2026-01-20)
谷底收益率: 6.34% (日期: 2026-01-26)
分析交易数: 10
风险评级: ✅ 良好
恢复需要盈利: 14.25%
测试用例
测试用例 1:正常回撤
def test_normal_drawdown():
"""测试用例 1: 正常回撤场景"""
calculator = MaxDrawdownCalculator()
fills = [
{"closedPnl": "100", "time": 1000000}, # +10%
{"closedPnl": "200", "time": 2000000}, # +20%,峰值
{"closedPnl": "-150", "time": 3000000}, # -15%,回撤
{"closedPnl": "100", "time": 4000000}, # +10%,恢复
]
result = calculator.calculate_max_drawdown_on_capital(fills, 1000)
assert result['max_drawdown_pct'] > 0
assert result['max_drawdown_pct'] < 100
print(f"✅ 测试用例 1 通过:最大回撤 = {result['max_drawdown_pct']:.2f}%")
测试用例 2:持续上涨无回撤
def test_no_drawdown():
"""测试用例 2: 持续上涨,无回撤"""
calculator = MaxDrawdownCalculator()
fills = [
{"closedPnl": "100", "time": 1000000},
{"closedPnl": "200", "time": 2000000},
{"closedPnl": "150", "time": 3000000},
{"closedPnl": "300", "time": 4000000},
]
result = calculator.calculate_max_drawdown_on_capital(fills, 1000)
# 持续上涨,最大回撤应该非常小(接近0)
assert result['max_drawdown_pct'] < 1
print(f"✅ 测试用例 2 通过:无回撤场景")
测试用例 3:极端回撤
def test_extreme_drawdown():
"""测试用例 3: 极端回撤场景"""
calculator = MaxDrawdownCalculator()
fills = [
{"closedPnl": "500", "time": 1000000}, # +50%
{"closedPnl": "-800", "time": 2000000}, # -80%,巨大回撤
{"closedPnl": "200", "time": 3000000}, # 尝试恢复
]
result = calculator.calculate_max_drawdown_on_capital(fills, 1000)
assert result['max_drawdown_pct'] > 50 # 应该是较大回撤
print(f"✅ 测试用例 3 通过:极端回撤 = {result['max_drawdown_pct']:.2f}%")
测试用例 4:杠杆影响对比
def test_leverage_impact():
"""测试用例 4: 杠杆对回撤的影响"""
calculator = MaxDrawdownCalculator()
fills = [
{"closedPnl": "1000", "time": 1000000}, # 盈利
{"closedPnl": "-500", "time": 2000000}, # 回撤
]
# 场景1: 无杠杆,本金 $10,000
result_no_leverage = calculator.calculate_max_drawdown_on_capital(fills, 10000)
# 场景2: 10倍杠杆,本金 $1,000
result_with_leverage = calculator.calculate_max_drawdown_on_capital(fills, 1000)
print(f"✅ 测试用例 4 通过:")
print(f" 无杠杆回撤: {result_no_leverage['max_drawdown_pct']:.2f}%")
print(f" 高杠杆回撤: {result_with_leverage['max_drawdown_pct']:.2f}%")
print(f" 差异倍数: {result_with_leverage['max_drawdown_pct'] / result_no_leverage['max_drawdown_pct']:.2f}x")
测试用例 5:数据不足
def test_insufficient_data():
"""测试用例 5: 数据不足场景"""
calculator = MaxDrawdownCalculator()
fills = [
{"closedPnl": "100", "time": 1000000},
]
result = calculator.calculate_max_drawdown_on_capital(fills, 1000)
assert result['max_drawdown_pct'] == 0
assert result['total_trades'] == 1
print("✅ 测试用例 5 通过:数据不足处理正确")
运行所有测试
if __name__ == "__main__":
print("开始运行最大回撤测试用例...\n")
test_normal_drawdown()
test_no_drawdown()
test_extreme_drawdown()
test_leverage_impact()
test_insufficient_data()
print("\n🎉 所有测试用例通过!")
算法优势
| 优势 | 说明 |
|---|---|
| ✅ 不受杠杆影响 | 基于真实本金,真实反映风险 |
| ✅ 不受出入金影响 | 使用净入金,避免计算偏差 |
| ✅ 复利计算 | 真实反映累计收益和回撤 |
| ✅ 溢出保护 | 限制极端值,防止计算错误 |
| ✅ 详细记录 | 记录峰值和谷底的时间和数值 |
参考资料
- Hyperliquid API 文档: https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api
- Apex Liquid Bot: https://apexliquid.bot/
- 源代码: apex_fork.py (行 1067-1205)
文档生成时间: 2026-02-03
作者: Apex Calculator Team