020e6104da
产物 attribution_<end>.json: factorContribution schema verbatim(代表行+ market/size/industry/specific 三桶殿后)+families/rolling/diagnostics 同胞 字段; 原子写; CLI 独立跑位。P1/P1B 计数断言 +1(size_log 辅助列); attribution 测试幂等自挂(先序 _REGISTRY.clear 防污染, 09-22 routes 同款)。 设计档 §7.5 组件表加 attribution.py 行(档随码走同 push)。 Co-Authored-By: Claude Code <noreply@anthropic.com>
74 lines
3.3 KiB
Python
74 lines
3.3 KiB
Python
# tests/factor/test_fundamental_p1b_library.py
|
||
"""P1-B 批财务因子表达式库: 四新域 6 个注册(5 因子,D14 双注册)契约锁定.
|
||
|
||
契约:
|
||
- 表达式裸标识符 ⊆ adapter 特征列 + close/log/cs_rank
|
||
- 全部 cs_rank 一层截面;E08/E12 负 IC 取负,E11 不定按正向注册待实证
|
||
- E13/D14 正向;vb 两因子直接消费 vb 域日频列(不经 close×share_capital)
|
||
"""
|
||
import re
|
||
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 import fundamental_library # noqa: F401 import 即注册
|
||
from sanguo_factor.fundamental_adapter import FEATURE_COLUMNS
|
||
from sanguo_factor.fundamental_library import FUNDAMENTAL_P1B_FACTORS
|
||
from sanguo_factor.fundamental_neutralize import FUND_NEU_SOURCES # noqa: F401 v2b 计数用
|
||
from sanguo_factor.registry import list_factors, get_factor
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _ensure_fundamental_registered():
|
||
"""其它测试模块清空 _REGISTRY 后逐测试幂等重注册(顺序依赖前科)."""
|
||
fundamental_library._register_all()
|
||
|
||
|
||
_ALLOWED = set(FEATURE_COLUMNS) | {"close", "cs_rank", "log"}
|
||
|
||
|
||
def test_p1b_6_factors_registered():
|
||
facs = {f["name"]: f for f in list_factors("fundamental")}
|
||
names = {n for n, _e, _d, _ic in FUNDAMENTAL_P1B_FACTORS}
|
||
assert len(names) == 6, f"P1-B 应 6 个注册(5 因子,D14 双注册),实际 {len(names)}"
|
||
for name in names:
|
||
assert name in facs, f"{name} 未注册"
|
||
assert facs[name]["category"] == "fundamental"
|
||
|
||
|
||
def test_p1b_expressions_only_reference_feature_columns():
|
||
for name, expression, _d, _ic in FUNDAMENTAL_P1B_FACTORS:
|
||
idents = set(re.findall(r"[A-Za-z_][A-Za-z0-9_]*", expression))
|
||
bad = idents - _ALLOWED
|
||
assert not bad, f"{name} 引用了未产出列: {bad} in {expression}"
|
||
|
||
|
||
def test_p1b_all_cross_sectional_rank_one_layer():
|
||
for name, expression, _d, _ic in FUNDAMENTAL_P1B_FACTORS:
|
||
assert expression.startswith("cs_rank("), f"{name}: {expression}"
|
||
assert expression.endswith(")")
|
||
|
||
|
||
def test_p1b_negative_ic_factors_flipped():
|
||
"""E08(CCC 低=效率高)/E12(户数降=筹码集中)取负;E13/D14/E11 正向."""
|
||
assert get_factor("fund_ccc")["expression"] == "cs_rank(-ccc)"
|
||
assert get_factor("fund_gdhs_chg")["expression"] == "cs_rank(-gdhs_chg)"
|
||
assert get_factor("fund_topholder_chg")["expression"] == "cs_rank(topholder_chg)"
|
||
# E11 方向不定(公告前+/除权后−混合)→ 正向注册,由 IC 实测定去留
|
||
assert get_factor("fund_send_total_12m")["expression"] == "cs_rank(send_total_12m)"
|
||
|
||
|
||
def test_p1b_vb_factors_consume_vb_columns():
|
||
"""D14 双注册直接消费 vb 域列(长史 1990 起含退市,分域验证用)."""
|
||
assert get_factor("fund_ep_vb")["expression"] == "cs_rank(ep_vb)"
|
||
assert get_factor("fund_bp_vb")["expression"] == "cs_rank(bp_vb)"
|
||
|
||
|
||
def test_p1b_registration_idempotent():
|
||
fundamental_library._register_all()
|
||
fundamental_library._register_all()
|
||
# P0 32 + P1 37 + P1-B 6 + P2 5 = 80;v2b 另挂 6 个 fund_*_neu(fundamental 类)
|
||
assert len(list_factors("fundamental")) == 89 + len(FUND_NEU_SOURCES) + 1 # +1=size_log 辅助暴露列(P3 §4.7)
|