diff --git a/sanguo_factor/metrics.py b/sanguo_factor/metrics.py new file mode 100644 index 0000000..8ca78e4 --- /dev/null +++ b/sanguo_factor/metrics.py @@ -0,0 +1,146 @@ +# sanguo_factor/metrics.py +"""批量评估指标:全向量化(RankIC=秩相关逐行),不依赖 alphalens. + +所有矩阵约定:pandas DataFrame,index=DatetimeIndex(日),columns=vt_symbol, +值=因子值或前瞻收益,NaN=缺失自动从当日截面剔除。 +""" +import warnings + +import numpy as np +import pandas as pd + +TRADING_DAYS_PER_YEAR = 244 + + +def _row_pearson(a: np.ndarray, b: np.ndarray, index: pd.Index) -> pd.Series: + """逐行 Pearson(输入已是秩),全 NaN/无方差行 → NaN.""" + mask = ~(np.isnan(a) | np.isnan(b)) + n = mask.sum(axis=1) + a0 = np.where(mask, a, np.nan) + b0 = np.where(mask, b, np.nan) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", RuntimeWarning) + am = np.nanmean(a0, axis=1, keepdims=True) + bm = np.nanmean(b0, axis=1, keepdims=True) + ad = np.where(mask, a0 - am, 0.0) + bd = np.where(mask, b0 - bm, 0.0) + denom = np.sqrt((ad ** 2).sum(axis=1) * (bd ** 2).sum(axis=1)) + with np.errstate(invalid="ignore", divide="ignore"): + ic = np.where(denom > 0, (ad * bd).sum(axis=1) / denom, np.nan) + ic = np.where(n >= 3, ic, np.nan) # <3 只无意义 + return pd.Series(ic, index=index) + + +def rank_corr_rows(A: pd.DataFrame, B: pd.DataFrame) -> pd.Series: + """逐日 Spearman:先各自按行取秩再逐行 Pearson.""" + cols = A.columns.intersection(B.columns) + idx = A.index.intersection(B.index) + a = A.loc[idx, cols].rank(axis=1).to_numpy(dtype=float) + b = B.loc[idx, cols].rank(axis=1).to_numpy(dtype=float) + return _row_pearson(a, b, idx) + + +def factor_turnover(F: pd.DataFrame) -> float: + """换手率 = 1 - 相邻两日因子秩相关均值.""" + if len(F) < 2: + return 0.0 + # Shift indices to align consecutive days + F_next = F.iloc[1:].reset_index(drop=True) + F_prev = F.iloc[:-1].reset_index(drop=True) + corr = rank_corr_rows(F_next, F_prev).replace([np.inf, -np.inf], np.nan).dropna() + return float(1.0 - corr.mean()) if len(corr) else 0.0 + + +def _quantile_mask(F: pd.DataFrame, lo_frac: float, hi_frac: float) -> pd.DataFrame: + """按行把因子值分位选mask(基于升序秩/当日有效数).""" + ranks = F.rank(axis=1, ascending=False) # 1=最大 + n = ranks.notna().sum(axis=1) + k = np.maximum((n * 0.1).round().astype(int), 1) + if lo_frac == 0.0: + return ranks.le(k, axis=0) & ranks.notna() + return ranks.ge(n - k + 1, axis=0) & ranks.notna() + + +def long_short_annual_return(F: pd.DataFrame, R: pd.DataFrame) -> float | None: + """多空年化:top10% - bottom10% 等权前瞻日收益均值,复利年化.""" + cols = F.columns.intersection(R.columns) + idx = F.index.intersection(R.index) + f, r = F.loc[idx, cols], R.loc[idx, cols] + top = _quantile_mask(f, 0.0, 0.1) + bot = _quantile_mask(f, 0.9, 1.0) + daily = (r.where(top).mean(axis=1) - r.where(bot).mean(axis=1)).dropna() + if daily.empty: + return None + return float((1.0 + daily.mean()) ** TRADING_DAYS_PER_YEAR - 1.0) + + +def decile_annual_returns(F: pd.DataFrame, R: pd.DataFrame) -> list[float | None]: + """十分组(D1因子最低→D10最高)等权年化收益.""" + cols = F.columns.intersection(R.columns) + idx = F.index.intersection(R.index) + f, r = F.loc[idx, cols], R.loc[idx, cols] + pct = f.rank(axis=1, ascending=True).div(f.notna().sum(axis=1), axis=0) + out: list[float | None] = [] + for d in range(10): + sel = (pct > d / 10) & (pct <= (d + 1) / 10) + daily = r.where(sel).mean(axis=1).dropna() + out.append( + float((1.0 + daily.mean()) ** TRADING_DAYS_PER_YEAR - 1.0) + if len(daily) else None + ) + return out + + +def monthly_ic(ic: pd.Series) -> list[dict]: + """IC 按月聚合(sparkline/详情图数据).""" + s = ic.dropna() + if s.empty: + return [] + g = s.groupby(s.index.to_period("M").to_timestamp()).mean() + return [{"month": t.strftime("%Y-%m"), "ic": round(float(v), 6)} for t, v in g.items()] + + +def classify(icir: float | None, t_stat: float | None) -> str: + """结论信号灯:|ICIR|>=0.3 且 |t|>=2 有效;否则 |t|>=1.5 观察;其余淘汰.""" + if icir is None or t_stat is None: + return "eliminated" + if abs(icir) >= 0.3 and abs(t_stat) >= 2.0: + return "effective" + if abs(t_stat) >= 1.5: + return "watch" + return "eliminated" + + +def _period_stats(ic: pd.Series) -> dict: + s = ic.dropna() + n = len(s) + if n < 2: + return {"count": int(n), "ic_mean": None, "ic_std": None, + "icir": None, "t_stat": None, "win_rate": None, + "monthly_ic": monthly_ic(ic)} + m = float(s.mean()) + sd = float(s.std()) + icir = m / sd if sd > 0 else None + t_stat = m / (sd / n ** 0.5) if sd > 0 else None + return { + "count": int(n), + "ic_mean": m, + "ic_std": sd, + "icir": icir, + "t_stat": t_stat, + "win_rate": float((s > 0).mean()), + "monthly_ic": monthly_ic(ic), + } + + +def summarize_factor(F: pd.DataFrame, R1: pd.DataFrame, R5: pd.DataFrame, R10: pd.DataFrame) -> dict: + """单因子全指标:三周期 IC 族 + 换手 + 多空 + 十分组 + 结论.""" + out: dict = {"turnover": factor_turnover(F)} + for p, R in (("1", R1), ("5", R5), ("10", R10)): + ic = rank_corr_rows(F, R) + stats = _period_stats(ic) + stats["ls_annual"] = long_short_annual_return(F, R) + stats["deciles"] = decile_annual_returns(F, R) + stats["conclusion"] = classify(stats["icir"], stats["t_stat"]) + out[p] = stats + return out diff --git a/tests/factor/test_metrics.py b/tests/factor/test_metrics.py new file mode 100644 index 0000000..93e2fdd --- /dev/null +++ b/tests/factor/test_metrics.py @@ -0,0 +1,101 @@ +# tests/factor/test_metrics.py +"""向量化指标:已知输入的精确断言.""" +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +import numpy as np +import pandas as pd +import pytest + +from sanguo_factor.metrics import ( + rank_corr_rows, factor_turnover, long_short_annual_return, + decile_annual_returns, monthly_ic, classify, summarize_factor, + TRADING_DAYS_PER_YEAR, +) + + +def _mat(values, cols=("A", "B", "C")): + n_rows = len(values) + idx = pd.to_datetime([f"2024-01-{i+1:02d}" for i in range(n_rows)]) + return pd.DataFrame(values, index=idx, columns=list(cols), dtype=float) + + +def test_rank_corr_perfect_monotonic(): + F = _mat([[1, 2, 3], [3, 2, 1], [1, 3, 2]]) + R = _mat([[10, 20, 30], [30, 20, 10], [10, 30, 20]]) + ic = rank_corr_rows(F, R) + assert (ic == 1.0).all() + + +def test_rank_corr_inverse(): + F = _mat([[1, 2, 3], [1, 2, 3]]) + R = _mat([[3, 2, 1], [30, 20, 10]]) + ic = rank_corr_rows(F, R) + assert (ic == -1.0).all() + + +def test_rank_corr_nan_propagates_row(): + F = _mat([[1, 2, 3], [np.nan, 2, 3]]) + R = _mat([[1, 2, 3], [1, 2, 3]]) + ic = rank_corr_rows(F, R) + assert not np.isnan(ic.iloc[0]) + assert np.isnan(ic.iloc[1]) # 2个有效值秩恒定 → 无方差 → nan + + +def test_turnover_zero_for_static(): + F = _mat([[1, 2, 3]] * 4) + assert factor_turnover(F) == pytest.approx(0.0) + + +def test_turnover_full_for_shuffled(): + F = _mat([[1, 2, 3], [3, 2, 1]]) # 完全逆序 → 秩相关-1 → 换手=2 + assert factor_turnover(F) == pytest.approx(2.0) + + +def test_long_short_direction(): + F = _mat([[1, 2, 3, 4, 5]] * 2, cols=tuple("ABCDE")) + R = _mat([[0.01, 0.02, 0.03, 0.04, 0.05], + [0.01, 0.02, 0.03, 0.04, 0.05]], cols=tuple("ABCDE")) + # 每日 top10%(1只)=E bottom10%=A → 日均 0.04 + expected = (1 + 0.04) ** TRADING_DAYS_PER_YEAR - 1 + assert long_short_annual_return(F, R) == pytest.approx(expected) + + +def test_decile_monotonic(): + F = _mat([list(range(1, 11))] * 3, cols=tuple("ABCDEFGHIJ")) + R = _mat([[c / 100 for c in range(1, 11)]] * 3, cols=tuple("ABCDEFGHIJ")) + dec = decile_annual_returns(F, R) + assert len(dec) == 10 + assert all(d is not None for d in dec) + assert dec == sorted(dec) # D1最低收益 → D10最高收益 单调 + + +def test_monthly_ic_shape(): + idx = pd.to_datetime(["2024-01-05", "2024-01-10", "2024-02-01"]) + out = monthly_ic(pd.Series([0.1, 0.2, -0.1], index=idx)) + assert out[0] == {"month": "2024-01", "ic": pytest.approx(0.15)} + assert out[1] == {"month": "2024-02", "ic": pytest.approx(-0.1)} + + +def test_classify_rules(): + assert classify(0.5, 5.0) == "effective" + assert classify(-0.4, -3.0) == "effective" # 负 IC 强因子同样有效(反向) + assert classify(0.1, 1.8) == "watch" + assert classify(0.0, 0.5) == "eliminated" + + +def test_summarize_factor_structure(): + rng = np.random.default_rng(7) + idx = pd.bdate_range("2024-01-01", periods=60) + base = np.tile(np.arange(10.0, 90.0, 1.0), (60, 1)) + F = pd.DataFrame(base + rng.normal(0, 0.5, base.shape), index=idx) + R1 = pd.DataFrame(-0.001 * base + rng.normal(0, 0.001, base.shape), index=idx, columns=F.columns) + R5, R10 = R1 * 5, R1 * 10 + out = summarize_factor(F, R1, R5, R10) + assert set(out) == {"1", "5", "10", "turnover"} + p1 = out["1"] + for key in ("count", "ic_mean", "ic_std", "icir", "t_stat", "win_rate", + "ls_annual", "monthly_ic", "deciles", "conclusion"): + assert key in p1 + assert p1["count"] == 60 + assert p1["ic_mean"] < 0 # 构造为负相关