如何通过 WebSocket 接入期货实时行情接口

实时行情数据是量化交易系统和行情展示应用的核心。相比传统的 HTTP 请求，WebSocket 能在客户端与服务器之间建立持续的双向连接，支持毫秒级数据推送，非常适合用于期货、股票、加密货币等高频场景。

本文将演示如何通过 Python 建立 WebSocket 连接、订阅期货实时行情、并实现自动重连与心跳机制。

一、为什么要使用 WebSocket

在金融行情数据中，HTTP 请求（REST API）通常用于：

• 查询历史数据；

• 拉取一次性快照。

而 WebSocket 接口 更适合：

• 实时订阅行情变化；

• 获取逐笔成交（Tick 数据）；

• 接收盘口深度（Depth）；

• 实时 K 线（Kline）推送。

例如，当期货价格每秒更新上百次时，HTTP 无法高效应对，而 WebSocket 可以持续不断地接收推送。

二、接口说明

期货业务连接地址格式如下：

wss://data.infoway.io/ws?business=common&apikey=YourApiKey

其中：

• business=common 这个地址可同时订阅外汇和期货的数据，股票有另外的地址

• apikey 为你的开发者密钥；

• 返回的数据为实时推送的 JSON 对象。

三、行情订阅代码结构

下面是完整的 Python 实现，支持自动重连与心跳机制：

import json
import time
import schedule
import threading
import websocket
from loguru import loggerclass WebsocketExample:def __init__(self):self.session = Noneself.ws_url = "wss://data.infoway.io/ws?business=common&apikey=yourApikey"self.reconnecting = Falseself.is_ws_connected = False  # 添加连接状态标志def connect_all(self):"""建立WebSocket连接并启动自动重连机制""""""申请API KEY: www.infoway.io"""try:self.connect(self.ws_url)self.start_reconnection(self.ws_url)except Exception as e:logger.error(f"Failed to connect to {self.ws_url}: {str(e)}")def start_reconnection(self, url):"""启动定时重连检查"""def check_connection():if not self.is_connected():logger.debug("Reconnection attempt...")self.connect(url)# 使用线程定期检查连接状态schedule.every(10).seconds.do(check_connection)def run_scheduler():while True:schedule.run_pending()time.sleep(1)threading.Thread(target=run_scheduler, daemon=True).start()def is_connected(self):"""检查WebSocket连接状态"""return self.session and self.is_ws_connecteddef connect(self, url):"""建立WebSocket连接"""try:if self.is_connected():self.session.close()self.session = websocket.WebSocketApp(url,on_open=self.on_open,on_message=self.on_message,on_error=self.on_error,on_close=self.on_close)# 启动WebSocket连接（非阻塞模式）threading.Thread(target=self.session.run_forever, daemon=True).start()except Exception as e:logger.error(f"Failed to connect to the server: {str(e)}")def on_open(self, ws):"""WebSocket连接建立成功后的回调"""logger.info(f"Connection opened")self.is_ws_connected = True  # 设置连接状态为Truetry:# 发送实时成交明细订阅请求trade_send_obj = {"code": 10000,"trace": "01213e9d-90a0-426e-a380-ebed633cba7a","data": {"codes": "USOIL"}}self.send_message(trade_send_obj)# 不同请求之间间隔一段时间time.sleep(5)# 发送实时盘口数据订阅请求depth_send_obj = {"code": 10003,"trace": "01213e9d-90a0-426e-a380-ebed633cba7a","data": {"codes": "USOIL"}}self.send_message(depth_send_obj)# 不同请求之间间隔一段时间time.sleep(5)# 发送实时K线数据订阅请求kline_data = {"arr": [{"type": 1,"codes": "USOIL"}]}kline_send_obj = {"code": 10006,"trace": "01213e9d-90a0-426e-a380-ebed633cba7a","data": kline_data}self.send_message(kline_send_obj)# 启动定时心跳任务schedule.every(30).seconds.do(self.ping)except Exception as e:logger.error(f"Error sending initial messages: {str(e)}")def on_message(self, ws, message):"""接收消息的回调"""try:logger.info(f"Message received: {message}")except Exception as e:logger.error(f"Error processing message: {str(e)}")def on_close(self, ws, close_status_code, close_msg):"""连接关闭的回调"""logger.info(f"Connection closed: {close_status_code} - {close_msg}")self.is_ws_connected = False  # 设置连接状态为Falsedef on_error(self, ws, error):"""错误处理的回调"""logger.error(f"WebSocket error: {str(error)}")self.is_ws_connected = False  # 发生错误时设置连接状态为Falsedef send_message(self, message_obj):"""发送消息到WebSocket服务器"""if self.is_connected():try:self.session.send(json.dumps(message_obj))except Exception as e:logger.error(f"Error sending message: {str(e)}")else:logger.warning("Cannot send message: Not connected")def ping(self):"""发送心跳包"""ping_obj = {"code": 10010,"trace": "01213e9d-90a0-426e-a380-ebed633cba7a"}self.send_message(ping_obj)# 使用示例
if __name__ == "__main__":ws_client = WebsocketExample()ws_client.connect_all()# 保持主线程运行try:while True:schedule.run_pending()time.sleep(1)except KeyboardInterrupt:logger.info("Exiting...")if ws_client.is_connected():ws_client.session.close()

四、代码逻辑解读

以上代码包含5个部分：

#1 建立连接

通过 websocket.WebSocketApp 建立连接后，会自动回调 on_open。

#2 订阅期货行情

在 on_open 中发送三类请求：

• 10000 → 实时成交明细；

• 10003 → 实时盘口；

• 10006 → 实时K线。

你可以更换合约代码，比如：

"data": {"codes": "NGAS"}  # 天然气

#3 自动重连

使用 schedule 每 10 秒检测一次连接状态，若断线则重新执行 connect()。

#4 心跳保活

服务器通常会在长时间无数据时断开空闲连接，因此每 30 秒发送一次心跳包 code=10010。

#5 消息处理

每当行情变化，服务器会推送 JSON 格式的消息，例如：

{"code":10000,"data": {"s": "IF2401.CFE","p": "4563.4","v": "2","t": 1731475200000}
}

可在 on_message 中解析并写入数据库或推送到前端。

五、错误码说明

错误码	说明
500	服务异常
501	请求频率超出一分钟限制
505	产品数量超出限制
506	参数缺失
515	参数不是 JSON 格式
513	WebSocket 心跳超时
401	认证错误，API Key 不正确，或者未将 API Key 放置在指定 header 或 query 中
404	API 接口不存在
427	WebSocket 连接数量超过套餐配额
429	请求频率限制，请求不符合您当前套餐配额
499	客户端主动中断，通常发生在网络不稳定的地区，请在良好网络环境连接
524	源服务器连接超时，通常发生在加速代理提供商到服务器之间不稳定导致，一般很快就能恢复