行情采集的断线重连：WebSocket 重连退避与数据补洞

做 A 股/港股量化的同学迟早会踩到同一个坑：行情 WebSocket 在前一天收盘后、或网络抖动时悄悄断开，第二天开盘你以为连着，其实数据已经断流半个多小时。盘后流水线跑得好好的，盘中却漏了一截，回测和信号全建立在残缺数据上。本文把我这几版行情客户端沉淀下来的重连与补洞做法整理出来，重点是可直接抄的代码片段。

一、为什么不能「断了就立刻重连」

最直觉的写法是在 on_close 里直接重新 connect。问题是当服务端限流、网关重启、或本地网络彻底挂掉时，客户端会以极高频率疯狂重连，瞬间把对方连接池打满，轻则被限流，重则被封 IP。正确做法是指数退避 + 抖动（exponential backoff with jitter）：第 1 次 1s，之后 2s、4s、8s……上限封到 30s，每次再叠加一个随机抖动，避免一批客户端同时重连形成共振。

import asyncio, random

class Backoff:
    def __init__(self, base=1.0, cap=30.0, factor=2.0):
        self.base, self.cap, self.factor = base, cap, factor
        self.attempt = 0
    def next_delay(self) -> float:
        exp = min(self.base * (self.factor ** self.attempt), self.cap)
        self.attempt += 1
        return random.uniform(0, exp)   # full jitter：在 [0, exp] 内取随机值
    def reset(self):
        self.attempt = 0

注意这里用 full jitter（在 [0, exp] 内取随机），而不是常见的 exp + 随机量。AWS 的「Exponential Backoff and Jitter」白皮书实测表明，full jitter 能把重试风暴的峰值压到接近理论下限，对服务端最友好。

二、心跳与超时，是两件不同的事

很多断线是「假死」：TCP 连接还在，但中间链路已经不通，on_close 根本不触发。必须主动发心跳（ping/pong），并设两个超时：

• 发送侧超时：超过 N 秒没发出去（写入缓冲区满/阻塞），主动断开；
• 接收侧超时：超过 M 秒没收到任何帧（含 pong），判定链路死掉，主动重连。

PING_INTERVAL = 15.0   # 每 15s 发一次 ping
RECV_TIMEOUT  = 40.0   # 40s 内没收到任何帧就认为死链

async def watchdog(recv_flag: asyncio.Event):
    while True:
        recv_flag.clear()
        try:
            await asyncio.wait_for(recv_flag.wait(), timeout=RECV_TIMEOUT)
        except asyncio.TimeoutError:
            raise ConnectionError("recv timeout, link suspected dead")

async def consumer(ws, recv_flag: asyncio.Event):
    async for msg in ws:
        recv_flag.set()           # 任何帧(含 pong)都算"活着"
        await handle(msg)

我一般发送间隔取 15s、接收超时取 40s：宁可误判一次重连，也不能长期假死。注意 pong 要计入 recv_flag——很多自研 client 只在收到业务帧时刷新，结果心跳 pong 不刷新标记，照样假死。

三、重连之后，数据怎么补（最容易被忽略）

重连成功只代表「从现在起有数据」，但断开那段时间里的 tick / 分钟线已经丢了。补洞分两层：

1. 交易所回补接口：很多行情源提供按时间区间拉历史成交/快照的 REST 接口。重连后先算「断线时刻 → 现在」的区间，用 REST 把缺失段补齐，再切回实时流。
2. 本地缓存去重：无论补洞还是实时，统一以「标的 + 时间戳 + 价位」做幂等键写入。这样就算补洞和实时流短暂重叠，也不会出现双份数据。

async def gap_fill(symbol: str, gap_start: float, now: float):
    candles = await rest_client.get_history(symbol, start=gap_start, end=now)
    for c in candles:
        await writer.upsert(c)        # 幂等写入，见第四节
    logging.info("gap filled %s: %d bars", symbol, len(candles))

补洞接口若也失败，不要无限重试：记一条告警并保留断点时间戳，等下一轮重连再补——宁可留一个已知缺口，也别让补洞请求把主流程拖死。

四、幂等写入：别让补洞和实时流重叠

补洞（REST）和实时（WebSocket）几乎一定会短暂重叠。写入层用唯一约束 + UPSERT 兜底：

CREATE TABLE IF NOT EXISTS bars (
    symbol TEXT NOT NULL, ts INTEGER NOT NULL, price REAL, volume REAL,
    PRIMARY KEY (symbol, ts)          -- 幂等键
);
INSERT INTO bars(symbol, ts, price, volume) VALUES (?, ?, ?, ?)
ON CONFLICT(symbol, ts) DO UPDATE SET price=excluded.price, volume=excluded.volume;

如果用 Redis 而非 SQLite，等价做法是对 key = {symbol}:{ts} 做 SETNX，已存在就跳过。核心是：写入层对重复免疫，上游才能放心并发补洞。

五、实测参数与结论

总结一句：重连解决「连不连得上」，补洞解决「数据完不完整」，幂等写入解决「补洞会不会造重复」，监控解决「你知不知道出问题了」——四件都得做，少一件都会在某天开盘悄悄坑你。