155 lines
6.9 KiB
Python
155 lines
6.9 KiB
Python
# tests/factor/test_composite_rolling.py
|
||
"""合成层 v2c(Rolling): 滚动制度因子——档案扫描/命名/0 权项过滤/失败留痕/幂等重建/回归."""
|
||
import importlib
|
||
import json
|
||
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 composite_library, composite_weighting, composite_rolling
|
||
from sanguo_factor.composite_weighting import (
|
||
WeightProfile, load_profile, resolve_weights, register_composite_from_profile,
|
||
)
|
||
from sanguo_factor.registry import get_factor, _REGISTRY
|
||
|
||
# 12 滚动档案 × scope(quant12/all18) × 年(2021-2026);排序稳定 = (scope 字典序, year 升序)
|
||
EXPECTED_NAMES = ([f"composite_v2c_all18_{y}" for y in range(2021, 2027)]
|
||
+ [f"composite_v2c_quant12_{y}" for y in range(2021, 2027)])
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _ensure_v2c_registered():
|
||
"""其它测试模块清空 _REGISTRY 后幂等重挂(与 test_composite_weighting 同模式)."""
|
||
composite_rolling._register_all()
|
||
|
||
|
||
def _split_top_terms(expr: str) -> list[str]:
|
||
"""按括号深度 0 切分合成表达式顶层加法项(term 内部可含 + 与括号)."""
|
||
body = expr[expr.index("(") + 1: expr.rindex(") / ")]
|
||
parts, depth, start, i = [], 0, 0, 0
|
||
while i < len(body):
|
||
if body[i] == "(":
|
||
depth += 1
|
||
elif body[i] == ")":
|
||
depth -= 1
|
||
elif depth == 0 and body.startswith(" + ", i):
|
||
parts.append(body[start:i])
|
||
start = i + 3
|
||
i += 3
|
||
continue
|
||
i += 1
|
||
parts.append(body[start:])
|
||
return parts
|
||
|
||
|
||
# ==================== 档案扫描与命名 ====================
|
||
|
||
def test_iter_rolling_profiles_discovers_12_stable_order():
|
||
"""扫描 weight_profiles/ 认 12 个 {scope}_roll_{year}.json,排序 (scope, year) 稳定;
|
||
v2a/v2b 等非约定名文件不纳入."""
|
||
got = composite_rolling._iter_rolling_profiles()
|
||
assert [name for name, _p in got] == EXPECTED_NAMES
|
||
assert all(p.name.endswith(".json") and "_roll_" in p.name for _n, p in got)
|
||
|
||
|
||
def test_all_12_rolling_factors_registered_composite():
|
||
"""导入即注册: 12 个 composite_v2c_* 因子全在 registry,category=composite."""
|
||
for name in EXPECTED_NAMES:
|
||
fac = get_factor(name)
|
||
assert fac is not None, f"{name} 未注册"
|
||
assert fac["category"] == "composite"
|
||
|
||
|
||
# ==================== 0 权项过滤(表达式项数) ====================
|
||
|
||
def test_quant12_rolling_expression_exact_12_terms():
|
||
"""quant12 滚动档案权重全正: 表达式恰 12 项,无 0.000000 项."""
|
||
expr = get_factor("composite_v2c_quant12_2021")["expression"]
|
||
assert len(_split_top_terms(expr)) == 12
|
||
assert "0.000000 * " not in expr
|
||
|
||
|
||
def test_all18_2021_drops_zero_weight_neu_terms():
|
||
"""all18 2021 制度 6 个 neu 源 0 权被跳过: 表达式恰 12 项,neu 源因子串不出现."""
|
||
expr = get_factor("composite_v2c_all18_2021")["expression"]
|
||
assert len(_split_top_terms(expr)) == 12
|
||
assert "0.000000 * " not in expr
|
||
profile = load_profile(str(next(
|
||
p for n, p in composite_rolling._iter_rolling_profiles()
|
||
if n == "composite_v2c_all18_2021")))
|
||
zero_srcs = [s for s, (_d, w) in profile.sources.items() if w <= 0]
|
||
assert len(zero_srcs) == 6 # 设计行为: 2021 neu 全 0
|
||
for src in zero_srcs:
|
||
assert get_factor(src)["expression"] not in expr, f"0 权源 {src} 不应入式"
|
||
|
||
|
||
def test_all_rolling_terms_count_matches_positive_weights():
|
||
"""派生断言(不钉死档案数值): 12 份滚动档案顶层项数 == 正权源数,
|
||
每个正权源以 6 位定点权重入式(all18_2022..2024 含零星 0 权 neu 源同样被跳过)."""
|
||
for name, path in composite_rolling._iter_rolling_profiles():
|
||
weights = resolve_weights(load_profile(str(path)))
|
||
positive = {s: w for s, w in weights.items() if w > 0}
|
||
expr = get_factor(name)["expression"]
|
||
terms = _split_top_terms(expr)
|
||
assert len(terms) == len(positive), f"{name} 项数 {len(terms)} != 正权源数 {len(positive)}"
|
||
for src, w in positive.items():
|
||
assert any(t.startswith(f"{w:.6f} * (") for t in terms), \
|
||
f"{name} 缺 {src} 权重项"
|
||
|
||
|
||
# ==================== 全零防线与失败留痕 ====================
|
||
|
||
def test_register_all_zero_profile_raises_value_error():
|
||
"""全零档案非法: 直接构造 WeightProfile(L1 全 0,绕过 load 校验)注册 → ValueError,
|
||
且不注册(可读错误优于 ZeroDivisionError)."""
|
||
name = "composite_v2c_zerotest"
|
||
_REGISTRY.pop(name, None)
|
||
sources = {n: (d, 0.0) for n, d in composite_library.QUANT_SOURCES}
|
||
with pytest.raises(ValueError, match="全零"):
|
||
register_composite_from_profile(
|
||
WeightProfile("t-zero", "t", "manual", sources, (), {}), name)
|
||
assert get_factor(name) is None
|
||
|
||
|
||
def test_register_failure_message_carries_filename(tmp_path, monkeypatch):
|
||
"""单档案加载失败 raise 且带文件名(防静默丢制度期)."""
|
||
bad = tmp_path / "quant12_roll_2020.json"
|
||
bad.write_text(json.dumps({"profile_id": "broken", "name": "t", "method": "m"}),
|
||
encoding="utf-8") # 缺 sources
|
||
monkeypatch.setattr(composite_rolling, "_iter_rolling_profiles",
|
||
lambda: [("composite_v2c_quant12_2020", bad)])
|
||
with pytest.raises(ValueError, match="quant12_roll_2020.json"):
|
||
composite_rolling._register_all()
|
||
|
||
|
||
# ==================== 幂等重建与回归 ====================
|
||
|
||
def test_registry_cleared_reload_rebuilds_independently():
|
||
"""registry 清空后 reload composite_rolling 可独立重建: v2c 12 个 +
|
||
依赖链(v1.1 源/neu/v2a/v2b)全部重挂(幂等)."""
|
||
_REGISTRY.clear()
|
||
assert get_factor("composite_v2c_quant12_2024") is None
|
||
importlib.reload(composite_rolling)
|
||
for name in EXPECTED_NAMES:
|
||
assert get_factor(name) is not None, f"{name} 未重建"
|
||
for dep in ("alpha16", "fund_bp_neu",
|
||
composite_weighting._DEFAULT_COMPOSITE_NAME,
|
||
composite_weighting._V2B_COMPOSITE_NAME):
|
||
assert get_factor(dep) is not None, f"依赖 {dep} 未重挂"
|
||
|
||
|
||
def test_v2a_v2b_default_factors_unaffected():
|
||
"""回归: v2c 上线后既有 v2a/v2b 默认因子不变(全正权档案,项数 12/18,权重入式)."""
|
||
v2a = get_factor(composite_weighting._DEFAULT_COMPOSITE_NAME)
|
||
assert v2a is not None
|
||
assert len(_split_top_terms(v2a["expression"])) == 12
|
||
v2b = get_factor(composite_weighting._V2B_COMPOSITE_NAME)
|
||
assert v2b is not None
|
||
assert len(_split_top_terms(v2b["expression"])) == 18
|
||
weights = resolve_weights(
|
||
load_profile(str(composite_weighting._DEFAULT_PROFILE_PATH)))
|
||
for src, w in weights.items():
|
||
assert f"{w:.6f} * (" in v2a["expression"], f"v2a 缺 {src} 项"
|