tradingSystem/scripts/test_wiring.py

# -*- coding: utf-8 -*-
"""
装配自检 (零外部依赖, 不连库)
==============================
运行: python scripts/test_wiring.py

用内存桩替换 repo/行情, 验证「服务层 → 核心逻辑 → 落表」整条链路的接线是否正确:
导入链、API 路由、调度表与守卫、单表访问守卫、交易日历、参数中心回退,
以及命令下达→方案落表→撤销、成交回放→批次入账→成本重算 两条主干流程。
"""
import os
import sys
import traceback
from datetime import date, datetime

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

RESULTS = []


def case(name):
    def deco(fn):
        RESULTS.append((name, fn))
        return fn
    return deco


# ================================================================ 内存桩
class FakeRepo:
    """pms_repo 的内存替身 (只实现服务层用到的函数, 语义与 SQL 版一致)。"""

    def __init__(self):
        self.params, self.commands, self.plans = {}, {}, []
        self.positions, self.lots, self.instructions = {}, [], {}
        self.proposals, self.ledger, self.reports, self.industry = {}, [], {}, {}
        self.cash_flows = []
        self._lot_id = 0

    # --- runtime param ---
    def all_params(self):
        return {k: {"param_value": v, "updated_by": "test", "updated_at": ""}
                for k, v in self.params.items()}

    def set_param(self, key, value, updated_by="user"):
        self.params[key] = str(value)
        return 1

    def get_param(self, key):
        return self.params.get(key)

    # --- command ---
    def next_command_seq(self, ymd):
        return len(self.commands) + 1

    def insert_command(self, *, command_id, cmd_class, cmd_type, ts_code, params, status,
                       issued_by=None, note=None, progress=None):
        kw = {"command_id": command_id, "cmd_class": cmd_class, "cmd_type": cmd_type,
              "ts_code": ts_code, "params": params, "status": status,
              "issued_by": issued_by, "note": note, "progress": progress}
        self.commands[kw["command_id"]] = {
            "id": len(self.commands) + 1, "command_id": kw["command_id"],
            "cmd_class": kw["cmd_class"], "cmd_type": kw["cmd_type"], "ts_code": kw["ts_code"],
            "params": kw["params"], "status": kw["status"], "progress": kw.get("progress") or {},
            "issued_by": kw.get("issued_by"), "issued_at": datetime.now(), "done_at": None,
            "note": kw.get("note")}
        return 1

    def get_command(self, cid):
        return self.commands.get(cid)

    def list_commands(self, *, statuses=None, cmd_class=None, limit=200):
        out = [c for c in self.commands.values()
               if (not statuses or c["status"] in statuses)
               and (not cmd_class or c["cmd_class"] == cmd_class)]
        return sorted(out, key=lambda c: -c["id"])[:limit]

    def update_command(self, cid, *, status=None, progress=None, done_at=None, note=None):
        c = self.commands.get(cid)
        if not c:
            return 0
        if status is not None:
            c["status"] = status
        if progress is not None:
            c["progress"] = progress
        if done_at is not None:
            c["done_at"] = done_at
        if note is not None:
            c["note"] = note
        return 1

    def supersede_param_commands(self, cmd_type, ts_code=None, keep_command_id=None):
        n = 0
        for c in self.commands.values():
            if (c["cmd_class"] == "param" and c["status"] == "EFFECTIVE"
                    and c["cmd_type"] == cmd_type and c["ts_code"] == ts_code
                    and c["command_id"] != keep_command_id):
                c["status"] = "SUPERSEDED"
                n += 1
        return n

    def latest_effective(self, cmd_type, ts_code=None):
        hits = [c for c in self.commands.values()
                if c["cmd_class"] == "param" and c["status"] == "EFFECTIVE"
                and c["cmd_type"] == cmd_type and (not ts_code or c["ts_code"] == ts_code)]
        return sorted(hits, key=lambda c: c["id"])[-1] if hits else None

    def list_effective_stock_params(self, cmd_types=None):
        return [c for c in self.commands.values()
                if c["cmd_class"] == "param" and c["status"] == "EFFECTIVE" and c["ts_code"]
                and (not cmd_types or c["cmd_type"] in cmd_types)]

    # --- plan ---
    def insert_plans(self, rows):
        for r in rows:
            d = dict(r)
            d.setdefault("filled_qty", 0)
            self.plans.append(d)
        return len(rows)

    def list_plans(self, *, command_id=None, statuses=None, ts_code=None, limit=500):
        out = [p for p in self.plans
               if (not command_id or p["command_id"] == command_id)
               and (not statuses or p["status"] in statuses)
               and (not ts_code or p["ts_code"] == ts_code)]
        return sorted(out, key=lambda p: p.get("priority", 100))[:limit]

    def update_plan(self, plan_id, *, status=None, filled_qty=None):
        for p in self.plans:
            if p["plan_id"] == plan_id:
                if status:
                    p["status"] = status
                if filled_qty is not None:
                    p["filled_qty"] = filled_qty
                return 1
        return 0

    def cancel_plans_of_command(self, cid):
        n = 0
        for p in self.plans:
            if p["command_id"] == cid and p["status"] in ("PENDING", "EXECUTING"):
                p["status"] = "CANCELLED"
                n += 1
        return n

    def set_plans_deadline(self, cid, dl):
        for p in self.plans:
            if p["command_id"] == cid:
                p["deadline"] = dl
        return 1

    # --- position / lot ---
    def list_positions(self, *, only_open=False):
        out = list(self.positions.values())
        return [p for p in out if int(p.get("total_qty") or 0) > 0] if only_open else out

    def get_position(self, code):
        return self.positions.get(code)

    def ensure_position(self, code):
        self.positions.setdefault(code, {"ts_code": code, "status": "PLANNED", "total_qty": 0,
                                         "avail_qty": 0, "frozen_reason": "NONE"})
        return 1

    def update_position(self, code, **fields):
        self.ensure_position(code)
        self.positions[code].update(fields)
        return 1

    def bump_position_qty(self, code, *, total_delta=0, avail_delta=0):
        self.ensure_position(code)
        p = self.positions[code]
        p["total_qty"] = max(0, int(p.get("total_qty") or 0) + total_delta)
        p["avail_qty"] = max(0, int(p.get("avail_qty") or 0) + avail_delta)
        return 1

    def reset_avail_all(self):
        for p in self.positions.values():
            p["avail_qty"] = p.get("total_qty", 0)
            p["t0_count_today"] = 0
        return len(self.positions)

    def list_lots(self, ts_code=None, *, status="OPEN", limit=1000):
        out = [l for l in self.lots
               if (not ts_code or l["ts_code"] == ts_code)
               and (not status or l["status"] == status)]
        return sorted(out, key=lambda l: (str(l["open_date"]), l["id"]))[:limit]

    def insert_lot(self, *, ts_code, lot_type, qty, open_price, open_date, instruction_id=None,
                   note=None):
        self._lot_id += 1
        self.lots.append({"id": self._lot_id, "ts_code": ts_code, "lot_type": lot_type,
                          "qty": int(qty), "open_price": float(open_price),
                          "open_date": open_date, "closed_qty": 0, "close_avg_price": None,
                          "realized_pnl": 0.0, "status": "OPEN",
                          "instruction_id": instruction_id, "note": note})
        return 1

    def close_lot_qty(self, lot_id, *, qty, close_price, realized_pnl):
        for l in self.lots:
            if l["id"] == lot_id:
                old_closed = int(l["closed_qty"])
                l["close_avg_price"] = ((float(l["close_avg_price"] or 0) * old_closed
                                         + close_price * qty) / (old_closed + qty))
                l["qty"] = max(0, int(l["qty"]) - qty)
                l["closed_qty"] = old_closed + qty
                l["realized_pnl"] += realized_pnl
                l["status"] = "CLOSED" if l["qty"] <= 0 else "OPEN"
                return 1
        return 0

    def update_lot(self, lot_id, **fields):
        for l in self.lots:
            if l["id"] == lot_id:
                l.update(fields)
                return 1
        return 0

    # --- instruction / proposal / ledger / report / industry ---
    # --- pms_cash_flow (第 14 张表; 费用只进现金账, 绝不摊成本 —— 协议 §5.5) ---
    def insert_cash_flow(self, *, ymd, kind, amount, ts_code=None, estimated=0,
                         trade_no=None, instruction_id=None, note=None):
        self.cash_flows.append({"ymd": ymd, "kind": kind, "amount": float(amount),
                                "ts_code": ts_code, "estimated": int(bool(estimated)),
                                "trade_no": trade_no, "instruction_id": instruction_id,
                                "note": note})
        return 1

    def insert_instruction(self, *, instruction_id, origin_type, origin_id, ts_code, action,
                           side, qty, limit_price=None, window_tdays=3, status="PROPOSED",
                           progress=None):
        kw = {"instruction_id": instruction_id, "origin_type": origin_type,
              "origin_id": origin_id, "ts_code": ts_code, "action": action, "side": side,
              "qty": qty, "limit_price": limit_price, "window_tdays": window_tdays,
              "status": status, "progress": progress or {}}
        kw.setdefault("exec_qty", 0)
        kw["created_at"] = kw["updated_at"] = datetime.now()
        self.instructions[kw["instruction_id"]] = kw
        return 1

    def list_instructions(self, *, statuses=None, side=None, ts_code=None, limit=300):
        out = [i for i in self.instructions.values()
               if (not statuses or i["status"] in statuses)
               and (not side or i.get("side") == side)
               and (not ts_code or i["ts_code"] == ts_code)]
        return out[:limit]

    def get_instruction(self, iid):
        return self.instructions.get(iid)

    # **签名必须跟真 repo 一模一样**, 不许写成 `**kw`。2026-08-03 实机: executor 多传了
    # 一个 limit_price=None, 真 repo 抛 TypeError、单子已发出而本端没记账, 而这里 `**kw`
    # 全收所以单测一路绿。桩比真依赖宽松 = 单测在替真依赖打掩护。
    # test_batch10 的 [M] 组会扫这一条, 别改回 **kw。
    def update_instruction(self, iid, *, status=None, exec_qty=None, exec_avg_price=None,
                           dispatch_ref=None, progress=None):
        i = self.instructions.get(iid)
        if not i:
            return 0
        for k, v in (("status", status), ("exec_qty", exec_qty),
                     ("exec_avg_price", exec_avg_price), ("dispatch_ref", dispatch_ref),
                     ("progress", progress)):
            if v is not None:
                i[k] = v
        i["updated_at"] = datetime.now()
        return 1

    def add_instruction_exec(self, iid, qty):
        i = self.instructions.get(iid)
        if i:
            i["exec_qty"] = int(i.get("exec_qty") or 0) + int(qty)
        return 1

    def insert_proposal(self, *, proposal_id, ts_code, action, qty, hard_numbers, expire_at,
                        judge_verdict=None, judge_reason=None, status="WAIT_USER"):
        kw = {"proposal_id": proposal_id, "ts_code": ts_code, "action": action, "qty": qty,
              "hard_numbers": hard_numbers, "expire_at": expire_at,
              "judge_verdict": judge_verdict, "judge_reason": judge_reason, "status": status}
        kw["status"] = kw.get("status", "WAIT_USER")
        kw["hard_numbers"] = kw.get("hard_numbers") or {}
        self.proposals[kw["proposal_id"]] = kw
        return 1

    def list_proposals(self, *, statuses=("WAIT_USER",), limit=200):
        return [p for p in self.proposals.values() if p["status"] in statuses][:limit]

    def get_proposal(self, pid):
        return self.proposals.get(pid)

    def decide_proposal(self, pid, status):
        p = self.proposals.get(pid)
        if p and p["status"] == "WAIT_USER":
            p["status"] = status
            return 1
        return 0

    def expire_proposals(self, now=None):
        return 0

    def insert_ledger(self, *, ts_code, action, arbiter, verdict, price_at, hard_numbers=None,
                      failed_checks=None, reason=None, ref_id=None):
        kw = {"ts_code": ts_code, "action": action, "arbiter": arbiter, "verdict": verdict,
              "price_at": price_at, "hard_numbers": hard_numbers or {},
              "failed_checks": failed_checks or [], "reason": reason, "ref_id": ref_id,
              "decided_at": datetime.now()}
        self.ledger.append(kw)
        return 1

    def list_ledger(self, *, ts_code=None, limit=200):
        return self.ledger[-limit:]

    def rule_rejected_today(self, since):
        # 内存桩里所有留痕都算"今天"; 只认规则闸的 REJECT (研判结论会变, 不参与当日去重)
        return {(r["ts_code"], r["action"]) for r in self.ledger
                if r.get("verdict") == "REJECT" and r.get("arbiter") == "rule"}

    def upsert_report(self, ymd, report):
        self.reports[int(ymd)] = report
        return 1

    def get_report(self, ymd):
        r = self.reports.get(int(ymd))
        return {"ymd": ymd, "report": r} if r else None

    def latest_report(self):
        if not self.reports:
            return None
        y = max(self.reports)
        return {"ymd": y, "report": self.reports[y]}

    def get_industry(self, code):
        return self.industry.get(code)

    def list_industry(self, limit=5000):
        return [{"ts_code": k, "industry": v} for k, v in self.industry.items()]

    def upsert_industry(self, rows):
        for r in rows:
            self.industry[r["ts_code"]] = r["industry"]
        return len(rows)


