2024年双十一那天，我的加密货币套利机器人在线上跑了3小时后彻底哑火。行情数据像断了线的风筝，订单簿数据停在错误的时间戳上，价差套利机会眼睁睁从指缝溜走。那一刻我才意识到，WebSocket 断线重连不是"锦上添花"，而是高频交易系统的"生死线"。
本文是我用血泪踩坑换来的实战经验，涵盖 Binance、Bybit、OKX、Deribit 四大主流交易所的 WebSocket 行情接入完整方案，包含可直接复制的 Python/Node.js 代码，以及常见报错的排查清单。
为什么你的 WebSocket 连接总是不稳定？
在深入代码之前，先理解问题的本质。加密货币交易所的 WebSocket 连接不稳定通常由以下原因造成：
- 网络抖动：云服务器到交易所服务器之间的网络路径可能经过多个运营商节点，任何一个节点故障都会导致连接中断。
- 交易所限流：Binance 单个 IP 每分钟最多建立 5 个新连接，超出后会被临时封禁 2 分钟。
- 心跳超时：大多数交易所要求客户端每 5-30 秒发送一次 ping，未响应的连接会被服务器主动关闭。
- 防火墙/NAT：企业网络或云服务器的安全组配置可能阻断长连接。
标准断线重连架构设计
一个健壮的重连机制需要包含以下组件：
"""
WebSocket 行情重连管理器 - Python 实现
支持：Binance / Bybit / OKX / Deribit
"""
import asyncio
import json
import time
import random
from dataclasses import dataclass, field
from typing import Callable, Optional, Dict, Any
from enum import Enum
import websockets
from websockets.client import WebSocketClientProtocol
class ConnectionState(Enum):
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
RECONNECTING = "reconnecting"
@dataclass
class ReconnectConfig:
max_retries: int = 10
base_delay: float = 1.0 # 基础重连延迟（秒）
max_delay: float = 60.0 # 最大重连延迟（秒）
exponential_base: float = 2.0 # 指数退避基数
jitter: bool = True # 是否添加随机抖动
@dataclass
class MarketDataHandler:
"""行情数据处理器基类"""
def on_orderbook(self, exchange: str, data: Dict[str, Any]):
"""处理订单簿数据"""
pass
def on_trade(self, exchange: str, data: Dict[str, Any]):
"""处理成交数据"""
pass
def on_ticker(self, exchange: str, data: Dict[str, Any]):
"""处理行情 ticker"""
pass
class MultiExchangeWebSocketManager:
"""
多交易所 WebSocket 行情管理
核心特性：
- 指数退避重连 + 抖动
- 自动心跳维持
- 多交易所并行管理
- 数据流丢失检测
"""
def __init__(
self,
config: ReconnectConfig = None,
handler: MarketDataHandler = None
):
self.config = config or ReconnectConfig()
self.handler = handler or MarketDataHandler()
self.connections: Dict[str, WebSocketClientProtocol] = {}
self.states: Dict[str, ConnectionState] = {}
self.retry_counts: Dict[str, int] = {}
self.last_message_time: Dict[str, float] = {}
self.subscriptions: Dict[str, list] = {}
self._running = False
# 交易所 WebSocket 端点配置
self.endpoints = {
"binance": "wss://stream.binance.com:9443/ws",
"bybit": "wss://stream.bybit.com/v5/public/spot",
"okx": "wss://ws.okx.com:8443/ws/v5/public",
"deribit": "wss://www.deribit.com/ws/api/v2"
}
# 心跳配置（各交易所要求不同）
self.ping_intervals = {
"binance": 60,
"bybit": 30,
"okx": 20,
"deribit": 10
}
# 数据超时阈值（秒）
self.data_timeouts = {
"binance": 10,
"bybit": 15,
"okx": 12,
"deribit": 8
}
def _calculate_delay(self, retry_count: int) -> float:
"""计算重连延迟：指数退避 + 抖动"""
delay = min(
self.config.base_delay * (self.config.exponential_base ** retry_count),
self.config.max_delay
)
if self.config.jitter:
# 添加 ±25% 随机抖动，避免多客户端同时重连
delay *= (0.75 + random.random() * 0.5)
return delay
async def connect(self, exchange: str, streams: list):
"""连接到指定交易所"""
if exchange not in self.endpoints:
raise ValueError(f"不支持的交易所: {exchange}")
self.subscriptions[exchange] = streams
self.states[exchange] = ConnectionState.DISCONNECTED
await self._establish_connection(exchange)
async def _establish_connection(self, exchange: str):
"""建立或重建连接"""
self.states[exchange] = ConnectionState.CONNECTING
try:
endpoint = self.endpoints[exchange]
async with websockets.connect(endpoint, ping_interval=None) as ws:
self.connections[exchange] = ws
self.states[exchange] = ConnectionState.CONNECTED
self.retry_counts[exchange] = 0
self.last_message_time[exchange] = time.time()
# 发送订阅请求
await self._subscribe(exchange, ws)
# 启动心跳协程
asyncio.create_task(self._heartbeat(exchange, ws))
# 启动数据超时检测
asyncio.create_task(self._monitor_data_timeout(exchange))
# 启动消息接收循环
await self._receive_loop(exchange, ws)
except websockets.exceptions.ConnectionClosed as e:
print(f"[{exchange}] 连接关闭: {e}")
await self._handle_reconnect(exchange)
except Exception as e:
print(f"[{exchange}] 连接错误: {e}")
await self._handle_reconnect(exchange)
async def _subscribe(self, exchange: str, ws: WebSocketClientProtocol):
"""发送订阅请求（各交易所格式不同）"""
streams = self.subscriptions.get(exchange, [])
if exchange == "binance":
# Binance: 直接在 URL 中指定 streams
pass
elif exchange == "bybit":
await ws.send(json.dumps({
"op": "subscribe",
"args": streams
}))
elif exchange == "okx":
await ws.send(json.dumps({
"op": "subscribe",
"args": [{"channel": s} for s in streams]
}))
elif exchange == "deribit":
await ws.send(json.dumps({
"method": "subscribe",
"params": {"channels": streams}
}))
async def _heartbeat(self, exchange: str, ws: WebSocketClientProtocol):
"""维持心跳连接"""
interval = self.ping_intervals.get(exchange, 30)
try:
while self.states.get(exchange) == ConnectionState.CONNECTED:
await asyncio.sleep(interval)
if self.states.get(exchange) == ConnectionState.CONNECTED:
await ws.ping()
except Exception:
pass
async def _monitor_data_timeout(self, exchange: str):
"""检测数据流是否超时"""
timeout = self.data_timeouts.get(exchange, 10)
while self.states.get(exchange) == ConnectionState.CONNECTED:
await asyncio.sleep(timeout / 2)
last_time = self.last_message_time.get(exchange, 0)
if time.time() - last_time > timeout:
print(f"[{exchange}] 数据流超时 ({timeout}s)，强制重连")
await self._handle_reconnect(exchange)
break
async def _receive_loop(self, exchange: str, ws: WebSocketClientProtocol):
"""消息接收主循环"""
async for message in ws:
self.last_message_time[exchange] = time.time()
try:
data = json.loads(message)
await self._process_message(exchange, data)
except json.JSONDecodeError:
print(f"[{exchange}] JSON 解析失败: {message[:100]}")
async def _process_message(self, exchange: str, data: Dict[str, Any]):
"""处理接收到的消息"""
# 根据消息类型分发到不同处理器
msg_type = data.get("e") or data.get("type") or data.get("channel", "")
if "depth" in str(msg_type).lower() or "orderbook" in str(data).lower():
self.handler.on_orderbook(exchange, data)
elif "trade" in str(msg_type).lower():
self.handler.on_trade(exchange, data)
elif "ticker" in str(msg_type).lower() or "bookTicker" in str(data).lower():
self.handler.on_ticker(exchange, data)
async def _handle_reconnect(self, exchange: str):
"""处理重连逻辑"""
retry_count = self.retry_counts.get(exchange, 0)
if retry_count >= self.config.max_retries:
print(f"[{exchange}] 达到最大重试次数 ({self.config.max_retries})，放弃重连")
self.states[exchange] = ConnectionState.DISCONNECTED
return
self.states[exchange] = ConnectionState.RECONNECTING
delay = self._calculate_delay(retry_count)
print(f"[{exchange}] {delay:.1f}秒后重连 (第 {retry_count + 1} 次)")
await asyncio.sleep(delay)
self.retry_counts[exchange] = retry_count + 1
# 尝试重建连接
try:
await self._establish_connection(exchange)
except Exception as e:
print(f"[{exchange}] 重连失败: {e}")
await self._handle_reconnect(exchange)
async def start_all(self, exchanges: Dict[str, list]):
"""启动所有交易所连接"""
self._running = True
tasks = [
self.connect(exchange, streams)
for exchange, streams in exchanges.items()
]
await asyncio.gather(*tasks)
async def stop(self):
"""停止所有连接"""
self._running = False
for exchange, ws in self.connections.items():
await ws.close()
self.states[exchange] = ConnectionState.DISCONNECTED
使用示例
if __name__ == "__main__":
class MyHandler(MarketDataHandler):
def on_orderbook(self, exchange: str, data: Dict[str, Any]):
print(f"[{exchange}] 订单簿更新: {data.get('s', 'UNKNOWN')}")
def on_trade(self, exchange: str, data: Dict[str, Any]):
print(f"[{exchange}] 成交: 价格={data.get('p')}, 数量={data.get('q')}")
async def main():
config = ReconnectConfig(
max_retries=10,
base_delay=2.0,
max_delay=120.0,
jitter=True
)
manager = MultiExchangeWebSocketManager(
config=config,
handler=MyHandler()
)
# 订阅配置
exchanges = {
"binance": ["btcusdt@depth20@100ms", "btcusdt@trade"],
"bybit": ["orderbook.50.BTCUSDT", "publicTrade.BTCUSDT"],
"okx": ["books5:BTC-USDT", "trades:BTC-USDT"],
"deribit": ["book.BTC-PERPETUAL", "trades.BTC-PERPETUAL"]
}
try:
await manager.start_all(exchanges)
except KeyboardInterrupt:
await manager.stop()
# 运行: asyncio.run(main())Node.js 实现方案
如果你更习惯 JavaScript/TypeScript 生态，以下是 Node.js 原生实现：
/**
* 多交易所 WebSocket 行情客户端 - Node.js 实现
* 使用原生 WebSocket + async/await
*/
const EventEmitter = require('events');
class ExchangeWebSocket extends EventEmitter {
constructor(exchange, config = {}) {
super();
this.exchange = exchange;
this.config = {
maxRetries: config.maxRetries || 10,
baseDelay: config.baseDelay || 1000,
maxDelay: config.maxDelay || 60000,
pingInterval: config.pingInterval || 30000,
dataTimeout: config.dataTimeout || 15000
};
this.ws = null;
this.retryCount = 0;
this.isRunning = false;
this.lastMessageTime = 0;
this.pingTimer = null;
this.timeoutTimer = null;
// 各交易所配置
this.endpoints = {
binance: 'wss://stream.binance.com:9443/ws',
bybit: 'wss://stream.bybit.com/v5/public/spot',
okx: 'wss://ws.okx.com:8443/ws/v5/public',
deribit: 'wss://www.deribit.com/ws/api/v2'
};
}
calculateDelay() {
// 指数退避 + 抖动
let delay = Math.min(
this.config.baseDelay * Math.pow(2, this.retryCount),
this.config.maxDelay
);
// 添加 ±25% 抖动
delay *= (0.75 + Math.random() * 0.5);
return delay;
}
async connect(streams = []) {
return new Promise((resolve, reject) => {
const endpoint = this.endpoints[this.exchange];
if (!endpoint) {
reject(new Error(不支持的交易所: ${this.exchange}));
return;
}
console.log([${this.exchange}] 正在连接...);
try {
this.ws = new WebSocket(endpoint);
this.ws.on('open', async () => {
console.log([${this.exchange}] 连接成功);
this.isRunning = true;
this.retryCount = 0;
this.lastMessageTime = Date.now();
// 发送订阅
await this.subscribe(streams);
// 启动心跳
this.startPing();
// 启动超时检测
this.startTimeoutMonitor();
this.emit('connected');
resolve();
});
this.ws.on('message', (data) => {
this.lastMessageTime = Date.now();
this.handleMessage(data);
});
this.ws.on('close', (code, reason) => {
console.log([${this.exchange}] 连接关闭: ${code} - ${reason});
this.cleanup();
this.handleReconnect(streams);
});
this.ws.on('error', (error) => {
console.error([${this.exchange}] 错误: ${error.message});
this.emit('error', error);
});
} catch (error) {
reject(error);
}
});
}
async subscribe(streams) {
if (!streams.length || this.ws.readyState !== WebSocket.OPEN) return;
const message = this.buildSubscribeMessage(streams);
this.ws.send(JSON.stringify(message));
console.log([${this.exchange}] 订阅: ${JSON.stringify(streams)});
}
buildSubscribeMessage(streams) {
// 各交易所订阅格式不同
switch (this.exchange) {
case 'bybit':
return { op: 'subscribe', args: streams };
case 'okx':
return { op: 'subscribe', args: streams.map(s => ({ channel: s })) };
case 'deribit':
return { method: 'subscribe', params: { channels: streams } };
default:
return streams;
}
}
handleMessage(data) {
try {
const message = JSON.parse(data);
this.emit('message', message);
// 路由到具体处理器
if (message.e) {
// Binance 格式
this.emit(message.e, message);
} else if (message.type) {
this.emit(message.type, message);
}
} catch (e) {
console.error([${this.exchange}] 消息解析失败);
}
}
startPing() {
this.pingTimer = setInterval(() => {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.ping();
}
}, this.config.pingInterval);
}
startTimeoutMonitor() {
this.timeoutTimer = setInterval(() => {
const elapsed = Date.now() - this.lastMessageTime;
if (elapsed > this.config.dataTimeout) {
console.warn([${this.exchange}] 数据超时 (${elapsed}ms)，重连中...);
this.ws.close();
}
}, this.config.dataTimeout / 2);
}
cleanup() {
this.isRunning = false;
if (this.pingTimer) clearInterval(this.pingTimer);
if (this.timeoutTimer) clearInterval(this.timeoutTimer);
}
async handleReconnect(streams) {
if (!this.isRunning) return;
if (this.retryCount >= this.config.maxRetries) {
console.error([${this.exchange}] 达到最大重试次数，停止重连);
this.emit('maxRetriesExceeded');
return;
}
const delay = this.calculateDelay();
console.log([${this.exchange}] ${Math.round(delay/1000)}秒后重连 (第 ${this.retryCount + 1} 次));
await new Promise(resolve => setTimeout(resolve, delay));
this.retryCount++;
try {
await this.connect(streams);
} catch (error) {
console.error([${this.exchange}] 重连失败: ${error.message});
}
}
close() {
this.isRunning = false;
if (this.ws) {
this.ws.close();
}
this.cleanup();
}
}
// 使用示例
async function main() {
const streams = {
binance: ['btcusdt@depth20@100ms', 'btcusdt@trade'],
bybit: ['orderbook.50.BTCUSDT', 'publicTrade.BTCUSDT'],
okx: ['books5:BTC-USDT', 'trades:BTC-USDT']
};
const clients = {};
for (const [exchange, params] of Object.entries(streams)) {
const client = new ExchangeWebSocket(exchange, {
maxRetries: 10,
baseDelay: 2000,
maxDelay: 120000
});
// 事件监听
client.on('connected', () => {
console.log(✓ ${exchange} 已连接);
});
client.on('depthUpdate', (data) => {
// 处理订单簿更新
});
client.on('trade', (data) => {
// 处理成交数据
});
client.on('error', (error) => {
console.error(✗ ${exchange} 错误:, error.message);
});
client.on('maxRetriesExceeded', () => {
console.error(✗ ${exchange} 重试次数耗尽);
});
clients[exchange] = client;
}
// 启动所有连接
const connections = Object.entries(clients).map(
([exchange, client]) => client.connect(streams[exchange])
);
await Promise.allSettled(connections);
// 优雅关闭
process.on('SIGINT', () => {
console.log('\n正在关闭所有连接...');
Object.values(clients).forEach(client => client.close());
process.exit(0);
});
}
main().catch(console.error);多交易所数据对齐策略
实战中最头疼的问题不是断线重连，而是多交易所数据的时间对齐。不同交易所的时钟可能存在毫秒级偏差，直接用各自的时间戳做跨交易所套利会出问题。
"""
多交易所数据时间对齐模块
核心思路：
1. 以某个交易所为基准（通常选择延迟最低的）
2. 其他交易所数据通过插值/外推对齐到基准时间戳
3. 使用滑动窗口平滑异常值
"""
import time
from collections import deque
from dataclasses import dataclass
from typing import Dict, Optional
import threading
@dataclass
class TimestampedData:
exchange: str
symbol: str
price: float
timestamp: float # 交易所原始时间戳
local_time: float # 本地接收时间
@dataclass
class AlignedData:
base_timestamp: float
data: Dict[str, Optional[TimestampedData]]
latency_ms: float # 相对基准的延迟
class MultiExchangeSynchronizer:
"""
多交易所行情数据时间同步器
算法说明：
- 使用滑动窗口记录每个交易所的数据到达时间差
- 动态计算各交易所的时钟偏差
- 基于偏差值对原始数据进行时间对齐
"""
def __init__(self, base_exchange: str = "binance", window_size: int = 100):
self.base_exchange = base_exchange
self.window_size = window_size
# 各交易所的时钟偏差（相对于基准交易所）
self.clock_offsets: Dict[str, deque] = {
"binance": deque(maxlen=window_size),
"bybit": deque(maxlen=window_size),
"okx": deque(maxlen=window_size),
"deribit": deque(maxlen=window_size)
}
self.latest_data: Dict[str, Optional[TimestampedData]] = {
"binance": None,
"bybit": None,
"okx": None,
"deribit": None
}
self._lock = threading.Lock()
def add_data(self, data: TimestampedData) -> Optional[AlignedData]:
"""添加新的行情数据，返回对齐后的数据（如果可对齐）"""
with self._lock:
exchange = data.exchange
local_time = data.local_time
# 更新最新数据
self.latest_data[exchange] = data
if exchange != self.base_exchange:
# 计算时钟偏移
# 假设：同一时刻，各交易所收到同一事件的本地时间应相近
# 偏移 = 交易所时间戳 - 本地接收时间 + 基准偏移
base_data = self.latest_data.get(self.base_exchange)
if base_data:
# 粗略估计偏移（实际场景需要更复杂的算法）
offset = data.timestamp - (local_time - base_data.local_time + base_data.timestamp)
self.clock_offsets[exchange].append(offset)
# 返回对齐数据
return self._align()
def _align(self) -> Optional[AlignedData]:
"""将所有交易所数据对齐到基准时间戳"""
base_data = self.latest_data.get(self.base_exchange)
if not base_data:
return None
base_time = base_data.timestamp
aligned = {}
max_latency = 0
for exchange, data in self.latest_data.items():
if data is None:
aligned[exchange] = None
continue
# 计算时钟偏差的中位数
offsets = list(self.clock_offsets.get(exchange, []))
clock_correction = sum(offsets) / len(offsets) if offsets else 0
# 对齐后的时间戳
adjusted_time = data.timestamp - clock_correction
latency = abs(adjusted_time - base_time) * 1000 # 毫秒
aligned[exchange] = data
max_latency = max(max_latency, latency)
return AlignedData(
base_timestamp=base_time,
data=aligned,
latency_ms=max_latency
)
def get_spread_opportunity(self, symbol: str, min_spread: float = 0.001) -> Optional[Dict]:
"""
检测跨交易所套利机会
要求：所有交易所数据已对齐，延迟小于阈值
"""
with self._lock:
aligned = self._align()
if not aligned or aligned.latency_ms > 100: # 超过100ms的对齐数据不可用
return None
prices = {}
for exchange, data in aligned.data.items():
if data and data.symbol == symbol:
prices[exchange] = data.price
if len(prices) < 2:
return None
max_price = max(prices.values())
min_price = min(prices.values())
spread = (max_price - min_price) / min_price
if spread >= min_spread:
max_ex = [k for k, v in prices.items() if v == max_price][0]
min_ex = [k for k, v in prices.items() if v == min_price][0]
return {
"symbol": symbol,
"buy_exchange": min_ex,
"sell_exchange": max_ex,
"buy_price": prices[min_ex],
"sell_price": prices[max_ex],
"spread_bps": round(spread * 10000, 2),
"latency_ms": round(aligned.latency_ms, 2),
"timestamp": aligned.base_timestamp
}
return None常见报错排查
错误 1：Connection Refused (10061/10060)
错误信息：websockets.exceptions.ConnectionRefusedError: [WinError 10061] 或 ConnectionResetError: [Errno 104]
原因分析：
- 目标服务器拒绝连接，通常是 IP 被交易所封禁
- 防火墙拦截了出站 WebSocket 端口（80/443）
- 交易所端点变更或维护
解决方案：
# 检查 IP 是否被封禁
import requests
def check_ban_status():
# Binance IP 检查
try:
r = requests.get("https://api.binance.com/wapi/v3/apiKey.html",
headers={"X-MBX-APIKEY": "YOUR_KEY"})
print(f"Binance API 状态: {r.status_code}")
except Exception as e:
print(f"可能已被封禁: {e}")
添加重试和备用端点
ENDPOINTS_BINANCE = [
"wss://stream.binance.com:9443/ws",
"wss://stream.binance.com/ws", # 备用
"wss://stream.bnbchain.org/ws", # BNB Chain 端点
]
使用健康检查选择最优端点
import asyncio
import aiohttp
async def health_check(endpoint):
try:
async with aiohttp.ClientSession() as session:
start = asyncio.get_event_loop().time()
async with session.ws_connect(endpoint.replace('wss://', 'https://')) as ws:
latency = (asyncio.get_event_loop().time() - start) * 1000
return (endpoint, latency, True)
except:
return (endpoint, 9999, False)
async def select_best_endpoint():
results = await asyncio.gather(*[health_check(ep) for ep in ENDPOINTS_BINANCE])
available = [r for r in results if r[2]]
if available:
return min(available, key=lambda x: x[1])[0]
raise Exception("所有端点均不可用")错误 2：429 Too Many Requests
错误信息：WebSocket connection closed: 1008: Too Many Requests
原因分析：
- 连接频率超限（Binance 单 IP 5连接/分钟）
- 订阅消息频率过快
- 同一账号多实例同时连接
解决方案：
import asyncio
import threading
class RateLimitedConnection:
"""带限流控制的连接管理器"""
def __init__(self, max_connections_per_minute=4, max_subscriptions_per_second=10):
self.connection_semaphore = asyncio.Semaphore(max_connections_per_minute)
self.subscription_limiter = AsyncRateLimiter(max_subscriptions_per_second)
self._connection_count = 0
self._last_reset = time.time()
self._lock = threading.Lock()
async def acquire_connection(self):
"""获取连接许可（带自动重置计数）"""
async with self.connection_semaphore:
with self._lock:
current_time = time.time()
if current_time - self._last_reset > 60:
self._connection_count = 0
self._last_reset = current_time
if self._connection_count >= 4:
wait_time = 60 - (current_time - self._last_reset)
print(f"连接数超限，等待 {wait_time:.0f} 秒")
await asyncio.sleep(wait_time)
self._connection_count = 0
self._last_reset = time.time()
self._connection_count += 1
return True
async def subscribe(self, ws, channels):
"""限流订阅"""
for channel in channels:
await self.subscription_limiter.acquire()
await ws.send(json.dumps({"op": "subscribe", "args": [channel]}))
print(f"订阅: {channel}")
await asyncio.sleep(0.2) # 每次订阅间隔 200ms
class AsyncRateLimiter:
"""异步速率限制器"""
def __init__(self, max_per_second):
self.max_per_second = max_per_second
self.interval = 1.0 / max_per_second
self.last_execution = 0
self._lock = asyncio.Lock()
async def acquire(self):
async with self._lock:
now = asyncio.get_event_loop().time()
wait_time = self.last_execution + self.interval - now
if wait_time > 0:
await asyncio.sleep(wait_time)
self.last_execution = asyncio.get_event_loop().time()错误 3：数据乱序与消息丢失
错误信息：订单簿更新序号跳号、深度数据出现负数数量
原因分析：
- 网络重传导致消息乱序到达
- 重连后未正确处理增量更新的序号验证
- TCP 层的丢包在 WebSocket 层表现为数据不连续
解决方案：
class OrderbookManager:
"""
订单簿状态管理器
包含序号验证和数据完整性检查
"""
def __init__(self, symbol: str, snapshot_depth: int = 20):
self.symbol = symbol
self.snapshot_depth = snapshot_depth
self.bids = {} # {price: quantity}
self.asks = {}
self.last_update_id = 0
self.last_sequence = 0
self._pending_updates = []
self._is_snapshot_loaded = False
def apply_snapshot(self, snapshot: dict):
"""加载完整快照"""
self.bids.clear()
self.asks.clear()
# Binance 格式
for price, qty in snapshot.get('bids', []):
self.bids[float(price)] = float(qty)
for price, qty in snapshot.get('asks', []):
self.asks[float(price)] = float(qty)
self.last_update_id = snapshot.get('lastUpdateId', 0)
self.last_sequence = self.last_update_id
self._is_snapshot_loaded = True
self._process_pending()
def apply_update(self, update: dict) -> bool:
"""
应用增量更新
返回值表示是否成功应用
"""
if not self._is_snapshot_loaded:
self._pending_updates.append(update)
return False
# 序号验证
new_seq = update.get('u') or update.get('seq') or update.get('sequence', 0)
# 允许序号跳跃（跳过丢失的消息，直接使用最新状态）
# 但拒绝序号倒退
if new_seq <= self.last_sequence and self.last_sequence > 0:
print(f"序号倒退: {new_seq} < {self.last_sequence}")
return False
self.last_sequence = new_seq
# 应用更新
for price, qty in update.get('b', update.get('bids', [])):
price, qty = float(price), float(qty)
if qty == 0:
self.bids.pop(price, None)
else:
if qty < 0:
print(f"非法数量（负数）: {qty}")
return False
self.bids[price] = qty
for price, qty in update.get('a', update.get('asks', [])):
price, qty = float(price), float(qty)
if qty == 0:
self.asks.pop(price, None)
else:
if qty < 0:
print(f"非法数量（负数）: {qty}")
return False
self.asks[price] = qty
return True
def _process_pending(self):
"""处理积压的更新"""
while self._pending_updates:
update = self._pending_updates.pop(0)
self.apply_update(update)
def get_top_of_book(self) -> dict:
"""获取最佳买卖价"""
if not self.bids or not self.asks:
return {}
best_bid = max(self.bids.items(), key=lambda x: x[0])
best_ask = min(self.asks.items(), key=lambda x: x[0])
return {
"symbol": self.symbol,
"bid_price": best_bid[0],
"bid_qty": best_bid[1],
"ask_price": best_ask[0],
"ask_qty": best_ask[1],
"spread": best_ask[0] - best_bid[0],