美股量化回測最佳化：以 Python 與 WebSocket 提升行情資料擷取效率

在運用 Python 開發美股量化策略、執行歷史回測的過程中，不少開發者都會遇到一項共同難題：透過 HTTP 輪詢擷取 Tick 資料、分鐘線行情時，不僅整體耗時漫長，還容易遭遇 API 限流、連線中斷、切換監控標的時發生連線異常等狀況。

本文實戰導向，結合 AllTick API 導入 WebSocket 長連線 + 動態訂閱 架構，從根源降低網路連線額外負荷，提升多檔標的行情擷取的效率與穩定性。內容包含原理說明、場景配置、完整可執行程式碼、常見問題除錯與延伸優化方向，適合量化研究者、程式開發者參考與導入。

一、傳統 HTTP 輪詢模式的缺失

HTTP 短連線輪詢雖然實作門檻低，但應用在批量行情擷取、多標的並行回測場景時，缺點會被大幅放大：

1. 連線資源耗損高

   HTTP 屬於短連線協定，每次發送請求都必須重新建立、關閉網路連線。若循序擷取十餘檔美股的歷史行情，連線累積的額外開銷會大幅拉長作業時間，不利大規模歷史資料前置處理與模型回測。

2. 切換訂閱易引發狀態異常

   多數傳統寫法在新增、移除監控標的時，會直接中斷現有連線再重新訂閱，不僅容易引發連線風暴，也會造成本機標的清單與伺服器端訂閱狀態不一致，中斷資料連續性。

3. 資料鏈冗餘拖累執行效能

   行情 API 預設會回傳完整欄位，然而量化回測僅需開盤價、最高價、最低價、收盤價、成交量等核心數據。若未搭配本機快取機制，重複呼叫 API 也會降低資料解析與回測的執行速度。

針對上述痛點，改用 WebSocket 常駐長連線搭配動態訂閱，是更適合高頻、批量行情擷取的技術選擇。

二、WebSocket 動態訂閱原理與場景配置

2.1 何謂動態訂閱

動態訂閱是指維持單一持續運作的 WebSocket 長連線，透過專屬指令線上新增或移除商品代碼，全程無須中斷、重建網路連線。此模式不同於 HTTP 輪詢與斷線重訂閱，能穩定支援多標的動態監控與連續資料接收。

依照 AllTick 官方介面規範，美股行情使用專屬 WebSocket 連線位址，所有訂閱、取消訂閱操作，皆統一透過 cmd_id=22004 指令管理，介面規範標準化，易於導入正式專案。

2.2 典型應用場景與參數規範

搭配量化回測、標的監控的實務場景，整理對應設定邏輯與驗證標準：

1. 多標的初始批次訂閱

   場景需求：組合策略批量回測，一次性訂閱多檔美股行情。

   設定規則：使用 cmd_id=22004、action=subscribe，商品代碼格式統一為 交易所:商品代碼。

   驗證標準：僅建立單一網路連線，不會產生多餘備援連線。

2. 增量新增監控標的

   場景需求：回測過程中擴充觀察標的，擴大取樣範圍。

   設定規則：沿用現有長連線，使用 cmd_id=22004 搭配 subscribe 指令，追加新的商品代碼清單。

   驗證標準：原有連線維持正常通訊，僅傳送增量訂閱指令。

3. 指定標的取消訂閱

   場景需求：剔除無效標的、精簡回測取樣池。

   設定規則：使用 cmd_id=22004、action=unsubscribe，填入欲取消訂閱的商品代碼。

   驗證標準：本機訂閱清單同步更新，停止接收對應標的行情資料。

4. 重複訂閱邊界處理

   場景需求：程式邏輯疏漏，導致同一標的重複發送訂閱指令。

   設定規則：維持原有指令格式，並在程式層加入本機去重機制。

   驗證標準：伺服器不會重複推送行情，避免資料冗餘。

5. 空清單指令攔截

   場景需求：誤傳空的商品代碼清單至介面。

   設定規則：程式事前判斷，攔截空白代碼清單。

   驗證標準：不傳送無效指令，避免觸發介面異常。

三、完整 Python 實作程式碼

以下程式實作連線初始化、動態訂閱 / 取消訂閱、資料驗證、異常捕捉等完整功能，替換為個人 API Token 即可直接執行，適用於量化行情資料擷取場景。

import websocket
import json

# 介面規範參考：AllTick 官方 API 文件
# 美股專屬 WebSocket 連線位址
WS_STOCK_URL = "wss://quote.alltick.co/quote-stock-b-ws-api?token=YOUR_TOKEN"
# 本機訂閱集合，用於去重與狀態同步
subscriptions = set()