class FakeQmtRepo:
    """ws 通道三表的内存替身 (pms_qmt_order / pms_qmt_inbox / pms_ws_state)。

    只实现 dispatcher 用得到的那几个 —— 出栈、落库、seq 水位是常驻进程 (app/ws/runner.py)
    的事, 由 test_batch6_units.py 的纯逻辑用例覆盖; 这里只管「服务层接线对不对」。
    """
    LIVE = ("QUEUED", "SENDING", "SENT", "ACCEPTED", "SUBMITTED", "PARTIAL")

    def __init__(self):
        self.state = {"id": 1, "last_seq": 0, "acked_seq": 0, "server_seq": 0,
                      "conn_state": "INIT", "heartbeat_at": None, "connected_at": None,
                      "resync_flag": 0, "last_error": None, "stat": {}}
        self.orders = {}
        self.inbox = {}                 # seq → 上行消息行 (pms_qmt_inbox)
        self.snapshots = []             # snapshot 上行 (对账的事实源, §5.7)

    def set_channel(self, conn_state="ONLINE", alive=True):
        """摆一个通道状态出来 —— 「进程活没活」和「连接通没通」是两件事。"""
        self.state["conn_state"] = conn_state
        self.state["heartbeat_at"] = datetime.now() if alive else None

    # ---- 以下与真 repo 同名同义 ----
    def get_state(self):
        return dict(self.state)

    def inbox_pending_count(self):
        return 0

    def queue_depth(self):
        d = {}
        for o in self.orders.values():
            d[o["status"]] = d.get(o["status"], 0) + 1
        return d

    def enqueue_order(self, *, instruction_id, parent_id, ts_code, side, qty, limit_price,
                      valid_until, intent="OPEN", note=None):
        self.orders[instruction_id] = {
            "instruction_id": instruction_id, "parent_id": parent_id,
            # 真表的列名是 parent_instruction_id, consume_ws_trades 反查用的是它。
            # 桩里两个键都放, 少一个的话 SMOKE_ 那道闸在单测里永远"看起来没生效"。
            "parent_instruction_id": parent_id, "ts_code": ts_code,
            "side": side, "qty": int(qty), "limit_price": float(limit_price),
            "valid_until": int(valid_until), "intent": intent, "note": note,
            "status": "QUEUED", "cancel_state": "NONE", "cancel_id": None}
        return 1

    def request_cancel(self, *, parent_id, cancel_id):
        n = 0
        for o in self.orders.values():
            if (o["parent_id"] == parent_id and o["cancel_state"] == "NONE"
                    and o["status"] in self.LIVE):
                o.update({"cancel_state": "REQUESTED", "cancel_id": cancel_id})
                n += 1
        return n

    def get_order(self, instruction_id):
        return self.orders.get(instruction_id)

    # ---- pms_qmt_inbox ----
    def put_trade(self, *, seq, instruction_id, ts_code, side, qty, price, fee=0.0,
                  trade_no=None, processed=0):
        """塞一笔 trade 上行。测试专用, 真 repo 里对应的是 runner 的落库路径。"""
        self.inbox[int(seq)] = {
            "seq": int(seq), "msg_type": "trade", "processed": int(processed),
            "process_note": None,
            "payload": {"instruction_id": instruction_id, "ts_code": ts_code, "side": side,
                        "qty": int(qty), "price": float(price),
                        "amount": round(float(price) * int(qty), 2), "fee": float(fee),
                        "trade_no": trade_no or f"T-{seq}"}}
        return self.inbox[int(seq)]

    def inbox_pending(self, limit=500):
        rows = [r for r in sorted(self.inbox.values(), key=lambda x: x["seq"])
                if int(r["processed"]) == 0]
        return [dict(r, payload=dict(r["payload"])) for r in rows[:int(limit)]]

    def inbox_mark(self, seqs, processed=1, note=None):
        n = 0
        for s in (seqs or []):
            if int(s) in self.inbox:
                self.inbox[int(s)].update({"processed": int(processed), "process_note": note})
                n += 1
        return n

    def inbox_orphan_count(self):
        return sum(1 for r in self.inbox.values() if int(r["processed"]) == 3)

    def put_snapshot(self, *, kind="positions", items=None, data=None, age_sec=0, seq=None):
        """塞一条 snapshot 上行 (§5.7)。age_sec 用来造"快照过期"。"""
        seq = int(seq if seq is not None else (max(self.inbox) + 1 if self.inbox else 1000))
        pl = {"kind": kind, "as_of": 1769500000000}
        if items is not None:
            pl["items"] = items
        if data is not None:
            pl["data"] = data
        self.snapshots.append({"seq": seq, "payload": pl, "age_sec": float(age_sec),
                               "received_at": datetime.now()})
        return pl

    def latest_snapshot(self, kind, *, scan=60):
        for s in sorted(self.snapshots, key=lambda x: x["seq"], reverse=True):
            if str(s["payload"].get("kind") or "").lower() == str(kind).lower():
                return dict(s)
        return None


def install_fakes(prices=None, positions=None, params=None, high5=None, prev_close=None):
    """把内存桩装到各模块上, 返回 FakeRepo 实例 (ws 通道桩挂在 .qmt 上)。"""
    from app.repo import downstream_repo, pms_repo, qmt_repo
    from app.services import industry, market, param_store, portfolio

    fake = FakeRepo()
    fake.qmt = FakeQmtRepo()
    for _n in ("get_state", "inbox_pending_count", "queue_depth", "enqueue_order",
               "request_cancel", "get_order", "inbox_pending", "inbox_mark",
               "inbox_orphan_count", "latest_snapshot"):
        setattr(qmt_repo, _n, getattr(fake.qmt, _n))
    fake.params.update(params or {})
    for p in (positions or []):
        base = {"ts_code": p["ts_code"], "status": "HOLDING", "frozen_reason": "NONE",
                "total_qty": 0, "avail_qty": 0, "base_qty": 0, "fill_qty": 0, "add_qty": 0,
                "dca_qty": 0, "t0_qty": 0, "avg_cost": None, "cushion_peak": 0,
                "realized_t_profit": 0, "t0_enabled": 0}
        base.update(p)
        fake.positions[p["ts_code"]] = base

    for mod in (pms_repo,):
        for name in dir(FakeRepo):
            if name.startswith("_"):
                continue
            setattr(mod, name, getattr(fake, name))
    market.get_price = lambda c: (prices or {}).get(c)
    market.get_prices = lambda cs_: {c: (prices or {}).get(c) for c in (cs_ or [])}
    market.get_refs = lambda c, **kw: {"support": None, "pressure": None, "stop": None,
                                       "source": "none"}
    market.get_ma5 = lambda c: (prices or {}).get(c)
    # plan_price 是规划期定价 (实时价 → 昨收兜底)。桩不接库, 一律按"有实时价"处理;
    # 昨收那条路单独给 prev_close 参数的用例覆盖。
    market.get_last_close = lambda c: (prev_close or {}).get(c)
    market.plan_price = lambda c: (
        {"price": (prices or {}).get(c), "source": "realtime"} if (prices or {}).get(c)
        else ({"price": (prev_close or {}).get(c), "source": "prev_close"}
              if (prev_close or {}).get(c) else {"price": None, "source": "none"}))
    market.get_high5 = lambda c: (high5 or prices or {}).get(c)
    market.day_snapshot = lambda c: ({} if not (prices or {}).get(c) else {
        "price": prices[c], "vwap": prices[c], "open": prices[c], "high": prices[c] * 1.02,
        "low": prices[c] * 0.98, "day_chg_from_open": 0.0, "bars": 60})
    industry.get_many = lambda cs_: {c: None for c in (cs_ or [])}
    industry.get = lambda c: None
    industry.ready = lambda: False
    industry.status = lambda: {"source": "", "ready": False, "count": 0, "hint": "test"}
    # 下游只读表默认桩 (单测不触网; 个别用例内再按需覆盖)
    downstream_repo.fetch_buy_plans = lambda **kw: []
    downstream_repo.fetch_filled_orders = lambda **kw: []
    downstream_repo.fetch_positions = lambda: {"rows": [], "columns": {"qty": None},
                                               "raw_count": 0}
    downstream_repo.fetch_refs = lambda c: None
    downstream_repo.category_columns = lambda force=False: {
        "columns": None, "code_col": None, "industry_col": None,
        "error": "单测桩: 未接行业表"}
    from app.repo import industry_repo
    industry_repo.probe = lambda force=False: {"form": None, "error": "单测桩: 未接 gp_hybk",
                                               "columns": None, "tried": []}
    industry_repo.primary_industry_map = lambda codes, level="l3": {}
    industry_repo.invalidate = lambda: None
    downstream_repo.latest_filled_order_id = lambda: "ANCHOR_0"
    # 回放游标预置成非空 —— 否则 replay_fills 会走「冷启动只对齐不追认」那条路 (见
    # ledger_service._seed_cursor), 下面那几个回放用例就测不到入账。冷启动本身另有专门用例。
    # 用 "0" 而不是随便一个字符串: next_cursor 只进不退, 且非数字 id 会退化成字典序比较,
    # 占位值若比真实 order_id 大 (比如 "SEED"), 游标就永远推不动了。
    fake.params.setdefault("PMS_REPLAY_CURSOR", "0")
    # 上游选股计划接口在单测里一律停用 (base 为空 → plan_feed 立即抛 PlanFeedError,
    # 不会发出任何 HTTP 请求)。要测候选池的用例自己 stub plan_feed.candidates。
    fake.params.setdefault("PMS_PLAN_API_BASE", "")
    param_store._cache.update({"at": 0.0, "data": {}, "error": None, "loaded": False})
    portfolio.save_neg_streak({})
    return fake


# ================================================================ 用例
@case("装配·全部模块可导入 (无循环依赖)")
def _():
    import importlib
    for m in ["app.core.sizer", "app.core.cushion", "app.core.command_spec", "app.core.planner",
              "app.core.recon", "app.core.tradedays", "app.db.session", "app.repo.pms_repo",
              "app.repo.downstream_repo", "app.services.param_store", "app.services.industry",
              "app.services.market", "app.services.plan_feed",
              "app.services.portfolio", "app.services.command_service",
              "app.services.ledger_service", "app.web.main", "app.scheduler"]:
        importlib.import_module(m)


@case("装配·API 路由齐全 (四块页面 + 运维)")
def _():
    from app.web.main import app
    paths = {r.path for r in app.routes}
    need = ["/", "/health", "/api/overview", "/api/params", "/api/commands",
            "/api/commands/catalog", "/api/commands/active", "/api/commands/{command_id}",
            "/api/commands/{command_id}/cancel", "/api/plans", "/api/positions",
            "/api/positions/{ts_code}/lots", "/api/instructions", "/api/ledger", "/api/report",
            "/api/proposals", "/api/proposals/{proposal_id}/decide", "/api/ops/replay",
            "/api/ops/reconcile", "/api/ops/daily-settle", "/api/ops/report",
            "/api/ops/downstream-schema", "/api/industry",
            "/api/upstream/plan", "/api/ops/plan-refresh"]
    missing = [p for p in need if p not in paths]
    assert not missing, missing
    import os as _os
    from app.web import main as web
    assert _os.path.exists(_os.path.join(web.STATIC_DIR, "index.html")), "单页文件缺失"


@case("装配·调度表覆盖设计 §10 全部调度位")
def _():
    from app import scheduler as sch
    names = set(sch.celery_app.conf.beat_schedule)
    assert names == {"plan_pull", "premarket", "command_poll", "replay_fills",
                     "intraday_exec", "signal_digest", "t0_close", "daily_settle",
                     "daily_report"}, names
    tasks = set(sch.celery_app.tasks)
    for t in ("pms.premarket", "pms.command_poll", "pms.replay_fills", "pms.daily_settle",
              "pms.daily_report", "pms.t0_close", "pms.intraday_exec", "pms.signal_digest",
              "pms.plan_pull"):
        assert t in tasks, t
    assert sch.in_session(datetime(2026, 7, 27, 10, 0)) is True
    assert sch.in_session(datetime(2026, 7, 27, 12, 0)) is False
    assert sch.in_session(datetime(2026, 7, 27, 14, 59)) is True


@case("装配·调度守卫: 非交易日跳过 / 休假模式跳过 / 异常不外抛")
def _():
    from app import scheduler as sch
    from app.core import tradedays as tdm
    orig = tdm.is_trade_day
    try:
        tdm.is_trade_day = lambda d=None: False

        @sch.guard(trade_day=True)
        def t1():
            raise AssertionError("非交易日不应执行")
        assert t1() == {"skipped": "not_trade_day"}

        tdm.is_trade_day = lambda d=None: True
        install_fakes(params={"PMS_GLOBAL_EXEC_HALT": "true"})

        @sch.guard(trade_day=True)
        def t2():
            raise AssertionError("休假模式不应执行")
        assert t2() == {"skipped": "exec_halt"}

        install_fakes(params={})

        @sch.guard(trade_day=True)
        def t3():
            raise ValueError("boom")
        r = t3()
        assert "error" in r and "boom" in r["error"], r
    finally:
        tdm.is_trade_day = orig


@case("装配·严格单表访问守卫 (JOIN 与多表一律拒绝)")
def _():
    from app.db.session import MultiTableSQL, assert_single_table
    assert_single_table("SELECT * FROM pms_position WHERE ts_code = :c")
    assert_single_table("UPDATE pms_lot SET qty = 0 WHERE id = :id")
    for bad in ["SELECT a.* FROM pms_position a JOIN pms_lot b ON a.ts_code = b.ts_code",
                "SELECT * FROM pms_position, pms_lot",
                "SELECT * FROM pms_plan WHERE ts_code IN (SELECT ts_code FROM pms_position)"]:
        try:
            assert_single_table(bad)
            assert False, f"未拦截: {bad}"
        except MultiTableSQL:
            pass


