392 lines
16 KiB
Python
392 lines
16 KiB
Python
# tests/factor/test_version_registry.py
|
|
"""版本注册表 TDD——spec §4.2 决议 D(版本化)/E(成熟度门墓园侧)."""
|
|
import json
|
|
import os
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from sanguo_factor.version_registry import (
|
|
VALID_STATUSES,
|
|
add_version,
|
|
append_event,
|
|
effective_version,
|
|
load_registry,
|
|
main,
|
|
open_gitea_issue,
|
|
render_promotion_issue,
|
|
save_registry,
|
|
transition,
|
|
upsert_factor,
|
|
writeback_card_death,
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def reg():
|
|
return {"factors": {}}
|
|
|
|
|
|
def _mk(reg, name="pledge_net_chg", hyp="H-2026-09-20-03"):
|
|
return upsert_factor(reg, name, hypothesis=hyp)
|
|
|
|
|
|
def test_load_missing_path_gives_empty(tmp_path):
|
|
got = load_registry(str(tmp_path / "nope.yaml"))
|
|
assert got == {"factors": {}}
|
|
|
|
|
|
def test_save_load_roundtrip(tmp_path, reg):
|
|
_mk(reg)
|
|
p = str(tmp_path / "factor_registry.yaml")
|
|
save_registry(p, reg)
|
|
assert load_registry(p) == reg
|
|
assert not [f for f in os.listdir(tmp_path) if f.endswith(".tmp")] # 原子写无残留
|
|
|
|
|
|
def test_upsert_default_incubating_and_idempotent(reg):
|
|
e1 = _mk(reg)
|
|
assert e1["status"] == "incubating"
|
|
e2 = upsert_factor(reg, "pledge_net_chg", hypothesis="H-2026-09-20-03")
|
|
assert e2 is e1 # 同名同假设幂等返回既有条目
|
|
|
|
|
|
def test_upsert_same_name_diff_hypothesis_raises(reg):
|
|
_mk(reg)
|
|
with pytest.raises(ValueError, match="假设不同"):
|
|
upsert_factor(reg, "pledge_net_chg", hypothesis="H-OTHER")
|
|
|
|
|
|
def test_transition_chain_and_skip_level(reg):
|
|
_mk(reg)
|
|
with pytest.raises(ValueError, match="非法状态迁移"):
|
|
transition(reg, "pledge_net_chg", "promoted") # incubating 不可跳级
|
|
transition(reg, "pledge_net_chg", "assessable")
|
|
transition(reg, "pledge_net_chg", "promoted", promoted_at="2026-09-22", promotion_t=4.01)
|
|
e = reg["factors"]["pledge_net_chg"]
|
|
assert e["status"] == "promoted" and e["promotion_t"] == 4.01
|
|
transition(reg, "pledge_net_chg", "decaying")
|
|
transition(reg, "pledge_net_chg", "retired")
|
|
|
|
|
|
def test_graveyard_needs_cause_and_is_terminal(reg):
|
|
_mk(reg)
|
|
transition(reg, "pledge_net_chg", "assessable")
|
|
with pytest.raises(ValueError, match="cause"):
|
|
transition(reg, "pledge_net_chg", "graveyard") # 缺死因被拒
|
|
transition(reg, "pledge_net_chg", "graveyard",
|
|
cause="IC 反号+12 因子共线 0.961", evidence_run="ev_20260921_120000_ab12")
|
|
with pytest.raises(ValueError, match="墓园不可"):
|
|
transition(reg, "pledge_net_chg", "assessable") # 墓园不可复活
|
|
|
|
|
|
def test_add_version_closes_previous_window(reg):
|
|
_mk(reg)
|
|
add_version(reg, "pledge_net_chg", commit="574808c",
|
|
params={"window": 180, "source": "pledge_detail"}, effective_from="2026-09-22")
|
|
add_version(reg, "pledge_net_chg", commit="aaaaaaa",
|
|
params={"window": 120}, effective_from="2026-10-15")
|
|
vs = reg["factors"]["pledge_net_chg"]["versions"]
|
|
assert vs[0]["effective_to"] == "2026-10-15" and vs[1]["effective_to"] is None
|
|
|
|
|
|
def test_effective_version_by_date(reg):
|
|
_mk(reg)
|
|
add_version(reg, "pledge_net_chg", commit="c1", params={}, effective_from="2026-09-22")
|
|
add_version(reg, "pledge_net_chg", commit="c2", params={}, effective_from="2026-10-15")
|
|
f = reg["factors"]["pledge_net_chg"]
|
|
assert effective_version(f, "2026-09-30")["commit"] == "c1"
|
|
assert effective_version(f, "2026-10-15")["commit"] == "c2"
|
|
assert effective_version(f, "2026-01-01") is None
|
|
|
|
|
|
def test_valid_statuses_complete():
|
|
assert VALID_STATUSES == frozenset(
|
|
{"incubating", "assessable", "promoted", "decaying", "retired", "graveyard"})
|
|
|
|
|
|
def test_append_event_jsonl(tmp_path, reg):
|
|
_mk(reg)
|
|
transition(reg, "pledge_net_chg", "assessable")
|
|
ev = append_event(str(tmp_path / "ev.jsonl"),
|
|
reg["factors"]["pledge_net_chg"], "incubating", "assessable")
|
|
assert ev["new"] == "assessable" and ev["name"] == "pledge_net_chg"
|
|
with open(tmp_path / "ev.jsonl") as f:
|
|
lines = [json.loads(x) for x in f if x.strip()]
|
|
assert len(lines) == 1
|
|
|
|
|
|
def test_render_promotion_issue_downstream_todos(reg):
|
|
_mk(reg)
|
|
transition(reg, "pledge_net_chg", "assessable")
|
|
transition(reg, "pledge_net_chg", "promoted", promoted_at="2026-09-22",
|
|
promotion_t=4.01)
|
|
title, body = render_promotion_issue(
|
|
"pledge_net_chg", reg["factors"]["pledge_net_chg"])
|
|
assert "合成层跟进" in title and "孵化梯" in body and "4.01" in body
|
|
|
|
|
|
def test_cli_transition_promotes_and_emits(tmp_path, reg):
|
|
_mk(reg)
|
|
transition(reg, "pledge_net_chg", "assessable")
|
|
p = str(tmp_path / "reg.yaml")
|
|
save_registry(p, reg)
|
|
rc = main(["transition", "pledge_net_chg", "promoted", "--registry", p,
|
|
"--events", str(tmp_path / "ev.jsonl"),
|
|
"--promoted-at", "2026-09-22", "--promotion-t", "4.01",
|
|
"--issue-dir", str(tmp_path / "issues")])
|
|
assert rc == 0
|
|
assert load_registry(p)["factors"]["pledge_net_chg"]["status"] == "promoted"
|
|
assert (tmp_path / "issues" / "pledge_net_chg_promoted_issue.md").exists()
|
|
with open(tmp_path / "ev.jsonl") as f:
|
|
assert json.loads(f.readline())["new"] == "promoted"
|
|
|
|
|
|
def test_cli_graveyard_without_cause_returns_1(tmp_path, reg):
|
|
_mk(reg)
|
|
transition(reg, "pledge_net_chg", "assessable")
|
|
p = str(tmp_path / "reg.yaml")
|
|
save_registry(p, reg)
|
|
rc = main(["transition", "pledge_net_chg", "graveyard", "--registry", p,
|
|
"--events", str(tmp_path / "ev.jsonl")])
|
|
assert rc == 1 # 缺死因被状态机拒,CLI 不 traceback
|
|
|
|
|
|
|
|
def test_open_gitea_issue_base_url_env_and_param(monkeypatch):
|
|
"""open_gitea_issue 地址三优先级:显式参数 > env SANGUO_GITEA_BASE_URL > 内网缺省.
|
|
|
|
与 strategy_registry.open_gitea_issue 同款(B' 双机: NAS=内网 IP,
|
|
VPS=FRP 127.0.0.1:13000)——修 10-01 架构审计 P1-13 尾:VPS 因子晋级
|
|
--gitea-open 曾因硬编码内网地址必失败静默降级.
|
|
"""
|
|
seen = {}
|
|
|
|
class _Resp:
|
|
def read(self):
|
|
return b'{"number": 7}'
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
seen["url"] = req.full_url
|
|
return _Resp()
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
|
|
monkeypatch.delenv("SANGUO_GITEA_BASE_URL", raising=False)
|
|
assert open_gitea_issue("t", "b", token="x") == 7
|
|
assert seen["url"] == ("http://192.168.2.154:3000/api/v1/repos/"
|
|
"sanguo/sanguo_vnpy_v2/issues")
|
|
|
|
monkeypatch.setenv("SANGUO_GITEA_BASE_URL", "http://127.0.0.1:13000")
|
|
assert open_gitea_issue("t", "b", token="x") == 7
|
|
assert seen["url"].startswith("http://127.0.0.1:13000/")
|
|
|
|
# 显式参数优先于 env
|
|
assert open_gitea_issue("t", "b", token="x", base_url="http://nas:3000") == 7
|
|
assert seen["url"].startswith("http://nas:3000/")
|
|
|
|
|
|
def test_graveyard_transition_writes_back_card(tmp_path, monkeypatch):
|
|
from sanguo_portfolio import pipeline_store as ps
|
|
import sanguo_factor.version_registry as vr
|
|
|
|
db = str(tmp_path / "pipeline.db")
|
|
monkeypatch.setenv("SANGUO_PIPELINE_DB", db)
|
|
hyp = ps.insert_hypothesis(db, {
|
|
"title": "t", "logic": "l", "expected_sign": "positive",
|
|
"falsifiable": "f", "data_needs": []})["id"]
|
|
|
|
reg_path = str(tmp_path / "reg.yaml")
|
|
reg = {"factors": {}}
|
|
vr.upsert_factor(reg, "llm_x", hypothesis=hyp, status="incubating")
|
|
vr.add_version(reg, "llm_x", commit="decompose",
|
|
params={"expression": "cs_rank(close)"},
|
|
effective_from="2026-10-01")
|
|
vr.save_registry(reg_path, reg)
|
|
|
|
rc = vr.main(["transition", "llm_x", "graveyard",
|
|
"--registry", reg_path,
|
|
"--events", str(tmp_path / "ev.jsonl"),
|
|
"--cause", "连续三月 t<2"])
|
|
assert rc == 0
|
|
item = next(i for i in ps.list_hypotheses(db) if i["id"] == hyp)
|
|
assert item["state"] == "graveyard"
|
|
assert "连续三月 t<2" in item["deathReason"]
|
|
assert item["factorId"] == "llm_x"
|
|
|
|
|
|
def test_writeback_skips_non_card_hypothesis(tmp_path):
|
|
import sanguo_factor.version_registry as vr
|
|
assert vr.writeback_card_death(None, {"factors": {}},
|
|
{"hypothesis": "H-2026-09-22-05",
|
|
"name": "f"}, "x") is False
|
|
|
|
|
|
def _insert_card(db):
|
|
from sanguo_portfolio import pipeline_store as ps
|
|
return ps.insert_hypothesis(db, {
|
|
"title": "t", "logic": "l", "expected_sign": "positive",
|
|
"falsifiable": "f", "data_needs": []})["id"]
|
|
|
|
|
|
def _read_missed(events_path):
|
|
with open(events_path, encoding="utf-8") as f:
|
|
return [json.loads(line) for line in f
|
|
if json.loads(line).get("event") == "death_writeback_missed"]
|
|
|
|
|
|
def test_writeback_card_not_in_db_leaves_trace(tmp_path, capsys):
|
|
"""P2-10: 库在但卡不在(None 语义)→stderr+events 附记,不再零痕迹返回."""
|
|
import sanguo_factor.version_registry as vr
|
|
|
|
db = str(tmp_path / "pipeline.db")
|
|
_insert_card(db) # 库存在有表,但回写目标 hyp-missing 不在
|
|
events = str(tmp_path / "ev.jsonl")
|
|
rc = vr.writeback_card_death(
|
|
db, {"factors": {}}, {"hypothesis": "hyp-missing",
|
|
"name": "f_ghost"}, "死因",
|
|
events_path=events)
|
|
assert rc is False
|
|
err = capsys.readouterr().err
|
|
assert "hyp-missing" in err and "f_ghost" in err
|
|
missed = _read_missed(events)
|
|
assert len(missed) == 1
|
|
assert missed[0]["hyp_id"] == "hyp-missing"
|
|
assert missed[0]["name"] == "f_ghost"
|
|
|
|
|
|
def test_writeback_relative_pipeline_db_absolutized(tmp_path, monkeypatch, capsys):
|
|
"""P2-10: pipeline_db 相对路径→stderr 警告+abs 绝对化(落库位置明确),回写真落库."""
|
|
import sanguo_factor.version_registry as vr
|
|
from sanguo_portfolio import pipeline_store as ps
|
|
|
|
monkeypatch.chdir(tmp_path)
|
|
db = "pipeline_rel.db" # 相对路径(防错误 cwd 落错库的病灶形态)
|
|
hyp = _insert_card(db)
|
|
events = str(tmp_path / "ev.jsonl")
|
|
rc = vr.writeback_card_death(
|
|
db, {"factors": {}}, {"hypothesis": hyp, "name": "f_rel"}, "死因",
|
|
events_path=events)
|
|
assert rc is True
|
|
err = capsys.readouterr().err
|
|
assert "相对路径" in err
|
|
assert os.path.abspath(db) in err # 警告点名绝对化后的落库位置
|
|
item = next(i for i in ps.list_hypotheses(db) if i["id"] == hyp)
|
|
assert item["state"] == "graveyard" # 回写命中真库
|
|
assert not os.path.exists(events) # 未命中场景不产 missed 事件
|
|
|
|
|
|
def test_writeback_missing_db_skipped_not_created(tmp_path, capsys):
|
|
"""P2-10: 卡片库不存在→stderr 警告+跳过回写,不再静默 CREATE 空库."""
|
|
import sanguo_factor.version_registry as vr
|
|
|
|
db = str(tmp_path / "sub" / "pipeline.db") # 连同父目录都不存在
|
|
events = str(tmp_path / "ev.jsonl")
|
|
rc = vr.writeback_card_death(
|
|
db, {"factors": {}}, {"hypothesis": "hyp-404",
|
|
"name": "f_lost"}, "死因",
|
|
events_path=events)
|
|
assert rc is False
|
|
err = capsys.readouterr().err
|
|
assert "不存在" in err
|
|
assert not os.path.exists(db) # 未建空库(_connect 静默 CREATE 已挡)
|
|
missed = _read_missed(events)
|
|
assert len(missed) == 1
|
|
assert missed[0]["hyp_id"] == "hyp-404"
|
|
assert missed[0]["name"] == "f_lost"
|
|
|
|
|
|
# —— #91①: 条目级 origin 写入方(建条即落;P3-11 消费方条目级优先) ——
|
|
def test_upsert_origin_lands_on_entry(reg):
|
|
entry = upsert_factor(reg, "fa_new", "hyp-1", origin="decomposer")
|
|
assert entry["origin"] == "decomposer"
|
|
assert reg["factors"]["fa_new"]["origin"] == "decomposer"
|
|
|
|
|
|
def test_upsert_without_origin_omits_key(reg):
|
|
entry = upsert_factor(reg, "fa_plain", "hyp-2")
|
|
assert "origin" not in entry
|
|
|
|
|
|
def test_upsert_invalid_origin_rejected(reg):
|
|
with pytest.raises(ValueError, match="非法 origin"):
|
|
upsert_factor(reg, "fa_bad", "hyp-3", origin="Decomposer")
|
|
|
|
|
|
def test_upsert_idempotent_return_does_not_overwrite_origin(reg):
|
|
first = upsert_factor(reg, "fa_x", "hyp-4") # 建条无 origin
|
|
again = upsert_factor(reg, "fa_x", "hyp-4", origin="decomposer")
|
|
assert again is first
|
|
assert "origin" not in reg["factors"]["fa_x"] # 幂等返回不补写
|
|
|
|
|
|
def test_upsert_factor_description_built_once_not_overwritten(reg):
|
|
"""2026-10-09 补丁:description 条目级自述,仅建条落盘,幂等不覆盖."""
|
|
e1 = upsert_factor(reg, "f1", "H-1",
|
|
description="20日均值动量,股价延续性的量度")
|
|
assert e1["description"] == "20日均值动量,股价延续性的量度"
|
|
# 同名同假设幂等返回:换 description 入参也不覆盖(改描述走升版本)
|
|
e2 = upsert_factor(reg, "f1", "H-1", description="新描述不应生效")
|
|
assert e2 is e1
|
|
assert reg["factors"]["f1"]["description"] == "20日均值动量,股价延续性的量度"
|
|
# 不传 description=条目无该键(存量因子形态)
|
|
upsert_factor(reg, "f2", "H-2")
|
|
assert "description" not in reg["factors"]["f2"]
|
|
|
|
|
|
class TestWritebackAggregation:
|
|
"""决议O:卡片死亡回写=全死才埋(任一因子死不再埋整卡)."""
|
|
|
|
def _card_db(self, tmp_path):
|
|
from sanguo_portfolio import pipeline_store as ps
|
|
db = str(tmp_path / "pipeline.db")
|
|
ps.insert_hypothesis(db, {
|
|
"title": "高管增持后跑赢", "logic": "x",
|
|
"expected_sign": "positive", "falsifiable": "y"})
|
|
with __import__("sqlite3").connect(db) as conn:
|
|
conn.execute(
|
|
"UPDATE hypotheses SET id='hyp-agg1' WHERE title='高管增持后跑赢'")
|
|
return db
|
|
|
|
def test_partial_death_keeps_card_alive(self, tmp_path, capsys):
|
|
db = self._card_db(tmp_path)
|
|
reg = {"factors": {
|
|
"fa_a": {"name": "fa_a", "hypothesis": "hyp-agg1",
|
|
"status": "graveyard"}, # 死的那个(触发者)
|
|
"fa_b": {"name": "fa_b", "hypothesis": "hyp-agg1",
|
|
"status": "incubating"},}} # 活着 → 不埋卡
|
|
ok = writeback_card_death(db, reg, reg["factors"]["fa_a"], "IC 两窗反号")
|
|
assert ok is False
|
|
out = capsys.readouterr().err
|
|
assert "部分因子存活" in out and "fa_b" in out
|
|
from sanguo_portfolio import pipeline_store as ps
|
|
card = [c for c in ps.list_hypotheses(db) if c["id"] == "hyp-agg1"][0]
|
|
assert card["state"] == "queued" # 卡留活跃池
|
|
|
|
def test_all_dead_buries_card_with_new_reason(self, tmp_path):
|
|
db = self._card_db(tmp_path)
|
|
reg = {"factors": {
|
|
"fa_a": {"name": "fa_a", "hypothesis": "hyp-agg1",
|
|
"status": "graveyard"},
|
|
"fa_b": {"name": "fa_b", "hypothesis": "hyp-agg1",
|
|
"status": "graveyard"},}} # 全灭 → 埋
|
|
ok = writeback_card_death(db, reg, reg["factors"]["fa_a"], "IC 两窗反号")
|
|
assert ok is True
|
|
from sanguo_portfolio import pipeline_store as ps
|
|
card = [c for c in ps.list_hypotheses(db) if c["id"] == "hyp-agg1"][0]
|
|
assert card["state"] == "graveyard"
|
|
assert card["deathReason"] == "全灭:触发因子 fa_a 入墓园: IC 两窗反号"
|
|
|
|
def test_zero_sibling_factor_buries(self, tmp_path):
|
|
"""卡下无其他因子(一卡一因子旧语境)→该因子死即全灭,照埋."""
|
|
db = self._card_db(tmp_path)
|
|
reg = {"factors": {
|
|
"fa_a": {"name": "fa_a", "hypothesis": "hyp-agg1",
|
|
"status": "graveyard"},}}
|
|
assert writeback_card_death(db, reg, reg["factors"]["fa_a"], "x") is True
|