部分订阅：重订阅丢失的，总是最后一批

一个 340 个标的的 watchlist，随交易所在 02:00 一起中断，02:11 重订阅完成。每一行日志都是 connected，每一批都显示"已发送"。三天后对账发现有两个标的——而且始终是同样两个——从 02:00 起就没有数据。它们是最后一批订阅请求里的最后两个 code。没有报错，没有告警，客户端会一直"健康"下去，直到有人把"已确认的订阅集合"和"请求过的订阅集合"放到一起比。

这个故障有明确的形状，而且不是断线的形状。连接是好的，重订阅发出去了，大部分标的照常流动。坏掉的是一批，在尾部，不带任何事件。

重连本身的机制——检测断开、退避、恢复状态——WebSocket 重连最佳实践已经讲过；重连产生的重复 tick 及其成因，三个订阅状态陷阱已经讲过。两者都没有讲的，是夹在中间的那种失败：重订阅在部分批次上成功，其余静默丢失。这就是本文。

一句话版本：重订阅是按批确认的。数确认数量只能证明"回来了一些"，集合比对才能证明"全都回来了"。

订阅为什么是分批的

WebSocket 订阅消息按套餐设上限：定价页列出了各档位的单连接标的数上限，超过上限的那一批会以 error 帧返回——WS_CONN_LIMIT 就是超出当前档位连接或标的限额时流的返回码。这个上限不是实现细节，它就是这类故障存在的原因。

几百个标的的 watchlist 无论哪个档位都没法一条消息发完，客户端必须拆分、发送、收集确认。每一个拆分点都是一个位置：这一批被接受、下一批没有被接受——被限流丢掉、因容量被拒、或者丢在一个从未被按 error 解析的帧里。丢掉至少一批的概率随批次数上升，也就是说，watchlist 越大，这个故障越常见。

而且丢失在 watchlist 上不是均匀分布的。第一批通常没事：它刚进入发送循环，循环外围的任何重试逻辑都会再触发它。尾部不一样。它是最后写进去的、最先被假设成功的、最没有人检查的——因为循环跑完，代码就进入恢复路径了，而恢复路径假设订阅已经就位。

部分订阅的三种形状

三种形状需要不同的处置，混在一起两个方向都昂贵。把"被拒"当"部分"进循环重试，是拿连接配额去撞一个必然复现的失败。把"部分"当"静默"去查存储，找的是一个还没发生的空洞——因为那些标的压根没被订阅上。

第一种至少给了一帧。第二种什么都不给——这正是它能藏在三天健康日志里的原因，也是本文剩下的篇幅都在讲它的原因。

比集合，不比数量

直觉的做法是数数："发出去十二个，回来十一个确认。"数数几乎是正确的想法，而"几乎"正是它危险的地方——它说不出缺的是哪几个，遇到交易所对同一个 code 确认两次就崩溃，并且在有人想让告警闭嘴的那一刻，退化成一个阈值（"差不多都到了"）。

经得起这些压力的做法是集合比对：

import json
import os
import time

import websocket  # pip install websocket-client

WS_URL = "wss://api.tiqex.io/api/v1/market/stream"
API_KEY = os.environ["TIQEX_API_KEY"]          # 只放服务端；严禁进仓库
BATCH_SIZE = 50                                 # 单批标的数上限按套餐而定；从配置里读你那一档的值


def chunk(codes: list[str], size: int) -> list[list[str]]:
    return [codes[i:i + size] for i in range(0, len(codes), size)]


def resubscribe(ws, market: str, codes: list[str], timeout_s: float = 5.0) -> dict:
    """分批发送订阅，然后等到每一个请求的 code 都被确认。"""
    expected = set(codes)
    confirmed: set[str] = set()
    for batch in chunk(codes, BATCH_SIZE):
        ws.send(json.dumps({"action": "subscribe", "market": market,
                            "symbols": batch}))

    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline and confirmed != expected:
        frame = json.loads(ws.recv())
        kind = frame.get("type")
        if kind == "subscribed":
            confirmed |= set(frame.get("symbols", []))  # 确认帧的字段名请以你档位的流文档为准
        elif kind == "error":
            # WS_CONN_LIMIT 或权限类错误码，重试会一模一样地复现
            return {"state": "rejected", "code": frame.get("code"),
                    "msg": frame.get("msg"), "confirmed": sorted(confirmed)}
        # 订阅确认期间到达的 tick 帧是正常的；其他类型不是数据

    missing = expected - confirmed
    if missing:
        # 不要在只确认了一部分时宣布成功
        return {"state": "partial", "missing": sorted(missing)}
    return {"state": "ok", "confirmed": len(confirmed)}

