aszer/etf
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e6ddea518c2775c23aa79298d3b796bbf03e8751
etf/strategies/rotation/report.py
aszerW e6ddea518c feat(report): 支持ETF净值和溢价率的绩效报告展示 
- 在生成绩效报告接口中新增code_config、index_data、etf_price_data和etf_nav_data_raw参数
- 计算溢价率并基于信号前一日数据进行校验和计算
- 打印最新调仓信号时增加ETF代码、ETF净值、溢价率及高溢价警告显示
- 调整信号数据基准日期展示，更准确反映信号计算依据
- 报告图表支持显示ETF净值和溢价率列，完善调仓信息视觉效果
- 统一处理跨市场ETF映射和特殊市场（如加密货币）情况，避免溢价率误报
- 完善打印表格和图表的列宽和格式，增强可读性
2026-03-25 22:02:05 +08:00

703 lines
28 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
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,
code_config: dict = None,
index_data: pd.DataFrame = None,
etf_price_data: pd.DataFrame = None,
etf_nav_data_raw: pd.DataFrame = None,
) -> dict:
"""
生成完整的绩效报告
Args:
backtest_result: 回测结果
code_list: ETF代码列表
code_name_map: 代码到名称映射
benchmark_name: 基准名称
save_path: 报告保存路径前缀
select_num: 选中数量
code_config: 代码配置（包含 name, etf, market），用于显示ETF映射
index_data: 指数价格数据
etf_price_data: ETF价格数据（用于计算溢价率）
etf_nav_data_raw: ETF净值数据（用于计算溢价率）
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 {}
code_config = code_config 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)
# 计算溢价率（需要ETF价格和ETF净值）
# 溢价率 = (ETF价格 - ETF净值) / ETF净值
# 注意：信号是基于前一日数据计算的，所以使用 signal_date - 1 的数据
etf_nav_data = {}
premium_data = {}
if etf_price_data is not None and etf_nav_data_raw is not None:
# 使用信号前一天的数据（因为信号是基于前一天收盘数据计算的）
signal_date = backtest_result.index[-1]
data_base_date = signal_date - pd.Timedelta(days=1)
# 确保数据基准日期在数据范围内
if data_base_date in etf_price_data.index and data_base_date in etf_nav_data_raw.index:
for code in code_list:
if code in etf_price_data.columns and code in etf_nav_data_raw.columns:
etf_price = etf_price_data.loc[data_base_date, code]
etf_nav = etf_nav_data_raw.loc[data_base_date, code]
if pd.notna(etf_price) and pd.notna(etf_nav) and etf_nav > 0:
premium = (etf_price - etf_nav) / etf_nav
premium_data[code] = premium
etf_nav_data[code] = etf_nav
# 打印最新调仓信号
_print_latest_signal(backtest_result, code_list, code_name_map, select_num, code_config, etf_nav_data, premium_data)
# 绘制图表
_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,
},
code_config=code_config,
etf_price_data=etf_price_data,
etf_nav_data_raw=etf_nav_data_raw,
)
# 返回指标字典
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, code_config: dict = None, etf_nav_data: dict = None, premium_data: dict = None):
"""打印最新调仓信号（支持ETF映射、ETF净值和溢价率显示）"""
code_config = code_config or {}
etf_nav_data = etf_nav_data or {}
premium_data = premium_data or {}
latest = _extract_latest_positions(backtest_result, code_list, code_name_map, select_num)
signal_date = latest["signal_date"]
signal_date_str = signal_date.strftime("%Y-%m-%d")
# 数据基准日期（信号是基于前一日数据计算的）
# 根据跨市场ETF映射方案：T+1日09:00计算的信号基于T日数据
data_base_date = signal_date - pd.Timedelta(days=1)
data_base_date_str = data_base_date.strftime("%Y-%m-%d")
print("\n")
print("=" * 135)
print(" 最新调仓信号 (下一交易日执行)")
print("=" * 135)
print(f" 信号日期: {signal_date_str} (基于 {data_base_date_str} 收盘数据)")
print()
# 表头 - 添加ETF净值和溢价率列
print(f' {"标的名称":<10} {"指数代码":>12} {"ETF代码":>12} {"仓位":>6} {"得分":>8} {"进场日期":>12} {"指数进场价":>10} {"指数最新价":>10} {"ETF净值":>10} {"溢价率":>8} {"操作":>6} {"持有天数":>8} {"盈亏":>10}')
print(" " + "-" * 155)
# 下期持仓（调入/维持）
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 ' '
# 获取ETF代码、ETF净值和溢价率
idx_code = pos["code"]
cfg = code_config.get(idx_code, {})
etf_code = cfg.get('etf', '—')
market = cfg.get('market', 'A')
if etf_code is None:
etf_code = '直接交易'
# 获取ETF净值和溢价率
if market == 'CRYPTO':
etf_nav_str = ' —'
premium_str = ' —'
else:
# ETF净值
etf_nav = etf_nav_data.get(idx_code)
if etf_nav is not None:
etf_nav_str = f'{etf_nav:>10.3f}'
else:
etf_nav_str = ' —'
# 溢价率
premium = premium_data.get(idx_code)
if premium is not None:
# 高溢价警告标记
warning = '⚠️' if premium > 0.02 else ''
premium_str = f'{premium:>+7.2%}{warning}'
else:
premium_str = ' —'
print(f' {pos["name"]:<10} {idx_code:>12} {etf_code:>12} {pos["weight"]:>6.0%} {score_str} {entry_date_str:>12} {entry_str} {pos["current_price"]:>10.2f} {etf_nav_str} {premium_str} {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 = ' —' # 调出品种无得分
# 获取ETF代码、ETF净值和溢价率
idx_code = pos["code"]
cfg = code_config.get(idx_code, {})
etf_code = cfg.get('etf', '—')
market = cfg.get('market', 'A')
if etf_code is None:
etf_code = '直接交易'
# 获取ETF净值和溢价率
if market == 'CRYPTO':
etf_nav_str = ' —'
premium_str = ' —'
else:
# ETF净值
etf_nav = etf_nav_data.get(idx_code)
if etf_nav is not None:
etf_nav_str = f'{etf_nav:>10.3f}'
else:
etf_nav_str = ' —'
# 溢价率
premium = premium_data.get(idx_code)
if premium is not None:
warning = '⚠️' if premium > 0.02 else ''
premium_str = f'{premium:>+7.2%}{warning}'
else:
premium_str = ' —'
print(f' {pos["name"]:<10} {idx_code:>12} {etf_code:>12} {pos["weight"]:>6.0%} {score_str} {entry_date_str:>12} {entry_str} {pos["current_price"]:>10.2f} {etf_nav_str} {premium_str} ▼调出 {days_str} {pnl_str}')
print("=" * 160)
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,
code_config: dict = None,
etf_price_data: pd.DataFrame = None,
etf_nav_data_raw: pd.DataFrame = None,
):
"""绘制报告图表（支持ETF净值和溢价率显示）"""
# 设置中文字体（macOS: Arial Unicode MS, Linux: WenQuanYi Zen Hei）
plt.rcParams["font.sans-serif"] = ["Arial Unicode MS", "WenQuanYi Zen Hei", "DejaVu Sans"]
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)
# 准备配置数据
code_config = code_config or {}
signal_date = backtest_result.index[-1]
# 数据基准日期（信号是基于前一日数据计算的）
data_base_date = signal_date - pd.Timedelta(days=1)
# 计算ETF净值和溢价率（使用数据基准日期的数据）
etf_nav_dict = {}
premium_dict = {}
if etf_price_data is not None and etf_nav_data_raw is not None:
# 确保数据基准日期在数据范围内
if data_base_date in etf_price_data.index and data_base_date in etf_nav_data_raw.index:
for code in code_list:
if code in etf_price_data.columns and code in etf_nav_data_raw.columns:
etf_price = etf_price_data.loc[data_base_date, code]
etf_nav = etf_nav_data_raw.loc[data_base_date, code]
if pd.notna(etf_price) and pd.notna(etf_nav) and etf_nav > 0:
premium = (etf_price - etf_nav) / etf_nav
etf_nav_dict[code] = etf_nav
premium_dict[code] = premium
# 计算表格行数
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 = latest["signal_date"]
signal_date_str = signal_date.strftime("%Y-%m-%d")
# 数据基准日期（信号是基于前一日数据计算的）
data_base_date = signal_date - pd.Timedelta(days=1)
data_base_date_str = data_base_date.strftime("%Y-%m-%d")
ax0.set_title(f"最新调仓信号 (信号日期: {signal_date_str}，基于 {data_base_date_str} 数据，下一交易日执行)", fontsize=14, fontweight="bold", loc="left", pad=15)
# 构建表格数据（添加ETF净值和溢价率列）
table_data = []
col_labels = ["标的名称", "指数代码", "仓位", "得分", "进场日期", "进场价", "最新价", "ETF净值", "溢价率", "操作", "持有天数", "盈亏"]
# 下期持仓（调入/维持）
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 "—"
# 获取ETF净值和溢价率
idx_code = pos["code"]
cfg = code_config.get(idx_code, {})
market = cfg.get('market', 'A')
if market == 'CRYPTO':
etf_nav_str = "—"
premium_str = "—"
else:
etf_nav = etf_nav_dict.get(idx_code)
premium = premium_dict.get(idx_code)
etf_nav_str = f"{etf_nav:.3f}" if etf_nav is not None else "—"
if premium is not None:
warning = "⚠️" if premium > 0.02 else ""
premium_str = f"{premium:+.2%}{warning}"
else:
premium_str = "—"
table_data.append([
pos["name"], pos["code"], f'{pos["weight"]:.0%}',
score_str, entry_date_str, entry_str, f'{pos["current_price"]:.2f}',
etf_nav_str, premium_str, 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 = "—" # 调出品种无得分
# 获取ETF净值和溢价率
idx_code = pos["code"]
cfg = code_config.get(idx_code, {})
market = cfg.get('market', 'A')
if market == 'CRYPTO':
etf_nav_str = "—"
premium_str = "—"
else:
etf_nav = etf_nav_dict.get(idx_code)
premium = premium_dict.get(idx_code)
etf_nav_str = f"{etf_nav:.3f}" if etf_nav is not None else "—"
if premium is not None:
warning = "⚠️" if premium > 0.02 else ""
premium_str = f"{premium:+.2%}{warning}"
else:
premium_str = "—"
table_data.append([
pos["name"], pos["code"], f'{pos["weight"]:.0%}',
score_str, entry_date_str, entry_str, f'{pos["current_price"]:.2f}',
etf_nav_str, premium_str, "调出", days_str, pnl_str
])
if table_data:
table = ax0.table(
cellText=table_data,
colLabels=col_labels,
loc="center",
cellLoc="center",
colWidths=[0.09, 0.09, 0.06, 0.07, 0.07, 0.07, 0.07, 0.07, 0.08, 0.06, 0.07, 0.07],
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}")
Reference in New Issue View Git Blame Copy Permalink