#!/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、/fields、/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 detail_of(payload) -> str: """从响应里取人类可读的错误说明。 `call()` 在 HTTPError 分支返回的是**原文串**（不是 dict），所以这里必须兼容两种形态 —— 直接对 str 调 .get() 会把自检脚本自己搞崩（真实踩过）。 """ if isinstance(payload, dict): return str(payload.get("detail") or payload) return str(payload) 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] # 选股策略（2026-09 重构后只含「怎么选」；资金/持仓/调仓/费率/区间移到回测组合与公共配置） STRATEGY = { "name": f"契约自检-高股息-{int(time.time())}", "description": "", # 故意留空：验证后端自动补全 "universe": {"exclude_st": True, "min_listing_days": 250}, "factors": [{"name": "dividend_yield", "weight": 1}], "conditions": [{"field": "dv_ratio", "op": "lte", "value": 30}], } # 回测组合（引用上面的选股策略 + 回测参数）—— 用于 [5] 的组合回测 Job COMBO = { "name": f"契约自检-组合-{int(time.time())}", "strategy_ids": [], # 创建策略后回填 "initial_capital": 1000000, "hold_count": 15, "hold_min_days": 0, "hold_max_days": 20, "rebalance_freq": "monthly", "period": ["2024-01-02", "2024-12-31"], } 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], ) # 选股策略不应持久化任何回测执行参数（重构核心约束） for forbidden in ("selection_interval_months", "rebalance_interval_months", "selection", "costs", "rebalance"): check( forbidden not in saved, f"选股策略不存回测参数字段 {forbidden}", str(saved.get(forbidden)), ) created_at = str(saved.get("created_at") or "") # ---------- 3. 原地更新 ---------- print("[3] 原地更新（策略库「编辑」依赖）", flush=True) upd = dict(STRATEGY) upd["conditions"] = [{"field": "dv_ratio", "op": "lt", "value": 15}] 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"))) conds = after.get("conditions") or [] check( len(conds) == 1 and conds[0].get("value") == 15, "条件已更新（dv_ratio < 15）", str(conds), ) # 选股策略不应再带回测执行参数字段（重构核心约束） for forbidden in ("selection", "costs", "rebalance", "price_adjustment"): check(forbidden not in after, f"选股策略不含回测参数字段 {forbidden}", str(after.get(forbidden))) 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( "dv_ratio" in formula or "≤" in formula or "<=" in formula or "30" in formula, "说明体现过滤条件", formula[:80].replace("\n", " "), ) check( "选股" in summary or "因子" in formula, "说明体现选股口径（选股策略不讲成本/成交，那些在回测组合里）", summary[:60], ) 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. 回测组合（取代旧的 /expand） ---------- print("[5] 回测组合" + ("" if args.skip_job else " + 真实组合回测 Job"), flush=True) # /expand 已随重构移除（回测改由组合驱动） st, _ = call( "POST", f"/api/strategies/{created_id}/expand", {"period": ["2024-01-02", args.end], "initial_capital": 1000000}, ) check(st in (404, 405), "/strategies/{id}/expand 已移除（回测改走 /api/combos）", f"HTTP {st}") combo = dict(COMBO) combo["strategy_ids"] = [created_id] combo["period"] = ["2024-01-02", args.end] st, saved_combo = call("POST", "/api/combos", combo) check(st == 200 and isinstance(saved_combo, dict), "POST /api/combos（保存组合）", f"HTTP {st}") combo_id = (saved_combo or {}).get("id") if isinstance(saved_combo, dict) else None check(bool(combo_id), "返回组合 id", str(combo_id)) # 组合库（列表 / 详情 / 原地更新）—— 前端「已保存的回测组合」卡片依赖这些读路径 st, combo_list = call("GET", "/api/combos") check(st == 200 and isinstance(combo_list, list), "GET /api/combos（组合库列表）", f"HTTP {st}") if isinstance(combo_list, list) and combo_id: check(any(c.get("id") == combo_id for c in combo_list), "列表含刚保存的组合", str(combo_id)) st, detail = call("GET", f"/api/combos/{combo_id}") check( st == 200 and (detail or {}).get("hold_count") == COMBO["hold_count"], "GET /api/combos/{id} 回读参数一致", f"HTTP {st}", ) # 原地更新：description 往返（前端「载入到表单 → 改 → 更新组合」链路） upd_combo = {**combo, "description": "契约自检：更新后的组合说明", "hold_count": 8} st, updated = call("PUT", f"/api/combos/{combo_id}", upd_combo) check( st == 200 and (updated or {}).get("description") == "契约自检：更新后的组合说明", "PUT /api/combos/{id} 说明可往返（组合库编辑依赖）", f"HTTP {st}", ) check( (updated or {}).get("hold_count") == 8, "PUT 更新持仓数生效（不是静默忽略）", str((updated or {}).get("hold_count")), ) # 未知字段必须 422：拼错键名被静默忽略 = 用户以为设上了、其实没生效（AGENT 禁止降级） st, _ = call("POST", "/api/combos/run", {**combo, "capital": 1}) check(st == 422, "组合含未知字段 → 422（不静默降级）", f"HTTP {st}") st, _ = call("POST", "/api/strategies", {**STRATEGY, "costs": {"commission_rate": 0.001}}) check(st == 422, "策略含旧版回测参数 → 422（不静默丢弃）", f"HTTP {st}") # ---------- 5.5 公共配置（全局唯一） ---------- print("[5.5] 公共配置（费率/印花税/滑点/复权，回测组合运行时快照）", flush=True) st, cfg = call("GET", "/api/config") check(st == 200 and isinstance(cfg, dict), "GET /api/config", f"HTTP {st}") if isinstance(cfg, dict): check(cfg.get("id") == "default", "公共配置是单例（id=default）", str(cfg.get("id"))) for key in ( "commission_rate", "stamp_tax_rate", "slippage_rate", "min_commission", "price_adjustment", "benchmark", ): check(key in cfg, f"公共配置含 {key}", str(cfg.get(key))) st, _ = call("PUT", "/api/config", {"slippage": 0.001}) # 故意写错键名：必须被拒 check(st == 422, "公共配置含未知字段 → 422（拼错键名不静默忽略）", f"HTTP {st}") # 策略库说明不得再出现回测执行词：旧自动文案（「每 6 个月调仓…含佣金…」）已被 # c5d6e7f8a9b0 重算，这里做数据层回归守卫，防止再出现自相矛盾的策略说明。 _, all_st = call("GET", "/api/strategies") stale_desc = [ (s or {}).get("id") for s in (all_st if isinstance(all_st, list) else []) if any( m in ((s or {}).get("description") or "") for m in ("佣金", "印花税", "滑点", "调仓", "复权口径") ) ] check(not stale_desc, "策略库无陈旧说明（不含调仓/成本词）", str(stale_desc)) # ---------- 5.6 字段库（过滤条件的字段目录，2026-10） ---------- print("[5.6] 字段库（条件字段目录：中文名/含义/类型收窄/自定义增删）", flush=True) st, fields = call("GET", "/api/condition-fields") check(st == 200 and isinstance(fields, list) and fields, "GET /api/condition-fields（首次读取自动 seed）", f"HTTP {st}，{len(fields) if isinstance(fields, list) else '?'} 条") by_name = {f.get("name"): f for f in (fields if isinstance(fields, list) else [])} check("dv_ratio" in by_name, "字段库含 dv_ratio", str(sorted(by_name)[:5])) for key in ("close", "ma60", "static.industry", "fundamental.roe", "dividend_yield"): check(key in by_name, f"字段库含 {key}", "缺失") dv = by_name.get("dv_ratio") or {} check(bool(dv.get("label")) and bool(dv.get("description")), "字段有中文名与含义（下拉要显示）", f"{dv.get('label')} / {(dv.get('description') or '')[:24]}") check(dv.get("kind") == "num" and dv.get("unit") == "%", "字段带类型与单位", f"{dv.get('kind')} {dv.get('unit')}") ind = by_name.get("static.industry") or {} check( set(ind.get("ops") or []) == {"eq", "ne", "in", "not_in"}, "文本字段比较符收窄为 等值/集合（不摆出恒为假的 >）", str(ind.get("ops")), ) check( set(dv.get("ops") or []) == {"gt", "gte", "lt", "lte", "eq", "ne"}, "数值字段比较符合法集合", str(dv.get("ops")), ) st, avail = call("GET", "/api/condition-fields/available") check(st == 200 and isinstance(avail, list), "GET /api/condition-fields/available", f"HTTP {st}，{len(avail) if isinstance(avail, list) else '?'} 条") # 单位：只能从注册表给的阶梯里选（界面单位），基准单位不可改 —— 换算在界面层做， # 库里/引擎里永远是基准单位，所以改单位不会让历史策略变义。 mv = by_name.get("total_mv") or {} check(mv.get("base_unit") == "万元", "字段响应带基准单位 base_unit（引擎口径）", str(mv.get("base_unit"))) check( [(u.get("unit"), u.get("factor")) for u in (mv.get("units") or [])] == [("万元", 1.0), ("亿元", 10000.0)], "总市值的可选界面单位 = 万元/亿元（系数 1/10000）", str(mv.get("units")), ) check( [(u.get("unit"), u.get("factor")) for u in (by_name.get("close") or {}).get("units") or []] == [("元", 1.0)], "没有备选单位的字段只有基准单位一项（界面不给选择）", str((by_name.get("close") or {}).get("units")), ) st, _ = call("PUT", "/api/condition-fields/total_mv", {"unit": "亿亿元"}) check(st == 422, "自由文本单位 → 422（单位只在你给的范围内选）", f"HTTP {st}") st, _ = call("PUT", "/api/condition-fields/total_mv", {"unit": "元"}) check(st == 422, "不在该字段阶梯里的单位 → 422（总市值没有「元」这一档）", f"HTTP {st}") st, unit_case = call("PUT", "/api/condition-fields/total_mv", {"unit": "亿元"}) check( st == 200 and (unit_case or {}).get("unit") == "亿元" and (unit_case or {}).get("base_unit") == "万元", "选界面单位 亿元：生效，且基准单位仍是 万元", f"HTTP {st} unit={(unit_case or {}).get('unit')} base={(unit_case or {}).get('base_unit')}", ) st, _ = call("PUT", "/api/condition-fields/total_mv", {"unit": "万元"}) check(st == 200, "单位改回基准单位 万元（自检收尾，不留痕）", f"HTTP {st}") st, _ = call("POST", "/api/condition-fields", {"name": "static.list_date"}) check(st == 422, "新增不可计算的字段 → 422（拒绝伪字段，避免永远选不出股票）", f"HTTP {st}") st, made = call("POST", "/api/condition-fields", {"name": "ps", "label": "自检-市销率"}) check(st == 200 and (made or {}).get("source") == "custom", "新增自定义字段（引擎支持的少用字段）", f"HTTP {st}") st, _ = call("POST", "/api/condition-fields", {"name": "ps"}) check(st == 422, "重复新增 → 422（不静默覆盖）", f"HTTP {st}") st, _ = call("PUT", "/api/condition-fields/ps", {"label": "自检-改名"}) check(st == 200, "编辑中文名/含义", f"HTTP {st}") st, _ = call("PUT", "/api/condition-fields/ps", {"kind": "str"}) check(st == 422, "改 kind → 422（类型是引擎事实，不许改）", f"HTTP {st}") st, _ = call("PUT", "/api/condition-fields/ps", {"enabled": False}) check(st == 200, "停用字段", f"HTTP {st}") _, picker = call("GET", "/api/condition-fields?include_disabled=false") check( "ps" not in {f.get("name") for f in (picker if isinstance(picker, list) else [])}, "停用后不出现在条件选择器里", "", ) st, _ = call("DELETE", "/api/condition-fields/close") check(st == 400, "删除内置字段 → 400（只能停用，否则下次读取又补回来）", f"HTTP {st}") st, _ = call("DELETE", "/api/condition-fields/ps") check(st == 200, "删除自定义字段", f"HTTP {st}") # ---------- 5.7 因子目录（必须是代码注册表的投影，不是可编辑配置） ---------- print("[5.7] 因子目录（注册表投影：口径/方向/回看/依赖列与代码一致）", flush=True) st, factors = call("GET", "/api/factors") check(st == 200 and isinstance(factors, list) and factors, "GET /api/factors", f"HTTP {st}") cat = {f.get("name"): f for f in (factors if isinstance(factors, list) else [])} check("dividend_yield" in cat, "目录含股息率因子（历史 bug：后加的因子曾长期缺失）", str(sorted(cat))) try: from app.quant.factors import ( list_factors as _registry, # noqa: PLC0415 - 自检可用时才导入 ) drift = [] for d in _registry(): row = cat.get(d.name) if row is None: drift.append(f"{d.name}: 缺失") continue for k in ("description", "formula", "brief", "frequency", "lookback", "direction", "requires"): want, got = getattr(d, k), row.get(k) # requires 在库里是 JSON 数组、代码里是 tuple：比语义，不比容器类型 same = list(got or []) == list(want or []) if k == "requires" else got == want if not same: drift.append(f"{d.name}.{k}: 库={got!r} 代码={want!r}") check(not drift, "目录与代码注册表逐字段一致（方向/口径/依赖列不许漂移）", "; ".join(drift[:3])) except ImportError as e: # 没装后端依赖时如实跳过，不假装通过 print(f" ⚠ 跳过注册表比对（{e}）：请用 PYTHONPATH=. 在 backend 下运行", flush=True) # ---------- 5.7b 因子参数化（窗口/方向可编辑，且真生效、可冻结） ---------- print("[5.7b] 因子参数化（模板 + 受控参数 + 参数化键 = 冻结口径）", flush=True) mv = cat.get("momentum_60") or {} check( mv.get("label") == "动量（窗口 60，越高越好）", "目录暴露中文名（含真实参数）", str(mv.get("label")), ) check( (mv.get("params") or {}).get("window") == 60 and mv.get("template") == "momentum", "目录暴露 template 与冻结参数", f"template={mv.get('template')} params={mv.get('params')}", ) specs = {s.get("name"): s for s in (mv.get("param_specs") or [])} check( specs.get("window", {}).get("maximum") == 500 and specs.get("window", {}).get("minimum") == 2 and specs.get("direction", {}).get("choices") == ["higher_is_better", "lower_is_better"], "暴露可编辑参数与允许范围（窗口 2~500、方向二选一）", f"window={specs.get('window')} direction={specs.get('direction')}", ) probe_key = "momentum(window=91,direction=lower_is_better)" try: st, tpls = call("GET", "/api/factors/templates") tpl_names = {t.get("name") for t in (tpls or [])} if isinstance(tpls, list) else set() check( st == 200 and {"momentum", "volatility", "volume_ratio", "dividend_yield"} <= tpl_names, "GET /api/factors/templates（模板：可编辑参数与默认值）", f"HTTP {st} {sorted(tpl_names)[:4]}", ) st, made = call("POST", "/api/factors", {"template": "momentum", "params": {"window": 91, "direction": "lower_is_better"}}) created_key = (made or {}).get("name") if isinstance(made, dict) else None check(st == 201 and created_key == probe_key, "POST /api/factors 新建参数化因子（参数写进名字）", f"HTTP {st} {created_key}") check( (made or {}).get("label") == "动量（窗口 91，越低越好）" and (made or {}).get("lookback") == 91 and (made or {}).get("source") == "custom", "新建行带回中文名/回看/来源（口径由引擎投影）", f"{made.get('label') if isinstance(made, dict) else made}", ) st, dup = call("POST", "/api/factors", {"template": "momentum", "params": {"window": 91, "direction": "lower_is_better"}}) check(st == 422 and "已存在" in detail_of(dup), "同参数不重复创建（422 + 已存在的名字）", f"HTTP {st} {detail_of(dup)[:60]}") for bad_params, why in ( ({"window": 999, "direction": "higher_is_better"}, "窗口越界"), ({"window": 90, "direction": "upper"}, "方向不在枚举里"), ({"window": 90, "direction": "higher_is_better", "foo": 1}, "多给了参数"), ): st, bad = call("POST", "/api/factors", {"template": "momentum", "params": bad_params}) check(st == 422, f"越界/非法参数被拒：{why}", f"HTTP {st} {detail_of(bad)[:50]}") st, bad_tpl = call("POST", "/api/factors", {"template": "no_such", "params": {}}) check(st == 422 and "未知因子模板" in detail_of(bad_tpl), "未知模板被拒", f"HTTP {st}") # 参数真的进了引擎：lookback 与方向都要跟着变 try: # 别名不要叫 _ok：本脚本的通过计数器就叫 _ok，遮蔽它会打印出函数对象 from app.quant.factors import get_factor as _gf # noqa: PLC0415 from app.quant.factors import is_resolvable as _resolvable # noqa: PLC0415 d91, _ = _gf(probe_key) check( d91.lookback == 91 and d91.direction == "lower_is_better" and "91" in d91.formula, "参数化键在引擎里真解析（回看/方向/公式都带上了 91）", f"lookback={d91.lookback} dir={d91.direction}", ) check( _resolvable("momentum(window=91)") is False, "缺参数的短键被拒（不许靠模板默认值兜底）", ) check( _resolvable("momentum_20(window=5,direction=higher_is_better)") is False, "实例名不能再带参数", ) except ImportError as e: print(f" ⚠ 跳过引擎解析比对（{e}）", flush=True) # 参数化因子能当过滤条件（字段库取得到、无单位），并能被策略引用后回读 try: from app.quant.condition_fields import get_field as _getf # noqa: PLC0415 fdef = _getf(probe_key) check( fdef is not None and fdef.kind == "num" and fdef.unit == "", "参数化因子可当过滤条件（kind=num 且不瞎挂单位）", f"{fdef}", ) except ImportError as e: print(f" ⚠ 跳过条件字段比对（{e}）", flush=True) st, probe_st2 = call("POST", "/api/strategies", { **STRATEGY, "name": f"契约自检-参数化因子-{int(time.time())}", "factors": [{"name": probe_key, "weight": 1}], "conditions": [], }) probe2_id = (probe_st2 or {}).get("id") if isinstance(probe_st2, dict) else None check(st == 200 and probe2_id, "参数化因子能被策略引用并落库", f"HTTP {st}") if probe2_id: _, back = call("GET", f"/api/strategies/{probe2_id}") got_f = ((back or {}).get("factors") or [{}])[0].get("name") check(got_f == probe_key, "回读策略：因子名（含参数）原样存回（历史不变义）", str(got_f)) _, doc = call("GET", f"/api/strategies/{probe2_id}/describe") doc_text = "\n".join( [str((doc or {}).get("formula") or "")] + [str(x) for x in ((doc or {}).get("steps") or [])] ) check(probe_key in doc_text, "策略说明书写明所用参数版本（读者要知道窗口是多少）", doc_text[:80]) pd2, _ = call("DELETE", f"/api/strategies/{probe2_id}") print(f"[cleanup] 删除参数化因子自检策略 {probe2_id} → HTTP {pd2}", flush=True) st, off = call("PATCH", "/api/factors", {"name": probe_key, "enabled": False}) check(st == 200 and (off or {}).get("enabled") is False, "PATCH 停用参数化因子", f"HTTP {st}") _, again = call("GET", "/api/factors") still = {f.get("name"): f.get("enabled") for f in (again or []) if isinstance(f, dict)} check(still.get(probe_key) is False, "停用状态不被幂等同步冲掉", str(still.get(probe_key))) st, on = call("PATCH", "/api/factors", {"name": probe_key, "enabled": True}) check(st == 200 and (on or {}).get("enabled") is True, "PATCH 启用（复原）", f"HTTP {st}") st, bi = call("PATCH", "/api/factors", {"name": "momentum_60", "enabled": False}) check(st == 422, "内置实例不能停用（开关由代码决定）", f"HTTP {st}") # （清理统一放在 finally 里，见下方：断言中途失败也不会留垃圾） finally: # 清理：目录没有删除接口（设计如此），按主键直接清；放在 finally 里， # 即使上面的断言抛错也不会给真人库留下「自检因子」 try: from app.infrastructure.persistence.sqlalchemy.models.factor import ( # noqa: PLC0415 FactorDefinitionModel as _M, ) from app.infrastructure.persistence.sqlalchemy.session import ( # noqa: PLC0415 SessionLocal as _SL, ) from sqlalchemy import delete as _delete # noqa: PLC0415 with _SL() as _s: _s.execute(_delete(_M).where(_M.name == probe_key)) _s.commit() print(f"[cleanup] 删除临时参数化因子 {probe_key}", flush=True) except Exception as e: # noqa: BLE001 - 清理失败要显式说出来，别静默留着 print(f" ⚠ 清理临时因子失败（请手工检查 factor_definition）：{e}", flush=True) # 条件payload的两种形态必须能被服务端接受（ref 字段比较 / in 文本集合） st, probe_st = call("POST", "/api/strategies", { **STRATEGY, "name": f"契约自检-条件形态-{int(time.time())}", "conditions": [{"field": "close", "op": "gt", "ref": "ma60"}, {"field": "static.industry", "op": "in", "value": ["银行"]}], }) check(st == 200, "条件支持「字段 vs 字段」(close>ma60) 与「属于」文本集合", f"HTTP {st}") if st == 200 and isinstance(probe_st, dict) and probe_st.get("id"): pd, _ = call("DELETE", f"/api/strategies/{probe_st['id']}") print(f"[cleanup] 删除条件形态自检策略 {probe_st.get('id')} → HTTP {pd}", flush=True) if not args.skip_job and combo_id: st, job = call("POST", f"/api/combos/{combo_id}/run") check(st == 200, "POST /api/combos/{id}/run（组合 → 回测 Job）", 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", ["选股策略库", "新建选股策略", "股票池"]), # 注意：只验 SSR 就能看到的静态文案。「直接运行/载入到表单」是组合列表 # 非空时客户端渲染出来的按钮，SSR 阶段列表还在加载，断言它们会假失败。 ("/backtest", ["回测组合", "选择选股策略", "持仓数 N", "已保存的回测组合"]), ("/settings", ["公共配置", "交易成本与行情口径", "全局唯一"]), # 字段库：说明卡是 SSR 静态内容，可断言；字段表格由客户端拉取后渲染 ("/fields", ["字段库", "打分因子 与 过滤条件 是什么关系", "新增字段"]), # 因子研究：说明「参数可改（新建参数化因子）、口径按代码收敛」必须是 SSR 文案 —— # 参数化之前这里写的是「不可修改」，那句话现在已经不成立（不能留假话） ("/factors", ["因子目录", "投影", "参数", "新建参数化因子"]), ("/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: # 先删引用它的组合（若有），再删策略 stc, combos = call("GET", "/api/combos") if isinstance(combos, list): for cb in combos: if created_id in (cb.get("strategy_ids") or []): sd, _ = call("DELETE", f"/api/combos/{cb.get('id')}") print(f"[cleanup] 删除自检组合 {cb.get('id')} → HTTP {sd}", flush=True) 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())