多套AI策略夏普比率，最大回撤批量计算程序，自动横向排名。
多策略夏普比率与最大回撤批量计算及横向排名工具
定位：策略研发阶段的「多策略横向对比与筛选」工具
适用：课程实验、策略竞赛、多因子模型对比
语言：Python 3.8+
一、实际应用场景描述
在《智能证券投资》课程或实际量化研究中，一个常见场景是：
"我有 10 套策略，到底哪套最好？"
具体来说，研究者可能需要：
1. 同时维护多套策略（不同因子组合、不同参数、不同模型）
2. 每套策略都有回测结果（日收益率序列）
3. 需要从 风险调整后收益 的角度进行横向对比
4. 最终选出 综合表现最优 的 1~3 套策略
如果手动逐个算夏普比率、最大回撤，再人工排序，既低效又容易出错。
二、引入痛点
🔴 痛点 1：指标口径不一致
问题 后果
有的算算术夏普，有的算对数夏普 不可比
有的年化用 252，有的用 365 结果偏差
有的用无风险利率，有的不用 口径混乱
👉 不同策略的"好"没有统一标准
🔴 痛点 2：批量计算重复劳动
- 10 套策略 = 10 次手动计算
- 每次改因子或参数，全部重算
- 没有自动化脚本
👉 研究效率极低
🔴 痛点 3：排名维度单一
- 只看收益率 → 忽略风险
- 只看夏普 → 忽略回撤
- 没有综合评分
👉 "最好"的定义模糊
三、核心逻辑讲解
✅ 夏普比率（Sharpe Ratio）
衡量 每承受一单位风险，获得多少超额收益：
SR = (策略日均收益 - 无风险利率) / 策略收益波动率 × √交易日数
夏普值 解读
< 0 跑输无风险利率
0 ~ 1 一般
1 ~ 2 较好
> 2 优秀（需警惕过拟合）
✅ 最大回撤（Maximum Drawdown）
衡量 从历史最高点到最低点的最大跌幅：
回撤_t = (净值_t - 历史最高净值) / 历史最高净值
最大回撤 = min(回撤_t)
最大回撤 解读
< 5% 非常稳健
5% ~ 15% 正常
15% ~ 25% 偏高
> 25% 高风险
✅ 综合排名机制
采用 百分位打分法（Percentile Ranking）：
综合得分 = w1 × 夏普百分位排名 + w2 × 回撤百分位排名
夏普越高越好，回撤越低越好，两者加权得到综合排名。
四、代码模块化实现
📁 项目结构
strategy_comparison/
│
├── config.py # 参数配置
├── data_loader.py # 数据加载
├── metrics.py # 指标计算（夏普、回撤）
├── ranker.py # 横向排名与综合评分
├── reporter.py # 报告生成与可视化
├── main.py # 程序入口
└── README.md
config.py
# config.py
from pathlib import Path
from datetime import date
# 数据目录
DATA_DIR = Path(__file__).parent / "data"
# 回测参数
START_DATE = date(2020, 1, 1)
END_DATE = date(2024, 12, 31)
RISK_FREE_RATE = 0.0 # 无风险利率（日化），设为 0 表示计算超额收益版
ANNUALIZE_FACTOR = 252 # 年化因子（A 股交易日约 252 天）
# 综合排名权重
SHARPE_WEIGHT = 0.5 # 夏普比率权重
DRAWDOWN_WEIGHT = 0.5 # 最大回撤权重（绝对值越大越差）
# 策略文件命名规则
# data/returns/strategy_{name}.csv
# 字段：date, return
data_loader.py
# data_loader.py
import pandas as pd
from pathlib import Path
from config import DATA_DIR, START_DATE, END_DATE
def load_strategy_returns(strategy_name: str) -> pd.Series:
"""
加载单套策略的日收益率序列
Args:
strategy_name: 策略名称，对应文件名 strategy_{name}.csv
Returns:
以 date 为索引的 return 序列
"""
file_path = DATA_DIR / "returns" / f"strategy_{strategy_name}.csv"
if not file_path.exists():
raise FileNotFoundError(f"策略文件不存在: {file_path}")
df = pd.read_csv(file_path, parse_dates=["date"])
df = df.set_index("date")["return"]
# 按日期范围过滤
mask = (df.index >= START_DATE) & (df.index <= END_DATE)
df = df[mask]
return df.sort_index()
def load_all_strategies(strategy_names: list[str]) -> dict[str, pd.Series]:
"""
批量加载多套策略的收益率数据
Returns:
{策略名: 日收益率 Series}
"""
results = {}
for name in strategy_names:
try:
results[name] = load_strategy_returns(name)
print(f" ✓ 已加载策略: {name} ({len(results[name])} 个交易日)")
except FileNotFoundError as e:
print(f" ✗ 跳过策略: {name}（文件不存在）")
print(f"\n[数据加载] 共加载 {len(results)} 套策略\n")
return results
metrics.py
# metrics.py
import numpy as np
import pandas as pd
from config import RISK_FREE_RATE, ANNUALIZE_FACTOR
def calc_sharpe_ratio(returns: pd.Series) -> float:
"""
计算年化夏普比率
公式: SR = (E[R] - Rf) / σ(R) × √T
"""
if len(returns) < 2:
return np.nan
excess_returns = returns - RISK_FREE_RATE
mean_excess = excess_returns.mean()
std_excess = excess_returns.std()
if std_excess == 0:
return np.nan
daily_sharpe = mean_excess / std_excess
annualized_sharpe = daily_sharpe * np.sqrt(ANNUALIZE_FACTOR)
return float(annualized_sharpe)
def calc_max_drawdown(returns: pd.Series) -> float:
"""
计算最大回撤（负值，如 -0.15 表示 15%）
公式: MDD = min((净值_t - 历史最高净值) / 历史最高净值)
"""
if len(returns) < 2:
return np.nan
nav = (1 + returns).cumprod() # 累积净值
running_max = nav.cummax() # 历史最高净值
drawdown = (nav - running_max) / running_max # 回撤序列
return float(drawdown.min())
def calc_total_return(returns: pd.Series) -> float:
"""计算总收益"""
if len(returns) < 1:
return np.nan
return float((1 + returns).prod() - 1)
def calc_win_rate(returns: pd.Series) -> float:
"""计算胜率"""
if len(returns) < 1:
return np.nan
return float((returns > 0).mean())
def calc_annualized_return(returns: pd.Series) -> float:
"""计算年化收益"""
if len(returns) < 2:
return np.nan
total_return = (1 + returns).prod() - 1
n_days = len(returns)
annualized = (1 + total_return) ** (ANNUALIZE_FACTOR / n_days) - 1
return float(annualized)
def evaluate_strategy(returns: pd.Series, strategy_name: str) -> dict:
"""
对单套策略计算全部绩效指标
"""
return {
"strategy": strategy_name,
"trading_days": len(returns),
"total_return": calc_total_return(returns),
"annualized_return": calc_annualized_return(returns),
"sharpe_ratio": calc_sharpe_ratio(returns),
"max_drawdown": calc_max_drawdown(returns),
"win_rate": calc_win_rate(returns),
}
ranker.py
# ranker.py
import numpy as np
import pandas as pd
from config import SHARPE_WEIGHT, DRAWDOWN_WEIGHT
def rank_strategies(metrics_df: pd.DataFrame) -> pd.DataFrame:
"""
对多套策略进行横向排名
排名规则：
- 夏普比率：降序排列（越高越好）
- 最大回撤：升序排列（越小越好，即越接近 0 越好）
- 综合得分：加权平均
"""
df = metrics_df.copy()
# 1. 夏普比率排名（降序 → 排名越小越好）
df["sharpe_rank"] = df["sharpe_ratio"].rank(ascending=False, method="average")
# 2. 最大回撤排名（升序 → 回撤越小排名越靠前）
# 注意：max_drawdown 是负数，所以升序 = 绝对值越小越靠前
df["drawdown_rank"] = df["max_drawdown"].rank(ascending=True, method="average")
# 3. 综合得分（百分制）
# 将排名转换为百分位得分
n = len(df)
df["sharpe_score"] = (n - df["sharpe_rank"] + 1) / n * 100
df["drawdown_score"] = (n - df["drawdown_rank"] + 1) / n * 100
# 4. 综合得分
df["composite_score"] = (
SHARPE_WEIGHT * df["sharpe_score"] +
DRAWDOWN_WEIGHT * df["drawdown_score"]
)
# 5. 最终排名
df["final_rank"] = df["composite_score"].rank(ascending=False, method="min")
# 排序
df = df.sort_values("final_rank").reset_index(drop=True)
return df
def format_metrics_df(df: pd.DataFrame) -> pd.DataFrame:
"""格式化输出"""
formatted = df[[
"final_rank", "strategy", "total_return", "annualized_return",
"sharpe_ratio", "max_drawdown", "win_rate",
"sharpe_rank", "drawdown_rank", "composite_score"
]].copy()
# 重命名列
formatted.columns = [
"综合排名", "策略名称", "总收益", "年化收益",
"夏普比率", "最大回撤", "胜率",
"夏普排名", "回撤排名", "综合得分"
]
# 格式化数值
formatted["总收益"] = formatted["总收益"].apply(lambda x: f"{x:.2%}")
formatted["年化收益"] = formatted["年化收益"].apply(lambda x: f"{x:.2%}")
formatted["夏普比率"] = formatted["夏普比率"].apply(lambda x: f"{x:.4f}")
formatted["最大回撤"] = formatted["最大回撤"].apply(lambda x: f"{x:.2%}")
formatted["胜率"] = formatted["胜率"].apply(lambda x: f"{x:.2%}")
formatted["综合得分"] = formatted["综合得分"].apply(lambda x: f"{x:.1f}")
return formatted
reporter.py
# reporter.py
import pandas as pd
import numpy as np
def print_rankings(formatted_df: pd.DataFrame):
"""打印排名表格"""
print("\n" + "=" * 90)
print(" 多策略夏普比率 & 最大回撤 横向排名")
print("=" * 90)
print(formatted_df.to_string(index=False))
print("=" * 90)
def print_summary(ranked_df: pd.DataFrame):
"""打印摘要统计"""
print("\n--- 摘要统计 ---")
best_overall = ranked_df.loc[ranked_df["final_rank"].idxmin()]
best_sharpe = ranked_df.loc[ranked_df["sharpe_ratio"].idxmax()]
best_drawdown = ranked_df.loc[ranked_df["max_drawdown"].idxmax()]
print(f"综合最优策略: {best_overall['strategy']}")
print(f" 夏普最优策略: {best_sharpe['strategy']} (SR={best_sharpe['sharpe_ratio']:.4f})")
print(f" 回撤最小策略: {best_drawdown['strategy']} (MDD={best_drawdown['max_drawdown']:.2%})")
# 夏普分布
sharpe_values = ranked_df["sharpe_ratio"].dropna()
print(f"\n夏普比率分布:")
print(f" 最高: {sharpe_values.max():.4f}")
print(f" 最低: {sharpe_values.min():.4f}")
print(f" 均值: {sharpe_values.mean():.4f}")
print(f" 标准差: {sharpe_values.std():.4f}")
def export_to_csv(ranked_df: pd.DataFrame, filepath: str = "strategy_rankings.csv"):
"""导出完整结果到 CSV"""
export_df = ranked_df.copy()
export_df.to_csv(filepath, index=False, encoding="utf-8-sig")
print(f"\n[导出] 完整结果已保存至: {filepath}")
main.py
# main.py
import pandas as pd
from data_loader import load_all_strategies
from metrics import evaluate_strategy
from ranker import rank_strategies, format_metrics_df
from reporter import print_rankings, print_summary, export_to_csv
def main():
# ========== 配置区 ==========
# 策略名称列表（对应 data/returns/strategy_{name}.csv）
STRATEGY_NAMES = [
"value_factor",
"momentum_factor",
"quality_factor",
"low_volatility",
"growth_factor",
"multi_factor_v1",
"multi_factor_v2",
"ml_lgbm",
"ml_xgboost",
"hybrid_smart",
]
# ============================
print("=" * 60)
print(" 多策略夏普比率 & 最大回撤 批量计算")
print("=" * 60)
# 1. 批量加载策略数据
print("\n[第一步] 加载策略收益率数据...")
strategies = load_all_strategies(STRATEGY_NAMES)
if len(strategies) == 0:
print("\n❌ 未加载到任何策略数据，请检查 data/returns/ 目录")
return
# 2. 逐策略计算绩效指标
print("[第二步] 计算各策略绩效指标...")
metrics_list = []
for name, returns in strategies.items():
metrics = evaluate_strategy(returns, name)
metrics_list.append(metrics)
metrics_df = pd.DataFrame(metrics_list)
# 3. 横向排名
print("[第三步] 执行横向排名...")
ranked_df = rank_strategies(metrics_df)
# 4. 格式化输出
formatted = format_metrics_df(ranked_df)
# 5. 打印报告
print_rankings(formatted)
print_summary(ranked_df)
# 6. 导出
export_to_csv(ranked_df)
print("\n✅ 全部完成！")
if __name__ == "__main__":
main()
五、README 文件
# 多策略夏普比率 & 最大回撤批量计算与横向排名工具
## 功能
- 批量计算多套策略的夏普比率和最大回撤
- 自动横向排名，支持加权综合评分
- 输出标准化排名表格和摘要统计
## 快速开始
### 1. 准备数据
在 data/returns/ 目录下放置各策略的日收益率文件：
strategy_value_factor.csv:
date,return
2020-01-02,0.0120
2020-01-03,-0.0050
### 2. 配置策略列表
编辑 main.py 中的 STRATEGY_NAMES 列表：
STRATEGY_NAMES = [
"value_factor",
"momentum_factor",
...
]
### 3. 运行
pip install pandas numpy
python main.py
## 输出示例
==========================================================
多策略夏普比率 & 最大回撤 横向排名
==========================================================
综合排名 策略名称 总收益 年化收益 夏普比率 最大回撤 胜率
1 hybrid_smart 45.20% 8.50% 1.5230 -12.30% 56.80%
2 multi_factor_v2 38.60% 7.20% 1.4120 -14.10% 54.30%
3 ml_lgbm 32.10% 6.10% 1.2800 -15.80% 52.90%
...
## 排名规则
| 指标 | 方向 | 说明 |
|---|---|---|
| 夏普比率 | 越高越好 | 风险调整后收益 |
| 最大回撤 | 越低越好 | 下行风险控制 |
| 综合得分 | 加权平均 | 可配置权重 |
默认权重：夏普 50% + 回撤 50%
可通过 config.py 调整 SHARPE_WEIGHT 和 DRAWDOWN_WEIGHT
六、核心知识点卡片
【知识点卡片：策略横向对比与排名】
1️⃣ 夏普比率（Sharpe Ratio）
- 衡量单位风险的超额收益
- 年化公式：SR_daily × √252
- 局限性：假设收益正态分布，对肥尾不敏感
2️⃣ 最大回撤（Maximum Drawdown）
- 衡量极端下行风险
- 比波动率更直观，投资者更敏感
- 局限性：对回测起止时间敏感
3️⃣ 百分位排名法
- 将不同量纲的指标统一到同一尺度
- 避免某单一指标主导排名
- 加权方案可根据研究目标调整
4️⃣ 多策略对比的注意事项
- 确保所有策略使用相同的时间区间
- 统一年化因子和交易日数
- 注意幸存者偏差（淘汰的策略是否被纳入对比）
- 样本外验证结果应单独标注
七、免责声明与风险提示
⚠️ 免责声明
- 本程序仅供 学术研究与课程实验 使用
- 不构成任何投资建议或投资依据
- 排名结果仅反映历史回测表现
⚠️ 风险提示
- 历史夏普比率高 ≠ 未来表现好
- 回测最大回撤可能被低估（未考虑极端行情）
- 多策略排名可能受时间区间选择影响
- 权重设置具有主观性，不同权重可能导致不同排名
- 建议结合样本外验证、滚动窗口分析综合判断
八、总结
本文构建了一个 多策略绩效指标批量计算与横向排名工具，核心要点：
1. 批量计算：一套代码覆盖所有策略，避免重复劳动
2. 统一口径：年化因子、无风险利率、排名方法全部标准化
3. 多维排名：同时考虑夏普比率和最大回撤，避免单一指标误导
4. 可扩展：可轻松加入 Sortino、Calmar、信息比率等指标
核心原则：
选策略不是选"回测最好看的"，而是选 风险调整后最稳健的。
本文代码仅供学习与技术交流，不构成任何投资建议，股市有风险，入市需谨慎！
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
