如何通过python-binance获取并维护完整的币安期货订单簿？

之前通过API拉取订单簿因请求频率过高导致IP被封，改用Websocket增量更新是正确方向。下面结合你已有的代码，一步步实现订单簿的完整维护逻辑。

一、先明确核心字段含义 #

先理清币安文档里的关键字段，避免后续逻辑混乱：

• lastUpdateId：订单簿快照的最后更新ID，作为初始基准
• u：Websocket事件的最终更新ID
• U：Websocket事件的起始更新ID
• pu：Websocket事件的前序更新ID（必须等于上一个事件的u）
• b/a：事件中的买卖档位数据，格式为[价格字符串, 数量字符串]

二、订单簿维护的核心规则落地 #

币安的维护规则可以拆解为以下可执行步骤：

1. 初始化本地订单簿：先通过API获取快照，作为初始状态，记录lastUpdateId
2. 过滤无效事件：丢弃所有u < lastUpdateId的事件（更新时间早于快照，无效）
3. 校验首个有效事件：第一个处理的事件必须满足U <= lastUpdateId <= u（确保覆盖快照到当前的所有更新）
4. 校验事件连续性：后续每个事件的pu必须等于上一个事件的u，否则说明断更，重新拉取快照
5. 更新本地档位：事件中的数量是绝对数值，数量>0则更新该档位；数量=0则移除该档位（不存在的档位直接忽略即可）

三、完整代码实现 #

下面是整合了快照获取、Websocket连接、订单簿维护的完整异步代码：

import asyncio
from binance import AsyncClient, BinanceSocketManager

# 替换为你的API密钥
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"

# 封装本地订单簿的状态与更新逻辑
class OrderBook:
    def __init__(self):
        self.bids = {}  # 买盘：{价格(float): 数量(float)}
        self.asks = {}  # 卖盘：{价格(float): 数量(float)}
        self.last_update_id = None  # 快照/最后一次有效事件的更新ID
        self.previous_event_u = None  # 上一个有效事件的u值

    def init_from_snapshot(self, snapshot):
        """用API快照初始化本地订单簿"""
        self.last_update_id = snapshot['lastUpdateId']
        # 转换快照数据为字典格式，方便快速更新
        self.bids = {float(price): float(qty) for price, qty in snapshot['bids']}
        self.asks = {float(price): float(qty) for price, qty in snapshot['asks']}
        self.previous_event_u = None

    def update_from_event(self, event):
        """用Websocket事件更新订单簿，返回是否更新成功（失败需重新拉快照）"""
        event_u = event['u']
        event_U = event['U']
        event_pu = event['pu']

        # 规则1：丢弃早于快照的事件
        if event_u < self.last_update_id:
            return True
        
        # 规则2：校验首个有效事件的覆盖范围
        if self.previous_event_u is None:
            if not (event_U <= self.last_update_id <= event_u):
                return False
        else:
            # 规则3：校验事件连续性
            if event_pu != self.previous_event_u:
                return False

        # 更新买盘档位
        for price_str, qty_str in event['b']:
            price = float(price_str)
            qty = float(qty_str)
            if qty > 0:
                self.bids[price] = qty
            else:
                self.bids.pop(price, None)  # 数量为0则移除，不存在则忽略

        # 更新卖盘档位
        for price_str, qty_str in event['a']:
            price = float(price_str)
            qty = float(qty_str)
            if qty > 0:
                self.asks[price] = qty
            else:
                self.asks.pop(price, None)

        # 更新状态
        self.previous_event_u = event_u
        self.last_update_id = event_u
        return True

async def get_futures_order_book_snapshot(client, symbol):
    """获取期货订单簿快照"""
    return await client.futures_order_book(symbol=symbol, limit=1000)

async def run_depth_stream(socket_manager, symbol, order_book):
    """运行深度流并维护订单簿"""
    print(f"启动{symbol}深度流连接")
    depth_socket = socket_manager.depth_socket(symbol=symbol)
    
    async with depth_socket as ds:
        while True:
            event = await ds.recv()
            # 尝试更新订单簿
            update_success = order_book.update_from_event(event)
            
            if not update_success:
                print(f"{symbol}订单簿更新中断，重新获取快照")
                # 重新拉取快照并初始化
                snapshot = await get_futures_order_book_snapshot(socket_manager._client, symbol)
                order_book.init_from_snapshot(snapshot)
                print(f"{symbol}快照重新加载完成，lastUpdateId: {order_book.last_update_id}")
            
            # 示例：打印当前最优买卖价
            best_bid = max(order_book.bids.keys()) if order_book.bids else 0
            best_ask = min(order_book.asks.keys()) if order_book.asks else 0
            print(f"\r{symbol} 买一: {best_bid:.4f} | 卖一: {best_ask:.4f}", end="")

async def main():
    try:
        # 创建异步客户端
        client = await AsyncClient.create(api_key=API_KEY, api_secret=API_SECRET)
        socket_manager = BinanceSocketManager(client)
        
        symbols = ["MANAUSDT"]
        tasks = []
        
        for symbol in symbols:
            # 初始化订单簿：先拉快照
            order_book = OrderBook()
            snapshot = await get_futures_order_book_snapshot(client, symbol)
            order_book.init_from_snapshot(snapshot)
            print(f"{symbol}初始订单簿加载完成，lastUpdateId: {order_book.last_update_id}")
            
            # 添加深度流任务
            tasks.append(run_depth_stream(socket_manager, symbol, order_book))
        
        await asyncio.gather(*tasks)
    
    except KeyboardInterrupt:
        print("\n程序被手动中断")
    
    finally:
        await client.close_connection()

if __name__ == '__main__':
    asyncio.run(main())

四、关键注意事项 #

• API权限：确保你的API密钥拥有期货市场的只读权限（无需交易权限）
• 数据类型转换：币安返回的价格/数量是字符串，必须转为浮点数才能正确排序和计算
• 断连处理：Websocket自动重连后，需要重新拉取快照（重连后的事件可能丢失连续性）
• 多币种扩展：每个币种单独维护自己的OrderBook实例，避免数据冲突
• 性能优化：如果不需要实时打印，可以去掉最优价打印逻辑，减少资源消耗

内容的提问来源于stack exchange，提问作者trapala85