凌晨三点行情监控脚本告警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: {...}} 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标的代码intervalK 线周期gap_start缺口起始时间gap_end缺口结束时间expected_count应有 K 线条数recovered_count实际回补条数statuspending / 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的记录预期回补前置检查发现已成功回补直接跳过本次请求不重复写入
