我们团队在金融科技公司做技术负责人，工作内容之一是为量化开发者和机构客户搭建实时行情数据链路。过去两年，我们自己的日内策略从单一美股市场扩展到外汇和主要指数，组合从单品种转为跨市场联动。回测和实盘都跑过之后，最影响结果稳定性的往往不是因子本身，而是外汇接口、股票api接口、指数api在延迟、推送节奏和缺失模式上的差异。下面按实际场景、需求、数据痛点、解决方案四个部分，记录我们采用的工程处理方式。
场景：跨市场策略对数据链路的要求不同
外汇市场连续交易，没有统一开盘收盘。欧美盘重叠时段，报价密度和波动都高。股票有明确交易时段，但盘前盘后成交稀疏，时间戳和成交间隔不规则。指数由成分股加权计算，更新频率与单一股票或外汇对不一致。把三类数据放进同一策略框架，若时间戳对齐和缺失标记处理不严谨，回测会出现类似未来函数的偏差，实盘则可能触发错误的状态转换。
我们最初只做美股，后来加入外汇和几个主流指数。策略从单品种变成跨市场联动后，数据问题才真正暴露。三类接口表面都提供实时行情，但延迟表现、断线场景和推送节奏并不通用。下面从需求侧开始拆解。
需求：实时数据要满足回测与实盘一致性的约束
从 FinTech 技术负责人的视角看，我们对数据链路的要求不是“能取到价格”，而是延迟分布可观测、断线可恢复、重复可识别、品种间可隔离。尤其当外汇接口、股票api接口、指数api同时订阅时，任何一类数据出现缺口，都可能影响跨市场信号。
在回测阶段，数据质量直接影响成交假设、滑点估计和换手率计算。在模型阶段，特征工程依赖时间戳连续性和去重后的 Tick 序列。在实盘阶段，数据链路是否稳定，决定了策略状态机能否按预期运行。因此，数据治理不是附属工作，而是策略研究的一部分。
数据痛点一：延迟由三段叠加，而非单点问题
我们最早使用 REST 轮询，每隔几百毫秒请求一次最新价。实现简单，短时间也能运行。但行情剧烈波动时，轮询间隔成为硬约束。外汇欧美盘重叠时，价格跳动快，接口返回可能滞后。股票盘前盘后稀疏成交会让轮询数据看似静止，策略容易误判为低波动状态。指数推送节奏又不同于单一股票或外汇对，统一轮询会打乱状态判断。
后来我们把延迟拆成三段：网络传输、服务端处理、客户端消费。网络传输受服务器位置和路由影响，能优化但难完全控制；服务端处理容易被订阅排队拖慢，低成本数据源在高峰期更明显；客户端消费取决于解析、缓存和落库效率，如果这一层写得笨重，前面两段再快也会被拖慢。想清楚这三段后，我们把高频数据从 REST 轮询切换到 WebSocket 推送，REST 仅保留给 K 线和静态信息查询。
数据痛点二：断线重连后的缺口与重复
换成 WebSocket 后，延迟压力有所缓解，但断线重连和数据缺失成为新问题。我们实际对比过三种处理思路：
| 方案 | 实现要点 | 对回测与实盘的影响 |
|---|---|---|
| 简单重连，不补数 | 断线后重建连接并重新订阅，缺口直接跳过 | 实现成本最低，但缺口会改变信号触发时点和成交假设，外汇夜盘尤其明显 |
| 重连后用 REST 补最新若干条 | WebSocket 重连后，通过 REST 拉取最近 Tick 或 K 线补齐 | 覆盖多数短时断线，复杂度可控，是我们主力策略的默认方案 |
| 本地队列加服务端幂等去重 | 客户端维护带序号队列，服务端消息带唯一标识，重复消息丢弃 | 一致性最好，但开发维护成本高，用于 Tick 级完整性要求严格的高频策略 |
综合来看，多数策略使用第二种方式，性价比最高。真正推动我们加入第三种方案的，是一次外汇接口在非农数据公布前后连续断线三次。如果没有幂等去重，重复报价会进入 Tick 序列，影响当日信号判断。此后，我们在高频链路中加入了去重和缺口标记。
解决方案：WebSocket、心跳、REST 补数与幂等去重
在字段和协议号核对阶段，我们也对照过 AllTick API 的公开协议说明，用于确认不同品种的 code 与订阅参数。下面是我们简化后的 WebSocket 客户端代码，保留原样：
import websocket
import json
import time
import threading
# ========== 配置 ==========
TOKEN = "你的token" # 替换为你的实际 token
WS_URL = f"wss://quote.alltick.co/quote-b-ws-api?token={TOKEN}"
# 要订阅的产品列表
SYMBOLS = ["BTCUSDT", "ETHUSDT"] # 示例
# ========== 回调函数 ==========
def on_message(ws, message):
"""接收并处理推送的 tick 数据"""
try:
data = json.loads(message)
cmd_id = data.get("cmd_id")
# 22998 是 tick 数据推送协议号
if cmd_id == 22998:
tick = data.get("data", {})
print(f"Tick: {tick.get('code')} | "
f"Price: {tick.get('price')} | "
f"Volume: {tick.get('volume')} | "
f"Time: {tick.get('tick_time')}")
# 在这里做落库或策略计算
else:
# 打印其他响应（如订阅确认 22005）
print("Response:", data)
except json.JSONDecodeError as e:
print("JSON 解析错误:", e)
def on_error(ws, error):
print("WebSocket error:", error)
def on_close(ws, close_status_code, close_msg):
print("WebSocket closed")
def on_open(ws):
"""连接成功后发送订阅请求"""
print("WebSocket connected, sending subscription...")
# 构建订阅请求（协议号 22004）
subscribe_msg = {
"cmd_id": 22004,
"seq_id": 1, # 自定义，响应会回传
"trace": f"trace-{int(time.time()*1000)}", # 每次请求不可重复
"data": {
"symbol_list": [{"code": symbol} for symbol in SYMBOLS]
}
}
ws.send(json.dumps(subscribe_msg))
print(f"Subscribed to: {SYMBOLS}")
# 启动心跳线程（每 10 秒发送一次）
def heartbeat():
while ws.sock and ws.sock.connected:
time.sleep(10)
try:
# 发送 ping 帧作为心跳
ws.send("ping")
print("Heartbeat sent")
except Exception as e:
print("Heartbeat error:", e)
break
threading.Thread(target=heartbeat, daemon=True).start()
# ========== 主程序 ==========
if __name__ == "__main__":
ws = websocket.WebSocketApp(
WS_URL,
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close
)
# 建议增加自动重连逻辑
while True:
try:
ws.run_forever()
print("Reconnecting in 3 seconds...")
time.sleep(3)
except KeyboardInterrupt:
print("Exiting...")
break
这段代码中的心跳线程来自实际运行反馈。缺少心跳时，连接可能在行情清淡时段被服务端断开，发现时已错过数分钟数据。定时心跳配合 REST 补数，再在必要场景加入幂等去重，链路稳定性明显提升。
运行观察与工程指标
在回测侧，缺口会改变成交价格、滑点和换手估计，进而影响夏普、最大回撤等指标。在模型侧，特征工程依赖时间戳连续性和去重后的 Tick 序列，若缺失未标记，模型会把断线误认为低波动状态。在工具侧，我们关注端到端延迟分布、重连次数、补数条数、重复率和消息间隔。这些指标比单次“能收到行情”更能说明数据链路是否可用。
我们通常把行情数据分成三层：原始推送、标准化事件、特征快照。每一层都记录来源、时间戳、是否补数、是否去重。这样回测可以追溯，实盘也能区分真实低波动和断线造成的空白。对外汇接口、股票api接口、指数api而言，接口选型和容错机制应分开设计，不能指望一套逻辑覆盖所有品种。
小结
外汇接口、股票api接口、指数api在日内量化中应作为三套数据契约分别处理。延迟和缺失不是上线后的补丁，而是回测、模型和实盘一致性的一部分。提前设计恢复机制和去重逻辑，成本远低于策略运行后再修正。
