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)

    # 打印最新调仓信号
    _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"] = ["Arial Unicode MS", "SimHei", "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)

    # 计算表格行数
    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}")