用 Python 订阅 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"
}