頻寬限制下貴金屬即時 API 多品種動態訂閱量化實務

研究前言

在貴金屬量化回測、盤中即時策略執行的流程之中，Tick 數據源的穩定供給，直接左右訊號時效性與回測結果的真實度。多數量化研究者一開始接入貴金屬即時 API 時，習慣使用「單一商品對應獨立 WebSocket」的簡易寫法；這套邏輯在地端測試直觀好除錯，但長時間壓測、多商品同時監控的場景下，會陸續暴發流量限制阻斷、行情斷流、訊息堆積等資料鏈路問題，最終導致策略訊號延遲、回測樣本出現空缺。

本文分享經過實盤長時間驗證的單一長連線動態訂閱架構，附上可直接重複使用 Python 完整程式、線上故障歸納、各場景邊界驗證規則，適用黃金、白銀、鉑金、鈀金同步數據擷取，可直接整合進量化回測框架、即時監控工具，減少資料異常對因子模型、收益回測的干擾。

一、實盤觀測到的數據鏈路故障

初期僅監控 XAUUSD、XAGUSD 兩項貴金屬，單品種單連線架構短期運行數據完整，看不出明顯缺陷。隨研究需求擴增，新增 XPTUSD、XPDUSD 至監控清單，系統連續執行 4 小時後，持續觀察到三類會干擾量化研究的數據異常：

1. API 持續正常推送 Tick 封包，但本機回呼佇列不斷溢位，指標運算、訊號生成執行緒落後，實盤交易訊號產生延遲、回測時間切片缺漏；

2. 貴金屬劇烈波動時段，多條連線同步觸發心跳重連，形成重連風暴，大幅提高觸發流量限制的機率；

3. 同時觸發帳號連線數上限、單一通道訊息頻率雙重門檻，部分貴金屬數據流暫時中斷，回測資料集產生空白區間。

二、量化研究對數據鏈路的硬性規範

適用貴金屬多因子模型、日內短線回測、即時策略監控的數據擷取系統，必須符合四項條件，維持數據連續完整：

1. 相容金融即時 API 的連線數、訊息頻率雙重限制，避免限流造成數據斷層，確保回測樣本無缺損；

2. 交易時段可動態新增、移除監控商品，切換訂閱過程不會遺失任何 Tick，不破壞回測時序連續性；

3. 數據接收、因子運算、行情持久化完全解耦，單一商品大量波動數據，不會堵塞全品種擷取流程；

4. 所有訂閱異動皆留存完整紀錄，方便後續異常數據溯源、回測誤差歸因驗證。

三、傳統多連線架構對量化研究的負面影響

1. 連線維護成本隨商品數線性成長

每新增一項貴金屬，就要建立獨立 WebSocket，心跳維持、斷線重連、訂閱補發的程式碼量會成倍增加。行情活躍期多通道同時推送 Tick，主執行緒循序處理封包，造成因子計算卡頓，回測與實盤訊號同步延遲。

2. 更容易觸發 API 流量限制，損害回測完整性

多數貴金屬即時 API 都會限制單一帳號最大並行連線數、單通道每秒訊息上限，多連線同時擷取很容易碰觸門檻，API 會暫停所有行情推送，讓回測資料出現空白，造成模型擬合、收益計算失真。

3. 調整監控商品一定產生數據斷層

多連線架構新增或移除商品時，必須關閉全部通道重新建立連線、完整重訂閱，重連的空窗期會遺失 Tick；大規模同步重連還會加重限流處罰，延長數據中斷時間。

4. 數據接收與運算高度耦合

Tick 接收、指標運算、資料庫寫入全部寫在同一個回呼函數，貴金屬快速漲跌時訊息快速堆積，即時因子更新延遲，實盤訊號失真，回測時序對照產生偏差。

5. 斷線後隱性數據缺失，難以釐清回測誤差源頭

網路斷線只重建 Socket，沒有補送訂閱指令，介面顯示連線正常，但完全收不到任何貴金屬 Tick，不會拋出明顯錯誤，研究者很容易將數據缺漏誤判為策略本身失效。

四、單一長連線動態訂閱標準實作方案

4.1 方案定義

動態訂閱機制仰賴單一長期維持的 WebSocket 通道，透過標準指令即時調整監控商品清單，新增、移除貴金屬全程不需中斷或重建連線。相較多通道分離、REST 輪詢兩種擷取方式，消除重連帶來的數據斷層，維持回測時序完整，是流量限制環境下最合適的量化數據擷取架構。

4.2 場域落地驗證對照表

4.3 Python 完整擷取程式（可直接嵌入量化框架）

import websocket
import json
from queue import Queue
import threading
import time

# 貴金屬/外匯通用 WebSocket 行情介面
WSS_URL = "wss://quote.xxx.co/quote-b-ws-api?token=YOUR_TOKEN"
# 全域 Tick 佇列：隔離數據接收與因子運算，避免計算阻塞擷取
tick_queue = Queue(maxsize=5000)
# 本機訂閱集合：去重、同步狀態，供異常數據溯源使用
subscriptions = set()

