187 lines
7.8 KiB
Python
187 lines
7.8 KiB
Python
# tests/factor/test_composite_weighting.py
|
|
"""合成层 v2a(Weighting): 档案加载校验 + L2 覆盖归一 + 表达式拼接 + 注册幂等 + v1.1 回归."""
|
|
import dataclasses
|
|
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
|
|
from sanguo_factor.composite_library import rank_term
|
|
from sanguo_factor.composite_weighting import (
|
|
WeightProfile, load_profile, resolve_weights, build_weighted_expression,
|
|
register_composite_from_profile,
|
|
)
|
|
from sanguo_factor.registry import get_factor
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _ensure_v2a_registered():
|
|
"""其它测试模块清空 _REGISTRY 后幂等重挂源+v2a(与 test_composite_library 同模式)."""
|
|
composite_weighting._register_all()
|
|
|
|
|
|
# ==================== L1 档案加载与校验 ====================
|
|
|
|
def _base_profile_dict() -> dict:
|
|
"""2 源小例档案骨架(load/resolve 测试用,不依赖 12 源表;违例由调用方注入)."""
|
|
return {
|
|
"profile_id": "t-2src", "name": "t 2源", "method": "manual",
|
|
"sources": {
|
|
"a": {"direction": "+", "fit_icir": 1.0, "weight": 0.6},
|
|
"b": {"direction": "-", "fit_icir": 0.5, "weight": 0.4},
|
|
},
|
|
"overrides": [],
|
|
}
|
|
|
|
|
|
def _write_profile(tmp_path, profile: dict) -> str:
|
|
"""写临时档案 JSON(测试自建,不碰真档案)."""
|
|
p = tmp_path / "profile.json"
|
|
p.write_text(json.dumps(profile, ensure_ascii=False), encoding="utf-8")
|
|
return str(p)
|
|
|
|
|
|
def test_load_real_archive_directions_match_v11_source_table():
|
|
"""真档案(只读): 12 源方向表与 QUANT_SOURCES 逐对一致,权重全正且和 ≈1."""
|
|
profile = load_profile(str(composite_weighting._DEFAULT_PROFILE_PATH))
|
|
assert profile.profile_id == "quant12-icirfit-v1"
|
|
assert len(profile.sources) == 12
|
|
assert profile.overrides == ()
|
|
declared = {src: d for src, (d, _w) in profile.sources.items()}
|
|
assert declared == dict(composite_library.QUANT_SOURCES)
|
|
weights = [w for _d, w in profile.sources.values()]
|
|
assert min(weights) > 0
|
|
assert sum(weights) == pytest.approx(1.0)
|
|
|
|
|
|
def test_load_real_archive_metadata_kept(tmp_path):
|
|
"""真档案(只读): 档案其余字段(fit_window/baseline)归入 metadata 不丢失."""
|
|
profile = load_profile(str(composite_weighting._DEFAULT_PROFILE_PATH))
|
|
assert profile.method == "icir_fit"
|
|
assert profile.metadata["fit_window"]["start"] == "2018-01"
|
|
assert profile.metadata["baseline"]["h1_icir"] == 0.885
|
|
assert "sources" not in profile.metadata # 已提字段不重复保留
|
|
|
|
|
|
def test_load_rejects_negative_weight(tmp_path):
|
|
bad = _base_profile_dict()
|
|
bad["sources"]["a"]["weight"] = -0.2
|
|
with pytest.raises(ValueError, match="权重"):
|
|
load_profile(_write_profile(tmp_path, bad))
|
|
|
|
|
|
def test_load_rejects_all_zero_weights(tmp_path):
|
|
bad = _base_profile_dict()
|
|
bad["sources"]["a"]["weight"] = 0
|
|
bad["sources"]["b"]["weight"] = 0
|
|
with pytest.raises(ValueError, match="> 0"):
|
|
load_profile(_write_profile(tmp_path, bad))
|
|
|
|
|
|
def test_load_rejects_unknown_override_source(tmp_path):
|
|
bad = _base_profile_dict()
|
|
bad["overrides"] = [{"source": "ghost", "weight": 0.1, "reason": "盘感"}]
|
|
with pytest.raises(ValueError, match="ghost"):
|
|
load_profile(_write_profile(tmp_path, bad))
|
|
|
|
|
|
def test_load_rejects_empty_override_reason(tmp_path):
|
|
bad = _base_profile_dict()
|
|
bad["overrides"] = [{"source": "a", "weight": 0.1, "reason": " "}]
|
|
with pytest.raises(ValueError, match="reason"):
|
|
load_profile(_write_profile(tmp_path, bad))
|
|
|
|
|
|
def test_load_rejects_negative_override_weight(tmp_path):
|
|
bad = _base_profile_dict()
|
|
bad["overrides"] = [{"source": "a", "weight": -0.1, "reason": "盘感"}]
|
|
with pytest.raises(ValueError, match="override"):
|
|
load_profile(_write_profile(tmp_path, bad))
|
|
|
|
|
|
# ==================== L2 覆盖层 resolve ====================
|
|
|
|
def _profile(sources: dict, overrides: tuple = ()) -> WeightProfile:
|
|
return WeightProfile("t", "t", "manual", dict(sources), tuple(overrides), {})
|
|
|
|
|
|
def test_resolve_without_override_equals_l1():
|
|
"""无覆盖 = L1 原样(仅浮点归一误差)."""
|
|
got = resolve_weights(_profile({"a": ("+", 0.6), "b": ("-", 0.4)}))
|
|
assert got == pytest.approx({"a": 0.6, "b": 0.4})
|
|
assert sum(got.values()) == pytest.approx(1.0)
|
|
|
|
|
|
def test_resolve_override_replaces_and_renormalizes():
|
|
"""有覆盖: 替换对应源权重后全组重归一 sum=1,未覆盖源保持 L1 相对结构."""
|
|
profile = _profile(
|
|
{"a": ("+", 0.6), "b": ("-", 0.4)},
|
|
overrides=({"source": "a", "weight": 0.2, "reason": "盘感降权"},),
|
|
)
|
|
got = resolve_weights(profile)
|
|
assert got["a"] == pytest.approx(1 / 3) # 0.2 / (0.2+0.4)
|
|
assert got["b"] == pytest.approx(2 / 3) # 未覆盖,按 L1 结构分享余量
|
|
assert sum(got.values()) == pytest.approx(1.0)
|
|
|
|
|
|
# ==================== L3 表达式拼接 ====================
|
|
|
|
def test_build_weighted_expression_exact_string():
|
|
"""拼接形态钉死: (w1*(项1) + w2*(项2)) / Σw,权重与和均 6 位定点."""
|
|
terms = [("cs_rank((close))", 0.25), ("cs_rank((volume))", 0.75)]
|
|
assert build_weighted_expression(terms) == \
|
|
"(0.250000 * (cs_rank((close))) + 0.750000 * (cs_rank((volume)))) / 1.000000"
|
|
|
|
|
|
def test_build_weighted_expression_denominator_is_weight_sum():
|
|
"""分母 = 权重和(非 1 归一假设): 0.2+0.3 → / 0.500000."""
|
|
assert build_weighted_expression([("x", 0.2), ("y", 0.3)]).endswith("/ 0.500000")
|
|
|
|
|
|
# ==================== 注册与 v1.1 回归 ====================
|
|
|
|
def test_v2a_registered_with_all_12_sources_and_profile_weights():
|
|
"""composite_quant12_v2a 已注册: composite 类,12 源项齐且权重值=档案 resolve 结果."""
|
|
fac = get_factor("composite_quant12_v2a")
|
|
assert fac is not None and fac["category"] == "composite"
|
|
weights = resolve_weights(
|
|
load_profile(str(composite_weighting._DEFAULT_PROFILE_PATH)))
|
|
expr = fac["expression"]
|
|
for src, direction in composite_library.QUANT_SOURCES:
|
|
term = rank_term(get_factor(src)["expression"], direction)
|
|
assert f"{weights[src]:.6f} * ({term})" in expr, \
|
|
f"{src} 项缺失或权重不符"
|
|
assert expr.endswith(f") / {sum(weights.values()):.6f}")
|
|
|
|
|
|
def test_register_rejects_direction_mismatch():
|
|
"""档案方向表与 QUANT_SOURCES 不一致 → ValueError,且不注册."""
|
|
real = load_profile(str(composite_weighting._DEFAULT_PROFILE_PATH))
|
|
flipped = dict(real.sources)
|
|
src0, (d0, w0) = next(iter(flipped.items()))
|
|
flipped[src0] = ("-" if d0 == "+" else "+", w0)
|
|
bad = dataclasses.replace(real, sources=flipped)
|
|
with pytest.raises(ValueError, match="QUANT_SOURCES"):
|
|
register_composite_from_profile(bad, "composite_quant12_v2a_bad")
|
|
assert get_factor("composite_quant12_v2a_bad") is None
|
|
|
|
|
|
def test_register_idempotent_double_call_keeps_expression():
|
|
"""重复注册不炸: 名字已注册直接跳过,表达式不变."""
|
|
before = get_factor("composite_quant12_v2a")["expression"]
|
|
profile = load_profile(str(composite_weighting._DEFAULT_PROFILE_PATH))
|
|
register_composite_from_profile(profile, "composite_quant12_v2a")
|
|
assert get_factor("composite_quant12_v2a")["expression"] == before
|
|
|
|
|
|
def test_v11_regression_quant12_equal_weight_unchanged():
|
|
"""v1.1 回归: 导入 v2a 后 composite_quant12 仍 == 独立重建的等权式(零影响)."""
|
|
expected = composite_library.build_composite_expression(
|
|
[rank_term(get_factor(n)["expression"], d)
|
|
for n, d in composite_library.QUANT_SOURCES]
|
|
)
|
|
assert get_factor("composite_quant12")["expression"] == expected
|