用Python构建可靠的Polymarket WebSocket行情监控器

交易机器人要做出正确的执行决策，离不开及时的市场数据。但连接上WebSocket、收到消息只是第一步——连接会断、消息会延迟，机器人还可能在重连后继续使用一份过时的订单簿快照。这些隐患恰恰是实盘环境中最容易被忽视的致命细节。

本文围绕Polymarket公开的市场WebSocket，用Python搭建一个小型监控器。它的核心不在于下单执行，而在于三件事：消费行情更新、判断数据何时不再可信、以及在连接失败后重建状态。

连接Polymarket市场WebSocket

Polymarket提供了一个公开的市场频道，用于推送实时订单簿和成交更新。官方文档中记录的端点为：

wss://ws-subscriptions-clob.polymarket.com/ws/market

市场频道通过资产ID（也称token ID）订阅。每个结果（outcome）都有自己独立的token ID，因此你需要为想监控的那个结果准备对应的ID。官方的Polymarket WebSocket文档描述了订阅格式与事件类型。

安装Python的WebSocket库：

pip install websockets

本文示例刻意使用原始WebSocket协议，而非把连接生命周期隐藏在某个大型框架里——目的是让整个连接、订阅、心跳、重连的流程清晰可见，便于理解每一步在发生什么。

Polymarket是一个基于区块链的预测市场平台，运行在Polygon网络上。用户可以针对现实世界事件的结果买卖份额，价格在0到1美元之间波动，代表该结果发生的隐含概率。每个市场通常有两个或多个结果（outcome），每个结果对应一个独立的ERC-1155代币，即token ID。例如，"某候选人赢得选举"是一个结果，"某候选人未赢得选举"是另一个结果，它们的token ID各不相同。

Polymarket的订单撮合通过一个名为CLOB（Central Limit Order Book，中央限价订单簿）的链下撮合引擎处理，成交后才结算到链上。WebSocket端点URL中的clob正源于此。这意味着行情数据反映的是链下订单状态，而非链上实时状态，订阅时需要使用CLOB层面的token ID而非合约地址。

读懂你收到的消息

对一个基础监控器来说，有三种消息类型最值得关注。

book 提供一份订单簿快照，含买卖各档位。它用于初始化或重建本地的市场状态。

price_change 报告价格档位的变动，其price_changes数组可能包含不同资产、不同档位的更新。该载荷还包含best_bid和best_ask字段，方便做盘口（top-of-book）监控。

last_trade_price 报告一笔已发生的成交。需要注意的是，最新成交价并不等于你下一笔订单能够成交的价格。

这里有一个关键区分：收到一条增量价格更新，并不意味着你本地对整个订单簿的视图已经被重建。重连之后尤其如此——单条价格更新无法替代一份完整快照的作用。

订单簿（Order Book）是一份按价格排列的买单与卖单列表。买单（bid）代表愿意买入的最高报价，卖单（ask）代表愿意卖出的最低报价。最优买价（best bid）和最优卖价（best ask）分别是当前买卖两侧的最佳报价，两者之差称为价差（spread）。

快照（snapshot）与增量更新（incremental update）是两种不同的数据形式：快照描述某一时刻的完整状态，增量更新只描述相对于上一状态的变化。book消息属于快照，price_change属于增量更新。增量更新依赖于一个正确的初始基准状态——如果基准本身已过时或缺失，后续叠加的增量将产生错误的结果。这正是文章反复强调"必须先收到新鲜快照才能信任增量更新"的根本原因。

构建带过时保护的监控器

下面的监控器只跟踪一个结果token。它从「未就绪」状态启动，订阅市场、发送必需的应用层心跳，只有在收到新鲜的book快照之后才将数据标记为就绪。

如果连接断开，监控器会以指数退避方式重试。每一次新连接都从未就绪状态开始，从而防止应用把旧快照当作当前数据来使用。

import asyncio
import json
import time
import websockets

WS_URL = "wss://ws-subscriptions-clob.polymarket.com/ws/market"
ASSET_ID = "YOUR_OUTCOME_TOKEN_ID"

class MarketMonitor:
    def __init__(self):
        self.best_bid = None
        self.best_ask = None
        self.book_ready = False
        self.last_message_at = None

    def reset(self):
        self.best_bid = None
        self.best_ask = None
        self.book_ready = False
        self.last_message_at = None

    def handle(self, message):
        event_type = message.get("event_type")
        self.last_message_at = time.monotonic()

        if event_type == "book":
            if message.get("asset_id") != ASSET_ID:
                return
            bids = [float(l["price"]) for l in message.get("bids", []) if float(l["size"]) > 0]
            asks = [float(l["price"]) for l in message.get("asks", []) if float(l["size"]) > 0]
            self.best_bid = max(bids) if bids else None
            self.best_ask = min(asks) if asks else None
            self.book_ready = True
            print("Fresh book snapshot received.")

        elif event_type == "price_change":
            for change in message.get("price_changes", []):
                if change.get("asset_id") != ASSET_ID:
                    continue
                if not self.book_ready:
                    continue
                if "best_bid" in change:
                    self.best_bid = float(change["best_bid"])
                if "best_ask" in change:
                    self.best_ask = float(change["best_ask"])

        if self.book_ready:
            print(f"Bid: {self.best_bid} | Ask: {self.best_ask} | Spread: {self.spread()}")

    def spread(self):
        if self.best_bid is None or self.best_ask is None:
            return None
        return round(self.best_ask - self.best_bid, 6)

