feat(report): 策略KPI数据统一来源,避免重复计算

修改内容:
1. strategies/rotation/report.py
   - 新增保存策略KPI到JSON文件的功能
   - 保存指标:累计收益、年化收益、夏普比率、最大回撤、Calmar比率、日胜率
   - 同时保存基准指标和回测区间信息
   - 输出路径: results/report_metrics.json

2. visualization/report_generator/generate_report.py
   - 加载策略KPI JSON文件(优先)
   - 直接使用轮动策略输出的指标,不再重复计算
   - 保留备用计算逻辑(JSON不存在时)
   - 确保HTML报告与轮动策略结果完全一致

效果:
- KPI指标统一从轮动策略回测结果获取
- 避免重复计算导致的数据不一致
- 数据来源清晰可追溯
- HTML报告指标与终端输出完全一致
This commit is contained in:
2026-05-08 22:25:22 +08:00
parent 5c87bea4fc
commit 861e590441
2 changed files with 87 additions and 1 deletions

View File

@@ -133,6 +133,37 @@ def generate_performance_report(
etf_nav_data_raw=etf_nav_data_raw, etf_nav_data_raw=etf_nav_data_raw,
) )
# 保存整体策略KPI到JSON文件
import json
metrics_dict = {
"策略": {
"累计收益": float(s_total_return),
"年化收益(自然日)": float(s_cagr_nat),
"年化收益(交易日)": float(s_cagr_trd),
"夏普比率": float(s_sharpe),
"最大回撤": float(s_max_dd),
"Calmar比率": float(s_calmar),
"日胜率": float(s_win_rate),
"回测区间": {
"开始": strategy_nav.index.min().strftime("%Y-%m-%d"),
"结束": strategy_nav.index.max().strftime("%Y-%m-%d"),
"交易天数": len(strategy_nav)
}
},
"基准": {
"累计收益": float(b_total_return),
"年化收益(自然日)": float(b_cagr_nat),
"夏普比率": float(b_sharpe),
"最大回撤": float(b_max_dd),
"名称": benchmark_name
}
}
metrics_path = f"{save_path}_metrics.json"
with open(metrics_path, 'w', encoding='utf-8') as f:
json.dump(metrics_dict, f, indent=2, ensure_ascii=False)
print(f"策略指标已保存: {metrics_path}")
# 返回指标字典 # 返回指标字典
return { return {
"累计收益": s_total_return, "累计收益": s_total_return,

View File

@@ -15,6 +15,7 @@ from datetime import datetime
import argparse import argparse
import os import os
import sys import sys
import json
class ReportGenerator: class ReportGenerator:
@@ -24,6 +25,7 @@ class ReportGenerator:
self.results_dir = results_dir self.results_dir = results_dir
self.summary_df = None self.summary_df = None
self.trades_df = None self.trades_df = None
self.metrics = None # 从JSON加载的策略KPI
def load_data(self): def load_data(self):
"""加载回测数据""" """加载回测数据"""
@@ -48,10 +50,63 @@ class ReportGenerator:
self.trades_df['进场日期'] = pd.to_datetime(self.trades_df['进场日期']) self.trades_df['进场日期'] = pd.to_datetime(self.trades_df['进场日期'])
self.trades_df['出场日期'] = pd.to_datetime(self.trades_df['出场日期']) self.trades_df['出场日期'] = pd.to_datetime(self.trades_df['出场日期'])
# 加载策略KPI JSON文件如果存在
metrics_path = os.path.join(self.results_dir, 'report_metrics.json')
if os.path.exists(metrics_path):
with open(metrics_path, 'r', encoding='utf-8') as f:
self.metrics = json.load(f)
print(f"✅ 加载策略指标: {metrics_path}")
else:
print(f"⚠️ 未找到策略指标文件: {metrics_path}")
print(f"✅ 数据加载成功: {len(self.trades_df)} 条交易记录") print(f"✅ 数据加载成功: {len(self.trades_df)} 条交易记录")
def calculate_kpis(self, trades_filtered=None): def calculate_kpis(self, trades_filtered=None):
"""计算关键指标""" """计算关键指标 - 优先使用轮动策略输出的指标"""
# 如果有从JSON加载的策略KPI直接使用
if self.metrics is not None and '策略' in self.metrics:
strategy_metrics = self.metrics['策略']
print("✅ 使用轮动策略输出的KPI指标")
# 计算调仓次数需要从trades数据获取
df = trades_filtered if trades_filtered is not None else self.trades_df
total_trades = len(df)
# 最佳品种从summary获取
if self.summary_df is not None:
symbol_col = self.summary_df['累计收益']
best_symbol = self.summary_df.loc[symbol_col.idxmax(), '品种代码']
else:
best_symbol = 'N/A'
# 平均持仓天数
avg_holding_days = df['持仓天数'].mean() if len(df) > 0 else 0
# 盈亏次数基于trades数据
# 转换持仓收益为数值
if df['持仓收益'].dtype == 'object':
returns_num = df['持仓收益'].str.rstrip('%').astype(float)
else:
returns_num = df['持仓收益']
win_count = (returns_num > 0).sum()
loss_count = len(df) - win_count
return {
'total_return': f"{strategy_metrics['累计收益'] * 100:.2f}",
'annual_return': f"{strategy_metrics['年化收益(自然日)'] * 100:.2f}",
'win_rate': f"{strategy_metrics['日胜率'] * 100:.2f}",
'max_drawdown': f"{strategy_metrics['最大回撤'] * 100:.2f}",
'sharpe_ratio': f"{strategy_metrics['夏普比率']:.2f}",
'total_trades': str(total_trades),
'best_symbol': best_symbol,
'avg_holding_days': f"{avg_holding_days:.1f}",
'win_count': int(win_count),
'loss_count': int(loss_count)
}
# 否则重新计算(备用方案)
print("⚠️ 未找到策略KPI重新计算...")
df = trades_filtered if trades_filtered is not None else self.trades_df df = trades_filtered if trades_filtered is not None else self.trades_df
# 使用净值计算真实收益 # 使用净值计算真实收益