claude_dev
fbf5d3f6ed
- AttributionResearch.vue submitRuling 补 catch：409/422/网络错误弹 ElMessage.error（含响应 detail，兜底「裁决提交失败」），照 IncubationLadder/PromotionReview 同款写法；finally 保留
- 新增失败路径测试：post 拒绝 {response:{data:{detail:'当月已有裁决'}}} → 弹出该文案、post 恰一次、卡不重拉
- verdict_replay.load_latest_monthly 与 research_card 端点：json.load 后非 dict 跳过（回退更旧文件 / replay_doc=None）
- 清理 test_verdict_replay.py unused import os、test_verdict_records.py unused import json
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
127 lines
5.3 KiB
Python
127 lines
5.3 KiB
Python
# tests/factor/test_verdict_replay.py
|
|
"""研判卡回放引擎 TDD——spec §4.7 P3 第一刀之 1(触发点普查+七档近似成绩)."""
|
|
import json
|
|
|
|
from sanguo_factor import verdict_replay as vr
|
|
from sanguo_factor.decay import COLLECTIVE_MEDIAN_FLOOR
|
|
|
|
|
|
def _points(name, months_ts):
|
|
return {name: [{"month": m, "t": t} for m, t in months_ts]}
|
|
|
|
|
|
def test_census_below_floor_months_only():
|
|
mp = {"a": [{"month": "2026-06", "t": 3.0}, {"month": "2026-07", "t": 0.5},
|
|
{"month": "2026-08", "t": 2.5}],
|
|
"b": [{"month": "2026-07", "t": 0.7}]}
|
|
eps = vr.census(mp, ["a", "b"])
|
|
assert [e["month"] for e in eps] == ["2026-07"]
|
|
assert eps[0]["n"] == 2
|
|
assert eps[0]["median"] == 0.6
|
|
assert eps[0]["median"] < COLLECTIVE_MEDIAN_FLOOR
|
|
|
|
|
|
def test_census_skips_none_and_empty():
|
|
mp = {"a": [{"month": "2026-06", "t": None}, {"month": "2026-07", "t": 0.1}]}
|
|
assert [e["month"] for e in vr.census(mp, ["a"])] == ["2026-07"]
|
|
assert vr.census({}, ["a"]) == []
|
|
assert vr.census({"a": []}, ["a"]) == []
|
|
|
|
|
|
def test_month_next_rollover():
|
|
assert vr._month_next("2026-09", 1) == "2026-10"
|
|
assert vr._month_next("2026-12", 1) == "2027-01"
|
|
assert vr._month_next("2026-11", 4) == "2027-03"
|
|
|
|
|
|
def _curve(days, ret):
|
|
return {d: ret for d in days}
|
|
|
|
|
|
def _biz_days(lo, n):
|
|
from datetime import date, timedelta
|
|
d0 = date.fromisoformat(lo)
|
|
out, d, i = [], d0, 0
|
|
while len(out) < n:
|
|
if d.weekday() < 5:
|
|
out.append(d.isoformat())
|
|
d += timedelta(days=1)
|
|
i += 1
|
|
assert i < 400
|
|
return out
|
|
|
|
|
|
def test_blend_weights_and_missing_day():
|
|
days = ["2026-01-05", "2026-01-06"]
|
|
c1 = {"2026-01-05": 0.01, "2026-01-06": 0.02}
|
|
c2 = {"2026-01-05": 0.03} # 01-06 缺日=0
|
|
eq = vr.blend_returns({"x": c1, "y": c2})
|
|
assert eq["2026-01-05"] == (0.01 + 0.03) / 2
|
|
assert eq["2026-01-06"] == 0.02 / 2
|
|
w = vr.blend_returns({"x": c1, "y": c2}, {"x": 3.0, "y": 1.0})
|
|
assert abs(w["2026-01-05"] - (3 * 0.01 + 0.03) / 4) < 1e-12
|
|
assert vr.blend_returns({}) == {}
|
|
|
|
|
|
def test_replay_episode_tiers():
|
|
days = _biz_days("2026-07-01", 70) # 覆盖 7/8/9 月窗
|
|
base_curves = {"all_weather": _curve(days, 0.001),
|
|
"momentum": _curve(days, 0.002),
|
|
"value": _curve(days, 0.0),
|
|
"small_cap": _curve(days, 0.002)}
|
|
eps = [{"month": "2026-06", "median": 0.6, "n": 4}]
|
|
out = vr.replay_episodes(eps, base_curves)
|
|
assert len(out) == 1 and out[0]["replayable"] is True
|
|
tiers = out[0]["tiers"]
|
|
assert set(tiers) == {str(i) for i in range(7)}
|
|
assert tiers["0"] == tiers["5"] # 档5=仓位不动同档0
|
|
assert abs(tiers["1"]["ret3m"] - 0.7 * tiers["0"]["ret3m"] - 0.3 * 0.0) < 2e-4
|
|
assert tiers["6"] == {"ret3m": 0.0, "max_dd": 0.0} # 清仓=现金
|
|
assert tiers["3"] is None and tiers["4"] is None # 需精确验证
|
|
assert tiers["2"]["ret3m"] != tiers["0"]["ret3m"] # 换权有效
|
|
|
|
|
|
def test_replay_episode_uncovered_window():
|
|
days = _biz_days("2026-07-01", 5) # 窗内不足 10 交易日
|
|
out = vr.replay_episodes([{"month": "2026-06", "median": 0.6, "n": 4}],
|
|
{"x": _curve(days, 0.01)})
|
|
assert out[0]["replayable"] is False and "tiers" not in out[0]
|
|
|
|
|
|
def test_load_latest_monthly_skips_corrupt(tmp_path):
|
|
tmp_path.joinpath("host_2026-08-31.json").write_text("not-json", encoding="utf-8")
|
|
good = {"as_of": "2026-09-30", "collective": {"is_collective_decay": True}}
|
|
tmp_path.joinpath("host_2026-09-30.json").write_text(
|
|
json.dumps(good), encoding="utf-8")
|
|
tmp_path.joinpath("verdict_replay.json").write_text("{}", encoding="utf-8") # 不中 glob
|
|
assert vr.load_latest_monthly(str(tmp_path))["as_of"] == "2026-09-30"
|
|
assert vr.load_latest_monthly(str(tmp_path / "empty")) is None
|
|
|
|
|
|
def test_build_doc_and_cli(tmp_path, monkeypatch):
|
|
monthly = tmp_path / "factor_monthly"
|
|
monthly.mkdir()
|
|
rep = {"as_of": "2026-09-30",
|
|
"verdicts": {"a": {"state": "ok", "ratio": 0.9},
|
|
"b": {"state": "alert", "ratio": 0.4}},
|
|
"monthly_points": {"a": [{"month": "2026-08", "t": 3.0},
|
|
{"month": "2026-09", "t": 0.5}],
|
|
"b": [{"month": "2026-09", "t": 0.7}]}}
|
|
monthly.joinpath("h_2026-09-30.json").write_text(json.dumps(rep), encoding="utf-8")
|
|
reg = {"strategies": {
|
|
"s1": {"stage": "live", "backtest_run": "t1"},
|
|
"s2": {"stage": "retired", "backtest_run": "t9"}, # 不算
|
|
"s3": {"stage": "live", "backtest_run": None}}} # 缺挂跳过
|
|
monkeypatch.setattr(vr, "_load_curves_for_registry",
|
|
lambda db, registry: {"s1": _curve(_biz_days("2026-10-01", 70), 0.001)})
|
|
doc = vr.build_doc(str(monthly), "fake.db", reg, "2026-09-30")
|
|
assert doc["since"] == "2026-08"
|
|
assert doc["strategies"] == ["s1"]
|
|
assert [e["month"] for e in doc["episodes"]] == ["2026-09"] # 触发但窗未满
|
|
assert doc["episodes"][0]["replayable"] is False
|
|
rc = vr.main(["--as-of", "2026-09-30", "--monthly-dir", str(monthly),
|
|
"--db", "fake.db"])
|
|
assert rc == 0
|
|
out = json.loads(monthly.joinpath("verdict_replay.json").read_text(encoding="utf-8"))
|
|
assert out["as_of"] == "2026-09-30"
|