sanguo_vnpy_v2/tests/factor/test_tears_data.py

"""Test tears_data - 分层序列序列化(方案A tears 端点数据源)."""
import sys
import os
_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))
if _VNPY_SRC not in sys.path:
    sys.path.insert(0, _VNPY_SRC)

import numpy as np
import pandas as pd
import pytest


# —— 纯函数(不依赖 alphalens,本地必跑) ——

def test_nav_compound():
    from sanguo_factor.tears_data import _nav
    nav = _nav(pd.Series([0.1, -0.2, 0.1]))
    assert list(np.round(nav.values, 6)) == [1.1, 0.88, 0.968]


def test_nav_nan_day_is_flat():
    from sanguo_factor.tears_data import _nav
    nav = _nav(pd.Series([np.nan, 0.1]))
    assert nav.iloc[0] == 1.0
    assert abs(nav.iloc[1] - 1.1) < 1e-9


def test_max_drawdown():
    from sanguo_factor.tears_data import _max_drawdown, _nav
    nav = _nav(pd.Series([0.1, -0.2, 0.1]))
    dd = _max_drawdown(nav)
    assert dd <= 0
    assert abs(dd - (0.88 / 1.1 - 1.0)) < 1e-9


def test_annualized_deoverlaps_period():
    from sanguo_factor.tears_data import _annualized
    s = pd.Series([0.001] * 252)  # 日均 0.001
    assert abs(_annualized(s, 1) - 0.252) < 1e-9
    # 同收益按 5D 重叠口径 → 年化除以 5(去重叠,三周期可比)
    assert abs(_annualized(s, 5) - 0.0504) < 1e-9


def test_monthly_ic_groups_by_month():
    from sanguo_factor.tears_data import _monthly_ic
    idx = pd.to_datetime(["2024-01-05", "2024-01-20", "2024-02-01"])
    g = _monthly_ic(pd.Series([0.1, 0.3, 0.2], index=idx))
    assert [x["month"] for x in g] == ["2024-01", "2024-02"]
    assert abs(g[0]["ic"] - 0.2) < 1e-9
    assert g[1]["ic"] == 0.2


# —— build_tears_data 全链(真实 alphalens 函数) ——

def _merged_data(n_days: int = 40, n_assets: int = 20, seed: int = 7) -> pd.DataFrame:
    """合成 alphalens factor_data:因子与前瞻收益秩正相关(rank IC 显著为正)."""
    rng = np.random.default_rng(seed)
    dates = pd.date_range("2024-01-02", periods=n_days, freq="B", name="date")
    assets = [f"S{i:03d}" for i in range(n_assets)]
    idx = pd.MultiIndex.from_product([dates, assets], names=["date", "asset"])
    factor = rng.normal(size=len(idx))
    df = pd.DataFrame({"factor": factor}, index=idx)
    # 横截面 rank(0~1)驱动前瞻收益 → IC>0;1/5/10D 噪声递减信号不变
    rk = pd.Series(factor, index=idx).groupby(level="date").rank(pct=True).values
    base = (rk - 0.5) * 0.04
    for p in (1, 5, 10):
        df[f"{p}D"] = base + rng.normal(0, 0.01, size=len(idx))
    df["factor_quantile"] = (
        pd.Series(factor, index=idx).groupby(level="date")
        .transform(lambda x: pd.qcut(x, 10, labels=False) + 1)
        .astype(int)
    )
    return df


@pytest.fixture(scope="module")
def built():
    pytest.importorskip("alphalens")
    from sanguo_factor.tears_data import build_tears_data
    return build_tears_data(_merged_data())


def test_build_tears_data_period_keys(built):
    assert set(built["periods"]) == {"1D", "5D", "10D"}


def test_build_tears_data_ic_positive(built):
    """合成因子与收益正相关 → 三周期 IC 均值/ICIR 为正,胜率过半."""
    for p in ("1D", "5D", "10D"):
        d = built["periods"][p]
        assert d["ic_mean"] > 0, p
        assert d["icir"] > 0, p
        assert d["t_stat"] > 2, p
        assert d["win_rate"] > 0.5, p


