在做量化和行情对接的过程里，贵金属实时报价一直是最容易出问题的环节。我前后因为数据延迟、连接断开、脏数据干扰踩过很多坑，在不断优化后，终于实现了一套稳定、干净、不卡顿的黄金 / 白银 / 铂金实时行情获取方案，今天把完整思路和代码分享给大家。
为什么你的黄金 API 总是不稳？
和普通股票不一样，贵金属是全球 24 小时轮动交易，伦敦、纽约、香港、悉尼四个市场不间断开盘。
时段交替、数据发布、网络波动，都很容易出现这些情况：
行情突然跳空，程序没抓到
几秒钟没有新价格，数据断档
WebSocket 莫名断开，再也连不上
报价重复、异常价格，影响判断
这些小问题，在真正使用时都会变成大麻烦。
我最实用的一套解决方案
踩坑多次后，我总结出最稳的结构：
用 WebSocket 长连接，不用轮询
一个连接同时订阅黄金、白银、铂金
自动重连，断了立刻恢复
数据三层清洗，保证干净可用
下面是直接复制就能运行的代码。
import websocket
import json
import time
def on_message(ws, message):
data = json.loads(message)
# 只处理实时行情
if 'tick' in data:
symbol = data['symbol']
price = data['price']
ts = data['time']
print(f"[{symbol}] {price} @ {ts}")
def on_error(ws, error):
print("连接出错:", error)
time.sleep(2)
ws.run_forever() # 自动重连
def on_close(ws):
print("连接关闭，正在重连...")
def on_open(ws):
# 一次订阅黄金、白银、铂金
sub_msg = {
"action": "subscribe",
"symbols": ["XAUUSD", "XAGUSD", "XPTUSD"]
}
ws.send(json.dumps(sub_msg))
if __name__ == "__main__":
ws = websocket.WebSocketApp(
"wss://api.alltick.co/websocket",
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close
)
ws.run_forever()
必做的三步：让行情数据更可靠
拿到数据不能直接用，简单处理一下，准确率提升特别多：
去重
同一个价格、同一时间戳，只保留一条。
过滤异常
突然暴涨暴跌的错误价格，直接丢掉。
本地缓存
存最近 200 条行情，断网一瞬间也不会空白。
进阶用法：实时计算指标，自动提醒
把行情和指标结合，不用额外接口，来一条数据算一次：
def on_message(ws, message):
data = json.loads(message)
if 'tick' in data:
price = data['price']
rsi = quick_rsi(price)
if rsi > 70:
print("黄金超买，注意风险")
elif rsi < 30:
print("黄金超卖，可以关注")
不管是做实时监控、策略提醒、量化交易，都非常实用。
最后想说
API 只是一个数据通道，真正决定好不好用的，是连接稳不稳、重连快不快、数据干不干净。
把这些小细节做好，黄金实时报价就可以做到24 小时稳定、不卡顿、不出错。
如果你也在对接行情、写量化策略，希望这篇实战分享能帮到你。
