運用 WebSocket 動態訂閱建置穩定 A 股即時行情串流｜量化數據管線實戰方案

7 月 13 日

在開發自用 A 股量化分析工具的過程中，我需要同時接入滬深交易所、港股、加密貨幣與大宗商品的多源即時行情 API，長期處理 Tick 逐筆數據、建構回測資料集。實作期間，WebSocket 長連線衍生各種穩定性問題，整整卡了數週的數據管線開發；本文分享一套經過實盤驗證、標準化的單連線動態增量訂閱架構，完整解決連線斷開、重連風暴、髒數據干擾等問題，附上可直接執行的 Python 原始碼。

前言：自建 A 股量化工具時，長連線行情串流踩過的穩定性坑

若未設計統一的訂閱抽象層，每一次新增或移除觀察清單的 A 股標的，都必須銷毀並重建 WebSocket 連線。這會引發大規模重連風暴，產生 Tick 數據空白區間，直接打斷日內高頻策略的訊號生成；同時開啟數十條並行 WebSocket 通道時，輕微網路波動就會造成 Socket 假存活狀態，本地回呼佇列堆積大量未處理的 A 股行情資訊。短時間連續發送訂閱、取消訂閱指令會產生競態條件，衍生幽靈訂閱、重複二級盤口推送等狀況，每個交易日都要耗費數小時翻閱日誌除錯。

初期我嘗試兩種傳統方案：REST 輪詢、單標的獨立連線。輪詢會帶來無法接受的延遲，不利日內高頻交易；一檔股票一條連線的架構，則很容易觸發 API 並行連線數上限。經過多輪迭代與線上優化，我建構單連線動態訂閱框架，統一規範所有 A 股即時行情 API 的連線管理、訂閱邏輯與數據解析。所有故障場景都能透過日誌與官方 API 規格重現驗證，下文完整拆解整套設計思路。

一、多資產 A 股行情 API 原生存在的相容痛點

1. 各供應商協定、欄位、時間戳規格完全割裂

不同數據商提供的 A 股即時行情 API，訂閱規則彼此互不兼容。部分端點只能在全新建立連線後才能修改觀察標的，且各資產類別有獨特的代碼命名規則：A 股統一 6 位數字代碼、港股搭配市場前綴、加密貨幣使用交易對字串。成交價欄位混用 price、last_price 兩種名稱，時間戳則同時存在秒、毫秒兩種單位，沒有統一標準。

合併 A 股與跨市場數據製作 K 線、執行量化回測時，時間偏移會直接破壞計算結果。若針對每一套行情 API 硬編碼專屬轉換邏輯，後續新增外匯、貴金屬品種時，必須完整重寫連線、訂閱、解析整套流程，大幅拉高回歸測試與迭代成本。

2. 高頻日內交易專屬的效能缺陷

切換 A 股標的強制完整重連：每次握手驗證、批量訂閱的週期會產生數據盲窗，短線高頻策略會遺漏關鍵進出場訊號。
心跳維護資源消耗龐大：數十條獨立 WebSocket 通道各自運行心跳偵測，微小網路波動就會批量斷線，同步重連的請求洪流壓垮 API 伺服器。
本地與伺服器訂閱狀態不同步：快速連續發送增減訂閱指令會造成訊息亂序，產生無明確錯誤日誌的隱性故障 —— 已取消的 A 股持續推送 Tick、新增標的完全沒有行情回傳。

二、核心概念：何謂動態增量訂閱

動態增量訂閱指重複使用一條持續存活的 WebSocket 長連線，全程不關閉、不重新初始化 Socket；開發者發送標準化指令框架，攜帶欲新增 / 移除的標的代碼清單，逐步更新有效訂閱範圍。

此模式和兩種舊式實作有明顯區分：REST 快照輪詢、單品種獨立長連線。最大優勢為單通道重複利用、訂閱範圍增量更新，同一套邏輯就能介接所有類別的即時行情 API，同時適用離線回測數據採集、線上實盤低延遲 Tick 串流兩大場景。

三、高頻 A 股場景可複製對照表

四、單連線動態訂閱架構：根治 A 股行情連線不穩定問題

1. 依市場分類標準化 WebSocket 端點 + 統一訂閱模型

行情介面分為兩套標準 WSS 長連線位址，區分股票與加密、外匯、貴金屬，不用自行拼接非標準網域：

A 股、港股、美股專屬股票通道：wss://quote.alltick.co/quo...
加密貨幣、外匯、貴金屬通道：wss://quote.alltick.co/quo...

兩組端點共用完全相同的 cmd_id=22004 訂閱框架結構，僅標的代碼依市場規則區分：A 股使用 6 位數字代碼、美股附加交易所前綴、加密貨幣直接使用交易對名稱。單一適配層可支援全類資產，大幅減少重複程式碼。

2. 增量更新訂閱，完全避免重連風暴

