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(); +}