feat:添加alpha模块

alpha/evaluation.py

@@ -0,0 +1,262 @@
+"""
+评估与报告模块 — 绩效评估指标、风险分析、可视化报告
+"""
+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("指标")