myquant/finance/backtest/report.py

"""
标准化回测报告。

与回测引擎解耦，后续换引擎只需改构造函数。
"""

from dataclasses import dataclass, field

import numpy as np
import pandas as pd


@dataclass
class BacktestReport:
    """回测报告"""

    # 核心收益指标
    total_return: float = 0.0
    cagr: float = 0.0
    max_drawdown: float = 0.0
    sharpe_ratio: float = 0.0
    calmar_ratio: float = 0.0
    annual_volatility: float = 0.0

    # 交易统计
    win_rate: float = 0.0
    profit_factor: float = 0.0
    total_trades: int = 0
    avg_hold_days: float = 0.0
    best_trade_pct: float = 0.0
    worst_trade_pct: float = 0.0

    # 序列数据
    equity_curve: pd.Series = field(default_factory=pd.Series)
    drawdown_curve: pd.Series = field(default_factory=pd.Series)
    monthly_returns: pd.Series = field(default_factory=pd.Series)
    trades_df: pd.DataFrame = field(default_factory=pd.DataFrame)

    # 原始 stats
    stats_dict: dict = field(default_factory=dict)

    @classmethod
    def from_vbt_result(cls, pf, close: pd.Series) -> "BacktestReport":
        """从 VectorBT Portfolio 结果构建报告。"""
        stats = pf.stats()

        def _pct(v):
            """VectorBT stats 值已为百分比 float，直接返回。"""
            if v is None:
                return 0.0
            try:
                return float(v)
            except (ValueError, TypeError):
                return 0.0

        def _duration_days(v):
            """Timedelta → 天数。"""
            if v is None:
                return 0.0
            try:
                return v.total_seconds() / 86400
            except AttributeError:
                return float(v) if v is not None else 0.0

        equity = pf.value()
        equity = pd.Series(equity.values, index=close.index[: len(equity)])

        if not isinstance(equity.index, pd.DatetimeIndex):
            equity.index = pd.to_datetime(equity.index, format="%Y%m%d")

        dd = equity / equity.cummax() - 1
        daily_ret = equity.pct_change().dropna()

        years = len(daily_ret) / 252 if len(daily_ret) > 0 else 1
        total_ret = (equity.iloc[-1] / equity.iloc[0] - 1) * 100 if len(equity) > 1 else 0
        cagr = ((total_ret / 100 + 1) ** (1 / years) - 1) * 100 if years > 0 else 0
        mdd = dd.min() * 100
        ann_vol = daily_ret.std() * np.sqrt(252) * 100 if len(daily_ret) > 0 else 0

        mean_ret = daily_ret.mean() * 252
        std_ret = daily_ret.std() * np.sqrt(252)
        sharpe = mean_ret / std_ret if std_ret > 0 else 0
        calmar = cagr / abs(mdd) if mdd != 0 else 0

        trades = pf.trades.records_readable if hasattr(pf, "trades") else pd.DataFrame()

        try:
            monthly = equity.resample("ME").last().pct_change()
        except Exception:
            monthly = pd.Series(dtype=float)

        pf_factor = stats.get("Profit Factor", 0)
        if pf_factor is None or np.isinf(float(pf_factor)):
            pf_factor = 0.0
        else:
            pf_factor = float(pf_factor)

        return cls(
            total_return=round(total_ret, 2),
            cagr=round(cagr, 2),
            max_drawdown=round(mdd, 2),
            sharpe_ratio=round(sharpe, 2),
            calmar_ratio=round(calmar, 2),
            annual_volatility=round(ann_vol, 2),
            win_rate=_pct(stats.get("Win Rate [%]", 0)),
            profit_factor=pf_factor,
            total_trades=int(stats.get("Total Trades", 0)),
            avg_hold_days=_duration_days(stats.get("Avg Winning Trade Duration", None)),
            best_trade_pct=_pct(stats.get("Best Trade [%]", 0)),
            worst_trade_pct=_pct(stats.get("Worst Trade [%]", 0)),
            equity_curve=equity,
            drawdown_curve=dd,
            monthly_returns=monthly,
            trades_df=trades,
            stats_dict={k: str(v) for k, v in stats.items()},
        )

    def to_dict(self) -> dict:
        """核心指标转字典。"""
        return {
            "total_return": self.total_return,
            "cagr": self.cagr,
            "max_drawdown": self.max_drawdown,
            "sharpe_ratio": self.sharpe_ratio,
            "calmar_ratio": self.calmar_ratio,
            "annual_volatility": self.annual_volatility,
            "win_rate": self.win_rate,
            "profit_factor": self.profit_factor,
            "total_trades": self.total_trades,
        }

    def summary(self) -> str:
        """一行摘要。"""
        return (
            f"收益={self.total_return:.1f}% "
            f"年化={self.cagr:.1f}% "
            f"回撤={self.max_drawdown:.1f}% "
            f"夏普={self.sharpe_ratio:.2f} "
            f"胜率={self.win_rate:.1f}% "
            f"交易={self.total_trades}笔"
        )

    def __repr__(self) -> str:
        return self.summary()