外汇行情API，实时行情接口怎么选？一次实盘滑点后的数据链路复盘 - 文章 - 开发者社区

我在基金公司做研究支持，平时会帮交易团队验证策略、搭数据管道。去年我参与过一个跑欧美盘的日内策略，回测阶段曲线很顺，切到实盘后却出现了成交价和信号价偏离的问题。最初我怀疑参数、撮合、滑点模型，后来把日志按时间戳摊开看，才发现根因在行情接入层：当时用的外汇行情API是一秒一次报价，信号触发和真实可交易价格之间隔了明显时间差。

这件事之后，我对“实时行情接口”的理解变了。它不只是取数工具，而是策略链路里的时间基准。下面按场景、需求、数据痛点、解决方案来拆。

场景：研究员的策略验证与交易团队实盘

研究岗和纯开发岗不同，我既要看模型，也要看数据在实盘中的表现。基金公司开发部门关注的是：策略能否复现、回测与实盘偏差是否可解释、数据延迟是否能被度量。专业交易者更直接：信号到了，能不能在预期价格附近成交。

所以我现在评估外汇行情API，不会先看宣传页，而是先问：这条报价从交易所/流动性源到策略内存，中间经过哪些环节？每个环节能不能打点？

需求：先定义“实时”的边界

很多人一上来比较品种和价格。我更建议先回到策略周期。小时级趋势跟踪，对一两秒延迟不敏感；事件驱动、剥头皮、短周期突破，几百毫秒就可能改变结果。先定可接受延迟上限，再选实时行情接口，效率更高。

第二是数据形态。分钟K线适合画图、算指标，但它是加工后的数据。等K线收完，信号可能已经落后。对入场敏感的策略，我会直接消费tick，再在本地聚合成需要的周期。

数据痛点：延迟不只在接口，也在本机

一条报价从产生到进入策略，我通常拆成四段：数据源到服务商、服务商到我的机器、公网传输、本机处理。前两段不可控，后两段可控。

我踩过的坑是：在WebSocket回调里直接算指标、写数据库。行情密集时回调被阻塞，后面的tick排队，表面像接口慢，实际是本机处理拖垮了链路。后来改成回调只入队，计算和存储交给独立任务，延迟明显下降。

另一个痛点是时间戳。没有时间戳的接口很难做延迟度量；有毫秒时间戳，也要先用NTP校准本机时钟，否则算出来的数字没有意义。

解决方案：WebSocket + tick + 轻量回调

要拿实时外汇报价，REST适合补历史、查K线，实时推送应走WebSocket。下面是我常用的最小可用样例。我这次用 AllTick API 的外汇 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 = ["EURUSD", "USDJPY"] 

# ========== 回调函数 ==========
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

工程化补充：可观测性与重连策略

心跳建议按文档频率发送，连接静默断开比延迟更危险。订阅请求往往会覆盖旧列表，追加品种时要重发完整列表。延迟指标不要只看均值，我会跑满交易日，统计中位数和p95。中位数低但尖峰高，优先查网络抖动和本机阻塞；整体高，先查机器位置和时钟。

重连后策略应暂停开新仓，直到报价恢复新鲜。可以用“最近一次tick时间”做阈值判断，超过阈值就停止交易。这样比盲目重连更稳。

上线前检查

外汇周末休市无推送是正常现象；时区统一UTC；成交报价和盘口数据不要混用。官网也会提示行情仅供参考，实盘前先用小仓位对照真实成交。外汇行情API没有统一答案，关键是延迟上限、tick接入、WebSocket链路和亲手测量。

picture.image

API Docs：https://alltick.co/apis/en

GitHub：https://github.com/alltick/alltick-realtime-forex-crypto-stock-tick-finance-websocket-api