aszerW
38a31357d1
新增 scripts/generate_legacy_report.py： - 使用新框架运行回测 - 将数据格式转换为原引擎格式 - 调用原引擎 generate_performance_report 生成报告 输出文件： - rotation_legacy_chart.png (净值曲线+回撤+持仓分布) - rotation_legacy_metrics.json (策略指标JSON) - rotation_legacy_nav.csv (净值曲线数据) 用法：python scripts/generate_legacy_report.py
164 lines
5.6 KiB
Python
164 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
使用新框架数据生成原引擎格式的报告
|
||
|
||
用法:
|
||
python scripts/generate_legacy_report.py
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import yaml
|
||
import pandas as pd
|
||
import numpy as np
|
||
from pathlib import Path
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
# 添加项目根目录到 sys.path
|
||
project_root = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
# 导入新框架
|
||
from strategies.rotation.strategy import RotationStrategy
|
||
|
||
# 导入原引擎报告生成模块
|
||
archive_path = project_root / 'archive' / 'legacy_core'
|
||
sys.path.insert(0, str(archive_path))
|
||
from report import generate_performance_report
|
||
from core.common.utils import calculate_cagr, calculate_max_drawdown, calculate_sharpe
|
||
|
||
|
||
def run_with_legacy_report():
|
||
"""运行新框架回测并生成原引擎格式报告"""
|
||
|
||
# 加载配置
|
||
config_path = 'config/strategies/rotation.yaml'
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
config = yaml.safe_load(f)
|
||
|
||
# 新框架回测
|
||
print("=" * 60)
|
||
print(" ETF轮动策略 回测系统 (新框架)")
|
||
print("=" * 60)
|
||
|
||
strategy = RotationStrategy.from_yaml(config_path)
|
||
data = strategy.get_data()
|
||
|
||
# 计算因子
|
||
print("\n计算因子...")
|
||
factor_df = strategy.compute_factors(data)
|
||
|
||
# 生成信号
|
||
print("\n生成信号...")
|
||
signals = strategy.generate_signals(factor_df)
|
||
|
||
# 执行回测
|
||
print("\n执行回测...")
|
||
result = strategy.run_backtest(data=data)
|
||
|
||
# 准备原引擎格式的数据
|
||
backtest_result = result['result'].copy()
|
||
|
||
if backtest_result is None:
|
||
print("回测失败，无法生成报告")
|
||
return
|
||
|
||
# 重命名列以匹配原引擎格式
|
||
backtest_result['轮动策略净值'] = backtest_result['策略净值']
|
||
backtest_result['轮动策略日收益率'] = backtest_result['策略日收益率']
|
||
|
||
# 1. 基准净值和基准日收益率
|
||
benchmark_data = data.get('benchmark_data')
|
||
if benchmark_data is not None and not benchmark_data.empty:
|
||
# 对齐基准数据到回测日期
|
||
benchmark_close = benchmark_data['close'] if 'close' in benchmark_data.columns else benchmark_data.iloc[:, 0]
|
||
benchmark_close_aligned = benchmark_close.reindex(backtest_result.index, method='ffill')
|
||
|
||
# 计算基准净值
|
||
benchmark_nav = (1 + benchmark_close_aligned.pct_change()).cumprod()
|
||
benchmark_nav = benchmark_nav / benchmark_nav.dropna().iloc[0] # 归一化起点为1
|
||
|
||
backtest_result['基准净值'] = benchmark_nav.values
|
||
backtest_result['基准日收益率'] = benchmark_close_aligned.pct_change().values
|
||
|
||
# 2. 各标的净值（指数价格）
|
||
index_close = data.get('index_close')
|
||
valid_codes = data['valid_codes']
|
||
|
||
for code in valid_codes:
|
||
if index_close is not None and code in index_close.columns:
|
||
# 计算该标的的净值曲线
|
||
price_series = index_close[code].loc[backtest_result.index]
|
||
nav_series = (1 + price_series.pct_change()).cumprod()
|
||
nav_series = nav_series / nav_series.iloc[0] if nav_series.iloc[0] > 0 else nav_series
|
||
backtest_result[f'净值_{code}'] = nav_series.values
|
||
backtest_result[code] = price_series.values # 当前价格
|
||
|
||
# 3. 得分列（从factor_df获取）
|
||
for code in valid_codes:
|
||
if code in factor_df.columns:
|
||
scores_aligned = factor_df[code].reindex(backtest_result.index, method='ffill')
|
||
backtest_result[f'得分_{code}'] = scores_aligned.values
|
||
|
||
# 4. 信号列（中文名）
|
||
backtest_result['信号'] = backtest_result['signal']
|
||
|
||
# 构建code_name_map和code_config
|
||
code_config = config.get('code_list', {})
|
||
code_name_map = {code: cfg.get('name', code) for code, cfg in code_config.items()}
|
||
|
||
# 准备ETF价格和净值数据（用于溢价率计算）
|
||
etf_data = data.get('etf_data')
|
||
etf_nav_data = data.get('etf_nav_data')
|
||
|
||
# ETF数据需要用ETF代码作为列名
|
||
etf_price_data = None
|
||
etf_nav_data_raw = None
|
||
|
||
if etf_data is not None:
|
||
# 转换列名：指数代码 -> ETF代码（通过etf_code_map）
|
||
etf_code_map = data.get('etf_code_map', {})
|
||
etf_price_data = pd.DataFrame(index=etf_data.index)
|
||
for idx_code, etf_code in etf_code_map.items():
|
||
if etf_code in etf_data.columns:
|
||
etf_price_data[idx_code] = etf_data[etf_code]
|
||
|
||
if etf_nav_data is not None:
|
||
etf_nav_data_raw = pd.DataFrame(index=etf_nav_data.index)
|
||
for idx_code, etf_code in etf_code_map.items():
|
||
if etf_code in etf_nav_data.columns:
|
||
etf_nav_data_raw[idx_code] = etf_nav_data[etf_code]
|
||
|
||
# 生成原引擎格式的报告
|
||
print("\n" + "=" * 60)
|
||
print(" 生成原引擎格式报告")
|
||
print("=" * 60)
|
||
|
||
save_path = 'results/rotation_legacy'
|
||
os.makedirs('results', exist_ok=True)
|
||
|
||
metrics = generate_performance_report(
|
||
backtest_result=backtest_result,
|
||
code_list=valid_codes,
|
||
code_name_map=code_name_map,
|
||
benchmark_name=config.get('benchmark_name', '沪深300指数'),
|
||
save_path=save_path,
|
||
select_num=config.get('select_num', 3),
|
||
code_config=code_config,
|
||
index_data=index_close,
|
||
etf_price_data=etf_price_data,
|
||
etf_nav_data_raw=etf_nav_data_raw,
|
||
)
|
||
|
||
print(f"\n报告文件已生成:")
|
||
print(f" - {save_path}_chart.png")
|
||
print(f" - {save_path}_metrics.json")
|
||
print(f" - {save_path}_nav.csv")
|
||
|
||
return metrics
|
||
|
||
|
||
if __name__ == '__main__':
|
||
run_with_legacy_report() |