外匯行情 API、股票、貴金屬、加密貨幣多源接入：該怎樣處理異質行情數據？

kalos
·
·
IPFS
·
在建置金融應用、量化研究平台的過程當中，經常需要同時匯入多種類別的市場行情，涵蓋外匯、貴金屬、加密貨幣與股票。倘若直接接入多家第三方行情 API，不同服務在驗證機制、商品代碼、時間格式、推送協定、存取頻率限制都存在相當大的差異。本文從專案踩坑的真實經驗出發，梳理跨市場行情接入常見的各類問題，並附上以 Python + WebSocket 實作的原型，協助開發者降低異質行情數據的整合難度。

一、跨市場行情整合常見痛點

過去開發一套多資產監控原型系統時，曾經同時接入三家外部數據源，分別負責外匯與貴金屬、加密貨幣、股票行情。各家服務的介面各自獨立，驗證規則、商品代碼、時間戳、回傳欄位互不相容，大量開發時間耗費在數據轉換與適配，反而壓縮了核心業務邏輯的開發時程。主要問題可以歸納為四點：

1. 商品代碼體系不一致

同一個交易標的，不同服務商的編寫格式會有差異。例如歐元兌美元會寫成 EURUSD 或是 EUR/USD；加密貨幣交易對有 BTC‑USDT、BTCUSDT 兩種寫法；股票代碼還需要加上交易所後綴。直接沿用另一個數據源的代碼查詢，往往會出現查無資料、數據缺失的狀況。

2. 時間戳與欄位資料型態異質

時間戳有秒、毫秒、時間字串等多種形式；價格、成交量欄位，部分介面回傳數值，部分則回傳字串。若沒有統一轉換型態，在時間序列統計、指標運算時很容易拋出例外，影響系統穩定性與數據可信度。

3. 推送機制、心跳與存取限制各不相同

部分 API 支援 WebSocket 長連線即時推送，部分僅能透過 HTTP 輪詢取得資料。各家的心跳保活、斷線重連、存取頻率限制規則皆為自訂，難以撰寫一套通用的客戶端，維護多個數據來源的成本會隨著市場數量上升。

4. 各市場交易時間模型存在差異

加密貨幣為全年無休 24 小時交易；外匯於週末休市；股票則有開盤、午休、例假日休市的規則。如果直接使用固定逾時時間做故障判斷，休市期間會產生大量無效告警，無法分辨是市場正常休市，還是數據來源真的發生故障，干擾系統監控作業。

二、多資產行情 API 的評估維度

選型的核心目標，是降低上層業務的適配負擔，減少異質數據源帶來的數據不一致風險，建議從以下四個面向進行評估：

  1. 驗證機制：不同資產類別是否可以使用同一組 Token 完成身份驗證，減少多組憑證的管理負擔。

  2. 協定一致性：不同商品的訂閱請求、推送回傳欄位的命名是否穩定、具備共通性。

  3. 文件完備度：是否提供完整可查閱的商品代碼清單與欄位說明。

  4. 執行約束說明：明確標示存取配額、限流規範，文件也需說明心跳、斷線復原的處理邏輯。

必須客觀理解：世上不存在欄位 100% 統一的第三方 API。就算是同一個服務商，不同資產也可能分屬不同接入端點，依然會存在細部欄位差異。工程上務實的做法，優先挑選協定標準化程度高的服務，再於應用層建構適配器，消化剩餘的格式差異。

在原型驗證階段，可以使用 AllTick API 進行測試。外匯、貴金屬、加密貨幣共用同一組 WebSocket 接入端點，美股、港股、A 股市場則使用另一組端點，兩套連線的訂閱協定維持一致，能夠有效減少適配層的程式開發量。

三、架構設計：透過內部 Quote 模型隔離外部數據來源差異

設計價值：把外部介面的變化封閉在適配層，確保量化運算、回測、上層業務邏輯不會因為第三方介面調整而受到影響。

無論外部 API 回傳的格式如何，在應用內部定義專屬的行情實體 Quote，只保留業務真正需要的核心欄位：數據來源標示、商品代碼、價格、成交量、毫秒等級時間戳。

所有外部原始行情數據流入系統之後，第一步就執行正規化轉換，映射為統一的Quote物件。上層模組，包含策略運算、指標統計、回測分析、系統監控，全部使用這套內部模型，完全不感知第三方原始 JSON 結構。

這個架構可以帶來兩項實質工程效益：

  1. 切換底層數據來源時，只需要修改數據轉換函式，上層業務、回測邏輯不需要更動；

  2. 新增某一個交易市場，僅補充數據來源設定，不需要大規模重構現有程式。

過去的專案便是依靠這套抽象設計，新增港股、美股行情的接入，僅耗費半天就完成測試上線。開發上建議一條原則：業務與回測邏輯，不要直接依賴第三方 API 的原始回傳結構。

四、原型實作：單一程序多路 WebSocket 行情接收

下方提供可除錯的原型程式碼，同一支程序維護多條獨立 WebSocket 連線，各市場的連線互相隔離，單一連線發生故障，不會影響其他數據流。適合前期技術驗證，若要上線生產環境，還需要加強容錯、資料持久化相關能力。

