Python 实战：从 HTTP 轮询到 WebSocket，打造毫秒级外汇行情网关

在构建金融交易系统时，我们常说“天下武功，唯快不破”。但在外汇交易的实战开发中，很多开发者往往卡在了第一步：如何优雅且高效地获取实时数据？

前阵子我在优化一个即时汇率换算模块，目标是同时监听 USD/JPY 和 EUR/USD 的波动。需求很明确：低延迟、低资源占用、高稳定性。

传统的 requests.get() 轮询方案在这里是行不通的。每一次 HTTP 请求都要经历三次握手、传输、断开，这种开销对于高频变化的行情数据来说是巨大的浪费。而且，你很难控制轮询的频率——太快了服务器当你是 DDoS，太慢了又捕捉不到瞬间的插针行情。

解决这个问题的标准答案就是 WebSocket。它允许建立一次连接后保持双向通信，服务器有新价格直接推送到客户端。我在对比了几个 API 文档后，选择了 AllTick API 作为演示对象，主要是看重它在断线重连和数据包结构的简洁性上做得比较符合开发直觉。

首先，摒弃复杂的框架，回归最基础的 websocket-client。

pip install websocket-client requests

接下来的核心代码涉及三个回调函数：on_open（建立连接时订阅）、on_message（接收数据）、on_error（错误处理）。

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    print(f"{data['symbol']} | {data['price']} | {data['time']}")

def on_open(ws):
    subscribe_msg = {
        "action": "subscribe",
        "symbols": ["EURUSD", "USDJPY"]
    }
    ws.send(json.dumps(subscribe_msg))

ws = websocket.WebSocketApp(
    "wss://api.alltick.co/forex/realtime",
    on_open=on_open,
    on_message=on_message
)

ws.run_forever()

当你订阅了多个货币对时，数据流的压力会变大。

import csv
from datetime import datetime

def save_tick(data):
    with open("forex_tick.csv", "a", newline="") as f:
        writer = csv.writer(f)
        writer.writerow([
            datetime.now(),
            data["symbol"],
            data["price"]
        ])

在处理这些并发数据时，我的经验是：千万不要在 on_message 里做耗时的计算逻辑。先把数据塞进队列（Queue）或者存下来，计算逻辑另起线程处理，否则会阻塞心跳，导致连接断开。

subscribe_msg = {
    "action": "subscribe",
    "symbols": ["EURUSD", "USDJPY", "GBPUSD", "AUDUSD"]
}

从 HTTP 转向 WebSocket，本质上是思维方式从“主动查询”到“事件驱动”的转变。如果你手头也有类似的监控需求，不妨试试上面的代码。你会发现，当数据流实时涌入控制台的那一刻，整个系统的“手感”完全不同了。