# -*- coding: utf-8 -*- """ 第十批单测: 静默失败专项 (零外部依赖, 不连库) ================================================= 运行: python scripts/test_batch10_units.py 2026-07-31 专项排查的产物。这一批守的不是某一条业务规则, 而是一类**故障形态**: 失败长得像成功。 本系统里大量写入函数「失败不抛异常, 只回 {"ok": False, ...} 或 0 行」。返回值一丢, 写入没发生, 而调用方照常往下走、照常回 ok=True、页面照常显示"已完成"。八条实例: 1 取不到现价时安全垫按 0 记 → 凭空触发保垫减仓 (卖出方向不设确认门槛, 直接出手) 2 不追高闸的当日涨幅恒为 None → 自主买入这道闸从来没真正跑过 3 清仓命令悄悄漏掉无价的票, 嘴上还说"全部 N 只" 4 参数表读不到时 HALT 开关按 False 放行 (fail-open) 5 连续天数的"上次推进日"键漏在白名单外 → 按日推进形同虚设 6 行业减仓算错分母 → 卖完仍超限却报 DONE 7 命令撤在途指令只改本端状态, 下游子单原封不动继续成交 8 信号去重键在落库**之前**就烧掉 → 落库失败后这条信号当天再也不会重来 [A] 组是这批里唯一一条**静态**用例: 它不测行为, 它扫源码, 守住"关键路径不许丢返回值" 这条纪律本身 —— 新加的调用点一旦又把返回值丢了, 这里立刻红。 """ import ast import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) CASES = [] def case(name): def deco(fn): CASES.append((name, fn)) return fn return deco ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # ================================================================ # [A] 关键路径禁止丢弃返回值 (静态扫描) # ================================================================ # 这些函数**写失败时不抛异常**, 只把失败写在返回值里。调用点必须接住。 # key = 函数名, value = 允许的调用者模块名 (None = 不限, 用于 _db(f, ...) 这种间接调用) SOFT_FAIL = { "set_param": {"param_store"}, # → {"ok": False, "error": ...} "save_neg_streak": {"portfolio"}, # → {"ok": False, "error": ...} "bump_once_guards": {"proposal_service"}, # → {"ok": False, "error": ...} "cancel_instruction": {"executor"}, # → {"ok": False, "message"|"error": ...} "dispatch": {"dispatcher"}, # → {"ok": False, "error": ...} "cancel": {"dispatcher"}, # → {"ok": False, "error": ...} "update_order": {"qmt_repo"}, # → 影响行数, 0 = 出口表里没这行 } # 确实可以丢的调用点写在这里, **必须带理由**。空着比乱加强。 ALLOWED = { # "app/xxx.py:123": "理由", } def _discarded_calls(path: str) -> list: """找出「整条语句就是一次调用、返回值没被任何人接住」的软失败调用。""" with open(path, encoding="utf-8") as f: tree = ast.parse(f.read(), path) out = [] for node in ast.walk(tree): if not isinstance(node, ast.Expr): # 表达式语句 = 返回值直接扔掉 continue call = node.value if isinstance(call, ast.Await): call = call.value if not isinstance(call, ast.Call): continue fn = call.func cand = [] if isinstance(fn, ast.Attribute): cand.append((getattr(fn.value, "id", None), fn.attr)) elif isinstance(fn, ast.Name) and fn.id == "_db" and call.args: # await _db(qmt_repo.update_order, ...) —— 线程池里跑的同一件事 a = call.args[0] if isinstance(a, ast.Attribute): cand.append((getattr(a.value, "id", None), a.attr)) for mod, attr in cand: if attr in SOFT_FAIL and (mod is None or mod in SOFT_FAIL[attr]): out.append((node.lineno, f"{mod}.{attr}")) return out @case("[A1] 关键路径禁止丢弃返回值: app/ 全库无「调用了却不看结果」的软失败写入") def _(): bad = [] for d, _dirs, files in os.walk(os.path.join(ROOT, "app")): for f in sorted(files): if not f.endswith(".py"): continue p = os.path.join(d, f) rel = os.path.relpath(p, ROOT) for lineno, what in _discarded_calls(p): if ALLOWED.get(f"{rel}:{lineno}"): continue bad.append(f"{rel}:{lineno} 丢弃了 {what}() 的返回值") assert not bad, ("以下调用点把「写失败」的返回值扔了 —— 写不进去时调用方一无所知, " "会照常报成功:\n " + "\n ".join(bad)) @case("[A2] 扫描器本身有效: 造一个丢返回值的调用, 必须扫得出来") def _(): # 守住守卫。扫描器写错 (比如 AST 节点类型判断反了) 会让 A1 永远绿, 那比没有更糟。 import tempfile src = ("def f(param_store, dispatcher, x):\n" " param_store.set_param('K', 1, 'test')\n" # ← 该被抓 " r = param_store.set_param('K', 2, 'test')\n" # ← 接住了, 放行 " if dispatcher.dispatch(x)['ok']:\n" # ← 用上了, 放行 " return r\n") with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as fh: fh.write(src) tmp = fh.name try: hits = _discarded_calls(tmp) assert [h[1] for h in hits] == ["param_store.set_param"], hits assert hits[0][0] == 2, hits finally: os.unlink(tmp) @case("[A3] 软失败清单没漏: 名单里的函数确实是「不抛异常只回 ok=False」") def _(): from app.services import param_store, portfolio, proposal_service # set_param: 不可修改的键 → 回 ok=False 而不是抛 r = param_store.set_param("PROXY_DB_URL", "x") assert isinstance(r, dict) and r.get("ok") is False and r.get("error"), r # 另外两个的失败分支在 [E]/[F] 组用桩验证, 这里只钉住"返回的是 dict 不是 None" assert portfolio.save_neg_streak.__doc__, "save_neg_streak 必须写明返回成败" assert proposal_service.bump_once_guards.__doc__ # ================================================================ # [B] 开关命令: 参数没写进去就不许报"已完成" # ================================================================ @case("[B1] HALT_BUY 参数写入失败 → 命令置 CANCELLED 并回 ok=False (不能报已完成)") def _(): from test_wiring import install_fakes from app.services import command_service, param_store from app.core import command_spec as cs install_fakes(prices={}) orig = param_store.set_param try: param_store.set_param = lambda k, v, by="user": {"ok": False, "error": "库挂了"} r = command_service.issue("HALT_BUY", {}, issued_by="test") # issue 内部即时排方案 # 刹车没踩上, 就不许说"已完成" assert r["ok"] is False, r assert r["status"] == cs.ST_CANCELLED, r assert "库挂了" in (r["errors"][0] if r["errors"] else ""), r assert any("开关未生效" in str(n) for n in r["plan"]["notes"]), r["plan"]["notes"] finally: param_store.set_param = orig @case("[B2] 参数写得进去时 HALT_BUY 照常 DONE (修复不能把正常路堵死)") def _(): from test_wiring import install_fakes from app.services import command_service, param_store from app.core import command_spec as cs fake = install_fakes(prices={}) r = command_service.issue("HALT_BUY", {}, issued_by="test") assert r["ok"] and r["status"] == cs.ST_DONE, r assert param_store.get("PMS_GLOBAL_BUY_HALT") is True, fake.params @case("[B3] instant 命令自身失败 (撤销一个不存在的命令) → CANCELLED 而不是 DONE") def _(): from test_wiring import install_fakes from app.services import command_service from app.core import command_spec as cs install_fakes(prices={}) r = command_service.issue("CANCEL_COMMAND", {"target_command_id": "CMD_NOPE_0001"}, issued_by="test") assert r["ok"] is False and r["status"] == cs.ST_CANCELLED, r assert "不存在" in str(r["errors"]), r # ================================================================ # [C] 撤在途指令: 下游拒了必须露出来 # ================================================================ @case("[C1] 撤单被下游拒 → 不计入 cancelled, 顶到 notes 第一条") def _(): from test_wiring import install_fakes from app.services import command_service, executor install_fakes(prices={}) calls = [] def fake_cancel(iid, reason=None): calls.append(iid) return {"ok": False, "message": f"下游拒绝撤单: {iid}"} orig = executor.cancel_instruction try: executor.cancel_instruction = fake_cancel items = [{"ts_code": "600000.SH", "action": "HALT", "cancel_instruction_id": "INS_A"}, {"ts_code": "000001.SZ", "action": "HALT", "cancel_instruction_id": "INS_B"}] r = command_service._cancel_marked_instructions(items) assert calls == ["INS_A", "INS_B"], calls # 真的调了下游, 不是只改本端 assert r["cancelled"] == [], r assert [x["instruction_id"] for x in r["failed"]] == ["INS_A", "INS_B"], r assert "下游拒绝撤单" in r["failed"][0]["why"], r finally: executor.cancel_instruction = orig @case("[C2] 部分成功: cancelled 数的是真撤掉的那些, 不是点名的条数") def _(): from test_wiring import install_fakes from app.services import command_service, executor install_fakes(prices={}) orig = executor.cancel_instruction try: executor.cancel_instruction = ( lambda iid, reason=None: {"ok": iid == "INS_OK", "message": "no"}) r = command_service._cancel_marked_instructions( [{"cancel_instruction_id": "INS_OK"}, {"cancel_instruction_id": "INS_BAD"}]) assert r["cancelled"] == ["INS_OK"] and len(r["failed"]) == 1, r finally: executor.cancel_instruction = orig @case("[C3] 方案生成器的 notes 措辞是「点名撤销」而不是「已撤销」") def _(): from app.core import planner as pl r = pl.plan_halt_buy(pending_buys=[{"ts_code": "600000.SH", "qty": 100, "instruction_id": "INS_1"}]) assert r["items"][0]["cancel_instruction_id"] == "INS_1", r # planner 还不知道撤没撤成, 不许承诺结果 assert "点名" in r["notes"][0], r["notes"] # ================================================================ # [D] 组合刹车: 该踩没踩上必须报错 # ================================================================ def _fake_totals(mv, equity, source="ws", why=""): """positions_view 的最小桩: 只喂刹车结算要看的那几个字段。""" return lambda **kw: {"held": [], "positions": [], "params": {}, "totals": {"portfolio_mv": mv, "total_asset": equity, "cash_source": source, "cash_why": why}} @case("[D1] PMS_BRAKE_UNTIL 写入失败 → 盘前准备报 errors, 不许 ok=True") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, param_store, portfolio install_fakes(prices={}) orig_set, orig_view = param_store.set_param, portfolio.positions_view try: # 高水位 100 万, 总资产 80 万 → 回撤 20%, 远超默认 5%, 必须踩刹车 portfolio.positions_view = _fake_totals(mv=700000.0, equity=800000.0) param_store.set_param = lambda k, v, by="user": ( {"ok": True} if k == "PMS_HIGH_WATER" else {"ok": False, "error": "库挂了"}) orig_get = param_store.get_float param_store.get_float = lambda k, d=0.0: (1000000.0 if k == "PMS_HIGH_WATER" else orig_get(k, d)) try: b = ls._settle_brake() finally: param_store.get_float = orig_get assert b["engaged"] is False, b # 没踩上就不能说踩上了 assert any("刹车未生效" in w for w in b.get("warnings", [])), b finally: param_store.set_param, portfolio.positions_view = orig_set, orig_view @case("[D2] 高水位写入失败 → 回撤按旧高点算并留 warning, 不静默") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, param_store, portfolio install_fakes(prices={}) orig_set, orig_view = param_store.set_param, portfolio.positions_view try: portfolio.positions_view = _fake_totals(mv=100000.0, equity=500000.0) param_store.set_param = lambda k, v, by="user": {"ok": False, "error": "库挂了"} b = ls._settle_brake() assert b["high_water"] == 0.0, b # 没写进去就不许当成写进去了 assert any("高水位" in w for w in b.get("warnings", [])), b finally: param_store.set_param, portfolio.positions_view = orig_set, orig_view @case("[D3] 高水位按总资产算: **主动减仓不该被当成回撤**") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, param_store, portfolio install_fakes(prices={}) orig_view = portfolio.positions_view try: # 100 万全在持仓, 总资产 100 万 → 高水位 100 万 portfolio.positions_view = _fake_totals(mv=1000000.0, equity=1000000.0) ls._settle_brake() assert param_store.get_float("PMS_HIGH_WATER") == 1000000.0 # 一条降仓 40% 的命令执行完: 市值 60 万、现金 40 万, 总资产没变。 # 按市值算会判成"回撤 40%"直接刹停 3 天 —— 用户按纪律减了仓, 系统当他亏了钱。 portfolio.positions_view = _fake_totals(mv=600000.0, equity=1000000.0) b = ls._settle_brake() assert b["drawdown"] == 0.0 and b["engaged"] is False, b # 真亏钱才该刹: 总资产掉到 92 万 → 回撤 8% ≥ 5% portfolio.positions_view = _fake_totals(mv=520000.0, equity=920000.0) b = ls._settle_brake() assert abs(b["drawdown"] - 0.08) < 1e-9 and b["engaged"] is True, b finally: portfolio.positions_view = orig_view @case("[D3b] 清空账本后不许误刹车 (账本清了但高水位还在的那一幕)") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, param_store, portfolio install_fakes(prices={}) orig_view = portfolio.positions_view try: # 2026-07-31 实机那一幕: 账本里躺过 1100 股 → 高水位被顶起来, 然后账本被清空。 # 按市值算的话回撤 100%, 清个账就把自主增持刹停 3 天。按总资产算, 钱还在账户里, # 回撤是 0 —— 这才对。 param_store.set_param("PMS_HIGH_WATER", 10461.0, "test") param_store.set_param("PMS_BRAKE_UNTIL", 0, "test") portfolio.positions_view = _fake_totals(mv=0.0, equity=2000000.0) b = ls._settle_brake() assert b["drawdown"] == 0.0 and b["engaged"] is False, b # 旧口径的高水位是市值量级, 必然小于总资产 → 第一次跑就被抬上来, 不需要迁移 assert b["high_water"] == 2000000.0, b finally: portfolio.positions_view = orig_view @case("[D3c] 资金快照不可信 → 整轮跳过刹车结算, 且**必须说出来**") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, portfolio install_fakes(prices={}) orig_view = portfolio.positions_view try: # cash_est 是 scale−市值 的虚数, 拿它当权益, 减仓照样会让"权益"跳水 —— # 换个地方犯同一个错。宁可这一轮不判, 但不判这件事不能没人知道。 portfolio.positions_view = _fake_totals( mv=600000.0, equity=None, source="estimate", why="ws 从未回过 funds 快照 (对端未实现 query_funds)") b = ls._settle_brake() assert b["skipped"] is True and b["engaged"] is False, b assert b["drawdown"] is None, b assert any("本轮无回撤保护" in w and "query_funds" in w for w in b.get("warnings", [])), b finally: portfolio.positions_view = orig_view @case("[D4] reset_ledger 的归零清单必须盖住这四个账本派生参数") def _(): import re src = open(os.path.join(ROOT, "scripts", "reset_ledger.py"), encoding="utf-8").read() m = re.search(r"RESET_PARAMS\s*=\s*\(([^)]*)\)", src, re.S) assert m, "reset_ledger 里找不到 RESET_PARAMS" got = set(re.findall(r'"([A-Z_]+)"', m.group(1))) need = {"PMS_RECON_STREAK", "PMS_RECON_STREAK_YMD", "PMS_HIGH_WATER", "PMS_BRAKE_UNTIL"} assert need <= got, f"清账时漏了这些派生参数: {need - got}" # ================================================================ # [E] 安全垫连负天数: 写不上要让日终结算整体报失败 # ================================================================ @case("[E1] save_neg_streak 失败 → 返回 ok=False (不再吞成 warning 回 None)") def _(): from test_wiring import install_fakes from app.services import portfolio from app.repo import pms_repo install_fakes(prices={}) orig = pms_repo.set_param try: def boom(*a, **kw): raise RuntimeError("库挂了") pms_repo.set_param = boom r = portfolio.save_neg_streak({"600000.SH": 3}) assert r["ok"] is False and "库挂了" in r["error"], r finally: pms_repo.set_param = orig @case("[E2] 连负天数写不上 → daily_settle 的 ok 必须是 False") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, portfolio install_fakes(prices={}) orig = portfolio.save_neg_streak try: portfolio.save_neg_streak = lambda m: {"ok": False, "error": "库挂了"} out = ls.daily_settle() assert out["ok"] is False, out assert any("安全垫" in e for e in out["errors"]), out["errors"] finally: portfolio.save_neg_streak = orig # ================================================================ # [F] 一次性守卫计数器: 没写上要留痕 # ================================================================ @case("[F1] update_position 影响 0 行 → bump_once_guards 回 ok=False") def _(): from test_wiring import install_fakes from app.services import proposal_service from app.repo import pms_repo install_fakes(prices={}) orig = pms_repo.update_position try: pms_repo.update_position = lambda code, **kw: 0 # 持仓行不存在 r = proposal_service.bump_once_guards("600000.SH", "FILL") assert r["ok"] is False and "0 行" in r["error"], r # 不涉及计数器的动作不该被误判成失败 assert proposal_service.bump_once_guards("600000.SH", "TRIM")["ok"] is True finally: pms_repo.update_position = orig @case("[F2] 计数器没写上 → 评审账本留一条 WARN 痕 (指令仍落表, 但纪律失效要有人知道)") def _(): from test_wiring import install_fakes from app.services import proposal_service from datetime import datetime fake = install_fakes(prices={"600000.SH": 10.0}) orig = proposal_service.bump_once_guards try: proposal_service.bump_once_guards = ( lambda code, act, hn=None, now=None: {"ok": False, "fields": {"fill_count": 1}, "error": "库挂了"}) iid = proposal_service._make_instruction( {"ts_code": "600000.SH", "action": "FILL", "side": "buy", "qty": 100, "reason": "测试", "hard_numbers": {}}, 10.0, datetime.now()) assert iid, iid warn = [x for x in fake.ledger if x.get("verdict") == "WARN" and "一次性守卫" in str(x.get("reason"))] assert len(warn) == 1, fake.ledger assert warn[0]["ref_id"] == iid, warn finally: proposal_service.bump_once_guards = orig # ================================================================ # [G] 信号去重键: 落库成功之后才算用掉 # ================================================================ @case("[G1] 落指令抛异常 → 当天的去重键不许被烧掉 (下一跳还能重来)") def _(): from test_wiring import install_fakes from app.services import signal_service from app.core import signal_rules as sr install_fakes(prices={"600000.SH": 10.0}) seen, out = set(), {"ignored": 0, "recorded": 0, "exits": [], "proposals": [], "errors": []} sig = {"msg_id": "M1", "ts_code": "600000.SH", "action": "SELL", "confidence": 0.95, "source": "test", "reason": "风控"} view = {"positions": [{"ts_code": "600000.SH", "total_qty": 1000, "avail_qty": 1000, "price": 10.0}]} prm = {} d = sr.digest(sig, view["positions"][0], prm) if d["action"] != sr.ACT_EXIT: return # 规则口径变了就跳过, 不假装测到了 key = sr.dedup_key(sig, 20260731) orig = signal_service._make_exit try: def boom(*a, **kw): raise RuntimeError("库挂了") signal_service._make_exit = boom try: signal_service._handle(sig, view, prm, seen, 20260731, False, out) except RuntimeError: pass assert key not in seen, ("落库失败却把去重键用掉了 —— 这条风控卖出信号今天" "再也不会被消化, 而页面只多一行 error") finally: signal_service._make_exit = orig @case("[G2] 落成功后去重键照常生效 (修复不能把去重关掉)") def _(): from test_wiring import install_fakes from app.services import signal_service from app.core import signal_rules as sr install_fakes(prices={"600000.SH": 10.0}) seen, out = set(), {"ignored": 0, "recorded": 0, "exits": [], "proposals": [], "errors": []} sig = {"msg_id": "M1", "ts_code": "600000.SH", "action": "SELL", "confidence": 0.95, "source": "test", "reason": "风控"} view = {"positions": [{"ts_code": "600000.SH", "total_qty": 1000, "avail_qty": 1000, "price": 10.0}]} d = sr.digest(sig, view["positions"][0], {}) if d["action"] != sr.ACT_EXIT: return orig = signal_service._has_inflight try: signal_service._has_inflight = lambda c: False signal_service._handle(sig, view, {}, seen, 20260731, False, out) assert len(out["exits"]) == 1, out assert sr.dedup_key(sig, 20260731) in seen, seen signal_service._handle(sig, view, {}, seen, 20260731, False, out) assert len(out["exits"]) == 1 and out["ignored"] == 1, out # 第二次被去重挡掉 finally: signal_service._has_inflight = orig # ================================================================ # [H] 日终结算: 对账拒绝时不许报 ok=True # ================================================================ @case("[H1] reconcile 回 ok=False → daily_settle 必须 ok=False 并说明原因") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls install_fakes(prices={}) orig = ls.reconcile try: ls.reconcile = lambda **kw: {"ok": False, "errors": ["两个源都无应答"], "diffs": [], "fixes": []} out = ls.daily_settle() assert out["ok"] is False, out assert any("对账未完成" in e and "无应答" in e for e in out["errors"]), out["errors"] finally: ls.reconcile = orig @case("[H2] 对账连续不一致升到 ERROR → daily_settle 同样不许报成功") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls install_fakes(prices={}) orig = ls.reconcile try: ls.reconcile = lambda **kw: {"ok": True, "severity": "ERROR", "streak": 3, "diffs": [{"ts_code": "600000.SH"}], "fixes": []} out = ls.daily_settle() assert out["ok"] is False, out assert any("连续 3 日" in e for e in out["errors"]), out["errors"] finally: ls.reconcile = orig # ================================================================ # [I] 取不到现价的票: 不许拿成本价冒充, 更不许凭空算出安全垫 # ================================================================ @case("[I1] 无价的票 cushion_pct 必须是 None, 不能是 0 (0 会被读成「不赚不亏」)") def _(): from test_wiring import install_fakes from app.services import portfolio fake = install_fakes(prices={"600000.SH": 12.0}) # 000001.SZ 故意没价 for code, cost in (("600000.SH", 10.0), ("000001.SZ", 10.0)): fake.insert_lot(ts_code=code, lot_type="BASE", qty=1000, open_price=cost, open_date="2026-07-01") fake.update_position(code, total_qty=1000, avail_qty=1000, avg_cost=cost) v = portfolio.positions_view() by = {x["ts_code"]: x for x in v["held"]} assert by["600000.SH"]["price_ok"] is True assert abs(by["600000.SH"]["cushion_pct"] - 0.2) < 1e-6, by["600000.SH"] # 取不到价的那只: price 用成本顶着好让市值不塌, 但垫子必须是"不知道" assert by["000001.SZ"]["price_ok"] is False, by["000001.SZ"] assert by["000001.SZ"]["cushion_pct"] is None, by["000001.SZ"] assert "000001.SZ" in v["price_missing"], v.get("price_missing") @case("[I2] 动作引擎跳过无价的票, 且**跳过这件事本身是可见的**") def _(): from app.core import action_engine as ae r = ae.scan(positions=[{"ts_code": "000001.SZ", "total_qty": 1000, "avail_qty": 1000, "avg_cost": 10.0, "price": 10.0, "price_ok": False, "cushion_pct": None, "cushion_peak": 0.0}], params={}, market={}) assert not r["candidates"], r assert any(s["ts_code"] == "000001.SZ" and "取不到现价" in s["why"] for s in r.get("skipped", [])), r @case("[I3] 清仓命令不许悄悄漏掉无价的票, 只数要对得上持仓只数") def _(): from app.core import planner as pl positions = [{"ts_code": "600000.SH", "total_qty": 1000, "avail_qty": 1000, "price": 10.0, "price_ok": True}, {"ts_code": "000001.SZ", "total_qty": 500, "avail_qty": 500, "price": 10.0, "price_ok": False}] r = pl.plan_liquidate_all(positions=positions, pending_buys=[]) codes = {i["ts_code"] for i in r["items"] if i["action"] in ("EXIT", "SELL")} assert codes == {"600000.SH", "000001.SZ"}, r["items"] need = [i for i in r["items"] if i.get("need_price")] assert [i["ts_code"] for i in need] == ["000001.SZ"], need assert r["ok"] is True, r # ================================================================ # [J] 不追高闸: 拿不到数就说拿不到, 不许当成"通过" # ================================================================ @case("[J1] 当日涨幅超上限 → NO_CHASE_DAYUP 拦住 (这道闸得真能拦)") def _(): from app.core import rule_gate as rg r = rg.check(side="buy", action="FILL", qty=100, price=11.0, ctx={"position": {"total_qty": 0, "avail_qty": 0}, "caps": None, "params": {"buy_halt_dayup": 0.05}, "day": {"price": 11.0, "day_chg_from_open": 0.09, "ma5": 11.0}, "flags": {}}) assert any("NO_CHASE_DAYUP" in f for f in r["failed"]), r @case("[J2] 取不到当日涨幅 → 留 DAYUP_MISSING 警示, 绝不当成校验通过") def _(): from app.core import rule_gate as rg r = rg.check(side="buy", action="FILL", qty=100, price=11.0, ctx={"position": {"total_qty": 0, "avail_qty": 0}, "caps": None, "params": {"buy_halt_dayup": 0.05}, "day": {"price": 11.0, "day_chg_from_open": None, "ma5": 11.0}, "flags": {}}) assert not any("NO_CHASE_DAYUP" in f for f in r["failed"]), r assert any("DAYUP_MISSING" in w for w in r["warnings"]), r @case("[J3] 自主提议给规则闸的 day 必须是真行情, 不是拿 price 拼出来的空壳") def _(): import inspect from app.services import proposal_service # 曾经是 {"vwap": price, "day_chg_from_open": None} —— 当日涨幅恒 None, 于是 # "不追高(涨幅)"这一项对所有自主买入从来没有真正跑过。这里直接验行为: 造一只 # 当日大涨的票, 走 _route_one, 规则闸必须拿到真涨幅并拦下来。 from datetime import datetime from test_wiring import install_fakes from app.services import market, proposal_service install_fakes(prices={"600000.SH": 11.0}) orig = market.day_snapshot try: market.day_snapshot = lambda c: {"price": 11.0, "vwap": 10.8, "open": 10.0, "day_chg_from_open": 0.10, "bars": 60} mkt = proposal_service._market_ctx([{"ts_code": "600000.SH"}], datetime.now()) assert mkt["600000.SH"]["day"]["day_chg_from_open"] == 0.10, mkt # 取快照抛异常也不能让整轮扫描崩, 但要留空让规则闸记 DAYUP_MISSING def boom(c): raise RuntimeError("行情库不可用") market.day_snapshot = boom mkt = proposal_service._market_ctx([{"ts_code": "600000.SH"}], datetime.now()) assert mkt["600000.SH"]["day"] == {}, mkt finally: market.day_snapshot = orig # ================================================================ # [K] 参数表读不到时, 安全开关按"拦"而不是按默认值放行 # ================================================================ @case("[K1] 参数表读失败 → HALT 开关 fail-closed 取 True (宁可多拦一轮)") def _(): from test_wiring import install_fakes from app.services import param_store from app.repo import pms_repo install_fakes(prices={}) orig, snap = pms_repo.all_params, dict(param_store._cache) try: def boom(): raise RuntimeError("DB 挂了") pms_repo.all_params = boom param_store.refresh(force=True) assert param_store._cache["error"], param_store._cache assert param_store.get("PMS_GLOBAL_BUY_HALT") is True assert param_store.get("PMS_GLOBAL_EXEC_HALT") is True # 非安全开关不受影响, 照常回初值 —— fail-closed 只用在"拦得住"的地方 assert param_store.get("PMS_AUTONOMY") in ("full", "propose_only", "off") finally: pms_repo.all_params = orig param_store._cache.clear() param_store._cache.update(snap) @case("[K2] 表读得到、只是没设过这个键 → 照常走默认值 (不能把'没设'当成'读不到')") def _(): from test_wiring import install_fakes from app.services import param_store install_fakes(prices={}) # 空参数表, 但**读得到** param_store.refresh(force=True) assert not param_store._cache["error"], param_store._cache assert param_store.get("PMS_GLOBAL_BUY_HALT") is not True, "误伤: 没设过被当成读不到" # ================================================================ # [L] 「没事」不许长得像「出事」: 全空输出要说清是哪一种空 # ================================================================ @case("[L1] 没有 PENDING 方案时, t-mat 要说破是「早就转完了」而不是「一条都没转成」") def _(): from test_wiring import install_fakes from app.services import executor fake = install_fakes(prices={}) fake.plans.extend([ {"id": 1, "plan_id": "P1", "command_id": "C1", "ts_code": "600000.SH", "action": "OPEN", "qty": 100, "status": "EXECUTING", "filled_qty": 0}, {"id": 2, "plan_id": "P2", "command_id": "C1", "ts_code": "600000.SH", "action": "FILL", "qty": 100, "status": "GATED", "filled_qty": 0}]) r = executor.materialize_plans() assert r["created"] == [] and r["skipped"] == [] and r["errors"] == [], r assert r["scanned"] == 0, r assert "不是**失败" in r["note"], r["note"] assert r["plans_by_status"].get("EXECUTING") == 1, r assert r["plans_by_status"].get("GATED") == 1, r @case("[L2] 一条方案都没有时也说得清 (与「转完了」是两回事)") def _(): from test_wiring import install_fakes from app.services import executor install_fakes(prices={}) r = executor.materialize_plans() assert r["scanned"] == 0 and "一条方案都没有" in r["note"], r @case("[L3] 账户空 + 账本空 → 重建预检判 EMPTY 而不是失败 (没活干 ≠ 出事)") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls install_fakes(prices={}) orig = ls.positions_source try: ls.positions_source = lambda: { "source": "ws", "mode": "ws_first", "rows": [], "age_sec": 191.4, "columns": {"qty": None, "avail": None, "cost": None}, "alerts": []} pf = ls.rebuild_preflight() assert pf["verdict"] == "EMPTY", pf assert pf["ok"] is True, pf # 两边都空 = 一致, 不是故障 assert all(s["ok"] for s in pf["steps"]), pf["steps"] assert "可以跳过" in pf["steps"][-1]["why"], pf["steps"] finally: ls.positions_source = orig @case("[L4] 账户空但账本非空 → 顺序反了, 这个才该拦 (别拿空集去核销持仓)") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls fake = install_fakes(prices={"600000.SH": 9.5}) fake.update_position("600000.SH", total_qty=1100, avail_qty=1100, avg_cost=9.273) orig = ls.positions_source try: ls.positions_source = lambda: { "source": "ws", "mode": "ws_first", "rows": [], "age_sec": 10.0, "columns": {"qty": None, "avail": None, "cost": None}, "alerts": []} pf = ls.rebuild_preflight() assert pf["verdict"] == "EMPTY" and pf["ok"] is False, pf assert "先清账本" in pf["steps"][-1]["why"] or "reset_ledger" in pf["hint"], pf finally: ls.positions_source = orig @case("[L6] 规模参数远大于账户资金 → 盘前就说破 (别等规则闸刷一屏 INSUFFICIENT_CASH)") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, portfolio install_fakes(prices={}) orig = portfolio.positions_view try: # 2026-07-31 实测的那组数: scale 200 万 × 上限 70% = 140 万方案空间, 账户只有 98.1 万 portfolio.positions_view = lambda **kw: { "held": [], "positions": [], "params": {}, "totals": {"scale": 2000000.0, "portfolio_mv": 0.0, "total_asset": 981448.56, "cash_source": "ws", "cash_why": ""}} s = ls._scale_vs_account() assert s["warning"] and "INSUFFICIENT_CASH" in s["warning"], s assert s["plan_ceiling"] > s["equity"], s # 规模压到账户以内就不该再吵 portfolio.positions_view = lambda **kw: { "held": [], "positions": [], "params": {}, "totals": {"scale": 1000000.0, "portfolio_mv": 0.0, "total_asset": 981448.56, "cash_source": "ws", "cash_why": ""}} assert not ls._scale_vs_account()["warning"], ls._scale_vs_account() finally: portfolio.positions_view = orig @case("[L7] 拿不到真实资金时不比 (估算值是 scale−市值, 拿它比 scale 是自己跟自己比)") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, portfolio install_fakes(prices={}) orig = portfolio.positions_view try: portfolio.positions_view = lambda **kw: { "held": [], "positions": [], "params": {}, "totals": {"scale": 2000000.0, "portfolio_mv": 0.0, "total_asset": None, "cash_source": "estimate", "cash_why": "ws 没回 funds 快照"}} assert not ls._scale_vs_account()["warning"], ls._scale_vs_account() finally: portfolio.positions_view = orig @case("[L5] 单测清单里的文件缺失不许算通过 (镜像旧了正好整批不存在)") def _(): import re src = open(os.path.join(ROOT, "scripts", "run_tests.py"), encoding="utf-8").read() # 原来缺文件只印一行"跳过"就继续, 于是镜像旧的时候新增那批测试一条没跑、 # 照样 ALL SUITES PASS。这是"没跑"被当成"通过"。 assert "missing" in src and "SUITE MISSING" in src, "run_tests 缺失分支没兜住" assert re.search(r"if missing:\s*\n\s*sys\.exit\(1\)", src), "缺失时没有退非零码" assert "code_fingerprint" in src, "run_tests 要打印代码指纹, 好识别跑的是哪份代码" # ================================================================ def main(): ok = fail = 0 for name, fn in CASES: try: fn() print(f" ok {name}") ok += 1 except Exception as e: print(f" FAIL {name}\n {type(e).__name__}: {e}") fail += 1 print("-" * 62) print(f"通过 {ok} 例, 失败 {fail} 例") if fail: print("BATCH10 FAIL") sys.exit(1) print("BATCH10 PASS") if __name__ == "__main__": main()