大家好，我是 HolySheep AI 的技术作者。今天跟大家聊聊我踩过的一个大坑——Binance WebSocket 断开重连的问题。
刚开始做量化交易的时候，我以为只要连上 Binance 的 WebSocket 就完事了。结果运行不到10分钟，连接就悄悄断了，订单数据全丢了，血亏。后来才知道，Binance 服务器会在 3分钟 后主动断开空闲连接，心跳机制才是 WebSocket 稳定运行的生命线。
这篇文章，我会从零开始，手把手教你实现一个健壮的 WebSocket 管理器，包含心跳检测、自动重连、异常处理。先剧透一下：如果你觉得自建太麻烦，立即注册 HolySheep API，它提供国内直连服务，延迟低于50ms，还能省去自己维护 WebSocket 的麻烦。
一、WebSocket 是什么？为什么要心跳？
1.1 概念解释（小白版）
想象你在跟朋友打电话。WebSocket 就像这条电话线，一旦接通，双方可以随时互相发消息，不像 HTTP 那样每次都要重新拨号。
问题来了：如果电话线一直开着但没人说话，运营商会以为线路坏了，直接给你断掉。Binance 服务器也是这样——如果你的连接 3分钟没有任何数据交互，它就会强制关闭。
1.2 心跳机制原理
心跳（Heartbeat）就是"我还活着"的信号。就像综艺节目里主持人喊"还有人吗？"，你要喊"在呢"回应。
Binance WebSocket 心跳流程：
- 客户端每 3分钟（180秒） 发送一个
{"method": "PING"} - 服务器回复
{"result": null, "id": 1} - 如果 60秒内没收到回复，说明连接已死，需要重连
二、为什么必须实现重连机制？
我自己踩过的坑：
- 网络抖动：路由器断一下，连接就断了
- 服务器维护：Binance 偶尔会重启服务
- IP被限流：请求太频繁会被临时封IP
- 本地程序崩溃：你的代码出bug了，进程直接退出
没有重连机制，意味着每次断线都是灾难。2019年有一次，我凌晨2点写的套利机器人断线了，早上起来一看，错过了一整波行情，白干三天。
三、Python 实现：心跳与重连管理器
3.1 基础版本（适合理解原理）
import time
import json
import threading
import websocket
class BinanceWebSocketManager:
"""Binance WebSocket 基础管理器 - 含心跳与重连"""
def __init__(self, stream_url="wss://stream.binance.com:9443/ws"):
self.url = stream_url
self.ws = None
self.connected = False
self.heartbeat_interval = 60 # 发送心跳间隔（秒）
self.last_ping_time = 0
self.max_reconnect_attempts = 10 # 最大重连次数
self.reconnect_delay = 1 # 初始重连延迟（秒）
self.listen_key = None
self._running = False
def connect(self, streams=None):
"""建立连接"""
if streams:
# 组合多 streams URL
stream_str = "/".join(streams)
url = f"wss://stream.binance.com:9443/stream?streams={stream_str}"
else:
url = self.url
print(f"[连接中] {url}")
try:
self.ws = websocket.WebSocketApp(
url,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
self._running = True
# 在独立线程中运行
self.ws_thread = threading.Thread(target=self._run, daemon=True)
self.ws_thread.start()
return True
except Exception as e:
print(f"[连接失败] {e}")
return False
def _run(self):
"""WebSocket 运行循环"""
while self._running:
if self.ws:
try:
self.ws.run_forever(ping_interval=30, ping_timeout=10)
except Exception as e:
print(f"[运行异常] {e}")
if self._running and self.connected:
self._reconnect()
def _on_open(self, ws):
"""连接建立时调用"""
print("[连接成功] WebSocket 已建立")
self.connected = True
self.last_ping_time = time.time()
def _on_message(self, ws, message):
"""收到消息时调用"""
try:
data = json.loads(message)
# 处理心跳响应
if data.get("result") is None and "id" in data:
print(f"[心跳响应] id={data['id']}")
self.last_ping_time = time.time()
else:
# 处理业务数据
self._handle_data(data)
except json.JSONDecodeError as e:
print(f"[解析错误] {e}, 原始数据: {message[:100]}")
def _handle_data(self, data):
"""处理业务数据 - 子类可重写"""
print(f"[收到数据] {json.dumps(data)[:100]}...")
def _on_error(self, ws, error):
"""错误处理"""
print(f"[WebSocket错误] {error}")
self.connected = False
def _on_close(self, ws, close_status_code, close_msg):
"""连接关闭时调用"""
print(f"[连接关闭] 状态码: {close_status_code}, 消息: {close_msg}")
self.connected = False
def _reconnect(self):
"""自动重连逻辑"""
print("[重连] 尝试重新连接...")
self._running = True
for attempt in range(1, self.max_reconnect_attempts + 1):
if not self._running:
break
# 指数退避：1s -> 2s -> 4s -> 8s... 最大32秒
delay = min(self.reconnect_delay * (2 ** (attempt - 1)), 32)
print(f"[重连] 第 {attempt}/{self.max_reconnect_attempts} 次，等待 {delay} 秒...")
time.sleep(delay)
if self.connect():
print("[重连] 成功！")
return
if attempt == self.max_reconnect_attempts:
print("[重连] 达到最大次数，停止重连")
self._running = False
def send_ping(self):
"""手动发送心跳"""
if self.ws and self.connected:
try:
self.ws.send(json.dumps({"method": "PING", "id": int(time.time())}))
self.last_ping_time = time.time()
print("[心跳] 已发送 PING")
except Exception as e:
print(f"[心跳失败] {e}")
def start_heartbeat(self):
"""启动心跳定时器"""
def heartbeat_loop():
while self._running:
if self.connected:
self.send_ping()
time.sleep(self.heartbeat_interval)
thread = threading.Thread(target=heartbeat_loop, daemon=True)
thread.start()
def close(self):
"""关闭连接"""
print("[关闭] 正在关闭 WebSocket...")
self._running = False
if self.ws:
self.ws.close()
self.connected = False
使用示例
if __name__ == "__main__":
manager = BinanceWebSocketManager()
# 订阅 K线 和 深度数据
streams = ["btcusdt@kline_1m", "btcusdt@depth20@100ms"]
if manager.connect(streams):
manager.start_heartbeat() # 启动心跳
time.sleep(60) # 运行60秒测试
manager.close()
else:
print("[错误] 无法建立连接")3.2 生产级版本（带订单簿和成交记录）
import time
import json
import logging
import threading
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Optional, Dict, List
from enum import Enum
import websocket
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ConnectionState(Enum):
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
RECONNECTING = "reconnecting"
FAILED = "failed"
@dataclass
class WebSocketConfig:
"""WebSocket 配置"""
stream_url: str = "wss://stream.binance.com:9443/ws"
ping_interval: int = 30 # ping 间隔（秒）
ping_timeout: int = 10 # ping 超时（秒）
reconnect_max_attempts: int = 20
reconnect_base_delay: float = 1.0
reconnect_max_delay: float = 60.0
heartbeat_check_interval: int = 30
heartbeat_timeout: int = 120 # 120秒没响应则认为断开
@dataclass
class OrderBookSnapshot:
"""订单簿快照"""
symbol: str
bids: Dict[str, str] = field(default_factory=dict) # 价格 -> 数量
asks: Dict[str, str] = field(default_factory=dict)
last_update_id: int = 0
class BinanceWebSocketManagerV2:
"""
生产级 Binance WebSocket 管理器
特性：
- 自动心跳检测
- 指数退避重连
- 订单簿本地维护
- 成交记录缓存
- 断线重连后自动订阅恢复
"""
def __init__(self, config: Optional[WebSocketConfig] = None):
self.config = config or WebSocketConfig()
self.ws: Optional[websocket.WebSocketApp] = None
self.state = ConnectionState.DISCONNECTED
self.streams: List[str] = []
self.subscribed_streams: set = set()
# 数据存储
self.order_books: Dict[str, OrderBookSnapshot] = {}
self.trade_cache: deque = deque(maxlen=1000)
# 回调函数
self.on_orderbook_update: Optional[Callable] = None
self.on_trade: Optional[Callable] = None
self.on_kline: Optional[Callable] = None
self.on_error: Optional[Callable] = None
self.on_state_change: Optional[Callable] = None
# 内部状态
self._running = False
self._ws_thread: Optional[threading.Thread] = None
self._heartbeat_thread: Optional[threading.Thread] = None
self._reconnect_attempt = 0
self._last_message_time = time.time()
self._lock = threading.Lock()
# 连接统计
self.stats = {
"messages_received": 0,
"reconnections": 0,
"last_error": None
}
def set_state(self, new_state: ConnectionState):
"""状态变更"""
if self.state != new_state:
logger.info(f"状态变更: {self.state.value} -> {new_state.value}")
self.state = new_state
if self.on_state_change:
self.on_state_change(new_state)
def connect(self, streams: List[str]) -> bool:
"""
连接到指定 streams
Args:
streams: 订阅列表，如 ["btcusdt@kline_1m", "btcusdt@depth20"]
Returns:
连接是否成功
"""
with self._lock:
self.streams = streams
self._reconnect_attempt = 0
return self._do_connect()
def _do_connect(self) -> bool:
"""执行连接"""
self.set_state(ConnectionState.CONNECTING)
# 构建多流 URL
stream_param = "/".join(self.streams)
url = f"wss://stream.binance.com:9443/stream?streams={stream_param}"
logger.info(f"正在连接: {url}")
try:
self.ws = websocket.WebSocketApp(
url,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
self._running = True
# 启动 WebSocket 线程
self._ws_thread = threading.Thread(
target=self._ws_run_loop,
name="websocket-runner",
daemon=True
)
self._ws_thread.start()
# 启动心跳检测线程
self._start_heartbeat_monitor()
return True
except Exception as e:
logger.error(f"连接失败: {e}")
self.set_state(ConnectionState.FAILED)
return False
def _ws_run_loop(self):
"""WebSocket 运行循环"""
while self._running:
if self.ws and self.state in [ConnectionState.CONNECTING, ConnectionState.CONNECTED]:
try:
self.ws.run_forever(
ping_interval=self.config.ping_interval,
ping_timeout=self.config.ping_timeout
)
except Exception as e:
logger.error(f"WebSocket 运行异常: {e}")
# 连接断开后的重连逻辑
if self._running and self.state != ConnectionState.CONNECTED:
self._schedule_reconnect()
def _on_open(self, ws):
"""连接建立"""
logger.info("WebSocket 连接已建立")
with self._lock:
self.subscribed_streams = set(self.streams)
self.set_state(ConnectionState.CONNECTED)
self._reconnect_attempt = 0
self._last_message_time = time.time()
def _on_message(self, ws, message: str):
"""消息处理"""
self.stats["messages_received"] += 1
self._last_message_time = time.time()
try:
data = json.loads(message)
# Binance Combined Stream 格式
if "stream" in data and "data" in data:
stream = data["stream"]
payload = data["data"]
else:
return
# 分发到对应处理器
if "depth" in stream:
self._handle_depth_update(stream, payload)
elif "kline" in stream:
self._handle_kline_update(stream, payload)
elif "trade" in stream:
self._handle_trade_update(stream, payload)
elif "aggTrade" in stream:
self._handle_agg_trade(stream, payload)
else:
logger.debug(f"未处理的数据类型: {stream}")
except json.JSONDecodeError as e:
logger.error(f"JSON 解析错误: {e}")
except Exception as e:
logger.error(f"消息处理异常: {e}")
def _handle_depth_update(self, stream: str, data: dict):
"""处理订单簿更新"""
symbol = stream.split("@")[0].upper()
if symbol not in self.order_books:
self.order_books[symbol] = OrderBookSnapshot(symbol=symbol)
ob = self.order_books[symbol]
ob.last_update_id = data.get("u", 0)
# 更新本地订单簿
for price, qty in data.get("b", []):
if float(qty) == 0:
ob.bids.pop(price, None)
else:
ob.bids[price] = qty
for price, qty in data.get("a", []):
if float(qty) == 0:
ob.asks.pop(price, None)
else:
ob.asks[price] = qty
if self.on_orderbook_update:
self.on_orderbook_update(symbol, ob)
def _handle_kline_update(self, stream: str, data: dict):
"""处理 K线更新"""
kline = data.get("k", {})
kline_data = {
"symbol": kline.get("s"),
"interval": kline.get("i"),
"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")),
"closed": kline.get("x") # K线是否关闭
}
if self.on_kline:
self.on_kline(kline_data)
def _handle_agg_trade(self, stream: str, data: dict):
"""处理聚合成交"""
trade = {
"symbol": data.get("s"),
"price": float(data.get("p")),
"quantity": float(data.get("q")),
"timestamp": data.get("T"),
"is_buyer_maker": data.get("m")
}
self.trade_cache.append(trade)
if self.on_trade:
self.on_trade(trade)
def _handle_trade_update(self, stream: str, data: dict):
"""处理单个成交"""
self._handle_agg_trade(stream, data)
def _on_error(self, ws, error):
"""错误处理"""
logger.error(f"WebSocket 错误: {error}")
self.stats["last_error"] = str(error)
if self.on_error:
self.on_error(error)
self.set_state(ConnectionState.FAILED)
def _on_close(self, ws, close_status_code, close_msg):
"""连接关闭"""
logger.info(f"连接关闭: {close_status_code} - {close_msg}")
self.set_state(ConnectionState.DISCONNECTED)
def _start_heartbeat_monitor(self):
"""启动心跳监控"""
def heartbeat_monitor():
while self._running:
try:
if self.state == ConnectionState.CONNECTED:
elapsed = time.time() - self._last_message_time
if elapsed > self.config.heartbeat_timeout:
logger.warning(
f"心跳超时: {elapsed:.1f}秒内无消息，触发重连"
)
self._schedule_reconnect()
break
elif elapsed > self.config.heartbeat_check_interval:
logger.debug(f"心跳检查: 上次消息 {elapsed:.1f}秒前")
time.sleep(5) # 每5秒检查一次
except Exception as e:
logger.error(f"心跳监控异常: {e}")
if self._heartbeat_thread and self._heartbeat_thread.is_alive():
return
self._heartbeat_thread = threading.Thread(
target=heartbeat_monitor,
name="heartbeat-monitor",
daemon=True
)
self._heartbeat_thread.start()
def _schedule_reconnect(self):
"""调度重连（指数退避）"""
self.set_state(ConnectionState.RECONNECTING)
with self._lock:
self._reconnect_attempt += 1
attempt = self._reconnect_attempt
if attempt > self.config.reconnect_max_attempts:
logger.error(f"已达到最大重连次数 ({self.config.reconnect_max_attempts})，停止重连")
self.set_state(ConnectionState.FAILED)
return
# 指数退避计算：1, 2, 4, 8, 16, 32, 60, 60...
delay = min(
self.config.reconnect_base_delay * (2 ** (attempt - 1)),
self.config.reconnect_max_delay
)
logger.info(f"计划 {delay:.1f}秒后重连 (第{attempt}次)")
# 在新线程中等待并重连
threading.Thread(
target=lambda: self._delayed_reconnect(delay),
daemon=True
).start()
def _delayed_reconnect(self, delay: float):
"""延迟重连"""
time.sleep(delay)
self.stats["reconnections"] += 1
self._do_connect()
def get_order_book(self, symbol: str) -> Optional[OrderBookSnapshot]:
"""获取订单簿快照"""
return self.order_books.get(symbol.upper())
def get_spread(self, symbol: str) -> Optional[float]:
"""获取当前买卖价差"""
ob = self.get_order_book(symbol)
if not ob or not ob.asks or not ob.bids:
return None
best_ask = min(float(p) for p in ob.asks.keys())
best_bid = max(float(p) for p in ob.bids.keys())
return best_ask - best_bid
def close(self):
"""关闭连接"""
logger.info("正在关闭 WebSocket...")
self._running = False
if self.ws:
try:
self.ws.close()
except Exception:
pass
self.set_state(ConnectionState.DISCONNECTED)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
==================== 使用示例 ====================
def main():
"""完整使用示例"""
manager = BinanceWebSocketManagerV2()
# 定义回调函数
def on_orderbook(symbol, orderbook):
spread = manager.get_spread(symbol)
best_bid = max(orderbook.bids.keys())
best_ask = min(orderbook.asks.keys())
print(f"[订单簿] {symbol} | 买: {best_bid} | 卖: {best_ask} | 价差: {spread}")
def on_kline(kline):
if kline["closed"]: # 只打印收线的K线
print(f"[K线] {kline['symbol']} | 收: {kline['close']} | 量: {kline['volume']}")
def on_trade(trade):
print(f"[成交] {trade['symbol']} | 价格: {trade['price']} | 数量: {trade['quantity']}")
def on_state_change(state):
print(f"[状态] {state.value}")
def on_error(error):
print(f"[错误] {error}")
# 设置回调
manager.on_orderbook_update = on_orderbook
manager.on_kline = on_kline
manager.on_trade = on_trade
manager.on_state_change = on_state_change
manager.on_error = on_error
# 订阅数据流
streams = [
"btcusdt@depth20@100ms", # 订单簿（20档，100ms更新）
"btcusdt@kline_1m", # 1分钟K线
"btcusdt@aggTrade" # 聚合成交
]
print("开始连接 Binance WebSocket...")
if manager.connect(streams):
print(f"连接成功！统计: {manager.stats}")
try:
# 持续运行
while True:
time.sleep(1)
except KeyboardInterrupt:
print("\n收到中断信号")
else:
print("连接失败")
manager.close()
print(f"最终统计: {manager.stats}")
if __name__ == "__main__":
main()3.3 JavaScript/Node.js 版本
// Binance WebSocket 管理器 - Node.js 版本
// 支持心跳、自动重连、订单簿维护
const WebSocket = require('ws');
class BinanceWebSocketManager {
constructor(options = {}) {
this.streams = options.streams || [];
this.baseUrl = 'wss://stream.binance.com:9443/stream';
// 配置参数
this.config = {
pingInterval: 30000, // ping 间隔（毫秒）
pingTimeout: 10000, // ping 超时
reconnectBaseDelay: 1000, // 基础重连延迟
reconnectMaxDelay: 60000, // 最大重连延迟
maxReconnectAttempts: 20,
heartbeatCheckInterval: 30000 // 心跳检查间隔
};
// 状态
this.ws = null;
this.state = 'DISCONNECTED';
this.reconnectAttempt = 0;
this.lastMessageTime = Date.now();
this.pingTimer = null;
this.reconnectTimer = null;
this.heartbeatTimer = null;
// 订单簿存储
this.orderBooks = new Map();
// 回调函数
this.callbacks = {
onOrderbookUpdate: options.onOrderbookUpdate || (() => {}),
onTrade: options.onTrade || (() => {}),
onKline: options.onKline || (() => {}),
onError: options.onError || console.error,
onStateChange: options.onStateChange || (() => {}),
onReconnect: options.onReconnect || (() => {})
};
}
setState(newState) {
if (this.state !== newState) {
console.log([状态变更] ${this.state} -> ${newState});
this.state = newState;
this.callbacks.onStateChange(newState);
}
}
connect(streams) {
this.streams = streams;
this.reconnectAttempt = 0;
return this._doConnect();
}
_doConnect() {
return new Promise((resolve, reject) => {
this.setState('CONNECTING');
const streamParam = this.streams.join('/');
const url = ${this.baseUrl}?streams=${streamParam};
console.log([连接中] ${url});
try {
this.ws = new WebSocket(url);
this.ws.on('open', () => {
console.log('[连接成功] WebSocket 已打开');
this.setState('CONNECTED');
this.reconnectAttempt = 0;
this.lastMessageTime = Date.now();
this._startPing();
this._startHeartbeatCheck();
resolve(true);
});
this.ws.on('message', (data) => {
this._handleMessage(data);
});
this.ws.on('error', (error) => {
console.error('[WebSocket错误]', error);
this.callbacks.onError(error);
this.setState('FAILED');
reject(error);
});
this.ws.on('close', (code, reason) => {
console.log([连接关闭] ${code} - ${reason});
this._stopTimers();
this.setState('DISCONNECTED');
this._scheduleReconnect();
});
} catch (error) {
console.error('[连接异常]', error);
this.setState('FAILED');
reject(error);
}
});
}
_handleMessage(data) {
this.lastMessageTime = Date.now();
try {
const message = JSON.parse(data);
// Binance Combined Stream 格式
if (message.stream && message.data) {
const stream = message.stream;
const payload = message.data;
if (stream.includes('@depth')) {
this._handleDepthUpdate(stream, payload);
} else if (stream.includes('@kline')) {
this._handleKlineUpdate(payload);
} else if (stream.includes('@aggTrade')) {
this._handleAggTrade(payload);
}
}
} catch (error) {
console.error('[消息解析错误]', error);
}
}
_handleDepthUpdate(stream, data) {
const symbol = stream.split('@')[0].toUpperCase();
if (!this.orderBooks.has(symbol)) {
this.orderBooks.set(symbol, { bids: {}, asks: {} });
}
const book = this.orderBooks.get(symbol);
// 更新 bids
for (const [price, qty] of data.b || []) {
if (parseFloat(qty) === 0) {
delete book.bids[price];
} else {
book.bids[price] = qty;
}
}
// 更新 asks
for (const [price, qty] of data.a || []) {
if (parseFloat(qty) === 0) {
delete book.asks[price];
} else {
book.asks[price] = qty;
}
}
this.callbacks.onOrderbookUpdate(symbol, book);
}
_handleKlineUpdate(data) {
const kline = data.k;
this.callbacks.onKline({
symbol: kline.s,
interval: kline.i,
open: parseFloat(kline.o),
high: parseFloat(kline.h),
low: parseFloat(kline.l),
close: parseFloat(kline.c),
volume: parseFloat(kline.v),
closed: kline.x
});
}
_handleAggTrade(data) {
this.callbacks.onTrade({
symbol: data.s,
price: parseFloat(data.p),
quantity: parseFloat(data.q),
timestamp: data.T,
isBuyerMaker: data.m
});
}
_startPing() {
// WebSocket 库会自动处理 ping/pong，但我们手动发一个应用层心跳
this.pingTimer = setInterval(() => {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
// 发送订阅请求作为心跳（带不同 ID）
const pingId = Date.now();
this.ws.send(JSON.stringify({
method: 'SUBSCRIBE',
params: ['btcusdt@trade'],
id: pingId
}));
console.log([心跳] 发送 ping id=${pingId});
}
}, this.config.pingInterval);
}
_startHeartbeatCheck() {
this.heartbeatTimer = setInterval(() => {
const elapsed = Date.now() - this.lastMessageTime;
if (elapsed > this.config.heartbeatCheckInterval * 2) {
console.warn([心跳检查] ${elapsed}ms 无消息，可能已断开);
}
// 如果超时太久，强制关闭触发重连
if (elapsed > this.config.pingTimeout * 3) {
console.error('[心跳超时] 强制断开连接');
this.ws?.terminate();
}
}, this.config.heartbeatCheckInterval);
}
_stopTimers() {
if (this.pingTimer) clearInterval(this.pingTimer);
if (this.heartbeatTimer) clearInterval(this.heartbeatTimer);
if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
}
_scheduleReconnect() {
if (this.reconnectAttempt >= this.config.maxReconnectAttempts) {
console.error('[重连] 达到最大重连次数，停止');
this.setState('FAILED');
return;
}
this.reconnectAttempt++;
// 指数退避
const delay = Math.min(
this.config.reconnectBaseDelay * Math.pow(2, this.reconnectAttempt - 1),
this.config.reconnectMaxDelay
);
console.log([重连] ${delay}ms 后重试 (第${this.reconnectAttempt}次));
this.callbacks.onReconnect(this.reconnectAttempt, delay);
this.reconnectTimer = setTimeout(() => {
this._doConnect().catch(() => {});
}, delay);
}
// 主动订阅新 streams
subscribe(streams) {
if (this.ws?.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify({
method: 'SUBSCRIBE',
params: streams,
id: Date.now()
}));
this.streams.push(...streams);
console.log([订阅] ${streams.join(', ')});
}
}
// 取消订阅
unsubscribe(streams) {
if (this.ws?.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify({
method: 'UNSUBSCRIBE',
params: streams,
id: Date.now()
}));
this.streams = this.streams.filter(s => !streams.includes(s));
console.log([取消订阅] ${streams.join(', ')});
}
}
getOrderBook(symbol) {
return this.orderBooks.get(symbol.toUpperCase());
}
getSpread(symbol) {
const book = this.getOrderBook(symbol);
if (!book) return null;
const bestBid = Math.max(...Object.keys(book.bids).map(Number));
const bestAsk = Math.min(...Object.keys(book.asks).map(Number));
return bestAsk - bestBid;
}
close() {
console.log('[关闭] 正在关闭 WebSocket...');
this._stopTimers();
this.setState('DISCONNECTED');
this.ws?.close();
}
}
// ==================== 使用示例 ====================
const streams = [
'btcusdt@depth20@100ms',
'btcusdt@kline_1m',
'btcusdt@aggTrade'
];
const manager = new BinanceWebSocketManager({
streams,
onOrderbookUpdate: (symbol, book) => {
const spread = manager.getSpread(symbol);
const bestBid = Math.max(...Object.keys(book.bids).map(Number));
const bestAsk = Math.min(...Object.keys(book.asks).map(Number));
console.log([订单簿] ${symbol} | 买:${bestBid} 卖:${bestAsk} 价差:${spread});
},
onKline: (kline) => {
if (kline.closed) {
console.log([K线] ${kline.symbol} 收:${kline.close} 量:${kline.volume});
}
},
onTrade: (trade) => {
console.log([成交] ${trade.symbol} 价格:${trade.price});
},
onStateChange: (state) => {
console.log(`