当前位置: 首页 > news >正文

多套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解决实际问题,如果你觉得这个工具好用,欢迎关注长安牧笛!

http://www.jsqmd.com/news/1072061/

相关文章:

  • Windows AirPlay 2接收器终极指南:5分钟让PC变身苹果设备无线投屏中心
  • 5分钟快速部署指南:让Windows电脑完美支持AirPlay 2投屏功能
  • 2026年乌鲁木齐先装后付装修生产厂家top5实践经验分享
  • 非同名入金与非同名代付为两类不同的异名资金操作:
  • 如何在5分钟内用Blender完成建筑建模?ArchiPack参数化插件深度解析
  • 终极图像隐写分析指南:如何使用ImageStrike一站式解决18种CTF挑战
  • 为什么92%的企业卡在Level 3?AISMM Level 4的4个隐藏准入门槛,及2026年前最后窗口期应对策略
  • 【绝密档案】奇点大会内部培训手册节选:AI人才成熟度5阶跃迁路线图(含L3→L4突破性训练包)
  • 原神模型导入神器GIMI:3分钟让你成为游戏角色造型师
  • Root 选举 + Beacon + TDMA 切换功能实现
  • 终极指南:三步快速上手开源制造执行系统openMES
  • 5分钟掌握Spek:免费开源的终极音频频谱分析器指南
  • AI预测模型的高盛下调黄金目标价500美元背后:金价定价逻辑重构预测模型
  • AltSnap:如何通过零注入架构实现Windows窗口管理的革命性突破?
  • Path of Building PoE2:流放之路2构建模拟器的终极指南
  • API Key怎么安全保存?环境变量、本地配置和团队权限管理清单
  • ClawHub曝供应链安全危机:23款冒牌插件潜伏AI代理生态,开发者险些“引狼入室“
  • 【操作系统】进程状态转换(三态模型/五态模型)
  • 【AISMM Level 1权威解码】:SITS 2026初始级企业必现的5大典型特征与避坑指南
  • 【每日一题】LeetCode 560. 和为 K 的子数组 TypeScript
  • 机器学习特征工程:从原始数据到模型输入
  • 终极指南:如何使用League-Toolkit的OP.GG数据功能提升英雄联盟游戏表现
  • 如何用5分钟将单张图片转换为专业PSD分层文件:Layerdivider完全指南
  • jQuery:那个改变前端的库,现在活到了 4.0
  • 3大核心技术突破:解密Bodymovin插件的高效动画转换机制
  • 3步掌握kohya_ss可视化训练监控:从新手到专家的终极指南
  • OpenRocket火箭设计软件:从零开始掌握专业级火箭仿真
  • 如何用Super IO实现Blender高效导入导出:新手也能掌握的完整指南
  • 数字音乐跨平台播放终极解决方案:一站式解决格式兼容性问题
  • 生产环境采样策略:如何平衡数据完整性与存储成本?