636 lines
34 KiB
Python
636 lines
34 KiB
Python
# -*- 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)
|