如何接入实时期货行情数据 - 详细操作步骤

期货市场通过标准化合约的交易，为投资者提供了在大宗商品、金融工具等方面进行风险对冲和投机的机会。量化交易以计算机模型为核心，通过历史数据和实时数据进行分析和策略执行，减少人为情绪对交易的干扰。由于期货市场的波动性强且价格变化迅速，量化交易者必须依赖实时准确的数据来做出交易决策。

期货行情数据接口是量化交易系统的基础模块，提供了交易者所需的实时价格、成交量、持仓量等关键市场数据。这些数据通过API自动传输给量化交易系统，使策略能够在毫秒级别内做出反应。API通常支持主流的期货交易所，如芝加哥商品交易所（CME）、上海期货交易所（SHFE）等，帮助量化交易者跨市场执行多资产策略。

注意延迟行据和实时数据的差别

目前你在市面上看到的几乎所有行情报价都存在延时，这里的延迟一般指的是交易的真实发生时间，到成交价格呈现在终端的时间，大概在15分钟左右。而实时数据则没有这个延迟，从交易发生到价格被传递出去，只有1秒以内的延迟。国内提供实时行情数据的供应商非常少，一般使用国外的产品。

实时行情数据接口一般用在哪些场景？

实时行情数据接口的应用场景主要包括：

量化交易策略执行

量化交易依赖于实时数据来识别市场上的价格微小变化并及时执行策略，例如套利策略、趋势跟随策略等。毫秒级别的价格信息可以让交易系统以最快的速度捕捉市场机会，优化交易收益。

交易平台的报价系统

在线期货交易平台需要为用户提供实时价格、深度图等市场信息，以支持用户下单。实时行情数据接口可以确保交易平台的数据准确性和及时性，提升用户体验，增加用户粘性。

算法交易

实时数据不仅用于执行交易，还用于监控市场风险和调整仓位。通过实时数据，量化交易系统可以设定动态风控模型，及时止损或止盈，确保投资组合的稳健运行。

期货市场对实时数据的需求非常高，使用高质量的实时行情数据接口可以帮助量化交易者和交易平台提高竞争优势，在快速变化的市场中占据领先地位。

如何接入期货实时行情API - 代码示例

下面以Infoway API提供的期货数据接口为例，通过接口查询美股原油的实时报价：

import requests
 
# 申请API KEY: www.infoway.io
api_url = 'https://data.infoway.io/common/batch_kline/1/10/USOIL'
 
# 设置请求头
headers = {
    'User-Agent': 'Mozilla/5.0',
    'Accept': 'application/json',
    'apiKey': 'yourApikey'
}
 
# 发送GET请求
response = requests.get(api_url, headers=headers)
 
# 输出结果
print(f"HTTP code: {response.status_code}")
print(f"message: {response.text}")

也可以通过Websocket查询：

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连接并启动自动重连机制"""
        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()