Simon
36fe018075
- docs/USAGE.md：
· 因子层一行改为「代码注册表投影 + 参数化实例，参数写在名字里以冻结口径」；
· 新增 `GET/POST/PATCH /api/factors`、`GET /api/factors/templates` 与
`/api/condition-fields`、`/fields` 的说明；
· 新增「参数化因子（2026-10）」块：受控范围、键必须写全参数（缺项就靠可改的
默认值兜底 = 追溯改义，所以拒绝）、口径文案按代码收敛、停用 ≠ 删除、
参数化因子也能当过滤条件；
· 自检清单一并更新（pytest 500 条；verify_strategy_workspace 145 项 skip-job；
verify_ui_alignment 8 页 160 项；新增 verify_unit_conversion、verify_factor_params）。
- scripts/：新增 verify_unit_conversion.py（单位只能在给定范围里选 + 界面单位⇄
基准单位换算）、verify_factor_params.py（参数暴露/界面新建/越界拒绝/停用语义，
跑完自动清掉临时因子）；verify_strategy_workspace.py 加 [5.7b] 因子参数化一节，
临时因子的清理挪进 finally（断言中途失败也不给真人库留垃圾）。
- .gitignore：docs/screenshots/ 是临时验证证据，不入库（文件留在磁盘）。
762 lines
43 KiB
Python
762 lines
43 KiB
Python
#!/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()) |