sanguo/sanguo_vnpy_v2
3
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(api): 归因端点接真——IS 五行读 latest_is+factorContribution 读 attribution JSON（P3 接线棒,spec §4.7） [vps]

Browse Source 
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
Claude Dev
2026-09-26 08:21:54 +08:00
parent e5b523de3f
commit 64ea2dc781
2 changed files with 126 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sanguo_api/routes_pipeline.py
+51 -6
View File
@@ -486,12 +486,49 @@ def research_card_verdict(body: dict) -> dict:
return {"ok": True, "record": rec}
_IS_ROWS = ("延迟成本", "冲击成本", "机会成本", "显性费用") # Perold 四分解(P3 后续接真)
# IS 五行(P3 接线棒):Perold 四分解+价格移动参考行,读偏差日报 is_daily 最新行。
# diff 口径=is_daily 无 traded_value 列,契约①的 bp 归一(金额/当日成交额)无分母
# 可除 → 按金额直显(f"{v:+,.2f}元",正=逆风);腿数注在 live 行内补量级语境。
_IS_METRICS = (
("延迟成本（落地差距）", "delay_cost"),
("冲击成本（落地差距）", "impact_cost"),
("机会成本（落地差距）", "opportunity_cost"),
("显性费用（落地差距）", "fee_total"),
("价格移动（市场漂移·参考）", "price_movement"),
)
def _attribution_dir() -> str:
return os.environ.get("SANGUO_ATTRIBUTION_DIR",
os.path.join("data", "attribution"))
def _latest_factor_contribution() -> list:
"""attribution_*.json 文件名降序最新一份 factorContribution 直通.
缺目录/缺文件/坏 JSON → fail-soft 返 [](warning 留痕,看板不 500 不 404)。
"""
try:
adir = _attribution_dir()
names = sorted((n for n in os.listdir(adir)
if n.startswith("attribution_") and n.endswith(".json")),
reverse=True)
if not names:
return []
with open(os.path.join(adir, names[0]), encoding="utf-8") as f:
doc = json.load(f)
rows = doc.get("factorContribution") if isinstance(doc, dict) else None
return rows if isinstance(rows, list) else []
except (OSError, ValueError) as exc: # 缺目录/坏 JSON/编码错一并按缺数据处理
logger.warning("attribution JSON 读取失败,按缺数据处理返空: %s", exc)
return []
@router.get("/pipeline/attribution")
def attribution() -> dict:
"""归因日报骨架(P3 第一刀):周报三指标真值+IS 四行占位;因子贡献留空待后续刀."""
"""归因日报(P3 接线棒):周报三指标真值+IS 五行读 latest_is(偏差日报)
+factorContribution 读 attribution JSON(CLI 产物直通);价格移动行带「参考」
不进评价,金额字段缺失如实 "—"(宁缺毋假)."""
from sanguo_portfolio import pipeline_store
reg = _load_strategy_registry()
th = pipeline_store.get_effective_thresholds(_pipeline_db())
@@ -522,8 +559,16 @@ def attribution() -> dict:
"diff": ("超" if vs is not None
and abs(vs) > th.shadow_weekly_dev_max else "过")
if vs is not None else "—"})
rows.extend({"metric": f"{label}（落地差距·P3 后续）",
"live": "—", "backtest": "—", "diff": "待接真"}
for label in _IS_ROWS)
latest_is = pipeline_store.latest_is(_pipeline_db())
legs = (latest_is or {}).get("legs_count")
for label, col in _IS_METRICS:
v = (latest_is or {}).get(col)
if v is None:
rows.append({"metric": label, "live": "—", "backtest": "—",
"diff": "—"})
else:
note = f"（{legs}腿）" if legs is not None else ""
rows.append({"metric": label, "live": f"{v:,.2f}元{note}",
"backtest": "—", "diff": f"{v:+,.2f}元"})
return {"period": latest_week or "周报待算", "liveVsBacktest": rows,
"factorContribution": []}
"factorContribution": _latest_factor_contribution()}
tests/portfolio/test_routes_pipeline.py
+75 -6
View File
@@ -22,6 +22,7 @@ def env(tmp_path, monkeypatch):
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)
@@ -29,6 +30,7 @@ def env(tmp_path, monkeypatch):
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:
@@ -51,7 +53,7 @@ def env(tmp_path, monkeypatch):
"paper_account_id": None},
}}, f, allow_unicode=True)
return {"pipeline_db": pipeline_db, "registry": registry, "monthly": monthly,
"issue_dir": issue_dir}
"issue_dir": issue_dir, "attribution": attribution_dir}
@pytest.fixture(scope="module")
@@ -510,9 +512,9 @@ def test_attribution_empty_before_weekly(client, token, env):
headers=_h(token)).json()
assert doc["period"] == "周报待算"
assert doc["factorContribution"] == []
# registry 两策略(s_live/s_shadow)皆无周报行 → 零策略行,仅 IS 四行占位
# (brief 原文 7 与其自身实现矛盾:无周报即 continue,凑不出 7;按实现真实语义取 4)
assert len(doc["liveVsBacktest"]) == 4
# 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):
@@ -536,6 +538,73 @@ def test_attribution_weekly_rows(client, token, env):
assert fr_row["live"] == "90.0%" and fr_row["diff"] == "-5.0%"
vs_row = rows["s_shadow·周收益差(vs回测)"]
assert vs_row["diff"] == "超"
is_rows = [r for r in doc["liveVsBacktest"] if "落地差距" in r["metric"]]
assert len(is_rows) == 4 and all(r["diff"] == "待接真" for r in is_rows)
# 无 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()
assert doc["factorContribution"] == contrib # 直通+文件名降序取最新
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"] == []