"""前端页面契约验证：按回测页实际下发的请求体与读取路径校验后端字段，防结构漂移。 基准是页面参数模型的「高股息案例预设」（`frontend/web/components/StrategyParamsForm.tsx` 的 `CASE_PRESET`）＋ 页面 `run()` 组装出的 spec 形状；结果侧的断言对应 `app/backtest/page.tsx` 的 ResultView（净值/回撤/个股曲线、买卖点、config_snapshot 口径）。 与 `verify_strategy_workspace.py` 的分工： - 本脚本 = 回测结果**结构契约**（跑一次完整 2020→ 区间，约 5 分钟）； - 另一个 = 策略库/说明/名称/选股直通**接口契约**（含一次 1 年回测，约 4 分钟）。 用法：PYTHONPATH=. .venv/bin/python ../scripts/verify_backtest_page_contract.py [--end 2024-12-31] """ from __future__ import annotations import argparse import json import time import urllib.request API = "http://127.0.0.1:8000" def _post(path: str, payload: dict) -> dict: req = urllib.request.Request( API + path, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(req, timeout=120) as r: # noqa: S310 return json.loads(r.read().decode()) def _get(path: str) -> dict: with urllib.request.urlopen(API + path, timeout=120) as r: # noqa: S310 return json.loads(r.read().decode()) def main() -> int: p = argparse.ArgumentParser() p.add_argument("--end", default="2024-12-31") args = p.parse_args() # —— 与 CASE_PRESET（高股息案例预设）逐字段一致；min_listing_days 用 250 —— spec = { "type": "backtest", "universe": {"exclude_st": True, "min_listing_days": 250}, "price_adjustment": "hfq", "factors": [{"name": "dividend_yield", "weight": 1}], "conditions": [{"field": "dv_ratio", "op": "lte", "value": 30}], "selection": { "top_n": 20, "hold_top_x": 20, "allow_substitute": False, "defer_buy": True, }, "rebalance": "monthly", "selection_interval_months": 6, "rebalance_interval_months": 6, "costs": { "commission_rate": 0.0003, "stamp_tax_rate": 0.0005, "slippage_rate": 0.001, "min_commission": 5, }, "initial_capital": 1_000_000, "period": ["2020-01-01", args.end], } print("[contract] POST /api/jobs（页面 submitJob 的请求体）…", flush=True) job = _post("/api/jobs", spec) job_id = job["job_id"] print(f"[contract] job_id={job_id} status={job['status']}") t0 = time.monotonic() while True: out = _get(f"/api/jobs/{job_id}") if out["status"] in ("success", "failed", "cancelled"): break if time.monotonic() - t0 > 1800: print("[contract] 超时") return 1 time.sleep(3) print(f"[contract] 终态 {out['status']}，耗时 {time.monotonic() - t0:.0f}s") if out["status"] != "success": print(f"[contract] 失败：{out.get('error')}") return 1 res = out["result"] s = res["summary"] # —— 页面 ResultView / BacktestMetrics 读取的字段 —— required = [ "summary", "equity_curve", "drawdown", "monthly_returns", "yearly_returns", "positions", "trades", "selection_history", "signal_history", "fills", "symbol_curves", "factor_curves", "turnover_pct", "unimplemented", "config_snapshot", ] missing = [k for k in required if k not in res] assert not missing, f"结果缺少页面读取的字段：{missing}" for k in ( "start", "end", "initial_capital", "final_equity", "total_return_pct", "annual_return_pct", "sharpe", "max_drawdown_pct", "volatility_pct", "win_rate_pct", "total_trades", "avg_turnover_pct", ): assert k in s, f"summary 缺字段 {k}" print(f"[contract] 指标：总收益 {s['total_return_pct']}% · 年化 {s['annual_return_pct']}%" f" · 回撤 {s['max_drawdown_pct']}% · 成交 {s['total_trades']}") # —— 整体收益趋势图 + 买卖点标注（page.tsx equityMarks 的数据源）—— equity_dates = {p["date"] for p in res["equity_curve"]} assert equity_dates, "净值曲线为空" buys = [a for a in res["fills"] if a["signal"] == "BUY"] sells = [a for a in res["fills"] if a["signal"] == "SELL"] assert buys and sells, f"买卖点为空：BUY={len(buys)} SELL={len(sells)}" on_curve = sum(1 for a in buys + sells if a["date"] in equity_dates) assert on_curve == len(buys) + len(sells), "存在落在净值曲线日期之外的买卖点（图上会丢失标注）" print(f"[contract] 净值曲线 {len(equity_dates)} 点；买卖点 BUY={len(buys)} SELL={len(sells)}" f"（全部可落到曲线日期上）") # —— 个股收益率趋势图 + 买卖点（page.tsx SymbolCurveChart 的数据源）—— curves = res["symbol_curves"] assert curves, "个股曲线为空" curve = curves[0] pt_dates = {p["date"] for p in curve["points"]} assert curve["points"], f"{curve['symbol']} 曲线无数据点" assert any(m["signal"] == "BUY" for m in curve["marks"]), "个股曲线缺 BUY 标注" mark_dates = {m["date"] for m in curve["marks"]} assert mark_dates <= pt_dates, f"个股买卖点日期不在曲线点上：{sorted(mark_dates - pt_dates)[:5]}" assert all(m["filled"] for m in curve["marks"]), "个股 marks 含未成交记录" print(f"[contract] 个股曲线 {len(curves)} 只；首只 {curve['symbol']} " f"{len(curve['points'])} 点 / {len(curve['marks'])} 个买卖点，全部落点成功" f"（期末 {curve['final_return_pct']}%）") # —— 页面顶部 Pill 读取口径与 n/x —— sel = res["config_snapshot"]["selection"] basis = res["config_snapshot"]["price_basis"] assert sel["top_n"] == 20 and sel["hold_top_x"] == 20, sel assert basis["adjust_mode"] == "hfq", basis print(f"[contract] config_snapshot.selection={sel}") print(f"[contract] config_snapshot.price_basis={basis}") # —— 买卖理由（「用数据说话」的字段契约）—— # 每个买卖点都必须带结构化理由，且关键数字（名次/综合分/因子值）齐全； # 文案由后端词表生成，前端只展示 —— 这里断言的是「页面要读的字段真的在」。 KNOWN = { "buy_enter_topn", "buy_defer_filled", "buy_skip_limit_up", "buy_skip_halted", "buy_skip_no_cash", "buy_skip_min_commission", "sell_drop_topn", "sell_force_tmax", "sell_defer_tmin", "sell_defer_halted", "sell_defer_limit_down", } missing_reason = [a for a in res["signal_history"] if not a.get("reason")] assert not missing_reason, f"有买卖点没有理由：{missing_reason[:3]}" bad_code = [a["reason"]["code"] for a in res["signal_history"] if a["reason"]["code"] not in KNOWN] assert not bad_code, f"出现词表外的理由代码：{sorted(set(bad_code))}" for a in res["signal_history"]: r = a["reason"] assert r["text"], f"{a['date']} {a['symbol']} 理由文案为空" assert isinstance(r["data"], dict), f"{a['date']} {a['symbol']} 理由缺少结构化数据" # 成交类理由必须能回答「第几名 / 多少候选 / 综合分 / 因子当时的值」 with_rank = [a for a in res["signal_history"] if isinstance(a["reason"]["data"].get("rank"), int)] assert with_rank, "没有任何理由给出名次（名次是买入选股的核心依据）" sample = next(a for a in res["signal_history"] if a["reason"]["code"] == "buy_enter_topn") sd = sample["reason"]["data"] assert sd["rank"] >= 1 and sd["total"] >= sd["rank"] and sd["top_n"] >= 1, sd assert "score" in sd and sd.get("factors"), f"买入理由缺少综合分/因子值：{sd}" print(f"[contract] 买卖理由 {len(res['signal_history'])} 条全部带理由与数字；样例 " f"{sample['date']} {sample['symbol']} {sample['reason']['code']} " f"rank={sd['rank']}/{sd['total']} score={sd['score']} factors={sd['factors']}") # —— 成交明细两端的理由（页面「为什么买 / 为什么卖」两列）—— trades = res["trades"] no_entry = [t for t in trades if not t.get("entry_reason")] no_exit = [t for t in trades if not t.get("exit_reason")] assert not no_entry, f"{len(no_entry)} 笔成交缺建仓理由" assert not no_exit, f"{len(no_exit)} 笔成交缺卖出理由" t0_ = trades[0] print(f"[contract] 成交明细两端理由齐全（{len(trades)} 笔）；样例 {t0_['symbol']} " f"买={t0_['entry_reason']['code']} 卖={t0_['exit_reason']['code']}") # —— 因子曲线（页面「因子曲线」区块 + 放大页的数据源）—— fcs = res["factor_curves"] assert fcs, "结果里没有因子曲线（页面会少一整块「买卖依据的水平」）" for fc in fcs: assert fc["name"] and fc["label"] and fc["direction"], fc assert fc["points"], f"因子 {fc['name']} 曲线无数据点" dates = [p_["date"] for p_ in fc["points"]] assert dates == sorted(dates), f"因子 {fc['name']} 曲线日期未按时间升序" assert len(set(dates)) == len(dates), f"因子 {fc['name']} 曲线有重复日期" used = {f["name"] for f in spec["factors"]} | {"dividend_yield"} names = {fc["name"] for fc in fcs} assert names & used, f"因子曲线与本次策略用到的因子对不上：{names} vs {used}" fc = fcs[0] print(f"[contract] 因子曲线 {len(fcs)} 条；首条 {fc['name']}（{fc['label']}，" f"单位 {fc.get('unit')}，{len(fc['points'])} 点，方向 {fc['direction']}）") # —— 未成交意图卡片（signal_history 中 filled=False 且带原因）—— rejects = [a for a in res["signal_history"] if not a["filled"] and a["reject_reason"]] print(f"[contract] 未成交意图 {len(rejects)} 条，样例：" f"{rejects[0]['date'] if rejects else '—'} {rejects[0]['reject_reason'] if rejects else ''}") print(f"[contract] unimplemented {len(res['unimplemented'])} 条（页面如实展示）") print("[contract] ✅ 页面契约验证通过") return 0 if __name__ == "__main__": raise SystemExit(main())