美股 Level2 深度數據標準化採集：WebSocket 單連線動態訂閱工程實作

前言

在跨境美股量化建模、歷史資料回測與即時策略運行的過程中，訂單簿深度資料的穩定性，直接左右 VWAP 因子、盤口流動性評分、買賣壓力權重等模型特徵的真實性。實務爬取 Tick 封包、對照多種行情介面後發現兩大普遍資料缺陷：其一為各廠商對空掛單檔位的處理邏輯不一致，造成樣本計算出現系統性偏移；其二為傳統 WebSocket 需斷線重連才能更換監控標的，導致時序樣本出現斷層，回測與實盤的特徵分佈無法對齊。

本文以標準化行情交互規範為基礎，提出單一長連線動態調整訂閱範圍的完整解決方案，包含資料失真成因拆解、多場景參數配置對照、線上採集故障排查邏輯、可直接部署的 Python 原始碼，適用於歷史 Tick 回放、高頻即時採集兩類量化場景，所有處理邏輯皆可透過原始行情封包重現驗證。

一、訂單簿空檔位造成量化模型偏差之根源

市面上多數美股 Level2 深度快照分為兩種空掛單處理機制，兩者皆會污染回測與實盤的特徵樣本：

1. 動態裁切陣列：單一價位掛單完全成交後，介面直接移除該陣列元素，買賣盤長度隨流動性變動。依索引分層計算流動性、多層價差、加權均價的模型會持續輸出錯誤數值，回測樣本與真實市場分佈產生落差，容易誤判策略收益。

2. 保留陣列結構、價格與掛單量歸零：若採集程式未新增過濾機制，0 值會納入所有加權運算。盤前、盤後流動性稀薄時，連續多個空檔位會人為壓低盤口均價，套利模型在回測產生大量虛假開倉訊號，放大過擬合風險。

本人在建構盤前流動性套利模型時曾遇到典型案例：單一個股連續六檔買盤瞬間被消化，未過濾空檔位的資料集計算出充足承接力道，回測收益表現亮眼；但套入真實行情時，同樣盤面會持續觸發開倉並產生浮虧，確認樣本失真為核心成因。

除此之外，多連線切換標的會衍生三項時序缺陷：

1. 斷線至重新完成 TCP 握手的窗口期無任何 Tick、深度快照輸入，時序特徵出現空白區段，時間序列模型訓練基礎被破壞；

2. 短時間批量切換美股、外匯、加密貨幣標的，大量並行握手請求容易觸發介面流量限制，行情推送中斷，回測與實盤可觀測樣本數量不一致；

3. 多條 WebSocket 常駐程式持續佔用記憶體，劇烈波動時 Tick 大量湧入，回呼隊列堆積，取樣時間戳持續延遲，高頻模型產生時序錯位。

二、單連線動態訂閱之標準定義

單連線動態訂閱指維持一條持續運作的 WebSocket 長連線，透過專屬指令攜帶新增 (add)、移除 (del) 動作與標代碼清單，即時調整監控範圍，全程不銷毀、重建 TCP 通道。

相較 REST 輪詢、斷線重訂閱模式，底層心跳機制、網路連線全程維持，僅更新本機標的管理集合，徹底消除連線重建帶來的資料空白，確保回測與即時採集的時序連續性一致。

三、動態訂閱配置規範與空檔位過濾機制

多場景參數對照表

空檔位標準化處理邏輯

符合規範的深度快照採用固定長度陣列儲存買賣價位，掛單完全消化僅將對應層級 size 設為 0，price 保留合法數值，不會回傳 null 或空字串。

量化資料採集層僅需新增一層過濾規則：僅將 size>0 的價位納入因子運算與樣本儲存，完全隔絕 0 值對 VWAP、流動性評分、多層價差模型的干擾，讓回測與即時採集使用同一套過濾標準。

底層通訊執行規則

1. 不同資產分離 WSS 通道，美股專屬連線位址：wss://quote.alltick.co/quo...；

2. 統一使用 cmd_id=22004 做為訂閱變更指令，透過 action 區分新增、移除；

3. 以集合結構於本機存放已訂閱標代碼，傳送指令前自動去重，避免重複接收相同標的行情；

4. 設定 10 秒週期心跳 ping，即時辨識 Socket 半斷線，預防無感覺的資料中斷。

四、量化採集常見故障檢測與補救機制

故障 1：大量 Tick 湧入造成回呼阻塞，取樣時序偏移

現象：行情劇烈波動時每秒數千筆 Tick 推送，on_message 佔用主執行緒，取樣時間戳延遲持續擴大

檢測方式：紀錄訊息佇列長度，單秒接收 Tick 超 5000 筆即判定壅塞

補救方案：回呼函數僅執行過濾與狀態標記，VWAP、流動性等耗時因子計算移交非同步執行緒池；優先過濾 size=0 空檔位，減少無效運算量。

故障 2：網路不穩產生 Socket 假活，持續接收殘缺深度快照