def test_build_tears_data_series_shapes(built):
    d1 = built["periods"]["1D"]
    n_nav = len(d1["nav_dates"])
    assert n_nav == 40
    assert d1["quantile_keys"] == [str(i) for i in range(1, 11)]
    for q in d1["quantile_keys"]:
        assert len(d1["quantile_nav"][q]) == n_nav, q
    assert len(d1["ls_nav"]) == n_nav
    assert len(d1["ic_dates"]) == d1["count"]
    assert len(d1["ic_dates"]) == len(d1["ic_values"])
    # 净值恒正(首值 = 1+首日收益,同 alphalens cum_returns 口径)
    assert all(v > 0 for v in d1["quantile_nav"]["1"])
    assert all(v > 0 for v in d1["ls_nav"])


def test_build_tears_data_quantile_monotonic(built):
    """信号由 rank 驱动 → Q10 年化 > Q1 年化,多空年化为正."""
    d1 = built["periods"]["1D"]
    assert d1["quantile_annual"]["10"] > d1["quantile_annual"]["1"]
    assert d1["ls_annual"] > 0
    assert d1["ls_max_dd"] <= 0


def test_build_tears_data_monthly_ic_present(built):
    d1 = built["periods"]["1D"]
    assert d1["monthly_ic"], "monthly_ic 不应为空(40 个交易日 ≥ 1 个月)"
    assert "month" in d1["monthly_ic"][0] and "ic" in d1["monthly_ic"][0]


def test_build_tears_data_json_serializable(built):
    """端点要 FileResponse 这个 dict → 必须整棵 json 可序列化."""
    import json
    json.dumps(built)


# —— 离散并列因子走真实 alphalens binning(NAS 2026-08-30 alpha16 实锤回归) ——

def test_discrete_factor_binning_survives():
    """alpha16 实锤形态:cs_rank 族因子截面值高度并列(中位日 25 值仅 6 唯一,
    -19.5×12/-5×9)→ alphalens quantize_factor 的 pd.qcut 不传 duplicates,
    边界必重合 → ValueError 被 no_raise 吞成全 NaN → binning 丢 100% →
    merged 空 → IC 全空 + tears IndexError。tears_data 模块级 qcut
    duplicates='drop' 补丁后:merged 非空、tears 出数、keys 动态短化不崩。"""
    pytest.importorskip("alphalens")
    import sanguo_factor.tears_data  # noqa: F401 — 触发 pd.qcut duplicates='drop' 补丁
    from alphalens.utils import get_clean_factor_and_forward_returns
    from sanguo_factor.tears_data import build_tears_data

    rng = np.random.default_rng(11)
    n_days, n_assets = 40, 25
    dates = pd.date_range("2024-01-02", periods=n_days, freq="B", name="date")
    assets = [f"S{i:03d}" for i in range(n_assets)]
    idx = pd.MultiIndex.from_product([dates, assets], names=["date", "asset"])
    levels = np.array([-19.5, -13.0, -12.0, -11.0, -10.0, -5.0])
    picks = rng.integers(0, len(levels), size=len(idx))
    factor = pd.Series(levels[picks], index=idx, name="factor")

    px = pd.DataFrame(
        100 + np.cumsum(rng.normal(0, 1, size=(n_days, n_assets)), axis=0),
        index=dates,
        columns=assets,
    )

    merged = get_clean_factor_and_forward_returns(
        factor=factor, prices=px, periods=(1,), quantiles=10, max_loss=1.0,
    )
    assert not merged.empty, "离散并列因子不应被 binning 丢光(修复回归)"

    tears = build_tears_data(merged, periods=(1,))
    d = tears["periods"]["1D"]
    assert d["count"] > 0
    assert d["quantile_keys"], "至少应有一组"
    assert len(d["ls_nav"]) == len(d["nav_dates"])
    import json
    json.dumps(tears)