@case("交易日历·执行窗口与剩余交易日 (跨周末)")
def _():
    from app.core import tradedays as tdm
    fri, mon = date(2026, 7, 24), date(2026, 7, 27)      # 2026-07-24 周五, 07-27 周一
    assert tdm.is_trade_day(fri) and tdm.is_trade_day(mon)
    assert tdm.is_trade_day(date(2026, 7, 25)) is False  # 周六
    assert tdm.next_trade_day(fri, 1) == mon
    assert tdm.window_deadline(mon, 1) == mon
    assert tdm.window_deadline(mon, 3) == date(2026, 7, 29)
    assert tdm.window_deadline(fri, 2) == mon            # 周五起 2 个交易日 = 周一
    assert tdm.trade_days_left(date(2026, 7, 29), mon) == 3
    assert tdm.trade_days_left(date(2026, 7, 20), mon) == 0
    assert tdm.ymd(mon) == 20260727


@case("参数中心·表值优先/文件回退/非法值拒绝")
def _():
    from app.services import param_store
    from config.settings import settings
    fake = install_fakes(params={"PMS_TOTAL_SCALE": "3000000", "PMS_AUTONOMY": "full"})
    assert param_store.get("PMS_TOTAL_SCALE") == 3_000_000.0
    assert param_store.get("PMS_AUTONOMY") == "full"
    assert param_store.get("PMS_STOCK_CAP") == settings.PMS_STOCK_CAP      # 未改写 → 文件初值
    assert param_store.get_bool("PMS_GLOBAL_BUY_HALT") is False            # 运行态默认
    assert param_store.set_param("PROXY_DB_URL", "x")["ok"] is False       # 基础设施不可改
    assert param_store.set_param("PMS_PORTFOLIO_CAP", 1.5)["ok"] is False  # 越界
    assert param_store.set_param("PMS_AUTONOMY", "auto")["ok"] is False    # 枚举非法
    assert param_store.set_param("PMS_PORTFOLIO_CAP", 0.5)["ok"] is True
    assert fake.params["PMS_PORTFOLIO_CAP"] == "0.5"
    assert param_store.get("PMS_PORTFOLIO_CAP") == 0.5
    sp = param_store.sizing_params()
    assert sp["scale"] == 3_000_000.0 and sp["portfolio_cap"] == 0.5


@case("命令服务·参数命令立即生效并覆盖旧记录")
def _():
    from app.services import command_service as csvc, param_store
    fake = install_fakes()
    r1 = csvc.issue("SET_SCALE", {"scale": 2_000_000})
    assert r1["ok"] and r1["status"] == "EFFECTIVE", r1
    assert param_store.get("PMS_TOTAL_SCALE") == 2_000_000.0
    r2 = csvc.issue("SET_SCALE", {"scale": 2_500_000})
    assert r2["ok"] and param_store.get("PMS_TOTAL_SCALE") == 2_500_000.0
    assert fake.commands[r1["command_id"]]["status"] == "SUPERSEDED"
    assert fake.commands[r2["command_id"]]["status"] == "EFFECTIVE"
    bad = csvc.issue("SET_SCALE", {"scale": -1})
    assert not bad["ok"] and any("OUT_OF_RANGE" in e for e in bad["errors"]), bad


@case("命令服务·个股参数命令投影到账本 (冻结/做T/止损价)")
def _():
    from app.services import command_service as csvc
    fake = install_fakes(positions=[{"ts_code": "600000.SH", "total_qty": 1000,
                                     "avg_cost": 10.0}])
    assert csvc.issue("FREEZE_STOCK", {"ts_code": "600000.SH"})["ok"]
    assert fake.positions["600000.SH"]["frozen_reason"] == "COMMAND_HALT"
    assert csvc.issue("T0_ENABLE", {"ts_code": "600000.SH", "t_ratio": "30%"})["ok"]
    assert fake.positions["600000.SH"]["t0_enabled"] == 1
    assert abs(float(fake.positions["600000.SH"]["t0_ratio"]) - 0.30) < 1e-9
    assert csvc.issue("SET_STOP_PRICE", {"ts_code": "600000.SH", "price": 9.2})["ok"]
    assert fake.positions["600000.SH"]["stop_ref"] == 9.2
    assert fake.positions["600000.SH"]["ref_source"] == "user"
    assert csvc.issue("UNFREEZE_STOCK", {"ts_code": "600000.SH"})["ok"]
    assert fake.positions["600000.SH"]["frozen_reason"] == "NONE"
    sp = csvc.effective_stock_params()["600000.SH"]
    assert sp["frozen"] is False and sp["t0"] is True and sp["stop_price"] == 9.2, sp
    # 做T比例超硬上限 1/3 被拒
    assert not csvc.issue("T0_ENABLE", {"ts_code": "600000.SH", "t_ratio": "40%"})["ok"]


@case("命令服务·降仓命令端到端 (方案落表 + 进度 + 撤销)")
def _():
    from app.services import command_service as csvc
    fake = install_fakes(
        prices={"600000.SH": 10.0, "000001.SZ": 8.0},
        params={"PMS_TOTAL_SCALE": "2000000", "PMS_WEAK_NEG_DAYS": "5"},
        positions=[{"ts_code": "600000.SH", "total_qty": 14000, "base_qty": 7000,
                    "fill_qty": 3500, "add_qty": 3500, "avg_cost": 8.93},
                   {"ts_code": "000001.SZ", "total_qty": 10000, "base_qty": 10000,
                    "avg_cost": 8.8}])
    from app.services import portfolio
    portfolio.save_neg_streak({"000001.SZ": 6})

    r = csvc.issue("REDUCE_EXPOSURE", {"pct": "5%", "window_tdays": 3})
    assert r["ok"] and r["status"] == "EXECUTING", r
    cid = r["command_id"]
    plans = fake.list_plans(command_id=cid)
    assert plans, "方案未落表"
    assert {p["action"] for p in plans} <= {"EXIT", "TRIM", "HALT"}, plans
    exit_p = [p for p in plans if p["action"] == "EXIT"]
    assert exit_p and exit_p[0]["ts_code"] == "000001.SZ", plans   # 弱票优先清
    assert all(p["deadline"] for p in plans)
    prog = fake.commands[cid]["progress"]
    assert prog["target_amount"] == 100_000.0 and prog["plan_count"] == len(plans), prog

    # 进度结算: 方案未成交 → 保持 EXECUTING; 全部成交 → DONE
    r2 = csvc.refresh_progress(cid)
    assert r2["commands"][0]["status"] in ("EXECUTING", "PARTIAL"), r2
    for p in plans:
        fake.update_plan(p["plan_id"], status="DONE", filled_qty=p["qty"])
    csvc.refresh_progress(cid)
    assert fake.commands[cid]["status"] == "DONE", fake.commands[cid]

    # 撤销: 已完成命令不可撤
    assert csvc.cancel(cid)["ok"] is False
    r3 = csvc.issue("REDUCE_EXPOSURE", {"pct": "3%"}, force_conflict=True)
    assert csvc.cancel(r3["command_id"])["ok"] is True
    assert fake.commands[r3["command_id"]]["status"] == "CANCELLED"
    assert all(p["status"] == "CANCELLED"
               for p in fake.list_plans(command_id=r3["command_id"]))


@case("命令服务·冲突拦截与强制下达")
def _():
    from app.services import command_service as csvc
    install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                  positions=[{"ts_code": "600000.SH", "total_qty": 14000, "base_qty": 7000,
                              "avg_cost": 9.0}])
    a = csvc.issue("REDUCE_EXPOSURE", {"pct": "5%"})
    assert a["ok"], a
    b = csvc.issue("INCREASE_EXPOSURE", {"pct": "5%"})
    assert not b["ok"] and b["conflicts"], b
    assert any("方向相反" in c["reason"] for c in b["conflicts"])
    c = csvc.issue("INCREASE_EXPOSURE", {"pct": "5%"}, force_conflict=True)
    assert c["ok"] and c["conflicts"], c        # 强制下达但冲突仍留痕


@case("命令服务·全局暂停买入撤在途买入指令并置开关")
def _():
    from app.services import command_service as csvc, param_store
    fake = install_fakes(prices={"600000.SH": 10.0})
    fake.insert_instruction(instruction_id="INS_1", origin_type="plan", origin_id="P1",
                            ts_code="600000.SH", action="OPEN", side="buy", qty=1000,
                            limit_price=10.0, status="DISPATCHED")
    fake.insert_instruction(instruction_id="INS_2", origin_type="plan", origin_id="P2",
                            ts_code="600000.SH", action="EXIT", side="sell", qty=500,
                            limit_price=10.0, status="DISPATCHED")
    r = csvc.issue("HALT_BUY", {})
    assert r["ok"] and r["status"] == "DONE", r
    assert fake.instructions["INS_1"]["status"] == "CANCELLED"
    assert fake.instructions["INS_2"]["status"] == "DISPATCHED"    # 卖出不受影响
    assert param_store.get_bool("PMS_GLOBAL_BUY_HALT") is True
    assert csvc.issue("RESUME_BUY", {})["ok"]
    assert param_store.get_bool("PMS_GLOBAL_BUY_HALT") is False


@case("账本服务·成交回放入账 → 批次/持仓/成本重算")
def _():
    from app.services import ledger_service as ls
    from app.repo import downstream_repo
    fake = install_fakes(prices={"600000.SH": 11.0})
    fake.insert_instruction(instruction_id="INS_A", origin_type="plan", origin_id="P1",
                            ts_code="600000.SH", action="OPEN", side="buy", qty=6000,
                            limit_price=10.0, status="DISPATCHED")
    # 建单时点必须显式钉死: 用 datetime.now() 会让本用例的通过与否取决于跑测试的钟点
    # (时间守卫要求 下发 ≤ 成交), 17:12 跑就会误判成外部成交 —— 实机上已踩到。
    fake.instructions["INS_A"]["created_at"] = "2026-07-27 09:30:00"
    # updated_at 故意置在成交之后: 认领必须看建单时点, 不能看 updated_at
    # (否则部分成交回写一次 updated_at, 后续成交就会被自己挡住)
    fake.instructions["INS_A"]["updated_at"] = "2026-07-27 23:59:59"
    fills = [{"order_id": 101, "ts_code": "600000.SH", "side": "buy", "qty": 6000,
              "price": 10.0, "done_time": "2026-07-27 09:40:00"}]
    orig = downstream_repo.fetch_filled_orders
    try:
        downstream_repo.fetch_filled_orders = lambda **kw: fills
        r = ls.replay_fills()
        assert r["ok"] and r["actions"] == 1, r
        assert fake.positions["600000.SH"]["total_qty"] == 6000
        assert fake.positions["600000.SH"]["avail_qty"] == 0          # T+1: 当日买入不可卖
        assert abs(float(fake.positions["600000.SH"]["avg_cost"]) - 10.0) < 1e-6
        assert fake.positions["600000.SH"]["base_qty"] == 6000
        assert abs(float(fake.positions["600000.SH"]["cushion_pct"]) - 0.10) < 1e-4
        assert fake.instructions["INS_A"]["status"] == "CONFIRMED"
        assert fake.params["PMS_REPLAY_CURSOR"] == "101"
        # 幂等: 游标已推进, 同一批不再重复入账
        downstream_repo.fetch_filled_orders = lambda **kw: []
        r2 = ls.replay_fills()
        assert r2["fills"] == 0 and fake.positions["600000.SH"]["total_qty"] == 6000
    finally:
        downstream_repo.fetch_filled_orders = orig


@case("账本服务·卖出回放按核销次序 + 摊薄成本下降")
def _():
    from app.services import ledger_service as ls
    from app.repo import downstream_repo
    fake = install_fakes(prices={"600000.SH": 12.0})
    fake.insert_lot(ts_code="600000.SH", lot_type="BASE", qty=6000, open_price=10.0,
                    open_date="2026-07-01")
    fake.insert_lot(ts_code="600000.SH", lot_type="ADD", qty=3000, open_price=11.0,
                    open_date="2026-07-20")
    fake.update_position("600000.SH", total_qty=9000, avail_qty=9000)
    fills = [{"order_id": 201, "ts_code": "600000.SH", "side": "sell", "qty": 3000,
              "price": 12.0, "done_time": "2026-07-27 10:00:00"}]
    orig = downstream_repo.fetch_filled_orders
    try:
        downstream_repo.fetch_filled_orders = lambda **kw: fills
        r = ls.replay_fills()
        assert r["ok"], r
        add_lot = [l for l in fake.lots if l["lot_type"] == "ADD"][0]
        assert add_lot["qty"] == 0 and add_lot["status"] == "CLOSED"   # ADD 先核销
        base_lot = [l for l in fake.lots if l["lot_type"] == "BASE"][0]
        assert base_lot["qty"] == 6000                                 # 底仓保留
        p = fake.positions["600000.SH"]
        assert p["total_qty"] == 6000 and p["avail_qty"] == 6000
        # 摊薄成本 = (60000+33000-36000)/6000 = 9.5
        assert abs(float(p["avg_cost"]) - 9.5) < 1e-6, p["avg_cost"]
    finally:
        downstream_repo.fetch_filled_orders = orig


