重连成功，图表却骗了你

凌晨三点，行情监控脚本告警：WebSocket 断开。五秒后重连成功，日志显示连接恢复，心跳正常。

你看了一眼 K 线图，没有跳空，没有断崖，价格曲线平滑。于是关掉告警继续睡觉。

两周后跑策略回测，发现那天的分钟线有几根不对。排查了很久才找到根因：凌晨断流那段时间缺了几分钟数据，而重连后 WebSocket 只推送了当前快照，没有把断流窗口的 K 线补回来。你看到的平滑曲线，其实是前端把缺失段直接连起来了。

重连成功 ≠ 数据连续。连接 watch dog 只能证明传输通道恢复了，不能证明断流期间的 K 线已经完整补回。补偿这件事必须由业务层主动完成：检测缺口、回补数据、校验完整性、留痕记录，缺一不可。

二、为什么 watch dog 发现断流，却不能证明数据完整

一个典型的 WebSocket watch dog 只做三件事：心跳超时检测、连接状态监控、重连触发。这三件事全部围绕“连接”展开。

它不回答以下问题：断流期间到底缺了几根 K 线？重连后收到的第一条数据，时间戳和断流前最后一条是否连续？回补的数据是不是完整覆盖了缺失窗口，还是只补了一部分？

连接是传输层的概念，K 线是业务层的概念。传输层的恢复，不能自动推导出业务数据的完整。这个推导必须由业务层主动完成。

常见的三种缺数据情况：

• 缺段：断流窗口内的 K 线全部缺失，重连后直接跳到当前时间，旧数据永久丢失。
• 缺条：窗口内部分 K 线缺失，比如 10 分钟内缺了 3 根，整体数据看起来连续，细查才发现空洞。
• 重复覆盖：重连后推送了部分重叠数据，但不是完整窗口，导致你误以为补全了，其实还有缺口。

这三种情况，如果不主动做缺口检测，都会被“连接正常”的状态掩盖。

三、缺口检测：拿“应该有”和“实际有”做集合差

缺口检测的核心逻辑很清晰：根据 K 线周期和你自己维护的交易日历，生成断流窗口内“应该有”的时间序列，然后与本地实际已落地的时间序列做集合差。

from typing import List from datetime import datetime def detect_gaps(expected_times: List[datetime], actual_times: List[datetime]) -> dict: """ 检测缺口：expected_times 是断流窗口内理论上应该存在的 bar 时间点， actual_times 是本地已确认写入的 bar 时间点（来自数据库记录或本地聚合后的 bar_time）。 返回缺失、重叠和意外时间点。 """ expected_set = set(expected_times) actual_set = set(actual_times) missing = sorted(expected_set - actual_set) overlap = sorted(expected_set & actual_set) unexpected = sorted(actual_set - expected_set) return { "gap_count": len(missing), "missing": missing, "overlap_count": len(overlap), "unexpected_count": len(unexpected), "is_complete": len(missing) == 0 and len(unexpected) == 0 }

expected_times 怎么来？根据请求的 interval 和断流起止时间推算。例如 1 分钟 K 线，断流 10 分钟，就应该有 10 个 expected_times。推算时必须参照交易日历，排除非交易时段、午休和节假日——否则会把不应该有 K 线的时段也计入“缺失”，导致 expected_count 虚高，回补窗口永远有一批“无法补上”的缺口。交易日历的维护是整个方案中最容易被低估成本的部分。

actual_times 怎么来？actual_times 不能直接等同于 WebSocket 收到的 bar 的时间戳，因为 WebSocket 推送的只是“那一刻看到的行情快照”，不是“已经持久化到本地的业务时间轴”。actual_times 应取自本地数据库已确认写入的 bar 的时间列表，或本地聚合后产生的有效业务时间轴。这样，实际存在的数据和应该存在的数据之间的差值，才是真正需要回补的缺口。

四、为什么必须用 REST 回补，而不是让 WebSocket 重新推

这是协议设计层面的取舍。WebSocket 是面向实时推送的轻量通道，服务端推送的是“当前及未来”的增量数据。如果要求它在重连后补推所有历史快照，服务端就需要为每一个连接维护快照缓存和增量序列——这与实时推送的定位相矛盾。因此，历史回补必须走另一条通道：REST K 线接口，按需拉取指定时间窗口的完整历史数组。

MCP 工具只适合 AI 按需查询，不应放进自动化监控的断流回补链路。REST K 线接口是这个场景下最合适的选择：一次请求拿一个完整时间窗口的 K 线，返回结构确定，不依赖连接状态。但要注意，REST K 线是回补查询通道，不承诺完整恢复所有断流数据。如果断流窗口超出了数据源的历史覆盖范围，或者该时段数据因其他原因不可用，回补可能只拿到部分数据，甚至完全拿不到。

