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三种接入方式，除了外汇还有股票、指数、期货、基金等产品。