tradingSystem/app/ws/runner.py

# -*- coding: utf-8 -*-
"""
QMT WebSocket 常驻连接进程 (pms-ws)
====================================
协议: `QMT_WS_PROTOCOL.md` V1.0 · 端点 `ws://192.168.16.98:8080` · 明文 ws + Ed25519 双向签名

启动:  docker compose --profile ws up -d pms-ws
       (本地调试:  python -m app.ws.runner)

为什么单独一个进程
------------------
既有调度是 Celery beat 的「定时起一个短命任务」模型, 而 ws 要求一条**常驻**长连接, 且
协议 §1 明确「PMS 不开第二条连接」——否则指令会乱序。所以连接由本进程独占持有:

    executor (celery worker)  --写表-->  pms_qmt_order(QUEUED)  --轮询-->  本进程 --> QMT
    QMT --> 本进程 --落表--> pms_qmt_inbox --> ledger 任务 (celery worker) --入账--> 账本

本进程只做**通道**的事: 连接、签名、序号、补发、出口出栈、上行落库与确认。
**不碰账本** —— 成交入账仍然发生在 worker 里 (设计 §4「成交回放与对账」那条链路),
这样账本变更只有一个来源, 也不至于让一个 DB 慢查询把心跳拖到 15 秒超时断连。

四个协程
--------
  _beat_loop        每 2 秒写一次存活心跳 + 刷参数快照。dispatcher 靠这个心跳判断
                    「进程还在不在」—— 连接断了是一回事 (还能重连), 进程没了是另一回事。
  _reader_loop      收帧 → 验签 → 落 inbox → 推进 seq 水位。**recv 带 15 秒超时**,
                    超时即视为对端失联, 抛出去让外层重连 (协议 §1 的 idle 规则)。
  _pinger_loop      每 5 秒一条 ping。
  _outbox_loop      每 0.5 秒扫 pms_qmt_order 的 QUEUED 与待撤单, 签名下发。

三条容易写错的地方, 都在对应位置有注释:
  1. **落库失败绝不能 ack** —— 落不了库就主动断线重连, 让 QMT 从 last_seq+1 重发。
  2. **水位是连续前缀**, 不是收到的最大 seq。中间缺一条就不能往前跨。
  3. **所有 DB 调用走 to_thread** —— SQLAlchemy 是同步的, 直接在事件循环里调用会连带
     把心跳和 recv 一起阻塞掉, 表现为莫名其妙的周期性断连。
"""
from __future__ import annotations

import asyncio
import contextlib
import json
import logging
import signal
import sys
from datetime import datetime

from config.settings import settings
from app.core import ws_codec as wsc
from app.repo import qmt_repo
from app.services import param_store

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(levelname)s [%(name)s] %(message)s")
logger = logging.getLogger("pms.ws")

BACKOFF = (1, 2, 5, 10, 30)          # 协议 §1 重连退避, 30 秒封顶, 无限重试


class PersistFailed(RuntimeError):
    """上行消息落库失败。**必须**冒泡到重连逻辑 —— 见 _handle_upstream 的注释。"""


class HandshakeFailed(RuntimeError):
    pass


async def _db(fn, *a, **kw):
    """所有阻塞 DB 调用的唯一入口。见模块头第 3 条。"""
    return await asyncio.to_thread(fn, *a, **kw)


