一、重连成功图表却骗了你凌晨三点行情监控脚本告警WebSocket 断开。五秒后重连成功日志显示连接恢复心跳正常。你看了一眼 K 线图没有跳空没有断崖价格曲线平滑。于是关掉告警继续睡觉。两周后跑策略回测发现那天的分钟线有几根不对。排查了很久才找到根因凌晨断流那段时间缺了几分钟数据而重连后 WebSocket 只推送了当前快照没有把断流窗口的 K 线补回来。你看到的平滑曲线其实是前端把缺失段直接连起来了。重连成功 ≠ 数据连续。 连接 watch dog 只能证明传输通道恢复了不能证明断流期间的 K 线已经完整补回。补偿这件事必须由业务层主动完成检测缺口、回补数据、校验完整性、留痕记录缺一不可。二、为什么 watch dog 发现断流却不能证明数据完整一个典型的 WebSocket watch dog 只做三件事心跳超时检测、连接状态监控、重连触发。这三件事全部围绕“连接”展开。它不回答以下问题断流期间到底缺了几根 K 线重连后收到的第一条数据时间戳和断流前最后一条是否连续回补的数据是不是完整覆盖了缺失窗口还是只补了一部分连接是传输层的概念K 线是业务层的概念。 传输层的恢复不能自动推导出业务数据的完整。这个推导必须由业务层主动完成。常见的三种缺数据情况缺段断流窗口内的 K 线全部缺失重连后直接跳到当前时间旧数据永久丢失。缺条窗口内部分 K 线缺失比如 10 分钟内缺了 3 根整体数据看起来连续细查才发现空洞。重复覆盖重连后推送了部分重叠数据但不是完整窗口导致你误以为补全了其实还有缺口。这三种情况如果不主动做缺口检测都会被“连接正常”的状态掩盖。三、缺口检测拿“应该有”和“实际有”做集合差缺口检测的核心逻辑很清晰根据 K 线周期和你自己维护的交易日历生成断流窗口内“应该有”的时间序列然后与本地实际已落地的时间序列做集合差。from typing import Listfrom datetime import datetimedef 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”: {…}}pass5.1 幂等回补同一窗口不重复写入每次发起 REST 回补前先查询 gap_report 表用 symbol interval gap_start gap_end 四个字段组成唯一键如果已有成功statusfull的回补记录直接跳过本次请求。这避免了因重试导致的重复写入和 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 empty5.3 恢复作业状态recovery_job每次从检测缺口到回补结束是一个 recovery_job。它记录整个过程的元信息便于监控和追溯。字段 说明job_id 唯一标识UUIDsymbol 标的代码interval K 线周期gap_start 缺口起始时间gap_end 缺口结束时间expected_count 应有 K 线条数recovered_count 实际回补条数status pending / fetching / full / partial / empty / failed / unrecoverableraw_snapshot_id 原始响应哈希created_at 作业创建时间completed_at 完成时间六、推荐表结构以下表结构用于支持断流回补的完整生命周期。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);七、失败分支处理失败场景 处理方式① REST 回补请求返回空 data 先标 empty确认参数和权限无误后若该时段超出历史覆盖范围则升级为 unrecoverable② 回补条数不足时间点未覆盖全部缺失 statuspartial明确标注哪些缺失时间点仍未补回③ 交易日历误判导致 expected 不准确 note 中标注“交易日历可能不准确”不强制补数④ interval 不一致回补用了和订阅不同的周期 阻断修正 interval 参数后重新拉取⑤ 同一段缺口被重复回补写入 利用 gap_report 的唯一约束拒绝重复幂等跳过⑥ raw_snapshot 未保存 statusfailednote 中标注“缺失原始快照”该条报告标记为不可复查八、单元测试思路编写单元测试时不依赖真实 WebSocket 或 REST 调用用模拟的 expected_times 和 recovered_klines 覆盖以下五种场景。测试用例 1正常无缺口输入missing_times[]recovered_klines 有对应的 bar预期is_completeTrue状态 full测试用例 2缺 3 根回补恰好覆盖输入missing_times 包含 3 个时间点recovered_klines 精确包含这 3 个时间点预期状态 fullrecovered_count expected_count测试用例 3recovered_count 够但 recovered_times 不覆盖 missing_times输入missing_times [T1, T2, T3]recovered_klines 有 3 根但时间点是 [T1, T2, T4]预期状态 partial因为 T3 仍缺失不能判定为 full测试用例 4REST 返回空输入recovered_klines[]request_errorNone预期状态 empty并进一步检查是否应升级为 unrecoverable测试用例 5重复回补幂等输入同一 symbol interval gap_start gap_end 已有 statusfull 的记录预期回补前置检查发现已成功回补直接跳过本次请求不重复写入九、整体流程整个断流补偿链路如下WebSocket 断流 → watch dog 触发重连 → 重连成功后立即执行 detect_gapsactual_times 取自本地数据库已落地的 bar 时间轴→ 确认缺失窗口 → 分段调用 REST K 线接口回补 → 每条回补数据经过 Decimal 及 OHLC 校验 → 判定覆盖状态full/partial/empty → 写入 gap_report 和 recovery_job 记录 → raw_snapshot 归档保存。全部完成后检查 status 分布。若存在 unrecoverable 或 partial通知下游策略该时段数据不完整相关回测结果需要复核。只有全部窗口标记为 full这次断流才算真正“补偿完毕”。十、TickDB 的工程边界上面这套断流检测、REST 回补和 gap_report 留痕流程是一套通用的工程方法不绑定任何特定数据源。如果你用 TickDB 做行情接入它在断流回补场景中承担两个明确的角色WebSocket 负责持续推送实时行情通过 WebSocket 通道到达是数据 watch dog 的输入。TickDB WebSocket 使用 api_key 作为 query 参数鉴权。REST K 线负责历史回补断流窗口的缺失 K 线通过 REST 接口拉取。TickDB REST 使用 X-API-Key Header 鉴权。两个通道各司其职不能混用。WebSocket 不是历史数据源REST K 线不是实时推送通道。MCP 是给 AI 工具按需查询用的不适合放进自动化监控的断流回补链路。所有端点、字段路径、时间戳语义以 TickDB 官方文档和你自己的实测为准。十一、断流回补的最小检查清单重连后有没有根据订阅的 interval 和交易日历生成 expected_times有没有用本地已写入的实际 bar 时间轴作为 actual_times做集合差确认缺口缺口回补是不是用 REST K 线接口按段拉取而不是靠 WebSocket 重新推送回补前有没有检查 gap_report 表防止重复写入同一缺口幂等full 状态的判定是否基于 recovered_times 完全覆盖 missing_times而不仅仅是数量partial 的记录有没有明确标注哪些时间点仍未补回empty 的记录有没有确认是否应转为 unrecoverable原始响应是否保存了 raw_snapshot供事后复查