Marico's space
最近折腾 Polymarket 交易机器人，实时行情、订单簿同步和低延迟这几个坑踩得不算少。这篇把问题说清楚：WebSocket（WebSocket 协议）、本地订单簿和异步执行，到底是怎么拼成一套可用的做市系统。
先说结论：交易策略的上限，往往不是由策略本身决定，而是由它拿到的数据、执行链路和状态管理能力决定。
假设你在 Polymarket 上运行一个自动化交易机器人，监控 BTC 5 分钟 Up 或 Down 市场。
机器人发现了一个机会。当前最优买价为 $0.49，最优卖价为 $0.51。它根据策略计算出新的买卖价格，准备下单。
但问题来了。
就在机器人计算下一步动作时，另一个交易者提交了一笔大单。订单簿发生变化，流动性转移，市场也跟着动了。
此时机器人作出的决策，依据的可能是已经过时的信息。
这是构建自动化做市商时最根本的难题之一。
只写出一个能算出理想价格的策略远远不够。基础设施还必须持续接收市场信息、处理更新、维护准确的订单簿状态，并可靠地执行决策。
下面会用 Python、WebSocket、异步编程和本地维护的订单簿，把这套基础设施拆开讲清楚。重点不只是用了哪些技术，而是每个组件为什么必须存在。
先从一个简单问题开始：交易机器人怎么知道市场变了？
最常见的方式，是反复通过 REST API（应用程序接口）请求最新市场信息。
比如：
Bot: What is the current market price? Exchange: $0.50 Bot: What is the current market price? Exchange: $0.51 Bot: What is the current market price? Exchange: $0.49
这种方式叫轮询。
应用不断向交易所询问最新信息。
轮询适合很多场景，例如查询账户、拉取历史数据和管理订单。
但如果把高频轮询作为实时行情更新的主要机制，就会遇到几个问题。
无效的 API 请求
即使市场完全没有变化，机器人也可能持续发请求。
反应延迟
一次轮询刚结束，市场可能立刻发生变化。应用要等到下一次请求才能发现变化。
额外的基础设施开销
高频请求会消耗网络资源，还会制造没必要的处理压力。
对于需要跟随流动性变化的做市系统，这些限制不是小问题。
机器人不必反复询问交易所，而是与交易所建立一条持久的 WebSocket 连接。
通信模型会变成这样：
REST Polling: Bot → Request → Exchange
Bot ← Response ← Exchange Bot → Request → Exchange
Bot ← Response ← Exchange WebSocket: Bot ←──── Persistent Connection ────→ Exchange Market update Market update Market update Market update
连接建立并完成订阅后，交易所会主动推送支持的市场更新。
这种方式特别适合监听订单簿变化和其他实时事件。
WebSocket 并不会消除网络延迟，但省掉了轮询带来的重复请求开销。
对于做市商来说，这就是很明确的架构优势。
写代码之前，先看看完整系统应该怎么组织。
做市机器人承担多项职责。如果全塞进一个 Python 大脚本里，维护难度很快就会失控。
更合理的做法，是把系统拆成彼此独立的组件。
一个简化架构如下：
POLYMARKET EXCHANGE | +----------+----------+ | | v v Market Data Execution Events WebSocket / API | | v v Market Data Handler Portfolio Manager | v Local Order Book | v Strategy Engine | v Risk Manager | v Execution Engine | v Order Management | v POLYMARKET
每个组件的职责如下。
Market Data Handler
接收传入的市场消息，校验消息，并转换成内部事件。
Local Order Book
为每个 token 维护当前已知的买盘、卖盘和可用流动性。
Strategy Engine
分析市场状态，计算候选买价和卖价。
Risk Manager
检查拟提交的订单是否符合库存、敞口、价格和资金限制。
Execution Engine
协调订单提交、替换和撤销。
Portfolio Manager
跟踪持仓、余额、成交，以及已实现或未实现盈亏（P&L）。
有一个非常重要的架构原则：市场数据和执行信息必须分开处理。
公开订单簿发生变化，不代表你的订单已经成交。
同样，订单提交成功，也不代表交易所已经接受或者执行了它。
每一种事件都要按照真实含义分别处理。
接下来从架构进入实现。
Python 提供了不少适合异步网络通信的工具。
这个示例会用到：
asyncio：执行异步任务。websockets：处理 WebSocket 通信。json：解析传入消息。安装依赖：
pip install websockets
一个简化的 WebSocket 客户端如下：
import asyncio
import json
import websockets async def listen_to_market(): websocket_url = "WSS_ENDPOINT_FROM_CURRENT_DOCS" async with websockets.connect( websocket_url, ping_interval= 20, ping_timeout= 20 ) as websocket: print("Connected to market data server") subscription = { "type": "subscribe", "channel": "market" } await websocket.send( json.dumps(subscription) ) async for message in websocket: data = json.loads(message) print("Received:", data) asyncio.run(listen_to_market())
这里有三个关键动作。
第一步，建立持久连接。
第二步，发送订阅消息，订阅目标市场。
第三步，持续监听传入消息。
关键区别在于：机器人不必反复请求最新市场信息，而是等待更新主动推送过来。
实现说明：上面的端点和订阅数据只是示例占位符。实际使用的 WebSocket 端点、频道、订阅格式和消息结构，必须以当前 Polymarket 官方文档为准。
实时交易系统不会只做一件事。
它可能同时执行：
如果这些操作使用阻塞式网络调用，其中一个慢操作就可能拖住其他任务。
这正是 Python asyncio 能派上用场的地方。
异步编程允许多个任务在其他任务等待网络操作时继续推进。
看这个例子：
import asyncio async def market_data_task(): while True: print("Listening to market data") await asyncio.sleep(1) async def order_monitor_task(): while True: print("Monitoring orders") await asyncio.sleep(2) async def portfolio_task(): while True: print("Updating portfolio") await asyncio.sleep(3) async def main(): await asyncio.gather( market_data_task(), order_monitor_task(), portfolio_task() ) asyncio.run(main())
这里有三个异步任务运行在同一个事件循环中。
一个任务等待时，其他任务还能继续执行。
这种模式特别适合交易机器人这类网络密集型应用。
不过，异步编程不代表任何场景下都会更快。
CPU 密集型计算依然可能阻塞事件循环。这类任务需要单独优化，或者放到独立进程中执行。
目标不是盲目追求异步，而是消除无意义的等待，让关键操作始终保持响应。
收到市场更新，只是第一步。
交易策略需要一种结构化方式描述当前市场。
这正是本地订单簿的用途。
订单簿保存某个 token 的买入需求和卖出需求。
假设当前市场如下。
Bid side
| Price | Quantity |
|---|---|
| $0.49 | 100 |
| $0.48 | 200 |
| $0.47 | 150 |
Ask side
| Price | Quantity |
|---|---|
| $0.51 | 120 |
| $0.52 | 180 |
| $0.53 | 100 |
最优买价是 $0.49。
最优卖价是 $0.51。
中间价为：
[
Midpoint = \frac{0.49 + 0.51}{2} = 0.50
]
这个中间价可以作为做市策略的一项输入。
但订单簿一直在变。
假设另一个交易者在 $0.50 提交买单，最优买价会立即改变。
策略应该能够直接获取这个更新，而不需要每次都重新请求完整订单簿。
本地订单簿的解决方案，就是在应用内部持续维护市场状态。
class LocalOrderBook: def __init__(self): self.bids = {} self.asks = {} def update_bid(self, price, quantity): if quantity <= 0: self.bids.pop(price, None) else: self.bids[price] = quantity def update_ask(self, price, quantity): if quantity <= 0: self.asks.pop(price, None) else: self.asks[price] = quantity def best_bid(self): return max(self.bids) if self.bids else None def best_ask(self): return min(self.asks) if self.asks else None def midpoint(self): bid = self.best_bid() ask = self.best_ask() if bid is None or ask is None: return None return (bid + ask) / 2
这个实现只用于展示基本原理。
生产环境还要处理初始快照、增量更新、可用的序列号、同步机制、token 标识，以及交易所精确的消息语义。
如果本地订单簿已经不一致，策略就不能继续把它当作可靠行情。
没有本地订单簿，策略只能依赖外部请求理解市场。
有了同步的本地订单簿，策略可以直接从内存读取最新且经过校验的市场状态。
这样既提高效率，也让系统结构更清晰。
这里还有一个关键设计决策。
WebSocket 处理器要不要在收到消息时直接计算策略并提交订单？
我不建议这么做。
想象一个市场在短时间内连续产生数百次更新。
如果每条消息都会触发策略计算、风险检查、订单提交和数据库操作，WebSocket 处理器很快就会成为瓶颈。
更稳妥的办法是拆分职责：
WebSocket | v
Message Parser | v
Local Order Book | v
Event Queue | v
Strategy Worker | v
Risk Manager | v
Execution Worker
WebSocket 处理器只负责接收和处理市场信息。
策略工作器负责分析市场状态。
执行工作器负责与交易所通信。
这些组件可以通过异步队列连接起来。
例如：
import asyncio async def market_data_producer(queue): while True: market_update = { "midpoint": 0.50 } await queue.put(market_update) await asyncio.sleep(1) async def strategy_consumer(queue): while True: update = await queue.get() try: midpoint = update["midpoint"] print( "Calculating quotes:", midpoint ) finally: queue.task_done() async def main(): queue = asyncio.Queue(maxsize= 1000) await asyncio.gather( market_data_producer(queue), strategy_consumer(queue) ) asyncio.run(main())
这是一个生产者—消费者架构。
生产者生成市场更新，消费者处理更新。
使用有界队列很重要，它能避免消费速度跟不上时内存无限增长。
在真实交易系统里，队列溢出必须触发明确的处理策略。
部分冗余的市场状态更新可以安全合并，但重要的订单和成交事件绝不能被静默丢弃。
做市机器人最危险的处境之一，就是拿着过时信息继续交易。
看这个例子：
Last valid market update: 10:00:00 Current time: 10:00:10 Data age: 10 seconds
机器人可能仍然认为中间价是 $0.50。
但真实市场可能已经发生了明显变化。
使用过期行情的策略会生成不合理的报价，让账户承担没有必要的市场风险。
一个简单的新鲜度检查器如下：
import time class MarketDataHealth: def __init__(self, max_age_seconds= 3): self.max_age_seconds = max_age_seconds self.last_update = None def record_update(self): self.last_update = time.monotonic() def is_fresh(self): if self.last_update is None: return False age = ( time.monotonic() - self.last_update ) return age <=