Python实时外汇行情接入实战：WebSocket与REST K线查询

最近外汇市场的消息面非常密集。欧洲央行刚结束议息会议，如期加息25个基点，把存款机制利率推到2.50%，同时上调了全年通胀预期——能源价格上涨是背后的主要推手。另一边，美国CPI数据刚落地，虽然整体符合预期，但核心通胀环比略超市场预期，美联储下周议息会议的加息概率从六成跳到了七成以上。

这两件事凑在一起，欧元兑美元的波动明显加大。ECB加息当天欧元不涨反跌，因为市场注意力立刻转向”美联储会不会跟着加”——如果两边都加，利差不变，欧元自然没有方向性行情。这种央行政策博弈期，汇率价格跳动快、消息面密集，做行情看板或者交易辅助工具的开发者，对实时数据接口的需求就特别迫切。

这篇文章从技术角度聊聊，怎么用Python从零接入一套实时外汇行情——包括WebSocket实时报价推送、REST历史K线查询，以及把两者组合起来做一个轻量监控脚本。不依赖任何量化框架，标准库加 requests 和 websocket-client 就能跑。

为什么要分开用WebSocket和REST

很多初学者一开始会问：直接定时轮询REST不行吗，为什么还要搞WebSocket？

实际用过就知道差别很大。像最近ECB决议和美联储议息这种事件，汇率反应往往在几秒钟内完成。用REST轮询，哪怕你每5秒请求一次，也会错过中间的波动轨迹。WebSocket适合盯实时跳动——价格一变服务端就推过来，延迟在毫秒级，做盘中监控和事件触发很顺手。

但WebSocket不适合拉历史数据，你不能让它”把过去90天的日K线给我”。REST反过来，拉历史K线、查最新快照都很方便，但频繁请求既浪费带宽又有速率限制。

所以正确的用法是：REST拉历史数据做初始化和回测，WebSocket订阅实时数据做增量更新。下面分别说。

REST：拉取历史K线做初始化

假设我们要做一个欧元兑美元的行情图表，首先需要拉一段历史日K线作为底图，观察近期的走势结构。接口的结构很规整：基础地址是 https://api.itick.org，外汇相关的REST路径都在 /forex/ 下面。

认证用HTTP Header token，不是常见的 X-API-Key 或 Authorization: Bearer，这个坑要注意。请求参数是 region 加 code 的组合——region 是市场代码（外汇用 GB），code 是货币对代码（如 EURUSD）。K线周期用 kType 数字编码：8代表日K，1代表1分钟，5代表1小时。

import requests

API_BASE = "https://api.itick.org"

TOKEN = "your_token_here"

headers = {

"accept": "application/json",

"token": TOKEN

}

def  get_forex_kline(region, code, k_type=8, limit=90):

"""

获取外汇历史K线

region: 市场代码，外汇固定用 GB

code: 货币对代码，如 EURUSD、GBPUSD、USDJPY

k_type: K线周期编码

1=1分钟 2=5分钟 3=15分钟 4=30分钟

5=1小时 8=日K 9=周K 10=月K

limit: 返回条数，最大500

"""

resp = requests.get(

f"{API_BASE}/forex/kline",

headers=headers,

params={

"region": region,

"code": code,

"kType": k_type,

"limit": limit

}

)

resp.raise_for_status()

return resp.json()["data"]

# 拉取EUR/USD近90根日K线

kline = get_forex_kline("GB", "EURUSD", k_type=8, limit=90)

# 返回的每根K线字段是缩写

# t=时间戳(毫秒) o=开 h=高 l=低 c=收 v=成交量 tu=成交额

latest = kline[-1]

print(f"最新日K: 开={latest['o']} 高={latest['h']} "

f"低={latest['l']} 收={latest['c']} 量={latest['v']}")

这里有个容易踩的坑：返回字段名是缩写的。价格字段用 o/h/l/c，成交量是 v，时间戳是 t（毫秒级）。不是 open/high/low/close/volume 那种长名。写代码的时候先打印一次返回值确认字段，别猜。

