668 lines
34 KiB
Python
668 lines
34 KiB
Python
# tests/portfolio/test_routes_pipeline.py
|
|
"""接真端点 TDD——spec §4.5 决议 L 六件之 4(前端零改动=删 mock 即接真).
|
|
|
|
环境注入全走 monkeypatch env(SANGUO_PIPELINE_DB/SANGUO_STRATEGY_REGISTRY/
|
|
SANGUO_PROMOTION_PASSPHRASE/SANGUO_FACTOR_MONTHLY_DIR),双机同代码各读各的.
|
|
"""
|
|
import json
|
|
import os
|
|
|
|
import pytest
|
|
import yaml
|
|
from fastapi.testclient import TestClient
|
|
|
|
from sanguo_api.app import create_app
|
|
from sanguo_api.auth import hash_password
|
|
from sanguo_portfolio import pipeline_store
|
|
|
|
|
|
@pytest.fixture()
|
|
def env(tmp_path, monkeypatch):
|
|
pipeline_db = str(tmp_path / "pipeline.db")
|
|
registry = str(tmp_path / "strategy_registry.yaml")
|
|
monthly = str(tmp_path / "factor_monthly")
|
|
issue_dir = str(tmp_path / "registry_events")
|
|
attribution_dir = str(tmp_path / "attribution")
|
|
monkeypatch.setenv("SANGUO_PIPELINE_DB", pipeline_db)
|
|
monkeypatch.setenv("SANGUO_STRATEGY_REGISTRY", registry)
|
|
monkeypatch.setenv("SANGUO_FACTOR_MONTHLY_DIR", monthly)
|
|
monkeypatch.setenv("SANGUO_PROMOTION_PASSPHRASE", "毕业口令")
|
|
monkeypatch.setenv("SANGUO_REGISTRY_ISSUE_DIR", issue_dir)
|
|
monkeypatch.setenv("SANGUO_STRATEGY_EVENTS",
|
|
str(tmp_path / "strategy_events.jsonl"))
|
|
monkeypatch.setenv("SANGUO_ATTRIBUTION_DIR", attribution_dir)
|
|
pipeline_store.init_db(pipeline_db)
|
|
os.makedirs(monthly, exist_ok=True)
|
|
with open(registry, "w", encoding="utf-8") as f:
|
|
yaml.safe_dump({"strategies": {
|
|
"s_shadow": {"name": "s_shadow", "hypothesis": "H-1",
|
|
"composite_version": "quant12_v2a", "stage": "shadow",
|
|
"backfilled": False,
|
|
"stage_history": [{"kind": "transition", "from": "paper",
|
|
"to": "shadow", "at": "2026-09-01"}],
|
|
"gate_snapshot": None, "review_issue": None,
|
|
"capital": None, "retired_cause": None,
|
|
"backtest_run": None, "paper_account_id": 1},
|
|
"s_live": {"name": "s_live", "hypothesis": "H-2",
|
|
"composite_version": "quant12_v2a", "stage": "live",
|
|
"backfilled": True, "stage_history": [],
|
|
"gate_snapshot": None, "review_issue": None,
|
|
"capital": {"mode": "占账户比例", "value": 30,
|
|
"at": "2026-09-18"},
|
|
"retired_cause": None, "backtest_run": None,
|
|
"paper_account_id": None},
|
|
}}, f, allow_unicode=True)
|
|
return {"pipeline_db": pipeline_db, "registry": registry, "monthly": monthly,
|
|
"issue_dir": issue_dir, "attribution": attribution_dir}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client(tmp_path_factory):
|
|
app = create_app(
|
|
db_path=str(tmp_path_factory.mktemp("routes") / "app.db"),
|
|
auth_config={"username": "admin",
|
|
"password_hash": hash_password("admin"),
|
|
"jwt_secret": "test-secret", "expire_minutes": 60},
|
|
max_workers=1,
|
|
)
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def token(client) -> str:
|
|
r = client.post("/api/v1/auth/login",
|
|
json={"username": "admin", "password": "admin"})
|
|
assert r.status_code == 200
|
|
return r.json()["token"]
|
|
|
|
|
|
def _h(token): return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def test_requires_auth(client):
|
|
assert client.get("/api/v1/pipeline/ladder").status_code == 401
|
|
|
|
|
|
def test_ladder_shape(client, token, env):
|
|
r = client.get("/api/v1/pipeline/ladder", headers=_h(token))
|
|
assert r.status_code == 200
|
|
items = r.json()["items"]
|
|
by = {i["name"]: i for i in items}
|
|
assert by["s_live"]["stage"] == "live"
|
|
assert by["s_live"]["capitalShare"] == "30%"
|
|
assert "补录" in by["s_live"]["note"]
|
|
assert by["s_shadow"]["stage"] == "shadow"
|
|
assert by["s_shadow"]["shadowWeeks"] >= 0
|
|
assert "retired" not in [i["stage"] for i in items]
|
|
|
|
|
|
def test_gate_config_get_put_audit(client, token, env):
|
|
r = client.get("/api/v1/pipeline/gate-config", headers=_h(token))
|
|
body = r.json()
|
|
assert body["defaults"]["backtest_sharpe_min"] == 1.0
|
|
assert body["effective"]["shadow_weeks_min"] == 4
|
|
assert body["audit"] == []
|
|
r2 = client.put("/api/v1/pipeline/gate-config", headers=_h(token),
|
|
json={"key": "shadow_te_annual_max", "value": 0.06,
|
|
"changed_by": "admin"})
|
|
assert r2.status_code == 200
|
|
r3 = client.get("/api/v1/pipeline/gate-config", headers=_h(token))
|
|
assert r3.json()["effective"]["shadow_te_annual_max"] == 0.06
|
|
assert r3.json()["audit"][0]["old_value"] == 0.05
|
|
bad = client.put("/api/v1/pipeline/gate-config", headers=_h(token),
|
|
json={"key": "nope", "value": 1, "changed_by": "admin"})
|
|
assert bad.status_code == 422
|
|
|
|
|
|
def test_graduate_passphrase_and_stage_gates(client, token, env):
|
|
# 口令错 → 403
|
|
r = client.post("/api/v1/pipeline/promotion/graduate", headers=_h(token),
|
|
json={"name": "s_shadow", "passphrase": "错的",
|
|
"allocMode": "固定金额", "allocValue": "50000"})
|
|
assert r.status_code == 403
|
|
# 非 shadow 态 → 422
|
|
r2 = client.post("/api/v1/pipeline/promotion/graduate", headers=_h(token),
|
|
json={"name": "s_live", "passphrase": "毕业口令",
|
|
"allocMode": "固定金额", "allocValue": "50000"})
|
|
assert r2.status_code == 422
|
|
# shadow 周数不足(2026-09-01 起算,monkeypatch 今天不可行——改用周报数据缺失路径:
|
|
# 周报三指标无数据 → 422 周报数据不足)
|
|
r3 = client.post("/api/v1/pipeline/promotion/graduate", headers=_h(token),
|
|
json={"name": "s_shadow", "passphrase": "毕业口令",
|
|
"allocMode": "固定金额", "allocValue": "50000"})
|
|
assert r3.status_code == 422
|
|
assert "周报" in r3.json()["detail"]
|
|
|
|
|
|
def test_graduate_success_flow(client, token, env):
|
|
# 补齐周报(s_shadow 满 4 周且四项全达标)→ 毕业成功
|
|
from datetime import date, timedelta
|
|
ws = date(2026, 9, 7)
|
|
for i in range(4):
|
|
pipeline_store.upsert_weekly(env["pipeline_db"], {
|
|
"strategy_id": "s_shadow",
|
|
"week_start": (ws + timedelta(weeks=i)).isoformat(),
|
|
"te_annual": 0.04, "fill_rate": 0.97, "vs_backtest": 0.008,
|
|
"weeks_counted": 5,
|
|
"computed_at": "2026-10-10T20:45:00"})
|
|
# registry 影子起算日挪到 5 周前使 shadow_weeks>=4(不写死日期)
|
|
with open(env["registry"], encoding="utf-8") as f:
|
|
reg = yaml.safe_load(f)
|
|
reg["strategies"]["s_shadow"]["stage_history"][0]["at"] = (
|
|
(date.today() - timedelta(weeks=5)).isoformat())
|
|
with open(env["registry"], "w", encoding="utf-8") as f:
|
|
yaml.safe_dump(reg, f, allow_unicode=True)
|
|
r = client.post("/api/v1/pipeline/promotion/graduate", headers=_h(token),
|
|
json={"name": "s_shadow", "passphrase": "毕业口令",
|
|
"allocMode": "固定金额", "allocValue": "50000",
|
|
"reviewIssue": 88})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
with open(env["registry"], encoding="utf-8") as f:
|
|
reg2 = yaml.safe_load(f)
|
|
e = reg2["strategies"]["s_shadow"]
|
|
assert e["stage"] == "live" and e["capital"]["value"] == 50000
|
|
assert e["gate_snapshot"]["passed"] is True
|
|
assert e["review_issue"] == 88
|
|
tmpl = os.path.join(env["issue_dir"], "s_shadow_graduate_issue.md")
|
|
assert os.path.exists(tmpl) # J1:模板必落盘
|
|
# 再毕业一次 → 422(已是 live)
|
|
r2 = client.post("/api/v1/pipeline/promotion/graduate", headers=_h(token),
|
|
json={"name": "s_shadow", "passphrase": "毕业口令",
|
|
"allocMode": "固定金额", "allocValue": "50000"})
|
|
assert r2.status_code == 422
|
|
|
|
|
|
def test_todos_aggregates_local_artifacts(client, token, env):
|
|
with open(os.path.join(env["monthly"], "nas_2026-09-30.json"), "w",
|
|
encoding="utf-8") as f:
|
|
json.dump({"verdicts": {"fa_gross_margin": {"state": "alert"}},
|
|
"collective": {"is_collective_decay": False}}, f)
|
|
with open(os.path.join(env["monthly"], "data_gaps.json"), "w",
|
|
encoding="utf-8") as f:
|
|
json.dump({"s02_sentiment_delta": ["corpus_sentiment"]}, f)
|
|
r = client.get("/api/v1/pipeline/todos", headers=_h(token))
|
|
items = r.json()["items"]
|
|
texts = " ".join(i["title"] for i in items)
|
|
assert "fa_gross_margin" in texts # 衰减告警(factor 月报 JSON)
|
|
assert "s02_sentiment_delta" in texts # 数据缺口(data_gaps.json)
|
|
for i in items:
|
|
assert i["path"].startswith("/pipeline/") # 每条可跳转
|
|
assert i["severity"] in ("info", "warn", "critical")
|
|
|
|
|
|
def test_factor_verdict_routes(client, token, env, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY",
|
|
str(tmp_path / "factor_registry.yaml"))
|
|
with open(tmp_path / "factor_registry.yaml", "w", encoding="utf-8") as f:
|
|
yaml.safe_dump({"factors": {"fa_gross_margin": {
|
|
"name": "fa_gross_margin", "hypothesis": "H-1",
|
|
"status": "assessable", "versions": []}}}, f, allow_unicode=True)
|
|
monkeypatch.setenv("SANGUO_FACTOR_EVENTS",
|
|
str(tmp_path / "registry_events.jsonl"))
|
|
r = client.post("/api/v1/pipeline/factors/fa_gross_margin/verdict",
|
|
headers=_h(token),
|
|
json={"verdict": "promote",
|
|
"answers": [True, True, True, True, True],
|
|
"note": "IC 两窗 t=3.1/2.8,共线 0.42"})
|
|
assert r.status_code == 200
|
|
with open(tmp_path / "factor_registry.yaml", encoding="utf-8") as f:
|
|
assert yaml.safe_load(f)["factors"]["fa_gross_margin"]["status"] == "promoted"
|
|
# revise=留观(不改状态,只留评审事件)
|
|
r2 = client.post("/api/v1/pipeline/factors/fa_gross_margin/verdict",
|
|
headers=_h(token),
|
|
json={"verdict": "revise", "answers": [True, True, False,
|
|
True, True],
|
|
"note": "三问不过,退回改造观察"})
|
|
assert r2.status_code == 200
|
|
with open(tmp_path / "factor_registry.yaml", encoding="utf-8") as f:
|
|
assert yaml.safe_load(f)["factors"]["fa_gross_margin"]["status"] == "promoted"
|
|
|
|
|
|
def test_factor_verdict_bootstraps_runtime_registry(client, token, tmp_path,
|
|
monkeypatch):
|
|
# M1:未设 env 时 verdict 端点默认写 data/ 运行副本,首触从 config/ 种子自举
|
|
monkeypatch.chdir(tmp_path) # 默认路径全 cwd 相对,隔离到 tmp
|
|
os.makedirs("config", exist_ok=True)
|
|
with open(os.path.join("config", "factor_registry.yaml"), "w",
|
|
encoding="utf-8") as f:
|
|
yaml.safe_dump({"factors": {"fa_x": {
|
|
"name": "fa_x", "hypothesis": "H-1", "status": "assessable",
|
|
"versions": []}}}, f, allow_unicode=True)
|
|
r = client.post("/api/v1/pipeline/factors/fa_x/verdict", headers=_h(token),
|
|
json={"verdict": "promote", "answers": [True] * 5,
|
|
"note": "首触自举后直写运行副本"})
|
|
assert r.status_code == 200
|
|
with open(os.path.join("data", "factor_registry.yaml"),
|
|
encoding="utf-8") as f:
|
|
assert yaml.safe_load(f)["factors"]["fa_x"]["status"] == "promoted"
|
|
assert os.path.exists(os.path.join("data", "registry_events.jsonl"))
|
|
|
|
|
|
def test_factors_registry_to_board(client, token, env, tmp_path, monkeypatch):
|
|
# 看板/候选下拉接真:注册表(运行副本)→PipelineFactor;graveyard 不上板
|
|
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY",
|
|
str(tmp_path / "factor_registry.yaml"))
|
|
with open(tmp_path / "factor_registry.yaml", "w", encoding="utf-8") as f:
|
|
yaml.safe_dump({"factors": {
|
|
"fa_fund_x": {"name": "fa_fund_x", "hypothesis": "H-1",
|
|
"status": "assessable",
|
|
"versions": [{"v": 1, "commit": "abc",
|
|
"params": {"source": "fundamentals_pit"},
|
|
"effective_from": "2026-01-01",
|
|
"effective_to": None}]},
|
|
"fa_dead": {"name": "fa_dead", "hypothesis": "H-2",
|
|
"status": "graveyard", "cause": "两窗反号",
|
|
"versions": []},
|
|
"fa_pledge_y": {"name": "fa_pledge_y", "hypothesis": "H-3",
|
|
"status": "promoted", "promotion_t": 4.01,
|
|
"versions": [{"v": 2, "commit": "def",
|
|
"params": {"source": "pledge_detail"},
|
|
"effective_from": "2026-02-01",
|
|
"effective_to": None}]},
|
|
}}, f, allow_unicode=True)
|
|
# eval_db 指向不存在文件(月度批评首班未跑)→ IC 值如实 null
|
|
monkeypatch.setenv("SANGUO_FACTOR_EVAL_DB",
|
|
str(tmp_path / "no_such_factor_eval.db"))
|
|
r = client.get("/api/v1/pipeline/factors", headers=_h(token))
|
|
assert r.status_code == 200
|
|
by = {i["id"]: i for i in r.json()["items"]}
|
|
assert "fa_dead" not in by # graveyard 不上板
|
|
fa = by["fa_fund_x"]
|
|
assert fa["icRecentT"] is None and fa["icFullT"] is None
|
|
assert fa["category"] == "fundamental" # source=fundamentals_pit
|
|
assert fa["version"] == "1"
|
|
pg = by["fa_pledge_y"]
|
|
assert pg["category"] == "pledge" # source 含 pledge
|
|
assert pg["promotedAtT"] == 4.01 # promotion_t 原样回传
|
|
assert pg["status"] == "promoted"
|
|
|
|
|
|
def test_factors_reads_injected_eval_db(client, token, env, tmp_path, monkeypatch):
|
|
# eval 库对齐月度链:app 注入路径优先,default_eval_db_path(data_backup 旧落点)只兜底
|
|
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY",
|
|
str(tmp_path / "factor_registry.yaml"))
|
|
with open(tmp_path / "factor_registry.yaml", "w", encoding="utf-8") as f:
|
|
yaml.safe_dump({"factors": {"fa_inj": {
|
|
"name": "fa_inj", "hypothesis": "H-9", "status": "assessable",
|
|
"versions": [{"v": 1, "commit": "abc", "params": {},
|
|
"effective_from": "2026-01-01",
|
|
"effective_to": None}]}},
|
|
}, f, allow_unicode=True)
|
|
# env 兜底路径故意不存在:只有命中注入路径才可能拿到 t
|
|
monkeypatch.setenv("SANGUO_FACTOR_EVAL_DB",
|
|
str(tmp_path / "no_such_eval.db"))
|
|
from datetime import date
|
|
from sanguo_api.routes_pipeline import set_pipeline_eval_db_path
|
|
from sanguo_factor import eval_store
|
|
inj_db = str(tmp_path / "factor_eval.db")
|
|
eval_store.init_db(inj_db)
|
|
end = date.today().isoformat()
|
|
run_id = eval_store.create_run(inj_db, "monthly_2026-09", "sz", 10, 1,
|
|
"2025-10-01", end, {})
|
|
eval_store.save_results(inj_db, run_id, [{
|
|
"factor": "fa_inj", "category": "fundamental", "expression": "x",
|
|
"metrics": {"1": {"t_stat": 2.7, "ic_mean": 0.03, "count": 240}}}])
|
|
eval_store.finish_run(inj_db, run_id, "done", 1)
|
|
set_pipeline_eval_db_path(inj_db)
|
|
try:
|
|
r = client.get("/api/v1/pipeline/factors", headers=_h(token))
|
|
assert r.status_code == 200
|
|
by = {i["id"]: i for i in r.json()["items"]}
|
|
assert by["fa_inj"]["icRecentT"] == 2.7 # t 来自注入库
|
|
assert by["fa_inj"]["lastEvalDate"] == end
|
|
finally:
|
|
set_pipeline_eval_db_path("") # 还原兜底,不影响他测
|
|
|
|
|
|
# ---- Task1(P2 defer): graduate 分支直测 + vs=0.0 回归 + 顺序双发 ----
|
|
|
|
def _set_shadow_since(registry_path: str, at: str) -> None:
|
|
"""把 s_shadow 的 shadow 起点改早(≥4 周),使影子周数 gate 可过."""
|
|
with open(registry_path, encoding="utf-8") as f:
|
|
reg = yaml.safe_load(f)
|
|
reg["strategies"]["s_shadow"]["stage_history"] = [
|
|
{"kind": "transition", "from": "paper", "to": "shadow", "at": at}]
|
|
with open(registry_path, "w", encoding="utf-8") as f:
|
|
yaml.safe_dump(reg, f, allow_unicode=True)
|
|
|
|
|
|
def _put_weekly(pipeline_db: str, name: str, te: float, fill: float, vs) -> None:
|
|
pipeline_store.upsert_weekly(pipeline_db, {
|
|
"strategy_id": name, "week_start": "2026-09-21",
|
|
"te_annual": te, "fill_rate": fill, "vs_backtest": vs,
|
|
"weeks_counted": 5, "computed_at": "2026-09-24T20:45:00"})
|
|
|
|
|
|
def test_graduate_503_no_passphrase(client, token, env, monkeypatch):
|
|
monkeypatch.delenv("SANGUO_PROMOTION_PASSPHRASE", raising=False)
|
|
r = client.post("/api/v1/pipeline/promotion/graduate",
|
|
json={"name": "s_shadow", "passphrase": "x"}, headers=_h(token))
|
|
assert r.status_code == 503
|
|
|
|
|
|
def test_graduate_403_wrong_passphrase(client, token, env):
|
|
r = client.post("/api/v1/pipeline/promotion/graduate",
|
|
json={"name": "s_shadow", "passphrase": "错的"},
|
|
headers=_h(token))
|
|
assert r.status_code == 403
|
|
|
|
|
|
def test_graduate_404_unknown(client, token, env):
|
|
r = client.post("/api/v1/pipeline/promotion/graduate",
|
|
json={"name": "s_unknown", "passphrase": "毕业口令"},
|
|
headers=_h(token))
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_graduate_422_wrong_stage(client, token, env):
|
|
r = client.post("/api/v1/pipeline/promotion/graduate",
|
|
json={"name": "s_live", "passphrase": "毕业口令"},
|
|
headers=_h(token))
|
|
assert r.status_code == 422
|
|
|
|
|
|
def test_graduate_422_no_weekly(client, token, env):
|
|
_set_shadow_since(env["registry"], "2026-08-20")
|
|
r = client.post("/api/v1/pipeline/promotion/graduate",
|
|
json={"name": "s_shadow", "passphrase": "毕业口令"},
|
|
headers=_h(token))
|
|
assert r.status_code == 422
|
|
assert "周报数据不足" in str(r.json())
|
|
|
|
|
|
def test_graduate_422_gate_fail_te_over(client, token, env):
|
|
_set_shadow_since(env["registry"], "2026-08-20")
|
|
_put_weekly(env["pipeline_db"], "s_shadow", te=0.06, fill=0.99, vs=0.0)
|
|
r = client.post("/api/v1/pipeline/promotion/graduate",
|
|
json={"name": "s_shadow", "passphrase": "毕业口令"},
|
|
headers=_h(token))
|
|
assert r.status_code == 422
|
|
body = r.json()
|
|
assert body["detail"]["error"].startswith("影子毕业 gate 未过")
|
|
# GateCheck 字段名=key,TE 项键=shadow_te_annual_max(brief 原文 name/te_annual 与仓库现状对齐)
|
|
assert any(c["key"] == "shadow_te_annual_max" and not c["passed"]
|
|
for c in body["detail"]["checks"])
|
|
|
|
|
|
def test_graduate_200_vs_zero_passes(client, token, env):
|
|
"""vs_backtest=0.0(完美贴合)必须过 gate——旧代码 `0.0 or 1.0`=1.0 误判超阈(本修 RED 主证)."""
|
|
_set_shadow_since(env["registry"], "2026-08-20")
|
|
_put_weekly(env["pipeline_db"], "s_shadow", te=0.01, fill=0.99, vs=0.0)
|
|
r = client.post("/api/v1/pipeline/promotion/graduate",
|
|
json={"name": "s_shadow", "passphrase": "毕业口令"},
|
|
headers=_h(token))
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is True # 返回体={ok,issue}(无 name 键,brief 原文对齐仓库现状)
|
|
|
|
|
|
def test_graduate_twice_second_422(client, token, env):
|
|
"""顺序双发:第一次毕业成功后 registry 已 live,第二次必须 422(并发锁的状态语义)."""
|
|
_set_shadow_since(env["registry"], "2026-08-20")
|
|
_put_weekly(env["pipeline_db"], "s_shadow", te=0.01, fill=0.99, vs=0.001)
|
|
body = {"name": "s_shadow", "passphrase": "毕业口令"}
|
|
r1 = client.post("/api/v1/pipeline/promotion/graduate", json=body,
|
|
headers=_h(token))
|
|
assert r1.status_code == 200
|
|
r2 = client.post("/api/v1/pipeline/promotion/graduate", json=body,
|
|
headers=_h(token))
|
|
assert r2.status_code == 422
|
|
|
|
|
|
def test_todos_graduate_candidate(client, token, env):
|
|
_set_shadow_since(env["registry"], "2026-08-20")
|
|
_put_weekly(env["pipeline_db"], "s_shadow", te=0.01, fill=0.99, vs=0.001)
|
|
r = client.get("/api/v1/pipeline/todos", headers=_h(token))
|
|
assert r.status_code == 200
|
|
ids = [item["id"] for item in r.json()["items"]]
|
|
assert "grad-s_shadow" in ids
|
|
|
|
|
|
# —— 追加:研判卡端点(P3 第一刀) ——
|
|
|
|
def _write_monthly(monthly_dir, as_of, is_decay, verdicts=None, points=None):
|
|
doc = {"as_of": as_of,
|
|
"collective": {"median": 1.2 if is_decay else 2.8,
|
|
"floor": 1.5, "is_collective_decay": is_decay, "n": 4},
|
|
"verdicts": verdicts or {},
|
|
"monthly_points": points or {}}
|
|
with open(os.path.join(monthly_dir, f"host_{as_of}.json"), "w",
|
|
encoding="utf-8") as f:
|
|
json.dump(doc, f, ensure_ascii=False)
|
|
|
|
|
|
def test_research_card_null_paths(client, token, env):
|
|
assert client.get("/api/v1/pipeline/research-card",
|
|
headers=_h(token)).json() == {"card": None}
|
|
_write_monthly(env["monthly"], "2026-09-30", False)
|
|
assert client.get("/api/v1/pipeline/research-card",
|
|
headers=_h(token)).json() == {"card": None}
|
|
|
|
|
|
def test_research_card_shape(client, token, env, tmp_path, monkeypatch):
|
|
# 裁决记录隔离到 tmp:卡上 ruling 断言不依赖 cwd 下 data/ 落盘
|
|
monkeypatch.setenv("SANGUO_VERDICT_RECORDS",
|
|
str(tmp_path / "verdict_records.jsonl"))
|
|
_write_monthly(env["monthly"], "2026-09-30", True,
|
|
verdicts={"fa": {"state": "alert", "ratio": 0.4},
|
|
"fb": {"state": "yellow", "ratio": 0.8},
|
|
"fc": {"state": "ok", "ratio": 0.6}},
|
|
points={"fa": [{"month": "2026-09", "t": 0.5}]})
|
|
card = client.get("/api/v1/pipeline/research-card",
|
|
headers=_h(token)).json()["card"]
|
|
assert card["id"] == "collective-2026-09"
|
|
assert card["triggeredAt"] == "2026-09-30"
|
|
assert card["medianT"] == 1.2 and card["threshold"] == 1.5
|
|
assert card["topDecayFactors"] == ["fa", "fc", "fb"] # alert 优先,ratio 升序
|
|
assert [o["level"] for o in card["options"]] == list(range(7))
|
|
assert card["ruling"] is None
|
|
assert "功课累积中" in card["replay"]["after3m"][0]["outcome"]
|
|
assert len(card["replay"]["after3m"]) == 7
|
|
|
|
|
|
def test_research_card_with_replay_doc(client, token, env):
|
|
_write_monthly(env["monthly"], "2026-09-30", True,
|
|
verdicts={"fa": {"state": "alert", "ratio": 0.4}})
|
|
with open(os.path.join(env["monthly"], "verdict_replay.json"), "w",
|
|
encoding="utf-8") as f:
|
|
json.dump({"since": "2025-01", "episodes": [
|
|
{"month": "2026-06", "median": 0.6, "n": 4, "replayable": True,
|
|
"tiers": {"0": {"ret3m": 0.023, "max_dd": -0.051},
|
|
"1": {"ret3m": 0.016, "max_dd": -0.036},
|
|
"2": {"ret3m": 0.031, "max_dd": -0.044},
|
|
"3": None, "4": None, "5": {"ret3m": 0.023, "max_dd": -0.051},
|
|
"6": {"ret3m": 0.0, "max_dd": 0.0}}}]}, f)
|
|
card = client.get("/api/v1/pipeline/research-card",
|
|
headers=_h(token)).json()["card"]
|
|
assert "2026-06" in card["replay"]["period"]
|
|
rows = {r["option"]: r["outcome"] for r in card["replay"]["after3m"]}
|
|
assert rows["档0 不动，继续观察"] == "+2.3%/回撤 -5.1%(直加·估)"
|
|
assert rows["档3 降调仓频率/只出不进"] == "需精确验证(现有回测工具)"
|
|
assert rows["档6 防御模式（最重）"] == "+0.0%/回撤 0.0%(直加·估)"
|
|
|
|
|
|
def test_verdict_post_lifecycle(client, token, env, monkeypatch, tmp_path):
|
|
recs = str(tmp_path / "verdict_records.jsonl")
|
|
monkeypatch.setenv("SANGUO_VERDICT_RECORDS", recs)
|
|
_write_monthly(env["monthly"], "2026-09-30", True)
|
|
r = client.post("/api/v1/pipeline/research-card/verdict",
|
|
headers=_h(token),
|
|
json={"month": "2026-09", "level": 2, "reason": "先挪钱"})
|
|
assert r.status_code == 200 and r.json()["record"]["level"] == 2
|
|
card = client.get("/api/v1/pipeline/research-card",
|
|
headers=_h(token)).json()["card"]
|
|
assert card["ruling"]["month"] == "2026-09"
|
|
assert card["ruling"]["reason"] == "先挪钱"
|
|
assert client.post("/api/v1/pipeline/research-card/verdict",
|
|
headers=_h(token),
|
|
json={"month": "2026-09", "level": 0, "reason": ""}).status_code == 409
|
|
assert client.post("/api/v1/pipeline/research-card/verdict",
|
|
headers=_h(token),
|
|
json={"month": "2026-09", "level": 9, "reason": ""}).status_code == 422
|
|
assert client.post("/api/v1/pipeline/research-card/verdict",
|
|
headers=_h(token),
|
|
json={"month": "2026-9", "level": 1, "reason": ""}).status_code == 422
|
|
|
|
|
|
# —— 追加:归因日报骨架(P3 第一刀) ——
|
|
|
|
def test_attribution_empty_before_weekly(client, token, env):
|
|
doc = client.get("/api/v1/pipeline/attribution",
|
|
headers=_h(token)).json()
|
|
assert doc["period"] == "周报待算"
|
|
assert doc["factorContribution"] == []
|
|
# registry 两策略(s_live/s_shadow)皆无周报行 → 零策略行,仅 IS 五行
|
|
# (无 is_daily 行 → 五行 live/diff 全 "—",细节见 test_attribution_is_rows_empty_when_no_data)
|
|
assert len(doc["liveVsBacktest"]) == 5
|
|
|
|
|
|
def test_attribution_weekly_rows(client, token, env):
|
|
# brief 原文 upsert all_weather/momentum 不在 env registry——端点按 registry
|
|
# 遍历查 latest_weekly,周报表孤行不上报;改以 registry 两策略承接 brief
|
|
# 同款数值,断言值/格式/阈语义逐字不变
|
|
for name, te, fr, vs in [("s_live", 0.031, 0.97, 0.011),
|
|
("s_shadow", 0.06, 0.90, -0.04)]:
|
|
pipeline_store.upsert_weekly(env["pipeline_db"], {
|
|
"strategy_id": name, "week_start": "2026-09-21",
|
|
"te_annual": te, "fill_rate": fr, "vs_backtest": vs,
|
|
"weeks_counted": 1, "computed_at": "2026-09-25T20:45:00"})
|
|
doc = client.get("/api/v1/pipeline/attribution",
|
|
headers=_h(token)).json()
|
|
assert doc["period"] == "2026-09-21"
|
|
rows = {r["metric"]: r for r in doc["liveVsBacktest"]}
|
|
te_row = rows["s_live·跟踪误差(年化)"]
|
|
assert te_row["live"] == "3.10%" and te_row["backtest"] == "阈 5.00%"
|
|
assert te_row["diff"] == "-1.90%"
|
|
fr_row = rows["s_shadow·成交率"]
|
|
assert fr_row["live"] == "90.0%" and fr_row["diff"] == "-5.0%"
|
|
vs_row = rows["s_shadow·周收益差(vs回测)"]
|
|
assert vs_row["diff"] == "超"
|
|
# 无 is_daily 行 → IS 五行如实 "—"(「待接真」占位文案已随 P3 接线移除)
|
|
is_rows = [r for r in doc["liveVsBacktest"]
|
|
if "落地差距" in r["metric"] or "市场漂移" in r["metric"]]
|
|
assert len(is_rows) == 5
|
|
assert all(r["live"] == "—" and r["diff"] == "—" for r in is_rows)
|
|
assert doc["factorContribution"] == []
|
|
|
|
|
|
# —— 追加:归因端点接真(P3 接线棒,spec §4.7) ——
|
|
# IS 五行 diff 口径:is_daily 无 traded_value 列,契约①的 bp 归一(金额/当日成交额)
|
|
# 无分母可除 → 按金额直显(f"{v:+,.2f}元",正=逆风),腿数注在 live 行内补量级语境;
|
|
# 金额 None → live/diff "—"(宁缺毋假)。
|
|
|
|
def test_attribution_is_rows_from_latest_is(client, token, env):
|
|
pipeline_store.upsert_is_daily(env["pipeline_db"], {
|
|
"day": "2026-09-25", "scope": "all", "legs_count": 3,
|
|
"legs_no_anchor": 1, "delay_cost": -1234.56, "impact_cost": 789.0,
|
|
"opportunity_cost": None, "fee_total": -45.67,
|
|
"price_movement": 210.5, "is_total": -280.73,
|
|
"identity_resid": 0.0, "computed_at": "2026-09-25T20:45:00"})
|
|
doc = client.get("/api/v1/pipeline/attribution", headers=_h(token)).json()
|
|
rows = {r["metric"]: r for r in doc["liveVsBacktest"]}
|
|
assert rows["延迟成本（落地差距）"] == {
|
|
"metric": "延迟成本（落地差距）", "live": "-1,234.56元（3腿）",
|
|
"backtest": "—", "diff": "-1,234.56元"}
|
|
assert rows["冲击成本（落地差距）"]["live"] == "789.00元（3腿）"
|
|
assert rows["冲击成本（落地差距）"]["diff"] == "+789.00元"
|
|
assert rows["机会成本（落地差距）"]["live"] == "—" # None 宁缺毋假
|
|
assert rows["机会成本（落地差距）"]["diff"] == "—"
|
|
assert rows["显性费用（落地差距）"]["live"] == "-45.67元（3腿）"
|
|
assert rows["显性费用（落地差距）"]["diff"] == "-45.67元"
|
|
assert rows["价格移动（市场漂移·参考）"]["live"] == "210.50元（3腿）"
|
|
assert rows["价格移动（市场漂移·参考）"]["diff"] == "+210.50元"
|
|
# IS 五行收尾且次序钉死(延迟/冲击/机会/显性/价格移动参考)
|
|
assert [r["metric"] for r in doc["liveVsBacktest"]] == [
|
|
"延迟成本（落地差距）", "冲击成本（落地差距）", "机会成本（落地差距）",
|
|
"显性费用（落地差距）", "价格移动（市场漂移·参考）"]
|
|
|
|
|
|
def test_attribution_is_rows_empty_when_no_data(client, token, env):
|
|
doc = client.get("/api/v1/pipeline/attribution", headers=_h(token)).json()
|
|
is_rows = [r for r in doc["liveVsBacktest"]
|
|
if "落地差距" in r["metric"] or "市场漂移" in r["metric"]]
|
|
assert len(is_rows) == 5 # 行仍在,不 404
|
|
assert all(r["live"] == "—" and r["diff"] == "—"
|
|
and r["backtest"] == "—" for r in is_rows)
|
|
|
|
|
|
def test_attribution_factor_contribution_from_json(client, token, env):
|
|
contrib = [
|
|
{"factor": "fa_pledge_net_chg", "month": "2026-09",
|
|
"contribution_ret": 0.0042, "cum_ret": 0.031, "exposure_avg": 0.35},
|
|
{"factor": "fa_res_target_upside", "month": "2026-09",
|
|
"contribution_ret": -0.0011, "cum_ret": -0.008, "exposure_avg": 0.12}]
|
|
adir = env["attribution"]
|
|
os.makedirs(adir, exist_ok=True)
|
|
with open(os.path.join(adir, "attribution_2026-09-24.json"), "w",
|
|
encoding="utf-8") as f:
|
|
json.dump({"factorContribution": [{"factor": "stale"}]}, f)
|
|
with open(os.path.join(adir, "attribution_2026-09-25.json"), "w",
|
|
encoding="utf-8") as f:
|
|
json.dump({"factorContribution": contrib}, f)
|
|
doc = client.get("/api/v1/pipeline/attribution", headers=_h(token)).json()
|
|
# 09-27 契位修复: 月度长表聚合为前端短表 {factorId, contributionPct}
|
|
# (近 12 月 Σ×100;文件名降序取最新);直通旧契曾致前端 toFixed 整页崩
|
|
assert doc["factorContribution"] == [
|
|
{"factorId": "fa_pledge_net_chg", "contributionPct": 0.42},
|
|
{"factorId": "fa_res_target_upside", "contributionPct": -0.11}]
|
|
|
|
|
|
def test_attribution_factor_contribution_missing_file(client, token, env):
|
|
# env 注入目录存在但空 → [];目录整个不存在同路径 fail-soft 返 [] 不 500
|
|
doc = client.get("/api/v1/pipeline/attribution", headers=_h(token)).json()
|
|
assert doc["factorContribution"] == []
|
|
|
|
|
|
def test_todos_factor_review_candidates(client, token, env, tmp_path, monkeypatch):
|
|
# 三矛盾②:触点②补源——assessable 且近窗 t≥2 → 待办「待晋级评审」;
|
|
# 无 IC 数据/t 不显著/非 assessable 一律不上板(与评审页下拉同门)
|
|
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY",
|
|
str(tmp_path / "factor_registry.yaml"))
|
|
with open(tmp_path / "factor_registry.yaml", "w", encoding="utf-8") as f:
|
|
yaml.safe_dump({"factors": {
|
|
"fa_hot": {"name": "fa_hot", "hypothesis": "H-1",
|
|
"status": "assessable", "versions": []},
|
|
"fa_nodata": {"name": "fa_nodata", "hypothesis": "H-2",
|
|
"status": "assessable", "versions": []},
|
|
"fa_weak": {"name": "fa_weak", "hypothesis": "H-3",
|
|
"status": "assessable", "versions": []},
|
|
"fa_prom": {"name": "fa_prom", "hypothesis": "H-4",
|
|
"status": "promoted", "promotion_t": 3.0,
|
|
"versions": []},
|
|
}}, f, allow_unicode=True)
|
|
monkeypatch.setenv("SANGUO_FACTOR_EVAL_DB",
|
|
str(tmp_path / "no_such_eval.db"))
|
|
from datetime import date
|
|
from sanguo_api.routes_pipeline import set_pipeline_eval_db_path
|
|
from sanguo_factor import eval_store
|
|
inj_db = str(tmp_path / "factor_eval.db")
|
|
eval_store.init_db(inj_db)
|
|
end = date.today().isoformat()
|
|
run_id = eval_store.create_run(inj_db, "monthly_2026-09", "sz", 10, 1,
|
|
"2025-10-01", end, {})
|
|
eval_store.save_results(inj_db, run_id, [
|
|
{"factor": "fa_hot", "category": "fundamental", "expression": "x",
|
|
"metrics": {"1": {"t_stat": 3.4, "ic_mean": 0.03, "count": 240}}},
|
|
{"factor": "fa_weak", "category": "fundamental", "expression": "y",
|
|
"metrics": {"1": {"t_stat": -0.2, "ic_mean": 0.001, "count": 240}}},
|
|
])
|
|
eval_store.finish_run(inj_db, run_id, "done", 1)
|
|
set_pipeline_eval_db_path(inj_db)
|
|
try:
|
|
r = client.get("/api/v1/pipeline/todos", headers=_h(token))
|
|
assert r.status_code == 200
|
|
items = r.json()["items"]
|
|
rev = [i for i in items if i["id"] == "review-fa_hot"]
|
|
assert len(rev) == 1
|
|
assert rev[0]["touchpoint"] == "promotion"
|
|
assert rev[0]["path"] == "/pipeline/review"
|
|
assert rev[0]["severity"] == "warn"
|
|
assert "3.40" in rev[0]["detail"] or "3.4" in rev[0]["detail"]
|
|
ids = [i["id"] for i in items]
|
|
assert "review-fa_nodata" not in ids # 无 IC 数据不上板
|
|
assert "review-fa_weak" not in ids # t<2 不上板
|
|
assert "review-fa_prom" not in ids # 非 assessable 不上板
|
|
finally:
|
|
set_pipeline_eval_db_path("")
|