这个函数有三处关键性质。

它比的是集合，不是确认数量。请求十二个只回来十个确认，是部分成功，按数量核对会把它当通过。集合版本还顺带免疫重复确认：交易所把一个 code 确认两次，集合不变。

它把 error 帧当作整个重订阅的终点，而不是在循环里继续重试。由 WS_CONN_LIMIT 或权限类错误码导致的拒绝，下一次尝试会一模一样地复现，而每一次盲目重试消耗的，都是真正要用这条连接的消费端需要的配额。

它容忍订阅确认期间到达的 tick 帧。流不会因为你正在确认就暂停；在这里假设严格的帧顺序，会制造一种只在压力下出现的间歇性故障——而压力场景是遇见它最坏的时机。

为什么缺的总在批次边界上

开头那个事故，看一眼位置就能诊断。两个失效标的正好是最后一批的最后两个 code，这个位置不是巧合。

如果标的随机失效，原因在上游——交易所、链路、市场。如果它们在订阅列表末尾连续地失效，原因就在分批循环里：最后一批发出去了但确认没回来，或者根本没发出去。连续性就是指纹。

这个观察把对账从一份报告变成一个诊断器：

import json
import urllib.request


def locate_lost_symbols(missing: list[str], codes: list[str],
                        size: int) -> list[dict]:
    """把每个缺失 code 映射到它在分批计划中的位置。"""
    plan = {code: (i // size, i % size) for i, code in enumerate(codes)}
    out = []
    for code in missing:
        if code in plan:
            out.append({"code": code, "batch": plan[code][0],
                        "position": plan[code][1]})
    contiguous = (
        len({d["batch"] for d in out}) == 1
        and sorted(d["position"] for d in out)
        == list(range(len(out)))
    )
    note = ("尾批连续缺失：故障在发送循环里"
            if contiguous else "离散缺失：排查上游或标的级问题")
    return out + [{"diagnosis": note}]


if __name__ == "__main__":
    import argparse

    p = argparse.ArgumentParser(description="端到端校验重订阅覆盖情况")
    p.add_argument("--market", default="crypto")
    p.add_argument("--symbols", nargs="+", required=True,
                   help="如 BTCUSDT ETHUSDT ...（crypto 代码，空格分隔）")
    args = p.parse_args()

    ws = websocket.create_connection(WS_URL, header=[f"X-API-Key: {API_KEY}"])
    result = resubscribe(ws, args.market, args.symbols)
    print(result)

    if result["state"] == "partial":
        # 在决定如何恢复之前，先用 REST 交叉核对这个洞
        req = urllib.request.Request(
            "https://api.tiqex.io/api/v1/market/crypto/quote"
            f"?symbols={','.join(result['missing'])}",
            headers={"X-API-Key": API_KEY})
        with urllib.request.urlopen(req, timeout=10) as resp:
            snapshot = json.load(resp)
        print("REST envelope:", snapshot.get("succ"), snapshot.get("code"))

        report = locate_lost_symbols(result["missing"], args.symbols, BATCH_SIZE)
        print(report[-1]["diagnosis"])

离散缺失和尾部连续缺失，成因不同，修法不同。上面这段代码一遍扫描就把两者分开，不用任何人去翻代码。还有一种第三类：缺失的 code 在计划里根本找不到——它从没进过 watchlist，那是配置问题，不是传输问题。

洞已经超出可取范围时

发现得早的部分订阅，一次重订阅就能修好。三天后才发现的，上面还叠了第二个问题：缺失的历史窗口可能已经超出 REST 接口能返回的范围——历史深度按套餐而定，各档位并不相同。

这改变了恢复的目标。重新订阅之后，实时流恢复了，但缺失的历史靠重试拿不回来——而重试恰恰是多数恢复循环在做的事。正确动作是把那个窗口标记为"不可恢复"，写进存储、带明确边界，让洞可审计而不是不可见。回补窗口的机制与边界，录制器一文已经讲过；这里只讲决策规则：早于回补窗口的部分订阅，是要记录的数据质量事件，不是要重试的传输错误。

这也是部分确认必须在日志里醒目的原因。一条带缺失 code 排序列表的 partial 记录，离事故报告只有一次 grep 的距离。同样的信息埋在成功路径里，就会把两个标的的故障拖成三天。

证明它：五分钟演练

一个从没抓到过丢批的重订阅检查，只是一个假设。把故障人为造出来只要几分钟，就能把它变成测试：

1. 准备一个大于你档位单批标的数上限的 watchlist，让分批循环真的跑起来。
2. 注入故障：在发送前丢掉最后一批——在发送循环的副本里删掉最后一个 chunk，或者加个开关让循环跳过它。
3. 运行 resubscribe，断言返回 partial，且缺失的正是被注入的那几个。
4. 运行 locate_lost_symbols，断言诊断结论是"尾部连续"。
5. 只对缺失的 code 重新订阅，断言确认集合与请求集合相等。

第三步抓实现缺陷：同样的输入，按数量核对的版本会返回 ok。第五步把恢复路径本身闭环——一个只补对一半的修复会留下一个更难看见的洞，因为检查已经"通过"过一次了。

演练要定期跑。恢复路径是代码，代码会腐坏：一次档位调整、一次 watchlist 加载器重写、一次平台对标的数上限的修改，每一个都能让一个通过的测试变成空洞地通过。

常见问题

重连成功但部分标的收不到 tick，是断线吗？

传输层不是。先把确认的订阅集合和请求的集合比对——部分订阅产生的正是这个症状，且任何地方都不报错。集合相等，故障才在订阅状态的下游，下一步才是逐标的的缺口检查。

集合比对放在哪里做——客户端还是监控？

两边都做，分工不同。放在客户端、内联，作为"宣布恢复完成"的条件——这是不让洞打开的那道闸。放在监控里，作为读取存储中确认集合的独立检查——因为客户端里那个检查自己也可能静默失效，而那正是最初的故障换了件外套回来。

单批标的数上限是平台的细节，该写死吗？

不要。从配置里读，给一个默认值，并把拒绝错误码当作权威——这个上限按套餐而定，它是你当前档位的属性，不是平台的属性。上限若变了，写死的代码会以令人困惑的方式失效——分出的批大小和假设的不一样——而配置化的上限加集合比对会响亮且正确地失败，并精确告诉你哪些 code 从未确认。

被拒和部分都意味着一些标的黑了，是同一起事故吗？

不是，这个区别决定值班时怎么处置。被拒是容量或权限问题：它会一模一样地复现，所以修法是配置，不是重试。部分是投递问题：只对缺失的 code 重新订阅就能修。把被拒送进部分恢复的循环，会在原故障之上再加一层配额压力。

部分订阅最多能潜伏多久？

取决于你的健康信号能容忍多久。如果健康意味着"socket 开着、进程活着"，可以无限期——那就是开头的事故。如果健康包含 subscription_confirm_ratio——每次重连后确认数除以请求数——潜伏期就是一次重连。这个比值一行代码就能算，它就是"你选择看见的故障"和"对账替你发现的故障"之间的区别。

下一步

下次重连之后，把确认的订阅集合和请求的集合比对一次。如果相等，这篇文章对你没有成本。如果不相等，你手里已经有一份缺失 code 的清单——而且它已经存在了不止你以为的这么久。

注册即可获得免费密钥，这个检查可以先在延迟数据上搭起来，再决定要不要升配，不需要信用卡。

获取免费 API 密钥，开始接入 →

本文引用的接口行为、帧类型（subscribed / tick / error）、错误码（如 WS_CONN_LIMIT）与套餐权限，来自 tiqex.io 官方文档——错误码页（/docs/errors）、认证页（/docs/auth）、开发者中心（/developers）与 WebSocket 实时行情指南（/guides/websocket-live-quotes）——以及定价页，均访问于 2026-10-20。确认帧的准确字段名与各档位标的数上限，请在生产使用前以你档位的流文档和控制台为准。字段与参数的可用性随市场与套餐而异，采用本文任何校验逻辑前，请先确认实际返回结构。文中内容仅用于说明工程实践，不构成投资建议，也不构成对未来价格的预测。