跨市場金融行情資料擷取實作：透過 Python WebSocket 取得 A股、港股、美股即時報價

標籤：Python、WebSocket、股票API、金融資料、數據採集

背景

當我們進行跨市場的金融資料研究、量化策略原型驗證，往往需要同時追蹤 A股、港股、美股的標的價格。傳統做法是開啟多套行情工具，分別查看不同市場的報價。

一旦遇到交易時段重疊，例如港股集合競價與美股盤前交易時間撞期，手動切換介面比對價格就會出現無法避免的延遲。這個問題會帶來兩個實務上的困擾：第一，人工觀測的資料無法直接對接程式，行情資料不能輸入下游計算模組；第二，人為延遲會干擾價格訊號判斷，不利模型驗證與邏輯測試。

因此我嘗試導入外部行情 API，把多個市場的即時行情匯流到 Python 程式當中，完成行情接收，並可進一步擴充資料儲存、指標計算等功能。

跨市場行情 API 的工程評估重點

網路上有不少服務宣稱提供全球股市行情，但若是做研究或原型開發，必須從系統可靠度的角度建立評估標準，優先順序如下：

1. 原生支援多個交易市場 API 需要原生支援 A股、港股、美股。部分服務僅專注單一海外市場，其餘市場必須另外引入第三方資料源。多個資料來源會衍生多套介面、欄位不一致、時間戳對齊、資料校驗等額外開發工作，拉高整體維護成本。

2. 長連線穩定性與傳輸延遲 開盤、集合競價時段，行情封包流量會暴增，也是 WebSocket 連線的壓力測試場景。如果連線發生卡頓或異常斷線，會直接造成即時行情中斷，事件驅動的研究原型就會失效，甚至產生錯誤樣本，影響測試結果。

3. 跨市場資料結構標準化 這一點會直接影響程式複雜度，也關係測試與模擬邏輯的一致性。倘若不同市場回傳的欄位名稱、價格精確度、時間戳格式互不相同，業務程式就需要撰寫大量分支做資料清洗與轉換，容易造成各市場處理邏輯分歧，提高開發與除錯成本。優先選擇使用同一套通訊協定、欄位規格的介面。

4. 呼叫配額與計費模式 必須先確認穩定性與資料規格符合需求，才進一步評估使用額度與成本。就算服務價格低廉，只要資料品質、連線穩定性不足，就不適合拿來做研究與資料擷取。

經過數家服務的比對測試，本次實作選用 AllTick API。此服務將 A股、港股、美股、外匯等品項封裝在同一套 WebSocket 協定，訂閱的互動邏輯完全一致，不需要針對各市場分開撰寫解析程式，方便在建置統一的資料擷取層，降低跨市場的適配成本。

Python 程式實作：單一 WebSocket 長連線訂閱多市場即時行情

以下範例基於 websocket‑client 函式庫，透過單條長連線同時訂閱 A股、港股、美股行情。程式可以在本機或是雲端環境測試執行，參考官方文件撰寫。

import json
import websocket

# 替換為你自己的存取權杖
TOKEN = "YOUR_TOKEN"

WS_URL = (
    "wss://quote.alltick.co/quote-stock-b-ws-api"
    f"?token={TOKEN}"
)

# A股、美股、港股測試標的清單
symbols = [
    {"code": "688036.SH"},
    {"code": "AAPL.US"},
    {"code": "700.HK"},
]


def on_open(ws):
    """WebSocket 連線建立完成，發送行情訂閱請求"""
    print("WebSocket connected")
    subscribe_req = {
        "cmd_id": 22002,
        "seq_id": 1,
        "trace": "matters_demo",
        "data": {
            "symbol_list": [
                {"code": item["code"], "depth_level": 1}
                for item in symbols
            ]
        }
    }
    ws.send(json.dumps(subscribe_req))


def on_message(ws, message):
    """行情推送回呼，可擴充資料儲存、指標計算等邏輯"""
    try:
        payload = json.loads(message)
        # 可擴充：寫入資料庫、訊息佇列、即時指標運算
        print(payload)
    except json.JSONDecodeError:
        print("Invalid JSON message")


def on_error(ws, error):
    print(f"WebSocket error: {error}")


def on_close(ws, code, msg):
    print(f"WebSocket closed, code:{code}, msg:{msg}")


if __name__ == "__main__":
    ws_app = websocket.WebSocketApp(
        WS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws_app.run_forever()

程式執行之後，終端機會持續輸出多個市場的即時行情，全部市場統一使用毫秒級時間戳，報價欄位規格一致，不需要撰寫大量條件判斷來區分市場。

📌 實務開發提醒

1. 這段僅是最小可執行範例，尚未實作斷線重連、異常告警、資料持久化；若是長時間執行擷取任務，建議加上迴圈重連與退避重試機制，抵銷公網網路抖動的影響。

2. 在 on_message 回呼函式當中，可以擴充邏輯，把原始行情寫入訊息佇列或是時間序列資料庫，供下游模組消費。

3. 需要留意介面限流規範，避免一次性訂閱過多標的觸發服務限制。

實作成效與適用場景

程式啟動後，A股、港股、美股行情會匯流至同一支程序輸出，不再仰賴多個軟體手動切換。在港股集合競價等高流量時段，資料傳送沒有明顯延遲，回傳報價與公開網頁行情大致對齊。

這套擷取方案適用於幾種場景：金融技術原型開發、量化研究即時資料收集、行情樣本資料集建構；標準化的原始資料流，可以減少多源資料適配帶來的邏輯差異，提升開發研究的效率。

總結

建構跨市場的行情擷取服務，不能只關心介面能不能拿到價格。市場覆蓋度、長連線穩定性、跨市場資料格式的一致性，會直接影響上層研究或原型系統的可靠度。透過 WebSocket 長連線，我們可以把 A股、港股、美股行情統一擷取，降低多市場資料處理的適配負擔。

本次實作使用的 AllTick API，憑藉統一的通訊協定簡化跨市場資料接入流程，適合作為金融資料原型開發、行情樣本採集的技術驗證來源。