结论摘要
本文将深入探讨订单簿（Order Book）微观结构的核心建模方法，重点解析信息非对称性如何影响价格发现机制。我将从限价订单簿的数学建模出发，结合 Python 代码实现真实市场数据的处理流程，并分享使用
HolySheep AI API 构建订单簿分析系统的实战经验。订单簿微观结构是理解市场深度、流动性分布和信息传递机制的基础，适合量化交易员、风险分析师和对高频数据建模感兴趣的开发者。
订单簿微观结构基础概念
订单簿是金融市场上买卖双方挂单信息的实时记录，包含价格、挂单量、订单时间等维度。从微观结构视角看，订单簿反映了市场参与者的供需博弈，是价格发现的直接载体。
订单簿的核心数据结构
一个标准的订单簿包含以下关键字段：
- bid_price / ask_price：买方报价（买一~买十）与卖方报价（卖一~卖十）
- bid_volume / ask_volume：对应价格档位的挂单总量
- order_count：档位内的订单笔数（反映订单分散度）
- timestamp：订单到达时间（高频场景精确到纳秒）
# Python 订单簿数据结构定义
from dataclasses import dataclass
from typing import List, Optional
from collections import deque
import time
@dataclass
class Order:
"""订单对象"""
order_id: str
price: float
volume: float
side: str # 'bid' or 'ask'
timestamp: float
participant_id: Optional[str] = None
@dataclass
class OrderBookLevel:
"""订单簿档位"""
price: float
volume: float
order_count: int
class OrderBook:
"""订单簿模拟器"""
def __init__(self, symbol: str, max_levels: int = 10):
self.symbol = symbol
self.max_levels = max_levels
self.bids: List[OrderBookLevel] = [] # 买单列表（价格从高到低）
self.asks: List[OrderBookLevel] = [] # 卖单列表（价格从低到高）
self.trades: deque = deque(maxlen=1000) # 成交记录
self.mid_price: float = 0.0
self.spread: float = 0.0
self._last_update: float = 0.0
def update_bid(self, price: float, volume: float, order_count: int = 1):
"""更新买单档位"""
# 找到对应价格的位置
for i, level in enumerate(self.bids):
if abs(level.price - price) < 1e-8:
if volume == 0:
self.bids.pop(i)
else:
self.bids[i] = OrderBookLevel(price, volume, order_count)
break
else:
if volume > 0:
self.bids.append(OrderBookLevel(price, volume, order_count))
# 保持价格排序
self.bids.sort(key=lambda x: x.price, reverse=True)
self.bids = self.bids[:self.max_levels]
self._update_market_state()
def update_ask(self, price: float, volume: float, order_count: int = 1):
"""更新卖单档位"""
for i, level in enumerate(self.asks):
if abs(level.price - price) < 1e-8:
if volume == 0:
self.asks.pop(i)
else:
self.asks[i] = OrderBookLevel(price, volume, order_count)
break
else:
if volume > 0:
self.asks.append(OrderBookLevel(price, volume, order_count))
self.asks.sort(key=lambda x: x.price)
self.asks = self.asks[:self.max_levels]
self._update_market_state()
def _update_market_state(self):
"""更新市场状态指标"""
if self.bids and self.asks:
self.mid_price = (self.bids[0].price + self.asks[0].price) / 2
self.spread = self.asks[0].price - self.bids[0].price
self._last_update = time.time()
def get_imbalance(self, levels: int = 5) -> float:
"""
计算订单簿不平衡度
返回值范围 [-1, 1]，正值表示买方压力大，负值表示卖方压力大
"""
bid_vol = sum(level.volume for level in self.bids[:levels])
ask_vol = sum(level.volume for level in self.asks[:levels])
if bid_vol + ask_vol == 0:
return 0.0
return (bid_vol - ask_vol) / (bid_vol + ask_vol)
def get_microprice(self, levels: int = 5, decay_factor: float = 0.9) -> float:
"""
计算微观价格（Microprice）
加权平均价格，考虑订单不平衡度调整
"""
if not self.bids or not self.asks:
return self.mid_price
imbalance = self.get_imbalance(levels)
# Microprice = Mid + imbalance * spread / 2
microprice = self.mid_price + imbalance * self.spread / 2
return microprice
def __repr__(self):
return (f"OrderBook(symbol={self.symbol}, mid={self.mid_price:.2f}, "
f"spread={self.spread:.4f}, bids={len(self.bids)}, asks={len(self.asks)})")
使用示例
book = OrderBook("BTC-USDT")
book.update_bid(42150.5, 2.5, order_count=3)
book.update_bid(42149.0, 1.8)
book.update_ask(42151.0, 3.2, order_count=5)
book.update_ask(42152.5, 1.5)
print(f"订单簿状态: {book}")
print(f"订单不平衡度: {book.get_imbalance():.4f}")
print(f"微观价格: {book.get_microprice():.2f}")
信息非对称性与价格发现机制
信息非对称性的来源
在订单簿微观结构中，信息非对称性主要体现在三个维度：
- 知情交易者（Informed Traders）：拥有私有信息的交易者，通常通过限价单在市场薄弱的边缘下单，等待价格向有利方向移动后被动成交
- 不知情交易者（Liquidity Traders）：以执行套保、资产配置为目的的交易者，对价格冲击不敏感，主要使用市价单
- 做市商（Market Makers）：持续提供流动性的中介，通过买卖价差获取收益，承担信息风险
价格发现的动态过程
知情交易者的订单流会向市场传递私有信息，做市商通过观察订单流的不平衡来调整报价。Glosten 和 Milgrom（1985）提出的价差分解模型指出：
# 信息非对称性建模：价差分解
import numpy as np
from scipy.stats import norm
class AsymmetricInfoModel:
"""
基于 Glosten-Milgrom 模型的价差分解
"""
def __init__(self, prob_informed: float = 0.3, base_value: float = 100.0):
self.p_informed = prob_informed # 知情交易者比例
self.v = base_value # 资产基础价值
self.spread = 0.0
self.adverse_selection = 0.0
self.order_processing = 0.0
self.inventory_cost = 0.0
def compute_spread_components(self, order_flow: int,
volatility: float = 1.0,
inventory_b担忧: float = 0.2) -> dict:
"""
分解价差的三个组成部分
Args:
order_flow: 订单流（正=净买入，负=净卖出）
volatility: 资产波动率
inventory_b担忧: 做市商库存担忧系数
Returns:
包含各组成部分的字典
"""
# 逆向选择成本（知情交易者导致）
# 知情交易者更可能在价值上升时买入，下跌时卖出
prob_buy_informed = 0.5 + 0.5 * np.sign(order_flow)
# 条件期望价值变动
expected_value_change = (
self.p_informed *
volatility *
norm.ppf(prob_buy_informed)
)
# 逆向选择成本 = E[价值变化 | 订单方向]
adverse_selection_cost = self.p_informed * volatility * 0.5
# 订单处理成本（固定成本）
order_processing_cost = 0.01 * self.v
# 库存成本
net_position = order_flow * 0.1 # 假设每单位订单对应0.1单位持仓
inventory_cost = abs(net_position) * inventory_b担忧 * volatility
# 总价差
half_spread = adverse_selection_cost + order_processing_cost + inventory_cost
total_spread = 2 * half_spread
self.adverse_selection = adverse_selection_cost
self.order_processing = order_processing_cost
self.inventory_cost = inventory_cost
self.spread = total_spread
return {
'total_spread': total_spread,
'adverse_selection_cost': adverse_selection_cost,
'order_processing_cost': order_processing_cost,
'inventory_cost': inventory_cost,
'adverse_selection_ratio': adverse_selection_cost / total_spread if total_spread > 0 else 0
}
def estimate_probability_informed(self, trade_indicator: list) -> float:
"""
基于交易数据序列估计知情交易者概率
Args:
trade_indicator: 交易方向序列 [1=买入, -1=卖出, 0=无成交]
Returns:
估计的知情交易者概率
"""
# 使用 PIN（Probability of Informed Trading）模型
buys = sum(1 for t in trade_indicator if t == 1)
sells = sum(1 for t in trade_indicator if t == -1)
no_trades = sum(1 for t in trade_indicator if t == 0)
total = len(trade_indicator)
if total == 0:
return 0.0
# 简化的 PIN 估计
alpha = self.p_informed # 使用先验
delta = sells / (buys + sells) if (buys + sells) > 0 else 0.5
# 订单到达率（简化假设）
mu = 0.1 # 知情交易者到达率
alpha_delta = alpha * delta
alpha_minus = alpha * (1 - delta)
# 计算似然（需要 MLE，这里用简化版本）
pin = alpha * (1 - delta) + alpha * delta # 简化
return pin
def get_optimal_quote(self, position: float,
target_inventory: float = 0.0) -> tuple:
"""
计算做市商最优报价
考虑库存偏离和逆向选择风险
"""
# 库存偏离
inv_adj = (position - target_inventory) * 0.05
# 基础报价
base_bid = self.v - self.spread / 2 - inv_adj
base_ask = self.v + self.spread / 2 - inv_adj
# 考虑订单不平衡的动态调整
imbalance_担忧 = self.get_order_imbalance_concern()
quote_spread = self.spread * (1 + imbalance_担忧)
return (base_bid, base_ask, quote_spread)
def get_order_imbalance_concern(self) -> float:
"""计算订单不平衡担忧因子"""
# 简化：基于知情概率
return self.p_informed * 0.5
实战应用示例
model = AsymmetricInfoModel(prob_informed=0.25, base_value=42150.0)
场景1：净买入订单流
result1 = model.compute_spread_components(order_flow=10, volatility=50.0)
print("=== 净买入场景 ===")
print(f"知情交易者比例: {model.p_informed}")
print(f"总价差: ${result1['total_spread']:.2f}")
print(f" - 逆向选择成本: ${result1['adverse_selection_cost']:.2f} ({result1['adverse_selection_ratio']:.1%})")
print(f" - 订单处理成本: ${result1['order_processing_cost']:.2f}")
print(f" - 库存成本: ${result1['inventory_cost']:.2f}")
场景2：净卖出订单流
result2 = model.compute_spread_components(order_flow=-10, volatility=50.0)
print("\n=== 净卖出场景 ===")
print(f"总价差: ${result2['total_spread']:.2f}")
print(f"逆向选择成本: ${result2['adverse_selection_cost']:.2f}")
价格冲击模型与滑点估算
订单对价格的冲击是量化策略回测中最重要的误差来源之一。线性冲击模型（Almgren-Chriss 模型）给出了价格冲击与订单规模的定量关系：
# 价格冲击模型与最优执行
import numpy as np
from scipy.optimize import minimize
class PriceImpactModel:
"""
价格冲击模型
使用 Almgren-Chriss 框架
"""
def __init__(self,
permanent_impact: float = 0.1,
temporary_impact: float = 0.5,
volatility: float = 2.0):
"""
Args:
permanent_impact: 永久冲击系数（知情程度）
temporary_impact: 临时冲击系数（流动性）
volatility: 资产波动率（日频）
"""
self.gamma = permanent_impact
self.eta = temporary_impact
self.sigma = volatility
def permanent_impact(self, trade_rate: float) -> float:
"""永久冲击：影响后续价格"""
return self.gamma * abs(trade_rate)
def temporary_impact(self, trade_rate: float) -> float:
"""临时冲击：立即执行的滑点"""
return self.eta * trade_rate
def total_impact(self, trade_rate: float) -> float:
"""总冲击"""
return self.permanent_impact(trade_rate) + self.temporary_impact(trade_rate)
def estimate_impact_from_trades(self, price_before: float,
price_after: float,
volume: float) -> dict:
"""从实际交易数据估算冲击系数"""
price_change = abs(price_after - price_before)
normalized_change = price_change / price_before
# 临时冲击 = 短期价格变动 / 交易量
# 永久冲击 = 长期价格变动 - 临时冲击
estimated_total = normalized_change / volume if volume > 0 else 0
return {
'estimated_total_impact': estimated_total,
'price_change_bps': price_change / price_before * 10000,
'volume': volume
}
def optimal_execution(self,
shares: float,
time_horizon: float,
risk_aversion: float,
n_periods: int = 100) -> dict:
"""
Almgren-Chriss 最优执行策略
最小化：执行成本 + 风险惩罚
Args:
shares: 总执行量
time_horizon: 执行时间窗口（天）
risk_aversion: 风险厌恶系数
n_periods: 离散时间点数
Returns:
每个时间点的最优交易量
"""
dt = time_horizon / n_periods
# 协方差矩阵
cov = self.sigma ** 2 * dt * np.eye(n_periods)
# 目标函数：执行路径的惩罚
def objective(x):
# x = 各时间段交易量
# 临时冲击成本
temp_cost = self.eta * np.sum(x ** 2)
# 永久冲击成本
perm_cost = self.gamma * np.sum(x) ** 2
# 库存风险
remaining = shares - np.cumsum(x)
variance = np.sum(np.array([remaining[i:] ** 2 for i in range(n_periods)]) * dt)
risk_penalty = risk_aversion * self.sigma ** 2 * variance
return temp_cost + perm_cost + risk_penalty
# 约束：总量等于 shares
constraints = {'type': 'eq', 'fun': lambda x: np.sum(x) - shares}
bounds = [(0, shares * 2)] * n_periods
# 初始猜测：均匀分配
x0 = np.ones(n_periods) * shares / n_periods
result = minimize(objective, x0, method='SLSQP',
bounds=bounds, constraints=constraints)
optimal_trades = result.x
execution_times = np.linspace(0, time_horizon, n_periods)
return {
'optimal_trades': optimal_trades,
'execution_times': execution_times,
'cumulative_trades': np.cumsum(optimal_trades),
'expected_cost': result.fun,
'execution_cost_per_share': result.fun / shares
}
def simulate_execution(self, shares: float,
impact_coefficient: float = 1.0) -> dict:
"""
模拟不同执行策略的成本
Args:
shares: 执行量
impact_coefficient: 冲击系数调整
"""
strategies = {
'TWAP': np.ones(100) * shares / 100, # 时间加权
'VWAP': np.ones(100) * shares / 100, # 成交量加权
'Aggressive': np.concatenate([
np.ones(20) * shares * 0.6 / 20, # 前60%快速
np.ones(80) * shares * 0.4 / 80 # 后40%缓慢
])
}
results = {}
for name, trades in strategies.items():
temp_cost = self.eta * impact_coefficient * np.sum(trades ** 2)
perm_cost = self.gamma * impact_coefficient * np.sum(trades) ** 2
total_cost = temp_cost + perm_cost
results[name] = {
'total_cost': total_cost,
'cost_bps': total_cost / shares * 10000,
'avg_slippage_bps': total_cost / shares * 10000
}
return results
实战应用
impact_model = PriceImpactModel(
permanent_impact=0.1, # 永久冲击系数
temporary_impact=0.5, # 临时冲击系数
volatility=2.0 # 日波动率2%
)
估算10000股执行的冲击
shares = 10000
price = 42150.0
impact = impact_model.total_impact(shares / 1000000) * price # 归一化后
print(f"执行 {shares} 股的估算冲击成本: ${impact:.2f}")
print(f"冲击成本 (bps): {impact / price * 10000:.2f} bps")
最优执行策略
optimal = impact_model.optimal_execution(
shares=10000,
time_horizon=1.0, # 1天执行完毕
risk_aversion=1e-6, # 风险厌恶系数
n_periods=100
)
print(f"\n最优执行策略预期成本: ${optimal['expected_cost']:.2f}")
print(f"每shares成本: ${optimal['execution_cost_per_share']:.4f}")
实战：构建订单簿分析系统
系统架构设计
一个完整的订单簿分析系统需要包含以下模块：
- 数据采集层：连接交易所 WebSocket，获取实时订单簿增量更新
- 数据处理层：解析、验证、存储订单簿快照和变化
- 特征计算层：计算市场深度、流动性指标、不平衡度等
- 分析/预测层：基于订单流预测短期价格变动
# 订单簿数据采集与处理（使用 HolySheep API 进行语义分析）
import json
import asyncio
import aiohttp
from typing import Dict, List, Optional
from datetime import datetime
class OrderBookCollector:
"""订单簿数据采集器"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.order_book_snapshot: Dict = {}
self.update_sequence: List[Dict] = []
self._session: Optional[aiohttp.ClientSession] = None
async def connect(self, symbol: str):
"""建立 WebSocket 连接（模拟）"""
# 实际生产环境中，这里应该是交易所的 WebSocket 地址
# 例如 Binance: wss://stream.binance.com:9443/ws
self._session = aiohttp.ClientSession()
print(f"已连接到 {symbol} 订单簿流")
async def fetch_historical_depth(self, symbol: str,
limit: int = 100) -> Dict:
"""
获取历史订单簿深度数据
REST API 调用示例
"""
# 注意：这是交易所的公开 API，与 HolySheep API 无关
# 仅用于获取市场数据
url = f"https://api.binance.com/api/v3/depth"
params = {'symbol': symbol, 'limit': limit}
async with self._session.get(url, params=params) as resp:
if resp.status == 200:
data = await resp.json()
return {
'lastUpdateId': data['lastUpdateId'],
'bids': [(float(p), float(q)) for p, q in data['bids']],
'asks': [(float(p), float(q)) for p, q in data['asks']],
'timestamp': datetime.now().isoformat()
}
else:
raise Exception(f"获取深度数据失败: {resp.status}")
async def analyze_order_book_with_llm(self,
order_book_data: Dict) -> Dict:
"""
使用 LLM 分析订单簿特征
HolySheep API 调用示例：
- 汇率优势：¥1=$1（官方¥7.3=$1），节省>85%
- 国内直连延迟<50ms
- 支持微信/支付宝充值
"""
# 准备分析上下文
bids = order_book_data['bids'][:10]
asks = order_book_data['asks'][:10]
bid_prices = [b[0] for b in bids]
ask_prices = [a[0] for a in asks]
# 计算基础指标
mid_price = (bid_prices[0] + ask_prices[0]) / 2
spread = ask_prices[0] - bid_prices[0]
spread_pct = spread / mid_price * 100
# 深度分析
depth_analysis = {
'mid_price': mid_price,
'spread': spread,
'spread_bps': spread / mid_price * 10000,
'top_10_bid_vol': sum(b[1] for b in bids),
'top_10_ask_vol': sum(a[1] for a in asks),
'bid_ask_imbalance': (sum(b[1] for b in bids) - sum(a[1] for a in asks)) /
(sum(b[1] for b in bids) + sum(a[1] for a in asks))
}
# 构建 LLM 分析 prompt
analysis_prompt = f"""作为一位专业的量化交易分析师，请分析以下订单簿数据：
当前价格: {mid_price:.2f}
买卖价差: {spread:.4f} ({spread_pct:.4f}%)
买一~买十总量: {depth_analysis['top_10_bid_vol']:.4f}
卖一~卖十总量: {depth_analysis['top_10_ask_vol']:.4f}
订单不平衡度: {depth_analysis['bid_ask_imbalance']:.4f}
买单价格档位（价格, 数量）:
{chr(10).join([f" 买{i+1}: {b[0]:.2f}, {b[1]:.4f}" for i, b in enumerate(bids)])}
卖单价格档位（价格, 数量）:
{chr(10).join([f" 卖{i+1}: {a[0]:.2f}, {a[1]:.4f}" for i, a in enumerate(asks)])}
请分析：
1. 市场流动性状态（紧松/深度/弹性）
2. 多空力量对比
3. 短期价格走势判断
4. 潜在支撑/阻力位
"""
# 调用 HolySheep API 进行语义分析
try:
async with self._session.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1", # $8/MTok output
"messages": [
{"role": "system", "content": "你是一位专业的量化交易分析师，擅长订单簿分析和价格预测。"},
{"role": "user", "content": analysis_prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
) as resp:
if resp.status == 200:
result = await resp.json()
analysis = result['choices'][0]['message']['content']
depth_analysis['llm_analysis'] = analysis
depth_analysis['llm_model_used'] = 'gpt-4.1'
else:
error_text = await resp.text()
depth_analysis['llm_analysis'] = f"API调用失败: {error_text}"
except Exception as e:
depth_analysis['llm_analysis'] = f"分析异常: {str(e)}"
return depth_analysis
async def process_order_book_stream(self, symbol: str):
"""处理实时订单簿流"""
await self.connect(symbol)
try:
# 获取初始快照
snapshot = await self.fetch_historical_depth(symbol, limit=100)
self.order_book_snapshot = snapshot
# 模拟增量更新处理
async for update in self._simulate_updates():
self._apply_update(update)
# 每100个更新进行一次深度分析
if len(self.update_sequence) % 100 == 0:
analysis = await self.analyze_order_book_with_llm(
self.order_book_snapshot
)
print(f"分析结果: {analysis.get('llm_analysis', 'N/A')[:200]}...")
finally:
await self.close()
def _apply_update(self, update: Dict):
"""应用订单簿更新"""
# 更新买卖盘
for price, qty in update.get('b', []):
self._update_side('bid', float(price), float(qty))
for price, qty in update.get('a', []):
self._update_side('ask', float(price), float(qty))
self.update_sequence.append(update)
def _update_side(self, side: str, price: float, qty: float):
"""更新指定方向的订单"""
key = f"{side}s" # bids or asks
if qty == 0:
# 删除订单
self.order_book_snapshot[key] = [
(p, v) for p, v in self.order_book_snapshot[key]
if abs(p - price) > 1e-8
]
else:
# 更新或添加
found = False
for i, (p, v) in enumerate(self.order_book_snapshot[key]):
if abs(p - price) < 1e-8:
self.order_book_snapshot[key][i] = (price, qty)
found = True
break
if not found:
self.order_book_snapshot[key].append((price, qty))
# 重新排序
reverse = (side == 'bid')
self.order_book_snapshot[key].sort(
key=lambda x: x[0], reverse=reverse
)
async def _simulate_updates(self):
"""模拟订单簿更新流"""
import random
import time
for _ in range(1000):
# 模拟更新数据
update = {
'e': 'depthUpdate',
'b': [(42150 + random.uniform(-0.5, 0.5),
random.uniform(0.1, 2.0)) for _ in range(random.randint(1, 5))],
'a': [(42151 + random.uniform(-0.5, 0.5),
random.uniform(0.1, 2.0)) for _ in range(random.randint(1, 5))]
}
yield update
await asyncio.sleep(0.05) # 50ms 间隔
async def close(self):
"""关闭连接"""
if self._session:
await self._session.close()
使用示例
async def main():
collector = OrderBookCollector(
api_key="YOUR_HOLYSHEEP_API_KEY", # 使用 HolySheep API Key
base_url="https://api.holysheep.ai/v1"
)
try:
# 获取实时数据
depth = await collector.fetch_historical_depth("BTCUSDT", limit=20)
# 使用 LLM 分析
analysis = await collector.analyze_order_book_with_llm(depth)
print("=== 订单簿分析结果 ===")
print(f"中间价: {analysis['mid_price']:.2f}")
print(f"价差: {analysis['spread']:.4f} ({analysis['spread_bps']:.2f} bps)")
print(f"订单不平衡度: {analysis['bid_ask_imbalance']:.4f}")
print(f"LLM 分析模型: {analysis.get('llm_model_used', 'N/A')}")
print(f"\nLLM 语义分析:\n{analysis.get('llm_analysis', 'N/A')}")
except Exception as e:
print(f"执行错误: {e}")
finally:
await collector.close()
运行
asyncio.run(main())
常见报错排查
问题1：订单簿数据不一致（Sequence Gap）
# 错误：WebSocket 订单簿更新序列不连续
错误信息：OrderBookCacheError: gap in sequence. last=123456, new=123789
class OrderBookCache:
"""
订单簿缓存处理序列不连续问题
"""
def __init__(self, max_snapshots: int = 5):
self.snapshots: deque = deque(maxlen=max_snapshots)
self.last_update_id: int = 0
self.pending_updates: List[Dict] = []
def apply_snapshot(self, snapshot: Dict) -> bool:
"""
应用初始快照，返回是否成功
重要：必须等待 WebSocket 第一条消息的 lastUpdateId >= 快照的 lastUpdateId
"""
snapshot_id = snapshot['lastUpdateId']
# 检查是否有待处理的更新
if self.pending_updates:
# 过滤掉序列号 <= snapshot_id 的更新
valid_updates = [
u for u in self.pending_updates
if u['u'] > snapshot_id
]
if not valid_updates:
# 所有待处理更新都已在快照中覆盖
self.pending_updates = []
else:
# 返回 False，提示需要重新同步
return False
self.snapshots.append(snapshot)
self.last_update_id = snapshot_id
return True
def apply_update(self, update: Dict) -> Optional[Dict]:
"""
应用增量更新，处理序列号不连续
Returns:
处理后的订单簿状态，如果需要重新同步则返回 None
"""
update_id = update['u']
first_update_id = update.get('U', 0) # 第一个订单 ID
final_update_id = update.get('u', 0) # 最后一个订单 ID
# 情况1：更新序列号 < 最后更新的ID，丢弃
if update_id <= self.last_update_id:
print(f"丢弃过期更新: {update_id} <= {self.last_update_id}")
return None
# 情况2：首次更新 ID > 最后更新 ID + 1，说明有间隙
if first_update_id > self.last_update_id + 1:
# 需要重新获取快照
print(f"检测到序列间隙: last={self.last_update_id}, first={first_update_id}")
self.pending_updates.append(update)
return None
# 情况3：正常顺序更新
return self._do_apply_update(update)
def _do_apply_update(self, update: Dict) -> Dict:
"""执行更新逻辑"""
# 处理买单更新
for price, qty in update.get('b', []):
self._update_order('bid', float(price), float(qty))
# 处理卖单更新
for price, qty in update.get('a', []):
self._update_order('ask', float(price), float(qty))
self.last_update_id = update['u']
return self.get_current_state()