新增、移除 A 股標的僅傳送輕量指令框架，Socket 底層連線持續保活。相較「一標的一連線」架構，只需維護一套心跳、斷線重連邏輯；網路短暫波動時僅單一通道重試，不會出現數十條連線同步握手的併發衝擊。日誌透過固定連線 ID 完整追蹤生命週期，方便回測數據異常溯源。

3. 本地記憶集合快取，消除幽靈訂閱

透過 subscriptions 集合儲存當下所有有效 A 股標的代碼，新增自動去重、取消訂閱同步清理快取；每筆傳入的 A 股 Tick 都會先驗證標的是否屬於有效集合，過濾因訊息亂序殘留的無效推送，避免回測資料庫寫入髒數據、實盤模型接收雜訊樣本，降低後續數據清洗成本。

4. 全域欄位映射 + 毫秒時間戳統一轉換

所有 A 股即時行情 API 的原始回傳資料，統一映射為內部標準結構：

code：標的代碼
last_price：最新成交價
volume：累計成交量
全部時間戳強制轉為毫秒精度。A 股與跨市場數據抵達業務層時格式完全一致，無需為不同來源分開撰寫時間轉換模組，杜絕 K 線聚合、量化策略計算時的時間偏移誤差。

5. 標準心跳與分層異常回呼，偵測隱藏 Socket 假死

設定 10 秒自動心跳維持連線活性，完整實作 on_message、on_error、on_close 三層回呼鏈。連續多次未收到伺服器 PONG 回應即判定通道異常，自動執行重連；並區分本地網路斷線、伺服器限流關閉兩類異常輸出差異化日誌，快速定位回測斷流、實盤行情中斷根因。

五、導入此架構對量化研究的實質效益

切換 A 股觀察池無數據斷層：批量擴充回測樣本、實盤調整監控標的時 Tick 串流不中斷，無空白採樣區間，確保回測與實盤數據分佈一致，降低模型過擬合風險。
擴充資產迭代成本大幅縮減：後續接入外匯、大宗商品行情 API，僅需補充標的代碼對應規則，不需重寫整套 WebSocket 連線、心跳、訂閱邏輯，樣本池擴充效率大幅提升。
全鏈路日誌可追溯：A 股訂閱指令、Tick 推送、心跳逾時、連線斷開完整留存紀錄，回測結果失真、實盤訊號偏移時，可透過連線 ID、標的代碼快速定位數據層問題。
硬體資源開銷可控：單一通道整合數十條獨立行情連線，心跳、訊息傳輸的網路 I/O 負荷明顯下降；24 小時不間斷採集 A 股回測數據，不會發生訊息佇列堆積、記憶體持續上漲的狀況。

線上與回測環境常見故障對應方案

1. 現象：A 股高頻 Tick 大量湧入，本地消費佇列持續堆積

偵測指標：未處理訊息佇列長度連續五分鐘持續上升

對應方式：於 on_message 內部實作非同步消費佇列並設定長度門檻，觸發門檻時輸出警告日誌，暫時取消低權重 A 股標的訂閱釋放運算資源，防止數據遺失。

2. 現象：網路波動造成 Socket 假存活，心跳逾時前不會觸發關閉回呼

偵測指標：連續三次傳送心跳封包，皆未收到行情 API 伺服器 PONG 回應

對應方式：新增 12 秒本地逾時守護機制，逾時主動斷開並重啟通道，避免無效連線持續佔用頻寬、接收無效 A 股 Tick 寫入回測資料庫。

3. 現象：快速增減標的產生並行競態，本地快取與伺服器訂閱狀態不一致

偵測指標：日誌持續出現已取消訂閱之 A 股 Tick 推送

對應方式：所有訂閱、取消指令循序發送，透過同步鎖保護 subscriptions 集合，完整傳送指令後才更新本地快取，禁止並行修改訂閱清單。

4. 現象：A 股標的代碼格式錯誤，訂閱無報錯、靜默失敗

偵測指標：發送訂閱指令後，長時間無對應標的 Tick 流入回測資料庫

對應方式：發送指令前增加代碼格式校驗，僅放行 6 位數字 A 股代碼，非法格式直接攔截並輸出錯誤日誌，減少無效 API 請求。

架構能力邊界說明

本動態訂閱架構支援單一 WebSocket 內自由增刪 A 股與跨市場標的，但開發前需確認業務需求，存在三項硬性限制：

不支援多條行情通道之間同步訂閱狀態
不提供 A 股歷史 Tick 批量回溯查詢功能
無法解析 cmd_id=22004 以外的私有擴充指令

Python 完整可執行程式碼（相容回測數據採集、實盤 Tick 串流）

import websocket
import json
import threading
import time

# A股/港股/美股 即時行情WebSocket通道
STOCK_WSS_URL = "wss://quote.alltick.co/quote-stock-b-ws-api?token=YOUR_TOKEN"
# 加密貨幣、外匯、貴金屬行情通道
CRYPTO_WSS_URL = "wss://quote.alltick.co/quote-b-ws-api?token=YOUR_TOKEN"

