量化資料串流優化：持久化 WebSocket 動態訂閱解決 API 限流與連線抖動

前言

即時逐筆 Tick 資料是量化回測、實盤交易策略的核心輸入來源。早期我使用 Python 搭配aiohttp非同步輪詢建構行情擷取管線，在多標的同時監控的場景下，持續遭遇 429 限流回應、連線反覆斷開、重連風暴等問題，直接造成回測資料斷層、實盤訊號延遲失真。

當時陸續測試信號量並行控管、本機短期快取、批量請求合併、指數退避重試等常見緩解手段，但這些方式僅能暫時削減流量高峰，無法從架構根源解決持續輪詢帶來的多餘請求。

後續依 AllTick WebSocket 訂閱規格重構完整資料擷取模組，將「主動重複拉取」轉換為「伺服器事件推送」，上線後限流攔截次數大幅降至極低，連線穩定性可透過心跳封包、訂閱日誌完整驗證。本文整理架構缺陷分析、單一長連線動態訂閱實作邏輯、可直接匯入回測與實盤系統的 Python 程式碼，以及長期執行累積的邊界問題對策，提供量化研究者與程式開發者參考交流。

一、REST 輪詢架構用於量化擷取的根本缺失

多數量化研究需要同時監控一籃子標的，非同步 HTTP 輪詢存在三項無法迴避的缺陷，直接影響回測可信度與實盤執行穩定度：

1. 瞬間突發流量觸發滑動視窗限流

   程式初始化多檔標的時，數十筆請求會在短時間集中送出；即便整體每秒請求量未超過 API 規格上限，權杖桶、滑動視窗演算法仍會判定為異常存取。信號量僅能抹平峰值，無法減少總請求次數；多組回測、多策略同時執行時，限流錯誤會成倍增加，造成行情斷訊、訊號延遲。

2. 無差別重複請求耗盡 API 配額

   多數標的數百毫秒內價格不會波動，但輪詢機制仍固定週期重複查詢，持續消耗呼叫額度。同時執行多份回測批次、多組實盤策略時，配額消耗速度明顯上升；本機快取只能拉長查詢間隔，無法根除主動發送請求的行為，大量歷史樣本採集時資源浪費尤為明顯。

3. 標的清單更動引發連線震盪

   策略參數調校、回測樣本篩選、自選標的增減時，REST 架構必須不斷建立、銷毀非同步任務，短時間湧現大量新請求，再度觸發限流。若改用短生命週期 WebSocket，每次調整監控清單都要重新建立 TCP 連線，握手成本高，還會發生本機訂閱清單與伺服器狀態不一致，導致回測樣本缺漏、實盤漏發進場訊號。

批量請求、快取、重試退避都屬於事後補償手段，無法解決輪詢持續產生請求的核心矛盾。AllTick WebSocket 動態訂閱將資料流轉為異動增量推送，架構層面減少九成以上多餘請求，維持回測、實盤雙場景的資料連續性。

二、單一持久連線動態訂閱核心邏輯

定義說明

動態增減訂閱：維持一條不中斷的 WebSocket 長連線，透過cmd_id=22004指令傳送新增、移除標的清單，調整監控範圍；全程不需關閉、重建 Socket，和 REST 輪詢、短連線反覆握手的傳統擷取方式有本質差異，完美對應量化時常調整觀察標的需求。

量化場域落地對照表

三、可銜接回測 / 實盤完整 Python 程式

import websocket
import json
import time

# WebSocket連線位址遵循AllTick官方通訊規範
STOCK_WSS_URL = "wss://quote.alltick.co/quote-stock-b-ws-api?token=YOUR_TOKEN"
# 全域訂閱集合，用於去重與狀態同步，避免幽靈訂閱干擾模型輸入
subscriptions = set()

def send_subscribe_frame(ws, action: str, code_list: list):
    """統一封裝訂閱指令，固定通訊識別cmd_id=22004"""
    if not code_list:
        return
    frame = {
        "cmd_id": 22004,
        "action": action,
        "code": code_list
    }
    ws.send(json.dumps(frame))

def on_open(ws):
    """連線建立回呼，載入策略預設觀察標的池"""
    init_codes = ["NASDAQ:AAPL", "NASDAQ:TSLA", "NYSE:JPM"]
    global subscriptions
    for c in init_codes:
        subscriptions.add(c)
    send_subscribe_frame(ws, init_codes)
    print("WebSocket連線建立，完成策略預設標的訂閱")

def on_message(ws, message):
    """行情推送回呼，增加髒資料過濾，確保模型輸入有效"""
    if not message:
        return
    data = json.loads(message)
    code = data.get("code", "")
    last_price = data.get("lastPrice", 0)
    # 過濾空標的、零價無效Tick，避免汙染回測與實盤資料集
    if not code or last_price <= 0:
        return
    # 此處可串接即時模型運算、回測資料儲存模組
    print(f"標的{code} 最新盤價：{last_price}")

def on_error(ws, error):
    print(f"WebSocket連線異常：{str(error)}")

