From 02fc66bf1bd51d05b8d81494112e6b6797e80938 Mon Sep 17 00:00:00 2001 From: Praying Date: Fri, 30 Jan 2026 23:08:39 +0800 Subject: [PATCH 1/2] fix: watch mark price --- ccxt-core/src/base_exchange/mod.rs | 8 +- ccxt-core/src/http_client/config.rs | 4 +- ccxt-core/src/ws_client/config.rs | 10 + ccxt-core/src/ws_client/mod.rs | 23 +- ccxt-exchanges/src/binance/parser/market.rs | 71 +++++- ccxt-exchanges/src/binance/parser/mod.rs | 5 +- .../src/binance/rest/market_data.rs | 62 ++++-- ccxt-exchanges/src/binance/ws/binance_impl.rs | 8 +- ccxt-exchanges/src/binance/ws/handlers.rs | 92 +++++++- ccxt-exchanges/src/binance/ws/mod.rs | 202 ++++++++++++++++-- ccxt-exchanges/tests/binance_ws_test.rs | 30 +-- examples/binance_ws_example.rs | 81 ++++--- 12 files changed, 488 insertions(+), 108 deletions(-) diff --git a/ccxt-core/src/base_exchange/mod.rs b/ccxt-core/src/base_exchange/mod.rs index 318ae73..e94dc99 100644 --- a/ccxt-core/src/base_exchange/mod.rs +++ b/ccxt-core/src/base_exchange/mod.rs @@ -93,10 +93,10 @@ impl BaseExchange { strategy_type: crate::retry_strategy::RetryStrategyType::Fixed, ..crate::retry_strategy::RetryConfig::default() }), - max_response_size: 10 * 1024 * 1024, // 10MB default - max_request_size: 10 * 1024 * 1024, // 10MB default - circuit_breaker: None, // Disabled by default for backward compatibility - pool_max_idle_per_host: 10, // Default: 10 idle connections per host + max_response_size: 128 * 1024 * 1024, // 128MB default + max_request_size: 10 * 1024 * 1024, // 10MB default + circuit_breaker: None, // Disabled by default for backward compatibility + pool_max_idle_per_host: 10, // Default: 10 idle connections per host pool_idle_timeout: Duration::from_secs(90), // Default: 90 seconds }; diff --git a/ccxt-core/src/http_client/config.rs b/ccxt-core/src/http_client/config.rs index 882cc7a..e79bc84 100644 --- a/ccxt-core/src/http_client/config.rs +++ b/ccxt-core/src/http_client/config.rs @@ -78,8 +78,8 @@ impl Default for HttpConfig { proxy: None, enable_rate_limit: true, retry_config: None, - max_response_size: 10 * 1024 * 1024, - max_request_size: 10 * 1024 * 1024, + max_response_size: 128 * 1024 * 1024, // 128MB default + max_request_size: 10 * 1024 * 1024, // 10MB default circuit_breaker: None, pool_max_idle_per_host: 10, pool_idle_timeout: Duration::from_secs(90), diff --git a/ccxt-core/src/ws_client/config.rs b/ccxt-core/src/ws_client/config.rs index 11330d0..e984241 100644 --- a/ccxt-core/src/ws_client/config.rs +++ b/ccxt-core/src/ws_client/config.rs @@ -156,6 +156,14 @@ pub struct WsConfig { /// /// Default: `DropOldest` to ensure most recent data is available. pub backpressure_strategy: BackpressureStrategy, + /// Maximum message size in bytes. + /// + /// Default: 128 MiB (134,217,728 bytes) if None. + pub max_message_size: Option, + /// Maximum frame size in bytes. + /// + /// Default: 32 MiB (33,554,432 bytes) if None. + pub max_frame_size: Option, } impl Default for WsConfig { @@ -175,6 +183,8 @@ impl Default for WsConfig { message_channel_capacity: DEFAULT_MESSAGE_CHANNEL_CAPACITY, write_channel_capacity: DEFAULT_WRITE_CHANNEL_CAPACITY, backpressure_strategy: BackpressureStrategy::default(), + max_message_size: Some(128 * 1024 * 1024), + max_frame_size: Some(32 * 1024 * 1024), } } } diff --git a/ccxt-core/src/ws_client/mod.rs b/ccxt-core/src/ws_client/mod.rs index fd2a068..43beda2 100644 --- a/ccxt-core/src/ws_client/mod.rs +++ b/ccxt-core/src/ws_client/mod.rs @@ -34,7 +34,8 @@ use tokio::net::TcpStream; use tokio::sync::{Mutex, RwLock, mpsc}; use tokio::time::{Duration, interval}; use tokio_tungstenite::{ - MaybeTlsStream, WebSocketStream, connect_async, tungstenite::protocol::Message, + MaybeTlsStream, WebSocketStream, connect_async_with_config, + tungstenite::protocol::{Message, WebSocketConfig}, }; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, instrument, warn}; @@ -154,9 +155,17 @@ impl WsClient { let url = self.config.url.clone(); info!("Initiating WebSocket connection"); + let mut ws_config = WebSocketConfig::default(); + if let Some(limit) = self.config.max_message_size { + ws_config.max_message_size = Some(limit); + } + if let Some(limit) = self.config.max_frame_size { + ws_config.max_frame_size = Some(limit); + } + match tokio::time::timeout( Duration::from_millis(self.config.connect_timeout), - connect_async(&url), + connect_async_with_config(&url, Some(ws_config), false), ) .await { @@ -214,6 +223,14 @@ impl WsClient { self.set_state(WsConnectionState::Connecting); let url = self.config.url.clone(); + let mut ws_config = WebSocketConfig::default(); + if let Some(limit) = self.config.max_message_size { + ws_config.max_message_size = Some(limit); + } + if let Some(limit) = self.config.max_frame_size { + ws_config.max_frame_size = Some(limit); + } + tokio::select! { biased; () = token.cancelled() => { @@ -223,7 +240,7 @@ impl WsClient { } result = tokio::time::timeout( Duration::from_millis(self.config.connect_timeout), - connect_async(&url), + connect_async_with_config(&url, Some(ws_config), false), ) => { match result { Ok(Ok((ws_stream, response))) => { diff --git a/ccxt-exchanges/src/binance/parser/market.rs b/ccxt-exchanges/src/binance/parser/market.rs index a00d58d..565a7cd 100644 --- a/ccxt-exchanges/src/binance/parser/market.rs +++ b/ccxt-exchanges/src/binance/parser/market.rs @@ -32,6 +32,54 @@ pub fn parse_market(data: &Value) -> Result { let active = status == "TRADING"; let margin = data["isMarginTradingAllowed"].as_bool().unwrap_or(false); + // Check for contract type to determine if this is a futures/swap market + let contract_type = data["contractType"].as_str(); + let is_contract = contract_type.is_some(); + + let market_type = if let Some(ct) = contract_type { + if ct == "PERPETUAL" { + MarketType::Swap + } else { + MarketType::Futures + } + } else { + MarketType::Spot + }; + + let mut linear = None; + let mut inverse = None; + let mut settle_id = None; + let mut settle = None; + + if is_contract { + if let Some(margin_asset) = data["marginAsset"].as_str() { + settle_id = Some(margin_asset.to_string()); + settle = Some(margin_asset.to_string()); // In Binance, asset ID is usually the code + + if margin_asset == quote_asset { + linear = Some(true); + inverse = Some(false); + } else if margin_asset == base_asset { + linear = Some(false); + inverse = Some(true); + } + } + } + + let unified_symbol = if is_contract { + if let Some(s) = &settle { + // For now simplified logic: base/quote:settle + // TODO: Add expiration handling for Futures (vs Swap) + format!("{}/{}:{}", base_asset, quote_asset, s) + } else { + format!("{}/{}", base_asset, quote_asset) + } + } else { + format!("{}/{}", base_asset, quote_asset) + }; + + let parsed_symbol = ccxt_core::symbol::SymbolParser::parse(&unified_symbol).ok(); + let mut price_precision: Option = None; let mut amount_precision: Option = None; let mut min_amount: Option = None; @@ -81,7 +129,16 @@ pub fn parse_market(data: &Value) -> Result { } } - let unified_symbol = format!("{}/{}", base_asset, quote_asset); + let unified_symbol = if is_contract { + if let Some(s) = &settle { + format!("{}/{}:{}", base_asset, quote_asset, s) + } else { + format!("{}/{}", base_asset, quote_asset) + } + } else { + format!("{}/{}", base_asset, quote_asset) + }; + let parsed_symbol = ccxt_core::symbol::SymbolParser::parse(&unified_symbol).ok(); Ok(Market { @@ -90,16 +147,16 @@ pub fn parse_market(data: &Value) -> Result { parsed_symbol, base: base_asset.clone(), quote: quote_asset.clone(), - settle: None, base_id: Some(base_asset), quote_id: Some(quote_asset), - settle_id: None, - market_type: MarketType::Spot, + settle_id, + settle, + market_type, active, margin, - contract: Some(false), - linear: None, - inverse: None, + contract: Some(is_contract), + linear, + inverse, taker: Decimal::from_str("0.001").ok(), maker: Decimal::from_str("0.001").ok(), contract_size: None, diff --git a/ccxt-exchanges/src/binance/parser/mod.rs b/ccxt-exchanges/src/binance/parser/mod.rs index e362664..db7a7ab 100644 --- a/ccxt-exchanges/src/binance/parser/mod.rs +++ b/ccxt-exchanges/src/binance/parser/mod.rs @@ -12,7 +12,10 @@ pub use balance::{parse_balance, parse_balance_with_type}; pub use funding::{parse_funding_rate, parse_funding_rate_history}; pub use margin::{parse_margin_adjustment, parse_margin_loan}; pub use market::{parse_currencies, parse_market}; -pub use misc::{parse_mark_prices, parse_stats_24hr, parse_trading_fee, parse_trading_limits}; +pub use misc::{ + parse_mark_prices, parse_stats_24hr, parse_trading_fee, parse_trading_limits, + parse_ws_mark_price, +}; pub use ohlcv::{parse_ohlcvs, parse_ws_ohlcv}; pub use order::parse_order; pub use orderbook::{parse_orderbook, parse_ws_orderbook}; diff --git a/ccxt-exchanges/src/binance/rest/market_data.rs b/ccxt-exchanges/src/binance/rest/market_data.rs index 8921e4e..6613562 100644 --- a/ccxt-exchanges/src/binance/rest/market_data.rs +++ b/ccxt-exchanges/src/binance/rest/market_data.rs @@ -3,13 +3,13 @@ //! This module contains all public market data methods that don't require authentication. //! These include ticker data, order books, trades, OHLCV data, and market statistics. -use super::super::{Binance, constants::endpoints, parser}; +use super::super::{Binance, BinanceEndpointRouter, constants::endpoints, parser}; use ccxt_core::{ Error, ParseError, Result, time::TimestampUtils, types::{ - AggTrade, BidAsk, IntoTickerParams, LastPrice, MarkPrice, OhlcvRequest, ServerTime, - Stats24hr, Ticker, Trade, TradingLimits, + AggTrade, BidAsk, EndpointType, IntoTickerParams, LastPrice, MarkPrice, OhlcvRequest, + ServerTime, Stats24hr, Ticker, Trade, TradingLimits, }, }; use reqwest::header::HeaderMap; @@ -44,7 +44,7 @@ impl Binance { /// /// Returns an error if the request fails or the response is malformed. pub(crate) async fn fetch_time_raw(&self) -> Result { - let url = format!("{}{}", self.urls().public, endpoints::TIME); + let url = format!("{}{}", self.get_rest_url_public(), endpoints::TIME); let response = self.base().http_client.get(&url, None).await?; response["serverTime"] @@ -67,7 +67,8 @@ impl Binance { /// } /// ``` pub async fn fetch_status(&self) -> Result { - let url = format!("{}{}", self.urls().sapi, endpoints::SYSTEM_STATUS); + // System status is specific to Spot/Margin (SAPI) + let url = format!("{}{}", self.sapi_endpoint(), endpoints::SYSTEM_STATUS); let response = self.base().http_client.get(&url, None).await?; // Response format: { "status": 0, "msg": "normal" } @@ -122,7 +123,7 @@ impl Binance { pub async fn fetch_markets( &self, ) -> Result>>> { - let url = format!("{}{}", self.urls().public, endpoints::EXCHANGE_INFO); + let url = format!("{}{}", self.get_rest_url_public(), endpoints::EXCHANGE_INFO); let data = self.base().http_client.get(&url, None).await?; let symbols = data["symbols"] @@ -236,7 +237,11 @@ impl Binance { endpoints::TICKER_24HR }; - let full_url = format!("{}{}", self.urls().public, endpoint); + let full_url = format!( + "{}{}", + self.rest_endpoint(&market, EndpointType::Public), + endpoint + ); let mut url = Url::parse(&full_url).map_err(|e| Error::exchange("Invalid URL", e.to_string()))?; @@ -294,7 +299,7 @@ impl Binance { .collect(); drop(cache); - let url = format!("{}{}", self.urls().public, endpoints::TICKER_24HR); + let url = format!("{}{}", self.get_rest_url_public(), endpoints::TICKER_24HR); let data = self.base().http_client.get(&url, None).await?; let tickers_array = data.as_array().ok_or_else(|| { @@ -355,7 +360,11 @@ impl Binance { ) -> Result { let market = self.base().market(symbol).await?; - let full_url = format!("{}{}", self.urls().public, endpoints::DEPTH); + let full_url = format!( + "{}{}", + self.rest_endpoint(&market, EndpointType::Public), + endpoints::DEPTH + ); let mut url = Url::parse(&full_url).map_err(|e| Error::exchange("Invalid URL", e.to_string()))?; @@ -392,7 +401,7 @@ impl Binance { let url = if let Some(l) = limit { format!( "{}{}?symbol={}&limit={}", - self.urls().public, + self.rest_endpoint(&market, EndpointType::Public), endpoints::TRADES, market.id, l @@ -400,7 +409,7 @@ impl Binance { } else { format!( "{}{}?symbol={}", - self.urls().public, + self.rest_endpoint(&market, EndpointType::Public), endpoints::TRADES, market.id ) @@ -482,7 +491,7 @@ impl Binance { let mut url = format!( "{}{}?symbol={}", - self.urls().public, + self.rest_endpoint(&market, EndpointType::Public), endpoints::AGG_TRADES, market.id ); @@ -562,7 +571,7 @@ impl Binance { let mut url = format!( "{}{}?symbol={}", - self.urls().public, + self.rest_endpoint(&market, EndpointType::Public), endpoints::HISTORICAL_TRADES, market.id ); @@ -624,12 +633,12 @@ impl Binance { let market = self.base().market(sym).await?; format!( "{}{}?symbol={}", - self.urls().public, + self.rest_endpoint(&market, EndpointType::Public), endpoints::TICKER_24HR, market.id ) } else { - format!("{}{}", self.urls().public, endpoints::TICKER_24HR) + format!("{}{}", self.get_rest_url_public(), endpoints::TICKER_24HR) }; let data = self.base().http_client.get(&url, None).await?; @@ -678,7 +687,7 @@ impl Binance { let url = format!( "{}{}?symbol={}", - self.urls().public, + self.rest_endpoint(&market, EndpointType::Public), endpoints::EXCHANGE_INFO, market.id ); @@ -1115,11 +1124,11 @@ impl Binance { let market = self.base().market(sym).await?; format!( "{}/ticker/bookTicker?symbol={}", - self.urls().public, + self.rest_endpoint(&market, EndpointType::Public), market.id ) } else { - format!("{}/ticker/bookTicker", self.urls().public) + format!("{}/ticker/bookTicker", self.get_rest_url_public()) }; let data = self.base().http_client.get(&url, None).await?; @@ -1172,9 +1181,13 @@ impl Binance { let url = if let Some(sym) = symbol { let market = self.base().market(sym).await?; - format!("{}/ticker/price?symbol={}", self.urls().public, market.id) + format!( + "{}/ticker/price?symbol={}", + self.rest_endpoint(&market, EndpointType::Public), + market.id + ) } else { - format!("{}/ticker/price", self.urls().public) + format!("{}/ticker/price", self.get_rest_url_public()) }; let data = self.base().http_client.get(&url, None).await?; @@ -1233,13 +1246,18 @@ impl Binance { let url = if let Some(sym) = symbol { let market = self.base().market(sym).await?; + // Use rest_endpoint to correctly select FAPI/DAPI format!( "{}/premiumIndex?symbol={}", - self.urls().fapi_public, + self.rest_endpoint(&market, EndpointType::Public), market.id ) } else { - format!("{}/premiumIndex", self.urls().fapi_public) + // Default to the exchange's configured futures type (FAPI/DAPI) + format!( + "{}/premiumIndex", + self.default_rest_endpoint(EndpointType::Public) + ) }; let data = self.base().http_client.get(&url, None).await?; diff --git a/ccxt-exchanges/src/binance/ws/binance_impl.rs b/ccxt-exchanges/src/binance/ws/binance_impl.rs index 10f69f7..f4b3c52 100644 --- a/ccxt-exchanges/src/binance/ws/binance_impl.rs +++ b/ccxt-exchanges/src/binance/ws/binance_impl.rs @@ -98,7 +98,7 @@ impl Binance { &self, symbol: &str, params: Option>, - ) -> Result { + ) -> Result { self.load_markets(false).await?; let market = self.base.market(symbol).await?; @@ -126,7 +126,7 @@ impl Binance { }; let ws = self.connection_manager.get_public_connection().await?; - ws.watch_ticker_internal(&binance_symbol, channel_name) + ws.watch_mark_price_internal(&binance_symbol, channel_name) .await } @@ -218,7 +218,7 @@ impl Binance { &self, symbols: Option>, params: Option>, - ) -> Result> { + ) -> Result> { self.load_markets(false).await?; let use_1s_freq = if let Some(p) = ¶ms { @@ -255,7 +255,7 @@ impl Binance { }; let ws = self.connection_manager.get_public_connection().await?; - ws.watch_tickers_internal(binance_symbols, channel_name) + ws.watch_mark_prices_internal(binance_symbols, channel_name) .await } diff --git a/ccxt-exchanges/src/binance/ws/handlers.rs b/ccxt-exchanges/src/binance/ws/handlers.rs index 85f6315..1c5dfe0 100644 --- a/ccxt-exchanges/src/binance/ws/handlers.rs +++ b/ccxt-exchanges/src/binance/ws/handlers.rs @@ -416,18 +416,55 @@ impl MessageRouter { let symbol_opt = payload.get("s").and_then(|s| s.as_str()); if let Some(symbol) = symbol_opt { - let active_streams = subscription_manager + let mut active_streams = subscription_manager .get_subscriptions_by_symbol(symbol) .await; + + tracing::debug!( + "Routing message for symbol {}: stream_name={}, active_subscriptions={}", + symbol, + stream_name, + active_streams.len() + ); + + // If no streams found for uppercase symbol (e.g. "BTCUSDT"), try lowercase (e.g. "btcusdt") + // This is common since Binance API sends uppercase symbols but we usually subscribe with lowercase + if active_streams.is_empty() { + let lower_symbol = symbol.to_lowercase(); + if lower_symbol != symbol { + let lower_streams = subscription_manager + .get_subscriptions_by_symbol(&lower_symbol) + .await; + if !lower_streams.is_empty() { + active_streams = lower_streams; + tracing::debug!( + "Found subscriptions for lowercased symbol {}: count={}", + lower_symbol, + active_streams.len() + ); + } + } + } + let mut fallback_sent = false; for sub in active_streams { - if sub.stream.starts_with(&stream_name) && sub.stream != stream_name { + // For mark price updates, the stream name is usually like "btcusdt@markPrice" or "btcusdt@markprice" + // But the user might have subscribed to "btcusdt@markprice@1s" + // The starts_with check handles the suffix @1s + tracing::debug!( + "Checking subscription: stream={}, expected_starts_with={}", + sub.stream, + stream_name + ); + + if sub.stream.starts_with(&stream_name) { if subscription_manager .send_to_stream(&sub.stream, payload.clone()) .await { fallback_sent = true; + tracing::debug!("Successfully routed to fallback stream: {}", sub.stream); } } } @@ -446,6 +483,19 @@ impl MessageRouter { return Ok(stream.to_string()); } + // Handle array messages (e.g. !ticker@arr, !miniTicker@arr) + if let Some(arr) = message.as_array() { + if let Some(first) = arr.first() { + if let Some(event_type) = first.get("e").and_then(|e| e.as_str()) { + match event_type { + "24hrTicker" => return Ok("!ticker@arr".to_string()), + "24hrMiniTicker" => return Ok("!miniTicker@arr".to_string()), + _ => {} + } + } + } + } + if let Some(event_type) = message.get("e").and_then(|e| e.as_str()) { match event_type { "outboundAccountPosition" @@ -463,6 +513,7 @@ impl MessageRouter { if let Some(symbol) = message.get("s").and_then(|s| s.as_str()) { let stream = match event_type { "24hrTicker" => format!("{}@ticker", symbol.to_lowercase()), + "24hrMiniTicker" => format!("{}@miniTicker", symbol.to_lowercase()), "depthUpdate" => format!("{}@depth", symbol.to_lowercase()), "aggTrade" => format!("{}@aggTrade", symbol.to_lowercase()), "trade" => format!("{}@trade", symbol.to_lowercase()), @@ -477,7 +528,7 @@ impl MessageRouter { return Err(Error::generic("Missing kline data")); } } - "markPriceUpdate" => format!("{}@markPrice", symbol.to_lowercase()), + "markPriceUpdate" => format!("{}@markprice", symbol.to_lowercase()), "bookTicker" => format!("{}@bookTicker", symbol.to_lowercase()), _ => { return Err(Error::generic(format!( @@ -728,4 +779,39 @@ mod tests { assert_eq!(received["e"], "24hrTicker"); assert_eq!(received["c"], "50000.00"); } + + #[tokio::test] + async fn test_handle_message_mark_price_fallback() { + let manager = Arc::new(crate::binance::ws::subscriptions::SubscriptionManager::new()); + let (tx, mut rx) = tokio::sync::mpsc::channel(100); + + // Subscription uses lowercase symbol "btcusdt" and specific stream "btcusdt@markprice@1s" + manager + .add_subscription( + "btcusdt@markprice@1s".to_string(), + "btcusdt".to_string(), + crate::binance::ws::subscriptions::SubscriptionType::MarkPrice, + tx, + ) + .await + .unwrap(); + + // Incoming message has uppercase symbol "BTCUSDT" and event "markPriceUpdate" + // extract_stream_name will derive "btcusdt@markprice" + let message = json!({ + "e": "markPriceUpdate", + "s": "BTCUSDT", + "p": "50000.00", + "E": 123456789 + }); + + // This should succeed thanks to the fallback logic + MessageRouter::handle_message(message, manager, None) + .await + .unwrap(); + + let received = rx.recv().await.unwrap(); + assert_eq!(received["e"], "markPriceUpdate"); + assert_eq!(received["p"], "50000.00"); + } } diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index 44bb503..2e17aa4 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/ccxt-exchanges/src/binance/ws/mod.rs @@ -410,6 +410,157 @@ impl BinanceWs { self.subscription_manager.active_count() } + /// Watches a single mark price stream (internal helper) + async fn watch_mark_price_internal( + &self, + symbol: &str, + channel_name: &str, + ) -> Result { + let stream = format!("{}@{}", symbol.to_lowercase(), channel_name); + tracing::debug!( + "watch_mark_price_internal: stream={}, symbol={}", + stream, + symbol + ); + + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + self.subscription_manager + .add_subscription( + stream.clone(), + symbol.to_string(), + SubscriptionType::MarkPrice, + tx, + ) + .await?; + + self.message_router.subscribe(vec![stream.clone()]).await?; + + loop { + if let Some(message) = rx.recv().await { + if message.get("result").is_some() { + tracing::debug!("Received subscription result for {}", stream); + continue; + } + + tracing::debug!("Received mark price message for {}", stream); + + match parser::parse_ws_mark_price(&message) { + Ok(mark_price) => { + let mut mark_prices = self.mark_prices.lock().await; + mark_prices.insert(mark_price.symbol.clone(), mark_price.clone()); + return Ok(mark_price); + } + Err(e) => { + tracing::warn!( + "Failed to parse mark price message for stream {}: {:?}. Payload: {:?}", + stream, + e, + message + ); + } + } + } else { + tracing::warn!("Subscription channel closed for {}", stream); + return Err(Error::network("Subscription channel closed")); + } + } + } + + /// Watches multiple mark price streams (internal helper) + async fn watch_mark_prices_internal( + &self, + symbols: Option>, + channel_name: &str, + ) -> Result> { + let streams: Vec = if let Some(syms) = symbols.as_ref() { + syms.iter() + .map(|s| format!("{}@{}", s.to_lowercase(), channel_name)) + .collect() + } else { + vec![format!("!{}@arr", channel_name)] + }; + + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + + for stream in &streams { + self.subscription_manager + .add_subscription( + stream.clone(), + "all".to_string(), + SubscriptionType::MarkPrice, + tx.clone(), + ) + .await?; + } + + self.message_router.subscribe(streams.clone()).await?; + + let mut result = HashMap::new(); + + loop { + if let Some(message) = rx.recv().await { + if message.get("result").is_some() { + continue; + } + + if let Some(arr) = message.as_array() { + for item in arr { + if let Ok(mark_price) = parser::parse_ws_mark_price(item) { + let symbol = mark_price.symbol.clone(); + + if let Some(syms) = &symbols { + if syms.contains(&symbol.to_lowercase()) { + result.insert(symbol.clone(), mark_price.clone()); + } + } else { + result.insert(symbol.clone(), mark_price.clone()); + } + + let mut mark_prices = self.mark_prices.lock().await; + mark_prices.insert(symbol, mark_price); + } else { + tracing::warn!("Failed to parse item in mark price array: {:?}", item); + } + } + + if let Some(syms) = &symbols { + if result.len() >= syms.len() { + return Ok(result); + } + } else { + // For array updates without specific symbols filter, return what we got + return Ok(result); + } + } else { + match parser::parse_ws_mark_price(&message) { + Ok(mark_price) => { + let symbol = mark_price.symbol.clone(); + result.insert(symbol.clone(), mark_price.clone()); + + let mut mark_prices = self.mark_prices.lock().await; + mark_prices.insert(symbol, mark_price); + + if let Some(syms) = &symbols { + if result.len() >= syms.len() { + return Ok(result); + } + } + } + Err(e) => { + tracing::warn!( + "Failed to parse mark price message: {:?}. Payload: {:?}", + e, + message + ); + } + } + } + } else { + return Err(Error::network("Subscription channel closed")); + } + } + } + /// 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); @@ -432,10 +583,21 @@ impl BinanceWs { continue; } - if let Ok(ticker) = parser::parse_ws_ticker(&message, None) { - let mut tickers = self.tickers.lock().await; - tickers.insert(ticker.symbol.clone(), ticker.clone()); - return Ok(ticker); + match parser::parse_ws_ticker(&message, None) { + Ok(ticker) => { + let mut tickers = self.tickers.lock().await; + tickers.insert(ticker.symbol.clone(), ticker.clone()); + return Ok(ticker); + } + Err(e) => { + tracing::warn!( + "Failed to parse ticker message for stream {}: {:?}. Payload: {:?}", + stream, + e, + message + ); + // Continue waiting for the next message instead of hanging silently + } } } else { return Err(Error::network("Subscription channel closed")); @@ -495,6 +657,8 @@ impl BinanceWs { let mut tickers = self.tickers.lock().await; tickers.insert(symbol, ticker); + } else { + tracing::warn!("Failed to parse item in ticker array: {:?}", item); } } @@ -503,18 +667,32 @@ impl BinanceWs { return Ok(result); } } else { + // If we received an array but we are not waiting for specific symbols, + // we can assume we got the update for "all" markets. + // However, !ticker@arr returns ALL tickers in one message usually. return Ok(result); } - } else if let Ok(ticker) = parser::parse_ws_ticker(&message, None) { - let symbol = ticker.symbol.clone(); - result.insert(symbol.clone(), ticker.clone()); + } else { + match parser::parse_ws_ticker(&message, None) { + Ok(ticker) => { + let symbol = ticker.symbol.clone(); + result.insert(symbol.clone(), ticker.clone()); - let mut tickers = self.tickers.lock().await; - tickers.insert(symbol, ticker); + let mut tickers = self.tickers.lock().await; + tickers.insert(symbol, ticker); - if let Some(syms) = &symbols { - if result.len() >= syms.len() { - return Ok(result); + if let Some(syms) = &symbols { + if result.len() >= syms.len() { + return Ok(result); + } + } + } + Err(e) => { + tracing::warn!( + "Failed to parse ticker message: {:?}. Payload: {:?}", + e, + message + ); } } } diff --git a/ccxt-exchanges/tests/binance_ws_test.rs b/ccxt-exchanges/tests/binance_ws_test.rs index c12c1ba..1e6bb83 100644 --- a/ccxt-exchanges/tests/binance_ws_test.rs +++ b/ccxt-exchanges/tests/binance_ws_test.rs @@ -122,12 +122,12 @@ mod tests { let exchange = Binance::new(ExchangeConfig::default()).unwrap(); match exchange.watch_mark_price("BTC/USDT:USDT", None).await { - Ok(ticker) => { + Ok(mark_price) => { println!("✓ watch_mark_price succeeded"); - println!(" Symbol: {}", ticker.symbol); - println!(" Mark Price: {:?}", ticker.last); - println!(" Index Price: {:?}", ticker.info.get("indexPrice")); - assert!(ticker.symbol.contains("BTC")); + println!(" Symbol: {}", mark_price.symbol); + println!(" Mark Price: {:?}", mark_price.mark_price); + println!(" Index Price: {:?}", mark_price.index_price); + assert!(mark_price.symbol.contains("BTC") || mark_price.symbol.contains("btc")); } Err(e) => { println!("✗ watch_mark_price failed: {}", e); @@ -148,11 +148,11 @@ mod tests { .watch_mark_price("ETH/USDT:USDT", Some(params)) .await { - Ok(ticker) => { + Ok(mark_price) => { println!("✓ watch_mark_price (3s update) succeeded"); - println!(" Symbol: {}", ticker.symbol); - println!(" Mark Price: {:?}", ticker.last); - assert!(ticker.symbol.contains("ETH")); + println!(" Symbol: {}", mark_price.symbol); + println!(" Mark Price: {:?}", mark_price.mark_price); + assert!(mark_price.symbol.contains("ETH") || mark_price.symbol.contains("eth")); } Err(e) => { println!("✗ watch_mark_price (3s update) failed: {}", e); @@ -172,17 +172,17 @@ mod tests { .watch_mark_prices(Some(symbols.clone()), None) .await { - Ok(tickers) => { + Ok(mark_prices) => { println!( - "✓ watch_mark_prices succeeded, received {} tickers", - tickers.len() + "✓ watch_mark_prices succeeded, received {} mark prices", + mark_prices.len() ); - for (symbol, ticker) in &tickers { - println!(" {} - Mark: {:?}", symbol, ticker.last); + for (symbol, mark_price) in &mark_prices { + println!(" {} - Mark: {:?}", symbol, mark_price.mark_price); } - assert!(tickers.len() > 0); + assert!(mark_prices.len() > 0); } Err(e) => { println!("✗ watch_mark_prices failed: {}", e); diff --git a/examples/binance_ws_example.rs b/examples/binance_ws_example.rs index d0082e2..fe8d4e9 100644 --- a/examples/binance_ws_example.rs +++ b/examples/binance_ws_example.rs @@ -6,6 +6,7 @@ #![allow(clippy::disallowed_methods)] use anyhow::{Context, Result}; +use ccxt_core::logging::{LogConfig, init_logging}; use ccxt_core::prelude::{Amount, ExchangeConfig, Price}; use ccxt_exchanges::binance::Binance; use rust_decimal::Decimal; @@ -16,7 +17,9 @@ use tokio::time::{Duration, sleep}; #[tokio::main] async fn main() -> Result<()> { println!("=== Binance WebSocket Example ===\n"); - + let mut config = LogConfig::development(); + config.show_span_events = false; + init_logging(&config); let exchange = Binance::new(ExchangeConfig::default()).context("Failed to initialize Binance exchange")?; @@ -136,29 +139,27 @@ async fn main() -> Result<()> { // Example 5: Monitor futures mark price (1-second updates) println!("\n5. Monitor futures mark price (BTC/USDT:USDT, 1-second updates)"); + // Initialize futures exchange for monitoring futures markets + let futures_exchange = Binance::new_swap(ExchangeConfig::default()) + .context("Failed to initialize Binance futures exchange")?; + let mut params = HashMap::new(); params.insert("use1sFreq".to_string(), json!(true)); - match exchange + match futures_exchange .watch_mark_price("BTC/USDT:USDT", Some(params)) .await { - Ok(ticker) => { + Ok(mark_price) => { println!(" ✓ Mark price connected successfully!"); - println!(" Symbol: {}", ticker.symbol); - println!( - " Mark Price: {}", - ticker.last.unwrap_or(Price(Decimal::ZERO)) - ); + println!(" Symbol: {}", mark_price.symbol); + println!(" Mark Price: {}", mark_price.mark_price); - if !ticker.info.is_empty() { - let info = &ticker.info; - if let Some(index_price) = info.get("indexPrice") { - println!(" Index Price: {}", index_price); - } - if let Some(funding_rate) = info.get("fundingRate") { - println!(" Funding Rate: {}", funding_rate); - } + if let Some(index_price) = mark_price.index_price { + println!(" Index Price: {}", index_price); + } + if let Some(funding_rate) = mark_price.last_funding_rate { + println!(" Funding Rate: {}", funding_rate); } } Err(e) => { @@ -173,19 +174,15 @@ async fn main() -> Result<()> { let futures_symbols = vec!["BTC/USDT:USDT".to_string(), "ETH/USDT:USDT".to_string()]; - match exchange + match futures_exchange .watch_mark_prices(Some(futures_symbols.clone()), None) .await { - Ok(tickers) => { - println!(" ✓ Received {} mark prices", tickers.len()); + Ok(mark_prices) => { + println!(" ✓ Received {} mark prices", mark_prices.len()); - for (symbol, ticker) in &tickers { - println!( - " {} - Mark: ${:.2}", - symbol, - ticker.last.unwrap_or(Price(Decimal::ZERO)) - ); + for (symbol, mark_price) in &mark_prices { + println!(" {} - Mark: ${}", symbol, mark_price.mark_price); } } Err(e) => { @@ -193,18 +190,32 @@ async fn main() -> Result<()> { } } - // Example 7: Continuous ticker monitoring instructions - println!("\n7. Continuous ticker monitoring"); + // Example 7: Continuous ticker monitoring + println!("\n7. Continuous ticker monitoring (Running for 5 seconds)"); println!(" Tip: Continuous monitoring requires the watch_ticker() method"); println!(" This method maintains WebSocket connection and continuously pushes updates"); - println!(" Example code:"); - println!(" ```rust"); - println!(" loop {{"); - println!(" let ticker = exchange.watch_ticker(\"BTC/USDT\", None).await?;"); - println!(" println!(\"Price: {{}}\", ticker.last.unwrap_or(Price(Decimal::ZERO)));"); - println!(" // Process ticker data..."); - println!(" }}"); - println!(" ```"); + println!(" Starting monitoring loop..."); + + let start = std::time::Instant::now(); + let duration = Duration::from_secs(5); + + while start.elapsed() < duration { + match exchange.watch_ticker("BTC/USDT", None).await { + Ok(ticker) => { + println!( + " [{}] {} - ${}", + chrono::Local::now().format("%H:%M:%S"), + ticker.symbol, + ticker.last.unwrap_or(Price(Decimal::ZERO)) + ); + } + Err(e) => { + eprintln!(" Error: {}", e); + break; + } + } + } + println!(" Stopped."); sleep(Duration::from_secs(2)).await; From d028eed9598d49956ef256112db60a67b37e3627 Mon Sep 17 00:00:00 2001 From: quran Date: Sat, 31 Jan 2026 00:24:00 +0800 Subject: [PATCH 2/2] fix(binance): correct markPrice stream name and enhance futures parsing - Fixed markPrice stream subscription from "@markprice" to "@markPrice" to match Binance API - Added futures contract expiry timestamp parsing from deliveryDate field - Enhanced symbol parsing to include expiry suffix for futures contracts - Updated unified symbol format to include expiry information for futures - Improved mark price subscription logic with proper channel management - Added timestamp to datetime conversion for futures expiry dates --- ccxt-exchanges/src/binance/parser/market.rs | 42 ++++++++++++++------- ccxt-exchanges/src/binance/ws/handlers.rs | 8 ++-- ccxt-exchanges/src/binance/ws/mod.rs | 30 ++++++++++----- 3 files changed, 53 insertions(+), 27 deletions(-) diff --git a/ccxt-exchanges/src/binance/parser/market.rs b/ccxt-exchanges/src/binance/parser/market.rs index 565a7cd..312e147 100644 --- a/ccxt-exchanges/src/binance/parser/market.rs +++ b/ccxt-exchanges/src/binance/parser/market.rs @@ -2,6 +2,7 @@ use super::value_to_hashmap; use ccxt_core::{ Result, error::{Error, ParseError}, + parser_utils::timestamp_to_datetime, types::{Market, MarketLimits, MarketPrecision, MarketType, MinMax}, }; use rust_decimal::Decimal; @@ -66,19 +67,26 @@ pub fn parse_market(data: &Value) -> Result { } } - let unified_symbol = if is_contract { - if let Some(s) = &settle { - // For now simplified logic: base/quote:settle - // TODO: Add expiration handling for Futures (vs Swap) - format!("{}/{}:{}", base_asset, quote_asset, s) - } else { - format!("{}/{}", base_asset, quote_asset) - } + let expiry_timestamp = if market_type == MarketType::Futures { + data["deliveryDate"].as_i64().filter(|ts| *ts > 0) } else { - format!("{}/{}", base_asset, quote_asset) + None }; - let parsed_symbol = ccxt_core::symbol::SymbolParser::parse(&unified_symbol).ok(); + let expiry_datetime = expiry_timestamp.and_then(timestamp_to_datetime); + + let expiry_suffix = if market_type == MarketType::Futures { + symbol.rfind('_').and_then(|pos| { + let suffix = &symbol[pos + 1..]; + if suffix.len() == 6 && suffix.chars().all(|c| c.is_ascii_digit()) { + Some(suffix.to_string()) + } else { + None + } + }) + } else { + None + }; let mut price_precision: Option = None; let mut amount_precision: Option = None; @@ -131,7 +139,15 @@ pub fn parse_market(data: &Value) -> Result { let unified_symbol = if is_contract { if let Some(s) = &settle { - format!("{}/{}:{}", base_asset, quote_asset, s) + if market_type == MarketType::Futures { + if let Some(expiry) = &expiry_suffix { + format!("{}/{}:{}-{}", base_asset, quote_asset, s, expiry) + } else { + format!("{}/{}:{}", base_asset, quote_asset, s) + } + } else { + format!("{}/{}:{}", base_asset, quote_asset, s) + } } else { format!("{}/{}", base_asset, quote_asset) } @@ -160,8 +176,8 @@ pub fn parse_market(data: &Value) -> Result { taker: Decimal::from_str("0.001").ok(), maker: Decimal::from_str("0.001").ok(), contract_size: None, - expiry: None, - expiry_datetime: None, + expiry: expiry_timestamp, + expiry_datetime, strike: None, option_type: None, percentage: Some(true), diff --git a/ccxt-exchanges/src/binance/ws/handlers.rs b/ccxt-exchanges/src/binance/ws/handlers.rs index 1c5dfe0..490e57a 100644 --- a/ccxt-exchanges/src/binance/ws/handlers.rs +++ b/ccxt-exchanges/src/binance/ws/handlers.rs @@ -528,7 +528,7 @@ impl MessageRouter { return Err(Error::generic("Missing kline data")); } } - "markPriceUpdate" => format!("{}@markprice", symbol.to_lowercase()), + "markPriceUpdate" => format!("{}@markPrice", symbol.to_lowercase()), "bookTicker" => format!("{}@bookTicker", symbol.to_lowercase()), _ => { return Err(Error::generic(format!( @@ -785,10 +785,10 @@ mod tests { let manager = Arc::new(crate::binance::ws::subscriptions::SubscriptionManager::new()); let (tx, mut rx) = tokio::sync::mpsc::channel(100); - // Subscription uses lowercase symbol "btcusdt" and specific stream "btcusdt@markprice@1s" + // Subscription uses lowercase symbol "btcusdt" and specific stream "btcusdt@markPrice@1s" manager .add_subscription( - "btcusdt@markprice@1s".to_string(), + "btcusdt@markPrice@1s".to_string(), "btcusdt".to_string(), crate::binance::ws::subscriptions::SubscriptionType::MarkPrice, tx, @@ -797,7 +797,7 @@ mod tests { .unwrap(); // Incoming message has uppercase symbol "BTCUSDT" and event "markPriceUpdate" - // extract_stream_name will derive "btcusdt@markprice" + // extract_stream_name will derive "btcusdt@markPrice" let message = json!({ "e": "markPriceUpdate", "s": "BTCUSDT", diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index 2e17aa4..192f2e9 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/ccxt-exchanges/src/binance/ws/mod.rs @@ -472,17 +472,26 @@ impl BinanceWs { symbols: Option>, channel_name: &str, ) -> Result> { - let streams: Vec = if let Some(syms) = symbols.as_ref() { - syms.iter() - .map(|s| format!("{}@{}", s.to_lowercase(), channel_name)) - .collect() - } else { - vec![format!("!{}@arr", channel_name)] - }; - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); - for stream in &streams { + let streams: Vec = if let Some(syms) = symbols.as_ref() { + let mut streams = Vec::with_capacity(syms.len()); + for sym in syms { + let symbol = sym.to_lowercase(); + let stream = format!("{}@{}", symbol, channel_name); + self.subscription_manager + .add_subscription( + stream.clone(), + symbol, + SubscriptionType::MarkPrice, + tx.clone(), + ) + .await?; + streams.push(stream); + } + streams + } else { + let stream = format!("!{}@arr", channel_name); self.subscription_manager .add_subscription( stream.clone(), @@ -491,7 +500,8 @@ impl BinanceWs { tx.clone(), ) .await?; - } + vec![stream] + }; self.message_router.subscribe(streams.clone()).await?;