# 本地有效A股標的快取，避免重複訂閱與幽靈推送
subscriptions = set()
# 統一訂閱固定指令ID
SUBSCRIBE_CMD_ID = 22004
ws_app = None

def send_subscribe_action(action: str, code_list: list):
    """單連線發送訂閱/取消指令，用於回測樣本、實盤標的動態調整"""
    global ws_app
    if not ws_app or not ws_app.sock or not ws_app.sock.connected:
        print("無活躍行情通道，跳過指令發送")
        return
    # 攔截空白標的清單，防止行情API拋出參數異常
    if not code_list:
        print("標的代碼清單為空，攔截訂閱指令")
        return
    # 清洗無效標的代碼
    valid_codes = []
    for code in code_list:
        if isinstance(code, str) and len(code.strip()) > 0:
            valid_codes.append(code.strip())
    # 同步更新本地訂閱快取
    if action == "subscribe":
        for c in valid_codes:
            subscriptions.add(c)
    elif action == "unsubscribe":
        for c in valid_codes:
            if c in subscriptions:
                subscriptions.remove(c)
    # 組建行情API標準訂閱框架
    req_frame = {
        "cmd_id": SUBSCRIBE_CMD_ID,
        "action": action,
        "code": valid_codes
    }
    ws_app.send(json.dumps(req_frame))
    print(f"執行{action}指令，標的清單（含A股）：{valid_codes}")

def on_open(ws):
    print("WebSocket行情通道建立完畢，執行A股初始批量訂閱（回測樣本）")
    init_codes = ["600000", "000001", "NASDAQ:AAPL"]
    send_subscribe_action("subscribe", init_codes)

def on_message(ws, message):
    """Tick統一接收邏輯，可直接對接回測入庫、實盤因子計算"""
    if not message or len(message.strip()) == 0:
        return
    try:
        data = json.loads(message)
        tick_code = data.get("code", "")
        # 過濾已取消訂閱標的之幽靈推送
        if tick_code not in subscriptions:
            return
        # 空值防護，避免量化計算、資料庫寫入報錯
        last_price = data.get("last_price", 0)
        volume = data.get("volume", 0)
        timestamp_ms = data.get("timestamp", 0)
        if last_price <= 0 or timestamp_ms <= 0:
            return
        # 業務擴充點：Tick寫入資料庫供回測 / 即時因子、策略訊號計算
        print(f"Tick數據 | code:{tick_code} 價格:{last_price} 成交量:{volume} 時間戳:{timestamp_ms}")
    except Exception as e:
        print(f"A股市行情數據解析異常：{str(e)}")

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

def on_close(ws, close_code, close_msg):
    print(f"行情通道斷開 代碼:{close_code} 資訊:{close_msg}，即將自動重連採集A股Tick數據")
    # 斷線清空本地訂閱快取，重連後重新初始化訂閱
    subscriptions.clear()

def run_ws_client():
    global ws_app
    while True:
        # 範例使用A股股票長連線通道
        ws_app = websocket.WebSocketApp(
            STOCK_WSS_URL,
            on_open=on_open,
            on_message=on_message,
            on_error=on_error,
            on_close=on_close
        )
        # 10秒心跳保活，12秒逾時門檻
        ws_app.run_forever(ping_interval=10, ping_timeout=12)
        # 斷開後等待3秒重試連線行情API
        time.sleep(3)

if __name__ == "__main__":
    client_thread = threading.Thread(target=run_ws_client)
    client_thread.daemon = True
    client_thread.start()
    # 模擬回測過程新增加密品種訂閱
    time.sleep(10)
    send_subscribe_action("subscribe", ["BTCUSDT"])
    # 模擬收盤清理閒置A股標的訂閱
    time.sleep(20)
    send_subscribe_action("unsubscribe", ["600000"])
    # 主執行緒持續阻塞，維持背景行情執行緒
    while True:
        time.sleep(1)

總結

本文圍繞量化回測、實盤高頻策略兩大核心場景，建構單連線 WebSocket 動態訂閱架構，系統性解決多源 A 股即時行情 API 協定割裂、連線不穩、數據偏移等工程痛點。透過統一訂閱指令、本地標的狀態快取、全域欄位與時間戳歸一化三層封裝，實現多市場行情標準化接入，顯著降低量化研究者的數據採集、樣本清洗、線上維運成本。

整套方案基於通用 WebSocket 規範實作，文中範例程式碼透過 AllTick API 完成長期線上數據採集驗證，可直接用於回測資料集建置、日內高頻策略實盤數據串流開發；後續擴充港股、貴金屬、外匯等資產樣本池時，僅需補充對應市場標的代碼規則，無需重構底層連線訂閱邏輯，具備長期迭代維護優勢。

CC BY-NC-ND 4.0 授权

喜欢我的作品吗？别忘了给予支持与赞赏，让我知道在创作的路上有你陪伴，一起延续这份热忱！

kalos

来自作者
相关推荐