Python 3 步接入实时外汇行情 API（WebSocket+REST 完整代码）
7 / 0 / 创建于 3天前
Labea 的个人博客
如果你做过量化交易或者外汇相关的工具开发，大概率遇到过这个问题——明明只需要实时拿到几个主流货币对的报价，结果在技术选型上卡了半天。HTTP 轮询延迟太高，WebSocket 接入又不知道从哪里下手。今天这篇文章，带你用 Python 3 步搞定外汇实时行情的接入，代码可以直接复制进你的项目里跑起来。
为什么需要 WebSocket 而不是 HTTP 轮询
在正式写代码之前，先花 30 秒理解一个核心问题：为什么外汇实时行情要用 WebSocket 而不是 HTTP 轮询？
外汇市场日均交易量超过 6 万亿美元，价格波动到了毫秒级。HTTP 轮询的方式是客户端每隔一段时间主动问一次服务端“现在价格多少？”，延迟通常在 300~800 毫秒，而且频繁请求还容易被限流。
而 WebSocket 建立一条持久化的双向通信通道后，服务端有新的价格数据就会主动推过来，延迟低至 5~50 毫秒。对于需要实时响应市场波动的量化策略来说，这个差距是决定性的。
简单来说：HTTP 轮询像每隔几秒刷新一次网页，WebSocket 则像一个一直开着的直播流——数字一变化，你立刻就能看到。
准备工作
1. 安装必要的 Python 库
只需要两个库，干干净净：
pip install websocket-client requests
Python 版本 3.8 以上就够了。
2. 获取 API 密钥
你需要一个支持外汇实时行情的数据接口。以 iTick API 为例，去官网注册账号后，在控制台就能看到你的 API Key。免费额度足够日常开发使用。
第 1 步：REST API 获取实时快照
在 WebSocket 没建立起来之前，REST API 很适合用来快速查询当前价格或获取历史 K 线。
实时报价（Real-Time Quote）
这个接口返回的是带盘口信息的外汇报价，ld（Latest Deal）字段就是当前最新成交价。
import requests
url = "https://api.itick.org/forex/quote?region=GB&code=EURUSD"
headers = {
"accept": "application/json",
"token": "YOUR_API_KEY" # 替换成你自己的 Token
}
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
print(f"EUR/USD 最新价: {data.get('data', {}).get('ld')}")
else:
print(f"请求失败: {response.text}")
批量实时报价（Batch Real-Time Quotes）
如果用 REST 做多货币对监控，用批量接口效率更高，一次请求就能拿到多个货币对的最新价。
def fetch_batch_quotes(symbols):
"""批量获取多个货币对的实时报价"""
codes = ",".join(symbols)
url = f"https://api.itick.org/forex/quotes?region=GB&codes={codes}"
headers = {"accept": "application/json", "token": "YOUR_API_KEY"}
response = requests.get(url, headers=headers)
if response.status_code == 200:
quotes = response.json().get('data', {})
for symbol, quote in quotes.items():
print(f"{symbol} 最新价: {quote.get('ld')}")
else:
print(f"批量请求失败: {response.text}")
fetch_batch_quotes(["EURUSD", "GBPUSD", "USDJPY"])
历史 K 线查询（Historical Candlesticks）
回测或补充历史数据时，可以用 /forex/kline 接口获取 OHLCV 数据。参数 kType 取值范围是 1-10，覆盖分钟线到月 K 线。
def fetch_historical_bars(symbol, ktype=1, limit=100):
url = f"https://api.itick.org/forex/kline?region=GB&code={symbol}&kType={ktype}&limit={limit}"
headers = {"accept": "application/json", "token": "YOUR_API_KEY"}
response = requests.get(url, headers=headers)
if response.status_code == 200:
bars = response.json().get('data', [])
print(f"已获取 {len(bars)} 根K线")
return bars
else:
print(f"获取K线失败: {response.text}")
return []
bars = fetch_historical_bars("EURUSD", ktype=1, limit=100)
实时盘口与成交查询
如果策略需要更细粒度的数据（比如要看买卖盘深度或者每一笔逐笔成交），iTick 也单独提供了这两个 REST 端点：
实时盘口:
GET /forex/depth?region=GB&code=EURUSD实时成交:
GET /forex/tick?region=GB&code=EURUSD
使用方法基本一致，只需更换 URL 即可。
第 2 步：WebSocket 实时推送
需要持续追踪市场变化时，WebSocket 是更合适的选择。
iTick 的外汇 WebSocket 地址是 wss://api.itick.org/forex，和股票用的 wss://api.itick.org/stock 是分开的。
import websocket
import json
WS_URL = "wss://api.itick.org/forex"
SYMBOLS = ["EURUSD", "USDJPY"]
def on_message(ws, message):
data = json.loads(message)
symbol = data.get('symbol')
price = data.get('price')
print(f"{symbol} 最新价: {price}")
def on_open(ws):
print("连接已建立，正在订阅货币对...")
subscribe_msg = {
"action": "subscribe",
"symbols": SYMBOLS
}
ws.send(json.dumps(subscribe_msg))
def on_error(ws, error):
print(f"连接出错: {error}")
def on_close(ws, close_status_code, close_msg):
print("连接已关闭")
ws = websocket.WebSocketApp(
WS_URL,
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close
)
print("正在连接外汇行情服务器...")
ws.run_forever()
运行这段代码，控制台就会开始输出实时行情。
第 3 步：生产级 WebSocket 接入（带断线重连）
生产环境里一个常见问题是：连接意外断开（网络闪断、服务器重启、防火墙空闲超时），程序必须能够自动恢复，否则策略会停在半路上。
下面是在 WebSocket 订阅基础上加入的重连机制，同时保留了心跳维持的逻辑：
import websocket
import json
import time
WS_URL = "wss://api.itick.org/forex"
SYMBOLS = ["EURUSD", "USDJPY", "GBPUSD"]
RECONNECT_DELAY = 3
def on_message(ws, message):
data = json.loads(message)
symbol = data.get('symbol')
price = data.get('price')
print(f"{symbol} 最新价: {price}")
def on_open(ws):
print("WebSocket 连接已建立")
subscribe_msg = {
"action": "subscribe",
"symbols": SYMBOLS
}
ws.send(json.dumps(subscribe_msg))
print(f"已订阅: {SYMBOLS}")
# 心跳线程（部分服务商需要主动发送心跳包）
def send_heartbeat():
import threading
while ws.sock and ws.sock.connected:
time.sleep(30)
try:
ws.send(json.dumps({"type": "ping"}))
print("心跳已发送")
except:
break
threading.Thread(target=send_heartbeat, daemon=True).start()
def on_error(ws, error):
print(f"WebSocket 错误: {error}")
def on_close(ws, close_status_code, close_msg):
print(f"连接已关闭 (code: {close_status_code}), {RECONNECT_DELAY} 秒后重连...")
time.sleep(RECONNECT_DELAY)
start_ws()
def start_ws():
ws = websocket.WebSocketApp(
WS_URL,
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close
)
ws.run_forever()
if __name__ == "__main__":
print("外汇实时行情监控启动...")
start_ws()
重连逻辑的核心： on_close 里调用 start_ws() 重建连接，保证程序在断开后能够自动恢复。
完整性能组合：历史 K 线 + 实时推送
在实际的量化系统里，WebSocket 和 REST 往往是配合使用的：用 REST 拉取初始的历史 K 线数据，再通过 WebSocket 持续接收实时增量更新。
import requests
import websocket
import json
import pandas as pd
# Step 1: REST 获取历史 K 线
hist_url = "https://api.itick.org/forex/kline"
params = {"region": "GB", "code": "EURUSD", "kType": "1", "limit": 100}
headers = {"accept": "application/json", "token": "YOUR_API_KEY"}
res = requests.get(hist_url, headers=headers, params=params).json()
df = pd.DataFrame(res.get('data', []))
df['timestamp'] = pd.to_datetime(df['t'], unit='s')
print(f"已加载 {len(df)} 条历史 K 线数据")
print(df[['timestamp', 'o', 'h', 'l', 'c']].tail())
# Step 2: WebSocket 接收实时数据
def on_message(ws, message):
tick = json.loads(message)
symbol = tick.get('symbol')
price = tick.get('price')
print(f"实时更新: {symbol} = {price}")
ws = websocket.WebSocketApp(
"wss://api.itick.org/forex",
on_message=on_message,
on_open=lambda ws: ws.send(json.dumps({"action": "subscribe", "symbols": ["EURUSD"]}))
)
print("开始实时数据推送...")
ws.run_forever()
写在最后
跑起来之后，你会发现实时行情接入其实没那么复杂。核心就这三步：环境配好、WebSocket 连上、回调函数写好。第 2 步和第 3 步的代码直接复制到你的项目里，改掉 API 地址和货币对列表，基本就能跑了。
对于量化策略开发来说，数据通了，一切才算真正开始。这套基础框架搭好后，你就可以把精力放在模型构建、信号计算这些更有意思的事情上了。
参考文档：https://blog.itick.org/quant-tools/qlib-itick-forex-stock-integration
GitHub：https://github.com/itick-org/
本作品采用《CC 协议》，转载必须注明作者和本文链接
关于 LearnKu
粤公网安备 44030502004330号
推荐文章：