在加密货币量化交易和高频数据采集场景中，Binance WebSocket 是获取实时行情的核心技术手段。我自己在 2023 年搭建多币种监控面板时，最初用 REST API 轮询，不仅延迟高达 500-800ms，还频繁触发 418/429 限流错误。切换到 WebSocket 订阅后，同等数据量下延迟降到 <50ms，服务器资源消耗减少 70%。
先看一个与本文看似无关、但直接影响你 API 成本的关键数字：
| 模型 | Output 价格 | 每月 100 万 Token 原价 | HolySheep 折算后 | 节省比例 |
|---|---|---|---|---|
| GPT-4.1 | $8/MTok | $8.00 | ¥8（约$1.1） | 86% |
| Claude Sonnet 4.5 | $15/MTok | $15.00 | ¥15（约$2.05） | 86% |
| Gemini 2.5 Flash | $2.50/MTok | $2.50 | ¥2.5（约$0.34） | 86% |
| DeepSeek V3.2 | $0.42/MTok | $0.42 | ¥0.42（约$0.058） | 86% |
HolySheep 按 ¥1=$1 无损结算（官方汇率为 ¥7.3=$1），对于需要同时调用大模型 API + 采集加密货币数据的开发者来说，组合使用能显著降低整体开发成本。注册即送免费额度：立即注册
WebSocket vs REST API：为什么订阅模式更优
在实时行情场景下，WebSocket 的优势是压倒性的：
- 延迟：WebSocket 推送延迟 <50ms，REST 轮询平均 300-800ms
- 资源消耗：WebSocket 维持长连接，单连接复用；REST 每请求新建连接
- 限流保护：Binance 对 WebSocket 连接更宽容（每分钟 5 个连接限制 vs REST 的复杂权重系统）
- 数据完整性：WebSocket 保证逐笔推送，不遗漏任何 tick
Python 实现：订阅多个交易对
# pip install websocket-client
import json
import threading
import time
from websocket import create_connection, WebSocketTimeoutException
class BinanceMultiStream:
def __init__(self, symbols, streams=None):
"""
symbols: 交易对列表，如 ['btcusdt', 'ethusdt', 'bnbusdt']
streams: 可选指定数据流类型，默认 ['trade', 'miniTicker']
"""
self.symbols = [s.lower() for s in symbols]
self.streams = streams or ['trade', 'miniTicker']
self.base_url = "wss://stream.binance.com:9443/stream"
self.ws = None
self.running = False
self.message_count = 0
self.start_time = None
def build_url(self):
"""构建组合订阅 URL（单连接多流）"""
# 格式：symbol@stream
stream_str = '/'.join([f"{s}@{t}" for s in self.symbols for t in self.streams])
return f"{self.base_url}?streams={stream_str}"
def connect(self):
"""建立 WebSocket 连接"""
url = self.build_url()
print(f"[连接] WebSocket URL: {url}")
print(f"[订阅] 交易对数量: {len(self.symbols)}, 数据流: {self.streams}")
self.ws = create_connection(url, timeout=30)
self.running = True
self.start_time = time.time()
print("[状态] 连接成功，开始接收数据...")
def on_message(self, ws, message):
"""消息处理回调"""
try:
data = json.loads(message)
# Binance Combined Stream 格式：{"stream":"btcusdt@trade","data":{...}}
stream = data.get('stream', '')
payload = data.get('data', {})
self.message_count += 1
# 示例：处理 Trade 数据
if 'trade' in stream:
symbol = payload.get('s', '')
price = payload.get('p', '')
quantity = payload.get('q', '')
print(f"[成交] {symbol} | 价格: {price} | 数量: {quantity}")
# 示例：处理迷你行情（包含OHLCV）
elif 'miniTicker' in stream:
symbol = payload.get('s', '')
close = payload.get('c', '')
high = payload.get('h', '')
low = payload.get('l', '')
print(f"[行情] {symbol} | 收盘: {close} | 最高: {high} | 最低: {low}")
except json.JSONDecodeError as e:
print(f"[错误] JSON 解析失败: {e}")
except Exception as e:
print(f"[错误] 消息处理异常: {e}")
def heartbeat(self):
"""心跳保活机制"""
while self.running:
try:
if self.ws:
self.ws.ping()
time.sleep(25) # Binance 建议 30 秒内 ping 一次
except Exception as e:
print(f"[心跳] 异常: {e}")
self.running = False
def receive_loop(self):
"""消息接收主循环"""
heartbeat_thread = threading.Thread(target=self.heartbeat, daemon=True)
heartbeat_thread.start()
while self.running:
try:
self.ws.settimeout(30)
message = self.ws.recv()
if message:
self.on_message(self.ws, message)
except WebSocketTimeoutException:
print("[超时] 等待消息...")
continue
except Exception as e:
print(f"[错误] 接收异常: {e}")
self.reconnect()
def reconnect(self, max_retries=5):
"""自动重连逻辑"""
retry_count = 0
while retry_count < max_retries and not self.running:
print(f"[重连] 第 {retry_count + 1}/{max_retries} 次尝试...")
time.sleep(2 ** retry_count) # 指数退避
try:
self.connect()
print("[重连] 成功恢复连接")
return
except Exception as e:
print(f"[重连] 失败: {e}")
retry_count += 1
print("[致命] 重连次数耗尽，程序退出")
def close(self):
"""关闭连接"""
self.running = False
if self.ws:
self.ws.close()
elapsed = time.time() - self.start_time if self.start_time else 0
print(f"[统计] 运行 {elapsed:.1f} 秒，共接收 {self.message_count} 条消息")
def run(self, symbols):
"""便捷启动入口"""
self.symbols = [s.lower() for s in symbols]
try:
self.connect()
self.receive_loop()
except KeyboardInterrupt:
print("\n[中断] 收到 Ctrl+C，正在关闭...")
finally:
self.close()
使用示例：订阅 BTC、ETH、BNB 的成交和迷你行情
if __name__ == "__main__":
client = BinanceMultiStream(
symbols=['btcusdt', 'ethusdt', 'bnbusdt', 'solusdt', 'adausdt'],
streams=['trade', 'miniTicker']
)
client.run(['btcusdt', 'ethusdt', 'bnbusdt', 'solusdt', 'adausdt'])JavaScript/Node.js 实现：异步订阅架构
// npm install ws
const WebSocket = require('ws');
class BinanceMultiStreamClient {
constructor(options = {}) {
this.symbols = options.symbols || [];
this.streams = options.streams || ['trade', 'miniTicker'];
this.baseUrl = 'wss://stream.binance.com:9443/stream';
this.ws = null;
this.reconnectAttempts = 0;
this.maxReconnectAttempts = options.maxReconnectAttempts || 10;
this.reconnectDelay = options.reconnectDelay || 1000;
this.messageHandler = options.onMessage || this.defaultHandler;
this.stats = { messages: 0, startTime: null };
}
buildUrl() {
// 构建组合流 URL
const streams = [];
for (const symbol of this.symbols) {
for (const stream of this.streams) {
streams.push(${symbol.toLowerCase()}@${stream});
}
}
return ${this.baseUrl}?streams=${streams.join('/')};
}
connect() {
return new Promise((resolve, reject) => {
const url = this.buildUrl();
console.log([连接] WebSocket: ${url});
console.log([订阅] ${this.symbols.length} 个交易对 × ${this.streams.length} 个数据流);
this.ws = new WebSocket(url, {
handshakeTimeout: 10000,
keepAlive: true,
keepAliveInterval: 25000
});
this.ws.on('open', () => {
console.log('[状态] WebSocket 连接已建立');
this.reconnectAttempts = 0;
this.stats.startTime = Date.now();
this.startHeartbeat();
resolve();
});
this.ws.on('message', (data) => {
try {
const message = JSON.parse(data.toString());
// Binance Combined Stream: { stream: "...", data: {...} }
const { stream, data: payload } = message;
this.stats.messages++;
this.messageHandler(stream, payload);
} catch (error) {
console.error('[错误] 消息解析失败:', error.message);
}
});
this.ws.on('error', (error) => {
console.error('[错误] WebSocket 错误:', error.message);
reject(error);
});
this.ws.on('close', (code, reason) => {
console.log([关闭] 连接断开 (${code}): ${reason || '无原因'});
this.scheduleReconnect();
});
// 超时处理
this.ws.on('timeout', () => {
console.warn('[警告] WebSocket 响应超时');
});
});
}
startHeartbeat() {
this.heartbeatInterval = setInterval(() => {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.ping();
console.log('[心跳] Ping 已发送');
}
}, 25000);
}
scheduleReconnect() {
if (this.reconnectAttempts >= this.maxReconnectAttempts) {
console.error('[致命] 重连次数已达上限');
process.exit(1);
}
this.reconnectAttempts++;
const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);
console.log([重连] ${delay/1000}秒后进行第 ${this.reconnectAttempts} 次重连...);
setTimeout(() => {
this.connect().catch(console.error);
}, delay);
}
defaultHandler(stream, data) {
const [symbol, streamType] = stream.split('@');
if (streamType === 'trade') {
console.log([成交] ${data.s} | 价格: ${data.p} | 数量: ${data.q} | 时间: ${new Date(data.T).toISOString()});
} else if (streamType === 'miniTicker') {
console.log([行情] ${data.s} | 收盘: ${data.c} | 24h涨跌: ${((parseFloat(data.c) - parseFloat(data.o)) / parseFloat(data.o) * 100).toFixed(2)}%);
}
}
close() {
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
}
if (this.ws) {
this.ws.close(1000, '客户端主动关闭');
}
const elapsed = (Date.now() - this.stats.startTime) / 1000;
console.log([统计] 运行 ${elapsed.toFixed(1)}s，接收 ${this.stats.messages} 条消息);
}
// 动态添加/移除交易对订阅
subscribe(symbols) {
const newSymbols = symbols.filter(s => !this.symbols.includes(s));
const removedSymbols = this.symbols.filter(s => !symbols.includes(s));
if (newSymbols.length > 0) {
console.log([订阅] 新增: ${newSymbols.join(', ')});
// 重新构建连接以包含新流
this.symbols = [...new Set([...this.symbols, ...newSymbols])];
this.ws.close();
this.connect().catch(console.error);
}
this.symbols = symbols; // 简化处理：全量更新
}
}
// 使用示例
const client = new BinanceMultiStreamClient({
symbols: ['btcusdt', 'ethusdt', 'bnbusdt', 'solusdt'],
streams: ['trade', 'miniTicker'],
maxReconnectAttempts: 10,
onMessage: (stream, data) => {
// 自定义消息处理逻辑
if (stream.includes('trade')) {
// 成交数据处理
}
}
});
client.connect().then(() => {
console.log('开始接收实时行情...');
// 示例：10秒后动态更新订阅列表
setTimeout(() => {
console.log('\n[更新] 动态调整订阅列表...');
client.subscribe(['btcusdt', 'ethusdt', 'adausdt', 'dotusdt']);
}, 10000);
}).catch(console.error);
// 优雅关闭
process.on('SIGINT', () => {
console.log('\n[中断] 正在关闭...');
client.close();
process.exit(0);
});支持的数据流类型一览
| 数据流 | 后缀 | 数据内容 | 更新频率 | 适用场景 |
|---|---|---|---|---|
| Trade | @trade | 逐笔成交 | 实时 | 高频交易、成交监控 |
| Mini Ticker | @miniTicker | 简洁行情(OHLCV汇总) | 实时 | 通用行情面板 |
| Ticker | @ticker | 完整24h统计 | 实时 | 技术分析、趋势监控 |
| Kline | @kline_{interval} | K线数据 | 按间隔更新 | 图表、指标计算 |
| Depth | @depth{level} | 订单簿深度 | 实时 | 做市策略、流动性分析 |
| Book Ticker | @bookTicker | 最优买卖价量 | 实时 | 价差监控、最优报价 |
Python 实现：K线订阅（支持多周期）
import json
import time
from websocket import create_connection
class BinanceKlineSubscriber:
"""K线数据订阅器（支持多交易对、多周期）"""
INTERVALS = {
'1m': '1m', '3m': '3m', '5m': '5m', '15m': '15m',
'30m': '30m', '1h': '1h', '2h': '2h', '4h': '4h',
'6h': '6h', '8h': '8h', '12h': '12h', '1d': '1d',
'3d': '3d', '1w': '1w', '1M': '1M'
}
def __init__(self, symbol, intervals=None):
self.symbol = symbol.lower()
self.intervals = intervals or ['1m', '5m', '1h']
self.base_url = "wss://stream.binance.com:9443/stream"
self.ws = None
self.klines = {} # 缓存最新K线
def build_streams(self):
"""构建K线流列表"""
streams = []
for interval in self.intervals:
if interval in self.INTERVALS:
streams.append(f"{self.symbol}@kline_{interval}")
return '/'.join(streams)
def connect(self):
url = f"{self.base_url}?streams={self.build_streams()}"
print(f"[连接] {self.symbol.upper()} K线订阅 | 周期: {self.intervals}")
self.ws = create_connection(url, timeout=30)
print("[成功] WebSocket 已连接")
def parse_kline(self, data):
"""解析K线数据"""
kline = data.get('k', {})
return {
'symbol': kline.get('s'),
'interval': kline.get('i'),
'open_time': kline.get('t'),
'open': float(kline.get('o')),
'high': float(kline.get('h')),
'low': float(kline.get('l')),
'close': float(kline.get('c')),
'volume': float(kline.get('v')),
'close_time': kline.get('T'),
'is_closed': kline.get('x'), # K线是否已关闭
}
def on_kline(self, kline_data):
"""K线回调（可被子类重写）"""
k = kline_data
print(f"[K线] {k['symbol']} {k['interval']} | "
f"O:{k['open']} H:{k['high']} L:{k['low']} C:{k['close']} "
f"V:{k['volume']:.2f} | {'✓闭合' if k['is_closed'] else '进行中'}")
# 缓存最新K线
key = f"{k['symbol']}_{k['interval']}"
self.klines[key] = kline_data
# K线闭合时执行策略（如计算指标、发信号）
if k['is_closed']:
self.on_kline_closed(k)
def on_kline_closed(self, kline_data):
"""K线闭合回调（用于策略执行）"""
# 示例：1小时K线闭合时检查突破
if kline_data['interval'] == '1h':
print(f"[信号] {kline_data['symbol']} 1H K线闭合，触发策略检查...")
def run(self):
"""主循环"""
self.connect()
try:
while True:
message = self.ws.recv()
data = json.loads(message)
kline = self.parse_kline(data.get('data', {}))
self.on_kline(kline)
except KeyboardInterrupt:
print("\n[中断] 停止订阅")
finally:
if self.ws:
self.ws.close()
使用示例
if __name__ == "__main__":
# 订阅 BTCUSDT 的 1分钟、5分钟、1小时K线
subscriber = BinanceKlineSubscriber(
symbol='btcusdt',
intervals=['1m', '5m', '1h', '4h']
)
subscriber.run()深度数据（Order Book）订阅
import json
import time
from collections import defaultdict, deque
from websocket import create_connection
class BinanceDepthCache:
"""本地订单簿维护器（支持增量更新）"""
def __init__(self, symbol, level=20, depth_limit=100):
self.symbol = symbol.lower()
self.level = level # 显示档位（如 20档）
self.depth_limit = depth_limit # 订阅档位（最大100）
self.base_url = "wss://stream.binance.com:9443/stream"
self.ws = None
# 本地订单簿缓存
self.bids = {} # {price: quantity}
self.asks = {} # {price: quantity}
self.last_update_id = None
self.snapshot_received = False
# 统计数据
self.update_count = 0
self.spread_history = deque(maxlen=100)
def build_url(self):
"""订阅深度流和最佳买卖价"""
return (f"{self.base_url}?streams="
f"{self.symbol}@depth{self.level}@100ms/"
f"{self.symbol}@bookTicker")
def connect(self):
url = self.build_url()
print(f"[连接] {self.symbol.upper()} 订单簿订阅 | 档位: {self.level}")
self.ws = create_connection(url, timeout=30)
def get_spread(self):
"""计算当前买卖价差"""
if self.bids and self.asks:
best_bid = max(self.bids.keys())
best_ask = min(self.asks.keys())
spread = best_ask - best_bid
spread_pct = (spread / best_bid) * 100
return best_bid, best_ask, spread, spread_pct
return None, None, None, None
def update_depth(self, data):
"""更新本地订单簿"""
# 首次接收全量快照
if 'lastUpdateId' in data:
self.last_update_id = data['lastUpdateId']
self.snapshot_received = True
# 清空旧数据
self.bids.clear()
self.asks.clear()
# 解析快照
for price, qty in data.get('bids', []):
self.bids[float(price)] = float(qty)
for price, qty in data.get('asks', []):
self.asks[float(price)] = float(qty)
print(f"[快照] lastUpdateId={self.last_update_id}, "
f"买卖各 {len(self.bids)}/{len(self.asks)} 档")
# 增量更新
elif 'u' in data:
update_id = data['u']
# 丢弃过期更新
if self.last_update_id and update_id <= self.last_update_id:
return
# 更新 bids
for price, qty in data.get('b', []):
p, q = float(price), float(qty)
if q == 0:
self.bids.pop(p, None)
else:
self.bids[p] = q
# 更新 asks
for price, qty in data.get('a', []):
p, q = float(price), float(qty)
if q == 0:
self.asks.pop(p, None)
else:
self.asks[p] = q
self.last_update_id = update_id
self.update_count += 1
# 排序并截断
self.bids = dict(sorted(self.bids.items(), reverse=True)[:self.level])
self.asks = dict(sorted(self.asks.items())[:self.level])
def display_depth(self):
"""可视化订单簿"""
best_bid, best_ask, spread, spread_pct = self.get_spread()
if not best_bid:
return
self.spread_history.append(spread)
avg_spread = sum(self.spread_history) / len(self.spread_history)
print(f"\n{'='*60}")
print(f"[深度] {self.symbol.upper()} | 更新 #{self.update_count} | "
f"买卖价差: {spread:.2f} ({spread_pct:.3f}%) | 平均: {avg_spread:.2f}")
print(f"{'价格':>12} | {'数量':>12} | {'档位':>6} | {'价格':>12} | {'数量':>12}")
print('-'*60)
bids = list(self.bids.items())[:self.level]
asks = list(self.asks.items())[:self.level]
for i in range(max(len(bids), len(asks))):
bid_p, bid_q = bids[i] if i < len(bids) else (None, None)
ask_p, ask_q = asks[i] if i < len(asks) else (None, None)
bid_str = f"{bid_p:>12.4f} | {bid_q:>12.4f}" if bid_p else " "*37
ask_str = f"{ask_p:>12.4f} | {ask_q:>12.4f}" if ask_p else " "*25
print(f"{bid_str} | {i+1:>6} | {ask_str}")
def on_book_ticker(self, data):
"""最佳买卖价更新"""
bid_p, bid_q = data.get('b', ''), data.get('B', '')
ask_p, ask_q = data.get('a', ''), data.get('A', '')
print(f"[报价] 最优买: {bid_p}×{bid_q} | 最优卖: {ask_p}×{ask_q}")
def run(self, display_interval=2):
"""主循环"""
self.connect()
last_display = time.time()
try:
while True:
message = self.ws.recv()
data = json.loads(message).get('data', {})
if '@depth' in json.loads(message).get('stream', ''):
self.update_depth(data)
elif 'bookTicker' in data:
self.on_book_ticker(data)
# 定期显示深度
if time.time() - last_display >= display_interval:
self.display_depth()
last_display = time.time()
except KeyboardInterrupt:
print("\n[中断] 停止订阅")
finally:
if self.ws:
self.ws.close()
if __name__ == "__main__":
# 订阅 BTCUSDT 20档订单簿
cache = BinanceDepthCache(symbol='btcusdt', level=20)
cache.run(display_interval=3)实战技巧：批量订阅管理器
"""
多交易对批量订阅管理器
功能：
1. 动态增删订阅
2. 流量控制（消息节流）
3. 多进程支持
4. 数据持久化（可选）
"""
import json
import time
import threading
from queue import Queue, Empty
from dataclasses import dataclass, field
from typing import Dict, List, Callable, Optional
from websocket import create_connection, WebSocketException
@dataclass
class StreamConfig:
symbol: str
streams: List[str] # ['trade', 'miniTicker', 'kline_1m']
@dataclass
class MessageStats:
received: int = 0
processed: int = 0
dropped: int = 0
errors: int = 0
start_time: float = field(default_factory=time.time)
class MultiStreamManager:
"""多流管理器"""
def __init__(self, config=None):
self.config = config or {}
self.streams: Dict[str, StreamConfig] = {}
self.ws = None
self.message_queue = Queue(maxsize=10000)
self.stats = MessageStats()
self.running = False
self.handlers: Dict[str, Callable] = {}
# 节流控制
self.throttle_ms = self.config.get('throttle_ms', 100)
self.last_process_time = {}
# 自动重连
self.reconnect_enabled = self.config.get('reconnect', True)
self.max_reconnect_wait = 30
# 数据缓冲（用于批量处理）
self.buffer_size = self.config.get('buffer_size', 100)
self.buffers: Dict[str, List] = {}
def add_stream(self, symbol: str, streams: List[str]):
"""添加订阅流"""
key = symbol.lower()
self.streams[key] = StreamConfig(symbol=key, streams=streams)
self.buffers[key] = []
print(f"[订阅+] {symbol.upper()} → {streams}")
def remove_stream(self, symbol: str):
"""移除订阅流"""
key = symbol.lower()
if key in self.streams:
del self.streams[key]
print(f"[订阅-] {symbol.upper()} 已移除")
def register_handler(self, stream_type: str, handler: Callable):
"""注册消息处理器"""
self.handlers[stream_type] = handler
def build_combined_url(self) -> str:
"""构建组合订阅URL"""
all_streams = []
for cfg in self.streams.values():
for stream in cfg.streams:
all_streams.append(f"{cfg.symbol}@{stream}")
base_url = "wss://stream.binance.com:9443/stream"
return f"{base_url}?streams={'/'.join(all_streams)}"
def connect(self):
"""建立连接"""
if not self.streams:
raise ValueError("至少需要一个订阅流")
url = self.build_combined_url()
print(f"[连接] 订阅 {len(self.streams)} 个交易对, {sum(len(s.streams) for s in self.streams.values())} 个流")
print(f"[URL] {url[:100]}...")
self.ws = create_connection(url, timeout=30)
self.running = True
self.stats.start_time = time.time()
def process_message(self, stream: str, data: dict):
"""消息处理（含节流）"""
# 提取流类型
parts = stream.split('@')
symbol = parts[0]
stream_type = parts[1] if len(parts) > 1 else 'unknown'
# 节流控制
throttle_key = f"{symbol}_{stream_type}"
now = time.time() * 1000
if throttle_key in self.last_process_time:
elapsed = now - self.last_process_time[throttle_key]
if elapsed < self.throttle_ms:
self.stats.dropped += 1
return
self.last_process_time[throttle_key] = now
self.stats.processed += 1
# 调用注册的处理器
if stream_type in self.handlers:
try:
self.handlers[stream_type](symbol, data)
except Exception as e:
print(f"[处理错误] {stream_type}: {e}")
self.stats.errors += 1
# 默认处理
if stream_type == 'trade':
print(f"[成交] {data.get('s')} {data.get('p')}×{data.get('q')}")
elif stream_type == 'miniTicker':
print(f"[行情] {data.get('s')} C:{data.get('c')} O:{data.get('o')}")
def receive_loop(self):
"""接收循环"""
while self.running:
try:
self.ws.settimeout(1)
message = self.ws.recv()
self.stats.received += 1
try:
parsed = json.loads(message)
stream = parsed.get('stream', '')
data = parsed.get('data', {})
self.process_message(stream, data)
except json.JSONDecodeError:
pass
except Empty:
continue
except WebSocketException as e:
print(f"[错误] {e}")
if self.reconnect_enabled:
self.attempt_reconnect()
else:
self.running = False
def attempt_reconnect(self, attempt=1):
"""重连尝试"""
if not self.reconnect_enabled:
return
wait_time = min(2 ** attempt, self.max_reconnect_wait)
print(f"[重连] {wait_time}秒后重试 (第{attempt}次)...")
time.sleep(wait_time)
try:
self.connect()
print("[重连] 成功")
except Exception as e:
print(f"[重连] 失败: {e}")
self.attempt_reconnect(attempt + 1)
def print_stats(self):
"""打印统计信息"""
elapsed = time.time() - self.stats.start_time
rate = self.stats.received / elapsed if elapsed > 0 else 0
print(f"\n[统计] 运行: {elapsed:.1f}s | "
f"接收: {self.stats.received} | "
f"处理: {self.stats.processed} | "
f"丢弃: {self.stats.dropped} | "
f"错误: {self.stats.errors} | "
f"速率: {rate:.1f} msg/s")
def run(self, duration=None):
"""运行"""
try:
self.connect()
stats_thread = threading.Thread(target=self._stats_loop, daemon=True)
stats_thread.start()
self.receive_loop()
except KeyboardInterrupt:
print("\n[中断]")
finally:
self.running = False
if self.ws:
self.ws.close()
self.print_stats()
def _stats_loop(self):
while self.running:
time.sleep(10)
self.print_stats()
使用示例
if __name__ == "__main__":
manager = MultiStreamManager({
'throttle_ms': 50, # 50ms 节流
'reconnect': True,
})
# 添加多个交易对
symbols = ['btcusdt', 'ethusdt', 'bnbusdt', 'solusdt', 'adausdt']
for sym in symbols:
manager.add_stream(sym, ['trade', 'miniTicker'])
# 自定义处理器
manager.register_handler('trade', lambda s, d: print(f"[自定义] {s}: {d.get('p')}"))
# 运行指定时长（秒）
manager.run(duration=30)常见报错排查
1. 连接被拒绝：WebSocket connection failed
错误信息：
WebSocketException: connection closed unexpectedly
可能原因：
- 网络代理/防火墙