""" ETF轮动策略 - 绩效报告模块 """ import numpy as np import pandas as pd import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from typing import Optional from core.common.utils import calculate_cagr, calculate_max_drawdown, calculate_sharpe def generate_performance_report( backtest_result: pd.DataFrame, code_list: list, code_name_map: dict = None, benchmark_name: str = "沪深300指数", save_path: str = "report", select_num: int = 1, ) -> dict: """ 生成完整的绩效报告 Args: backtest_result: 回测结果 code_list: ETF代码列表 code_name_map: 代码到名称映射 benchmark_name: 基准名称 save_path: 报告保存路径前缀 select_num: 选中数量 Returns: dict: 绩效指标字典 """ import os os.makedirs(os.path.dirname(save_path) if os.path.dirname(save_path) else ".", exist_ok=True) code_name_map = code_name_map or {} strategy_nav = backtest_result["轮动策略净值"] strategy_ret = backtest_result["轮动策略日收益率"] benchmark_nav = backtest_result["基准净值"] benchmark_ret = backtest_result["基准日收益率"] # 计算绩效指标 s_cagr_nat = calculate_cagr(strategy_nav, "natural_days") s_cagr_trd = calculate_cagr(strategy_nav, "trading_days") s_total_return = strategy_nav.iloc[-1] - 1 s_sharpe = calculate_sharpe(strategy_ret) s_max_dd, s_dd_start, s_dd_end = calculate_max_drawdown(strategy_nav) s_win_rate = (strategy_ret > 0).sum() / len(strategy_ret) s_calmar = s_cagr_nat / abs(s_max_dd) if s_max_dd != 0 else np.inf b_cagr_nat = calculate_cagr(benchmark_nav, "natural_days") b_cagr_trd = calculate_cagr(benchmark_nav, "trading_days") b_total_return = benchmark_nav.iloc[-1] - 1 b_sharpe = calculate_sharpe(benchmark_ret) b_max_dd, _, _ = calculate_max_drawdown(benchmark_nav) # 打印绩效表格 print("\n" + "=" * 70) print(" 绩效评估报告") print("=" * 70) print(f" 回测区间: {strategy_nav.index.min().date()} ~ {strategy_nav.index.max().date()}") print(f" 交易天数: {len(strategy_nav)}") print("-" * 70) print(f' {"指标":<25} {"轮动策略":>15} {"基准(" + benchmark_name + ")":>18}') print("-" * 70) print(f' {"累计收益":<25} {s_total_return:>14.2%} {b_total_return:>17.2%}') print(f' {"CAGR(自然日口径)":<25} {s_cagr_nat:>14.2%} {b_cagr_nat:>17.2%}') print(f' {"CAGR(交易日口径)":<25} {s_cagr_trd:>14.2%} {b_cagr_trd:>17.2%}') print(f' {"年化夏普比率":<25} {s_sharpe:>14.2f} {b_sharpe:>17.2f}') print(f' {"最大回撤":<25} {s_max_dd:>14.2%} {b_max_dd:>17.2%}') print(f' {"Calmar比率":<23} {s_calmar:>14.2f} {"--":>17}') print(f' {"日胜率":<25} {s_win_rate:>14.2%} {"--":>17}') print(f' {"最大回撤区间":<22} {str(s_dd_start.date()):>10} ~ {str(s_dd_end.date())}') print("=" * 70) # 打印最新调仓信号 _print_latest_signal(backtest_result, code_list, code_name_map, select_num) # 绘制图表 _plot_report_chart( backtest_result, code_list, code_name_map, benchmark_name, save_path, select_num, metrics={ "累计收益": s_total_return, "年化收益": s_cagr_nat, "夏普比率": s_sharpe, "最大回撤": s_max_dd, "Calmar比率": s_calmar, "日胜率": s_win_rate, } ) # 返回指标字典 return { "累计收益": s_total_return, "CAGR_自然日": s_cagr_nat, "CAGR_交易日": s_cagr_trd, "夏普比率": s_sharpe, "最大回撤": s_max_dd, "Calmar比率": s_calmar, "日胜率": s_win_rate, "基准累计收益": b_total_return, "基准CAGR_自然日": b_cagr_nat, "基准夏普比率": b_sharpe, "基准最大回撤": b_max_dd, } def _print_latest_signal(backtest_result: pd.DataFrame, code_list: list, code_name_map: dict, select_num: int): """打印最新调仓信号""" latest = _extract_latest_positions(backtest_result, code_list, code_name_map, select_num) signal_date_str = latest["signal_date"].strftime("%Y-%m-%d") print("\n") print("=" * 100) print(" 最新调仓信号 (下一交易日执行)") print("=" * 100) print(f" 数据截止: {signal_date_str}") print() # 表头 print(f' {"品种名称":<8} {"代码":>10} {"仓位":>6} {"得分":>8} {"进场日期":>12} {"进场价":>10} {"最新价":>10} {"操作":>6} {"持有天数":>8} {"盈亏":>10}') print(" " + "-" * 115) # 下期持仓（调入/维持） for pos in latest["positions"]: pnl_str = f'{pos["pnl"]:>+9.2%}' if pos["pnl"] is not None else ' —' days_str = f'{pos["holding_days"]:>7}天' if pos["holding_days"] is not None else ' —' entry_str = f'{pos["entry_price"]:>10.2f}' if pos["entry_price"] is not None else ' —' entry_date_str = pos["entry_date"].strftime("%Y-%m-%d") if pos.get("entry_date") else ' —' score_str = f'{pos["score"]:>8.2f}' if pos["score"] is not None else ' —' flag = '▲' if pos["action"] == "调入" else ' ' print(f' {pos["name"]:<8} {pos["code"]:>10} {pos["weight"]:>6.0%} {score_str} {entry_date_str:>12} {entry_str} {pos["current_price"]:>10.2f} {flag}{pos["action"]:>4} {days_str} {pnl_str}') # 需调出的品种 if latest["exit_positions"]: print() print(" 需调出:") for pos in latest["exit_positions"]: pnl_str = f'{pos["pnl"]:>+9.2%}' if pos["pnl"] is not None else ' —' days_str = f'{pos["holding_days"]:>7}天' if pos["holding_days"] is not None else ' —' entry_str = f'{pos["entry_price"]:>10.2f}' if pos["entry_price"] is not None else ' —' entry_date_str = pos["entry_date"].strftime("%Y-%m-%d") if pos.get("entry_date") else ' —' score_str = ' —' # 调出品种无得分 print(f' {pos["name"]:<8} {pos["code"]:>10} {pos["weight"]:>6.0%} {score_str} {entry_date_str:>12} {entry_str} {pos["current_price"]:>10.2f} ▼调出 {days_str} {pnl_str}') print("=" * 120) def _extract_latest_positions(backtest_result: pd.DataFrame, code_list: list, code_name_map: dict, select_num: int) -> dict: """提取最新持仓和下期调仓建议""" last_date = backtest_result.index[-1] last_row = backtest_result.iloc[-1] # 当前持仓 current_signal = last_row["信号"] if select_num == 1: current_codes = [current_signal] else: current_codes = current_signal.split(",") # 下期建议 score_cols = [f"得分_{code}" for code in code_list if f"得分_{code}" in backtest_result.columns] scores = pd.to_numeric(last_row[score_cols], errors="coerce") top_n = scores.nlargest(select_num) next_codes = [c.replace("得分_", "") for c in top_n.index] # 计算持仓信息 positions_info = [] weight = 1.0 / select_num for code in next_codes: name = code_name_map.get(code, code) action = "维持" if code in current_codes else "调入" # 获取当前价格和得分 current_price = last_row.get(code, 0) score = scores.get(f"得分_{code}", None) # 计算持仓信息（如果是维持的仓位） entry_date = None entry_price = None holding_days = None pnl = None if action == "维持": # 找到该标的最近一次连续持仓的起始日期 signal_series = backtest_result["信号"] mask = signal_series == code if select_num == 1 else signal_series.str.contains(code, regex=False, na=False) # 找到连续持仓段（从后往前找） is_holding = mask.values dates = backtest_result.index # 从最后一天往前遍历，找到连续持仓的起始点 entry_date = None for i in range(len(is_holding) - 1, -1, -1): if is_holding[i]: entry_date = dates[i] else: break if entry_date is not None: entry_price = backtest_result.loc[entry_date, code] holding_days = (last_date - entry_date).days if entry_price and entry_price != 0: pnl = current_price / entry_price - 1 positions_info.append({ "code": code, "name": name, "weight": weight, "score": score, "action": action, "current_price": current_price, "entry_date": entry_date, "entry_price": entry_price, "holding_days": holding_days, "pnl": pnl, }) # 需调出的品种信息 exit_positions = [] for code in current_codes: if code not in next_codes: name = code_name_map.get(code, code) current_price = last_row.get(code, 0) # 计算调出品种的持仓信息（最近一次连续持仓） signal_series = backtest_result["信号"] mask = signal_series == code if select_num == 1 else signal_series.str.contains(code, regex=False, na=False) # 找到连续持仓段（从后往前找） is_holding = mask.values dates = backtest_result.index entry_price = None holding_days = None pnl = None # 从最后一天往前遍历，找到连续持仓的起始点 entry_date = None for i in range(len(is_holding) - 1, -1, -1): if is_holding[i]: entry_date = dates[i] else: break if entry_date is not None: entry_price = backtest_result.loc[entry_date, code] holding_days = (last_date - entry_date).days if entry_price and entry_price != 0: pnl = current_price / entry_price - 1 exit_positions.append({ "code": code, "name": name, "weight": weight, "score": None, # 调出品种无得分 "action": "调出", "current_price": current_price, "entry_date": entry_date, "entry_price": entry_price, "holding_days": holding_days, "pnl": pnl, }) return { "signal_date": last_date, "current_codes": current_codes, "next_codes": next_codes, "positions": positions_info, "exit_positions": exit_positions, } def _plot_report_chart( backtest_result: pd.DataFrame, code_list: list, code_name_map: dict, benchmark_name: str, save_path: str, select_num: int, metrics: dict = None, ): """绘制报告图表""" # 设置中文字体（优先使用基础镜像中已存在的字体） plt.rcParams["font.sans-serif"] = [ "WenQuanYi Zen Hei", # 基础镜像已安装 "WenQuanYi Micro Hei", # 将要安装 "DejaVu Sans", "SimHei", "Arial Unicode MS" ] plt.rcParams["axes.unicode_minus"] = False strategy_nav = backtest_result["轮动策略净值"] benchmark_nav = backtest_result["基准净值"] strategy_ret = backtest_result["轮动策略日收益率"] # 计算绩效指标（如果没有传入） if metrics is None: from core.common.utils import calculate_cagr, calculate_max_drawdown, calculate_sharpe s_cagr_nat = calculate_cagr(strategy_nav, "natural_days") s_total_return = strategy_nav.iloc[-1] - 1 s_sharpe = calculate_sharpe(strategy_ret) s_max_dd, s_dd_start, s_dd_end = calculate_max_drawdown(strategy_nav) s_win_rate = (strategy_ret > 0).sum() / len(strategy_ret) s_calmar = s_cagr_nat / abs(s_max_dd) if s_max_dd != 0 else np.inf metrics = { "累计收益": s_total_return, "年化收益": s_cagr_nat, "夏普比率": s_sharpe, "最大回撤": s_max_dd, "Calmar比率": s_calmar, "日胜率": s_win_rate, } # 提取最新调仓信息 latest = _extract_latest_positions(backtest_result, code_list, code_name_map, select_num) # 计算表格行数 n_table_rows = len(latest["positions"]) + len(latest["exit_positions"]) signal_table_height = max(2.0, 0.6 + n_table_rows * 0.35) metrics_table_height = 1.2 fig = plt.figure(figsize=(14, 10 + signal_table_height + metrics_table_height + 8)) gs = fig.add_gridspec(5, 1, height_ratios=[signal_table_height, metrics_table_height, 3, 1, 1.2], hspace=0.35) # 面板0: 最新调仓信号表 ax0 = fig.add_subplot(gs[0]) ax0.axis("off") signal_date_str = latest["signal_date"].strftime("%Y-%m-%d") ax0.set_title(f"最新调仓信号 (数据截止: {signal_date_str}，下一交易日执行)", fontsize=14, fontweight="bold", loc="left", pad=15) # 构建表格数据 table_data = [] col_labels = ["品种名称", "代码", "仓位", "得分", "进场日期", "进场价", "最新价", "操作", "持有天数", "盈亏"] # 下期持仓（调入/维持） for pos in latest["positions"]: pnl_str = f'{pos["pnl"]:+.2%}' if pos["pnl"] is not None else "—" days_str = f'{pos["holding_days"]}' if pos["holding_days"] is not None else "—" entry_str = f'{pos["entry_price"]:.2f}' if pos["entry_price"] is not None else "—" entry_date_str = pos["entry_date"].strftime("%m-%d") if pos.get("entry_date") else "—" score_str = f'{pos["score"]:.2f}' if pos["score"] is not None else "—" table_data.append([ pos["name"], pos["code"], f'{pos["weight"]:.0%}', score_str, entry_date_str, entry_str, f'{pos["current_price"]:.2f}', pos["action"], days_str, pnl_str ]) # 需调出的品种 for pos in latest["exit_positions"]: pnl_str = f'{pos["pnl"]:+.2%}' if pos["pnl"] is not None else "—" days_str = f'{pos["holding_days"]}' if pos["holding_days"] is not None else "—" entry_str = f'{pos["entry_price"]:.2f}' if pos["entry_price"] is not None else "—" entry_date_str = pos["entry_date"].strftime("%m-%d") if pos.get("entry_date") else "—" score_str = "—" # 调出品种无得分 table_data.append([ pos["name"], pos["code"], f'{pos["weight"]:.0%}', score_str, entry_date_str, entry_str, f'{pos["current_price"]:.2f}', "调出", days_str, pnl_str ]) if table_data: table = ax0.table( cellText=table_data, colLabels=col_labels, loc="center", cellLoc="center", colWidths=[0.10, 0.10, 0.07, 0.08, 0.08, 0.08, 0.08, 0.07, 0.08, 0.08], bbox=[0, 0, 1, 1], # 使用完整宽度 ) table.auto_set_font_size(False) table.set_fontsize(10) table.scale(1, 2.0) # 行高与绩效表格一致 # 表头深色 for j in range(len(col_labels)): table[0, j].set_facecolor("#2C3E50") table[0, j].set_text_props(color="white", fontweight="bold") # 数据行按操作着色 for i in range(len(table_data)): action = table_data[i][7] # 操作列在第8列 if action == "调入": color = "#d4edda" # 绿色 elif action == "调出": color = "#f8d7da" # 红色 else: color = "#fff3cd" # 黄色 for j in range(len(col_labels)): table[i + 1, j].set_facecolor(color) # 面板1: 策略绩效指标对比表（转置：行为策略/基准，列为指标） ax1 = fig.add_subplot(gs[1]) ax1.axis("off") ax1.set_title("策略绩效对比", fontsize=14, fontweight="bold", loc="left", pad=10) # 计算基准指标 from core.common.utils import calculate_cagr, calculate_max_drawdown, calculate_sharpe benchmark_ret = backtest_result["基准日收益率"] b_cagr_nat = calculate_cagr(benchmark_nav, "natural_days") b_total_return = benchmark_nav.iloc[-1] - 1 b_sharpe = calculate_sharpe(benchmark_ret) b_max_dd, _, _ = calculate_max_drawdown(benchmark_nav) # 构建绩效对比表格（转置） start_date = strategy_nav.index.min().strftime("%Y-%m-%d") end_date = strategy_nav.index.max().strftime("%Y-%m-%d") # 列标题（指标），第一列添加"策略" perf_col_labels = ["策略", "开始时间", "结束时间", "累计收益", "年化收益", "最大回撤", "夏普比率", "Calmar比率", "日胜率"] # 策略行数据（包含行标题） strategy_row = [ "轮动策略", start_date, end_date, f"{metrics.get('累计收益', 0):.2%}", f"{metrics.get('年化收益', 0):.2%}", f"{metrics.get('最大回撤', 0):.2%}", f"{metrics.get('夏普比率', 0):.2f}", f"{metrics.get('Calmar比率', 0):.2f}", f"{metrics.get('日胜率', 0):.2%}", ] # 基准行数据（包含行标题） benchmark_row = [ f"基准({benchmark_name})", start_date, end_date, f"{b_total_return:.2%}", f"{b_cagr_nat:.2%}", f"{b_max_dd:.2%}", f"{b_sharpe:.2f}", "—", "—", ] # 表格数据：2行（策略、基准） perf_table_data = [strategy_row, benchmark_row] # 使用与调仓表格相同的列宽计算方式，确保总宽度一致 # 调仓表格有10列，这里也有9列，使用相似的宽度分配 perf_col_widths = [0.10, 0.10, 0.10, 0.10, 0.10, 0.10, 0.10, 0.10, 0.10] perf_table = ax1.table( cellText=perf_table_data, colLabels=perf_col_labels, loc="center", cellLoc="center", colWidths=perf_col_widths, bbox=[0, 0, 1, 1], # 使用完整宽度，与调仓表格一致 ) perf_table.auto_set_font_size(False) perf_table.set_fontsize(10) # 字体大小与调仓表格一致 perf_table.scale(1, 2.0) # 行高与调仓表格一致 # 表头样式（第一行） for j in range(len(perf_col_labels)): perf_table[0, j].set_facecolor("#2C3E50") perf_table[0, j].set_text_props(color="white", fontweight="bold") # 数据行样式 # 策略行浅绿背景 for j in range(len(perf_col_labels)): perf_table[1, j].set_facecolor("#d4edda") # 基准行浅蓝背景 for j in range(len(perf_col_labels)): perf_table[2, j].set_facecolor("#cce5ff") # 第一列（策略名称）加粗 for i in range(2): perf_table[i + 1, 0].set_text_props(fontweight="bold") # 面板2: 净值曲线 ax2 = fig.add_subplot(gs[2]) ax2.plot(strategy_nav.index, strategy_nav.values, label="轮动策略", linewidth=2, color="#E74C3C") ax2.plot(benchmark_nav.index, benchmark_nav.values, label=benchmark_name, linewidth=1.5, color="#3498DB", alpha=0.8) chart_colors = plt.cm.tab20.colors show_legend_n = min(len(code_list), 10) for i, code in enumerate(code_list): name = code_name_map.get(code, code) lbl = name if i < show_legend_n else None ax2.plot(backtest_result.index, backtest_result[f"净值_{code}"].values, label=lbl, linewidth=0.8, alpha=0.4, color=chart_colors[i % len(chart_colors)]) ax2.set_title("ETF轮动策略 - 净值曲线", fontsize=16, fontweight="bold") ax2.set_ylabel("净值") ax2.legend(loc="upper left", fontsize=8, ncol=2) ax2.grid(True, alpha=0.3) ax2.set_yscale("log") # 面板3: 回撤曲线 ax3 = fig.add_subplot(gs[3]) cummax = strategy_nav.cummax() drawdown = (strategy_nav - cummax) / cummax ax3.fill_between(drawdown.index, drawdown.values, 0, alpha=0.5, color="#E74C3C") ax3.set_title("策略回撤", fontsize=12) ax3.set_ylabel("回撤") ax3.grid(True, alpha=0.3) # 面板4: 持仓分布 ax4 = fig.add_subplot(gs[4]) signal_series = backtest_result["信号"] for i, code in enumerate(code_list): name = code_name_map.get(code, code) if select_num > 1: mask = signal_series.str.contains(code, regex=False, na=False) else: mask = signal_series == code if mask.any(): ax4.fill_between(signal_series.index, i, i + 0.8, where=mask, alpha=0.7, color=chart_colors[i % len(chart_colors)], label=name) ylabels = [code_name_map.get(c, c) for c in code_list] ax4.set_title("每日持仓分布", fontsize=12) ax4.set_yticks(range(len(ylabels))) ax4.set_yticklabels(ylabels, fontsize=7) ax4.grid(True, alpha=0.3) chart_path = f"{save_path}_chart.png" plt.savefig(chart_path, dpi=150, bbox_inches="tight") plt.close() print(f"\n报告图表已保存: {chart_path}")