雲端量化開發筆記：以 WebSocket 實現外匯無斷流動態訂閱架構

無論是個人開發小型量化工具，或是於雲伺服器、容器叢集建置專業行情擷取服務，幾乎都會遇到「執行中動態調整外匯監控貨幣對」的需求。若訂閱更新的程式架構設計不完善，很容易出現 Tick 報價斷層、延遲雜訊堆積、雲端運算負載飆升等狀況，連帶造成線上實盤訊號失真、離線回測數據缺漏，大幅影響量化策略的可信度。

一、量化開發中兩大動態調整訂閱場景

在建置自動化交易、行情監控系統時，需要即時增刪貨幣對的狀況主要分為兩類：

1. 手動互動式行情儀表板

   依據當下市場波動度、流動性手動切換觀察標的，短時間內可能多次調整訂閱清單，多用於盤中套利機會掃描、跨品種走勢對照研究。若處理不當，每次切換都會產生數據空窗，錯過關鍵價差訊號。

2. 自動輪動策略引擎

   透過量化多因子門檻自動篩選、替換監控池，訂閱調整呈碎片化、高頻特性，常搭配定時批量回測作業。頻繁發送訂閱指令不僅加重伺服器負載，也會累積大量無效行情佔用雲端頻寬。

許多初學者會使用「直接中斷 WebSocket、重新全量訂閱」的簡易寫法，看似快速完成功能，實則暗藏諸多缺陷：重連期間遺失連續 Tick，回測樣本不完整；持續反覆建立銷毀連線，容易觸發 API 存取限制，長期運行的雲服務穩定性大打折扣。

我們架構設計的核心目標：全程維持單一 WebSocket 長連線不中斷、切換標的時零數據遺失、減少無效報價傳輸降低雲資源消耗、控管本機訊息佇列負載，確保實盤與回測使用完全一致的行情數據標準。

二、動態切換引發數據異常的根本緣由

現行多數外匯即時行情 API 皆採用「單一傳輸通道承載多個標題」的運作模式，伺服器僅依據客戶端傳送的指令推送數據，不會自動優化訂閱清單的異動。未做好狀態分離管控時，會衍生三種直接干擾量化研究的問題：

1. 全量重置訂閱邏輯

   每次切換前清空所有訂閱品種，再重新批量訂閱，兩段操作之間會形成數據空白區間，導致 K 線重製、滑點統計、歷史回測大量缺樣，策略模擬結果完全失真。

2. 只新增、不註銷過期標的

   持續新增觀察貨幣對，卻沒有寫入取消訂閱邏輯，隨時間累積大量無關行情，容器或雲主機內的訊息佇列持續膨脹，拖慢指標運算與模型推論速度。

3. 短時間連續發送訂閱請求

   手動操作或自動策略觸發的連續調整，會打亂伺服器推送時序，大量重複、延遲的髒數據湧入本地，增加數據清洗、寫入資料庫的額外運算成本。

透過雲日誌回放、壓測復原場景後可定位核心問題：開發者未區分「當前正在生效的標的集合」與「欲切換的目標集合」，缺少集合差集運算、短時異動防抖緩衝兩道關鍵處理步驟。

三、適合雲量化專案的分層優化架構

3.1 雙模組解耦，隔離連線與訂閱狀態

將整段行情處理流程拆為兩個互不耦合的獨立模組，從底層避免切換標的造成的數據抖動，亦相容容器彈性伸縮部署：

• 連線管理層：專責 WebSocket 心跳維護、網路異常自動重連，無論監控標的如何調整，傳輸通道持續維持連線，杜絕無預警斷流；

• 訂閱狀態管理層：負責比對新舊貨幣對清單、拆分新增與註銷請求，所有訂閱異動僅在此模組內執行，不會更動底層長連線狀態。

3.2 以集合差集實現增量更新，捨棄全量覆蓋

不使用陣列直接覆蓋清單，改用 Set 結構分別儲存當前生效、目標監控兩組貨幣對，透過集合運算區分兩類操作：

待新增標的 = 目標集合 − 當前生效集合

待註銷標的 = 當前生效集合 − 目標集合

分開傳送 subscribe、unsubscribe 請求，不會一次性清空全部品種，徹底消除數據空白窗口，同時避免長期累積無效行情。我開發行情擷取、回測前置處理流程時，會選用 AllTick API 作為數據來源，統一規格的請求報文，差集計算完成的清單可直接封裝傳送，大幅縮減雲端對接與除錯的時間成本。

3.3 200ms 防抖緩衝，合併短時間多重異動

手動調整面板、自動輪動策略都會在短時間內多次修改監控清單。新增 200 毫秒的緩衝視窗，視窗內所有訂閱變更統一合併運算，僅執行一次更新請求，減少對外 API 呼叫次數，降低伺服器與雲端頻寬負擔。

3.4 本地延遲 Tick 過濾機制

即便已傳送取消訂閱指令，伺服器仍會短暫推送滯後的舊品種報價。在數據進入指標運算、回測邏輯前增加過濾判斷，直接丟棄已註銷標的之延遲數據，避免髒數據干擾量化輸出結果。

開發核心思維總結

處理動態訂閱時，應捨去逐筆處理單一指令的思維，轉為「集合狀態持續管控」的邏輯。隨時維護一份精準的有效標的清單，以集合差值做增量更新，程式可讀性、後續除錯與回測問題復原難度都會大幅降低。整套架構的優化核心，不只是實現切換標的功能，而是在彈性雲環境中維持完整連續的數據流，縮小實盤與離線回測之間的策略評估落差。

四、可直接部署雲端之 Python 實作程式

import json
import websocket
from typing import Set

# 雲服務當前生效訂閱貨幣對
current_symbols: Set = {"EURUSD", "GBPUSD"}
# 切換後目標監控標的池
target_symbols: Set = {"EURUSD", "USDJPY", "AUDUSD"}

# 透過集合差集計算新增、待註銷清單
add_list = list(target_symbols - current_symbols)
remove_list = list(current_symbols - target_symbols)

def refresh_subscription(ws_conn):
    global current_symbols
    # 傳送新增訂閱請求
    if add_list:
        sub_cmd = json.dumps({"action": "subscribe", "params": add_list})
        ws_conn.send(sub_cmd)
    # 傳送取消訂閱請求
    if remove_list:
        unsub_cmd = json.dumps({"action": "unsubscribe", "params": remove_list})
        ws_conn.send(unsub_cmd)
    # 同步更新本地紀錄之有效標的
    current_symbols = target_symbols.copy()

五、雲端落地心得

這套分層增量訂閱架構輕量無額外算力耗損，同時相容單台雲主機、彈性容器叢集兩種部署模式，可同時支撐本地輕型行情監控、雲端自動交易、離線批量回測等各式量化任務。

透過「長連線持續保活、短時異動合併、延遲數據過濾」三層優化，一次解決行情斷流、冗餘頻寬消耗、高頻 API 請求三大常見問題，統一線上實盤與離線回測的行情擷取邏輯，減少因數據處理落差帶來的策略誤差，適合長期維運的量化開發流程。