需要特别强调：这是一个教学示例，而非完整的交易数据服务。它用best_bid和best_ask做盘口监控，并不维护一份随每次增量变化而更新的完整本地订单簿。生产实现应当正确应用价格档位更新、校验消息、跟踪数据时间戳，并对照官方文档测试当前协议。

为什么监控器要重置状态

设想一个场景：连接断开时最优卖价是 $0.48，断线期间市场移动到了 $0.53。如果机器人重连后继续沿用旧的 $0.48，策略可能误以为一个有吸引力的入场点仍然存在，而实际上它早已消失。

重置状态正是为了避免把这个旧值当作当前值。监控器会等待一份新的book快照，才宣布它的盘口视图「就绪」。这一原则同样应当贯穿到交易机器人的决策层：

• 市场数据过时时不要提交新交易；
• 不要假设重连成功就等于本地订单簿已同步；
• 在信任增量更新之前，先用新鲜快照重建状态；
• 持仓与挂单的对账，要与公开市场数据分开独立处理。

一条市场数据连接并不能告诉你自己的订单是否被接受或成交。账户与订单状态需要有自己可靠的跟踪流程。

显式检测过时数据

一个处于connected状态的socket，并不保证策略需要的数据是新鲜的。

生产级监控器应当记录最后一次有效行情更新的时间，并定义一个可接受的最大数据年龄。如果在这段时间内没有合格的更新到达，就应把数据流标记为stale，并阻止依赖它的交易决策。这个阈值要贴合应用本身——一个仪表盘可以容忍更长延迟，而一个想捕捉微小价格波动的策略则不行。

同样重要的是，要区分「安静的市场」和「断掉的连接」。没有价格变动并不必然意味着数据流不健康。心跳响应、连接状态、服务器时间戳、应用层新鲜度检查，这些应当被综合考量。

Polymarket文档规定的市场与用户频道都要求每十秒发送一次PING，服务器以PONG回应。因此示例中禁用了WebSocket库自带的ping帧（ping_interval=None），改为发送协议要求的应用层心跳。

应用层心跳（application-level heartbeat）与WebSocket协议层的ping/pong帧是两个不同的机制。WebSocket标准协议内置了ping/pong帧，由客户端或服务器发送以检测连接是否存活，通常由WebSocket库自动处理。而Polymarket要求的PING是一个应用层消息——即以文本或JSON形式发送的自定义消息，与协议层ping帧无关。两者可以同时存在，但作用不同：协议层ping检测TCP连接是否断开，应用层ping则验证服务端业务逻辑是否仍在响应。示例代码中将库的ping_interval设为None并手动发送应用层PING，正是为了满足Polymarket协议的具体要求，同时避免库的自动ping与应用层心跳产生冲突。

投入实盘前还需要完善什么

一个可靠的生产实现需要更多的保障措施：

完整订单簿维护：book事件提供快照，后续的price_change事件会增加、更新或移除价格档位。本地订单簿必须正确应用这些变化，包括size变为零的更新。

快照与更新的一致性：如果应用丢消息或发生重连，应当重建订单簿，而不是假设之前的状态仍然有效。

结构化日志：记录连接尝试、重连、数据过时事件、畸形消息和快照恢复，同时避免记录私密凭证。

背压处理：当更新到达速度快过应用处理速度时，无界队列会增加内存占用并拖慢决策。应测量处理延迟，并选择与策略匹配的队列策略。

执行安全阀：下单前要校验市场数据的年龄、价格上限、可用余额和组合敞口。仅凭数据新鲜，并不能让一笔交易变得安全。

测试：模拟断连、畸形消息、空订单簿、缺失字段和延迟更新，验证机器人在市场状态未就绪或过时时拒绝行动。

背压（backpressure）是流处理系统中的一个核心概念，描述下游消费者处理速度跟不上上游生产者发送速度时产生的压力积累。在WebSocket行情场景中，如果行情更新的到达速率超过策略逻辑的处理速率，消息会在内存队列中堆积。无界队列（unbounded queue）意味着队列长度没有上限，极端情况下会耗尽内存。更隐蔽的问题是决策延迟：机器人可能正在处理数秒前的消息，以为在响应当前市场，实际上早已滞后。常见的应对策略包括：设置队列最大长度并在满队时丢弃旧消息、只保留每个资产的最新一条消息、或在处理前检查消息时间戳以跳过已过期数据。

小结

Polymarket WebSocket监控器远不只是一个「连接并打印价格」的脚本。它是一个有状态的组件，必须清楚自己的数据何时有效、何时已过时、何时需要重建对市场的视图。

从快照恢复、心跳处理、新鲜度检查和带上限的重连退避入手，能为市场数据系统打下更扎实的基础。Dexoryn Labs维护了一个公开的Polymarket跟单机器人仓库，供评估现有自动化方案的开发者参考。

可靠的市场数据不保证盈利——它的价值在于防止机器人基于一份它本不该再信任的信息做出决策。本文仅作教育用途，自动化交易涉及财务风险，在接入实盘下单逻辑之前，请务必在非交易环境中充分测试。