@case("账本服务·首次回放只对齐游标, 不把下游历史当外部成交追认")
def _():
    from app.repo import downstream_repo
    from app.services import ledger_service as ls
    fake = install_fakes()
    fake.params.pop("PMS_REPLAY_CURSOR", None)          # 冷启动: 从没设过游标
    hist = [{"order_id": f"BUY_X_{i}", "ts_code": "600000.SH", "side": "buy", "qty": 100,
             "price": 10.0, "done_time": "2020-01-01 09:40:00"} for i in range(500)]
    orig = downstream_repo.fetch_filled_orders
    try:
        downstream_repo.fetch_filled_orders = lambda **kw: hist
        downstream_repo.latest_filled_order_id = lambda: "SELL_ZZZ_999"
        r = ls.replay_fills()
        # 关键: 一条都不能入账。旧系统多年的历史成交若被并入 BASE, 摊薄成本与安全垫全错,
        # 而补仓/加仓/保垫减仓都挂在安全垫上 —— 一错就是整条纪律链。
        assert r.get("seeded") and r["fills"] == 0 and r["actions"] == 0, r
        assert r["cursor"] == "SELL_ZZZ_999", r
        assert not fake.lots and not fake.positions, "冷启动不该产生任何批次或持仓"
        assert fake.params["PMS_REPLAY_CURSOR"] == "SELL_ZZZ_999"

        # 游标就位后, 新成交照常入账
        new = [{"order_id": "ZZZ_NEW", "ts_code": "600000.SH", "side": "buy", "qty": 100,
                "price": 10.0, "done_time": "2026-07-29 09:40:00"}]
        downstream_repo.fetch_filled_orders = lambda **kw: new
        r2 = ls.replay_fills()
        assert r2["ok"] and r2["actions"] == 1, r2

        # 显式全量: 游标设 ALL 才追认历史
        fake.params["PMS_REPLAY_CURSOR"] = ls.CURSOR_ALL
        downstream_repo.fetch_filled_orders = lambda **kw: hist[:3]
        r3 = ls.replay_fills()
        assert r3["fills"] == 3, r3
    finally:
        downstream_repo.fetch_filled_orders = orig


@case("自主提议·一次性守卫真的写得进去, 且当日被拒的不再每分钟重评")
def _():
    from datetime import date
    from app.repo import pms_repo
    from app.services import proposal_service as ps
    fake = install_fakes(positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                     "avail_qty": 6000, "base_qty": 6000, "avg_cost": 10.0}])
    # 三个计数器此前只被读、从没被写过 —— 设计 §6 的三条一次性约束等于一直没生效
    ps.bump_once_guards("600000.SH", "FILL", {})
    assert fake.positions["600000.SH"]["fill_count"] == 1
    ps.bump_once_guards("600000.SH", "ADD", {})
    assert fake.positions["600000.SH"]["last_add_date"] == date.today()
    ps.bump_once_guards("600000.SH", "DCA", {"stage": 2})
    assert fake.positions["600000.SH"]["dca_count"] == 2, "dca_count 记的是档序不是次数"
    ps.bump_once_guards("600000.SH", "TRIM", {})          # 减持无一次性约束, 不该动任何列
    assert fake.positions["600000.SH"]["fill_count"] == 1

    # 当日已被规则闸拒过的 (代码, 动作) 要进 skip —— 否则超上限时每分钟重评一次、
    # 每分钟往评审账本灌一行, 判分锚被自己的噪声埋掉
    pms_repo.insert_ledger(ts_code="600000.SH", action="DCA", arbiter="rule",
                           verdict="REJECT", price_at=10.0, reason="超总仓上限")
    pms_repo.insert_ledger(ts_code="000001.SZ", action="ADD", arbiter="judge",
                           verdict="REJECT", price_at=10.0, reason="研判驳回")
    keys = ps._rejected_today_keys()
    assert ("600000.SH", "DCA") in keys, keys
    assert ("000001.SZ", "ADD") not in keys, "研判驳回不该进当日去重 —— 研判结论会变"


@case("账本服务·对账补仓位取下游成本价, 不拿现价充数")
def _():
    from app.core import recon as rc
    from app.repo import downstream_repo
    from app.services import ledger_service as ls, market
    fake = install_fakes(prices={"600000.SH": 10.0, "000001.SZ": 30.0})
    # 下游有两只票: A 真实成本 20 (现价 10, 实亏 50%), B 下游没给成本价
    downstream_repo.fetch_positions = lambda: {
        "columns": {"qty": "total_quantity", "cost": "cost_price"}, "raw_count": 2,
        "rows": [{"ts_code": "600000.SH", "qty": 1000, "avail_qty": 1000, "cost": 20.0,
                  "price": 10.0},
                 {"ts_code": "000001.SZ", "qty": 500, "avail_qty": 500, "cost": None,
                  "price": 30.0}]}
    r = ls.reconcile(apply_fix=True)
    assert r["ok"], r
    by = {f["ts_code"]: f for f in r["fixes"]}
    # A: 必须记 20 而不是 10 —— 记成 10 的话安全垫是 0, 这只实亏 50% 的票就永远不进补仓评估
    assert by["600000.SH"]["price"] == 20.0, by["600000.SH"]
    assert by["600000.SH"]["price_source"] == "下游成本价"
    # B: 下游没成本价才退到现价, 且留痕注明是估的
    assert by["000001.SZ"]["price"] == 30.0 and "兜底" in by["000001.SZ"]["price_source"]
    # 落到账本上: 摊薄成本 = 真实成本, 安全垫 = (10-20)/20 = -50%
    pos = fake.positions["600000.SH"]
    assert abs(float(pos["avg_cost"]) - 20.0) < 1e-6, pos
    assert abs(float(pos["cushion_pct"]) - (-0.5)) < 1e-4, pos

    # 纯逻辑侧: cost_map 缺项时逐只独立回退, 不会一只没成本就全退现价
    fixes = rc.build_recon_fixes(
        [{"ts_code": "A", "delta": 100}, {"ts_code": "B", "delta": 200}],
        price_map={"A": 1.0, "B": 2.0}, cost_map={"A": 9.0})
    assert [f["price"] for f in fixes] == [9.0, 2.0], fixes


@case("账本服务·对账以下游为准 + 连续不一致升级")
def _():
    from app.services import ledger_service as ls, param_store
    from app.repo import downstream_repo
    fake = install_fakes(prices={"600000.SH": 10.0})
    fake.insert_lot(ts_code="600000.SH", lot_type="BASE", qty=1000, open_price=10.0,
                    open_date="2026-07-01")
    fake.update_position("600000.SH", total_qty=1000, avail_qty=1000)
    orig = downstream_repo.fetch_positions
    try:
        downstream_repo.fetch_positions = lambda: {
            "rows": [{"ts_code": "600000.SH", "qty": 1500, "avail_qty": 1500}],
            "columns": {"code": "stock_code", "qty": "current_qty"}, "raw_count": 1}
        r = ls.reconcile()
        assert r["ok"] and len(r["diffs"]) == 1 and r["diffs"][0]["delta"] == 500, r
        assert fake.positions["600000.SH"]["total_qty"] == 1500          # 以下游为准
        assert any(l["lot_type"] == "RECON" for l in fake.lots)          # 修正留痕
        assert any(x.get("action") == "RECON" for x in fake.ledger)
        assert r["severity"] == "WARN"
        assert r["streak"] == 1, r

        # 每跑一趟对账下游都比账本多 100 股 —— 这样每一趟都真的"不一致", 隔离出
        # 「同一天多趟到底加不加」这一个变量
        nxt = [1500]

        def _drift():
            nxt[0] += 100
            return {"rows": [{"ts_code": "600000.SH", "qty": nxt[0]}],
                    "columns": {"qty": "current_qty"}, "raw_count": 1}
        downstream_repo.fetch_positions = _drift

        # 同一交易日内再对账多少趟, 连续天数都**不动** —— 盘中轻对账每分钟跑一次,
        # 按次累加的话"连续 3 日"三分钟就到了, 日报还会写出"连续 175 日"这种数。
        for _ in range(3):
            r = ls.reconcile()
        assert r["diffs"] and r["streak"] == 1 and r["severity"] == "WARN", r

        # 跨交易日才推进: 把"上次推进日"往前拨, 等价于隔了一天再跑。
        # 必须走 set_param (不能直接改 fake.params) —— 一是要过 ParamStore 的缓存失效,
        # 二是**顺带守住白名单**: STREAK_YMD 键当初就是漏在白名单外, set_param 静默拒写、
        # prev_ymd 永远读回 0, 按日推进形同虚设。这里 ok 断言就是那道锁。
        for _i in range(2):
            w = param_store.set_param(ls.STREAK_YMD_KEY, 20260701 + _i, "test")
            assert w["ok"], w
            r = ls.reconcile()
        assert r["streak"] == 3, r                                      # 连续第 3 日
        assert r["severity"] == "ERROR", r
    finally:
        downstream_repo.fetch_positions = orig


@case("账本服务·下游读空绝不清账 (2026-07-29 实际发生过的 22 只全核销)")
def _():
    from app.services import ledger_service as ls
    from app.repo import downstream_repo
    fake = install_fakes(prices={f"6000{i:02d}.SH": 10.0 for i in range(22)})
    for i in range(22):
        code = f"6000{i:02d}.SH"
        fake.insert_lot(ts_code=code, lot_type="BASE", qty=1000, open_price=10.0,
                        open_date="2026-07-01")
        fake.update_position(code, total_qty=1000, avail_qty=1000)
    orig = downstream_repo.fetch_positions
    try:
        # 对端模拟环境重启, 下游持仓表读回空集。「一夜清仓」与「读空」在数据上完全同形,
        # 但代价不对称: 拦错了只是晚一天修账, 放错了是 22 个持仓的批次结构不可逆地没了。
        downstream_repo.fetch_positions = lambda: {
            "rows": [], "columns": {"code": "stock_code", "qty": "current_qty"},
            "raw_count": 0}
        r = ls.reconcile()
        assert r.get("blocked"), "下游读空必须拦截, 不能照着清账"
        assert r["severity"] == "ERROR", r["severity"]
        assert not r["fixes"], "拦截时不得产生任何修正"
        for i in range(22):
            assert fake.positions[f"6000{i:02d}.SH"]["total_qty"] == 1000, "持仓不得被动过"
        r2 = ls.reconcile(force=True)          # 人工确认下游读数无误后放行
        assert not r2.get("blocked"), "force 应绕过限制"
    finally:
        downstream_repo.fetch_positions = orig


@case("账本服务·日常小幅漂移仍照常自动修正 (不被爆炸半径误伤)")
def _():
    from app.services import ledger_service as ls
    from app.repo import downstream_repo
    fake = install_fakes(prices={f"6000{i:02d}.SH": 10.0 for i in range(22)})
    for i in range(22):
        code = f"6000{i:02d}.SH"
        fake.insert_lot(ts_code=code, lot_type="BASE", qty=1000, open_price=10.0,
                        open_date="2026-07-01")
        fake.update_position(code, total_qty=1000, avail_qty=1000)
    orig = downstream_repo.fetch_positions
    try:
        rows = [{"ts_code": f"6000{i:02d}.SH", "qty": 1000} for i in range(22)]
        rows[0]["qty"] = 1500                       # 22 只里只有 1 只对不上
        downstream_repo.fetch_positions = lambda: {
            "rows": rows, "columns": {"qty": "current_qty"}, "raw_count": 22}
        r = ls.reconcile()
        assert not r.get("blocked"), "小幅漂移不该被拦 —— 那正是对账的价值"
        assert fake.positions["600000.SH"]["total_qty"] == 1500, "以下游为准"
    finally:
        downstream_repo.fetch_positions = orig


@case("账本服务·日报生成 (关注区 + 次日除权检测快照)")
def _():
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                     "avg_cost": 9.0}])
    rep = ls.build_daily_report()
    assert rep["totals"]["portfolio_mv"] == 60_000.0
    assert rep["snapshot"]["600000.SH"] == {"qty": 6000, "price": 10.0}
    assert any(a["type"] == "行业约束停用" for a in rep["attention"]), rep["attention"]
    assert fake.reports[rep["ymd"]]["ymd"] == rep["ymd"]


@case("账本服务·除权检测走通 (10送10 → 批次按比例调整)")
def _():
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 5.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 2000,
                                     "avg_cost": 10.0}])
    fake.insert_lot(ts_code="600000.SH", lot_type="BASE", qty=2000, open_price=10.0,
                    open_date="2026-07-01")
    fake.upsert_report(20260726, {"snapshot": {"600000.SH": {"qty": 1000, "price": 10.0}}})
    r = ls.detect_and_apply_ex_right()
    assert r["ex_rights"] and abs(r["ex_rights"][0]["ratio"] - 2.0) < 1e-6, r
    lot = fake.lots[0]
    assert lot["qty"] == 4000 and abs(lot["open_price"] - 5.0) < 1e-6, lot


@case("盘前准备·T+1 可卖重置")
def _():
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 10.0},
                         positions=[{"ts_code": "600000.SH", "total_qty": 6000, "avail_qty": 0,
                                     "avg_cost": 9.0}])
    r = ls.premarket()
    assert r["avail_reset"] >= 1, r
    assert fake.positions["600000.SH"]["avail_qty"] == 6000


