透過 WebSocket 接收即時資料事件

用 API 串流經濟資料事件

每當 Narwhal 記錄到新的資料發布、修訂、行事曆變更或殖利率曲線時，事件串流就會發送一則訊息。你可以選擇國家、資料族和事件類型，每則訊息都帶有 cursor，可用來從中斷處繼續。

1. 準備 Python 和 API 金鑰 用 pip install "websockets>=14" 安裝 Python 的 websockets 套件。14 及以上版本才支援這裡用到的 additional_headers 參數。請把金鑰存放在 NARWHAL_API_KEY 環境變數中。REST、MCP 和各類串流都使用同一把金鑰。你可以取得一把免費金鑰。
2. 連線並發送訂閱 伺服器和應用程式在 WebSocket 握手標頭中發送 Authorization: Bearer YOUR_KEY；串流票證僅用於 Narwhal 網站。讀取 ready 後，請在 5 秒內把 subscribe 作為第一個封包發送。該腳本請求美國和墨西哥的價格與勞動力事件。空清單表示選擇該篩選條件的所有值，因此 "event_types": [] 會接收所有事件類型。
   Python（websockets）：訂閱美國和墨西哥的價格與勞動力事件

   import asyncio, json, os
   import websockets
   
   
   async def main(cursor=None):
       headers = {"Authorization": f"Bearer {os.environ['NARWHAL_API_KEY']}"}
       async with websockets.connect(
           "wss://api.narwhalapi.com/v1/events/stream", additional_headers=headers
       ) as ws:
           print(json.loads(await ws.recv()))  # ready
           subscribe = {
               "action": "subscribe",
               "countries": ["USA", "MEX"],
               "families": ["prices", "labour"],
               "event_types": [],
           }
           if cursor:
               subscribe["cursor"] = cursor
           await ws.send(json.dumps(subscribe))
           async for message in ws:
               frame = json.loads(message)
               if "cursor" in frame:
                   cursor = frame["cursor"]  # store this to resume after a drop
               print(frame)
   
   
   asyncio.run(main())

3. 讀取訊息並儲存游標 伺服器先回覆 subscribed，然後發送事件物件以及 heartbeat 和 pong 封包，每個都帶有 cursor。請儲存最新的游標。再發送第二則 subscribe 訊息會讓連線以 resubscribe_required 關閉，因此需要更換篩選條件時請重新連線。一次訂閱最多可包含 60 個 countries、12 個 families 和 9 個 event_types，用戶端封包上限為 2,048 位元組。

串流錯誤以 {"type":"error","code":"..."} 的形式到達。錯誤碼有 stream_limit、rate_limited、subscribe_required、quota_unavailable、unauthorized、slow_consumer、feed_unavailable、revoked、resubscribe_required、command_limit、invalid_command、invalid_frame、cursor_expired、cursor_future 和 shutdown。

串流預測機率

預測串流會針對你選擇的主題或事件發送 Kalshi 和 Polymarket 的機率。訂閱最多可包含 40 個 topics 和 200 個 event_ids。伺服器先發送每個事件的 snapshot，之後在機率變化時發送 update，每個事件每分鐘最多一次；事件關閉或結算時，會發送 status 訊息。

import asyncio, json, os
import websockets


async def main():
    headers = {"Authorization": f"Bearer {os.environ['NARWHAL_API_KEY']}"}
    async with websockets.connect(
        "wss://api.narwhalapi.com/v1/predictions/stream", additional_headers=headers
    ) as ws:
        print(json.loads(await ws.recv()))  # ready
        subscribe = {"action": "subscribe", "topics": ["usa-fed"], "event_ids": []}
        await ws.send(json.dumps(subscribe))
        async for message in ws:
            frame = json.loads(message)
            if frame["type"] in ("snapshot", "update", "status"):
                print(frame)


asyncio.run(main())

heartbeat 每 30 秒到達一次。重新連線會得到新的快照，不會補發舊訊息，因此連線中斷後只需重新訂閱。

哪種方式適合你的任務

連線中斷會怎樣？

對於事件串流，請帶退避地重新連線，並在 subscribe 訊息中帶上你儲存的最後一個游標，串流會重送該游標之後的所有事件。cursor_expired 表示游標已超出保留的歷史範圍：請不帶游標重新開始，並根據目前資料重建。cursor_future 表示游標領先於日誌，請使用伺服器傳回的游標。

對於預測串流，請重新連線並再次訂閱。你會收到每個事件的新快照，但斷線期間發生的變化不會補發。

為什麼我在 5 秒後被中斷了？

如果連線在 ready 之後 5 秒內沒有發送 subscribe 訊息，事件串流會以 subscribe_required 關閉它。停止讀取的用戶端會以 slow_consumer 關閉。每個腳本中的接收迴圈在訂閱之後仍會持續讀取。

串流的費用是多少？

每個連線會從你的額度中扣除 1 次呼叫，每個帳戶都可以使用串流。保持一整天的連線，費用也只是同樣的 1 次呼叫。當帳戶已達到允許的最大並行串流連線數時，新連線會收到 stream_limit 並被關閉。每次呼叫的價格請參閱定價。

範例：墨西哥勞動力資料發布

在 Python 腳本的訂閱設定下，2026 年 9 月 25 日記錄的一次墨西哥勞動力資料發布會以下面的事件物件到達。type 為 dataset.published，resource_id 與 resource_version 標識所發布的那次資料。請在處理事件之前先儲存 cursor，這樣重新連線時就能從它之後繼續。

{
  "cursor": "1",
  "event_id": "2cdf117d-0000-4000-8000-000000000001",
  "type": "dataset.published",
  "country": "MEX",
  "family": "labour",
  "resource_id": "775d11a3-0000-4000-8000-000000000001",
  "resource_version": "822e107c-0000-4000-8000-000000000001",
  "recorded_at": "2026-09-25T15:44:56Z"
}