| """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 |
|
|
