外汇期货实时行情 API 使用教程
由bq0m7vas创建，最终由bq0m7vas 被浏览 2 用户
在量化交易、行情监控、风控系统或金融数据服务中，实时行情数据是最基础、也是最重要的数据来源之一。本文将通过一个真实的 HTTP API 示例，手把手教你如何查询外汇期货的实时 K 线数据。
一、什么是外汇期货？
外汇期货（FX Futures） 是在正规交易所上市交易的、以某种货币对为标的的期货合约。
与外汇现货（如 MT4/MT5 中的 GBP/USD）相比，外汇期货具有以下特点：
- 集中交易：在 CME 等交易所撮合成交
- 真实成交量和持仓量（OI）
- 有明确的合约到期日
- 强监管、数据透明
常见外汇期货示例（CME）
| 代码 | 含义 |
|---|---|
| 6E | 欧元期货（EUR/USD） |
| 6J | 日元期货（USD/JPY） |
| 6B | 英镑期货（GBP/USD） |
| M6B | 微型英镑期货 |
二、什么是外汇期货「主力连续合约」？
外汇期货是按月份交割的，例如：
- M6BH26（2026年3月）
- M6BM26（2026年6月）
如果直接使用单一月份合约，会频繁遇到换月、数据断层、技术指标失真
主力连续合约（Continuous Contract）
为了解决这个问题，行情系统通常提供主力连续合约：
将当前成交量最大的主力月份合约，按规则自动拼接成一条连续行情
M6B1! 的含义拆解
| 部分 | 含义 |
|---|---|
| M6B | 微型英镑期货 |
| 1 | 最近主力合约 |
| ! | 连续合约标识 |
M6B1! = 微型英镑期货的主力连续合约
注意：主力连续合约只能用于行情分析，不能直接下单交易。
三、API 接口简介
本文使用的是Infoway API的 HTTP GET 行情接口，用于查询 外汇期货的 K 线数据。
接口地址格式
https://data.infoway.io/common/v2/batch_kline/{klineType}/{klineNum}/{codes}
下面是一个最简单、可直接运行的示例，用于查询 M6B1! 的 1 分钟 K 线数据。
import requests
api_url = 'https://data.infoway.io/common/v2/batch_kline/1/10/M6B1!'
headers = {
'User-Agent': 'Mozilla/5.0',
'Accept': 'application/json',
'apiKey': 'yourApikey' # 申请API Key: www.infoway.io
}
response = requests.get(api_url, headers=headers)
print(f"HTTP code: {response.status_code}")
print(f"message: {response.text}")
我们首先需要构造 API 请求 URL：
api_url = 'https://data.infoway.io/common/v2/batch_kline/1/10/M6B1!'
这里的参数含义是：
| 参数 | 值 | 含义 |
|---|---|---|
| klineType | 1 | 1 分钟 K 线 |
| klineNum | 10 | 最近 10 根 K 线 |
| codes | M6B1! | 微型英镑期货主力连续合约 |
四、通过WebSocket查询外汇期货实时行情
我们也可以通过WS来订阅外汇期货的实时行情，使用下面的代码我们可以建立WS链接，并且查询英镑期货的K线、交易明细、盘口等实时数据。而且包含重连机制，可无脑复制使用：
import json
import time
import schedule
import threading
import websocket
from loguru import logger
# 申请API Key: www.infoway.io
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": "M6B1!"}
}
self.send_message(trade_send_obj)
# 不同请求之间间隔一段时间
time.sleep(5)
# 发送实时盘口数据订阅请求
depth_send_obj = {
"code": 10003,
"trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
"data": {"codes": "M6B1!"}
}
self.send_message(depth_send_obj)
# 不同请求之间间隔一段时间
time.sleep(5)
# 发送实时K线数据订阅请求
kline_data = {
"arr": [
{
"type": 1,
"codes": "M6B1!"
}
]
}
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()
\