feat: signal abstraction layer + sizer + HS300 universe + PnL/IC reports

eval/metrics.py

@@ -0,0 +1,90 @@
+"""Information-coefficient metrics for alpha signals."""
+from typing import Any
+
+import pandas as pd
+
+
+def _summarize(ic: pd.Series, rank_ic: pd.Series) -> dict[str, Any]:
+    """Aggregate per-period IC series into summary statistics."""
+    ic = ic.dropna()
+    rank_ic = rank_ic.dropna()
+
+    ic_mean = float(ic.mean()) if len(ic) else float("nan")
+    ic_std = float(ic.std()) if len(ic) else float("nan")
+    rank_ic_mean = float(rank_ic.mean()) if len(rank_ic) else float("nan")
+    rank_ic_std = float(rank_ic.std()) if len(rank_ic) else float("nan")
+
+    return {
+        "ic_mean": ic_mean,
+        "ic_std": ic_std,
+        "ir": ic_mean / ic_std if ic_std else float("nan"),
+        "rank_ic_mean": rank_ic_mean,
+        "rank_ic_std": rank_ic_std,
+        "rank_ir": rank_ic_mean / rank_ic_std if rank_ic_std else float("nan"),
+        "hit_rate": float((rank_ic > 0).mean()) if len(rank_ic) else float("nan"),
+        "n_periods": int(len(rank_ic)),
+        "ic_series": ic,
+        "rank_ic_series": rank_ic,
+    }
+
+
+def _cross_sectional(signals_df: pd.DataFrame, returns_df: pd.DataFrame) -> dict[str, Any]:
+    """Per-date IC across stocks (requires >= 2 stocks)."""
+    dates = signals_df.index
+    ic_vals, rank_ic_vals, idx = [], [], []
+    for dt in dates:
+        s = signals_df.loc[dt]
+        r = returns_df.loc[dt]
+        mask = s.notna() & r.notna()
+        if mask.sum() < 2:
+            continue
+        sv, rv = s[mask], r[mask]
+        # A degenerate (constant) vector makes correlation undefined.
+        if sv.nunique() < 2 or rv.nunique() < 2:
+            continue
+        ic_vals.append(sv.corr(rv))
+        rank_ic_vals.append(sv.corr(rv, method="spearman"))
+        idx.append(dt)
+    ic = pd.Series(ic_vals, index=idx, dtype=float)
+    rank_ic = pd.Series(rank_ic_vals, index=idx, dtype=float)
+    return _summarize(ic, rank_ic)
+
+
+def _rolling_single(
+    signals_df: pd.DataFrame, returns_df: pd.DataFrame, window: int = 20
+) -> dict[str, Any]:
+    """Rolling time-series IC for the single-stock case.
+
+    With one stock there is no cross-section, so we measure how well the signal
+    tracks forward returns over a trailing window instead.
+    """
+    col = signals_df.columns[0]
+    s = signals_df[col]
+    r = returns_df[col]
+    ic = s.rolling(window).corr(r)
+    rank_ic = s.rank().rolling(window).corr(r.rank())
+    return _summarize(ic, rank_ic)
+
+
+def evaluate_cross_sectional(
+    signals_df: pd.DataFrame, returns_df: pd.DataFrame
+) -> dict[str, Any]:
+    """Evaluate a signal's predictive power against forward returns.
+
+    Args:
+        signals_df: DataFrame indexed by date, one column per stock, signal values.
+        returns_df: DataFrame indexed by date, one column per stock, forward returns.
+
+    Returns:
+        Dict with ``ic_mean``, ``ic_std``, ``ir``, ``rank_ic_mean``,
+        ``rank_ic_std``, ``rank_ir``, ``hit_rate``, ``n_periods`` and the
+        per-period ``ic_series`` / ``rank_ic_series`` (for plotting).
+    """
+    cols = signals_df.columns.intersection(returns_df.columns)
+    idx = signals_df.index.intersection(returns_df.index)
+    signals_df = signals_df.loc[idx, cols]
+    returns_df = returns_df.loc[idx, cols]
+
+    if len(cols) >= 2:
+        return _cross_sectional(signals_df, returns_df)
+    return _rolling_single(signals_df, returns_df)