""" 标准化回测报告。 与回测引擎解耦,后续换引擎只需改构造函数。 """ 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()