""" 评估与报告模块 — 绩效评估指标、风险分析、可视化报告 """ from dataclasses import dataclass from typing import Dict, List, Optional, Tuple import numpy as np import pandas as pd # --------------------------------------------------------------------------- # 绩效评估器 # --------------------------------------------------------------------------- class PerformanceEvaluator: """ 绩效评估器 — 计算常见量化指标 """ def __init__( self, equity_curve: pd.DataFrame, benchmark_returns: Optional[pd.Series] = None, risk_free_rate: float = 0.03, ): """ equity_curve : DataFrame, 必须包含 'return' 或 'nav' 列, index=date benchmark_returns : Series, 基准日收益率, index=date risk_free_rate : 年化无风险利率 """ self.equity = equity_curve.copy() self.rf = risk_free_rate # 确保有日收益率列 if "daily_return" not in self.equity.columns: if "return" in self.equity.columns: self.equity["daily_return"] = self.equity["return"].diff().fillna(0) elif "nav" in self.equity.columns: self.equity["daily_return"] = self.equity["nav"].pct_change().fillna(0) else: raise ValueError("equity_curve 必须包含 'return'/'nav' 列") self.benchmark_returns = benchmark_returns self._metrics: Dict[str, float] = {} # ----- 基础指标 ----- def total_return(self) -> float: """累计收益率""" if "nav" in self.equity.columns: return self.equity["nav"].iloc[-1] - 1.0 return (1 + self.equity["daily_return"]).prod() - 1.0 def annual_return(self, periods_per_year: int = 252) -> float: """年化收益率""" total = self.total_return() years = self._year_frac(periods_per_year) return (1 + total) ** (1 / years) - 1.0 if years > 0 else 0.0 def annual_volatility(self, periods_per_year: int = 252) -> float: """年化波动率""" return self.equity["daily_return"].std() * np.sqrt(periods_per_year) def max_drawdown(self) -> float: """最大回撤""" if "nav" in self.equity.columns: nav = self.equity["nav"] else: nav = (1 + self.equity["daily_return"]).cumprod() running_max = nav.cummax() drawdown = (nav - running_max) / running_max return drawdown.min() def max_drawdown_duration(self) -> int: """最长回撤持续天数""" if "nav" in self.equity.columns: nav = self.equity["nav"] else: nav = (1 + self.equity["daily_return"]).cumprod() running_max = nav.cummax() is_dd = nav < running_max # 统计最长连续 True 段 dd_groups = (is_dd != is_dd.shift()).cumsum() dd_durations = ( is_dd.groupby(dd_groups).sum() ) return int(dd_durations.max()) if not dd_durations.empty else 0 # ----- 风险调整收益 ----- def sharpe_ratio(self, periods_per_year: int = 252) -> float: """夏普比率""" excess = self.equity["daily_return"] - self.rf / periods_per_year return ( excess.mean() / (excess.std() + 1e-12) * np.sqrt(periods_per_year) ) def calmar_ratio(self) -> float: """Calmar 比率 (年化收益 / 最大回撤绝对值)""" mdd = abs(self.max_drawdown()) return self.annual_return() / mdd if mdd > 0 else 0.0 def sortino_ratio(self, periods_per_year: int = 252) -> float: """Sortino 比率 (只考虑下行波动)""" returns = self.equity["daily_return"] downside = returns[returns < 0] downside_vol = downside.std() * np.sqrt(periods_per_year) excess_annual = self.annual_return() - self.rf return excess_annual / (downside_vol + 1e-12) def win_rate(self) -> float: """胜率（日收益率 > 0 的比例）""" returns = self.equity["daily_return"] return (returns > 0).mean() def profit_loss_ratio(self) -> float: """盈亏比 (平均盈利 / 平均亏损绝对值)""" returns = self.equity["daily_return"] avg_win = returns[returns > 0].mean() avg_loss = abs(returns[returns < 0].mean()) return avg_win / (avg_loss + 1e-12) # ----- 相对基准指标 ----- def alpha(self, periods_per_year: int = 252) -> float: """Jensen's Alpha 相对基准""" if self.benchmark_returns is None: return 0.0 aligned = self._align_benchmark() strat_ret = aligned["strat"] bench_ret = aligned["bench"] excess_strat = strat_ret - self.rf / periods_per_year excess_bench = bench_ret - self.rf / periods_per_year # 线性回归 cov = np.cov(excess_strat, excess_bench) beta = cov[0, 1] / (cov[1, 1] + 1e-12) alpha_daily = excess_strat.mean() - beta * excess_bench.mean() return alpha_daily * periods_per_year def beta(self) -> float: """Beta""" if self.benchmark_returns is None: return 1.0 aligned = self._align_benchmark() cov = np.cov(aligned["strat"], aligned["bench"]) return cov[0, 1] / (cov[1, 1] + 1e-12) def information_ratio(self, periods_per_year: int = 252) -> float: """信息比率""" if self.benchmark_returns is None: return 0.0 aligned = self._align_benchmark() tracking_error = (aligned["strat"] - aligned["bench"]).std() return ( (aligned["strat"].mean() - aligned["bench"].mean()) / (tracking_error + 1e-12) * np.sqrt(periods_per_year) ) # ----- 报告 ----- def full_report(self) -> Dict[str, float]: """生成完整绩效报告字典""" self._metrics = { "累计收益率(%)": self.total_return() * 100, "年化收益率(%)": self.annual_return() * 100, "年化波动率(%)": self.annual_volatility() * 100, "最大回撤(%)": self.max_drawdown() * 100, "最长回撤天数": self.max_drawdown_duration(), "夏普比率": self.sharpe_ratio(), "Calmar比率": self.calmar_ratio(), "Sortino比率": self.sortino_ratio(), "日胜率(%)": self.win_rate() * 100, "盈亏比": self.profit_loss_ratio(), "Alpha(%)": self.alpha() * 100, "Beta": self.beta(), "信息比率": self.information_ratio(), } return self._metrics def summary(self) -> str: """打印可读报告""" report = self.full_report() lines = ["=" * 50, " 量化策略绩效评估报告", "=" * 50] for key, value in report.items(): if isinstance(value, float): lines.append(f" {key}: {value:.4f}") else: lines.append(f" {key}: {value}") lines.append("=" * 50) return "\n".join(lines) # ----- 内部方法 ----- def _year_frac(self, periods_per_year: int) -> float: n = len(self.equity) return n / periods_per_year def _align_benchmark(self) -> pd.DataFrame: """对齐策略与基准日收益率""" strat = self.equity["daily_return"] bench = self.benchmark_returns merged = pd.concat([strat, bench], axis=1, join="inner").dropna() merged.columns = ["strat", "bench"] return merged # --------------------------------------------------------------------------- # 报告生成器 # --------------------------------------------------------------------------- class ReportGenerator: """生成交易分析报告（Markdown / Excel）""" def __init__(self, evaluator: PerformanceEvaluator, trades: Optional[List] = None): self.evaluator = evaluator self.trades = trades or [] def to_markdown(self) -> str: """输出 Markdown 格式报告""" metrics = self.evaluator.full_report() lines = [ "# 量化策略回测报告", "", "## 绩效指标", "", "| 指标 | 数值 |", "|------|------|", ] for key, value in metrics.items(): lines.append(f"| {key} | {value:.4f} |") # 交易统计 if self.trades: lines.extend(["", "## 交易统计", ""]) trades_df = pd.DataFrame( [ { "date": t.date, "stock": t.stock, "side": t.side, "quantity": t.quantity, "price": t.price, } for t in self.trades ] ) total_trades = len(trades_df) buy_trades = len(trades_df[trades_df["side"] == "buy"]) sell_trades = len(trades_df[trades_df["side"] == "sell"]) lines.append(f"- 总交易笔数: {total_trades}") lines.append(f"- 买入笔数: {buy_trades}") lines.append(f"- 卖出笔数: {sell_trades}") return "\n".join(lines) def to_dataframe(self) -> pd.DataFrame: """将绩效指标导出为 DataFrame""" metrics = self.evaluator.full_report() return pd.DataFrame( list(metrics.items()), columns=["指标", "数值"] ).set_index("指标")