import asyncio
import json
import os
import uuid
from dataclasses import dataclass

import websockets

BASE = "wss://quote.alltick.co"

FEEDS = {
    "multi": {  # 外匯、貴金屬、加密貨幣
        "path": "/quote-b-ws-api",
        "codes": ["EURUSD", "GOLD", "BTCUSDT"],
    },
    "stock": {  # 美股、港股、A股
        "path": "/quote-stock-b-ws-api",
        "codes": ["AAPL.US", "700.HK", "600519.SH"],
    },
}

heartbeat = {"cmd_id": 22000, "seq_id": 1, "trace": "heartbeat", "data": {}}


@dataclass
class Quote:
    feed: str
    code: str
    price: float
    volume: float
    ts_ms: int


def normalize(feed, tick):
    """將外部原始tick資料正規化為內部Quote模型，供業務、回測模組呼叫"""
    return Quote(
        feed=feed,
        code=tick["code"],
        price=float(tick["price"]),
        volume=float(tick["volume"]),
        ts_ms=int(tick["tick_time"]),
    )


async def run_feed(name, cfg, queue):
    token = "填入你申請取得的AllTick_token"
    uri = f"{BASE}{cfg['path']}?token={token}"
    subscribe = {
        "cmd_id": 22004,
        "seq_id": 1,
        "trace": str(uuid.uuid4()),
        "data": {"symbol_list": [{"code": c} for c in cfg["codes"]]},
    }
    while True:  # 斷線自動重新連線
        try:
            async with websockets.connect(uri) as ws:
                await ws.send(json.dumps(subscribe))

                async def beat():
                    while True:
                        await asyncio.sleep(10)
                        await ws.send(json.dumps(heartbeat))

                task = asyncio.create_task(beat())
                try:
                    async for raw in ws:
                        msg = json.loads(raw)
                        if msg.get("cmd_id") == 22998:
                            await queue.put(normalize(name, msg["data"]))
                finally:
                    task.cancel()
        except (websockets.ConnectionClosed, OSError):
            await asyncio.sleep(3)


async def consumer(queue):
    while True:
        q = await queue.get()
        # 可在此擴充：寫入資料庫、推送業務模組、回測資料前置處理
        print(q.feed, q.code, q.price, q.ts_ms)


async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        consumer(queue),
        *[run_feed(n, c, queue) for n, c in FEEDS.items()],
    )


asyncio.run(main())

開發注意事項

  1. 各市場使用獨立 WebSocket 連線，分別處理訂閱、心跳保活、斷線重試；解析完畢的資料傳送到共用非同步佇列統一消費。

  2. 同一個 WebSocket 工作階段當中，重複發送訂閱指令會覆蓋原本的訂閱清單。動態新增商品時，需要把完整的商品清單重新送出。

  3. 接入新的數據來源，務必列印原始封包確認時間戳單位，時間單位錯誤，會直接造成時間序列、回測結果全部錯亂。

五、完成數據正規化之後仍要留意的工程細節

就算完成內部模型轉換，不等於整條數據鏈路已經完備，有幾點會直接影響數據可靠度，開發階段必須處理：

  1. 依不同市場設計故障告警邏輯

    不要使用全域單一的逾時門檻。加密貨幣的監控逾時可以設定較短；外匯、股票則要搭配交易日曆判斷。外匯週末沒有行情屬於正常休市，不應該觸發故障告警。

  2. 留意浮點數運算的精度問題

    外匯報價普遍是五位小數，A 股兩位小數，加密貨幣最高可到八位小數。牽涉金額、淨值計算場景，建議使用Decimal型別運算，避免浮點數累積誤差，導致業務計算、回測結果出現偏移。

  3. 歷史 K 線資料同樣需要執行正規化

    即時行情透過 WebSocket 取得，歷史 K 線大多呼叫 REST 介面，兩套介面回傳結構往往不一樣。取得歷史資料之後，同樣轉換成Quote內部模型再寫入資料庫。讓歷史回測、即時業務可以使用同一套資料物件，縮小實盤與回測之間的邏輯落差。

六、結語

多資產行情整合並沒有捷徑。選用數據來源時，優先挑選協定規範、文件完善的服務；架構層面上透過適配層隔離第三方介面的異質差異，保障上層業務、量化回測模組的穩定性。

開發多資產相關系統的時候，優先定義統一的內部行情模型，能夠有效降低後續迭代、切換數據來源的改造成本，提升整條數據鏈路的可維護性。在原型開發階段，也可以參考如 AllTick API 這類協定一致性較佳的行情服務，減少適配層的開發負荷。

免責聲明：本文為技術實務分享，範例程式僅適用原型驗證。部署至生產系統，必須補齊例外捕捉、數據校驗、持久化儲存、監控告警、權限安全等機制。

CC BY-NC-ND 4.0 授权
已推荐到频道：时事・趋势
kalos
  • 来自作者
  • 相关推荐

外匯量化資料工程實務：透過外匯匯率 API 實現即時、分鐘級歷史資料擷取與行情介面統一

美股Tick數據購買的實戰反思：量化回測表現亮眼，實盤卻出現顯著落差的成因與數據選型思路

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