Yuxuan Yan
91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
"""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)
|