实时行情接口如何实现数据订阅 - 文章 - 开发者社区

一、行情订阅模式介绍

主流行情 API 分为轮询拉取与长连接订阅两类模式。

轮询拉取：客户端主动定时请求接口获取快照，延迟高、请求量大，不适合高频 Tick、秒级 K 线实时跟踪；
长连接订阅：客户端与服务端建立持久连接，服务端行情变动后主动推送增量数据，低延迟、资源占用更低，是量化、自动化交易系统的标准选型。

行情订阅细分为全量订阅、增量订阅、单标的定向订阅三类，生产环境普遍采用定向增量订阅，仅接收关注品种数据，减少无效流量。

二、实时数据连接机制

行业标准实时推送载体为 WebSocket，基于 HTTP 握手升级长连接，核心运行流程：

客户端发起 HTTP 握手请求，携带身份鉴权参数；
服务端校验权限，完成协议升级，建立双向持久通道；
客户端发送订阅指令，声明需要监听的交易品种；
服务端缓存订阅关系，标的行情更新时主动推送结构化报文；
链路断开后客户端执行自动重连，重新发送订阅指令恢复数据流。

配套保障机制：心跳保活、断线重连、报文序号校验、重复数据过滤，用于解决网络抖动导致的断流、数据重复问题。

三、实战接入：建立行情订阅服务（核心章节）

3.1 接入前置规范

鉴权准备：申请接口访问密钥，所有长连接握手需携带身份凭证；
订阅指令规范：采用 JSON 格式上报订阅列表，支持批量添加 / 取消标的；
数据报文规范：推送消息统一包含标的代码、时间戳、最新价、买卖盘、成交量等标准化字段；
运维配套：实现心跳上报、异常日志落盘、断连自动重试、订阅状态本地缓存。

3.2 完整业务流程设计

初始化连接客户端，加载鉴权配置、订阅品种清单、重连重试次数、心跳间隔；
发起 WebSocket 握手，等待服务端连接成功回执；
连接就绪后，同步发送批量订阅指令，注册需要监听的全部品种；
持续监听服务端下行推送报文，完成数据解析、格式化、入库 / 策略计算；
定时向服务端发送心跳包，维持链路活性；
捕获连接异常、超时、服务端主动断开事件，执行指数退避重连，重连成功后重新订阅标的；
程序退出时主动发送取消订阅指令，正常关闭长连接释放资源。

3.3 生产环境关键优化点

数据处理解耦：网络接收线程与业务计算线程分离，避免解析阻塞导致消息堆积；
消息去重：利用报文中唯一序列 ID 过滤重复推送数据；
时区统一转换：将接口返回时间戳统一转为 UTC 毫秒标准，规避回测、统计日期边界错乱；
流量管控：仅订阅业务所需标的，不开启全市场广播订阅，降低带宽与解析开销；
压测验证：上线前通过 API 压测工具模拟高并发 Tick 推送，验证客户端消息吞吐能力。

3.4 服务可用性补充说明

量化系统对行情连续性要求高，完整订阅服务必须包含降级逻辑：长连接多次重连失败时，自动切换短时轮询兜底，保障基础行情数据不中断。

四、Python 实战：WebSocket 订阅数据

基于标准 websocket-client 库实现基础订阅客户端，包含鉴权、订阅、消息解析、断线重连基础骨架。

import json
import time
import websocket

# 全局配置
API_KEY = "your_access_key"
SUBSCRIBE_SYMBOLS = ["XAUUSD", "EURUSD"]
WS_URL = "wss://api.alltick.co/ws"

def send_subscribe(ws):
    # 构造批量订阅指令
    sub_msg = {
        "action": "subscribe",
        "symbols": SUBSCRIBE_SYMBOLS,
        "token": API_KEY
    }
    ws.send(json.dumps(sub_msg))
    print("已发送订阅请求")

def on_open(ws):
    send_subscribe(ws)

def on_message(ws, message):
    data = json.loads(message)
    # 解析实时行情数据
    symbol = data.get("symbol")
    price = data.get("price")
    ts = data.get("timestamp")
    print(f"行情推送 | {symbol} 价格:{price} 时间戳:{ts}")

def on_error(ws, error):
    print("连接异常：", error)

def on_close(ws, close_code, close_msg):
    print("连接断开，3秒后自动重连")
    time.sleep(3)
    start_client()

def start_client():
    ws_app = websocket.WebSocketApp(
        WS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    # 循环保活
    ws_app.run_forever()

if __name__ == "__main__":
    start_client()

五、Java 实战：实现行情监听模块

使用 Java-WebSocket 开源组件构建异步监听服务，封装独立行情监听类，适配后端量化服务工程化开发。

import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import java.net.URI;
import java.util.List;

public class MarketSubscribeClient extends WebSocketClient {
    private final String token;
    private final List<String> symbolList;

    public MarketSubscribeClient(URI serverUri, String token, List<String> symbolList) {
        super(serverUri);
        this.token = token;
        this.symbolList = symbolList;
    }

    // 连接建立完成，发送订阅指令
    @Override
    public void onOpen(ServerHandshake handshakedata) {
        String subJson = String.format("{\"action\":\"subscribe\",\"token\":\"%s\",\"symbols\":%s}",
                token, symbolList.toString());
        send(subJson);
        System.out.println("订阅指令已下发");
    }

    // 接收实时推送行情
    @Override
    public void onMessage(String message) {
        // 自行引入JSON工具解析报文，处理行情数据
        System.out.println("收到行情数据：" + message);
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("连接断开，准备重连");
        new Thread(() -> {
            try {
                Thread.sleep(3000);
                reconnect();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }

    @Override
    public void onError(Exception ex) {
        ex.printStackTrace();
    }

调用示例：

import java.net.URI;
import java.util.List;

public class MarketApplication {
    public static void main(String[] args) throws Exception {
        URI uri = new URI("wss://api.alltick.co/ws");
        List<String> symbols = List.of("XAUUSD", "GBPUSD");
        MarketSubscribeClient client = new MarketSubscribeClient(uri, "your_token", symbols);
        client.connect();
    }
}

六、总结

实时行情订阅优先选用 WebSocket 长连接模式，相比轮询具备更低延迟与更高吞吐，是量化系统标准方案；完整订阅服务需要覆盖鉴权、订阅下发、消息解析、心跳保活、断线重连、数据清洗全链路逻辑；
Python、Java 可基于成熟开源组件快速搭建监听客户端，生产环境需做好线程解耦、消息去重、流量管控等优化，保证数据流稳定；
开发测试阶段可选用 AllTick API，平台提供 7 天全功能免费试用，免费周期开放全市场标的订阅权限，同时支持 WebSocket 并发压测，便于验证客户端高负载下的数据接收稳定性，适合策略回测、自动化交易系统前期开发验证。

参考文档：https://apis.alltick.co/
GitHub：https://github.com/alltick/alltick-realtime-forex-crypto-stock-tick-finance-websocket-api