@case("执行器·方案转指令 (GATED 批不转, HALT 直接完结)")
def _():
    from app.services import command_service as csvc, executor
    fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 14000,
                                     "base_qty": 7000, "fill_qty": 3500, "add_qty": 3500,
                                     "avail_qty": 14000, "avg_cost": 9.0}])
    r = csvc.issue("REDUCE_EXPOSURE", {"pct": "3%"})
    assert r["ok"], r
    # 掺一条 GATED 方案 (建仓补足批) 与一条 HALT 方案
    fake.insert_plans([
        {"plan_id": "P_G", "command_id": r["command_id"], "ts_code": "600000.SH",
         "action": "FILL", "qty": 3000, "amount": 30000, "priority": 20,
         "deadline": "2026-07-30", "status": "GATED", "reason": "回踩补足批"},
        {"plan_id": "P_H", "command_id": r["command_id"], "ts_code": "600000.SH",
         "action": "HALT", "qty": 0, "amount": 0, "priority": 10,
         "deadline": "2026-07-30", "status": "PENDING", "reason": "撤在途买入"}])
    m = executor.materialize_plans()
    assert m["created"], m
    codes = {fake.instructions[i]["ts_code"] for i in m["created"]}
    assert codes == {"600000.SH"}, codes
    assert all(fake.instructions[i]["status"] == "PROPOSED" for i in m["created"])
    gated = [p for p in fake.plans if p["plan_id"] == "P_G"][0]
    assert gated["status"] == "GATED", gated              # 未解锁的批次不转指令
    halt = [p for p in fake.plans if p["plan_id"] == "P_H"][0]
    assert halt["status"] == "DONE", halt
    assert any("HALT" in s["why"] for s in m["skipped"]), m["skipped"]
    # 幂等: 再跑一次不重复建指令
    n1 = len(fake.instructions)
    executor.materialize_plans()
    assert len(fake.instructions) == n1


@case("执行器·出手一跳 (影子模式): 择时→规则闸→下发→子单→评审留痕")
def _():
    from datetime import datetime as _dt
    from app.services import executor
    fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                     "avail_qty": 6000, "avg_cost": 9.0}])
    fake.insert_instruction(instruction_id="INS_S", origin_type="plan", origin_id="P1",
                            ts_code="600000.SH", action="TRIM", side="sell", qty=3000,
                            window_tdays=3, status="PROPOSED",
                            progress={"deadline": "2026-07-29", "is_command": True,
                                      "children": []})
    # 周一 10:05, 距截止 3 个交易日 → 当日配额 1000
    r = executor.run_tick(now=_dt(2026, 7, 27, 10, 5))
    assert r["ok"] and len(r["fired"]) == 1, r
    fired = r["fired"][0]
    assert fired["qty"] == 1000 and fired["mode"] == "shadow", fired
    ins = fake.instructions["INS_S"]
    assert ins["status"] == "DISPATCHED"
    assert len(ins["progress"]["children"]) == 1
    assert ins["progress"]["children"][0]["qty"] == 1000
    assert ins["progress"]["dispatched_at"]
    assert any(x["verdict"] == "PASS" and x["arbiter"] == "rule" for x in fake.ledger)
    # 同一天再跳一次: 配额已出完, 不重复下发
    r2 = executor.run_tick(now=_dt(2026, 7, 27, 10, 6))
    assert not r2["fired"] and r2["waited"], r2
    assert len(fake.instructions["INS_S"]["progress"]["children"]) == 1


@case("执行器·规则闸拦截时不下发且落拒绝留痕")
def _():
    from datetime import datetime as _dt
    from app.services import executor
    fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                     "avail_qty": 6000, "avg_cost": 9.0}])
    fake.insert_instruction(instruction_id="INS_B", origin_type="plan", origin_id="P2",
                            ts_code="600000.SH", action="OPEN", side="buy", qty=1000,
                            window_tdays=1, status="PROPOSED",
                            progress={"deadline": "2026-07-27", "is_command": True,
                                      "children": []})
    from app.services import param_store
    param_store.set_param("PMS_GLOBAL_BUY_HALT", True, "test")
    r = executor.run_tick(now=_dt(2026, 7, 27, 10, 5))
    assert not r["fired"] and r["rejected"], r
    assert any("BUY_HALT" in f for f in r["rejected"][0]["failed"]), r["rejected"]
    assert fake.instructions["INS_B"]["status"] == "PROPOSED"       # 未下发
    assert any(x["verdict"] == "REJECT" for x in fake.ledger)


@case("执行器·试算模式只算不发不落库")
def _():
    from datetime import datetime as _dt
    from app.services import executor
    fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                     "avail_qty": 6000, "avg_cost": 9.0}])
    fake.insert_instruction(instruction_id="INS_D", origin_type="plan", origin_id="P3",
                            ts_code="600000.SH", action="EXIT", side="sell", qty=2000,
                            window_tdays=1, status="PROPOSED",
                            progress={"deadline": "2026-07-27", "children": []})
    r = executor.run_tick(now=_dt(2026, 7, 27, 10, 5), dry_run=True)
    assert r["fired"] and r["fired"][0]["dry_run"] is True, r
    assert r["fired"][0]["qty"] == 2000                      # 末日全出
    assert fake.instructions["INS_D"]["status"] == "PROPOSED"
    assert not fake.instructions["INS_D"]["progress"]["children"]
    assert not fake.ledger


@case("执行器·窗口收口: 成交回写方案 + 足额置确认 + 命令类部分完成")
def _():
    from datetime import datetime as _dt
    from app.services import executor
    fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                     "avail_qty": 6000, "avg_cost": 9.0}])
    fake.insert_plans([{"plan_id": "P_F", "command_id": "CMD_X", "ts_code": "600000.SH",
                        "action": "TRIM", "qty": 2000, "amount": 20000, "priority": 30,
                        "deadline": "2026-07-27", "status": "EXECUTING", "reason": "收利润"}])
    fake.insert_instruction(instruction_id="INS_F", origin_type="plan", origin_id="P_F",
                            ts_code="600000.SH", action="TRIM", side="sell", qty=2000,
                            status="DISPATCHED",
                            progress={"deadline": "2026-07-27", "is_command": True,
                                      "children": []})
    # exec_qty 不是 insert_instruction 的形参 (真 repo 没有), 造数据直接改桩里的字段
    fake.instructions["INS_F"]["exec_qty"] = 2000
    fake.insert_instruction(instruction_id="INS_P", origin_type="plan", origin_id="P_F",
                            ts_code="600000.SH", action="TRIM", side="sell", qty=2000,
                            status="DISPATCHED",
                            progress={"deadline": "2026-07-20", "is_command": True,
                                      "children": []})
    fake.instructions["INS_P"]["exec_qty"] = 500
    fake.insert_instruction(instruction_id="INS_A2", origin_type="proposal", origin_id="PR1",
                            ts_code="600000.SH", action="ADD", side="buy", qty=1000,
                            status="DISPATCHED",
                            progress={"deadline": "2026-07-20", "is_command": False,
                                      "children": []})
    r = executor.sweep_windows(now=_dt(2026, 7, 27, 15, 10))
    assert fake.instructions["INS_F"]["status"] == "CONFIRMED", fake.instructions["INS_F"]
    assert [p for p in fake.plans if p["plan_id"] == "P_F"][0]["filled_qty"] == 500
    assert any(x["instruction_id"] == "INS_P" for x in r["partial"]), r
    assert fake.instructions["INS_P"]["progress"]["window_verdict"]["verdict"] == "PARTIAL"
    assert "INS_A2" in r["expired"], r                    # 自主类窗口耗尽即作废
    assert fake.instructions["INS_A2"]["status"] == "EXPIRED"


@case("组合快照·真钱取 ws 资金快照(含当日回笼); 取不到则标明是估的")
def _():
    from app.services import portfolio
    fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"})
    t = portfolio.positions_view()["totals"]
    assert t["cash_source"] == "estimate" and t["cash_avail"] is None, t
    assert t["cash_est"] == 2_000_000.0, t          # scale − 市值, 是个估算值

    fake.qmt.put_snapshot(kind="funds", data={"total_asset": 981448.56,
                                              "available_cash": 971251.56,
                                              "sell_return_today": 48900.0})
    v = portfolio.positions_view()
    t = v["totals"]
    assert t["cash_source"] == "ws", t
    # 当日卖出回笼 T+0 可用, 必须算进去 —— 不算就把"卖一只买另一只"判成资金不足
    assert t["cash_avail"] == round(971251.56 + 48900.0, 2), t
    assert t["total_asset"] == 981448.56, t
    # 这一行是整件事的由来: 按 scale 估出来 200 万, 账户真钱只有 102 万, 差着一倍
    assert t["cash_est"] > t["cash_avail"], (t["cash_est"], t["cash_avail"])
    # caps_ctx 要把真钱与来源一起带给规则闸, 否则闸门只能降级
    caps = portfolio.caps_ctx(v, ts_code="600000.SH")
    assert caps["cash_source"] == "ws" and caps["cash_avail"] == t["cash_avail"], caps


@case("对账事实源·ws 快照新鲜 → 以 ws 为准建账 (表空集不再当事实)")
def _():
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 10.0})
    fake.qmt.put_snapshot(items=[{"ts_code": "600000.SH", "total_qty": 1100,
                                  "avail_qty": 1100, "cost_price": 9.273}])
    src = ls.positions_source()
    assert src["source"] == ls.SRC_WS, src
    assert src["rows"][0]["qty"] == 1100 and src["rows"][0]["cost"] == 9.273, src["rows"]
    assert not src["alerts"], src["alerts"]      # 表是空集, 不算"不一致"
    # 账本空 + ws 有持仓 = 首次建账, 爆炸半径闸放行, 成本价取下游 cost_price
    r = ls.reconcile(apply_fix=True)
    assert r["source"] == "ws" and not r.get("blocked"), r
    assert fake.positions["600000.SH"]["total_qty"] == 1100, fake.positions
    assert abs(float(fake.positions["600000.SH"]["avg_cost"]) - 9.273) < 1e-6, \
        "开仓价必须取下游 cost_price —— 拿现价当成本会让安全垫齐刷刷是 0"


@case("对账事实源·前缀式代码归一成点式 (不归一会造出一轮双向全量重写)")
def _():
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 10.0})
    fake.qmt.put_snapshot(items=[{"stock_code": "SH600000", "total_quantity": 600,
                                  "cost_price": 10.0}])
    src = ls.positions_source()
    assert src["rows"][0]["ts_code"] == "600000.SH", src["rows"]


@case("对账事实源·ws 与表对不上 → 用 ws 但必须留痕告警")
def _():
    from app.repo import downstream_repo
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 10.0})
    fake.qmt.put_snapshot(items=[{"ts_code": "600000.SH", "total_qty": 1100}])
    downstream_repo.fetch_positions = lambda: {
        "rows": [{"ts_code": "600000.SH", "qty": 700, "avail_qty": 700, "cost": 9.0,
                  "frozen": 0, "price": 10.0}],
        "columns": {"qty": "total_quantity"}, "raw_count": 1}
    src = ls.positions_source()
    assert src["source"] == ls.SRC_WS, src["source"]
    assert src["rows"][0]["qty"] == 1100, "以 ws 为准"
    codes = [a["code"] for a in src["alerts"]]
    assert "SOURCE_DISAGREE" in codes, src["alerts"]


@case("对账事实源·ws 快照过期 → 退回表并说明退回原因")
def _():
    from app.repo import downstream_repo
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 10.0},
                         params={"PMS_RECON_WS_SNAPSHOT_MAX_AGE_SEC": "900"})
    fake.qmt.put_snapshot(items=[{"ts_code": "600000.SH", "total_qty": 1100}], age_sec=5000)
    downstream_repo.fetch_positions = lambda: {
        "rows": [{"ts_code": "600000.SH", "qty": 700, "avail_qty": 700, "cost": 9.0,
                  "frozen": 0, "price": 10.0}],
        "columns": {"qty": "total_quantity"}, "raw_count": 1}
    src = ls.positions_source()
    assert src["source"] == ls.SRC_TABLE, src["source"]
    assert src["rows"][0]["qty"] == 700
    msg = " ".join(a["message"] for a in src["alerts"])
    assert "过期" in msg, src["alerts"]


@case("对账事实源·表应答空集仍算有效应答 → 归 table (空集是数据, 不是缺数据)")
def _():
    from app.services import ledger_service as ls
    install_fakes(prices={"600000.SH": 10.0})    # 默认桩: fetch_positions 成功返回 rows=[]
    src = ls.positions_source()
    # 这一条守着 force 的生路: 归 table 才会走到爆炸半径闸, 人工确认后 force 能放行;
    # 若把"应答了空"误判成 source=none, 对账入口就把 force 挡死了 (曾经真挡死过)
    assert src["source"] == ls.SRC_TABLE, src
    assert src["rows"] == [], src


@case("对账事实源·两个源都**无应答** + 本端有持仓 → 入口就拒, **force 也不放行**")
def _():
    from app.repo import downstream_repo
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 10.0})
    fake.insert_lot(ts_code="600000.SH", lot_type="BASE", qty=1000, open_price=10.0,
                    open_date="2026-07-01")
    fake.update_position("600000.SH", total_qty=1000, avail_qty=1000)

    def boom():
        raise RuntimeError("Can't connect to MySQL server on '192.168.16.153'")
    downstream_repo.fetch_positions = boom       # 没塞快照 + 表查询异常 = 谁都没应答
    src = ls.positions_source()
    assert src["source"] == ls.SRC_NONE, src
    # force 的语义是「人工已确认下游读数正确」, 而这里根本没有读数可供确认 —— 没有任何数字,
    # 人也无从确认。真要清账走 reset_ledger.py, 别拿空壳子当"下游事实"去核销批次。
    for force in (False, True):
        r = ls.reconcile(apply_fix=True, force=force)
        assert r.get("blocked"), f"force={force} 也必须拒绝对账"
        assert r["severity"] == "ERROR" and not r["fixes"], r
    assert fake.positions["600000.SH"]["total_qty"] == 1000, "持仓一股都不许被动"


