#!/usr/bin/env python """策略研究工作台契约自检（本轮新增能力的端到端验证）。 覆盖「创建策略 → 展开 → 回测 → 选股 → 对比」全链路，以及三处前端契约： - `GET /api/stocks/names`：前端全站股票名称缓存的唯一数据源（形状必须是 {symbol: name}）； - `POST /api/strategies`：说明为空时必须自动补全（需求：策略必须有说明）； - `PUT /api/strategies/{id}`：原地更新且 id/created_at 不变（策略库「编辑」依赖）； - `GET /api/strategies/{id}/describe`、`POST /api/strategies/describe`：说明 + 计算公式； - 回测结果里 `symbol_curves/positions/trades` 必须带 `name`（前端「代码必须配名称」依赖）； - 选股结果 `candidates[].name` 与 `config_snapshot`（选股 → 回测直通依赖它取回当时的规则）； - 页面 SSR：/strategies、/backtest、/experiments 必须 200 且含关键区块； - **归档链路**：`GET /api/experiments` 的 `X-Total-Count` 与 kind/q 过滤、归档详情含 `data_version`/`job_id`、`/experiments/{id}` 归档页 SSR 能渲染、`DELETE` 语义正确。 用法： cd backend && PYTHONPATH=. .venv/bin/python ../scripts/verify_strategy_workspace.py # 跳过长回测（只验接口与页面）： ... --skip-job """ from __future__ import annotations import argparse import json import time import urllib.error import urllib.request API = "http://127.0.0.1:8000" WEB = "http://127.0.0.1:3000" _ok = 0 _bad = 0 def check(cond: bool, label: str, detail: str = "") -> None: global _ok, _bad if cond: _ok += 1 print(f" ✅ {label}" + (f" — {detail}" if detail else ""), flush=True) else: _bad += 1 print(f" ❌ {label}" + (f" — {detail}" if detail else ""), flush=True) def call_raw(method: str, path: str, body: object | None = None, timeout: float = 60.0): """返回 (status, headers, payload)：需要读响应头（X-Total-Count 等）时用。""" data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request( API + path, data=data, method=method, headers={"Content-Type": "application/json"} ) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, {k.lower(): v for k, v in r.headers.items()}, json.loads( r.read().decode() or "null" ) except urllib.error.HTTPError as e: return e.code, {}, e.read().decode()[:300] def call(method: str, path: str, body: object | None = None, timeout: float = 60.0): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request( API + path, data=data, method=method, headers={"Content-Type": "application/json"} ) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, json.loads(r.read().decode() or "null") except urllib.error.HTTPError as e: return e.code, e.read().decode()[:300] def get_text(url: str, timeout: float = 120.0) -> tuple[int, str]: try: with urllib.request.urlopen(url, timeout=timeout) as r: return r.status, r.read().decode("utf-8", "ignore") except urllib.error.HTTPError as e: return e.code, e.read().decode("utf-8", "ignore")[:200] STRATEGY = { "name": f"契约自检-高股息-{int(time.time())}", "description": "", # 故意留空：验证后端自动补全 "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, }, } def main() -> int: p = argparse.ArgumentParser() p.add_argument("--end", default="2024-12-31", help="回测结束日（默认 1 年，控制在 ~2 分钟）") p.add_argument("--skip-job", action="store_true", help="跳过真实回测 Job（只验接口与页面）") args = p.parse_args() created_id = "" try: # ---------- 1. 名称接口 ---------- print("[1] 股票名称接口（全站名称缓存的唯一数据源）", flush=True) st, names = call("GET", "/api/stocks/names") check(st == 200 and isinstance(names, dict), "GET /api/stocks/names 返回 dict", f"HTTP {st}") if isinstance(names, dict): check(len(names) > 5000, "名称条数 > 5000", f"{len(names)} 条") check(names.get("600519.SH") == "贵州茅台", "含 600519.SH 贵州茅台", str(names.get("600519.SH"))) st2, one = call("GET", "/api/stocks/600519.SH") check(st2 == 200, "GET /api/stocks/{symbol} 未被 /names 抢占（路由顺序）", f"HTTP {st2}") # ---------- 2. 创建策略（说明自动补全） ---------- print("[2] 创建策略 + 说明自动补全", flush=True) st, saved = call("POST", "/api/strategies", STRATEGY) check(st == 200 and isinstance(saved, dict), "POST /api/strategies", f"HTTP {st}") if not isinstance(saved, dict): return 1 created_id = saved.get("id") or "" check(bool(created_id), "返回策略 id", created_id) check( bool((saved.get("description") or "").strip()), "说明为空时被自动补全（策略必须有说明）", (saved.get("description") or "")[:80], ) check( saved.get("selection_interval_months") == 6 and saved.get("rebalance_interval_months") == 6, "m/y 字段持久化（策略 JSON blob，无需迁移）", f"m={saved.get('selection_interval_months')} y={saved.get('rebalance_interval_months')}", ) created_at = str(saved.get("created_at") or "") # ---------- 3. 原地更新 ---------- print("[3] 原地更新（策略库「编辑」依赖）", flush=True) upd = dict(STRATEGY) upd["selection"] = {**STRATEGY["selection"], "top_n": 15, "hold_top_x": 10} st, after = call("PUT", f"/api/strategies/{created_id}", upd) check(st == 200, "PUT /api/strategies/{id}", f"HTTP {st}") if isinstance(after, dict): check(after.get("id") == created_id, "id 不变", str(after.get("id"))) check( (after.get("selection") or {}).get("top_n") == 15, "参数已更新（n=15）", str((after.get("selection") or {}).get("top_n")), ) check( str(after.get("created_at") or "") == created_at, "created_at 未被刷新（避免「改一下就排最前」）", f"{created_at} → {after.get('created_at')}", ) st, notfound = call("PUT", "/api/strategies/STG-NOT-EXIST", upd) check(st == 404, "更新不存在的策略 → 404", f"HTTP {st}") # 改名目标必须**每次运行都不同**：曾经用固定名字，脚本被中断（未走到 cleanup）时 # 会留下同名策略，导致下次运行在这里收到正确的 400 重名拒绝、却被误判为失败。 rename_to = f"契约自检-改名-{int(time.time())}" st, dup = call("PUT", f"/api/strategies/{created_id}", {**upd, "name": rename_to}) check(st == 200, "改名成功（未撞车）", f"HTTP {st} → {rename_to}") # 重名应当被拒（400），这是产品行为，必须验到 st, conflict = call("PUT", f"/api/strategies/{created_id}", {**upd, "name": "高股息 Top20（案例口径）"}) check(st == 400, "改成已存在的策略名 → 400（重名保护）", f"HTTP {st}") # ---------- 4. 说明与公式 ---------- print("[4] 说明 / 计算公式（describe_strategy）", flush=True) st, doc = call("GET", f"/api/strategies/{created_id}/describe") check(st == 200 and isinstance(doc, dict), "GET /strategies/{id}/describe", f"HTTP {st}") if isinstance(doc, dict): summary = doc.get("summary") or "" formula = doc.get("formula") or "" check(bool(summary), "summary 非空（一句话说明）", summary[:90]) check("dividend_yield" in formula, "公式含因子名", "dividend_yield") check("dv_ratio" in formula, "公式含过滤条件字段", "dv_ratio") check("30" in formula, "公式含条件阈值", "30") check( any(k in formula for k in ("佣金", "印花", "滑点", "commission")), "公式含成本项", ) check(bool(doc.get("steps")), "steps 非空（执行步骤）", f"{len(doc.get('steps') or [])} 步") spec_probe = { "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}], # 注意：allow_substitute 与 defer_buy 互斥；只给 defer_buy 会因后端默认 # allow_substitute=True 触发 422（这正是在前端表单里用三态单选表达的原因） "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, "min_commission": 5}, "initial_capital": 1000000, "period": ["2020-01-01", args.end], } st, doc2 = call("POST", "/api/strategies/describe", spec_probe) check(st == 200 and bool((doc2 or {}).get("formula")), "POST /strategies/describe（未保存参数也可预览）", f"HTTP {st}") # 互斥校验必须仍然生效（前端三态单选正是为避免踩到它） bad = dict(spec_probe) bad["selection"] = {"top_n": 20, "allow_substitute": True, "defer_buy": True} st, _ = call("POST", "/api/strategies/describe", bad) check(st == 422, "allow_substitute 与 defer_buy 同真被拒（前端用三态单选规避）", f"HTTP {st}") # ---------- 5. 展开 + 回测 ---------- print("[5] 展开为 Spec" + ("" if args.skip_job else " + 真实回测 Job"), flush=True) st, spec = call( "POST", f"/api/strategies/{created_id}/expand", {"period": ["2024-01-02", args.end], "initial_capital": 1000000}, ) check(st == 200 and isinstance(spec, dict), "POST /strategies/{id}/expand", f"HTTP {st}") if isinstance(spec, dict): check(spec.get("selection", {}).get("top_n") == 15, "展开沿用了更新后的参数", str(spec.get("selection"))) if not args.skip_job and isinstance(spec, dict): st, job = call("POST", "/api/jobs", spec) check(st == 200, "POST /api/jobs（策略 → 回测）", f"HTTP {st}") job_id = (job or {}).get("job_id") if isinstance(job, dict) else None if job_id: t0 = time.monotonic() out = None while time.monotonic() - t0 < 900: _, out = call("GET", f"/api/jobs/{job_id}") if (out or {}).get("status") in ("success", "failed", "cancelled"): break time.sleep(6) status = (out or {}).get("status") check(status == "success", f"回测 Job 终态 success（耗时 {time.monotonic()-t0:.0f}s）", str(status)) res = ((out or {}).get("result") or {}) if isinstance(out, dict) else {} curves = res.get("symbol_curves") or [] pos = res.get("positions") or [] trades = res.get("trades") or [] check(bool(curves), "结果含个股曲线", f"{len(curves)} 条") check( all(c.get("name") for c in curves[:5]), "symbol_curves[].name 已填充（前端代码必须配名称）", str([c.get("name") for c in curves[:3]]), ) check( bool(pos) and all(p.get("name") for p in pos[:5]), "positions[].name 已填充", str([p.get("name") for p in pos[:3]]), ) check( all(t.get("name") for t in trades[:5]), "trades[].name 已填充", str([t.get("name") for t in trades[:3]]), ) # ---------- 6. 选股 + 直通契约 ---------- print("[6] 选股结果的 name 与「选股 → 回测」直通契约", flush=True) st, sel = call( "POST", "/api/selections", { "universe": {"exclude_st": True, "min_listing_days": 250}, "as_of": "2024-07-01", "method": "score", "factors": [{"name": "dividend_yield", "weight": 1}], "conditions": [{"field": "dv_ratio", "op": "lte", "value": 30}], "top_n": 20, }, timeout=300, ) check(st == 200 and isinstance(sel, dict), "POST /api/selections", f"HTTP {st}") if isinstance(sel, dict): run = sel.get("result", sel) cands = run.get("candidates") or [] check(bool(cands), "选股返回候选", f"{len(cands)} 只") check( all(c.get("name") for c in cands[:5]), "candidates[].name 已填充", str([c.get("name") for c in cands[:3]]), ) snap = run.get("config_snapshot") or {} check( bool(snap.get("factors")) and "top_n" in snap and "conditions" in snap, "config_snapshot 含规则（回测页据此预填参数）", f"keys={sorted(snap.keys())[:6]}", ) sid = sel.get("selection_id") if sid: st, again = call("GET", f"/api/selections/{sid}") check(st == 200, "GET /selections/{id} 可读回（直通按钮依赖）", f"HTTP {st}") # ---------- 6.5 归档（存档）链路 ---------- print("[6.5] 回测存档：列表过滤 / 总数 / 详情元数据 / 归档页 / 删除", flush=True) st, exps = call("GET", "/api/experiments?limit=200") check(st == 200 and isinstance(exps, list), "GET /api/experiments 列表", f"HTTP {st}") check(bool(exps), "已有归档记录", f"{len(exps) if isinstance(exps, list) else 0} 条") exp_id = "" if isinstance(exps, list) and exps: exp_id = exps[0].get("id") or "" st, headers, _ = call_raw("GET", "/api/experiments?limit=1") total = headers.get("x-total-count") check( total is not None and total.isdigit() and int(total) >= 1, "列表通过 X-Total-Count 暴露总数（不再静默截断在 50）", f"X-Total-Count={total}", ) st, only_bt = call("GET", "/api/experiments?kind=backtest&limit=200") check( st == 200 and isinstance(only_bt, list) and all(e.get("kind") == "backtest" for e in only_bt), "kind=backtest 过滤生效", f"{len(only_bt) if isinstance(only_bt, list) else '?'} 条", ) st, none_hit = call("GET", "/api/experiments?q=zzz-no-such-experiment") check( st == 200 and isinstance(none_hit, list) and len(none_hit) == 0, "q= 过滤生效（不存在的关键词 → 0 条）", f"{len(none_hit) if isinstance(none_hit, list) else '?'} 条", ) if exp_id: st, det = call("GET", f"/api/experiments/{exp_id}") check(st == 200 and isinstance(det, dict), f"GET /api/experiments/{exp_id} 详情", f"HTTP {st}") if isinstance(det, dict): check("job_id" in det, "详情含来源作业 id", str(det.get("job_id"))) check("data_version" in det, "详情含数据快照指纹字段", str(det.get("data_version"))) check(bool(det.get("spec")), "详情含归档 spec（复现依据）") res = det.get("result") or {} meta = res.get("archive_meta") or {} if meta: check( meta.get("curves_total") is not None, "结果含 archive_meta（曲线存储完整度）", f"stored={meta.get('curves_stored')} total={meta.get('curves_total')} " f"truncated={meta.get('truncated')}", ) check( not meta.get("truncated") or bool(meta.get("curves_stored")), "若被裁剪则如实标注（不静默丢曲线）", str(meta.get("truncated")), ) else: check(False, "结果含 archive_meta（曲线存储完整度）", "缺失：归档未带完整度元数据") # ---------- 7. 页面 SSR ---------- print("[7] 页面可访问性与关键区块", flush=True) pages = [ ("/strategies", ["策略库", "计算公式", "一句话说明"]), ("/backtest", ["选股回测", "保存为策略", "计算公式"]), ("/experiments", ["实验对比", "参数"]), ] if exp_id: # 归档页必须能回答「选股条件」与「交易执行依据」——这是本页存在的理由 pages.append((f"/experiments/{exp_id}", ["选股条件", "交易执行依据", "归档"])) for path, keywords in pages: code, html = get_text(WEB + path) check(code == 200, f"{path} HTTP 200", str(code)) missing = [k for k in keywords if k not in html] check(not missing, f"{path} 含关键区块", f"缺失 {missing}" if missing else "全部命中") # ---------- 7.1 归档页按类型逐类验证 ---------- # 为什么单列一节：归档结果的**结构随 kind 变化**（backtest / factor_test / selection）， # 只验回测归档会漏掉「非回测归档按回测字段渲染 → 整页 500」这类问题（真实踩过： # 4 条 factor_test + 1 条 selection 归档从列表点进去全部 500）。这里对库里 # **每一种**出现的归档类型各取一条真实归档验证：必须 200，且非回测类型不得 # 出现回测专属区块（净值曲线 / 交易执行依据），必须出现该类型自己的区块。 print("[7.1] 归档页按 kind 逐类验证（防结构错配 500）", flush=True) _, listed = call("GET", "/api/experiments?limit=200") by_kind: dict[str, str] = {} for row in listed if isinstance(listed, list) else []: by_kind.setdefault(str(row.get("kind")), str(row.get("id"))) if not by_kind: check(False, "归档列表可用于逐类验证", "列表为空（先跑一次回测/因子测试）") for kind, aid in sorted(by_kind.items()): code, html = get_text(f"{WEB}/experiments/{aid}") check(code == 200, f"归档页 {kind}（{aid}）HTTP 200", str(code)) if code != 200: continue if kind == "backtest": need = ["交易执行依据", "整体收益趋势"] forbid: list[str] = [] elif kind == "factor_test": need = ["因子测试配置", "IC"] forbid = ["交易执行依据（撮合价", "整体收益趋势"] elif kind == "selection": need = ["选股条件", "选出"] forbid = ["交易执行依据（撮合价"] else: need, forbid = ["归档"], [] missing = [k for k in need if k not in html] check(not missing, f"归档页 {kind} 含该类型专属区块", f"缺失 {missing}" if missing else "全部命中") leaked = [k for k in forbid if k in html] check( not leaked, f"归档页 {kind} 不出现回测专属口径", f"误出现 {leaked}（会让非回测归档看起来像跑过调仓成交）" if leaked else "未出现", ) finally: if created_id: st, _ = call("DELETE", f"/api/strategies/{created_id}") print(f"[cleanup] 删除自检策略 {created_id} → HTTP {st}", flush=True) print(f"\n结果：{_ok} 项通过 / {_bad} 项失败", flush=True) return 1 if _bad else 0 if __name__ == "__main__": raise SystemExit(main())