# -*- 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._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", 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() await self._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): """启动自检: 恢复水位 + 把上次崩在半路的 SENDING 退回队列。""" try: await _db(qmt_repo.ensure_state) 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.exception("启动自检失败: %s", e) def _operable(self) -> bool: """通道是否**有可能**工作 (已启用 + 密钥齐)。不满足就不写心跳 —— 见 _beat_loop。""" seed, peer = self._secrets() return bool(self._p("enabled", False) and seed and peer) 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", 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)。未配齐前不连接, 也不写心跳 —— " "dispatcher 会因此一律拒发, 这是对的", error=True) 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, 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", 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", 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", 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 _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)