claude_dev
0dd81d709f
①后端tears序列化:sanguo_factor/tears_data.py新模块,alphalens已算分层序列(日度IC/月度聚合/十分组累计净值/多空Q10−Q1净值+最大回撤/去重叠年化/因子秩自相关)在分析时序列化为{factor}_tears.json,与tearsheet同源;FactorReport加tears_paths,GET /task/{id}/tears/{factor}端点(token header),_persist_factor同步落DB
②前端tears页:TearsPanel.vue按设计稿tab①——指标条7格+月度IC柱(红正绿负)/累计IC线双轴+月度IC热力图(年×月,CSS格)+分组累计净值Q1/Q5/Q10+多空净值(琥珀+面积+○最大回撤标注)+十分组年化(±5%虚线)+IC衰减(1/5/10D),1/5/10D全页联动;Result.vue的iframe→原生渲染;旧任务404自动回退iframe旧alphalens报告
③加入对比(设计稿tab②纯前端):factorCompare store(localStorage持久化,2~6个)+排行榜行内「＋对比/✓已选」列+详情页死按钮做实(选中青色态)+全局底部托盘CompareTray(chips可删/清空/对比N因子→)+对比页Compare.vue(指标并排·行最优青色高亮/累计IC叠加多线/月度IC序列Pearson相关性矩阵前端算/十分组小倍数SVG)
测试:tears_data纯函数5+全链真实alphalens6(合成因子IC>0/分层单调/JSON可序列化)+analyzer写盘/容错2+端点401/404/200共3;factor+api+orchestrator 317全绿;npm run build绿
130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
"""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)
|