File size: 2,339 Bytes
590a501 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | """Factor evaluation utilities compatible with qlib and GP outputs."""
from __future__ import annotations
import numpy as np
import pandas as pd
import scipy.stats
def cross_sectional_rank(series: pd.Series) -> pd.Series:
return series.rank(pct=True)
def daily_rank_ic(factor_df: pd.DataFrame, label_col: str = "label", factor_col: str = "factor") -> pd.Series:
"""Compute daily Rank IC from long-format panel (date, symbol, factor, label)."""
ics = []
dates = []
for dt, group in factor_df.groupby("date"):
if group[factor_col].nunique() <= 1 or group[label_col].nunique() <= 1:
continue
ic, _ = scipy.stats.spearmanr(group[factor_col], group[label_col])
if np.isfinite(ic):
ics.append(ic)
dates.append(dt)
return pd.Series(ics, index=dates, name="rank_ic")
def ic_summary(ic_series: pd.Series) -> dict:
if ic_series.empty:
return {"ic_mean": np.nan, "ic_std": np.nan, "icir": np.nan, "pos_ratio": np.nan}
return {
"ic_mean": float(ic_series.mean()),
"ic_std": float(ic_series.std()),
"icir": float(ic_series.mean() / (ic_series.std() + 1e-8)),
"pos_ratio": float((ic_series > 0).mean()),
"n_days": len(ic_series),
}
def quantile_spread(
factor_df: pd.DataFrame,
label_col: str = "label",
factor_col: str = "factor",
n_groups: int = 5,
) -> pd.DataFrame:
"""Long-short spread by factor quantile groups."""
rows = []
for dt, group in factor_df.groupby("date"):
if len(group) < n_groups * 2:
continue
group = group.copy()
group["group"] = pd.qcut(group[factor_col].rank(method="first"), n_groups, labels=False)
grp_ret = group.groupby("group")[label_col].mean()
rows.append({"date": dt, "long_short": grp_ret.iloc[-1] - grp_ret.iloc[0]})
return pd.DataFrame(rows)
def evaluate_factor_panel(factor_df: pd.DataFrame) -> dict:
ic = daily_rank_ic(factor_df)
spread = quantile_spread(factor_df)
result = ic_summary(ic)
if not spread.empty:
result["long_short_ann_return"] = float(spread["long_short"].mean() * 252)
result["long_short_sharpe"] = float(
spread["long_short"].mean() / (spread["long_short"].std() + 1e-8) * np.sqrt(252)
)
return result
|