From d02001ef5e749ed4b4ae3b270f403ea0078dbe03 Mon Sep 17 00:00:00 2001 From: Praying Date: Sun, 25 Jan 2026 20:31:23 +0800 Subject: [PATCH 1/3] fix(binance): refactor WebSocket client to fix subscription payload and state management Refactor BinanceWs to use MessageRouter and SubscriptionManager. Fix subscription payload format to satisfy Binance requirements. Implement watch_*_internal helper methods. Update Binance exchange implementation to use new internal methods. Ensure synchronous subscriptions() method. Add integration test for payload verification. --- .gitignore | 3 + ccxt-exchanges/src/binance/ws/binance_impl.rs | 332 +------- ccxt-exchanges/src/binance/ws/handlers.rs | 28 +- ccxt-exchanges/src/binance/ws/mod.rs | 730 +++++++++++++++--- .../src/binance/ws/subscriptions.rs | 9 + .../binance_ws_subscribe_payload_test.rs | 57 ++ 6 files changed, 739 insertions(+), 420 deletions(-) create mode 100644 ccxt-exchanges/tests/binance_ws_subscribe_payload_test.rs diff --git a/.gitignore b/.gitignore index e13ee82..d7ffe45 100644 --- a/.gitignore +++ b/.gitignore @@ -47,3 +47,6 @@ _bmad/ _bmad-output/ docs/ AGENTS.md +CLAUDE.md +GEMINI.md + diff --git a/ccxt-exchanges/src/binance/ws/binance_impl.rs b/ccxt-exchanges/src/binance/ws/binance_impl.rs index 67a9be7..086b247 100644 --- a/ccxt-exchanges/src/binance/ws/binance_impl.rs +++ b/ccxt-exchanges/src/binance/ws/binance_impl.rs @@ -276,9 +276,6 @@ impl Binance { since: Option, limit: Option, ) -> Result> { - const MAX_RETRIES: u32 = 50; - const MAX_TRADES: usize = 1000; - self.base.load_markets(false).await?; let market = self.base.market(symbol).await?; @@ -287,48 +284,8 @@ impl Binance { let ws = self.create_ws(); ws.connect().await?; - ws.subscribe_trades(&binance_symbol).await?; - - let mut retries = 0; - - while retries < MAX_RETRIES { - if let Some(msg) = ws.client.receive().await { - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - if let Ok(trade) = parser::parse_ws_trade(&msg, Some(&market)) { - let mut trades_map = ws.trades.lock().await; - let trades = trades_map - .entry(symbol.to_string()) - .or_insert_with(VecDeque::new); - - if trades.len() >= MAX_TRADES { - trades.pop_front(); - } - trades.push_back(trade); - - let mut result: Vec = trades.iter().cloned().collect(); - - if let Some(since_ts) = since { - result.retain(|t| t.timestamp >= since_ts); - } - - if let Some(limit_size) = limit { - if result.len() > limit_size { - result = result.split_off(result.len() - limit_size); - } - } - - return Ok(result); - } - } - - retries += 1; - tokio::time::sleep(Duration::from_millis(100)).await; - } - - Err(Error::network("Timeout waiting for trade data")) + ws.watch_trades_internal(symbol, &binance_symbol, since, limit, Some(&market)) + .await } /// Streams OHLCV data for a unified symbol @@ -339,9 +296,6 @@ impl Binance { since: Option, limit: Option, ) -> Result> { - const MAX_RETRIES: u32 = 50; - const MAX_OHLCVS: usize = 1000; - self.base.load_markets(false).await?; let market = self.base.market(symbol).await?; @@ -350,53 +304,12 @@ impl Binance { let ws = self.create_ws(); ws.connect().await?; - ws.subscribe_kline(&binance_symbol, timeframe).await?; - - let mut retries = 0; - - while retries < MAX_RETRIES { - if let Some(msg) = ws.client.receive().await { - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - if let Ok(ohlcv) = parser::parse_ws_ohlcv(&msg) { - let cache_key = format!("{}:{}", symbol, timeframe); - let mut ohlcvs_map = ws.ohlcvs.lock().await; - let ohlcvs = ohlcvs_map.entry(cache_key).or_insert_with(VecDeque::new); - - if ohlcvs.len() >= MAX_OHLCVS { - ohlcvs.pop_front(); - } - ohlcvs.push_back(ohlcv); - - let mut result: Vec = ohlcvs.iter().cloned().collect(); - - if let Some(since_ts) = since { - result.retain(|o| o.timestamp >= since_ts); - } - - if let Some(limit_size) = limit { - if result.len() > limit_size { - result = result.split_off(result.len() - limit_size); - } - } - - return Ok(result); - } - } - - retries += 1; - tokio::time::sleep(Duration::from_millis(100)).await; - } - - Err(Error::network("Timeout waiting for OHLCV data")) + ws.watch_ohlcv_internal(symbol, &binance_symbol, timeframe, since, limit) + .await } /// Streams the best bid/ask data for a unified symbol pub async fn watch_bids_asks(&self, symbol: &str) -> Result { - const MAX_RETRIES: u32 = 50; - self.base.load_markets(false).await?; let market = self.base.market(symbol).await?; @@ -405,32 +318,7 @@ impl Binance { let ws = self.create_ws(); ws.connect().await?; - let stream_name = format!("{}@bookTicker", binance_symbol); - ws.client - .subscribe(stream_name, Some(symbol.to_string()), None) - .await?; - - let mut retries = 0; - - while retries < MAX_RETRIES { - if let Some(msg) = ws.client.receive().await { - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - if let Ok(bid_ask) = parser::parse_ws_bid_ask(&msg) { - let mut bids_asks_map = ws.bids_asks.lock().await; - bids_asks_map.insert(symbol.to_string(), bid_ask.clone()); - - return Ok(bid_ask); - } - } - - retries += 1; - tokio::time::sleep(Duration::from_millis(100)).await; - } - - Err(Error::network("Timeout waiting for BidAsk data")) + ws.watch_bids_asks_internal(symbol, &binance_symbol).await } /// Streams account balance changes (private user data stream) @@ -438,8 +326,6 @@ impl Binance { self: Arc, params: Option>, ) -> Result { - const MAX_RETRIES: u32 = 100; - self.base.load_markets(false).await?; let account_type = if let Some(p) = ¶ms { @@ -467,7 +353,6 @@ impl Binance { }; let ws = self.create_authenticated_ws(); - ws.connect().await?; if fetch_snapshot { let account_type_enum = account_type.parse::().ok(); @@ -481,34 +366,7 @@ impl Binance { } } - let mut retries = 0; - - while retries < MAX_RETRIES { - if let Some(msg) = ws.client.receive().await { - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - if let Some(event_type) = msg.get("e").and_then(|e| e.as_str()) { - if matches!( - event_type, - "balanceUpdate" | "outboundAccountPosition" | "ACCOUNT_UPDATE" - ) { - if let Ok(()) = ws.handle_balance_message(&msg, account_type).await { - let balances = ws.balances.read().await; - if let Some(balance) = balances.get(account_type) { - return Ok(balance.clone()); - } - } - } - } - } - - retries += 1; - tokio::time::sleep(Duration::from_millis(100)).await; - } - - Err(Error::network("Timeout waiting for balance data")) + ws.watch_balance_internal(account_type).await } /// Watches authenticated order updates via the user data stream @@ -522,89 +380,7 @@ impl Binance { self.base.load_markets(false).await?; let ws = self.create_authenticated_ws(); - ws.connect().await?; - - loop { - if let Some(msg) = ws.client.receive().await { - if let Value::Object(data) = msg { - if let Some(event_type) = data.get("e").and_then(serde_json::Value::as_str) { - if event_type == "executionReport" { - let order = user_data::parse_ws_order(&data); - - let mut orders = ws.orders.write().await; - let symbol_orders = orders - .entry(order.symbol.clone()) - .or_insert_with(HashMap::new); - symbol_orders.insert(order.id.clone(), order.clone()); - drop(orders); - - if let Some(exec_type) = - data.get("x").and_then(serde_json::Value::as_str) - { - if exec_type == "TRADE" { - if let Ok(trade) = - BinanceWs::parse_ws_trade(&Value::Object(data.clone())) - { - let mut trades = ws.my_trades.write().await; - let symbol_trades = trades - .entry(trade.symbol.clone()) - .or_insert_with(VecDeque::new); - - symbol_trades.push_front(trade); - if symbol_trades.len() > 1000 { - symbol_trades.pop_back(); - } - } - } - } - - return self.filter_orders(&ws, symbol, since, limit).await; - } - } - } - } else { - tokio::time::sleep(Duration::from_millis(100)).await; - } - } - } - - /// Filters cached orders by symbol, time range, and limit - async fn filter_orders( - &self, - ws: &BinanceWs, - symbol: Option<&str>, - since: Option, - limit: Option, - ) -> Result> { - let orders_map = ws.orders.read().await; - - let mut orders: Vec = if let Some(sym) = symbol { - orders_map - .get(sym) - .map(|symbol_orders| symbol_orders.values().cloned().collect()) - .unwrap_or_default() - } else { - orders_map - .values() - .flat_map(|symbol_orders| symbol_orders.values().cloned()) - .collect() - }; - - if let Some(since_ts) = since { - orders.retain(|order| order.timestamp.is_some_and(|ts| ts >= since_ts)); - } - - orders.sort_by(|a, b| { - let ts_a = a.timestamp.unwrap_or(0); - let ts_b = b.timestamp.unwrap_or(0); - ts_b.cmp(&ts_a) - }); - - if let Some(lim) = limit { - orders.truncate(lim); - } - - Ok(orders) + ws.watch_orders_internal(symbol, since, limit).await } /// Watches authenticated user trade updates @@ -615,43 +391,8 @@ impl Binance { limit: Option, _params: Option>, ) -> Result> { - const MAX_RETRIES: u32 = 100; - let ws = self.create_authenticated_ws(); - ws.connect().await?; - - let mut retries = 0; - - while retries < MAX_RETRIES { - if let Some(msg) = ws.client.receive().await { - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - if let Some(event_type) = msg.get("e").and_then(|e| e.as_str()) { - if event_type == "executionReport" { - if let Ok(trade) = BinanceWs::parse_ws_trade(&msg) { - let symbol_key = trade.symbol.clone(); - - let mut trades_map = ws.my_trades.write().await; - let symbol_trades = - trades_map.entry(symbol_key).or_insert_with(VecDeque::new); - - symbol_trades.push_front(trade); - if symbol_trades.len() > 1000 { - symbol_trades.pop_back(); - } - } - } - } - } else { - tokio::time::sleep(Duration::from_millis(100)).await; - } - - retries += 1; - } - - ws.filter_my_trades(symbol, since, limit).await + ws.watch_my_trades_internal(symbol, since, limit).await } /// Watches authenticated futures position updates @@ -662,62 +403,7 @@ impl Binance { limit: Option, _params: Option>, ) -> Result> { - const MAX_RETRIES: u32 = 100; - let ws = self.create_authenticated_ws(); - ws.connect().await?; - - let mut retries = 0; - - while retries < MAX_RETRIES { - if let Some(msg) = ws.client.receive().await { - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - if let Some(event_type) = msg.get("e").and_then(|e| e.as_str()) { - if event_type == "ACCOUNT_UPDATE" { - if let Some(account_data) = msg.get("a") { - if let Some(positions_array) = - account_data.get("P").and_then(|p| p.as_array()) - { - for position_data in positions_array { - if let Ok(position) = - BinanceWs::parse_ws_position(position_data) - { - let symbol_key = position.symbol.clone(); - let side_key = position - .side - .clone() - .unwrap_or_else(|| "both".to_string()); - - let mut positions_map = ws.positions.write().await; - let symbol_positions = positions_map - .entry(symbol_key) - .or_insert_with(HashMap::new); - - if position.contracts.unwrap_or(0.0).abs() < 0.000001 { - symbol_positions.remove(&side_key); - if symbol_positions.is_empty() { - positions_map.remove(&position.symbol); - } - } else { - symbol_positions.insert(side_key, position); - } - } - } - } - } - } - } - } else { - tokio::time::sleep(Duration::from_millis(100)).await; - } - - retries += 1; - } - - let symbols_ref = symbols.as_deref(); - ws.filter_positions(symbols_ref, since, limit).await + ws.watch_positions_internal(symbols, since, limit).await } } diff --git a/ccxt-exchanges/src/binance/ws/handlers.rs b/ccxt-exchanges/src/binance/ws/handlers.rs index 45b6aac..76a5252 100644 --- a/ccxt-exchanges/src/binance/ws/handlers.rs +++ b/ccxt-exchanges/src/binance/ws/handlers.rs @@ -58,13 +58,14 @@ impl MessageRouter { } /// Starts the message router - pub async fn start(&self) -> Result<()> { + pub async fn start(&self, url_override: Option) -> Result<()> { if self.is_connected() { self.stop().await?; } + let url = url_override.unwrap_or_else(|| self.ws_url.clone()); let config = ccxt_core::ws_client::WsConfig { - url: self.ws_url.clone(), + url: url.clone(), ..Default::default() }; let client = ccxt_core::ws_client::WsClient::new(config); @@ -79,7 +80,8 @@ impl MessageRouter { let subscription_manager = self.subscription_manager.clone(); let is_connected = self.is_connected.clone(); let reconnect_config = self.reconnect_config.clone(); - let ws_url = self.ws_url.clone(); + + let ws_url = url; let handle = tokio::spawn(async move { Self::message_loop( @@ -119,7 +121,12 @@ impl MessageRouter { pub async fn restart(&self) -> Result<()> { self.stop().await?; tokio::time::sleep(Duration::from_millis(100)).await; - self.start().await + self.start(None).await + } + + /// Returns the configured WebSocket URL + pub fn get_url(&self) -> String { + self.ws_url.clone() } /// Returns the current connection state @@ -296,6 +303,19 @@ impl MessageRouter { } if let Some(event_type) = message.get("e").and_then(|e| e.as_str()) { + match event_type { + "outboundAccountPosition" + | "balanceUpdate" + | "executionReport" + | "listStatus" + | "ACCOUNT_UPDATE" + | "ORDER_TRADE_UPDATE" + | "listenKeyExpired" => { + return Ok("!userData".to_string()); + } + _ => {} + } + if let Some(symbol) = message.get("s").and_then(|s| s.as_str()) { let stream = match event_type { "24hrTicker" => format!("{}@ticker", symbol.to_lowercase()), diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index eb9715e..ad3db65 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/ccxt-exchanges/src/binance/ws/mod.rs @@ -19,7 +19,6 @@ use ccxt_core::error::{Error, Result}; use ccxt_core::types::{ Balance, BidAsk, MarkPrice, MarketType, OHLCV, Order, OrderBook, Position, Ticker, Trade, }; -use ccxt_core::ws_client::{WsClient, WsConfig}; use serde_json::Value; use std::collections::{HashMap, VecDeque}; use std::sync::Arc; @@ -28,10 +27,10 @@ use tokio::sync::{Mutex, RwLock}; /// Binance WebSocket client wrapper pub struct BinanceWs { - pub(crate) client: Arc, + pub(crate) message_router: Arc, + pub(crate) subscription_manager: Arc, listen_key: Arc>>, listen_key_manager: Option>, - auto_reconnect_coordinator: Arc>>, pub(crate) tickers: Arc>>, pub(crate) bids_asks: Arc>>, #[allow(dead_code)] @@ -48,8 +47,7 @@ pub struct BinanceWs { impl std::fmt::Debug for BinanceWs { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("BinanceWs") - .field("is_connected", &self.client.is_connected()) - .field("state", &self.client.state()) + .field("is_connected", &self.message_router.is_connected()) .finish_non_exhaustive() } } @@ -57,23 +55,22 @@ impl std::fmt::Debug for BinanceWs { impl BinanceWs { /// Creates a new Binance WebSocket client pub fn new(url: String) -> Self { - let config = WsConfig { - url, - connect_timeout: 10000, - ping_interval: 180000, - reconnect_interval: 5000, - max_reconnect_attempts: 5, - auto_reconnect: true, - enable_compression: false, - pong_timeout: 90000, - ..Default::default() - }; + let subscription_manager = Arc::new(SubscriptionManager::new()); + let message_router = Arc::new(MessageRouter::new(url, subscription_manager.clone())); + + // Start the router immediately + let router_clone = message_router.clone(); + tokio::spawn(async move { + if let Err(e) = router_clone.start(None).await { + tracing::error!("Failed to start MessageRouter: {}", e); + } + }); Self { - client: Arc::new(WsClient::new(config)), + message_router, + subscription_manager, listen_key: Arc::new(RwLock::new(None)), listen_key_manager: None, - auto_reconnect_coordinator: Arc::new(Mutex::new(None)), tickers: Arc::new(Mutex::new(HashMap::new())), bids_asks: Arc::new(Mutex::new(HashMap::new())), mark_prices: Arc::new(Mutex::new(HashMap::new())), @@ -89,23 +86,22 @@ impl BinanceWs { /// Creates a WebSocket client with a listen key manager pub fn new_with_auth(url: String, binance: Arc) -> Self { - let config = WsConfig { - url, - connect_timeout: 10000, - ping_interval: 180000, - reconnect_interval: 5000, - max_reconnect_attempts: 5, - auto_reconnect: true, - enable_compression: false, - pong_timeout: 90000, - ..Default::default() - }; + let subscription_manager = Arc::new(SubscriptionManager::new()); + let message_router = Arc::new(MessageRouter::new(url, subscription_manager.clone())); + + // Start the router immediately + let router_clone = message_router.clone(); + tokio::spawn(async move { + if let Err(e) = router_clone.start(None).await { + tracing::error!("Failed to start MessageRouter: {}", e); + } + }); Self { - client: Arc::new(WsClient::new(config)), + message_router, + subscription_manager, listen_key: Arc::new(RwLock::new(None)), listen_key_manager: Some(Arc::new(ListenKeyManager::new(binance))), - auto_reconnect_coordinator: Arc::new(Mutex::new(None)), tickers: Arc::new(Mutex::new(HashMap::new())), bids_asks: Arc::new(Mutex::new(HashMap::new())), mark_prices: Arc::new(Mutex::new(HashMap::new())), @@ -121,32 +117,27 @@ impl BinanceWs { /// Connects to the WebSocket server pub async fn connect(&self) -> Result<()> { - self.client.connect().await?; - - let mut coordinator_guard = self.auto_reconnect_coordinator.lock().await; - if coordinator_guard.is_none() { - let coordinator = self.client.clone().create_auto_reconnect_coordinator(); - coordinator.start().await; - *coordinator_guard = Some(coordinator); - tracing::info!("Auto-reconnect coordinator started"); + if self.is_connected() { + return Ok(()); } + self.message_router.start(None).await?; + + // No auto-reconnect coordinator needed as MessageRouter handles it internally. + // The router's loop manages connection state and retries. + Ok(()) } /// Disconnects from the WebSocket server pub async fn disconnect(&self) -> Result<()> { - let mut coordinator_guard = self.auto_reconnect_coordinator.lock().await; - if let Some(coordinator) = coordinator_guard.take() { - coordinator.stop().await; - tracing::info!("Auto-reconnect coordinator stopped"); - } + self.message_router.stop().await?; if let Some(manager) = &self.listen_key_manager { manager.stop_auto_refresh().await; } - self.client.disconnect().await + Ok(()) } /// Connects to the user data stream @@ -157,9 +148,17 @@ impl BinanceWs { ))?; let listen_key = manager.get_or_create().await?; - let _user_stream_url = format!("wss://stream.binance.com:9443/ws/{}", listen_key); - self.client.connect().await?; + let base_url = self.message_router.get_url(); + let base_url = if base_url.ends_with('/') { + &base_url[..base_url.len() - 1] + } else { + &base_url + }; + + let url = format!("{}/{}", base_url, listen_key); + + self.message_router.start(Some(url)).await?; manager.start_auto_refresh().await; *self.listen_key.write().await = Some(listen_key); @@ -187,32 +186,69 @@ impl BinanceWs { /// Subscribes to the ticker stream for a symbol pub async fn subscribe_ticker(&self, symbol: &str) -> Result<()> { let stream = format!("{}@ticker", symbol.to_lowercase()); - self.client - .subscribe(stream, Some(symbol.to_string()), None) - .await + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::Ticker, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream]).await } /// Subscribes to the 24-hour ticker stream for all symbols pub async fn subscribe_all_tickers(&self) -> Result<()> { - self.client - .subscribe("!ticker@arr".to_string(), None, None) - .await + let stream = "!ticker@arr".to_string(); + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + "all".to_string(), + SubscriptionType::Ticker, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream]).await } /// Subscribes to real-time trade executions for a symbol pub async fn subscribe_trades(&self, symbol: &str) -> Result<()> { let stream = format!("{}@trade", symbol.to_lowercase()); - self.client - .subscribe(stream, Some(symbol.to_string()), None) - .await + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::Trades, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream]).await } /// Subscribes to the aggregated trade stream for a symbol pub async fn subscribe_agg_trades(&self, symbol: &str) -> Result<()> { let stream = format!("{}@aggTrade", symbol.to_lowercase()); - self.client - .subscribe(stream, Some(symbol.to_string()), None) - .await + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::Trades, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream]).await } /// Subscribes to the order book depth stream @@ -227,10 +263,18 @@ impl BinanceWs { } else { format!("{}@depth{}", symbol.to_lowercase(), levels) }; + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); - self.client - .subscribe(stream, Some(symbol.to_string()), None) - .await + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::OrderBook, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream]).await } /// Subscribes to the diff order book stream @@ -248,53 +292,96 @@ impl BinanceWs { } else { format!("{}@depth", symbol.to_lowercase()) }; + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); - self.client - .subscribe(stream, Some(symbol.to_string()), None) - .await + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::OrderBook, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream]).await } /// Subscribes to Kline (candlestick) data for a symbol pub async fn subscribe_kline(&self, symbol: &str, interval: &str) -> Result<()> { let stream = format!("{}@kline_{}", symbol.to_lowercase(), interval); - self.client - .subscribe(stream, Some(symbol.to_string()), None) - .await + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::Kline(interval.to_string()), + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream]).await } /// Subscribes to the mini ticker stream for a symbol pub async fn subscribe_mini_ticker(&self, symbol: &str) -> Result<()> { let stream = format!("{}@miniTicker", symbol.to_lowercase()); - self.client - .subscribe(stream, Some(symbol.to_string()), None) - .await + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::Ticker, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream]).await } /// Subscribes to the mini ticker stream for all symbols pub async fn subscribe_all_mini_tickers(&self) -> Result<()> { - self.client - .subscribe("!miniTicker@arr".to_string(), None, None) - .await + let stream = "!miniTicker@arr".to_string(); + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + "all".to_string(), + SubscriptionType::Ticker, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream]).await } /// Cancels an existing subscription pub async fn unsubscribe(&self, stream: String) -> Result<()> { - self.client.unsubscribe(stream, None).await + self.subscription_manager + .remove_subscription(&stream) + .await?; + self.message_router.unsubscribe(vec![stream]).await } /// Receives the next available message pub async fn receive(&self) -> Option { - self.client.receive().await + None } /// Indicates whether the WebSocket connection is active pub fn is_connected(&self) -> bool { - self.client.is_connected() + self.message_router.is_connected() } /// Returns the current connection state. pub fn state(&self) -> ccxt_core::ws_client::WsConnectionState { - self.client.state() + if self.message_router.is_connected() { + ccxt_core::ws_client::WsConnectionState::Connected + } else { + ccxt_core::ws_client::WsConnectionState::Disconnected + } } /// Returns the list of active subscriptions. @@ -303,24 +390,33 @@ impl BinanceWs { /// This method retrieves the actual subscriptions from the underlying WsClient's /// subscription manager, providing accurate state tracking. pub fn subscriptions(&self) -> Vec { - self.client.subscriptions() + let subs = self.subscription_manager.get_all_subscriptions_sync(); + subs.into_iter().map(|s| s.stream).collect() } /// Returns the number of active subscriptions. pub fn subscription_count(&self) -> usize { - self.client.subscription_count() + self.subscription_manager.active_count() } /// Watches a single ticker stream (internal helper) async fn watch_ticker_internal(&self, symbol: &str, channel_name: &str) -> Result { let stream = format!("{}@{}", symbol.to_lowercase(), channel_name); - self.client - .subscribe(stream.clone(), Some(symbol.to_string()), None) + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::Ticker, + tx, + ) .await?; + self.message_router.subscribe(vec![stream.clone()]).await?; + loop { - if let Some(message) = self.client.receive().await { + if let Some(message) = rx.recv().await { if message.get("result").is_some() { continue; } @@ -330,6 +426,8 @@ impl BinanceWs { tickers.insert(ticker.symbol.clone(), ticker.clone()); return Ok(ticker); } + } else { + return Err(Error::network("Subscription channel closed")); } } } @@ -348,14 +446,25 @@ impl BinanceWs { vec![format!("!{}@arr", channel_name)] }; + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + for stream in &streams { - self.client.subscribe(stream.clone(), None, None).await?; + self.subscription_manager + .add_subscription( + stream.clone(), + "all".to_string(), + SubscriptionType::Ticker, + tx.clone(), + ) + .await?; } + self.message_router.subscribe(streams.clone()).await?; + let mut result = HashMap::new(); loop { - if let Some(message) = self.client.receive().await { + if let Some(message) = rx.recv().await { if message.get("result").is_some() { continue; } @@ -379,7 +488,7 @@ impl BinanceWs { } if let Some(syms) = &symbols { - if result.len() == syms.len() { + if result.len() >= syms.len() { return Ok(result); } } else { @@ -393,11 +502,13 @@ impl BinanceWs { tickers.insert(symbol, ticker); if let Some(syms) = &symbols { - if result.len() == syms.len() { + if result.len() >= syms.len() { return Ok(result); } } } + } else { + return Err(Error::network("Subscription channel closed")); } } } @@ -439,10 +550,18 @@ impl BinanceWs { format!("{}@depth", symbol.to_lowercase()) }; - self.client - .subscribe(stream.clone(), Some(symbol.to_string()), None) + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::OrderBook, + tx, + ) .await?; + self.message_router.subscribe(vec![stream.clone()]).await?; + tokio::time::sleep(Duration::from_millis(500)).await; let _snapshot = self @@ -450,7 +569,7 @@ impl BinanceWs { .await?; loop { - if let Some(message) = self.client.receive().await { + if let Some(message) = rx.recv().await { if message.get("result").is_some() { continue; } @@ -528,6 +647,8 @@ impl BinanceWs { } } } + } else { + return Err(Error::network("Subscription channel closed")); } } } @@ -547,6 +668,9 @@ impl BinanceWs { )); } + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let mut streams = Vec::new(); + for symbol in &symbols { let stream = if update_speed == 100 { format!("{}@depth@100ms", symbol.to_lowercase()) @@ -554,11 +678,20 @@ impl BinanceWs { format!("{}@depth", symbol.to_lowercase()) }; - self.client - .subscribe(stream, Some(symbol.clone()), None) + streams.push(stream.clone()); + + self.subscription_manager + .add_subscription( + stream, + symbol.clone(), + SubscriptionType::OrderBook, + tx.clone(), + ) .await?; } + self.message_router.subscribe(streams).await?; + tokio::time::sleep(Duration::from_millis(500)).await; for symbol in &symbols { @@ -571,7 +704,7 @@ impl BinanceWs { let mut update_count = 0; while update_count < symbols.len() { - if let Some(message) = self.client.receive().await { + if let Some(message) = rx.recv().await { if message.get("result").is_some() { continue; } @@ -593,6 +726,8 @@ impl BinanceWs { } } } + } else { + return Err(Error::network("Subscription channel closed")); } } @@ -606,6 +741,164 @@ impl BinanceWs { Ok(result) } + /// Watches the best bid/ask data for a unified symbol (internal helper) + async fn watch_bids_asks_internal(&self, symbol: &str, market_id: &str) -> Result { + let stream = format!("{}@bookTicker", market_id.to_lowercase()); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::BookTicker, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream.clone()]).await?; + + loop { + if let Some(message) = rx.recv().await { + if message.get("result").is_some() { + continue; + } + + if let Ok(bid_ask) = parser::parse_ws_bid_ask(&message) { + let mut bids_asks_map = self.bids_asks.lock().await; + bids_asks_map.insert(symbol.to_string(), bid_ask.clone()); + return Ok(bid_ask); + } + } else { + return Err(Error::network("Subscription channel closed")); + } + } + } + + /// Streams trade data for a unified symbol (internal helper) + async fn watch_trades_internal( + &self, + symbol: &str, + market_id: &str, + since: Option, + limit: Option, + market: Option<&ccxt_core::types::Market>, + ) -> Result> { + let stream = format!("{}@trade", market_id.to_lowercase()); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::Trades, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream.clone()]).await?; + + // Wait for at least one trade or use a loop with timeout if we want to mimic "polling" until data arrives? + // Usually watch_trades returns the latest trades. + + loop { + if let Some(message) = rx.recv().await { + if message.get("result").is_some() { + continue; + } + + if let Ok(trade) = parser::parse_ws_trade(&message, market) { + let mut trades_map = self.trades.lock().await; + let trades = trades_map + .entry(symbol.to_string()) + .or_insert_with(VecDeque::new); + + // Constants from binance_impl.rs + const MAX_TRADES: usize = 1000; + if trades.len() >= MAX_TRADES { + trades.pop_front(); + } + trades.push_back(trade); + + let mut result: Vec = trades.iter().cloned().collect(); + + if let Some(since_ts) = since { + result.retain(|t| t.timestamp >= since_ts); + } + + if let Some(limit_size) = limit { + if result.len() > limit_size { + result = result.split_off(result.len() - limit_size); + } + } + + return Ok(result); + } + } else { + return Err(Error::network("Subscription channel closed")); + } + } + } + + /// Streams OHLCV data for a unified symbol (internal helper) + async fn watch_ohlcv_internal( + &self, + symbol: &str, + market_id: &str, + timeframe: &str, + since: Option, + limit: Option, + ) -> Result> { + let stream = format!("{}@kline_{}", market_id.to_lowercase(), timeframe); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::Kline(timeframe.to_string()), + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream.clone()]).await?; + + loop { + if let Some(message) = rx.recv().await { + if message.get("result").is_some() { + continue; + } + + if let Ok(ohlcv) = parser::parse_ws_ohlcv(&message) { + let cache_key = format!("{}:{}", symbol, timeframe); + let mut ohlcvs_map = self.ohlcvs.lock().await; + let ohlcvs = ohlcvs_map.entry(cache_key).or_insert_with(VecDeque::new); + + const MAX_OHLCVS: usize = 1000; + if ohlcvs.len() >= MAX_OHLCVS { + ohlcvs.pop_front(); + } + ohlcvs.push_back(ohlcv); + + let mut result: Vec = ohlcvs.iter().cloned().collect(); + + if let Some(since_ts) = since { + result.retain(|o| o.timestamp >= since_ts); + } + + if let Some(limit_size) = limit { + if result.len() > limit_size { + result = result.split_off(result.len() - limit_size); + } + } + + return Ok(result); + } + } else { + return Err(Error::network("Subscription channel closed")); + } + } + } + /// Returns cached ticker snapshot pub async fn get_cached_ticker(&self, symbol: &str) -> Option { let tickers = self.tickers.lock().await; @@ -623,6 +916,257 @@ impl BinanceWs { user_data::handle_balance_message(message, account_type, &self.balances).await } + /// Watches for balance updates (internal helper) + async fn watch_balance_internal(&self, account_type: &str) -> Result { + self.connect_user_stream().await?; + + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + "!userData".to_string(), + "user".to_string(), + SubscriptionType::Balance, + tx, + ) + .await?; + + loop { + if let Some(message) = rx.recv().await { + if let Some(event_type) = message.get("e").and_then(|e| e.as_str()) { + if matches!( + event_type, + "balanceUpdate" | "outboundAccountPosition" | "ACCOUNT_UPDATE" + ) { + if let Ok(()) = self.handle_balance_message(&message, account_type).await { + let balances = self.balances.read().await; + if let Some(balance) = balances.get(account_type) { + return Ok(balance.clone()); + } + } + } + } + } else { + return Err(Error::network("Subscription channel closed")); + } + } + } + + /// Watches for order updates (internal helper) + async fn watch_orders_internal( + &self, + symbol: Option<&str>, + since: Option, + limit: Option, + ) -> Result> { + self.connect_user_stream().await?; + + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + "!userData".to_string(), + "user".to_string(), + SubscriptionType::Orders, + tx, + ) + .await?; + + loop { + if let Some(message) = rx.recv().await { + if let Value::Object(data) = message { + if let Some(event_type) = data.get("e").and_then(serde_json::Value::as_str) { + if event_type == "executionReport" { + let order = user_data::parse_ws_order(&data); + + let mut orders = self.orders.write().await; + let symbol_orders = orders + .entry(order.symbol.clone()) + .or_insert_with(HashMap::new); + symbol_orders.insert(order.id.clone(), order.clone()); + drop(orders); + + if let Some(exec_type) = + data.get("x").and_then(serde_json::Value::as_str) + { + if exec_type == "TRADE" { + if let Ok(trade) = + BinanceWs::parse_ws_trade(&Value::Object(data.clone())) + { + let mut trades = self.my_trades.write().await; + let symbol_trades = trades + .entry(trade.symbol.clone()) + .or_insert_with(VecDeque::new); + + symbol_trades.push_front(trade); + if symbol_trades.len() > 1000 { + symbol_trades.pop_back(); + } + } + } + } + + return self.filter_orders(symbol, since, limit).await; + } + } + } + } else { + return Err(Error::network("Subscription channel closed")); + } + } + } + + /// Watches for my trades (internal helper) + async fn watch_my_trades_internal( + &self, + symbol: Option<&str>, + since: Option, + limit: Option, + ) -> Result> { + self.connect_user_stream().await?; + + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + "!userData".to_string(), + "user".to_string(), + SubscriptionType::MyTrades, + tx, + ) + .await?; + + loop { + if let Some(msg) = rx.recv().await { + if let Some(event_type) = msg.get("e").and_then(|e| e.as_str()) { + if event_type == "executionReport" { + if let Ok(trade) = BinanceWs::parse_ws_trade(&msg) { + let symbol_key = trade.symbol.clone(); + + let mut trades_map = self.my_trades.write().await; + let symbol_trades = + trades_map.entry(symbol_key).or_insert_with(VecDeque::new); + + symbol_trades.push_front(trade); + if symbol_trades.len() > 1000 { + symbol_trades.pop_back(); + } + + drop(trades_map); + return self.filter_my_trades(symbol, since, limit).await; + } + } + } + } else { + return Err(Error::network("Subscription channel closed")); + } + } + } + + /// Watches for positions (internal helper) + async fn watch_positions_internal( + &self, + symbols: Option>, + since: Option, + limit: Option, + ) -> Result> { + self.connect_user_stream().await?; + + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + self.subscription_manager + .add_subscription( + "!userData".to_string(), + "user".to_string(), + SubscriptionType::Positions, + tx, + ) + .await?; + + loop { + if let Some(msg) = rx.recv().await { + if let Some(event_type) = msg.get("e").and_then(|e| e.as_str()) { + if event_type == "ACCOUNT_UPDATE" { + if let Some(account_data) = msg.get("a") { + if let Some(positions_array) = + account_data.get("P").and_then(|p| p.as_array()) + { + for position_data in positions_array { + if let Ok(position) = + BinanceWs::parse_ws_position(position_data) + { + let symbol_key = position.symbol.clone(); + let side_key = position + .side + .clone() + .unwrap_or_else(|| "both".to_string()); + + let mut positions_map = self.positions.write().await; + let symbol_positions = positions_map + .entry(symbol_key) + .or_insert_with(HashMap::new); + + if position.contracts.unwrap_or(0.0).abs() < 0.000001 { + symbol_positions.remove(&side_key); + if symbol_positions.is_empty() { + positions_map.remove(&position.symbol); + } + } else { + symbol_positions.insert(side_key, position); + } + } + } + + let symbols_ref = symbols.as_deref(); + return self.filter_positions(symbols_ref, since, limit).await; + } + } + } + } + } else { + return Err(Error::network("Subscription channel closed")); + } + } + } + + /// Filters cached orders by symbol, time range, and limit + async fn filter_orders( + &self, + symbol: Option<&str>, + since: Option, + limit: Option, + ) -> Result> { + let orders_map = self.orders.read().await; + + let mut orders: Vec = if let Some(sym) = symbol { + orders_map + .get(sym) + .map(|symbol_orders| symbol_orders.values().cloned().collect()) + .unwrap_or_default() + } else { + orders_map + .values() + .flat_map(|symbol_orders| symbol_orders.values().cloned()) + .collect() + }; + + if let Some(since_ts) = since { + orders.retain(|order| order.timestamp.is_some_and(|ts| ts >= since_ts)); + } + + orders.sort_by(|a, b| { + let ts_a = a.timestamp.unwrap_or(0); + let ts_b = b.timestamp.unwrap_or(0); + ts_b.cmp(&ts_a) + }); + + if let Some(lim) = limit { + orders.truncate(lim); + } + + Ok(orders) + } + /// Parses a WebSocket trade message fn parse_ws_trade(data: &Value) -> Result { user_data::parse_ws_trade(data) diff --git a/ccxt-exchanges/src/binance/ws/subscriptions.rs b/ccxt-exchanges/src/binance/ws/subscriptions.rs index 0ee6ecf..9da0930 100644 --- a/ccxt-exchanges/src/binance/ws/subscriptions.rs +++ b/ccxt-exchanges/src/binance/ws/subscriptions.rs @@ -175,6 +175,15 @@ impl SubscriptionManager { subs.values().cloned().collect() } + /// Returns all registered subscriptions synchronously (non-blocking) + pub fn get_all_subscriptions_sync(&self) -> Vec { + if let Ok(subs) = self.subscriptions.try_read() { + subs.values().cloned().collect() + } else { + Vec::new() + } + } + /// Returns all subscriptions associated with a symbol pub async fn get_subscriptions_by_symbol(&self, symbol: &str) -> Vec { let index = self.symbol_index.read().await; diff --git a/ccxt-exchanges/tests/binance_ws_subscribe_payload_test.rs b/ccxt-exchanges/tests/binance_ws_subscribe_payload_test.rs new file mode 100644 index 0000000..9d7f244 --- /dev/null +++ b/ccxt-exchanges/tests/binance_ws_subscribe_payload_test.rs @@ -0,0 +1,57 @@ +use ccxt_exchanges::binance::ws::BinanceWs; +use futures_util::{SinkExt, StreamExt}; +use serde_json::Value; +use std::time::Duration; +use tokio_tungstenite::tungstenite::protocol::Message; + +#[tokio::test] +async fn test_binance_ws_subscribe_payload() { + // Create a TCP listener for the mock WebSocket server + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let ws_url = format!("ws://127.0.0.1:{}/ws", port); + + let server_task = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut ws_stream = tokio_tungstenite::accept_async(stream).await.unwrap(); + + // 2. Expect a subscription message + if let Some(msg) = ws_stream.next().await { + let msg = msg.unwrap(); + if let Message::Text(text) = msg { + let json: Value = serde_json::from_str(&text).unwrap(); + + // 3. Verify the payload format + // Expected: {"method": "SUBSCRIBE", "params": ["btcusdt@ticker"], "id": ...} + assert_eq!(json["method"], "SUBSCRIBE"); + assert!(json["params"].is_array()); + let params = json["params"].as_array().unwrap(); + assert_eq!(params[0], "btcusdt@ticker"); + assert!(json["id"].is_number()); + + // Send response to keep connection alive if needed + let response = serde_json::json!({ + "result": null, + "id": json["id"] + }); + ws_stream + .send(Message::Text(response.to_string().into())) + .await + .unwrap(); + } + } + }); + + // 4. Client connection and subscription + let binance_ws = BinanceWs::new(ws_url); + binance_ws.connect().await.unwrap(); + + // Give it a moment to connect + tokio::time::sleep(Duration::from_millis(100)).await; + + // Subscribe + binance_ws.subscribe_ticker("BTC/USDT").await.unwrap(); + + // Wait for server verification + server_task.await.unwrap(); +} From 44e5e9d3fc58273709ee5aba40feb557f111ff28 Mon Sep 17 00:00:00 2001 From: Praying Date: Thu, 29 Jan 2026 17:09:46 +0800 Subject: [PATCH 2/3] chore(websocket): refactor binance websocket --- ccxt-core/src/base_exchange/mod.rs | 2 +- ccxt-core/src/http_client/builder.rs | 2 +- ccxt-core/src/retry_strategy.rs | 2 +- ccxt-exchanges/src/binance/mod.rs | 72 ++- .../src/binance/ws/connection_manager.rs | 134 +++++ ccxt-exchanges/src/binance/ws/handlers.rs | 75 ++- ccxt-exchanges/src/binance/ws/mod.rs | 17 +- .../src/binance/ws_exchange_impl.rs | 566 +++++++++++------- 8 files changed, 651 insertions(+), 219 deletions(-) create mode 100644 ccxt-exchanges/src/binance/ws/connection_manager.rs diff --git a/ccxt-core/src/base_exchange/mod.rs b/ccxt-core/src/base_exchange/mod.rs index 05299f8..318ae73 100644 --- a/ccxt-core/src/base_exchange/mod.rs +++ b/ccxt-core/src/base_exchange/mod.rs @@ -32,7 +32,7 @@ use tokio::sync::{Mutex, RwLock}; use tracing::{debug, info, warn}; /// Base exchange implementation -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct BaseExchange { /// Exchange configuration pub config: ExchangeConfig, diff --git a/ccxt-core/src/http_client/builder.rs b/ccxt-core/src/http_client/builder.rs index 26a1f80..14bff19 100644 --- a/ccxt-core/src/http_client/builder.rs +++ b/ccxt-core/src/http_client/builder.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use super::config::HttpConfig; /// HTTP client with retry and rate limiting support -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct HttpClient { client: Client, config: HttpConfig, diff --git a/ccxt-core/src/retry_strategy.rs b/ccxt-core/src/retry_strategy.rs index 65ef3d3..ca9a816 100644 --- a/ccxt-core/src/retry_strategy.rs +++ b/ccxt-core/src/retry_strategy.rs @@ -174,7 +174,7 @@ impl RetryConfig { } /// Retry strategy. -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct RetryStrategy { config: RetryConfig, } diff --git a/ccxt-exchanges/src/binance/mod.rs b/ccxt-exchanges/src/binance/mod.rs index 32f08dd..257e601 100644 --- a/ccxt-exchanges/src/binance/mod.rs +++ b/ccxt-exchanges/src/binance/mod.rs @@ -33,7 +33,7 @@ pub use signed_request::{HttpMethod, SignedRequestBuilder}; pub use time_sync::{TimeSyncConfig, TimeSyncManager}; pub use urls::BinanceUrls; /// Binance exchange structure. -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct Binance { /// Base exchange instance. base: BaseExchange, @@ -42,9 +42,13 @@ pub struct Binance { /// Time synchronization manager for caching server time offset. time_sync: Arc, /// Persistent WebSocket connection reference for state tracking. + #[deprecated(note = "Use connection_manager instead")] ws_connection: Arc>>, + /// Centralized WebSocket connection manager + pub connection_manager: Arc, } +#[allow(deprecated)] impl Binance { /// Creates a new Binance instance using the builder pattern. /// @@ -93,11 +97,27 @@ impl Binance { let options = BinanceOptions::default(); let time_sync = Arc::new(TimeSyncManager::new()); + // Determine WS URL for ConnectionManager + let urls = if base.config.sandbox { + BinanceUrls::testnet() + } else { + BinanceUrls::production() + }; + // Apply overrides logic similar to self.urls() + let ws_url = if let Some(url) = base.config.url_overrides.get("ws") { + url.clone() + } else { + urls.ws.clone() + }; + + let connection_manager = Arc::new(ws::BinanceConnectionManager::new(ws_url)); + Ok(Self { base, options, time_sync, ws_connection: Arc::new(RwLock::new(None)), + connection_manager, }) } @@ -136,11 +156,44 @@ impl Binance { }; let time_sync = Arc::new(TimeSyncManager::with_config(time_sync_config)); + // Determine WS URL + let mut urls = if base.config.sandbox { + BinanceUrls::testnet() + } else { + BinanceUrls::production() + }; + + // Apply overrides + if let Some(ws_url) = base.config.url_overrides.get("ws") { + urls.ws.clone_from(ws_url); + } + if let Some(ws_fapi_url) = base.config.url_overrides.get("wsFapi") { + urls.ws_fapi.clone_from(ws_fapi_url); + } + if let Some(ws_dapi_url) = base.config.url_overrides.get("wsDapi") { + urls.ws_dapi.clone_from(ws_dapi_url); + } + if let Some(ws_eapi_url) = base.config.url_overrides.get("wsEapi") { + urls.ws_eapi.clone_from(ws_eapi_url); + } + + let ws_url = match options.default_type { + DefaultType::Swap | DefaultType::Futures => match options.default_sub_type { + Some(DefaultSubType::Inverse) => urls.ws_dapi.clone(), + _ => urls.ws_fapi.clone(), + }, + DefaultType::Option => urls.ws_eapi.clone(), + _ => urls.ws.clone(), + }; + + let connection_manager = Arc::new(ws::BinanceConnectionManager::new(ws_url)); + Ok(Self { base, options, time_sync, ws_connection: Arc::new(RwLock::new(None)), + connection_manager, }) } @@ -174,11 +227,28 @@ impl Binance { }; let time_sync = Arc::new(TimeSyncManager::with_config(time_sync_config)); + // Determine WS URL (Futures/Swap) + let mut urls = if base.config.sandbox { + BinanceUrls::testnet() + } else { + BinanceUrls::production() + }; + + if let Some(ws_fapi_url) = base.config.url_overrides.get("wsFapi") { + urls.ws_fapi.clone_from(ws_fapi_url); + } + + // Default to FAPI/Linear for new_swap if not configured otherwise + let ws_url = urls.ws_fapi.clone(); + + let connection_manager = Arc::new(ws::BinanceConnectionManager::new(ws_url)); + Ok(Self { base, options, time_sync, ws_connection: Arc::new(RwLock::new(None)), + connection_manager, }) } diff --git a/ccxt-exchanges/src/binance/ws/connection_manager.rs b/ccxt-exchanges/src/binance/ws/connection_manager.rs new file mode 100644 index 0000000..557f6e4 --- /dev/null +++ b/ccxt-exchanges/src/binance/ws/connection_manager.rs @@ -0,0 +1,134 @@ +use crate::binance::Binance; +use crate::binance::ws::BinanceWs; +use ccxt_core::error::Result; +use std::sync::Arc; +use tokio::sync::RwLock; + +const MAX_STREAMS_PER_CONNECTION: usize = 200; + +/// Manages multiple WebSocket connections for Binance to handle sharding and isolation +#[derive(Debug)] +pub struct BinanceConnectionManager { + /// Pool of public data connections (shards) + public_shards: RwLock>>, + /// Dedicated connection for private user data + private_client: RwLock>>, + /// Base WebSocket URL + base_ws_url: String, +} + +impl BinanceConnectionManager { + /// Creates a new connection manager + pub fn new(base_ws_url: String) -> Self { + Self { + public_shards: RwLock::new(Vec::new()), + private_client: RwLock::new(None), + base_ws_url, + } + } + + /// Gets or creates a public connection suitable for a new subscription + pub async fn get_public_connection(&self) -> Result> { + let mut shards = self.public_shards.write().await; + + // Try to find an existing shard with capacity + for shard in shards.iter() { + if shard.subscription_count() < MAX_STREAMS_PER_CONNECTION { + return Ok(shard.clone()); + } + } + + // If no shard available or all full, create a new one + let new_shard = Arc::new(BinanceWs::new(self.base_ws_url.clone())); + + // Connect the new shard immediately + new_shard.connect().await?; + + shards.push(new_shard.clone()); + Ok(new_shard) + } + + /// Gets or creates the dedicated private connection + pub async fn get_private_connection(&self, binance: &Arc) -> Result> { + let mut client_lock = self.private_client.write().await; + + if let Some(client) = client_lock.as_ref() { + if client.is_connected() { + return Ok(client.clone()); + } + } + + // Create new authenticated client + let new_client = Arc::new(BinanceWs::new_with_auth( + self.base_ws_url.clone(), + binance.clone(), + )); + + // Connect user stream (this handles listenKey generation and management) + new_client.connect_user_stream().await?; + + *client_lock = Some(new_client.clone()); + Ok(new_client) + } + + /// Returns the number of active public shards + pub async fn active_shards_count(&self) -> usize { + self.public_shards.read().await.len() + } + + /// Disconnects all active connections + pub async fn disconnect_all(&self) -> Result<()> { + let mut shards = self.public_shards.write().await; + for shard in shards.iter() { + let _ = shard.disconnect().await; + } + shards.clear(); + + let mut private = self.private_client.write().await; + if let Some(client) = private.take() { + let _ = client.disconnect().await; + } + + Ok(()) + } + + /// Checks if any connection is active (non-blocking) + pub fn is_connected(&self) -> bool { + if let Ok(shards) = self.public_shards.try_read() { + if !shards.is_empty() { + // Check if at least one shard is connected? + for shard in shards.iter() { + if shard.is_connected() { + return true; + } + } + } + } + + if let Ok(private) = self.private_client.try_read() { + if let Some(client) = private.as_ref() { + if client.is_connected() { + return true; + } + } + } + + false + } + + /// Returns a list of all active subscriptions across all connections + pub fn get_all_subscriptions(&self) -> Vec { + let mut subs = Vec::new(); + if let Ok(shards) = self.public_shards.try_read() { + for shard in shards.iter() { + subs.extend(shard.subscriptions()); + } + } + if let Ok(private) = self.private_client.try_read() { + if let Some(client) = private.as_ref() { + subs.extend(client.subscriptions()); + } + } + subs + } +} diff --git a/ccxt-exchanges/src/binance/ws/handlers.rs b/ccxt-exchanges/src/binance/ws/handlers.rs index 76a5252..7608f1e 100644 --- a/ccxt-exchanges/src/binance/ws/handlers.rs +++ b/ccxt-exchanges/src/binance/ws/handlers.rs @@ -285,8 +285,15 @@ impl MessageRouter { ) -> Result<()> { let stream_name = Self::extract_stream_name(&message)?; + // Check for "data" field (Combined Stream format) and unwrap if present + let payload = if message.get("stream").is_some() && message.get("data").is_some() { + message.get("data").cloned().unwrap_or(message) + } else { + message + }; + let sent = subscription_manager - .send_to_stream(&stream_name, message) + .send_to_stream(&stream_name, payload) .await; if sent { @@ -533,3 +540,69 @@ pub async fn watch_ticker_internal( } } } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use std::sync::Arc; + + #[test] + fn test_extract_stream_name_combined() { + let message = json!({ + "stream": "btcusdt@ticker", + "data": { + "e": "24hrTicker", + "s": "BTCUSDT" + } + }); + + let stream = MessageRouter::extract_stream_name(&message).unwrap(); + assert_eq!(stream, "btcusdt@ticker"); + } + + #[test] + fn test_extract_stream_name_raw() { + let message = json!({ + "e": "24hrTicker", + "s": "BTCUSDT" + }); + + let stream = MessageRouter::extract_stream_name(&message).unwrap(); + assert_eq!(stream, "btcusdt@ticker"); + } + + #[tokio::test] + async fn test_handle_message_unwrapping() { + let manager = Arc::new(crate::binance::ws::subscriptions::SubscriptionManager::new()); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + manager + .add_subscription( + "btcusdt@ticker".to_string(), + "BTCUSDT".to_string(), + crate::binance::ws::subscriptions::SubscriptionType::Ticker, + tx, + ) + .await + .unwrap(); + + let message = json!({ + "stream": "btcusdt@ticker", + "data": { + "e": "24hrTicker", + "s": "BTCUSDT", + "c": "50000.00" + } + }); + + MessageRouter::handle_message(message, manager) + .await + .unwrap(); + + let received = rx.recv().await.unwrap(); + assert!(received.get("stream").is_none()); + assert_eq!(received["e"], "24hrTicker"); + assert_eq!(received["c"], "50000.00"); + } +} diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index ad3db65..c09e7f0 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/ccxt-exchanges/src/binance/ws/mod.rs @@ -2,13 +2,16 @@ //! //! Provides WebSocket real-time data stream subscriptions for the Binance exchange +/// Connection manager module +pub mod connection_manager; mod handlers; mod listen_key; mod streams; mod subscriptions; -mod user_data; +pub(crate) mod user_data; // Re-export public types for backward compatibility +pub use connection_manager::BinanceConnectionManager; pub use handlers::MessageRouter; pub use listen_key::ListenKeyManager; pub use streams::*; @@ -25,6 +28,9 @@ use std::sync::Arc; use std::time::Duration; use tokio::sync::{Mutex, RwLock}; +const MAX_TRADES: usize = 1000; +const MAX_OHLCVS: usize = 1000; + /// Binance WebSocket client wrapper pub struct BinanceWs { pub(crate) message_router: Arc, @@ -366,7 +372,7 @@ impl BinanceWs { } /// Receives the next available message - pub async fn receive(&self) -> Option { + pub fn receive(&self) -> Option { None } @@ -812,8 +818,6 @@ impl BinanceWs { .entry(symbol.to_string()) .or_insert_with(VecDeque::new); - // Constants from binance_impl.rs - const MAX_TRADES: usize = 1000; if trades.len() >= MAX_TRADES { trades.pop_front(); } @@ -873,7 +877,6 @@ impl BinanceWs { let mut ohlcvs_map = self.ohlcvs.lock().await; let ohlcvs = ohlcvs_map.entry(cache_key).or_insert_with(VecDeque::new); - const MAX_OHLCVS: usize = 1000; if ohlcvs.len() >= MAX_OHLCVS { ohlcvs.pop_front(); } @@ -1261,8 +1264,8 @@ include!("binance_impl.rs"); mod tests { use super::*; - #[test] - fn test_binance_ws_creation() { + #[tokio::test] + async fn test_binance_ws_creation() { let ws = BinanceWs::new(WS_BASE_URL.to_string()); assert!(ws.listen_key.try_read().is_ok()); } diff --git a/ccxt-exchanges/src/binance/ws_exchange_impl.rs b/ccxt-exchanges/src/binance/ws_exchange_impl.rs index 994b387..2ad7f02 100644 --- a/ccxt-exchanges/src/binance/ws_exchange_impl.rs +++ b/ccxt-exchanges/src/binance/ws_exchange_impl.rs @@ -16,6 +16,7 @@ use ccxt_core::{ use rust_decimal::Decimal; use std::collections::HashMap; +use std::sync::Arc; use tokio::sync::mpsc; use super::Binance; @@ -47,44 +48,25 @@ impl WsExchange for Binance { // ==================== Connection Management ==================== async fn ws_connect(&self) -> Result<()> { - // Create a public WebSocket connection - let ws = self.create_ws(); - ws.connect().await?; - - // Store the connected WebSocket instance for state tracking - let mut conn = self.ws_connection.write().await; - *conn = Some(ws); - + // Ensure at least one public connection is active + let _ = self.connection_manager.get_public_connection().await?; Ok(()) } async fn ws_disconnect(&self) -> Result<()> { - // Get and clear the stored WebSocket connection - let mut conn = self.ws_connection.write().await; - if let Some(ws) = conn.take() { - ws.disconnect().await?; - } - Ok(()) + self.connection_manager.disconnect_all().await } fn ws_is_connected(&self) -> bool { - // Use try_read to avoid blocking - if let Ok(conn) = self.ws_connection.try_read() { - if let Some(ref ws) = *conn { - return ws.is_connected(); - } - } - false + self.connection_manager.is_connected() } fn ws_state(&self) -> WsConnectionState { - // Use try_read to avoid blocking - if let Ok(conn) = self.ws_connection.try_read() { - if let Some(ref ws) = *conn { - return ws.state(); - } + if self.connection_manager.is_connected() { + WsConnectionState::Connected + } else { + WsConnectionState::Disconnected } - WsConnectionState::Disconnected } // ==================== Public Data Streams ==================== @@ -96,44 +78,49 @@ impl WsExchange for Binance { // Get market info let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); + let stream = format!("{}@ticker", binance_symbol); + + // Get shared public connection + let ws = self.connection_manager.get_public_connection().await?; + + // Create subscription channel + let (tx, mut rx) = mpsc::unbounded_channel(); + + // Register with manager + ws.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + super::ws::SubscriptionType::Ticker, + tx, + ) + .await?; - // Create WebSocket and connect - let ws = self.create_ws(); - ws.connect().await?; - - // Subscribe to ticker stream - ws.subscribe_ticker(&binance_symbol).await?; + // Send subscribe command + ws.message_router.subscribe(vec![stream]).await?; - // Create a channel for streaming data - let (tx, rx) = mpsc::unbounded_channel::>(); + // Create user channel + let (user_tx, user_rx) = mpsc::unbounded_channel::>(); - // Spawn a task to receive messages and forward them + // Spawn parser task let market_clone = market.clone(); tokio::spawn(async move { - loop { - if let Some(msg) = ws.receive().await { - // Skip subscription confirmations - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - // Parse ticker - match super::parser::parse_ws_ticker(&msg, Some(&market_clone)) { - Ok(ticker) => { - if tx.send(Ok(ticker)).is_err() { - break; // Receiver dropped - } - } - Err(e) => { - let _ = tx.send(Err(e)); + while let Some(msg) = rx.recv().await { + match super::parser::parse_ws_ticker(&msg, Some(&market_clone)) { + Ok(ticker) => { + if user_tx.send(Ok(ticker)).is_err() { + break; } } + Err(e) => { + let _ = user_tx.send(Err(e)); + } } } }); // Convert receiver to stream - let stream = ReceiverStream::new(rx); + let stream = ReceiverStream::new(user_rx); Ok(Box::pin(stream)) } @@ -141,54 +128,66 @@ impl WsExchange for Binance { // Load markets self.load_markets(false).await?; - // Create WebSocket and connect - let ws = self.create_ws(); - ws.connect().await?; + // Create aggregator channel + let (agg_tx, mut agg_rx) = mpsc::unbounded_channel::(); - // Subscribe to all requested symbols + // Subscribe to all requested symbols (potentially across shards) let mut markets = HashMap::new(); for symbol in symbols { let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); - ws.subscribe_ticker(&binance_symbol).await?; - markets.insert(binance_symbol, market); + let stream = format!("{}@ticker", binance_symbol); + + markets.insert(binance_symbol.clone(), market); + + // Get connection (might be different for each symbol if sharding active) + let ws = self.connection_manager.get_public_connection().await?; + let (tx, mut rx) = mpsc::unbounded_channel(); + + ws.subscription_manager + .add_subscription( + stream.clone(), + symbol.clone(), + super::ws::SubscriptionType::Ticker, + tx, + ) + .await?; + + ws.message_router.subscribe(vec![stream]).await?; + + // Spawn parser task for this symbol + let agg_tx_clone = agg_tx.clone(); + let market_clone = self.base.market(symbol).await?; + + tokio::spawn(async move { + while let Some(msg) = rx.recv().await { + if let Ok(ticker) = super::parser::parse_ws_ticker(&msg, Some(&market_clone)) { + let _ = agg_tx_clone.send(ticker); + } + } + }); } - // Create channel for streaming - let (tx, rx) = mpsc::unbounded_channel::>>(); + drop(agg_tx); + + // Create user channel + let (user_tx, user_rx) = mpsc::unbounded_channel::>>(); - // Spawn receiver task - let markets_clone = markets; + // Spawn aggregator task tokio::spawn(async move { let mut tickers: HashMap = HashMap::new(); - loop { - if let Some(msg) = ws.receive().await { - // Skip subscription confirmations - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - // Get symbol from message - if let Some(symbol_str) = msg.get("s").and_then(|s| s.as_str()) { - let binance_symbol = symbol_str.to_lowercase(); - if let Some(market) = markets_clone.get(&binance_symbol) { - if let Ok(ticker) = super::parser::parse_ws_ticker(&msg, Some(market)) { - tickers.insert(ticker.symbol.clone(), ticker); + while let Some(ticker) = agg_rx.recv().await { + tickers.insert(ticker.symbol.clone(), ticker); - // Send current state of all tickers - let ticker_vec: Vec = tickers.values().cloned().collect(); - if tx.send(Ok(ticker_vec)).is_err() { - break; - } - } - } - } + let ticker_vec: Vec = tickers.values().cloned().collect(); + if user_tx.send(Ok(ticker_vec)).is_err() { + break; } } }); - let stream = ReceiverStream::new(rx); + let stream = ReceiverStream::new(user_rx); Ok(Box::pin(stream)) } @@ -204,44 +203,50 @@ impl WsExchange for Binance { let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); - // Create WebSocket and connect - let ws = self.create_ws(); - ws.connect().await?; - // Subscribe to orderbook stream let levels = limit.unwrap_or(20); - ws.subscribe_orderbook(&binance_symbol, levels, "100ms") + let stream = format!("{}@depth{}@100ms", binance_symbol, levels); + + // Get shared public connection + let ws = self.connection_manager.get_public_connection().await?; + + // Create subscription channel + let (tx, mut rx) = mpsc::unbounded_channel(); + + // Register with manager + ws.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + super::ws::SubscriptionType::OrderBook, + tx, + ) .await?; - // Create channel - let (tx, rx) = mpsc::unbounded_channel::>(); + // Send subscribe command + ws.message_router.subscribe(vec![stream]).await?; - // Spawn receiver task + // Create user channel + let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + + // Spawn parser task let symbol_clone = symbol.to_string(); tokio::spawn(async move { - loop { - if let Some(msg) = ws.receive().await { - // Skip subscription confirmations - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - // Parse orderbook - pass String as required by the parser - match super::parser::parse_ws_orderbook(&msg, symbol_clone.clone()) { - Ok(orderbook) => { - if tx.send(Ok(orderbook)).is_err() { - break; - } - } - Err(e) => { - let _ = tx.send(Err(e)); + while let Some(msg) = rx.recv().await { + match super::parser::parse_ws_orderbook(&msg, symbol_clone.clone()) { + Ok(orderbook) => { + if user_tx.send(Ok(orderbook)).is_err() { + break; } } + Err(e) => { + let _ = user_tx.send(Err(e)); + } } } }); - let stream = ReceiverStream::new(rx); + let stream = ReceiverStream::new(user_rx); Ok(Box::pin(stream)) } @@ -252,43 +257,49 @@ impl WsExchange for Binance { // Get market info let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); + let stream = format!("{}@trade", binance_symbol); + + // Get shared public connection + let ws = self.connection_manager.get_public_connection().await?; + + // Create subscription channel + let (tx, mut rx) = mpsc::unbounded_channel(); + + // Register with manager + ws.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + super::ws::SubscriptionType::Trades, + tx, + ) + .await?; - // Create WebSocket and connect - let ws = self.create_ws(); - ws.connect().await?; - - // Subscribe to trades stream - ws.subscribe_trades(&binance_symbol).await?; + // Send subscribe command + ws.message_router.subscribe(vec![stream]).await?; - // Create channel - let (tx, rx) = mpsc::unbounded_channel::>>(); + // Create user channel + let (user_tx, user_rx) = mpsc::unbounded_channel::>>(); - // Spawn receiver task + // Spawn parser task let market_clone = market.clone(); tokio::spawn(async move { - loop { - if let Some(msg) = ws.receive().await { - // Skip subscription confirmations - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - // Parse trade - match super::parser::parse_ws_trade(&msg, Some(&market_clone)) { - Ok(trade) => { - if tx.send(Ok(vec![trade])).is_err() { - break; - } - } - Err(e) => { - let _ = tx.send(Err(e)); + while let Some(msg) = rx.recv().await { + match super::parser::parse_ws_trade(&msg, Some(&market_clone)) { + Ok(trade) => { + if user_tx.send(Ok(vec![trade])).is_err() { + break; } } + Err(e) => { + let _ = user_tx.send(Err(e)); + } } } }); - let stream = ReceiverStream::new(rx); + // Convert receiver to stream + let stream = ReceiverStream::new(user_rx); Ok(Box::pin(stream)) } @@ -306,102 +317,249 @@ impl WsExchange for Binance { // Convert timeframe to Binance format let interval = timeframe.to_string(); + let stream = format!("{}@kline_{}", binance_symbol, interval); + + // Get shared public connection + let ws = self.connection_manager.get_public_connection().await?; + + // Create subscription channel + let (tx, mut rx) = mpsc::unbounded_channel(); + + // Register with manager + ws.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + super::ws::SubscriptionType::Kline(interval), + tx, + ) + .await?; - // Create WebSocket and connect - let ws = self.create_ws(); - ws.connect().await?; - - // Subscribe to kline stream - ws.subscribe_kline(&binance_symbol, &interval).await?; + // Send subscribe command + ws.message_router.subscribe(vec![stream]).await?; - // Create channel - let (tx, rx) = mpsc::unbounded_channel::>(); + // Create user channel + let (user_tx, user_rx) = mpsc::unbounded_channel::>(); - // Spawn receiver task + // Spawn parser task tokio::spawn(async move { - loop { - if let Some(msg) = ws.receive().await { - // Skip subscription confirmations - if msg.get("result").is_some() || msg.get("id").is_some() { - continue; - } - - // Parse OHLCV from kline message - match super::parser::parse_ws_ohlcv(&msg) { - Ok(ohlcv_f64) => { - // Convert OHLCV (f64) to Ohlcv (Decimal) - let ohlcv = Ohlcv { - timestamp: ohlcv_f64.timestamp, - open: Price::from( - Decimal::try_from(ohlcv_f64.open).unwrap_or_default(), - ), - high: Price::from( - Decimal::try_from(ohlcv_f64.high).unwrap_or_default(), - ), - low: Price::from( - Decimal::try_from(ohlcv_f64.low).unwrap_or_default(), - ), - close: Price::from( - Decimal::try_from(ohlcv_f64.close).unwrap_or_default(), - ), - volume: Amount::from( - Decimal::try_from(ohlcv_f64.volume).unwrap_or_default(), - ), - }; - if tx.send(Ok(ohlcv)).is_err() { - break; - } - } - Err(e) => { - let _ = tx.send(Err(e)); + while let Some(msg) = rx.recv().await { + // Parse OHLCV from kline message + match super::parser::parse_ws_ohlcv(&msg) { + Ok(ohlcv_f64) => { + // Convert OHLCV (f64) to Ohlcv (Decimal) + let ohlcv = Ohlcv { + timestamp: ohlcv_f64.timestamp, + open: Price::from( + Decimal::try_from(ohlcv_f64.open).unwrap_or_default(), + ), + high: Price::from( + Decimal::try_from(ohlcv_f64.high).unwrap_or_default(), + ), + low: Price::from(Decimal::try_from(ohlcv_f64.low).unwrap_or_default()), + close: Price::from( + Decimal::try_from(ohlcv_f64.close).unwrap_or_default(), + ), + volume: Amount::from( + Decimal::try_from(ohlcv_f64.volume).unwrap_or_default(), + ), + }; + if user_tx.send(Ok(ohlcv)).is_err() { + break; } } + Err(e) => { + let _ = user_tx.send(Err(e)); + } } } }); - let stream = ReceiverStream::new(rx); + let stream = ReceiverStream::new(user_rx); Ok(Box::pin(stream)) } // ==================== Private Data Streams ==================== async fn watch_balance(&self) -> Result> { - // Check credentials self.base .check_required_credentials() .map_err(|_| Error::authentication("API credentials required for watch_balance"))?; - // For user data streams, we need an Arc to create authenticated WS - // This is a limitation of the current design - Err(Error::not_implemented( - "watch_balance requires Arc for authenticated WebSocket. \ - Use create_authenticated_ws() directly for now.", - )) + let binance_arc = Arc::new(self.clone()); + let ws = self + .connection_manager + .get_private_connection(&binance_arc) + .await?; + + let (tx, mut rx) = mpsc::unbounded_channel(); + + ws.subscription_manager + .add_subscription( + "!userData".to_string(), + "user".to_string(), + super::ws::SubscriptionType::Balance, + tx, + ) + .await?; + + let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + let account_type = self.options.default_type.to_string(); + let balances_cache = ws.balances.clone(); + + tokio::spawn(async move { + while let Some(msg) = rx.recv().await { + if let Some(event_type) = msg.get("e").and_then(|e| e.as_str()) { + if matches!( + event_type, + "balanceUpdate" | "outboundAccountPosition" | "ACCOUNT_UPDATE" + ) { + if let Ok(()) = super::ws::user_data::handle_balance_message( + &msg, + &account_type, + &balances_cache, + ) + .await + { + let balances = balances_cache.read().await; + if let Some(balance) = balances.get(&account_type) { + if user_tx.send(Ok(balance.clone())).is_err() { + break; + } + } + } + } + } + } + }); + + let stream = ReceiverStream::new(user_rx); + Ok(Box::pin(stream)) } - async fn watch_orders(&self, _symbol: Option<&str>) -> Result> { - // Check credentials + async fn watch_orders(&self, symbol: Option<&str>) -> Result> { self.base .check_required_credentials() .map_err(|_| Error::authentication("API credentials required for watch_orders"))?; - Err(Error::not_implemented( - "watch_orders requires Arc for authenticated WebSocket. \ - Use create_authenticated_ws() directly for now.", - )) + let binance_arc = Arc::new(self.clone()); + let ws = self + .connection_manager + .get_private_connection(&binance_arc) + .await?; + + let (tx, mut rx) = mpsc::unbounded_channel(); + + ws.subscription_manager + .add_subscription( + "!userData".to_string(), + "user".to_string(), + super::ws::SubscriptionType::Orders, + tx, + ) + .await?; + + let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + let symbol_filter = symbol.map(ToString::to_string); + let orders_cache = ws.orders.clone(); + + tokio::spawn(async move { + while let Some(msg) = rx.recv().await { + if let Some(data) = msg.as_object() { + if let Some(event_type) = data.get("e").and_then(|e| e.as_str()) { + if event_type == "executionReport" { + let order = super::ws::user_data::parse_ws_order(data); + + // Update cache + { + let mut orders = orders_cache.write().await; + let symbol_orders = orders + .entry(order.symbol.clone()) + .or_insert_with(HashMap::new); + symbol_orders.insert(order.id.clone(), order.clone()); + } + + // Filter and send + if let Some(s) = &symbol_filter { + if &order.symbol != s { + continue; + } + } + + if user_tx.send(Ok(order)).is_err() { + break; + } + } + } + } + } + }); + + let stream = ReceiverStream::new(user_rx); + Ok(Box::pin(stream)) } - async fn watch_my_trades(&self, _symbol: Option<&str>) -> Result> { - // Check credentials + async fn watch_my_trades(&self, symbol: Option<&str>) -> Result> { self.base .check_required_credentials() .map_err(|_| Error::authentication("API credentials required for watch_my_trades"))?; - Err(Error::not_implemented( - "watch_my_trades requires Arc for authenticated WebSocket. \ - Use create_authenticated_ws() directly for now.", - )) + let binance_arc = Arc::new(self.clone()); + let ws = self + .connection_manager + .get_private_connection(&binance_arc) + .await?; + + let (tx, mut rx) = mpsc::unbounded_channel(); + + ws.subscription_manager + .add_subscription( + "!userData".to_string(), + "user".to_string(), + super::ws::SubscriptionType::MyTrades, + tx, + ) + .await?; + + let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + let symbol_filter = symbol.map(ToString::to_string); + let trades_cache = ws.my_trades.clone(); + + tokio::spawn(async move { + while let Some(msg) = rx.recv().await { + if let Some(event_type) = msg.get("e").and_then(|e| e.as_str()) { + if event_type == "executionReport" { + if let Ok(trade) = super::ws::user_data::parse_ws_trade(&msg) { + // Update cache + { + let mut trades = trades_cache.write().await; + let symbol_trades = trades + .entry(trade.symbol.clone()) + .or_insert_with(std::collections::VecDeque::new); + symbol_trades.push_front(trade.clone()); + if symbol_trades.len() > 1000 { + symbol_trades.pop_back(); + } + } + + // Filter and send + if let Some(s) = &symbol_filter { + if &trade.symbol != s { + continue; + } + } + + if user_tx.send(Ok(trade)).is_err() { + break; + } + } + } + } + } + }); + + let stream = ReceiverStream::new(user_rx); + Ok(Box::pin(stream)) } // ==================== Subscription Management ==================== @@ -459,13 +617,7 @@ impl WsExchange for Binance { } fn subscriptions(&self) -> Vec { - // Use try_read to avoid blocking - if let Ok(conn) = self.ws_connection.try_read() { - if let Some(ref ws) = *conn { - return ws.subscriptions(); - } - } - Vec::new() + self.connection_manager.get_all_subscriptions() } } From d3b601e6f44298ce6a320cae3ba7d4c60364554d Mon Sep 17 00:00:00 2001 From: Praying Date: Thu, 29 Jan 2026 20:31:25 +0800 Subject: [PATCH 3/3] fix(websocket): clean unused fn --- ccxt-exchanges/src/binance/ws/handlers.rs | 32 +---------------------- 1 file changed, 1 insertion(+), 31 deletions(-) diff --git a/ccxt-exchanges/src/binance/ws/handlers.rs b/ccxt-exchanges/src/binance/ws/handlers.rs index 7608f1e..d2a8cba 100644 --- a/ccxt-exchanges/src/binance/ws/handlers.rs +++ b/ccxt-exchanges/src/binance/ws/handlers.rs @@ -4,9 +4,9 @@ use crate::binance::Binance; use ccxt_core::error::{Error, Result}; +use ccxt_core::types::OrderBook; use ccxt_core::types::financial::{Amount, Price}; use ccxt_core::types::orderbook::{OrderBookDelta, OrderBookEntry}; -use ccxt_core::types::{OrderBook, Ticker}; use rust_decimal::Decimal; use serde_json::Value; use std::collections::HashMap; @@ -511,36 +511,6 @@ pub async fn fetch_orderbook_snapshot( Ok(snapshot) } -/// Watches a single ticker stream -pub async fn watch_ticker_internal( - ws_client: &ccxt_core::ws_client::WsClient, - symbol: &str, - channel_name: &str, - tickers: &Mutex>, - parser: &dyn Fn(&Value, Option<&ccxt_core::types::Market>) -> Result, -) -> Result { - let stream = format!("{}@{}", symbol.to_lowercase(), channel_name); - - ws_client - .subscribe(stream.clone(), Some(symbol.to_string()), None) - .await?; - - loop { - if let Some(message) = ws_client.receive().await { - if message.get("result").is_some() { - continue; - } - - if let Ok(ticker) = parser(&message, None) { - let mut tickers_map = tickers.lock().await; - tickers_map.insert(ticker.symbol.clone(), ticker.clone()); - - return Ok(ticker); - } - } - } -} - #[cfg(test)] mod tests { use super::*;