五、回补实现：分段拉取、幂等写入、状态判定

检测到缺口后，下一步是用 REST K 线接口把缺失窗口的数据拉回来。拉回的数据不能直接入库，需要经过完整性校验，并且保证同一缺口只回补一次。

def fetch_kline_gap(symbol: str, interval: str, start: datetime, end: datetime) -> dict: """ 用 REST K 线接口回补缺失窗口。 实际调用时替换为具体数据源的端点、鉴权方式和参数。 返回结构包含 recovered_times 列表，用于后续覆盖检查。 """ # 教学骨架，具体实现以数据源官方文档为准 # 示例：resp = requests.get(kline_url, headers={"X-API-Key": key}, # params={"symbol": symbol, "interval": interval, # "start": start, "end": end}) # 返回：{"klines": [...], "raw_snapshot": {...}} pass

5.1 幂等回补：同一窗口不重复写入

每次发起 REST 回补前，先查询gap_report表，用symbol + interval + gap_start + gap_end四个字段组成唯一键，如果已有成功（status=full）的回补记录，直接跳过本次请求。这避免了因重试导致的重复写入和 API 配额浪费。

5.2 状态判定：full 必须检查 recovered_times 覆盖 missing_times

状态full不能只看recovered_count >= expected_count。必须将回补返回的 K 线时间列表recovered_times与缺失列表missing_times做集合差，确认missing_times是recovered_times的子集，并且所有期望的时间点都已补回。只有时间点完全覆盖，且数据经过解析校验无误，才能标记为full。

def determine_status(missing_times: List[datetime], recovered_klines: List[dict], request_error: str = None) -> str: """根据缺失时间点与回补时间点的覆盖关系，判定回补状态。""" if request_error: return "failed" if not recovered_klines: return "empty" recovered_times = {bar["time"] for bar in recovered_klines} missing_set = set(missing_times) if missing_set.issubset(recovered_times): return "full" elif recovered_times & missing_set: return "partial" else: return "empty"

5.3 恢复作业状态：recovery_job

每次从检测缺口到回补结束，是一个 recovery_job。它记录整个过程的元信息，便于监控和追溯。

六、推荐表结构

以下表结构用于支持断流回补的完整生命周期。

6.1 gap_report 表

CREATE TABLE gap_report ( id BIGSERIAL PRIMARY KEY, symbol VARCHAR(20) NOT NULL, interval VARCHAR(5) NOT NULL, gap_start TIMESTAMP NOT NULL, gap_end TIMESTAMP NOT NULL, expected_count INT NOT NULL, recovered_count INT NOT NULL DEFAULT 0, raw_snapshot_id VARCHAR(32), status VARCHAR(20) NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','fetching','full','partial','empty','failed','unrecoverable')), note TEXT, reported_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), UNIQUE (symbol, interval, gap_start, gap_end) );

6.2 raw_snapshot 存储

原始响应不拆散入库，以 JSON 文件形式存储在对象存储或本地文件系统。raw_snapshot_id作为外键关联，文件路径如snapshots/{YYYY-MM}/{job_id}.json。

6.3 recovery_job 表

CREATE TABLE recovery_job ( job_id UUID PRIMARY KEY, symbol VARCHAR(20) NOT NULL, interval VARCHAR(5) NOT NULL, gap_start TIMESTAMP NOT NULL, gap_end TIMESTAMP NOT NULL, expected_count INT NOT NULL, recovered_count INT NOT NULL DEFAULT 0, status VARCHAR(20) NOT NULL DEFAULT 'pending', raw_snapshot_id VARCHAR(32), created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), completed_at TIMESTAMP WITH TIME ZONE );

七、失败分支处理

八、单元测试思路

编写单元测试时，不依赖真实 WebSocket 或 REST 调用，用模拟的 expected_times 和 recovered_klines 覆盖以下五种场景。

测试用例 1：正常无缺口

• 输入：missing_times=[]，recovered_klines有对应的 bar
• 预期：is_complete=True，状态full

测试用例 2：缺 3 根，回补恰好覆盖

• 输入：missing_times包含 3 个时间点，recovered_klines精确包含这 3 个时间点
• 预期：状态full，recovered_count == expected_count

测试用例 3：recovered_count 够，但 recovered_times 不覆盖 missing_times

• 输入：missing_times = [T1, T2, T3]，recovered_klines有 3 根但时间点是[T1, T2, T4]
• 预期：状态partial，因为T3仍缺失，不能判定为full

测试用例 4：REST 返回空

• 输入：recovered_klines=[]，request_error=None
• 预期：状态empty，并进一步检查是否应升级为unrecoverable

测试用例 5：重复回补幂等

• 输入：同一symbol + interval + gap_start + gap_end已有status=full的记录
• 预期：回补前置检查发现已成功回补，直接跳过本次请求，不重复写入