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

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

本文将使用Infoway API的期货行情接口，演示如何通过 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 logger
 
class WebsocketExample:
    def __init__(self):
        self.session = None
        self.ws_url = "wss://data.infoway.io/ws?business=common&apikey=yourApikey"
        self.reconnecting = False
        self.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_connected
 
    def 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  # 设置连接状态为True
        
        try:
            # 发送实时成交明细订阅请求
            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  # 设置连接状态为False
 
    def on_error(self, ws, error):
        """错误处理的回调"""
        logger.error(f"WebSocket error: {str(error)}")
        self.is_ws_connected = False  # 发生错误时设置连接状态为False
 
    def 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 中解析并写入数据库或推送到前端。

五、错误码说明