qlib/scripts/verify_backtest_page_contract.py

"""前端页面契约验证：按回测页实际下发的请求体与读取路径校验后端字段，防结构漂移。

基准是页面参数模型的「高股息案例预设」（`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())