WebSocket でリアルタイムにイベントを受信

API で経済イベントをストリーミングする

イベントストリームは、Narwhal が新しいデータ発表、改定、カレンダーの変更、利回り曲線を記録するたびにメッセージを送ります。国、データファミリー、イベントタイプを選べます。どのメッセージにも cursor が付くので、途切れたところから再開できます。

1. Python と API キーを準備する Python の websockets パッケージを pip install "websockets>=14" でインストールします。ここで使う additional_headers 引数はバージョン14以降が対応しています。キーは 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 が付きます。最新のカーソルを保存してください。2回目の subscribe メッセージを送ると、接続は resubscribe_required で閉じられるため、フィルターを変えたいときは再接続します。購読できるのは、countries が最大60、families が最大12、event_types が最大9で、クライアントのフレームは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 のオッズを送ります。購読できるのは topics が最大40、event_ids が最大200です。サーバーは各イベントの snapshot を送り、オッズが変わると update を送ります。更新はイベントごとに1分に1回までです。イベントが終了または確定すると 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回分の呼び出しが引かれます。どのアカウントでもストリーミングできます。1日中開いたままの接続でも、費用は同じ1回分です。アカウントが開ける接続数の上限にすでに達している場合、新しい接続は stream_limit を受け取って閉じられます。1回あたりの価格は料金を参照してください。

計算例：メキシコの労働統計の発表

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"
}