etf/strategies/rotation/report.py

"""
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)

    # 绘制图表
    _plot_report_chart(
        backtest_result, code_list, code_name_map,
        benchmark_name, save_path, select_num
    )

    # 返回指标字典
    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 _plot_report_chart(
    backtest_result: pd.DataFrame,
    code_list: list,
    code_name_map: dict,
    benchmark_name: str,
    save_path: str,
    select_num: int,
):
    """绘制报告图表"""
    plt.rcParams["font.sans-serif"] = ["Arial Unicode MS", "SimHei", "DejaVu Sans"]
    plt.rcParams["axes.unicode_minus"] = False

    strategy_nav = backtest_result["轮动策略净值"]
    benchmark_nav = backtest_result["基准净值"]

    fig, axes = plt.subplots(3, 1, figsize=(14, 12))

    # 面板1: 净值曲线
    ax1 = axes[0]
    ax1.plot(strategy_nav.index, strategy_nav.values,
             label="轮动策略", linewidth=2, color="#E74C3C")
    ax1.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
        ax1.plot(backtest_result.index, backtest_result[f"净值_{code}"].values,
                 label=lbl, linewidth=0.8, alpha=0.4,
                 color=chart_colors[i % len(chart_colors)])

    ax1.set_title("ETF轮动策略 - 净值曲线", fontsize=16, fontweight="bold")
    ax1.set_ylabel("净值")
    ax1.legend(loc="upper left", fontsize=8, ncol=2)
    ax1.grid(True, alpha=0.3)
    ax1.set_yscale("log")

    # 面板2: 回撤曲线
    ax2 = axes[1]
    cummax = strategy_nav.cummax()
    drawdown = (strategy_nav - cummax) / cummax
    ax2.fill_between(drawdown.index, drawdown.values, 0, alpha=0.5, color="#E74C3C")
    ax2.set_title("策略回撤", fontsize=12)
    ax2.set_ylabel("回撤")
    ax2.grid(True, alpha=0.3)

    # 面板3: 持仓分布
    ax3 = axes[2]
    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():
            ax3.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]
    ax3.set_title("每日持仓分布", fontsize=12)
    ax3.set_yticks(range(len(ylabels)))
    ax3.set_yticklabels(ylabels, fontsize=7)
    ax3.grid(True, alpha=0.3)

    plt.tight_layout()
    chart_path = f"{save_path}_chart.png"
    plt.savefig(chart_path, dpi=150, bbox_inches="tight")
    plt.close()
    print(f"\n报告图表已保存: {chart_path}")