現象：弱網環境連線半斷，心跳逾時前不會觸發 on_close，持續取得不完整訂單簿資料

檢測方式：連續 3 次心跳未收到 pong 回應，判定連線失效

補救方案：客戶端自建心跳計數器，逾時主動斷線重連；重連後讀取本機訂閱清單批量恢復監控，樣本不會缺失。

故障 3：短時間連續增刪訂閱，執行緒競態產生幽靈訂閱

現象：已執行 del 移除的標的仍持續推送 Tick，資料集混入不需要的品項樣本

檢測方式：比對即時 Tick 代碼與本機訂閱集合，找出已移除卻仍推送的標的

補救方案：訂閱指令傳送加上執行緒鎖，封包送出成功後才更新本機集合，維持狀態一致。

故障 4：標代碼命名空間缺失，訂閱無提示失效

現象：遺漏 NASDAQ: 前綴，介面不回報錯誤，該標的完全無法採集資料，回測樣本缺漏

檢測方式：對照官方美股標代碼清單，驗證前綴格式

補救方案：本機建立代碼校驗字典，傳送指令前攔截格式錯誤代碼，輸出標準錯誤日誌方便後續樣本溯源。

五、機制適用邊界說明

支援場景：單條運作中的 WebSocket 連線，透過 cmd_id=22004 自由增減美股、外匯、加密貨幣標的，一站式完成多品項行情取樣；

不支援場景：多條 WebSocket 之間同步訂閱狀態、透過該指令回溯歷史 Tick、使用非 22004 私有指令修改監控清單。

六、完整 Python 行情採集原始碼（動態訂閱 + 空檔位過濾）

import websocket
import json
import threading

# 美股專屬行情WSS通道，規範參考官方API文件
WS_STOCK_URL = "wss://quote.alltick.co/quote-stock-b-ws-api?token=YOUR_TOKEN"
# 本機存放訂閱標的，用於去重與斷線復原
subscriptions = set()

def send_subscribe_cmd(ws, action, code_list):
    """統一封裝訂閱變更指令，固定cmd_id=22004"""
    if not code_list:
        return
    unique_codes = list(set(code_list))
    req_payload = {
        "cmd_id": 22004,
        "action": action,
        "code": unique_codes
    }
    ws.send(json.dumps(req_payload))
    # 同步更新本機訂閱清單
    if action == "add":
        for code in unique_codes:
            subscriptions.add(code)
    elif action == "del":
        for code in unique_codes:
            if code in subscriptions:
                subscriptions.remove(code)

def on_open(ws):
    # 連線建立後批量初始化訂閱，對應回測基礎樣本
    init_tickers = ["NASDAQ:AAPL", "NASDAQ:TSLA"]
    send_subscribe_cmd(ws, "add", init_tickers)
    print("WebSocket連線建立，完成初始標的訂閱")

def on_message(ws, message):
    """行情回呼，執行空檔位過濾，輸出有效盤口樣本"""
    if not message:
        return
    data = json.loads(message)
    bids = data.get("bids", [])
    asks = data.get("asks", [])
    valid_bids = []
    valid_asks = []
    # 過濾空檔位，僅保留size>0有效掛單
    for level in bids:
        price = level.get("price", 0)
        size = level.get("size", 0)
        if size > 0 and price > 0:
            valid_bids.append({"price": price, "size": size})
    for level in asks:
        price = level.get("price", 0)
        size = level.get("size", 0)
        if size > 0 and price > 0:
            valid_asks.append({"price": price, "size": size})
    # 此處可接入因子計算、樣本持久化邏輯
    print("有效買盤前3檔樣本：", valid_bids[:3])

def on_error(ws, error):
    print("WebSocket連線異常紀錄：", error)

def on_close(ws, close_code, close_msg):
    print("連線中斷，待復原之監控標的清單：", subscriptions)

if __name__ == "__main__":
    ws_client = websocket.WebSocketApp(
        WS_STOCK_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    # 10秒自動心跳，提早偵測半斷線狀態
    ws_client.run_forever(ping_interval=10)

七、此架構對量化研究與即時採集的實質增益

1. 因子與回測資料一致性提升：透過固定長度陣列搭配 size 欄位辨識空檔位，根除索引偏移、0 值汙染等問題，盤口流動性、VWAP、分層價差等核心因子，在回測集與即時取樣集擁有相同分佈，降低模型過擬合機率；

2. 多標的並行採集效率提升：單一長連線可動態調整監控範圍，無需重複建立 TCP 連線，不存在行情空白區段，適合多因子同步回測研究；

3. 採集程序資源消耗可控：單一連線承載多品項行情接收，減少記憶體佔用；心跳機制提早捕捉網路異常，斷線後自動恢復全部訂閱，減少手動維護成本與樣本缺漏；

4. 資料異常完整可溯源：整套行情交換邏輯依 AllTick API 標準規範設計，訂閱封包、深度快照格式皆可對照官方文件，因子輸出異常時可逐幀比對原始 Tick 封包定位資料瑕疵，方便模型誤差複盤。