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가 있습니다. 최신 커서를 저장하세요. 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분에 한 번까지입니다. 이벤트가 마감되거나 확정되면 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로 닫힙니다. 각 스크립트의 수신 루프는 구독한 뒤에도 계속 읽습니다.

스트리밍 비용은 얼마인가요?

연결 한 번에 크레딧에서 호출 한 번이 차감되며, 모든 계정이 스트리밍할 수 있습니다. 하루 종일 열어 둔 연결도 같은 한 번의 호출 비용입니다. 계정이 열 수 있는 스트림 연결 수가 최대인 상태에서 새 연결을 시도하면 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"
}