外匯接口的實作筆記：在量化開發與獨立協作之間，我如何處理即時與歷史行情

這篇不是產品推薦，也不是所謂的「最佳實踐清單」。它更像是一份開發過程中的紀錄：我如何從手動看盤，走到必須依賴程式化行情；又如何在資料覆蓋、時間基準、斷線重連與成本之間，慢慢找到一個可以運作的平衡。

一、創業案例：從一張手動更新的看板說起

最開始的項目很小。我們想做一個外匯行情看板，主要看 EUR/USD、GBP/USD、USD/JPY，順便給策略研究提供一些歷史K線。那時候大家覺得，看價格而已，網頁或交易軟體就夠了。確實，如果只是一個人偶爾確認一下價格，手動查看完全沒問題。

但項目一旦進入協作，問題就出現了：

• 有人看的是延遲幾秒的網頁價格，有人看的是交易軟體裡的報價，彼此對不上。

• 策略側需要歷史K線，手動匯出既慢又容易出錯。

• 後來想加價格預警，才發現根本沒有穩定的數據來源。

• 隊友之間開始花大量時間確認「你看到的價格是哪個時間點的」。

那段經歷讓我意識到，程式的世界裡，行情不是一個數字，而是一條資料流。只要你要開發行情看板、量化策略、價格預警，直接使用外匯接口就會方便很多。程式透過 API 取得標準化行情，後面的計算、儲存與策略處理，才有共同的基礎。這不是為了追求技術感，而是為了讓協作有可重複的依據。

二、數據痛點：取得到資料，不代表資料可用

我後來在選擇外匯接口時，會先看三個面向：資料覆蓋、資料粒度、接口方式。這三點看起來很基本，但真正踩過坑之後，才知道它們各自藏著不同的問題。

資料覆蓋與粒度

常見的 EUR/USD、GBP/USD、USD/JPY 等貨幣對當然要覆蓋。如果做短週期量化，還要確認是否提供 Tick 資料，因為 Tick 能記錄更細的價格變化。只做日線研究時，K線可能夠用；但一旦涉及短週期策略，Tick 與時間戳的品質就會直接影響回測結果。

接口方式的差異

接口方式也很重要。REST API 適合查詢歷史資料，例如取得某個時間範圍的K線；WebSocket 更適合即時行情，建立連線後可以持續接收伺服器推送的資料。兩者不是互相替代，而是分工。用輪詢去硬撐即時行情，通常只會讓延遲和資源消耗同時變差。

那些真正麻煩的細節

實際開發中，更難處理的往往是下面這些：

• WebSocket 斷線後沒有自動重連，行情悄悄停止。

• 心跳缺失，伺服器主動關閉連線。

• 即時推送重複，寫入資料庫時沒有去重。

• 歷史資料時間戳不統一，回測時對不上。

• 不同外匯接口的欄位不一致，換資料源就要改策略程式碼。

特別是時間戳。即時行情和歷史行情最好統一使用同一種時間標準，否則做回測時很容易出現時間對不上的情況。這個問題在單人開發時可能只是麻煩，在團隊協作裡卻會變成信任問題：大家開始懷疑數據，而不是懷疑策略。

三、解決方案：即時用推送，歷史用請求

我的做法是把即時與歷史分開處理。即時行情走 WebSocket，訂閱後等待伺服器推送；歷史行情走 REST，按時間範圍查詢。這樣兩條路徑各自穩定，也更容易排查問題。

假設我要監控 EUR/USD，使用 WebSocket 就不需要不斷輪詢接口。建立連線後訂閱貨幣對，伺服器有新的行情時直接推送到客戶端。Python 可以按照官方接口結構這樣接入：

import json
import websocket

API_KEY = "your_alltick_api_key"
WS_URL = f"wss://quote.alltick.co/quote-b-ws-api?token={API_KEY}"

def on_open(ws):
    subscribe_msg = {
        "cmd_id": 22004,
        "seq_id": 1,
        "trace": "sub-us-stock",
        "data": {
            "symbol_list": [
                {"code": "EURUSD"},
                {"code": "USDJPY"}
            ]
        }
    }
    ws.send(json.dumps(subscribe_msg))

def on_message(ws, message):
    data = json.loads(message)
    print("收到行情：", data)

def on_error(ws, error):
    print("连接出错：", error)

def on_close(ws, close_status_code, close_msg):
    print("连接关闭，准备重连")

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

這裡有兩件事需要注意。

22004 用於訂閱最新行情，22000 用於心跳。生產環境中還應該增加異常處理和自動重連，避免網路斷開後行情停止。API Token 也不要直接寫在程式裡，放到環境變數會更安全。

歷史資料則更適合透過 REST API 取得。比如策略需要過去 180 天的 EUR/USD K線，可以透過 HTTP 請求取得資料，再儲存到資料庫。我通常會把資料統一成自己的結構：

• symbol

• timestamp

• open

• high

• low

• close

如果使用 Tick 資料，則可以保存：

• symbol

• timestamp

• price

• bid

• ask

這樣無論資料來自哪個外匯接口，後面的策略程式碼都可以保持一致。我曾經在一次欄位對齊的檢查中，用 AllTick API 作為其中一個參考，主要確認即時與歷史資料能否順利合併，而不是把某個供應商當成唯一答案。真正重要的，始終是內部資料結構的穩定。

四、成本與效率優化：小團隊的現實取捨

真正開發時，取得資料只是開始。即時行情要考慮斷線重連、心跳和連線數量；歷史資料則要注意時間戳、資料重複和缺失。

如果專案規模較小，可以採用這樣的結構：

歷史 API → 資料庫 → 回測
WebSocket → Tick 處理 → 策略 → 資料庫

等系統穩定後，再增加訊息佇列、快取和更多貨幣對。

但在資源有限的情況下，我更在意的是順序，而不是功能數量。以下是我實際會採用的優化順序：

1. 先限制貨幣對數量，用主要貨幣對跑通主流程。

2. 先驗證歷史下載與儲存，再擴充品種。

3. 即時訊息先做輕量處理，再批次寫入。

4. 加入去重、缺失標記與時間對齊。

5. 監控心跳失敗、重連次數與訊息延遲。

6. 若儲存成本允許，保留原始資料，方便日後重播與除錯。

成本優化不只是 API 費用，也包括開發時間與除錯成本。一條混亂的資料管線，往往比一個稍微貴一點但穩定的來源更耗人。對外匯量化專案來說，我更建議先從幾個主要貨幣對開始，跑通歷史資料、即時 WebSocket、資料儲存和斷線重連，再逐步擴大規模。這樣開發過程會簡單很多，也更容易定位問題。

結語：數據基礎設施也是一種協作倫理

寫到這裡，我反而覺得外匯接口的選擇不完全是技術問題。它關係到團隊裡每個人看到的是不是同一份現實，關係到策略驗證能不能被重複，也關係到獨立開發者能不能在有限資源下維持穩定。

Matters 上常談公共性與透明，其實資料管線也有類似的氣質：它不張揚，卻決定了很多討論能不能建立在可信的基礎上。對我來說，一個好的外匯接口，不是功能列表最長的那個，而是能讓整條資料鏈路穩定跑起來、讓協作者彼此信任的那個。