做港股量化策略时，数据质量决定策略回测和实盘的一致性。不知道大家有没有遇到过这样的情况：策略在回测时表现不错，一到实盘就各种偏差，查了半天发现是实时行情数据有缺口。WebSocket 连接状态显示正常，消息也在不断接收，但回看 tick 数据时，某些时间段就是空的。这种问题在通过港股股票数据接口获取实时行情时尤其容易发生。一次轻微网络抖动可能不会触发断开重连，但中间一段推送就这么丢了。如果不做完整性校验，后面合成 K 线、计算技术指标都会埋下隐患。
数据断层对量化策略的影响
实时行情数据依赖长连接按时间顺序推送。理想情况下，每条 tick 都有时间戳，策略可以根据时间顺序生成分钟线、成交分布、量价因子等。但网络传输并不稳定。短时间延迟、连接抖动、本地处理速度跟不上，都可能在“连接正常”的情况下造成数据缺失。比如某只港股在交易时段内出现：
| 时间 | 数据状态 |
|---|---|
| 10:00:01 | 正常接收 |
| 10:00:02 | 正常接收 |
| 10:00:03-10:00:15 | 没有数据 |
| 10:00:16 | 恢复接收 |
系统看到只是“数据暂停了十几秒”，但市场可能已经产生了大量成交。如果策略此时依赖实时 tick 计算信号，就可能因为数据缺失而错过触发点或错误触发。
用时间戳检测缺失区间
我的做法是保留每条 tick 的原始时间戳，检查相邻数据之间的间隔。如果间隔明显超过正常范围，就记录为疑似缺失区间。一个基础的时间戳检测逻辑如下：
last_time = None
def check_data(timestamp):
global last_time
if last_time:
gap = timestamp - last_time
if gap > 5:
print("检测到数据间隔:", gap)
last_time = timestamp
实际使用时，判断标准要结合股票活跃度调整。比如恒指成分股可能每秒都有报价，而一些成交清淡的细价股可能十几秒才有一个 tick。如果用同一套阈值，会误报或漏报。
序号检测与心跳兜底
除了时间戳，消息序号也是有效手段。如果行情推送中包含递增编号，可以直接观察编号是否连续。例如收到：
- 20001
- 20002
- 20003
- 20007
中间缺失 20004 到 20006，可以快速定位。如果接口没有提供序号，也可以增加心跳检测：程序定时检查最新行情时间，长时间没有更新就记录连接状态并触发重新订阅。这两种方法可以组合使用，提高检测准确度。
在实时行情流程中嵌入校验
实际量化系统中，我会把行情接收和数据校验拆成两个模块。行情模块负责接收，校验模块负责判断异常。以 AllTick 的股票数据接口为例，其 WebSocket 行情订阅返回 tick 数据后，可以很方便地加入时间间隔判断：
import websocket
import json
last_timestamp = None
def on_message(ws, message):
global last_timestamp
data = json.loads(message)
timestamp = int(data["timestamp"])
if last_timestamp:
gap = timestamp - last_timestamp
if gap > 5000:
print("可能存在缺失:", gap)
last_timestamp = timestamp
print(
data["symbol"],
data["price"]
)
ws = websocket.WebSocketApp(
"wss://api.alltick.co/stock/websocket",
on_message=on_message
)
ws.run_forever()
发现异常后，可以把缺失时间段保存下来，后续通过历史接口补充数据。这样行情记录能保持完整，策略运行也能更稳定。
数据恢复与防重写入
发现缺失区间只是数据质量管理的一部分，恢复流程同样关键。当系统确认某段时间数据缺少后，需要重新请求对应时间范围的数据，并检查补充的数据是否和已有记录保持连续。同时要注意避免重复写入，否则补回的数据可能影响成交量统计和 K 线计算。具体步骤：
- 记录缺口起止时间。
- 请求缺口区间的历史 tick 数据。
- 校验补充数据的时间戳是否连续。
- 写入前查重，防止重复记录。
量化交易中数据完整性是底线
使用港股股票数据接口搭建实时系统时，数据稳定性并不只是看接口有没有返回数据。真正可靠的行情系统，需要关注每一次推送是否连续，时间是否准确。数据完整性校验应该成为行情处理流程中的基础环节。提前记录异常区间，比等到策略实盘出现偏差后再排查更加有效，也能让后续的数据分析和策略测试更加稳定。