@case("对账事实源·两个源都无应答 + 账本也空 → 不报 ERROR (刚清账等对端装持仓)")
def _():
    from app.repo import downstream_repo
    from app.services import ledger_service as ls
    install_fakes()

    def boom():
        raise RuntimeError("库不可达")
    downstream_repo.fetch_positions = boom
    r = ls.reconcile(apply_fix=True)
    # 盘中轻对账每分钟一跳, 这个状态下反复刷 ERROR 只会把真告警埋掉
    assert not r.get("blocked") and r["severity"] == "OK", r
    assert "无可对之账" in (r.get("note") or ""), r


@case("ws 入账·三分流: 真单入账 / 联调单跳过 / **孤儿成交挂起不入账**")
def _():
    from app.repo import qmt_repo
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 10.0})
    fake.insert_instruction(instruction_id="INS_R", origin_type="plan", origin_id="P1",
                            ts_code="600000.SH", action="OPEN", side="buy", qty=100,
                            status="DISPATCHED")
    # 真单: 出口表有行、父指令不带 SMOKE_
    fake.qmt.enqueue_order(instruction_id="INS_R-1", parent_id="INS_R", ts_code="600000.SH",
                           side="buy", qty=100, limit_price=10.0, valid_until=0)
    fake.qmt.put_trade(seq=1, instruction_id="INS_R-1", ts_code="600000.SH", side="buy",
                       qty=100, price=10.0, fee=5.0)
    # 联调单: 出口表有行, 父指令带 SMOKE_
    fake.qmt.enqueue_order(instruction_id="INS_S-1", parent_id="SMOKE_INS_S",
                           ts_code="600000.SH", side="buy", qty=100, limit_price=9.0,
                           valid_until=0)
    fake.qmt.put_trade(seq=2, instruction_id="INS_S-1", ts_code="600000.SH", side="buy",
                       qty=100, price=9.0)
    # 孤儿: 出口表里根本没有这个 instruction_id (上下游停机后遗留的旧 inbox 行)
    fake.qmt.put_trade(seq=3, instruction_id="INS_GHOST-1", ts_code="600183.SH", side="buy",
                       qty=100, price=88.0)
    r = ls.consume_ws_trades()
    assert r["trades"] == 1, f"只有真单该进入账流程: {r}"
    assert r.get("smoke_skipped") == 1 and r.get("orphan_held") == 1, r
    assert fake.qmt.inbox[1]["processed"] == qmt_repo.PROC_BOOKED, fake.qmt.inbox[1]
    assert fake.qmt.inbox[2]["processed"] == qmt_repo.PROC_DIGESTED, fake.qmt.inbox[2]
    # 关键: 孤儿标 3 (挂起) 而不是 2 (已消化) —— 2 的语义是"确认过不用管", 混了就分不出漏账
    assert fake.qmt.inbox[3]["processed"] == qmt_repo.PROC_ORPHAN, fake.qmt.inbox[3]
    # 账本只认那一笔真单; 孤儿的标的一股都不许长出来
    assert [l["ts_code"] for l in fake.lots] == ["600000.SH"], fake.lots
    assert fake.positions["600000.SH"]["total_qty"] == 100
    assert "600183.SH" not in fake.positions or \
           fake.positions["600183.SH"]["total_qty"] == 0, fake.positions
    # 告警必须活着传出来 —— mapped["alerts"] 曾经直接赋值把它冲掉过
    assert any(a.get("type") == "ORPHAN_WS_TRADE" for a in r["alerts"]), r["alerts"]
    assert qmt_repo.inbox_orphan_count() == 1
    # 费用只出现金流水、不进成本 (§5.5): 真单那 5 元记一条 FEE, 摊薄成本仍是 10.0
    assert r["fees"] == 1 and len(fake.cash_flows) == 1, fake.cash_flows
    assert fake.cash_flows[0]["kind"] == "FEE" and fake.cash_flows[0]["amount"] == -5.0
    assert abs(float(fake.positions["600000.SH"]["avg_cost"]) - 10.0) < 1e-6, fake.positions


@case("ws 入账·出口表反查报错时一笔都不判 (留在待入账, 下一跳重来)")
def _():
    from app.repo import qmt_repo
    from app.services import ledger_service as ls
    fake = install_fakes(prices={"600000.SH": 10.0})
    fake.qmt.put_trade(seq=9, instruction_id="INS_X-1", ts_code="600000.SH", side="buy",
                       qty=100, price=10.0)

    def boom(_iid):
        raise RuntimeError("Lost connection to MySQL server during query")
    qmt_repo.get_order = boom
    r = ls.consume_ws_trades()
    # 原先这里把异常当"查不到"→ 当真单入账: 一次连接超时就能凭空造出一笔外部成交。
    assert not r["ok"] and r["errors"], r
    assert r["trades"] == 0 and not r.get("orphan_held"), r
    assert fake.qmt.inbox[9]["processed"] == qmt_repo.PROC_PENDING, fake.qmt.inbox[9]
    assert not fake.lots, fake.lots


@case("ws 入账·孤儿成交在通道状态里露出数字 (挂起的账不能没人报)")
def _():
    from app.services import dispatcher
    fake = install_fakes()
    assert dispatcher.channel_status()["orphan_held"] == 0
    fake.qmt.put_trade(seq=5, instruction_id="INS_GHOST-2", ts_code="600000.SH", side="sell",
                       qty=100, price=10.0, processed=3)
    assert dispatcher.channel_status()["orphan_held"] == 1


@case("下发通道·影子回执 / ws 出口队列 / 进程与连接两级降级 / 撤销走本地置状态")
def _():
    from datetime import datetime as _dtm
    from app.services import dispatcher, executor, param_store
    fake = install_fakes()
    assert dispatcher.mode() == "shadow"
    assert set(dispatcher.describe()["modes"]) == {"shadow", "ws"}, dispatcher.describe()
    d = dispatcher.dispatch(instruction_id="INS_1", ts_code="600000.SH", side="sell",
                            qty=1000, limit_price=9.98)
    assert d["ok"] and d["ref"] == "manual:INS_1" and "人工" in d["note"], d
    # 已作废的通道名不能再被设进来
    for dead in ("bad_mode", "plan_x", "channel_y"):
        assert param_store.set_param("PMS_DISPATCH_MODE", dead)["ok"] is False, dead
    assert param_store.set_param("PMS_DISPATCH_MODE", "ws")["ok"] is True
    vu = _dtm.now().replace(hour=14, minute=45, second=0, microsecond=0)

    def _send(iid, side="sell", qty=1000, px=9.98):
        return dispatcher.dispatch(instruction_id=iid, parent_id=iid.rsplit("_D", 1)[0],
                                   ts_code="600000.SH", side=side, qty=qty,
                                   limit_price=px, valid_until=vu, intent="TRIM")

    # ① ws 进程没在跑 (心跳陈旧): 买卖一律拒发 —— 排进队列也没人发, 装作成功更危险
    fake.qmt.set_channel("OFFLINE", alive=False)
    for side in ("sell", "buy"):
        r = _send(f"INS_2{side}_D01", side=side, qty=1000)
        assert r["ok"] is False and r["mode"] == "ws" and "未在线" in r["error"], r
    assert not fake.qmt.orders, "进程不在时不该往出口表里塞东西"

    # ② 进程在、连接断: 协议 §6.3 —— 停发增持, 减持照常入队等重连
    fake.qmt.set_channel("OFFLINE", alive=True)
    rb = _send("INS_3buy_D01", side="buy", qty=1000)
    assert rb["ok"] is False and "§6.3" in rb["error"], rb
    rs = _send("INS_3sell_D01", side="sell", qty=1000)
    assert rs["ok"] and "重连后" in rs["note"], rs
    assert fake.qmt.orders["INS_3sell_D01"]["status"] == "QUEUED"
    assert fake.qmt.orders["INS_3sell_D01"]["parent_id"] == "INS_3sell"

    # ③ 连接正常: 买入也放行, 落出口表即返回 (不等 QMT 的 ack)
    fake.qmt.set_channel("ONLINE", alive=True)
    rb2 = _send("INS_4buy_D01", side="buy", qty=1000, px=9.876)
    assert rb2["ok"] and rb2["ref"] == "INS_4buy_D01", rb2
    o = fake.qmt.orders["INS_4buy_D01"]
    assert o["limit_price"] == 9.88 and o["valid_until"] > 10 ** 12, o   # 2位小数 + 毫秒

    # ④ 不合协议的参数在本地就拦下, 不换对端一个 BAD_PARAM
    bad = _send("INS_5buy_D01", side="buy", qty=150)                     # 买入非整百
    assert bad["ok"] is False and "不合协议" in bad["error"], bad
    assert "INS_5buy_D01" not in fake.qmt.orders

    # ⑤ 撤单: 标记父指令名下所有在途子单, 由 ws 进程逐张发 cancel_order
    c = dispatcher.cancel(instruction_id="INS_3sell")
    assert c["ok"] and c["cancelled"] == 1 and c["cancel_id"].startswith("CXL-"), c
    assert fake.qmt.orders["INS_3sell_D01"]["cancel_state"] == "REQUESTED"
    assert dispatcher.cancel(instruction_id="INS_NOBODY")["cancelled"] == 0
    param_store.set_param("PMS_DISPATCH_MODE", "shadow")

    fake.insert_instruction(instruction_id="INS_C", origin_type="plan", origin_id="P1",
                            ts_code="600000.SH", action="TRIM", side="sell", qty=1000,
                            status="DISPATCHED", progress={"children": []})
    r = executor.cancel_instruction("INS_C")
    assert r["ok"] and fake.instructions["INS_C"]["status"] == "CANCELLED", r
    assert executor.cancel_instruction("INS_C")["ok"] is False


def _prop_fakes(**kw):
    """自主提议用的组合: A 票厚垫创新高(可加仓), B 票垫子回吐过半(可保垫减仓)。"""
    return install_fakes(
        prices={"600000.SH": 11.0, "000001.SZ": 10.4},
        high5={"600000.SH": 11.0, "000001.SZ": 12.0},     # B 没创新高, 只该出 TRIM
        params={"PMS_TOTAL_SCALE": "2000000", **(kw.get("params") or {})},
        positions=[{"ts_code": "600000.SH", "total_qty": 6000, "avail_qty": 6000,
                    "base_qty": 6000, "avg_cost": 10.0, "cushion_peak": 0.0,
                    "target_pct": 0.06},
                   {"ts_code": "000001.SZ", "total_qty": 6000, "avail_qty": 6000,
                    "base_qty": 6000, "avg_cost": 10.0, "cushion_peak": 0.08,
                    "target_pct": 0.06}])


@case("自主提议·propose_only: 减持自动执行 / 增持入队 / 研判未接通即降级留痕")
def _():
    from app.services import proposal_service as ps
    fake = _prop_fakes(params={"PMS_AUTONOMY": "propose_only"})
    r = ps.scan_and_route()
    assert r["ok"], r
    ex = {(x["ts_code"], x["action"]) for x in r["executed"]}
    qd = {(x["ts_code"], x["action"]) for x in r["queued"]}
    assert ("000001.SZ", "TRIM") in ex, r          # 减持方向不设确认门槛
    assert ("600000.SH", "ADD") in qd, r           # 增持入队
    assert r["degraded"] is True, r                # 研判未接通 → 降级
    assert any("研判不可用" in x["why"] for x in r["queued"]), r["queued"]
    # 减持落了指令, 增持落了提议
    trim_ins = [i for i in fake.instructions.values() if i["action"] == "TRIM"]
    assert trim_ins and trim_ins[0]["side"] == "sell" and trim_ins[0]["qty"] == 2000, trim_ins
    add_prop = [p for p in fake.proposals.values() if p["action"] == "ADD"]
    assert add_prop and add_prop[0]["qty"] == 2700, add_prop
    assert add_prop[0]["hard_numbers"]["price"] == 11.0
    # 再扫一轮不重复提 (在途去重)
    r2 = ps.scan_and_route()
    assert not r2["executed"] and not r2["queued"], r2
    assert any(s.get("why") == "已有在途提议/指令" for s in r2["skipped"]), r2["skipped"]