class WsRunner:
    def __init__(self):
        self._stop = asyncio.Event()
        self._ws = None
        self._msg_seq = 0
        self._last_seq = 0                # 已落库的连续水位
        self._acked_seq = 0               # 已发出 ack_seq 的水位
        self._pending_seq = set()         # 乱序暂存 (正常恒空)
        self._unacked = 0                 # 距上次 ack 又落了几条
        self._baselined = False           # 是否已对齐对端序号起点 (见 _set_baseline)
        self._db_ready = False            # 通道三表是否可用 (缺表时空转重试, 不写心跳)
        self._params = {}
        self._stat = {"rx": 0, "tx": 0, "trades": 0, "rejects": 0, "reconnects": 0,
                      "last_rx_at": None, "last_tx_at": None}

    # ============================================================ 配置
    def _p(self, key, default):
        return self._params.get(key, default)

    async def _refresh_params(self):
        """业务参数走 ParamStore (页面可调, 5 秒缓存); 密钥与端点只走 .env。"""
        def _load():
            return {
                "enabled": param_store.get_bool("PMS_QMT_WS_ENABLED", False),
                "url": param_store.get("PMS_QMT_WS_URL", settings.PMS_QMT_WS_URL),
                "heartbeat_sec": param_store.get_int("PMS_QMT_HEARTBEAT_SEC", 5),
                "idle_timeout_sec": param_store.get_int("PMS_QMT_IDLE_TIMEOUT_SEC", 15),
                "ack_batch": param_store.get_int("PMS_QMT_ACK_BATCH", 20),
                "ack_interval_sec": param_store.get_float("PMS_QMT_ACK_INTERVAL_SEC", 2.0),
                "outbox_poll_sec": param_store.get_float("PMS_QMT_OUTBOX_POLL_SEC", 0.5),
                "beat_sec": param_store.get_int("PMS_QMT_HEARTBEAT_DB_SEC", 2),
                "max_attempts": param_store.get_int("PMS_QMT_SEND_MAX_ATTEMPTS", 3),
                "connect_timeout": param_store.get_int("PMS_QMT_CONNECT_TIMEOUT_SEC", 10),
            }
        try:
            self._params = await _db(_load)
        except Exception as e:                  # 参数表读不到就用上一份 / 文件初值
            if not self._params:
                self._params = {"enabled": settings.PMS_QMT_WS_ENABLED,
                                "url": settings.PMS_QMT_WS_URL, "heartbeat_sec": 5,
                                "idle_timeout_sec": 15, "ack_batch": 20,
                                "ack_interval_sec": 2.0, "outbox_poll_sec": 0.5,
                                "beat_sec": 2, "max_attempts": 3, "connect_timeout": 10}
            logger.warning("参数刷新失败, 沿用上一份: %s", _brief_err(e))

    @staticmethod
    def _secrets() -> tuple:
        """(私钥 seed, 对端公钥)。只从 .env 注入 —— 协议 §10.1.1: 不入库、不进 ParamStore。"""
        return ((settings.PMS_QMT_SIGN_SEED_HEX or "").strip(),
                (settings.PMS_QMT_PEER_PUBKEY_B64 or "").strip())

    # ============================================================ 主循环
    async def run(self):
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            with contextlib.suppress(NotImplementedError):
                loop.add_signal_handler(sig, self._request_stop, sig)

        await self._refresh_params()
        # 注意: 不在这里做 _boot —— 建表检查放进连接循环, 通道没启用时就完全不碰库,
        # 表建好之后也能自己恢复, 不必重启容器。
        beat = asyncio.create_task(self._beat_loop(), name="beat")
        try:
            await self._connection_loop()
        finally:
            beat.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await beat
            await self._shutdown()

    def _request_stop(self, sig):
        logger.info("收到 %s, 开始优雅退出 (停止取新单 → 刷水位 → 最后一次 ack → 关连接)",
                    getattr(sig, "name", sig))
        self._stop.set()

    async def _boot(self) -> bool:
        """启动自检: 三表在不在 → 恢复水位 → 把上次崩在半路的 SENDING 退回队列。

        返回 False 表示还不能开工 (通常是表没建), 由连接循环退回空转并定期重试。
        """
        try:
            await _db(qmt_repo.ensure_state)
        except Exception as e:
            self._db_ready = False
            msg = _brief_err(e)
            if _looks_like_missing_table(msg):
                # 缺表是**部署少跑了一步**, 不是异常。给一行照着做就能好的话,
                # 而不是六十行 SQLAlchemy 堆栈 —— 后者会把真正该看的信息埋掉。
                logger.error(
                    "ws 通道三表还没建 (%s)。先建表再起本进程:\n"
                    "      docker compose run --rm pms-web python scripts/init_db.py --yes\n"
                    "      docker compose run --rm pms-web python scripts/check_db.py\n"
                    "    建完**不用重启容器**, 本进程每 60 秒自己重试。在此之前不写心跳, "
                    "dispatcher 会据此拒发指令 —— 这是对的。", msg)
            else:
                logger.error("通道表自检失败 (每 60 秒重试): %s", msg)
            return False

        self._db_ready = True
        try:
            st = await _db(qmt_repo.get_state)
            stored = int(st.get("last_seq") or 0)
            # last_seq 落库是按批刷的, 可能落后于实际已落库的消息 —— 以 inbox 为准重算。
            # 少 ack 只是让 QMT 多留一会儿; 多 ack 会让数据永久丢失, 故从保守值往前推。
            self._last_seq = await _db(qmt_repo.inbox_recover_watermark, stored)
            self._acked_seq = min(int(st.get("acked_seq") or 0), self._last_seq)
            if self._last_seq != stored:
                logger.warning("水位由 inbox 重算: 落库值 %s → 实际 %s", stored, self._last_seq)
            n = await _db(qmt_repo.reset_stuck_sending)
            if n:
                # 重发是安全的: 协议 §2.2 规定重复 instruction_id 不会二次下单,
                # 只回 ack{duplicate:true} 带当前状态。幂等键就是为这一刻准备的。
                logger.warning("有 %s 张委托上次卡在 SENDING, 已退回队列重发 (幂等键兜底)", n)
        except Exception as e:
            logger.error("启动自检失败, 水位按 0 起算 (下轮重连会重试): %s", _brief_err(e))
            return False
        return True

    def _operable(self) -> bool:
        """通道是否**有可能**工作 (已启用 + 密钥齐 + 表已建)。不满足就不写心跳 —— 见 _beat_loop。"""
        seed, peer = self._secrets()
        return bool(self._p("enabled", False) and seed and peer and self._db_ready)

    async def _beat_loop(self):
        """存活心跳。**连不上 QMT 也照跳** —— 它证明的是「进程在」而不是「连接通」,
        dispatcher 靠这两者的区别决定卖出要不要继续入队 (协议 §6.3 减持不挡)。

        但通道压根没启用或密钥没配齐时**故意不跳**: 那种状态下队列里的单子永远发不出去,
        让 dispatcher 直接判定进程不可用、一律拒发, 比让它把卖出排进一个死队列强。
        """
        while not self._stop.is_set():
            try:
                await self._refresh_params()
                if self._operable():
                    await _db(qmt_repo.beat, self._stat)
            except Exception as e:
                logger.warning("心跳写入失败 (库不可用?): %s", _brief_err(e))
            with contextlib.suppress(asyncio.TimeoutError):
                await asyncio.wait_for(self._stop.wait(), timeout=self._p("beat_sec", 2))

    async def _connection_loop(self):
        attempt = 0
        while not self._stop.is_set():
            if not self._p("enabled", False):
                await self._idle_note("PMS_QMT_WS_ENABLED=False, 通道未启用 (页面可开)")
                continue
            seed, peer = self._secrets()
            if not seed or not peer:
                await self._idle_note(
                    "缺少 Ed25519 密钥: PMS_QMT_SIGN_SEED_HEX / PMS_QMT_PEER_PUBKEY_B64 "
                    "须在 .env 注入 (协议 §10.1.1)。生成与自检: "
                    "python scripts/gen_keys.py [--check]。未配齐前不连接, 也不写心跳 —— "
                    "dispatcher 会因此一律拒发, 这是对的", error=True)
                continue
            if not self._db_ready and not await self._boot():
                # 建表提示已在 _boot 里打过一次, 这里只安静等 —— 不重复刷屏
                with contextlib.suppress(asyncio.TimeoutError):
                    await asyncio.wait_for(self._stop.wait(), timeout=60)
                continue

            url = self._p("url", settings.PMS_QMT_WS_URL)
            try:
                await _db(qmt_repo.set_conn, "CONNECTING")
                await self._session(url, seed, peer)
                attempt = 0                       # 正常断开 (优雅退出) 才会走到这
            except asyncio.CancelledError:
                raise
            except Exception as e:
                self._stat["reconnects"] += 1
                delay = BACKOFF[min(attempt, len(BACKOFF) - 1)]
                attempt += 1
                logger.warning("连接中断 (%s: %s), %s 秒后重连 [第 %s 次]",
                               type(e).__name__, e, delay, attempt)
                with contextlib.suppress(Exception):
                    await _db(qmt_repo.set_conn, "OFFLINE", last_error=f"{type(e).__name__}: {e}")
                with contextlib.suppress(asyncio.TimeoutError):
                    await asyncio.wait_for(self._stop.wait(), timeout=delay)

    async def _idle_note(self, msg: str, error: bool = False, every: int = 60):
        """未启用/缺密钥时的空转。**故意不写心跳** —— 让 dispatcher 判定进程不可用。"""
        (logger.error if error else logger.info)("[ws 空转] %s", msg)
        with contextlib.suppress(Exception):
            await _db(qmt_repo.set_conn, "OFFLINE", last_error=msg, beat=False)
        with contextlib.suppress(asyncio.TimeoutError):
            await asyncio.wait_for(self._stop.wait(), timeout=every)

    # ============================================================ 一次会话
    async def _session(self, url: str, seed: str, peer: str):
        import websockets                       # 延迟导入: 未启用通道时不强求装这个包

        logger.info("连接 QMT %s ...", url)
        async with websockets.connect(
                url, ping_interval=None,        # 关掉库自带 ping, 用协议层的 ping/pong
                open_timeout=self._p("connect_timeout", 10),
                close_timeout=5, max_size=4 * 1024 * 1024) as ws:
            self._ws = ws
            await self._handshake(seed, peer)
            await _db(qmt_repo.set_conn, "ONLINE", connected_at=datetime.now(),
                      last_error="")
            logger.info("握手完成, 通道在线 (本端水位 last_seq=%s)", self._last_seq)
            tasks = [asyncio.create_task(self._reader_loop(peer), name="reader"),
                     asyncio.create_task(self._pinger_loop(seed), name="pinger"),
                     asyncio.create_task(self._outbox_loop(seed), name="outbox"),
                     asyncio.create_task(self._ack_loop(seed), name="ack")]
            # 停机信号单独占一个 waiter: reader 阻塞在 15 秒 recv 上, 光等它自己醒来会让
            # 优雅退出白白拖满一个 idle 周期。谁先结束就收工, 剩下的直接 cancel。
            stopper = asyncio.create_task(self._stop.wait(), name="stopper")
            try:
                done, _ = await asyncio.wait(tasks + [stopper],
                                             return_when=asyncio.FIRST_COMPLETED)
                for t in done:                  # 让第一个异常冒泡去触发重连
                    if t is not stopper and t.exception():
                        raise t.exception()
            finally:
                for t in tasks + [stopper]:
                    t.cancel()
                await asyncio.gather(*tasks, stopper, return_exceptions=True)
                self._ws = None

    async def _handshake(self, seed: str, peer: str):
        """§4.1 hello → §5.1 hello_ack。last_seq 决定 QMT 从哪一条开始补发。"""
        await self._send(wsc.T_HELLO, wsc.hello_payload(self._last_seq), seed)
        try:
            raw = await asyncio.wait_for(self._ws.recv(), timeout=15)
        except asyncio.TimeoutError as e:
            raise HandshakeFailed("等 hello_ack 超时 15 秒") from e
        env = wsc.parse(raw, peer_pubkey_b64=peer)
        if env["type"] != wsc.T_HELLO_ACK:
            raise HandshakeFailed(f"握手期收到非 hello_ack 消息: {env['type']}")
        pl = env["payload"]
        server_seq = int(pl.get("server_seq") or 0)
        resync = bool(pl.get("resync_required"))
        resume_from = int(pl.get("resume_from") or 0)
        if resume_from:
            resync = await self._set_baseline(resume_from, "hello_ack.resume_from") or resync
            # 注意**不要**在这里就置 _baselined: 对端很可能只是把我们传的 last_seq 加一原样
            # 回填 (冷启动时那就是 1), 而它真正要发的第一条其实是 10001。基线还得靠首条
            # 消息兜一次 —— _set_baseline 本身幂等, 已对齐的话是空操作。
        await _db(qmt_repo.set_conn, "CONNECTING", server_seq=server_seq, resync=resync)
        if resync:
            # §5.1 / §6.2: 对端补不齐我们要的区间 (日志已滚动)。此时**不能**装作没事 ——
            # 中间那段成交我们永远拿不到了, 必须走全量快照对账, 对不齐就停一切自主动作。
            logger.error("resync 触发 (server_seq=%s, resume_from=%s, 本端水位 %s): "
                         "补发不全, 已置 resync_flag —— 请走全量对账后经页面清除标记",
                         server_seq, resume_from, self._last_seq)
        if env.get("seq") is not None:
            await self._persist_upstream(env)

    async def _set_baseline(self, first_seq: int, why: str) -> bool:
        """对齐对端的序号起点。返回「是否为真缺口」(真缺口要置 resync)。

        对端 seq 跨重启不回退, 首次对接时它可能已经是几万了; 而我们按 §4.1 传 last_seq=0。
        不对齐基线的话水位会永远卡在 0 —— 具体后果见 ws_codec.cold_start_baseline 的注释,
        那是一种完全静默的死法, 所以这里宁可多打两行日志。
        """
        baseline, is_gap = wsc.cold_start_baseline(self._last_seq, first_seq)
        if baseline == self._last_seq:
            return False
        old = self._last_seq
        self._last_seq = baseline
        self._pending_seq = {s for s in self._pending_seq if s > baseline}
        self._acked_seq = max(self._acked_seq, baseline)
        with contextlib.suppress(Exception):
            await _db(qmt_repo.save_watermark, baseline, self._acked_seq)
        if is_gap:
            logger.error("序号缺口: 本端水位 %s, 对端从 %s 起 (%s) —— 中间 %s 条永久缺失。"
                         "基线已推到 %s 保证通道继续可用, 但必须走全量对账", old, first_seq,
                         why, first_seq - old - 1, baseline)
        else:
            logger.info("冷启动: 本端无历史水位, 按对端起点 %s (%s) 建立基线 %s",
                        first_seq, why, baseline)
        return is_gap

    # ============================================================ 上行
    async def _reader_loop(self, peer: str):
        idle = self._p("idle_timeout_sec", 15)
        while not self._stop.is_set():
            try:
                raw = await asyncio.wait_for(self._ws.recv(), timeout=idle)
            except asyncio.TimeoutError as e:
                # §1: 任一侧 15 秒未收到对端消息即主动断开重连。我们每 5 秒发 ping,
                # 对方立即回 pong —— 15 秒还静默, 这条连接已经不能用了。
                raise ConnectionError(f"{idle} 秒未收到对端任何消息, 主动断开") from e
            self._stat["rx"] += 1
            self._stat["last_rx_at"] = datetime.now().strftime("%H:%M:%S")
            try:
                env = wsc.parse(raw, peer_pubkey_b64=peer)
            except wsc.CodecError as e:
                # §2.1: 验签失败直接丢弃, **不执行任何业务动作**。不回 reject ——
                # 连不上信任的对端时, 多说一句话只是多给攻击者一个探测面。
                logger.error("上行消息校验失败, 已丢弃 (%s): %s", e.code, e.message)
                continue
            await self._handle_upstream(env)

    async def _handle_upstream(self, env: dict):
        type_, pl = env["type"], env.get("payload") or {}
        seq = env.get("seq")

        if seq is None:
            # pong 没有业务内容, 不带 seq 很正常, 静默即可 (它的作用是让 recv 不超时)。
            # 其余类型缺 seq 则是对端实现问题: 这条消息进不了水位也就无法确认, 要吼一声。
            if type_ != wsc.T_PONG:
                logger.warning("上行 %s 缺少 seq, 无法纳入水位与补发, 仅记日志: %s", type_,
                               json.dumps(pl, ensure_ascii=False)[:200])
                await self._apply_side_effects(type_, pl, env)
            return
        if not self._baselined:
            # 兜底: 对端 hello_ack 没给 resume_from 时, 用第一条消息的 seq 建立基线。
            # 有 resume_from 的话握手时已经对齐过, 这里是个空操作。
            self._baselined = True
            if await self._set_baseline(int(seq), "首条上行消息"):
                with contextlib.suppress(Exception):
                    await _db(qmt_repo.set_conn, "ONLINE", resync=True)
        if int(seq) <= self._last_seq:
            return                              # 第一层去重: 补发时同一条消息 seq 不变

        await self._persist_upstream(env)       # 先落库
        await self._apply_side_effects(type_, pl, env)   # 再更新通道状态

    async def _persist_upstream(self, env: dict):
        """落 inbox 并推进水位。**落库失败绝不能 ack**。

        协议 §4.5 写得很清楚: 确认前必须已持久化, 否则 QMT 清了消息、PMS 又崩在落库前,
        那段数据就永久丢了。所以这里落库失败不是「记个日志继续」, 而是抛 PersistFailed
        主动断线 —— 重连时 hello 带的还是旧 last_seq, QMT 会把这段重新发一遍。
        协议的补发机制正是为这种情况准备的, 用它比自己攒重试队列稳妥得多。
        """
        type_, pl, seq = env["type"], env.get("payload") or {}, int(env["seq"])
        needs_ledger = type_ in wsc.NEEDS_LEDGER
        note = None if needs_ledger else f"{type_} 由通道进程消化, 不入账"
        try:
            r = await _db(qmt_repo.inbox_put, seq=seq, msg_id=str(env.get("msg_id") or ""),
                          msg_type=type_, payload=pl, msg_ts=int(env.get("ts") or 0),
                          corr_id=env.get("corr_id") or pl.get("instruction_id"),
                          dedup_key=wsc.dedup_key(type_, pl),
                          processed=0 if needs_ledger else 2, note=note)
        except Exception as e:
            raise PersistFailed(f"上行 seq={seq} ({type_}) 落库失败, 断线让对端重发: {e}") from e
        if r == qmt_repo.PUT_DUP_KEY:
            logger.warning("成交 %s 重复推送 (seq=%s), 已占位不入账",
                           pl.get("trade_no"), seq)
        self._last_seq, self._pending_seq = wsc.next_watermark(
            self._last_seq, self._pending_seq, seq)
        self._unacked += 1
        if self._pending_seq:
            logger.warning("上行乱序: 水位卡在 %s, 暂存 %s 条 (待缺口补齐)",
                           self._last_seq, len(self._pending_seq))
        if self._unacked >= self._p("ack_batch", 20):
            await self._flush_ack()

    async def _apply_side_effects(self, type_: str, pl: dict, env: dict):
        """把上行消息落到 pms_qmt_order 的状态上。**只动通道状态, 不动账本**。"""
        iid = pl.get("instruction_id") or env.get("corr_id")
        try:
            if type_ == wsc.T_ACK:
                await _db(qmt_repo.update_order, iid,
                          status=str(pl.get("status") or wsc.ST_ACCEPTED).upper(),
                          broker_order_id=pl.get("broker_order_id"))
                if pl.get("duplicate"):
                    logger.info("[ack] %s 幂等命中 (对端已受理过), 当前状态 %s",
                                iid, pl.get("status"))
            elif type_ == wsc.T_REJECT:
                self._stat["rejects"] += 1
                code = str(pl.get("code") or "")
                await _db(qmt_repo.update_order, iid, status=qmt_repo.OS_REJECTED,
                          reject_code=code[:32], reject_reason=str(pl.get("reason") or "")[:300],
                          final_at=datetime.now())
                # 设计 §13: 指令下发失败不自动重发。可重试码也只是记下来,
                # 由 executor 下一跳按最新行情重新决定 —— 换价重发的判断权在择时模块。
                lvl = logger.error if code == "INTERNAL" else logger.warning
                lvl("[reject] %s %s: %s (retryable=%s, 不自动重发)", iid, code,
                    pl.get("reason"), pl.get("retryable"))
            elif type_ == wsc.T_ORDER_UPDATE:
                await self._on_order_update(iid, pl)
            elif type_ == wsc.T_TRADE:
                self._stat["trades"] += 1
                if not wsc.trade_amount_ok(pl):
                    logger.warning("[trade] %s 金额自洽性存疑: price×qty ≠ amount (%s)",
                                   pl.get("trade_no"), json.dumps(pl, ensure_ascii=False)[:200])
                logger.info("[trade] %s %s %s股 @%s (待 worker 入账)", iid,
                            pl.get("ts_code"), pl.get("qty"), pl.get("price"))
            elif type_ == wsc.T_PERSIST:
                if not pl.get("ok"):
                    # §5.6: 落库失败不影响成交事实, 但要留一条下游不一致告警
                    logger.error("[persist_result] 下游落库失败 %s scope=%s: %s", iid,
                                 pl.get("scope"), pl.get("error"))
        except Exception as e:
            # 通道状态更新失败不该拖垮连接 —— 账本不靠它, 靠 inbox 里的 trade。
            logger.warning("通道状态更新失败 %s %s: %s", type_, iid, _brief_err(e))

    async def _on_order_update(self, iid: str, pl: dict):
        status = str(pl.get("status") or "").upper()
        row = await _db(qmt_repo.get_order, iid) or {}
        cur = str(row.get("status") or "").upper()
        if wsc.is_final(cur) and status != cur:
            # §5.4/§7.1: 终态唯一、不可再变、只推一次。真收到第二条终态说明对端有 bug ——
            # 此时**保住第一条**: 账本记的是"到底成交了多少", 被一条迟到的 CANCELLED 覆盖掉
            # 已经 FILLED 的记录, 比丢一条消息严重得多。
            logger.error("[order_update] %s 已是终态 %s, 又收到 %s —— 已忽略, 请核对对端实现",
                         iid, cur, status)
            return
        fields = {"status": status, "cum_qty": int(pl.get("cum_qty") or 0)}
        if pl.get("cum_avg_price") is not None:
            fields["cum_avg_price"] = float(pl["cum_avg_price"])
        if pl.get("leaves_qty") is not None:
            fields["leaves_qty"] = int(pl["leaves_qty"])
        if pl.get("final") or wsc.is_final(status):
            fields["final_at"] = datetime.now()
            requested = bool(row.get("cancel_state") in
                             (qmt_repo.CANCEL_REQUESTED, qmt_repo.CANCEL_SENT))
            # §7.2: 落库层 CANCELLED/EXPIRED 都写 cancelled, 但协议层对端会如实区分。
            # 这里两边对一下 —— 对不上说明有一侧记错了, 值得看一眼, 但仍以对端为准
            # (我们没发过撤单却收到 CANCELLED, 更可能是我们漏记而不是对方乱填)。
            if status == wsc.ST_CANCELLED and not requested:
                logger.warning("[order_update] %s 回 CANCELLED 但本地未发过撤单 —— "
                               "疑似到期撤 (EXPIRED) 被填成主动撤, 请核对", iid)
            elif status == wsc.ST_EXPIRED and requested:
                logger.warning("[order_update] %s 回 EXPIRED 但本地已请求撤单 —— "
                               "撤单可能在到期后才到, 按到期处理", iid)
            logger.info("[order_update] %s 终态 %s, 累计成交 %s 股", iid, status,
                        fields["cum_qty"])
        await _db(qmt_repo.update_order, iid, **fields)

    # ============================================================ 确认
    async def _ack_loop(self, seed: str):
        while not self._stop.is_set():
            with contextlib.suppress(asyncio.TimeoutError):
                await asyncio.wait_for(self._stop.wait(),
                                       timeout=self._p("ack_interval_sec", 2.0))
            await self._flush_ack(seed)

    async def _flush_ack(self, seed: str = None):
        """§4.5 累积确认: 只发当前**连续**水位。先把水位落库, 再 ack。"""
        if self._last_seq <= self._acked_seq:
            self._unacked = 0
            return
        seed = seed or self._secrets()[0]
        try:
            await _db(qmt_repo.save_watermark, self._last_seq, self._acked_seq)
        except Exception as e:
            logger.warning("水位落库失败, 本轮不 ack (宁可让对端多留一会儿): %s", _brief_err(e))
            return
        try:
            await self._send(wsc.T_ACK_SEQ, wsc.ack_seq_payload(self._last_seq), seed)
        except Exception as e:
            logger.warning("ack_seq 发送失败 (下轮重试): %s", _brief_err(e))
            return
        self._acked_seq = self._last_seq
        self._unacked = 0
        with contextlib.suppress(Exception):
            await _db(qmt_repo.save_watermark, self._last_seq, self._acked_seq)

    # ============================================================ 下行
    async def _pinger_loop(self, seed: str):
        while not self._stop.is_set():
            with contextlib.suppress(asyncio.TimeoutError):
                await asyncio.wait_for(self._stop.wait(),
                                       timeout=self._p("heartbeat_sec", 5))
            if self._stop.is_set():
                return
            await self._send(wsc.T_PING, {}, seed)

    async def _outbox_loop(self, seed: str):
        """出口出栈: QUEUED → 签名 → send → SENT。顺带处理待撤单。

        连接类异常**往上抛**去触发重连; 其它异常 (取数失败、单条参数不合法) 记日志继续 ——
        「故障即守成」是不产生新指令, 不是一有毛病就把整条通道停掉。
        """
        while not self._stop.is_set():
            try:
                await self._drain_outbox(seed)
                await self._drain_cancels(seed)
            except asyncio.CancelledError:
                raise
            except Exception as e:
                if _is_conn_error(e):
                    raise
                logger.exception("出口轮询异常 (不产生新指令, 下轮继续): %s", _brief_err(e))
            with contextlib.suppress(asyncio.TimeoutError):
                await asyncio.wait_for(self._stop.wait(),
                                       timeout=self._p("outbox_poll_sec", 0.5))

    async def _drain_outbox(self, seed: str):
        if self._stop.is_set():
            return                              # 优雅退出: 不再取新单
        rows = await _db(qmt_repo.next_queued, 20)
        now = wsc.now_ms()
        for r in rows:
            iid = r["instruction_id"]
            if int(r["valid_until"] or 0) <= now:
                # 排队期间就过期了。发出去只会立刻换回一个 EXPIRED, 白跑一趟还占对端一条
                # 记录 —— 本地作废更干净, executor 下一跳会按新行情重新出手。
                await _db(qmt_repo.abort_order, iid, "排队期间 valid_until 已过, 未发出即作废")
                logger.warning("[outbox] %s 有效期已过, 本地作废未发出", iid)
                continue
            if not await _db(qmt_repo.claim_order, iid):
                continue                        # 被别的进程抢走了 (滚动重启的瞬间)
            try:
                payload = wsc.place_order_payload(
                    instruction_id=iid, ts_code=r["ts_code"], side=r["side"],
                    qty=int(r["qty"]), limit_price=r["limit_price"],
                    valid_until=int(r["valid_until"]), intent=r.get("intent") or "OPEN",
                    note=r.get("note") or "")
                await self._send(wsc.T_PLACE, payload, seed, corr_id=iid)
            except Exception as e:
                # 连接断了不算这张单的"失败次数" —— 那是通道的问题不是这张单的问题,
                # 否则一次几秒的网络抖动就能把三张待发单全部烧成 SEND_FAILED。
                conn = _is_conn_error(e)
                await _db(qmt_repo.requeue_order, iid, f"{type(e).__name__}: {e}",
                          self._p("max_attempts", 3), not conn)
                logger.warning("[outbox] %s 发送失败%s: %s", iid,
                               " (连接问题, 不计失败次数)" if conn else "", e)
                raise
            await _db(qmt_repo.mark_sent, iid)
            logger.info("[outbox] 已发出 %s %s %s %s股 @%.2f", iid, r["ts_code"],
                        r["side"], r["qty"], float(r["limit_price"]))

    async def _drain_cancels(self, seed: str):
        rows = await _db(qmt_repo.next_cancel_requests, 20)
        for r in rows:
            iid = r["instruction_id"]
            if r["status"] in (qmt_repo.OS_QUEUED, qmt_repo.OS_SENDING):
                # 还没发出去就要撤 —— 直接本地作废, 不必跑一趟下游
                await _db(qmt_repo.abort_order, iid, "发出前撤销")
                await _db(qmt_repo.mark_cancel_sent, iid)
                logger.info("[cancel] %s 尚未发出, 本地作废", iid)
                continue
            payload = wsc.cancel_order_payload(
                cancel_id=r.get("cancel_id") or wsc.new_cancel_id(_ymd()),
                instruction_id=iid)
            await self._send(wsc.T_CANCEL, payload, seed, corr_id=iid)
            await _db(qmt_repo.mark_cancel_sent, iid)
            logger.info("[cancel] 已发出撤单 %s", iid)

    async def _send(self, type_: str, payload: dict, seed: str, corr_id=None):
        if self._ws is None:
            raise ConnectionError(f"连接未就绪, 无法发送 {type_}")
        self._msg_seq += 1
        env = wsc.build(type_, payload, seed_hex=seed,
                        msg_id=wsc.new_msg_id(_ymd(), self._msg_seq), corr_id=corr_id)
        await self._ws.send(wsc.dumps(env))
        self._stat["tx"] += 1
        self._stat["last_tx_at"] = datetime.now().strftime("%H:%M:%S")

    # ============================================================ 收尾
    async def _shutdown(self):
        """优雅退出: 刷水位 → 最后一次 ack → 关连接 → 置 STOPPED 并**清空心跳**。

        清心跳是关键一步: 不清的话, 停机后的 stale 窗口 (默认 15 秒) 里 dispatcher 仍
        认为进程活着, 卖出指令还会往队列里排 —— 而已经没人会发它们了。
        """
        with contextlib.suppress(Exception):
            if self._ws is not None:
                await asyncio.wait_for(self._flush_ack(), timeout=3)
        with contextlib.suppress(Exception):
            await _db(qmt_repo.save_watermark, self._last_seq, self._acked_seq)
        with contextlib.suppress(Exception):
            if self._ws is not None:
                await asyncio.wait_for(self._ws.close(code=1001, reason="pms shutdown"),
                                       timeout=5)
        with contextlib.suppress(Exception):
            await _db(qmt_repo.mark_stopped, "进程正常退出")
        logger.info("已退出 (水位 last_seq=%s / acked=%s, 收 %s 发 %s)",
                    self._last_seq, self._acked_seq, self._stat["rx"], self._stat["tx"])


