Python 3 步接入实时外汇行情 API（WebSocket+REST 完整代码）

如果你做过量化交易或者外汇相关的工具开发，大概率遇到过这个问题——明明只需要实时拿到几个主流货币对的报价，结果在技术选型上卡了半天。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/