@case("自主提议·full + 研判通过: 增持直接落指令; 深档补仓仍强制确认")
def _():
    from app.services import judge, proposal_service as ps
    orig = judge.request
    try:
        judge.request = lambda c, context=None, **kw: {"verdict": judge.PASS,
                                                       "reason": "研判通过(桩)",
                                                       "degraded": False, "raw": None}
        fake = _prop_fakes(params={"PMS_AUTONOMY": "full"})
        r = ps.scan_and_route()
        assert r["degraded"] is False, r
        ex = {(x["ts_code"], x["action"]) for x in r["executed"]}
        assert ("600000.SH", "ADD") in ex and ("000001.SZ", "TRIM") in ex, r
        assert not r["queued"], r
        add_ins = [i for i in fake.instructions.values() if i["action"] == "ADD"]
        assert add_ins and add_ins[0]["side"] == "buy" and add_ins[0]["qty"] == 2700, add_ins
        assert add_ins[0]["progress"]["is_command"] is False

        # 深档补仓: 即使档位 full、研判通过, 也必须入队等用户点头
        fake2 = install_fakes(prices={"600519.SH": 8.4}, high5={"600519.SH": 9.9},
                              params={"PMS_TOTAL_SCALE": "2000000", "PMS_AUTONOMY": "full"},
                              positions=[{"ts_code": "600519.SH", "total_qty": 6000,
                                          "avail_qty": 6000, "base_qty": 6000,
                                          "avg_cost": 10.0, "cushion_peak": 0.0,
                                          "target_pct": 0.06}])
        r2 = ps.scan_and_route()
        qd = {(x["ts_code"], x["action"]) for x in r2["queued"]}
        assert ("600519.SH", "DCA") in qd, r2
        assert any("深档" in x["why"] for x in r2["queued"]), r2["queued"]
        prop = [p for p in fake2.proposals.values() if p["action"] == "DCA"][0]
        assert prop["qty"] == 3000 and prop["hard_numbers"]["stage"] == 2, prop
    finally:
        judge.request = orig


@case("自主提议·研判驳回与规则闸拦截各自留痕")
def _():
    from app.services import judge, proposal_service as ps
    orig = judge.request
    try:
        judge.request = lambda c, context=None, **kw: {"verdict": judge.REJECT,
                                                       "reason": "形态走坏, 不宜加仓",
                                                       "degraded": False, "raw": None}
        fake = _prop_fakes(params={"PMS_AUTONOMY": "full"})
        r = ps.scan_and_route()
        rej = {(x["ts_code"], x["action"], x["by"]) for x in r["rejected"]}
        assert ("600000.SH", "ADD", "judge") in rej, r
        assert any(x["arbiter"] == "judge" and x["verdict"] == "REJECT" for x in fake.ledger)
        assert not any(i["action"] == "ADD" for i in fake.instructions.values())
    finally:
        judge.request = orig

    # 规则闸拦截: 全局暂停买入
    fake2 = _prop_fakes(params={"PMS_AUTONOMY": "full", "PMS_GLOBAL_BUY_HALT": "true"})
    r2 = ps.scan_and_route()
    rej2 = {(x["ts_code"], x["action"], x["by"]) for x in r2["rejected"]}
    assert ("600000.SH", "ADD", "rule") in rej2, r2
    assert any("BUY_HALT" in f for x in r2["rejected"] for f in x["failed"]), r2
    assert any(x["arbiter"] == "rule" and x["verdict"] == "REJECT" for x in fake2.ledger)
    # 减持不受暂停买入影响, 照样执行
    assert ("000001.SZ", "TRIM") in {(x["ts_code"], x["action"]) for x in r2["executed"]}, r2


@case("自主提议·档位 off 与休假模式不扫描; 试算不落表")
def _():
    from app.services import proposal_service as ps
    _prop_fakes(params={"PMS_AUTONOMY": "off"})
    r = ps.scan_and_route()
    assert r["candidates"] == 0 and any("off" in s["why"] for s in r["skipped"]), r

    _prop_fakes(params={"PMS_AUTONOMY": "full", "PMS_GLOBAL_EXEC_HALT": "true"})
    r2 = ps.scan_and_route()
    assert any("休假" in s["why"] for s in r2["skipped"]), r2

    fake = _prop_fakes(params={"PMS_AUTONOMY": "propose_only"})
    r3 = ps.scan_and_route(dry_run=True)
    assert (r3["executed"] or r3["queued"]) and all(
        x.get("dry_run") for x in r3["executed"] + r3["queued"]), r3
    assert not fake.instructions and not fake.proposals and not fake.ledger


@case("研判闸·未配置即不可用, 动作不在研判范围则直接放行")
def _():
    from app.services import judge
    install_fakes()
    assert judge.available() is False
    st = judge.status()
    assert st["available"] is False and "未配置" in st["reason"], st
    r = judge.request({"action": "ADD", "ts_code": "600000.SH", "qty": 100})
    assert r["verdict"] == judge.UNAVAILABLE and r["degraded"] is True, r
    r2 = judge.request({"action": "TRIM", "ts_code": "600000.SH", "qty": 100})
    assert r2["verdict"] == judge.PASS and r2["degraded"] is False, r2   # TRIM 不在研判范围


class FakeRedis:
    """Redis Stream 的最小替身 (消费组 + xreadgroup + ack)。"""

    def __init__(self, msgs=None):
        self.msgs = dict(msgs or {})
        self.acked, self.groups = [], []

    def xgroup_create(self, key, group, id="$", mkstream=False):
        self.groups.append((key, group))

    def xreadgroup(self, group, consumer, streams, count=10, block=0):
        key = list(streams)[0]
        m = self.msgs.pop(key, [])
        return [(key, m)] if m else []

    def xack(self, key, group, msg_id):
        self.acked.append(msg_id)

    def xlen(self, key):
        return len(self.msgs.get(key, []))

    def xinfo_groups(self, key):
        return [{"name": g, "pending": 0} for k, g in self.groups if k == key]


def _install_signal_fakes(fake, sell_msgs=None, intraday_msgs=None):
    """把两条流的假客户端装上, 返回 {db: FakeRedis}。"""
    import json as _json
    from datetime import datetime as _dt
    from app.services import signal_service as ss
    from config.settings import settings as _st
    ymd = _dt.now().strftime("%Y-%m-%d")
    r2 = FakeRedis({f"intraday_signals:{ymd}": list(intraday_msgs or [])})
    r3 = FakeRedis({"bionic:signals:llm_sell_actions": list(sell_msgs or [])})
    by_db = {_st.SIGNAL_REDIS_DB_INTRADAY: r2, _st.SIGNAL_REDIS_DB_ACTIONS: r3}
    ss._client = lambda db: by_db[db]
    return by_db


def _sell_msg(mid, code, conf, reason="逻辑走坏"):
    import json as _json
    return (mid, {"data": _json.dumps({"ts_code": code, "action": "SELL",
                                       "confidence": conf, "llm_reason": reason},
                                      ensure_ascii=False)})


@case("信号消化·高置信风控卖出转清仓指令; 中等置信落提议; 当日去重")
def _():
    from app.services import signal_service as ss
    fake = install_fakes(prices={"600000.SH": 10.0, "000001.SZ": 8.0},
                         params={"PMS_TOTAL_SCALE": "2000000"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                     "avail_qty": 6000, "avg_cost": 10.0},
                                    {"ts_code": "000001.SZ", "total_qty": 3000,
                                     "avail_qty": 3000, "avg_cost": 8.0}])
    by_db = _install_signal_fakes(fake, sell_msgs=[
        _sell_msg("3-1", "600000.SH", 92),      # 高置信 → 清仓
        _sell_msg("3-2", "000001.SZ", 80),      # 中置信 → 提议
        _sell_msg("3-3", "600519.SH", 95),      # 没持仓 → 忽略
    ])
    r = ss.consume()
    assert r["ok"], r
    assert [x["ts_code"] for x in r["exits"]] == ["600000.SH"], r
    assert r["exits"][0]["qty"] == 6000
    assert [x["ts_code"] for x in r["proposals"]] == ["000001.SZ"], r
    assert r["proposals"][0]["qty"] == 1000                  # 3000 的 1/3
    assert r["ignored"] >= 1, r
    assert len(by_db[3].acked) == 3, by_db[3].acked          # 三条都 ACK

    ins = [i for i in fake.instructions.values() if i["action"] == "EXIT"]
    assert ins and ins[0]["side"] == "sell" and ins[0]["progress"]["from_signal"] is True
    assert any(x["arbiter"] == "rule" and x["action"] == "EXIT" for x in fake.ledger)
    prop = [p for p in fake.proposals.values() if p["action"] == "TRIM"]
    assert prop and prop[0]["hard_numbers"]["signal_source"] == "risk_sell", prop

    # 同一条信号再来一次: 当日去重 + 在途检查, 不重复下指令
    n_ins = len(fake.instructions)
    _install_signal_fakes(fake, sell_msgs=[_sell_msg("3-4", "600000.SH", 92)])
    r2 = ss.consume()
    assert not r2["exits"] and len(fake.instructions) == n_ins, r2


@case("信号消化·BUY 只留痕不买; 关闭开关即不消化; 试算不落表不ACK")
def _():
    from app.services import signal_service as ss
    fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                     "avail_qty": 6000, "avg_cost": 10.0}])
    _install_signal_fakes(fake, intraday_msgs=[
        ("1-1", {"ts_code": "600000.SH", "action": "BUY", "confidence": "0.95"})])
    r = ss.consume()
    assert r["recorded"] == 1 and not r["exits"], r
    assert not any(i["side"] == "buy" for i in fake.instructions.values())
    assert any(x["action"] == "SIGNAL" for x in fake.ledger), fake.ledger

    fake2 = install_fakes(prices={"600000.SH": 10.0},
                         params={"PMS_TOTAL_SCALE": "2000000", "PMS_SIGNAL_ENABLED": "false"},
                         positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                     "avail_qty": 6000, "avg_cost": 10.0}])
    by = _install_signal_fakes(fake2, sell_msgs=[_sell_msg("3-9", "600000.SH", 95)])
    r2 = ss.consume()
    assert "skipped" in r2 and not fake2.instructions, r2

    fake3 = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
                          positions=[{"ts_code": "600000.SH", "total_qty": 6000,
                                      "avail_qty": 6000, "avg_cost": 10.0}])
    by3 = _install_signal_fakes(fake3, sell_msgs=[_sell_msg("3-10", "600000.SH", 95)])
    r3 = ss.consume(dry_run=True)
    assert r3["exits"] and r3["exits"][0]["dry_run"] is True, r3
    assert not fake3.instructions and not fake3.ledger
    assert by3[3].acked == [], "试算不该 ACK"


@case("装配·执行相关路由与调度接线到位")
def _():
    from app.web.main import app
    from app import scheduler as sch
    paths = {r.path for r in app.routes}
    for p in ("/api/ops/materialize", "/api/ops/exec-tick", "/api/ops/sweep-windows",
              "/api/instructions/{instruction_id}/cancel", "/api/dispatch-mode",
              "/api/ops/scan-proposals", "/api/ops/digest-signals", "/api/signal-status"):
        assert p in paths, p
    import inspect
    src = inspect.getsource(sch.intraday_exec)
    assert "executor" in src and "run_tick" in src, "调度器未接执行器"
    assert "proposal_service" in src, "调度器未接自主提议扫描"
    assert "signal_service" in inspect.getsource(sch.signal_digest), "调度器未接信号消化"


# ---------------------------------------------------------------- runner
# ================================================================ 行业数据源实探
def _real_industry():
    """把被 install_fakes 打过桩的 industry 恢复成真实实现 (reload 就地替换属性)。"""
    import importlib
    from app.services import industry as ind
    importlib.reload(ind)
    ind.invalidate()
    return ind


def _real_downstream():
    """同上, 恢复 downstream_repo —— install_fakes 把 category_columns 也打了桩。"""
    import importlib
    from app.repo import downstream_repo as dr
    importlib.reload(dr)
    return dr


@case("行业源·gp_stock_category 查得到才算 ready (参数填了不等于数据源能用)")
def _():
    from app.repo import downstream_repo
    fake = install_fakes(params={"PMS_SECTOR_SOURCE": "gp_stock_category"})
    ind = _real_industry()
    orig = downstream_repo.probe_category
    try:
        downstream_repo.probe_category = lambda c: {
            "value": "电力设备", "code_col": "stock_code", "industry_col": "industry",
            "columns": ["stock_code", "industry"], "error": None}
        assert ind.ready() is True
        st = ind.status()
        assert st["ready"] is True and st["count"] == st["probe"]["tried"] > 0, st
        assert "命中" in st["hint"], st["hint"]
        assert ind.get("600000.SH") == "电力设备"
    finally:
        downstream_repo.probe_category = orig
    _ = fake


@case("行业源·一只都查不到 → ready=False, 约束按未配置停用 (绝不静默失效)")
def _():
    from app.repo import downstream_repo
    install_fakes(params={"PMS_SECTOR_SOURCE": "gp_stock_category"})
    ind = _real_industry()
    orig = downstream_repo.probe_category
    try:
        # ① 表里没有这些票 (无报错)
        downstream_repo.probe_category = lambda c: {
            "value": None, "code_col": None, "industry_col": None, "columns": None,
            "error": None}
        assert ind.ready() is False
        st = ind.status()
        assert st["ready"] is False and "一只都没查到" in st["hint"], st["hint"]
        assert "表里没有这些票" in st["hint"] and "导出下游表结构" in st["hint"], st["hint"]
        # ② 表根本查不了 (有报错) —— 错误必须冒到 hint 上, 不能吞
        ind.invalidate()
        downstream_repo.probe_category = lambda c: {
            "value": None, "code_col": None, "industry_col": None, "columns": None,
            "error": "stock_code: ProgrammingError: Table doesn't exist"}
        assert ind.ready() is False
        assert "Table doesn't exist" in ind.status()["hint"]
    finally:
        downstream_repo.probe_category = orig