def send_subscribe_frame(ws, code_list, action="subscribe"):
    """封裝標準訂閱/取消訂閱指令"""
    if not isinstance(code_list, list) or len(code_list) == 0:
        return
    frame = {
        "cmd_id": 22004,
        "action": action,
        "code": code_list
    }
    ws.send(json.dumps(frame))
    # 同步更新本機訂閱狀態，留存完整操作軌跡
    if action == "subscribe":
        for code in code_list:
            subscriptions.add(code)
    elif action == "unsubscribe":
        for code in code_list:
            if code in subscriptions:
                subscriptions.remove(code)

def on_open(ws):
    """通道建立後執行初始全品種貴金屬訂閱"""
    init_metals = ["XAUUSD", "XAGUSD", "XPTUSD"]
    send_subscribe_frame(ws, init_metals, action="subscribe")
    print("初始貴金屬訂閱完成，目前監控商品：", subscriptions)

def on_message(ws, message):
    """僅將封包送入佇列，不執行複雜運算，確保擷取不中斷"""
    if not message:
        return
    try:
        data = json.loads(message)
        code = data.get("code", "")
        price = data.get("price", 0)
        # 過濾空值、零價髒數據，避免汙染回測資料集
        if code and price > 0:
            tick_queue.put(data)
    except json.JSONDecodeError:
        return

def tick_consumer():
    """獨立執行緒：因子計算、Tick 持久化、量化訊號生成"""
    while True:
        tick_data = tick_queue.get()
        code = tick_data["code"]
        price = tick_data["price"]
        # 此處填入貴金屬因子、回測資料儲存邏輯
        print(f"處理行情 {code}，最新報價：{price}")
        tick_queue.task_done()

def on_error(ws, error):
    print("WebSocket 行情通道異常：", error)

def on_close(ws, close_code, close_msg):
    print("行情通道斷線，等待自動重連，本機留存訂閱清單：", subscriptions)

if __name__ == "__main__":
    # 啟動獨立消費執行緒，分離數據接收與量化運算
    consumer_thread = threading.Thread(target=tick_consumer, daemon=True)
    consumer_thread.start()

    ws_app = websocket.WebSocketApp(
        WSS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    # 每10秒發送心跳，維持長連線穩定
    ws_app.run_forever(ping_interval=10)

五、量化擷取常見異常與對應兜底方案

異常 1：劇烈波動時 Tick 佇列溢位

現象：黃金、白銀大幅走勢期間，tick_queue 持續滿載，消費執行緒處理速度跟不上 API 推送速率。

監控指標：定時輸出佇列長度，連續 30 秒超過 3000 筆即判定異常。

解決方案：設定佇列容量上限，溢出時捨棄早期舊封包；依商品拆分多組消費執行緒分散負載。

異常 2：網路抖動造成 Socket 假存活，無報價流入

現象：短暫網路中斷，心跳尚未觸發逾時判斷，介面顯示連線正常，但長時間沒有任何貴金屬 Tick。

監控機制：記錄各商品最後一筆數據時間戳，單一商品 15 秒無新數據判定通道假活。

兜底邏輯：定時同步本機訂閱清單送出訂閱指令，修復數據串流。

異常 3：頻繁增刪商品產生競態，出現幽靈訂閱

現象：短時間多次切換監控清單，本機集合與伺服器訂閱狀態不一致，不需要的商品持續推送數據干擾因子運算。

檢核方式：完整留存每筆訂閱操作日誌，對照 API 回覆驗證一致性。

兜底邏輯：訂閱變更指令循序執行，使用執行緒鎖保護訂閱集合，同一時間僅執行一次異動。

異常 4：商品代碼格式錯誤，訂閱無聲失敗

現象：使用 XAU/USD 分隔格式發送請求，API 不會回報錯誤，但完全接收不到對應行情，回測缺損。

檢核方式：對照官方商品代碼清單核對 code 欄位格式。

標準規範：統一使用 XAUUSD 無分隔標準代碼，透過常數清單集中管理所有貴金屬商品。

六、架構適用邊界

1. 支援：單一 WebSocket 通道內動態新增、移除任意貴金屬標的，適用多商品同步回測、盤中即時監控；

2. 不支援：多條 WebSocket 之間同步訂閱狀態、歷史 Tick 批次回溯查詢；

3. 功能限制：僅相容 cmd_id=22004 標準訂閱指令，私有擴充指令無法使用。

研究總結

貴金屬量化模型、日內高頻回測對於 Tick 數據連續性、時序完整度要求極高，傳統多連線架構容易受 API 限流、同步重連風暴干擾，造成資料集失真，直接影響因子有效性與策略收益估算。

採用單一長連線搭配動態訂閱架構，結合訊息佇列解耦、本機狀態去重、多層異常兜底機制，可穩定執行多貴金屬同步數據擷取。透過 AllTick API 標準化 WebSocket 訂閱指令，不需反覆中斷重建通道，即可彈性調整監控商品清單；整套程式邏輯透明、操作軌跡完整可追溯，能直接整合進各類量化回測工具，具備高度重複利用價值。