def on_close(ws, close_code, close_msg):
    print(f"連線中斷，關閉代碼：{close_code}")
    # 清空快取，防止重連後重複訂閱造成資料冗餘
    global subscriptions
    subscriptions.clear()

# 工具函式：批量新增監控標的，對應策略迭代調參
def add_sub_codes(ws, new_codes: list):
    global subscriptions
    need_add = [c for c in new_codes if c not in subscriptions]
    if need_add:
        for c in need_add:
            subscriptions.add(c)
        send_subscribe_frame(ws, need_add)
        print(f"增量新增監控標的：{need_add}")

# 工具函式：批量移除監控標的，對應回測樣本篩選
def remove_sub_codes(ws, del_codes: list):
    global subscriptions
    need_del = [c for c in del_codes if c in subscriptions]
    if need_del:
        for c in need_del:
            subscriptions.discard(c)
        send_subscribe_frame(ws, "remove", need_del)
        print(f"移除監控標的：{need_del}")

if __name__ == "__main__":
    ws_app = websocket.WebSocketApp(
        STOCK_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. 全程維持單一 WebSocket 長連線，調整標的僅傳送訂閱指令，不執行ws.close()重建，減少 TCP 握手對實盤延遲的影響；

2. 使用集合管理本機訂閱狀態，發送指令前先行去重，避免重複 Tick 造成模型重複運算、回測結果偏離；

3. 內建心跳偵測，及早辨識無聲斷線，防範無感知遺漏行情導致策略失效。

四、長期實盤與回測常見問題排解

1. 劇烈波動時大量 Tick 湧入，主執行緒阻塞

現象：標的大幅震盪期每秒數百筆 Tick 進入，同步回呼阻塞主執行緒，訊息佇列持續膨脹、回測取樣時間大幅拉長。

偵測指標：訊息佇列長度、單筆 Tick 解析耗時、模型單次訊號生成延遲。

對策：建立非同步緩衝佇列，將 Tick 原始解析與量化模型運算分離；高耗時指標計算、回測取樣移至獨立執行緒池，不阻塞行情接收主執行緒。

2. 網路抖動產生半斷假活連線，無錯誤提示靜態遺漏行情

現象：網路不穩時連線進入半斷狀態，不會觸發 on_error、on_close，但伺服器持續推送、本機無法接收，回測資料出現空洞、實盤漏發進場訊號。

偵測指標：連續未收到心跳 pong 回應次數。

對策：開啟內建心跳機制，連續 3 次無 pong 回應則自動斷線重連，重連後重新傳送全部監控標的清單補足行情。

3. 短時間連續增減標的，本機與伺服器訂閱狀態錯亂

現象：快速切換回測樣本、批次調整監控池，短時間連續呼叫增減函式，本機集合與伺服器訂閱不一致，出現重複推送或完全缺標的行情，回測無法重現。

偵測方式：列印送出指令的標的清單，與本機集合即時比對。

對策：所有訂閱操作加上執行緒互斥鎖，同一時間僅能送出一筆指令，傳送完畢後才更新本機狀態。

4. 標的代碼缺少交易所前綴，訂閱無任何資料回傳

現象：僅傳入 AAPL 這類簡寫，未攜帶交易所命名空間，指令送出無錯誤回覆，但完全沒有 Tick 推送，回測、實盤皆無資料。

偵測方式：攔截 WebSocket 原始封包，檢查 code 欄位格式。

對策：封裝標的格式化工具，強制拼接交易所前綴，送出指令前驗證格式合法性。

五、功能邊界說明

✅ 支援：單一長連線內透過cmd_id=22004動態增減任意標的，適合多策略並行、回測樣本動態篩選；

❌ 不支援：多條 WebSocket 間同步訂閱狀態、透過該指令取回歷史完整 Tick、呼叫非標準私有擴充指令。

六、架構切換後量化開發與維運實質收益

團隊完整切換動態訂閱架構後，回測、實盤兩大場域皆有可量化改善：

1. 限流警示幾乎消失，可移除 asyncio 信號量、批量合併、指數退避三層防護邏輯，量化程式複雜度下降約 40%；

2. 標的增減無 TCP 重連成本，策略迭代、回測篩選時訊號延遲穩定，無瞬間流量突發；

3. 頻寬與運算消耗明顯降低，僅價格異動才推送 Tick，長時間橫盤標的不會產生封包；

4. 完整可追溯的訂閱、心跳、行情日誌，回測空洞、實盤延遲可快速定位，不需翻閱海量 HTTP 請求紀錄。

結語

對量化研究者來說，傳統 REST 輪詢先天存在流量不穩定的問題，持續的限流、連線抖動會直接破壞回測可信度與實盤穩定性。改用單一 WebSocket 長連線動態訂閱，以增量推送取代重複輪詢，是低成本、高穩定的資料擷取方案。整套 Python 模組輕量無複雜依賴，依 AllTick 標準 WebSocket 規格即可快速整合至回測框架與實盤策略，有效避開 API 限流、連線震盪帶來的資料失真，提升模型重現度與自動化交易連續性。