@case("行业源·gp_hybk: 三级 884 / 每票只认 bk_code 最小的主行业 / 代码写法自适应")
def _():
    from app.repo import industry_repo as ir
    install_fakes(params={"PMS_SECTOR_SOURCE": "gp_hybk"})
    ind = _real_industry()
    import importlib
    importlib.reload(ir)
    orig = ir.fetch_all
    try:
        # 表里用前缀式 SH600000; 一票挂 二级×1 + 三级×2
        TBL = [
            {"gp_code": "SH600000", "bk_code": 881155, "bk_name": "银行(二级)"},
            {"gp_code": "SH600000", "bk_code": 884219, "bk_name": "股份制银行"},
            {"gp_code": "SH600000", "bk_code": 884101, "bk_name": "全国性银行"},
            {"gp_code": "SZ000001", "bk_code": 884219, "bk_name": "股份制银行"},
        ]
        calls = []

        def _fa(sql, params=None, source=None):
            assert source == "index", source          # 必须走 199, 不是 153 代理
            want = set((params or {}).values())
            calls.append(sorted(want))
            return [r for r in TBL if r["gp_code"] in want]
        ir.fetch_all = _fa
        p = ir.probe(force=True)
        assert p["form"] == "prefix", p               # 点式先试没命中, 自动落到前缀式
        m = ir.fetch_industries(["600000.SH", "000001.SZ"], level="l3")
        assert m["600000.SH"] == [("884101", "全国性银行"), ("884219", "股份制银行")], m
        # 主行业 = bk_code 升序第一个 (884101 < 884219), 二级 881 不能混进来
        pm = ir.primary_industry_map(["600000.SH", "000001.SZ"], level="l3")
        assert pm == {"600000.SH": "全国性银行", "000001.SZ": "股份制银行"}, pm
        assert ir.primary_industry_map(["600000.SH"], level="l2") == {"600000.SH": "银行(二级)"}
        # 服务层: ready 有证据, get_many 批量且按日缓存
        ind.invalidate()
        assert ind.ready() is True
        assert ind.get_many(["600000.SH", "000001.SZ"]) == {
            "600000.SH": "全国性银行", "000001.SZ": "股份制银行"}
        n = len(calls)
        ind.get_many(["600000.SH", "000001.SZ"])      # 二次不再打库
        assert len(calls) == n, calls[n:]
        assert ind.get("600000.SH") == "全国性银行"
        st = ind.status()
        assert st["ready"] is True and st["level"] == "l3" and "884" in st["hint"], st
    finally:
        ir.fetch_all = orig
        ir.invalidate()


@case("行业源·gp_hybk 三种写法都不命中 → ready=False, 错误说清是表能查还是查不了")
def _():
    from app.repo import industry_repo as ir
    install_fakes(params={"PMS_SECTOR_SOURCE": "gp_hybk"})
    ind = _real_industry()
    import importlib
    importlib.reload(ir)
    orig = ir.fetch_all
    try:
        ir.fetch_all = lambda sql, params=None, source=None: []      # 能查, 但没有这些票
        ind.invalidate()
        assert ind.ready() is False
        st = ind.status()
        assert st["ready"] is False and "用不了" in st["hint"], st["hint"]
        assert "三种代码写法都没命中" in st["hint"], st["hint"]
        assert "DB_MYSQL_URL" in st["hint"], st["hint"]     # 指到该改哪儿
        # 查不了的情况: 错误原文要冒上来
        ir.invalidate()

        def _boom(sql, params=None, source=None):
            raise RuntimeError("Table 'db_gp_cj.gp_hybk' doesn't exist")
        ir.fetch_all = _boom
        ind.invalidate()
        assert ind.ready() is False
        assert "doesn't exist" in ind.status()["hint"], ind.status()["hint"]
    finally:
        ir.fetch_all = orig
        ir.invalidate()


@case("行业源·先 SHOW COLUMNS 认列再查: 没有代码列时直接给可执行的报错")
def _():
    install_fakes(params={"PMS_SECTOR_SOURCE": "gp_stock_category"})
    ind = _real_industry()
    dr = _real_downstream()          # category_columns 也被 install_fakes 打过桩
    orig_fa, orig_fo = dr.fetch_all, dr.fetch_one
    try:
        # ① 实机遇到的情况: 表在, 但没有 stock_code 这类代码列
        dr._cat_cols.update({"at": 0.0, "data": None})
        dr.fetch_all = lambda sql, p=None: [{"Field": "gp_name"}, {"Field": "sw_l1"}]
        meta = dr.category_columns(force=True)
        assert meta["code_col"] is None and "没有可识别的代码列" in meta["error"], meta
        assert "gp_name" in meta["error"], meta["error"]      # 把实际列名报出来才可执行
        assert dr.probe_category("600000.SH")["value"] is None
        assert ind.ready() is False
        # ② 列认出来了: 代码写法逐个试, 命中即返回
        dr._cat_cols.update({"at": 0.0, "data": None})
        dr.fetch_all = lambda sql, p=None: [{"Field": "ts_code"}, {"Field": "sw_l1"}]
        seen = []

        def _one(sql, p=None):
            seen.append((p or {}).get("code"))
            return {"ts_code": "SH600000", "sw_l1": "银行"} if (p or {}).get("code") == "SH600000" else None
        dr.fetch_one = _one
        dr.category_columns(force=True)
        r = dr.probe_category("600000.SH")
        assert r["value"] == "银行" and r["industry_col"] == "sw_l1", r
        assert seen == ["600000.SH", "SH600000"], seen     # 点式先试, 再前缀式
        ind.invalidate()
        assert ind.ready() is True
    finally:
        dr.fetch_all, dr.fetch_one = orig_fa, orig_fo
        dr._cat_cols.update({"at": 0.0, "data": None})


@case("行业源·custom_table 映射表为空同样不算 ready (空表 = 全票 None = 约束失效)")
def _():
    fake = install_fakes(params={"PMS_SECTOR_SOURCE": "custom_table"})
    ind = _real_industry()
    assert ind.ready() is False
    assert "是空的" in ind.status()["hint"], ind.status()["hint"]
    fake.industry = {}
    ind.invalidate()
    from app.repo import pms_repo
    orig = pms_repo.list_industry
    try:
        pms_repo.list_industry = lambda limit=5000: [
            {"ts_code": "600000.SH", "industry": "银行", "updated_at": ""}]
        assert ind.ready() is True
        assert ind.get("600000.SH") == "银行"
    finally:
        pms_repo.list_industry = orig


# ================================================================ 候选池 (上游 /plan)
def _stub_sp(csvc, mapping):
    """替换 effective_stock_params, 返回 restore 闭包。"""
    orig = csvc.effective_stock_params
    csvc.effective_stock_params = lambda: dict(mapping)
    return lambda: setattr(csvc, "effective_stock_params", orig)


@case("候选池·plan_api 独占: 价格现取 / 无价剔除 / 黑名单与持仓剔除 / 白名单压过计划票")
def _():
    from app.services import command_service as csvc, plan_feed
    install_fakes(prices={"600418.SH": 12.5, "300952.SZ": 30.0, "600000.SH": 10.0,
                          "600104.SH": 18.0},
                  params={"PMS_TOTAL_SCALE": "980000", "PMS_CANDIDATE_SOURCE": "plan_api"})
    restore_sp = _stub_sp(csvc, {"600000.SH": {"white": True}, "600104.SH": {"black": True}})
    orig = plan_feed.candidates
    try:
        plan_feed.candidates = lambda **kw: {
            "date": "2026-07-29", "considered": 4, "eligible": 4, "dropped": {},
            "items": [{"ts_code": "600418.SH", "score": 242.24, "theme": "整车"},
                      {"ts_code": "688717.SH", "score": 242.05, "theme": "储能"},
                      {"ts_code": "300952.SZ", "score": 241.91, "theme": "传感器"},
                      {"ts_code": "600104.SH", "score": 241.18, "theme": "整车"}]}
        pool = csvc._candidates({"held": [{"ts_code": "605598.SH"}]})
    finally:
        plan_feed.candidates = orig
        restore_sp()
    by = {c["ts_code"]: c for c in pool}
    # 688717.SH 行情里没有价 → 必须剔除 (planner 对 price<=0 只是静默跳过, 那样少票看不出来)
    assert "688717.SH" not in by, by
    assert "600104.SH" not in by, "黑名单票不得进池"
    assert by["600418.SH"]["price"] == 12.5 and by["600418.SH"]["src"] == "plan_api"
    assert by["600418.SH"]["theme"] == "整车" and by["600418.SH"]["sector"] is None
    # 白名单是用户点名的票, 分必须压过任何计划票 —— planner 只认 score 一把尺子
    assert by["600000.SH"]["src"] == "whitelist"
    plan_max = max(c["score"] for c in pool if c["src"] == "plan_api")
    assert by["600000.SH"]["score"] > plan_max, (by["600000.SH"]["score"], plan_max)


@case("候选池·盘前无实时价回落昨收 (db13 存的是当日分钟线, 盘前那张 key 不存在)")
def _():
    from app.services import command_service as csvc, plan_feed
    install_fakes(prices={}, prev_close={"600418.SH": 12.5, "300952.SZ": 30.0},
                  params={"PMS_CANDIDATE_SOURCE": "plan_api"})
    restore_sp = _stub_sp(csvc, {})
    orig = plan_feed.candidates
    try:
        plan_feed.candidates = lambda **kw: {
            "date": "2026-07-30", "considered": 3, "eligible": 3, "dropped": {},
            "items": [{"ts_code": "600418.SH", "score": 242.24, "theme": "整车"},
                      {"ts_code": "300952.SZ", "score": 241.91, "theme": "传感器"},
                      {"ts_code": "688717.SH", "score": 242.05, "theme": "储能"}]}
        pool = csvc._candidates({"held": []})
    finally:
        plan_feed.candidates = orig
        restore_sp()
    by = {c["ts_code"]: c for c in pool}
    # 有昨收的进池并标明价格来源; 实时价与昨收都没有的才剔除
    assert set(by) == {"600418.SH", "300952.SZ"}, by
    assert by["600418.SH"]["price"] == 12.5
    assert by["600418.SH"]["price_source"] == "prev_close", by["600418.SH"]
    assert "688717.SH" not in by


@case("候选池·上游计划不可用: 池为空, 绝不静默回退旧 trading_buy_plan 表")
def _():
    from app.repo import downstream_repo
    from app.services import command_service as csvc, plan_feed
    install_fakes(prices={"601111.SH": 9.0}, params={"PMS_CANDIDATE_SOURCE": "plan_api"})
    restore_sp = _stub_sp(csvc, {})
    orig_c, orig_bp = plan_feed.candidates, downstream_repo.fetch_buy_plans

    def _boom(**kw):
        raise plan_feed.PlanFeedError("接口超时")
    try:
        plan_feed.candidates = _boom
        downstream_repo.fetch_buy_plans = lambda **kw: [
            {"ts_code": "601111.SH", "price": 9.0, "score": 0.8}]
        pool = csvc._candidates({"held": []})
    finally:
        plan_feed.candidates, downstream_repo.fetch_buy_plans = orig_c, orig_bp
        restore_sp()
    assert pool == [], f"上游拿不到时必须让候选池为空, 实际拿到 {pool}"


@case("候选池·both: 计划票与旧表票取并集, 同码只留计划那份")
def _():
    from app.repo import downstream_repo
    from app.services import command_service as csvc, plan_feed
    install_fakes(prices={"600418.SH": 12.5, "601111.SH": 9.0},
                  params={"PMS_CANDIDATE_SOURCE": "both"})
    restore_sp = _stub_sp(csvc, {})
    orig_c, orig_bp = plan_feed.candidates, downstream_repo.fetch_buy_plans
    try:
        plan_feed.candidates = lambda **kw: {
            "date": "2026-07-29", "considered": 1, "eligible": 1, "dropped": {},
            "items": [{"ts_code": "600418.SH", "score": 242.24, "theme": "整车"}]}
        downstream_repo.fetch_buy_plans = lambda **kw: [
            {"ts_code": "600418.SH", "price": 11.0, "score": 0.9},      # 同码, 计划优先
            {"ts_code": "601111.SH", "price": 9.0, "score": 0.8}]
        pool = csvc._candidates({"held": []})
    finally:
        plan_feed.candidates, downstream_repo.fetch_buy_plans = orig_c, orig_bp
        restore_sp()
    by = {c["ts_code"]: c for c in pool}
    assert set(by) == {"600418.SH", "601111.SH"}, by
    assert by["600418.SH"]["src"] == "plan_api" and by["600418.SH"]["price"] == 12.5
    assert by["601111.SH"]["src"] == "buy_plan" and by["601111.SH"]["price"] == 9.0


def main():
    # 静音日志: 本套里有好几条用例**故意**触发异常与告警来验证「守成」行为
    # (调度守卫吞异常、外部成交告警、连续对账升级 ERROR、窗口耗尽告警),
    # 这些 ERROR 栈打在测试输出里会被误读成失败。判定标准只看断言, 不看日志。
    import logging
    logging.disable(logging.CRITICAL)
    passed, failed = 0, 0
    for name, fn in RESULTS:
        try:
            fn()
            print(f"  PASS  {name}")
            passed += 1
        except Exception:
            print(f"  FAIL  {name}")
            traceback.print_exc()
            failed += 1
    print("-" * 60)
    if failed:
        print(f"FAILED: {failed} / {passed + failed}")
        sys.exit(1)
    print(f"ALL PASS ({passed} cases)")


if __name__ == "__main__":
    main()