""" ETF轮动策略报告生成器 ======================= 从回测数据生成精美的 HTML 策略报告 使用方法: python generate_report.py python generate_report.py --start 2024-01-01 --end 2024-12-31 """ import pandas as pd import numpy as np from jinja2 import Template from datetime import datetime import argparse import os import sys class ReportGenerator: """策略报告生成器""" def __init__(self, results_dir='results'): self.results_dir = results_dir self.summary_df = None self.trades_df = None def load_data(self): """加载回测数据""" # 加载汇总数据 summary_path = os.path.join(self.results_dir, 'report_summary.csv') if not os.path.exists(summary_path): raise FileNotFoundError(f"找不到汇总数据文件: {summary_path}") self.summary_df = pd.read_csv(summary_path) # 转换百分比 for col in ['胜率', '平均收益', '累计收益', '最大单次收益', '最大单次亏损']: if col in self.summary_df.columns: self.summary_df[col] = self.summary_df[col].str.rstrip('%').astype(float) # 加载交易记录 trades_path = os.path.join(self.results_dir, 'report_trades.csv') if not os.path.exists(trades_path): raise FileNotFoundError(f"找不到交易记录文件: {trades_path}") self.trades_df = pd.read_csv(trades_path) self.trades_df['进场日期'] = pd.to_datetime(self.trades_df['进场日期']) self.trades_df['出场日期'] = pd.to_datetime(self.trades_df['出场日期']) print(f"✅ 数据加载成功: {len(self.trades_df)} 条交易记录") def calculate_kpis(self, trades_filtered=None): """计算关键指标""" df = trades_filtered if trades_filtered is not None else self.trades_df # 转换持仓收益为数值（去除百分号） if df['持仓收益'].dtype == 'object': df = df.copy() df['持仓收益_num'] = df['持仓收益'].str.rstrip('%').astype(float) else: df = df.copy() df['持仓收益_num'] = df['持仓收益'] # 总收益 total_return = df['持仓收益_num'].sum() # 年化收益 days = (df['出场日期'].max() - df['出场日期'].min()).days if days > 0: annual_return = total_return / (days / 365.0) else: annual_return = 0 # 胜率 win_count = (df['持仓收益_num'] > 0).sum() total_count = len(df) win_rate = (win_count / total_count * 100) if total_count > 0 else 0 loss_count = total_count - win_count # 夏普比率 daily_returns = df.groupby('出场日期')['持仓收益_num'].sum() if daily_returns.std() > 0: sharpe_ratio = daily_returns.mean() / daily_returns.std() * np.sqrt(252) else: sharpe_ratio = 0 # 最大回撤（简化计算） cum_returns = df['持仓收益_num'].cumsum() running_max = cum_returns.cummax() drawdown = (cum_returns - running_max) max_drawdown = drawdown.min() if len(drawdown) > 0 else 0 # 调仓次数 total_trades = len(df) # 最佳品种 symbol_returns = df.groupby('品种代码')['持仓收益_num'].sum() best_symbol = symbol_returns.idxmax() if len(symbol_returns) > 0 else 'N/A' # 平均持仓天数 avg_holding_days = df['持仓天数'].mean() if len(df) > 0 else 0 return { 'total_return': f"{total_return:.2f}", 'annual_return': f"{annual_return:.2f}", 'win_rate': f"{win_rate:.2f}", 'max_drawdown': f"{max_drawdown:.2f}", 'sharpe_ratio': f"{sharpe_ratio:.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) } def prepare_chart_data(self, trades_filtered=None): """准备图表数据""" df = trades_filtered if trades_filtered is not None else self.trades_df # 转换持仓收益为数值 if df['持仓收益'].dtype == 'object': df = df.copy() df['持仓收益_num'] = df['持仓收益'].str.rstrip('%').astype(float) else: df = df.copy() df['持仓收益_num'] = df['持仓收益'] df_sorted = df.sort_values('出场日期') # 净值曲线数据 df_sorted['累计收益'] = df_sorted['持仓收益_num'].cumsum() nav_values = df_sorted['累计收益'].tolist() nav_dates = df_sorted['出场日期'].dt.strftime('%Y-%m-%d').tolist() # 月度收益数据 df_copy = df_sorted.copy() df_copy['年月'] = df_copy['出场日期'].dt.to_period('M') monthly = df_copy.groupby('年月')['持仓收益_num'].sum().reset_index() monthly['年月_str'] = monthly['年月'].astype(str) monthly_dates = monthly['年月_str'].tolist() monthly_values = monthly['持仓收益_num'].round(2).tolist() # 品种收益排行 symbol_returns = df.groupby('品种代码')['持仓收益_num'].sum().sort_values() symbol_names = [] symbol_returns_list = [] for code, ret in symbol_returns.items(): name = self.summary_df[self.summary_df['品种代码'] == code] if len(name) > 0: symbol_names.append(name.iloc[0]['品种名称']) else: symbol_names.append(code) symbol_returns_list.append(round(ret, 2)) # 唯一品种列表 symbols = df['品种代码'].unique().tolist() return { 'nav_dates': nav_dates, 'nav_values': [round(v, 2) for v in nav_values], 'monthly_dates': monthly_dates, 'monthly_values': monthly_values, 'symbol_names': symbol_names, 'symbol_returns': symbol_returns_list, 'symbols': symbols } def generate(self, start_date=None, end_date=None, output_dir='reports'): """生成报告""" print("🚀 开始生成策略报告...") # 加载数据 self.load_data() # 筛选数据 if start_date: start_date = pd.to_datetime(start_date) if end_date: end_date = pd.to_datetime(end_date) trades_filtered = self.trades_df.copy() if start_date: trades_filtered = trades_filtered[trades_filtered['出场日期'] >= start_date] if end_date: trades_filtered = trades_filtered[trades_filtered['出场日期'] <= end_date] print(f"📊 筛选后数据: {len(trades_filtered)} 条记录") # 计算指标 kpis = self.calculate_kpis(trades_filtered) chart_data = self.prepare_chart_data(trades_filtered) # 准备交易记录 trades_display = trades_filtered.copy() trades_display['进场日期'] = trades_display['进场日期'].dt.strftime('%Y-%m-%d') trades_display['出场日期'] = trades_display['出场日期'].dt.strftime('%Y-%m-%d') trades_list = trades_display.to_dict('records') # 读取模板 template_path = os.path.join(os.path.dirname(__file__), 'template.html') with open(template_path, 'r', encoding='utf-8') as f: template = Template(f.read()) # 渲染模板 html = template.render( report_date=datetime.now().strftime('%Y-%m-%d %H:%M:%S'), start_date=start_date.strftime('%Y-%m-%d') if start_date else trades_filtered['出场日期'].min().strftime('%Y-%m-%d'), end_date=end_date.strftime('%Y-%m-%d') if end_date else trades_filtered['出场日期'].max().strftime('%Y-%m-%d'), trades=trades_list, **kpis, **chart_data ) # 创建输出目录 os.makedirs(output_dir, exist_ok=True) # 保存报告 output_file = os.path.join( output_dir, f'strategy_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.html' ) with open(output_file, 'w', encoding='utf-8') as f: f.write(html) print(f"✅ 报告已生成: {output_file}") print(f"📁 文件大小: {os.path.getsize(output_file) / 1024:.1f} KB") print(f"🌐 在浏览器中打开: file://{os.path.abspath(output_file)}") return output_file def main(): """主函数""" parser = argparse.ArgumentParser(description='生成ETF轮动策略报告') parser.add_argument('--start', type=str, help='开始日期 (YYYY-MM-DD)') parser.add_argument('--end', type=str, help='结束日期 (YYYY-MM-DD)') parser.add_argument('--output', type=str, default='reports', help='输出目录') args = parser.parse_args() try: generator = ReportGenerator() generator.generate( start_date=args.start, end_date=args.end, output_dir=args.output ) except Exception as e: print(f"❌ 生成失败: {e}") import traceback traceback.print_exc() sys.exit(1) if __name__ == '__main__': main()