拉完日K线之后，你可以自己算一下近期的波动幅度。比如EUR/USD最近在1.16附近来回震荡，如果你用日K的最高价减最低价再除以收盘价，算出来的日均波幅大概在0.4%左右——这个数字对你后面设计告警阈值很有用。

WebSocket：订阅实时报价

历史K线拉完之后，需要把最新价格实时推过来。像下周美联储议息这种事件，价格跳动可能在几分钟内完成，用REST轮询完全跟不上。外汇WebSocket的地址是 wss://api.itick.org/forex，跟REST同一个域名，只是协议换成了WebSocket。

连接的时候把token放在HTTP Header里：

import json

import threading

import time

import websocket

WS_URL = "wss://api.itick.org/forex"

TOKEN = "your_token_here"

authenticated = False

heartbeat_timer = None

def  start_heartbeat(ws):

"""服务端要求每30秒发一次心跳，否则可能断开"""

def  ping():

while  True:

time.sleep(30)

ws.send(json.dumps({

"ac": "ping",

"params": str(int(time.time() * 1000))

}))

global heartbeat_timer

heartbeat_timer = threading.Thread(target=ping, daemon=True)

heartbeat_timer.start()

def  subscribe(ws):

"""

订阅格式：{"ac": "subscribe", "params": "EURUSD$GB", "types": "quote"}

params 用 标的$市场 的格式，多标的用逗号分隔

types 可选 quote/tick/depth/kline@1 等

"""

ws.send(json.dumps({

"ac": "subscribe",

"params": "EURUSD$GB,GBPUSD$GB",

"types": "quote"

}))

WebSocket的状态机有个关键点：不能连上就发订阅。连接刚建立时服务端会先回一条 {"code":1,"msg":"Connected Successfully"}，这时候还不能订阅。必须等到收到鉴权成功消息 {"resAc":"auth","code":1,"msg":"authenticated"} 之后，才能发subscribe。很多人第一次接的时候卡在这里——连接成功了但订阅一直报错，就是因为没等鉴权。

def  on_open(ws):

print("连接已建立，等待鉴权...")

def  on_message(ws, message):

global authenticated

payload = json.loads(message)

# 等鉴权成功后再订阅，同时启动心跳

if payload.get("resAc") == "auth"  and payload.get("code") == 1  and  not authenticated:

authenticated = True

print("鉴权成功，开始订阅行情")

subscribe(ws)

start_heartbeat(ws)

return

# 心跳响应直接忽略

if payload.get("resAc") == "pong":

return

# 处理实时报价推送

data = payload.get("data") or {}

if data.get("type") == "quote":

# 推送字段和REST一致：s=标的 ld=最新价 o/h/l=当日开高低 t=毫秒时间戳

symbol = data.get("s")

price = data.get("ld")

ts = time.strftime("%H:%M:%S", time.localtime(data.get("t", 0) / 1000))

print(f"[{ts}] {symbol}: {price}")

ws = websocket.WebSocketApp(

WS_URL,

header=[f"token: {TOKEN}"],

on_open=on_open,

on_message=on_message,

)

ws.run_forever()

整个流程梳理一下：建立连接 → 等鉴权 → 发订阅 → 收推送 → 持续心跳。顺序不能乱，否则订阅会失败。

如果你想在美联储决议公布那一刻收到提醒，可以在 on_message 里加一个价格变动阈值判断——比如当EUR/USD在10秒内波动超过30个点就触发告警。这种用WebSocket做的实时监控，比轮询方案的响应速度快一个数量级。

REST查最新报价：补一个快照接口

WebSocket是增量推送，但有些场景你只需要一个当前快照，不想维护长连接。比如你写个定时任务，每小时记录一次主要货币对的最新价，这种就没必要一直挂着WebSocket。REST的报价接口刚好合适：

def  get_forex_quote(region, code):

"""获取实时报价快照"""

resp = requests.get(

f"{API_BASE}/forex/quote",

headers=headers,

params={"region": region, "code": code}

)

return resp.json()["data"]

# 一次查多个主要货币对

for pair in ["EURUSD", "GBPUSD", "USDJPY"]:

q = get_forex_quote("GB", pair)