def on_open(ws):
    """連線建立後，執行初始批次訂閱"""
    print("WebSocket 連線已建立，開始執行初始標的訂閱")
    # 示範標的：納斯達克 Apple、Tesla
    init_codes = ["NASDAQ:AAPL", "NASDAQ:TSLA"]
    subscriptions.update(init_codes)
    # 組合標準訂閱指令
    sub_msg = {
        "cmd_id": 22004,
        "action": "subscribe",
        "code": init_codes
    }
    ws.send(json.dumps(sub_msg))

def on_message(ws, message):
    """接收行情資料，並過濾異常資料，確保回測資料品質"""
    if not message:
        return
    try:
        data = json.loads(message)
        code = data.get("code", "")
        price = data.get("price", 0)
        open_24h = data.get("open_24h", 0)
        # 過濾空白資料、異常數值
        if not code or price <= 0 or open_24h <= 0:
            return
        print(f"標的：{code} | 最新價格：{price} | 24小時開盤價：{open_24h}")
    except json.JSONDecodeError:
        return

def on_error(ws, error):
    """捕捉連線異常，方便除錯與維護"""
    print(f"連線異常：{str(error)}")

def on_close(ws, close_code, close_msg):
    """連線關閉時，清空本機訂閱狀態"""
    print(f"連線已關閉，關閉代碼：{close_code}，備註：{close_msg}")
    subscriptions.clear()

def add_subscribe(ws, code_list):
    """增量新增訂閱標的，沿用現有長連線"""
    if not code_list:
        return
    new_codes = [c for c in code_list if c not in subscriptions]
    if not new_codes:
        return
    subscriptions.update(new_codes)
    msg = {
        "cmd_id": 22004,
        "action": "subscribe",
        "code": new_codes
    }
    ws.send(json.dumps(msg))
    print(f"增量訂閱完成：{new_codes}")

def remove_subscribe(ws, code_list):
    """取消指定標的之訂閱"""
    if not code_list:
        return
    remove_codes = [c for c in code_list if c in subscriptions]
    if not remove_codes:
        return
    for c in remove_codes:
        subscriptions.discard(c)
    msg = {
        "cmd_id": 22004,
        "action": "unsubscribe",
        "code": remove_codes
    }
    ws.send(json.dumps(msg))
    print(f"取消訂閱完成：{remove_codes}")

if __name__ == "__main__":
    # 初始化 WebSocket 客戶端
    ws_app = websocket.WebSocketApp(
        WS_STOCK_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 資料造成回調堆積

   現象：高頻行情持續推送，導致回調函數壅塞，整體資料處理速度下降。

   處理方式：將資料接收與量化運算邏輯拆分，透過佇列實現非同步消費，勿在回調函數內執行回測、因子運算等複雜運算。

2. 網路震盪引發連線假死

   現象：長時間未收到新行情，程式無報錯、也未觸發關閉回調。

   處理方式：仰賴心跳機制維持連線，新增資料接收逾時判斷，逾時後自動執行重新連線。

3. 頻繁增刪標的導致狀態錯位

   現象：已取消訂閱的標的仍持續接收資料，新增標的無法正常取得行情。

   處理方式：為訂閱、取消訂閱流程加入執行鎖，限制同一時間僅執行單一指令，作業完成後檢核本機訂閱狀態。

4. 商品代碼格式錯誤，造成無聲訂閱失敗

   現象：程式正常執行，卻永遠無法接收目標標的行情。

   處理方式：嚴格遵循 交易所:商品代碼 格式，發送訂閱前增加格式與文字檢核。

五、功能使用邊界說明

導入前須清楚本架構的適用範圍與限制：

• 支援：在單一 WebSocket 連線內，動態新增、移除商品代碼，彈性調整回測標的池與監控範圍。

• 不支援：多組連線之間同步訂閱狀態、透過現有指令回溯歷史 Tick 資料、呼叫非 cmd_id=22004 的私有指令。

六、額外效能最佳化建議

在 WebSocket 架構基礎上，搭配量化回測的作業特性，可進一步優化全鏈路執行效率：

1. 精簡欄位：僅擷取回測、因子模型所需的核心欄位，縮減資料傳輸體積。

2. 本機快取：將重複使用的歷史行情儲存至本機，避免重複呼叫 API。

3. 儲存格式最佳化：採用 HDF5、Parquet 欄式儲存格式管理行情資料，相較傳統表格檔案，能有效提升回測階段的讀寫速度。

七、總結

相較傳統 HTTP 輪詢，WebSocket 動態訂閱架構有效降低網路連線額外負荷，解決美股行情擷取速度緩慢、連線不穩定、容易被限流等問題，適用於中長期歷史回測、高頻資料取樣、多標的組合策略研究等主流量化場景。

整套程式邏輯標準、介面規範明確，可快速整合至各類 Python 量化框架。搭配欄位篩選、本機快取、高效儲存等機制，能夠全面優化從資料擷取到策略回測的全流程效能，具備實務導入價值。