claude_dev
200ea620a0
infra 验收核查发现: 快照只含有质押的股票,全解押股退出名单后事件行 join_asof backward 会把旧值永久 stale 带到全部后续日期——与 detail 族 UNFREEZE 区间重建的解押归零语义直接矛盾(解押后风险解除应拿 0 档, stale 会漏掉改善分)。修=loader 在「最后出现期的下一实际期」插 0 行, 消失次期起归 0;再质押后续真行自然覆盖;期序列=树内实际有数据的期 (树完整假设,failed 期周六班自愈)。测试: 消失股 D 0106 有行 0707 起 消失→次期起 0 永久(修复前 stale 8.0 实锤 RED);既有 C 股断言由 stale 语义勘误为真实在押(0707 补行)。 节假日空周=0 行 0 列 parquet(NAS 实测形态,周六班收口)→ loader 列 检查跳过与「省略文件」同效果,asof 回退节前最近真期——补 0×0 字面 夹具钉死该防线(防未来重构破坏列检查)。全套 1483 绿(+2 用例) [nas] Co-Authored-By: Claude Code <noreply@anthropic.com>
198 lines
9.1 KiB
Python
198 lines
9.1 KiB
Python
# tests/factor/test_fundamental_p1c_adapter.py
|
||
"""P1-C 批财务因子适配层: 研报共识 5 + 质押 4(spec §20.13 联合设计,2026-09-22).
|
||
|
||
口径锚(P1-C 开工包):
|
||
- 研报族 PIT=条目 publishDate;180 天窗两端 cum 口径(同 E11 手法);
|
||
rating_pos 门=窗内研报 ≥3;fwd_ep 门=窗内带 EPS 条目 ≥1(空值不算
|
||
分母);target=最新带目标价条目+180 天新鲜度门(null 目标价不进流
|
||
→ 不遮蔽上一份带价研报)
|
||
- pledge_ratio PIT=周五快照(asof 取 ≤asof 最近期);股无行=0(无质押
|
||
=真实零,与缺域 NaN 语义区分)
|
||
- pledge_detail 区间重建: 生效开始=max(PF_START_DATE, NOTICE_DATE)
|
||
(公告日才可见,宁晚毋早);结束=UNFREEZE_DATE(解押即视为可见——EM 无
|
||
解押公告日列);asof∈[开始,结束) ⇒ 在押;IS_CONTROL_SHAREHOLDER=='1'
|
||
行=实控人质押直标(免股东名 join)
|
||
- pledge_net_180 = 180 天窗(解押笔数 − 新增质押笔数);计数列聚合即
|
||
cast Int64(u32 环回坑,P2 实锤先例)
|
||
"""
|
||
import sys, os
|
||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
|
||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0")))
|
||
|
||
import pytest
|
||
|
||
from sanguo_factor.fundamental_adapter import build_fundamental_features
|
||
|
||
A, B, C, BANK, D = "600000.SSE", "000001.SZSE", "300001.SZSE", "601398.SSE", "600004.SSE"
|
||
|
||
_RES_COLS = ["res_coverage", "res_rating_pos", "res_fwd_eps",
|
||
"res_target_price", "res_rating_chg"]
|
||
_PL_COLS = ["pledge_ratio", "pledge_ctrl", "pledge_warn_line", "pledge_net_180"]
|
||
|
||
|
||
def _dt(day: str):
|
||
import datetime as _d
|
||
return _d.datetime.strptime(day, "%Y-%m-%d")
|
||
|
||
|
||
def val(df, vt: str, day: str, col: str):
|
||
row = df.filter((df["vt_symbol"] == vt) & (df["datetime"] == _dt(day)))
|
||
assert row.height == 1, f"grid 缺行 {vt} {day}"
|
||
v = row[col][0]
|
||
return None if v is None else float(v)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def feat(synthetic_static):
|
||
return build_fundamental_features(
|
||
[A, B, C, BANK, D], "2021-01-01", "2024-06-30", data_dir=synthetic_static,
|
||
columns=_RES_COLS + _PL_COLS)
|
||
|
||
|
||
# ---------- R 族: res_coverage(180 天窗研报数) ----------
|
||
|
||
def test_res_coverage_window(feat):
|
||
# A 事件(窗右开两端差): 07-03 时窗含 e1..e3 → 3;08-29(−180d=03-02,
|
||
# e1@03-10 未出窗) → 3;24-06-30(−180d=2024-01-02,e1..e4 出窗)仅 e5 → 1
|
||
assert val(feat, A, "2023-03-10", "res_coverage") == 1.0
|
||
assert val(feat, A, "2023-07-03", "res_coverage") == 3.0
|
||
assert val(feat, A, "2023-08-29", "res_coverage") == 3.0
|
||
assert val(feat, A, "2024-06-30", "res_coverage") == 1.0
|
||
|
||
|
||
def test_res_coverage_zero_after_slideout(feat):
|
||
# C 单份 2022-06-30: 2023-08-29 窗已滑出 → 0(曾覆盖后窗空=0 值,
|
||
# 与首研报前 NaN 区分)
|
||
assert val(feat, C, "2023-08-29", "res_coverage") == 0.0
|
||
|
||
|
||
def test_res_coverage_nan_before_first(feat):
|
||
assert val(feat, A, "2023-03-09", "res_coverage") is None
|
||
# B 无研报 → 全 NaN(关注缺失≠零关注,与质押族 fill 0 语义相反)
|
||
assert val(feat, B, "2023-12-31", "res_coverage") is None
|
||
|
||
|
||
# ---------- R 族: res_rating_pos(买入+增持占比,门 n≥3) ----------
|
||
|
||
def test_res_rating_pos_value_and_gate(feat):
|
||
assert val(feat, A, "2023-07-03", "res_rating_pos") == pytest.approx(2.0 / 3.0)
|
||
# 门: 窗内 <3 条 → NaN(06-15 窗 e1,e2 共 2 条;24-06-30 窗仅 e5;
|
||
# C 单条)
|
||
assert val(feat, A, "2023-06-15", "res_rating_pos") is None
|
||
assert val(feat, A, "2024-06-30", "res_rating_pos") is None
|
||
assert val(feat, C, "2022-08-29", "res_rating_pos") is None # 窗 1 条
|
||
|
||
|
||
# ---------- R 族: res_fwd_eps(窗内 mean EPS,空值不算分母) ----------
|
||
|
||
def test_res_fwd_eps(feat):
|
||
assert val(feat, A, "2023-03-10", "res_fwd_eps") == 1.0 # 仅 e1
|
||
assert val(feat, A, "2023-07-03", "res_fwd_eps") == pytest.approx(1.05) # (1.0+1.1)/2,e3 无 eps
|
||
assert val(feat, A, "2023-08-29", "res_fwd_eps") == 1.05 # e1 未出窗(−180d=03-02)
|
||
# 24-06-30 窗仅 e5(无 eps) → NaN
|
||
assert val(feat, A, "2024-06-30", "res_fwd_eps") is None
|
||
|
||
|
||
# ---------- R 族: res_target_price(最新可见+180 天新鲜度门) ----------
|
||
|
||
def test_res_target_price_freshness(feat):
|
||
assert val(feat, A, "2023-07-03", "res_target_price") == 18.0
|
||
assert val(feat, A, "2023-08-29", "res_target_price") == 18.0 # e3 距 57 天
|
||
# e4(22.0,2023-11-17)距 2024-06-30=226 天 >180 → 过期 NaN;
|
||
# e5 tgt=None 不进流不遮蔽 → 整列 NaN(门失效会得 22.0)
|
||
assert val(feat, A, "2024-06-30", "res_target_price") is None
|
||
# 首个带价条目(e2 05-12)前 → NaN
|
||
assert val(feat, A, "2023-05-11", "res_target_price") is None
|
||
|
||
|
||
# ---------- R 族: res_rating_chg(窗内净上调次数) ----------
|
||
|
||
def test_res_rating_chg(feat):
|
||
assert val(feat, A, "2023-07-03", "res_rating_chg") == -1.0 # e3 下调
|
||
assert val(feat, A, "2023-08-29", "res_rating_chg") == -1.0
|
||
# 24-06-30 窗仅 e5(维持) → 0
|
||
assert val(feat, A, "2024-06-30", "res_rating_chg") == 0.0
|
||
assert val(feat, A, "2023-11-17", "res_rating_chg") == 0.0 # 窗(e3,e4)下调+上调相抵
|
||
|
||
|
||
# ---------- Q 族: pledge_ratio(周五快照 asof;无行 fill 0) ----------
|
||
|
||
def test_pledge_ratio_snapshot_asof(feat):
|
||
assert val(feat, A, "2023-01-05", "pledge_ratio") == 0.0 # 首期前 fill 0
|
||
assert val(feat, A, "2023-01-06", "pledge_ratio") == 5.0
|
||
assert val(feat, A, "2023-07-06", "pledge_ratio") == 5.0 # ≤asof 最近期=0106
|
||
assert val(feat, A, "2023-07-07", "pledge_ratio") == 6.0
|
||
assert val(feat, C, "2023-07-07", "pledge_ratio") == 20.0
|
||
# B 快照无行(无质押=真实零)
|
||
assert val(feat, B, "2023-12-31", "pledge_ratio") == 0.0
|
||
|
||
|
||
def test_pledge_ratio_disappearing_stock_fresh_zero(feat):
|
||
"""全解押股退出快照名单 → 下一期起归 0(非 join_asof 永久 stale 旧值).
|
||
|
||
快照只含有质押的股票,D 0106 期有行(8.0) 0707 期起消失=全解押;
|
||
若无消失修复,asof 会把 0106 的 8.0 带到全部后续日期(与 detail 族
|
||
区间重建的解押语义矛盾——解押后风险解除应拿 0 档).
|
||
"""
|
||
assert val(feat, D, "2023-01-06", "pledge_ratio") == 8.0
|
||
assert val(feat, D, "2023-07-06", "pledge_ratio") == 8.0 # 仍在押
|
||
assert val(feat, D, "2023-07-07", "pledge_ratio") == 0.0 # 消失次期归 0
|
||
assert val(feat, D, "2024-06-30", "pledge_ratio") == 0.0 # 永久 0 非 stale
|
||
|
||
|
||
def test_pledge_ratio_empty_holiday_week_file_skipped(feat):
|
||
"""0 行 0 列节假日空周文件(NAS 实测形态)→ loader 列检查跳过,
|
||
asof 回退最近真期(0707),非 NaN 非误 0."""
|
||
assert val(feat, A, "2023-10-06", "pledge_ratio") == 6.0
|
||
assert val(feat, A, "2023-12-31", "pledge_ratio") == 6.0 # 空文件后仍回退 0707
|
||
|
||
|
||
# ---------- Q 族: pledge_ctrl(实控人在押 max 占所持;区间重建) ----------
|
||
|
||
def test_pledge_ctrl_interval_rebuild(feat):
|
||
# p1 生效=max(start 02-10, notice 02-15)=02-15(PIT: 公告日才可见)
|
||
assert val(feat, A, "2023-02-14", "pledge_ctrl") == 0.0
|
||
assert val(feat, A, "2023-02-15", "pledge_ctrl") == 30.0
|
||
# 05-20 起 p1+p2 在押 → 实控人 max(30,50)=50
|
||
assert val(feat, A, "2023-06-30", "pledge_ctrl") == 50.0
|
||
# p2 09-10 解押 → 实控人在押仅 p1 → 30
|
||
assert val(feat, A, "2023-10-01", "pledge_ctrl") == 30.0
|
||
# C 无实控人质押行 → 0
|
||
assert val(feat, C, "2023-12-31", "pledge_ctrl") == 0.0
|
||
assert val(feat, B, "2023-12-31", "pledge_ctrl") == 0.0
|
||
|
||
|
||
# ---------- Q 族: pledge_warn_line(在押最高预估平仓线) ----------
|
||
|
||
def test_pledge_warn_line(feat):
|
||
assert val(feat, A, "2023-02-14", "pledge_warn_line") == 0.0
|
||
assert val(feat, A, "2023-06-30", "pledge_warn_line") == 25.0 # max(15,25)
|
||
assert val(feat, A, "2023-10-01", "pledge_warn_line") == 15.0 # p2 解押 → max(15,8)
|
||
assert val(feat, C, "2023-01-01", "pledge_warn_line") == 12.0 # 2021 全程在押
|
||
|
||
|
||
# ---------- Q 族: pledge_net_180(180 天窗解押−新增笔数) ----------
|
||
|
||
def test_pledge_net_chg(feat):
|
||
# A 事件: 02-15 新增 / 05-20 新增 / 08-01 新增 / 09-10 解押
|
||
assert val(feat, A, "2023-02-14", "pledge_net_180") == 0.0 # 首事件前 fill 0
|
||
assert val(feat, A, "2023-02-15", "pledge_net_180") == -1.0
|
||
assert val(feat, A, "2023-06-30", "pledge_net_180") == -2.0 # 新增 p1+p2
|
||
# 10-01 窗 (2023-04-04,10-01]: 新增 p3 − 解押 p2 相抵,但 180 天前
|
||
# 端 cum=−1(至 02-15) → 两端差 = −2 −(−1) = −1
|
||
assert val(feat, A, "2023-10-01", "pledge_net_180") == -1.0
|
||
# C 2021 事件早已出 180 天窗 → 0;B 无质押史 → 0
|
||
assert val(feat, C, "2023-01-01", "pledge_net_180") == 0.0
|
||
assert val(feat, B, "2023-12-31", "pledge_net_180") == 0.0
|
||
|
||
|
||
# ---------- 引用瘦身路径(仅 P1-C 域列请求) ----------
|
||
|
||
def test_p1c_columns_subset(synthetic_static):
|
||
df = build_fundamental_features(
|
||
[A], "2023-06-01", "2023-12-31", data_dir=synthetic_static,
|
||
columns=["pledge_ctrl", "res_coverage"])
|
||
assert df.columns == ["vt_symbol", "datetime", "pledge_ctrl", "res_coverage"]
|
||
assert val(df, A, "2023-06-30", "pledge_ctrl") == 50.0
|
||
assert val(df, A, "2023-07-03", "res_coverage") == 3.0
|