def _ymd() -> int:
    return int(datetime.now().strftime("%Y%m%d"))


def _brief_err(e: BaseException) -> str:
    """异常压成一行。

    SQLAlchemy 的报错里会把整条 SQL 和全部参数带上, 动辄十几行; 再叠一层
    logger.exception 的堆栈, 一个「表还没建」能刷出六十行, 真正有用的那半句反而被埋掉。
    日志是给人看的, 定位靠的是第一句话。
    """
    first = str(e).splitlines()[0].strip()
    return f"{type(e).__name__}: {first[:220]}"


def _looks_like_missing_table(msg: str) -> bool:
    """这条报错是不是「表不存在」。MySQL 是 1146, ShardingSphere 代理回的是 10002。"""
    low = msg.lower()
    return ("does not exist" in low or "doesn't exist" in low
            or "1146" in low or "10002" in low)


def _is_conn_error(e: BaseException) -> bool:
    """这个异常是不是"连接没了"。

    按类名判断而不是 isinstance, 是为了不在模块顶层 import websockets ——
    通道未启用时那个包可以不装, 而本模块 (以及导入它的 check 脚本) 仍要能加载。
    """
    if isinstance(e, (ConnectionError, OSError, asyncio.IncompleteReadError)):
        return True
    return type(e).__name__ in ("ConnectionClosed", "ConnectionClosedOK",
                                "ConnectionClosedError", "WebSocketException",
                                "InvalidHandshake", "InvalidStatus", "InvalidStatusCode",
                                "HandshakeFailed", "PersistFailed")


async def main():
    logger.info("pms-ws 启动 · 协议 V1.0 · 端点 %s · 模式 %s", settings.PMS_QMT_WS_URL,
                "启用" if settings.PMS_QMT_WS_ENABLED else "未启用 (空转)")
    await WsRunner().run()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        sys.exit(0)