# ld=最新价 chp=涨跌幅 ch=涨跌额 o/h/l=当日开高低

print(f"{pair}: 最新={q['ld']} 涨跌幅={q['chp']}%")

报价接口的返回字段比K线多几个：ch 是涨跌额，chp 是涨跌幅（百分比），ts 是交易状态。这些字段做仪表盘显示够用了。

最近EUR/USD就在1.16附近窄幅波动，ECB加息后欧元不但没涨反而小跌，说明市场当前更关注美联储那边的政策预期。这种”利差变化”类的分析，你可以用报价接口同时拉EURUSD和USDJPY，对比看美元指数的整体强弱。

组合起来：一个轻量行情监控脚本

把REST初始化和WebSocket实时推送结合，就是一个最基本的行情监控器——启动时先拉历史K线作为基线，然后订阅实时报价做增量更新。这个模式在央行决议日特别有用：你可以在决议公布前把脚本挂着，价格一动就能立刻看到。

import json

import threading

import time

import requests

import websocket

API_BASE = "https://api.itick.org"

WS_URL = "wss://api.itick.org/forex"

TOKEN = "your_token_here"

headers = {"accept": "application/json", "token": TOKEN}

# 启动时先拉历史日K作为基线

def  init_kline(code):

resp = requests.get(

f"{API_BASE}/forex/kline",

headers=headers,

params={"region": "GB", "code": code, "kType": 8, "limit": 30}

)

data = resp.json()["data"]

recent = [bar["c"] for bar in data[-5:]]

print(f"{code} 近5日收盘价: {recent}")

# WebSocket部分

authenticated = False

alerts = [] # 可以在这里加自己的告警逻辑

def  on_message(ws, message):

global authenticated

payload = json.loads(message)

if payload.get("resAc") == "auth"  and payload.get("code") == 1  and  not authenticated:

authenticated = True

ws.send(json.dumps({

"ac": "subscribe",

"params": "EURUSD$GB",

"types": "quote"

}))

return

data = payload.get("data") or {}

if data.get("type") == "quote":

price = data['ld']

# 这里可以加自己的告警判断，比如价格突破某个区间

print(f"实时更新 {data['s']}: {price}")

init_kline("EURUSD")

ws = websocket.WebSocketApp(

WS_URL,

header=[f"token: {TOKEN}"],

on_message=on_message,

)

ws.run_forever()

这个脚本虽然简单，但骨架已经完整：REST负责历史数据初始化，WebSocket负责实时增量，两者字段格式统一，切换的时候不需要做数据格式转换。

几个实际踩过的坑

Header字段名是 token，不是 X-API-Key 也不是 Authorization。我第一次接的时候用了 X-API-Key，一直返回鉴权失败，翻了文档才发现。

必须等鉴权成功再订阅。刚连上就发subscribe会被服务端拒绝，错误消息是 cannot be resolved action。用 resAc == "auth" 判断鉴权状态最靠谱。

K线周期是数字编码，不是字符串 "1d" 或 "1m"。8是日K，5是1小时，1是1分钟。这个没有默认值，必须传对。

WebSocket心跳不能省。30秒一次ping，超过1分钟不发服务端会主动断开。用一个daemon线程跑心跳就行，别阻塞主消息循环。

返回字段是缩写。ld 是最新价，chp 是涨跌幅，v 是量。不要用 last_price、change_percent 这种直觉字段名，会直接KeyError。

小结

外汇行情接入的技术栈其实不复杂：REST拉历史，WebSocket收实时，统一用token鉴权。关键是把状态机走对——连接、鉴权、订阅、心跳这四个环节顺序不能乱，字段名对照文档用缩写，剩下的就是业务逻辑了。

最近这个阶段，欧洲央行刚加完息，美联储下周要议息，汇率市场的消息密度会持续比较高。有一套顺手的实时数据接入方案，不管是做监控工具还是自己看盘，都能少走很多弯路。

接口的完整文档可以在这里查阅，支持REST、WebSocket和FIX三种接入方式，除了外汇还有股票、指数、期货、基金等产品。

本作品采用《CC 协议》，转载必须注明作者和本文链接