feat: add strict benchmark-relative performance metrics

src/quant_engine/metrics.py

@@ -130,6 +130,62 @@ def summary(r: pd.Series, rf: float = 0.0) -> Mapping[str, float]:
     }
 
 
+def benchmark_summary(
+    portfolio_returns: pd.Series,
+    benchmark_returns: pd.Series,
+    *,
+    risk_free_daily: float = 0.0,
+    annualization: int = TRADING_DAYS_PER_YEAR,
+) -> Mapping[str, float]:
+    """计算成本后组合相对基准的严格对齐绩效。
+
+    与通用 ``summary`` 不同，本函数拒绝静默清洗或日期 inner join。alpha
+    使用日频回归截距的几何年化；基准方差不足时 alpha/beta 为 NaN，明确
+    表示回归不可估计。
+    """
+    portfolio, benchmark = _validate_benchmark_inputs(
+        portfolio_returns,
+        benchmark_returns,
+    )
+    if isinstance(annualization, bool) or not isinstance(annualization, int):
+        raise TypeError("annualization must be an integer")
+    if annualization <= 0:
+        raise ValueError("annualization must be positive")
+    if not np.isfinite(risk_free_daily):
+        raise ValueError("risk_free_daily must be finite")
+
+    active = portfolio - benchmark
+    active_std = float(active.std())
+    tracking_error = active_std * float(np.sqrt(annualization))
+    information_ratio = (
+        float(active.mean()) / active_std * float(np.sqrt(annualization))
+        if active_std >= 1e-30
+        else float("nan")
+    )
+
+    adjusted_portfolio = portfolio - risk_free_daily
+    adjusted_benchmark = benchmark - risk_free_daily
+    benchmark_variance = float(adjusted_benchmark.var())
+    if benchmark_variance < 1e-30:
+        beta = float("nan")
+        alpha = float("nan")
+    else:
+        beta = float(adjusted_portfolio.cov(adjusted_benchmark) / benchmark_variance)
+        alpha_daily = float((adjusted_portfolio - beta * adjusted_benchmark).mean())
+        alpha = (
+            float((1.0 + alpha_daily) ** annualization - 1.0)
+            if alpha_daily > -1.0
+            else float("nan")
+        )
+    return {
+        "n_observations": len(portfolio),
+        "tracking_error": tracking_error,
+        "information_ratio": information_ratio,
+        "alpha": alpha,
+        "beta": beta,
+    }
+
+
 # ── 内部 ──────────────────────────────────────
 
 
@@ -138,3 +194,29 @@ def _clean(r: pd.Series) -> pd.Series:
     if not isinstance(r, pd.Series):
         raise TypeError(f"expected pd.Series, got {type(r).__name__}")
     return r.replace([np.inf, -np.inf], np.nan).dropna()
+
+
+def _validate_benchmark_inputs(
+    portfolio_returns: pd.Series,
+    benchmark_returns: pd.Series,
+) -> tuple[pd.Series, pd.Series]:
+    if not isinstance(portfolio_returns, pd.Series):
+        raise TypeError("portfolio_returns must be a pandas Series")
+    if not isinstance(benchmark_returns, pd.Series):
+        raise TypeError("benchmark_returns must be a pandas Series")
+    if not portfolio_returns.index.equals(benchmark_returns.index):
+        raise ValueError("portfolio and benchmark returns must use matching indexes")
+    if not portfolio_returns.index.is_unique:
+        raise ValueError("portfolio and benchmark indexes must be unique")
+    if len(portfolio_returns) < 2:
+        raise ValueError("benchmark metrics require at least two observations")
+
+    portfolio = portfolio_returns.astype(float, copy=True)
+    benchmark = benchmark_returns.astype(float, copy=True)
+    if not np.isfinite(portfolio.to_numpy()).all() or not np.isfinite(
+        benchmark.to_numpy()
+    ).all():
+        raise ValueError("portfolio and benchmark returns must be finite")
+    if (portfolio < -1.0).any() or (benchmark < -1.0).any():
+        raise ValueError("simple returns cannot be less than -1")
+    return portfolio, benchmark

src/quant_engine/research_pipeline.py

@@ -23,7 +23,7 @@ from quant_engine.execution import (
     simulate_daily_ledger_with_audit,
     simulate_multi_day_with_audit,
 )
-from quant_engine.metrics import summary as metrics_summary
+from quant_engine.metrics import benchmark_summary, summary as metrics_summary
 from quant_engine.portfolio_construction import scores_to_weight_table
 
 __all__ = [
@@ -99,6 +99,10 @@ class FactorBacktestResult:
             self.valuation_prices,
         )
 
+    def benchmark_stats(self, benchmark_returns: pd.Series) -> Mapping[str, float]:
+        """计算成本后日收益相对同日基准的 TE、IR、alpha 与 beta。"""
+        return benchmark_summary(self.returns, benchmark_returns)
+
 
 def _validate_datetime_index(index: pd.Index, name: str) -> pd.DatetimeIndex:
     if not isinstance(index, pd.DatetimeIndex):