claude_dev
b7ce364b11
10-04 首班起挂 NAS 月度批评链第四段(run_factor_monthly_standalone.sh): - 15 截面增量前置: 12 quant 源+size_log+在位者 composite_quant12_v2a_szneul +挑战者 composite_quant12_szneul, 各跑 daily_section 220 历日短窗幂等 merge——NAS 截面无每日班(09-25/26 手动快照起点), 批前必然新鲜 - as_of(自然月末)撞周末/节假日: latest_bar_date 归一 ≤as_of 最近日线 bar 日(000001 走 symbol 索引秒回), 防 daily_section sanity raise - 归因双跑同窗同参(全窗锚 2021-07-01 与首跑件一致, 挑战者受其截面 自然窗限制; 挑战者落工作区 challenger/ 子目录=端点 glob 非递归防顶掉) - cp 双投递生效位(--effective-dir): 在位者件→data/attribution/ 根 glob 取最新; 挑战者件→challenger/attribution_<end>_<因子名>.json 带因子名防顶掉; #69 影子 A/B 四周对照的归因侧供给 - fail loud: 截面断供/归因挂/投递缺任一→容器非零→班次 final=1 (静默旧件=红线); 生效位目录容器挂 rw /effective-data, 代码卷恒 :ro - 档随码走: 因子档归因行拆两行(独立 CLI+归因批)+runbook 三段链→四段 (VPS 侧批评三段不镜像 stage4, 归因双跑落 NAS, VPS 归因件=infra 偏差日报链另行供给) - 测试 428→438(归一日/编排序/同窗断言/挑战者子目录/双投递文件名/ fail loud/CLI 冒烟共 10 件) Co-Authored-By: Claude Code <noreply@anthropic.com>
182 lines
7.7 KiB
Python
182 lines
7.7 KiB
Python
"""attribution_batch(月度批评 stage4)单测: 归一日/编排序/挑战者子目录/fail loud."""
|
|
import os
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from sanguo_factor import attribution_batch as ab
|
|
|
|
|
|
@pytest.fixture()
|
|
def tiny_bar_db(tmp_path):
|
|
"""最小 dbbardata: 000001 三根日 bar, 查询只吃
|
|
symbol+exchange+interval+datetime 列."""
|
|
p = tmp_path / "bars.db"
|
|
conn = sqlite3.connect(p)
|
|
conn.execute("CREATE TABLE dbbardata (symbol TEXT, exchange TEXT, "
|
|
"interval TEXT, datetime TEXT, close_price REAL)")
|
|
rows = [
|
|
("000001", "SZSE", "d", "2026-10-02 15:00:00", 10.0), # 周五
|
|
("000001", "SZSE", "d", "2026-10-05 15:00:00", 10.5), # 周一
|
|
("000001", "SZSE", "d", "2026-10-06 15:00:00", 10.6), # 周二
|
|
]
|
|
conn.executemany("INSERT INTO dbbardata VALUES (?,?,?,?,?)", rows)
|
|
conn.commit()
|
|
conn.close()
|
|
return str(p)
|
|
|
|
|
|
class TestLatestBarDate:
|
|
def test_as_of_on_bar_day_returns_it(self, tiny_bar_db):
|
|
assert ab.latest_bar_date(tiny_bar_db, "2026-10-02") == "2026-10-02"
|
|
|
|
def test_as_on_weekend_falls_back_to_prior_bar(self, tiny_bar_db):
|
|
# 2026-10-03/04=周六日, 无 bar → 归一到 10-02
|
|
assert ab.latest_bar_date(tiny_bar_db, "2026-10-04") == "2026-10-02"
|
|
|
|
def test_no_bar_at_all_raises(self, tmp_path):
|
|
p = tmp_path / "empty.db"
|
|
conn = sqlite3.connect(p)
|
|
conn.execute("CREATE TABLE dbbardata (symbol TEXT, exchange TEXT, "
|
|
"interval TEXT, datetime TEXT, close_price REAL)")
|
|
conn.commit()
|
|
conn.close()
|
|
with pytest.raises(ValueError, match="bar"):
|
|
ab.latest_bar_date(str(p), "2026-10-04")
|
|
|
|
|
|
class TestRunAttributionBatch:
|
|
@pytest.fixture()
|
|
def spies(self, monkeypatch):
|
|
calls = {"sections": [], "attr": []}
|
|
|
|
def fake_daily_section(name, end, out_dir, lookback_days=220, **kw):
|
|
calls["sections"].append((name, end, out_dir))
|
|
return {"ok": True, "factor": name, "date": end,
|
|
"merge": {"added_dates": 1}, "sanity": {"non_null": 4000}}
|
|
|
|
def fake_attr_main(argv):
|
|
calls["attr"].append(argv)
|
|
return 0
|
|
|
|
monkeypatch.setattr(ab, "run_daily_section", fake_daily_section)
|
|
monkeypatch.setattr(ab, "attribution_main", fake_attr_main)
|
|
# 归一日走真库的用例单独直连; 其余用例 mock 掉避免假库查询
|
|
monkeypatch.setattr(ab, "latest_bar_date",
|
|
lambda db, as_of: "2026-10-02")
|
|
return calls
|
|
|
|
def _run(self, tmp_path, **kw):
|
|
return ab.run_attribution_batch(
|
|
as_of="2026-10-04", sections_dir=str(tmp_path / "secs"),
|
|
work_dir=str(tmp_path / "attr"), vnpy_db=str(tmp_path / "x.db"),
|
|
industry_path=str(tmp_path / "ind.parquet"), **kw)
|
|
|
|
def test_section_order_incumbent_challenger_last(self, spies, tmp_path):
|
|
self._run(tmp_path)
|
|
names = [c[0] for c in spies["sections"]]
|
|
# 12 quant 源 + size_log + 在位者 + 挑战者 = 15; 组合两名在末
|
|
assert len(names) == 15
|
|
assert names[-2:] == ["composite_quant12_v2a_szneul",
|
|
"composite_quant12_szneul"]
|
|
assert set(names[:13]) >= {"vma_60", "alpha16", "size_log"}
|
|
|
|
def test_all_sections_use_normalized_end(self, spies, tmp_path,
|
|
tiny_bar_db):
|
|
# as_of=10-04(周日) → 归一 10-02; 全部截面同端
|
|
ab.run_attribution_batch(
|
|
as_of="2026-10-04", sections_dir=str(tmp_path / "secs"),
|
|
work_dir=str(tmp_path / "attr"), vnpy_db=tiny_bar_db,
|
|
industry_path=str(tmp_path / "ind.parquet"))
|
|
assert {c[1] for c in spies["sections"]} == {"2026-10-02"}
|
|
|
|
def test_two_attribution_calls_same_window_challenger_subdir(
|
|
self, spies, tmp_path):
|
|
res = self._run(tmp_path)
|
|
assert len(spies["attr"]) == 2
|
|
a1, a2 = spies["attr"]
|
|
d1 = dict(zip(a1[::2], a1[1::2]))
|
|
d2 = dict(zip(a2[::2], a2[1::2]))
|
|
assert d1["--factor"] == "composite_quant12_v2a_szneul"
|
|
assert d2["--factor"] == "composite_quant12_szneul"
|
|
assert d1["--out-dir"] == str(tmp_path / "attr")
|
|
assert d2["--out-dir"] == str(tmp_path / "attr" / "challenger")
|
|
# 同窗同参(对照可比)
|
|
assert d1["--start"] == d2["--start"] == "2021-07-01"
|
|
assert d1["--end"] == d2["--end"]
|
|
assert d1["--top-n"] == d2["--top-n"] == "200"
|
|
assert d1["--reb-days"] == d2["--reb-days"] == "63"
|
|
assert res["as_of"] == "2026-10-04"
|
|
assert res["end_sec"] == "2026-10-02"
|
|
|
|
def test_section_failure_propagates(self, spies, tmp_path, monkeypatch):
|
|
def boom(name, end, out_dir, lookback_days=220, **kw):
|
|
if name == "alpha16":
|
|
raise ValueError(f"截面跑批错误: [{name}]")
|
|
return {"ok": True, "merge": {"added_dates": 1},
|
|
"sanity": {"non_null": 4000}}
|
|
|
|
monkeypatch.setattr(ab, "run_daily_section", boom)
|
|
with pytest.raises(ValueError, match="alpha16"):
|
|
self._run(tmp_path)
|
|
# 断供后不再继续(归因不跑)
|
|
assert spies["attr"] == []
|
|
|
|
def test_attribution_failure_raises(self, spies, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(ab, "attribution_main",
|
|
lambda argv: (_ for _ in ()).throw(
|
|
RuntimeError("attribution CLI failed")))
|
|
with pytest.raises(RuntimeError, match="attribution"):
|
|
self._run(tmp_path)
|
|
|
|
def test_effective_dir_double_delivery(self, spies, tmp_path, monkeypatch):
|
|
"""双投递: 在位者件→生效位根; 挑战者件→challenger/ 带因子名."""
|
|
eff = tmp_path / "effective"
|
|
|
|
def fake_attr_main(argv):
|
|
d = dict(zip(argv[::2], argv[1::2]))
|
|
os.makedirs(d["--out-dir"], exist_ok=True)
|
|
# CLI 实际产物名=attribution_<end>.json
|
|
(Path(d["--out-dir"]) / "attribution_2026-10-02.json").write_text(
|
|
"{}", encoding="utf-8")
|
|
return 0
|
|
|
|
monkeypatch.setattr(ab, "attribution_main", fake_attr_main)
|
|
res = ab.run_attribution_batch(
|
|
as_of="2026-10-04", sections_dir=str(tmp_path / "secs"),
|
|
work_dir=str(tmp_path / "attr"), vnpy_db=str(tmp_path / "x.db"),
|
|
industry_path=str(tmp_path / "ind.parquet"),
|
|
effective_dir=str(eff))
|
|
assert (eff / "attribution_2026-10-02.json").exists()
|
|
assert (eff / "challenger" /
|
|
"attribution_2026-10-02_composite_quant12_szneul.json").exists()
|
|
assert res["effective"] == [
|
|
str(eff / "attribution_2026-10-02.json"),
|
|
str(eff / "challenger" /
|
|
"attribution_2026-10-02_composite_quant12_szneul.json")]
|
|
|
|
|
|
def test_main_cli_smoke(tmp_path, monkeypatch):
|
|
"""main 组装 argv→run_attribution_batch, 返回 0."""
|
|
seen = {}
|
|
|
|
def fake_run(as_of, sections_dir, work_dir, vnpy_db, industry_path, **kw):
|
|
seen.update(as_of=as_of, sections_dir=sections_dir,
|
|
work_dir=work_dir, vnpy_db=vnpy_db,
|
|
industry_path=industry_path, **kw)
|
|
return {"as_of": as_of, "end_sec": as_of,
|
|
"sections": [], "attribution": [], "effective": []}
|
|
|
|
monkeypatch.setattr(ab, "run_attribution_batch", fake_run)
|
|
rc = ab.main([
|
|
"--as-of", "2026-09-30",
|
|
"--sections-dir", str(tmp_path / "s"),
|
|
"--work-dir", str(tmp_path / "w"),
|
|
"--vnpy-db", str(tmp_path / "b.db"),
|
|
"--industry-path", str(tmp_path / "i.parquet"),
|
|
])
|
|
assert rc == 0
|
|
assert seen["as_of"] == "2026-09-30"
|
|
assert seen["start"] == "2021-07-01" # 缺省全窗锚(与首跑件一致)
|