From a01d51c80fcb234ff4aec1c980a6d2c3f3365a1e Mon Sep 17 00:00:00 2001 From: quran Date: Sat, 7 Feb 2026 09:21:22 +0800 Subject: [PATCH] refactor(binance): standardize HTTP client usage and improve WebSocket subscription handling - Replace direct http_client calls with unified public_get method for all public API endpoints - Add comprehensive documentation for WebSocket URL routing by market type in connection manager - Implement automatic time sync and retry mechanism for signed requests with timestamp errors - Enhance rate limiter to handle additional header variations including SAPI endpoints - Refactor WebSocket subscription system to support multiple subscribers per stream - Remove manual MessageRouter spawning in WebSocket constructors for better lifecycle management - Consolidate signed request logic into reusable builder pattern with proper error handling - Optimize futures margin operations using new signed request builder methods --- ccxt-exchanges/src/binance/mod.rs | 26 ++++ ccxt-exchanges/src/binance/rate_limiter.rs | 7 +- .../src/binance/rest/futures/funding.rs | 6 +- .../src/binance/rest/futures/margin.rs | 145 +++--------------- .../src/binance/rest/market_data.rs | 32 ++-- ccxt-exchanges/src/binance/signed_request.rs | 37 +++-- .../src/binance/ws/connection_manager.rs | 11 ++ ccxt-exchanges/src/binance/ws/mod.rs | 24 --- .../src/binance/ws/subscriptions.rs | 44 +++++- 9 files changed, 146 insertions(+), 186 deletions(-) diff --git a/ccxt-exchanges/src/binance/mod.rs b/ccxt-exchanges/src/binance/mod.rs index 8cdace9..e8201de 100644 --- a/ccxt-exchanges/src/binance/mod.rs +++ b/ccxt-exchanges/src/binance/mod.rs @@ -717,6 +717,32 @@ impl Binance { } Ok(response) } + + /// Executes a public GET request with unified error checking and rate limiter updates. + /// + /// This method wraps `http_client.get()` to provide: + /// - Rate limiter updates from response headers + /// - API error checking (Binance may return errors in the body with HTTP 200) + /// + /// All public REST API methods should use this instead of calling `http_client.get()` directly. + pub(crate) async fn public_get( + &self, + url: &str, + headers: Option, + ) -> Result { + let data = self.base().http_client.get(url, headers).await?; + + // Update rate limiter from response headers + if let Some(resp_headers) = data.get("responseHeaders") { + let rate_info = rate_limiter::RateLimitInfo::from_headers(resp_headers); + if rate_info.has_data() { + self.rate_limiter().update(rate_info); + } + } + + // Check for API errors in response body + self.check_response(data) + } } #[cfg(test)] diff --git a/ccxt-exchanges/src/binance/rate_limiter.rs b/ccxt-exchanges/src/binance/rate_limiter.rs index 48fc8ce..f623c9b 100644 --- a/ccxt-exchanges/src/binance/rate_limiter.rs +++ b/ccxt-exchanges/src/binance/rate_limiter.rs @@ -82,10 +82,15 @@ impl RateLimitInfo { let mut info = Self::default(); if let Some(obj) = headers.as_object() { - // Extract X-MBX-USED-WEIGHT-1M + // Extract X-MBX-USED-WEIGHT-1M (also fallback to x-mbx-used-weight without -1m suffix, + // and x-sapi-used-uid-weight-1m for SAPI endpoints) if let Some(weight) = obj .get("x-mbx-used-weight-1m") .or_else(|| obj.get("X-MBX-USED-WEIGHT-1M")) + .or_else(|| obj.get("x-mbx-used-weight")) + .or_else(|| obj.get("X-MBX-USED-WEIGHT")) + .or_else(|| obj.get("x-sapi-used-uid-weight-1m")) + .or_else(|| obj.get("X-SAPI-USED-UID-WEIGHT-1M")) .and_then(|v| v.as_str()) .and_then(|s| s.parse::().ok()) { diff --git a/ccxt-exchanges/src/binance/rest/futures/funding.rs b/ccxt-exchanges/src/binance/rest/futures/funding.rs index 5ab4446..59b719b 100644 --- a/ccxt-exchanges/src/binance/rest/futures/funding.rs +++ b/ccxt-exchanges/src/binance/rest/futures/funding.rs @@ -46,7 +46,7 @@ impl Binance { let _ = write!(request_url, "{}={}&", key, value); } - let response = self.base().http_client.get(&request_url, None).await?; + let response = self.public_get(&request_url, None).await?; let data = if market.linear.unwrap_or(true) { &response @@ -91,7 +91,7 @@ impl Binance { } } - let response = self.base().http_client.get(&request_url, None).await?; + let response = self.public_get(&request_url, None).await?; let mut rates = BTreeMap::new(); @@ -173,7 +173,7 @@ impl Binance { let _ = write!(request_url, "{}={}&", key, value); } - let response = self.base().http_client.get(&request_url, None).await?; + let response = self.public_get(&request_url, None).await?; let history_array = response.as_array().ok_or_else(|| { Error::from(ParseError::invalid_format( diff --git a/ccxt-exchanges/src/binance/rest/futures/margin.rs b/ccxt-exchanges/src/binance/rest/futures/margin.rs index 15e8ffc..d70fce4 100644 --- a/ccxt-exchanges/src/binance/rest/futures/margin.rs +++ b/ccxt-exchanges/src/binance/rest/futures/margin.rs @@ -2,7 +2,6 @@ use super::super::super::{Binance, signed_request::HttpMethod}; use ccxt_core::{Error, ParseError, Result, types::MarketType}; -use reqwest::header::HeaderMap; use rust_decimal::Decimal; use serde_json::Value; use std::collections::{BTreeMap, HashMap}; @@ -70,118 +69,45 @@ impl Binance { /// Set position mode (hedge mode or one-way mode). pub async fn set_position_mode(&self, dual_side: bool, params: Option) -> Result { - self.check_required_credentials()?; - let use_dapi = params .as_ref() .and_then(|p| p.get("type")) .and_then(serde_json::Value::as_str) .is_some_and(|t| t == "delivery" || t == "coin_m"); - let mut request_params = BTreeMap::new(); - request_params.insert("dualSidePosition".to_string(), dual_side.to_string()); - - if let Some(params) = params { - if let Some(obj) = params.as_object() { - for (key, value) in obj { - if key == "type" { - continue; - } - if let Some(v) = value.as_str() { - request_params.insert(key.clone(), v.to_string()); - } else if let Some(v) = value.as_bool() { - request_params.insert(key.clone(), v.to_string()); - } else if let Some(v) = value.as_i64() { - request_params.insert(key.clone(), v.to_string()); - } - } - } - } - - let timestamp = self.get_signing_timestamp().await?; - let auth = self.get_auth()?; - let signed_params = - auth.sign_with_timestamp(&request_params, timestamp, Some(self.options().recv_window))?; - let url = if use_dapi { format!("{}/positionSide/dual", self.urls().dapi_private) } else { format!("{}/positionSide/dual", self.urls().fapi_private) }; - let mut headers = HeaderMap::new(); - auth.add_auth_headers_reqwest(&mut headers); - - let body = serde_json::to_value(&signed_params).map_err(|e| { - Error::from(ParseError::invalid_format( - "data", - format!("Failed to serialize params: {}", e), - )) - })?; - - let data = self - .base() - .http_client - .post(&url, Some(headers), Some(body)) - .await?; + let builder = self + .signed_request(url) + .method(HttpMethod::Post) + .param("dualSidePosition", dual_side) + .merge_json_params(params); + let data = builder.execute().await?; Ok(data) } /// Fetch current position mode. pub async fn fetch_position_mode(&self, params: Option) -> Result { - self.check_required_credentials()?; - let use_dapi = params .as_ref() .and_then(|p| p.get("type")) .and_then(serde_json::Value::as_str) .is_some_and(|t| t == "delivery" || t == "coin_m"); - let mut request_params = BTreeMap::new(); - - if let Some(params) = params { - if let Some(obj) = params.as_object() { - for (key, value) in obj { - if key == "type" { - continue; - } - if let Some(v) = value.as_str() { - request_params.insert(key.clone(), v.to_string()); - } - } - } - } - - let timestamp = self.get_signing_timestamp().await?; - let auth = self.get_auth()?; - let signed_params = - auth.sign_with_timestamp(&request_params, timestamp, Some(self.options().recv_window))?; - - let query_string = signed_params - .iter() - .map(|(k, v)| format!("{}={}", k, v)) - .collect::>() - .join("&"); - let url = if use_dapi { - format!( - "{}/positionSide/dual?{}", - self.urls().dapi_private, - query_string - ) + format!("{}/positionSide/dual", self.urls().dapi_private) } else { - format!( - "{}/positionSide/dual?{}", - self.urls().fapi_private, - query_string - ) + format!("{}/positionSide/dual", self.urls().fapi_private) }; - let mut headers = HeaderMap::new(); - auth.add_auth_headers_reqwest(&mut headers); + let builder = self.signed_request(url).merge_json_params(params); - let data = self.base().http_client.get(&url, Some(headers)).await?; + let data = builder.execute().await?; if let Some(dual_side) = data.get("dualSidePosition") { if let Some(value) = dual_side.as_bool() { @@ -205,35 +131,7 @@ impl Binance { amount: Decimal, params: Option, ) -> Result { - self.check_required_credentials()?; - let market = self.base().market(symbol).await?; - let mut request_params = BTreeMap::new(); - request_params.insert("symbol".to_string(), market.id.clone()); - request_params.insert("amount".to_string(), amount.abs().to_string()); - request_params.insert( - "type".to_string(), - if amount > Decimal::ZERO { - "1".to_string() - } else { - "2".to_string() - }, - ); - - if let Some(params) = params { - if let Some(obj) = params.as_object() { - for (key, value) in obj { - if let Some(v) = value.as_str() { - request_params.insert(key.clone(), v.to_string()); - } - } - } - } - - let timestamp = self.get_signing_timestamp().await?; - let auth = self.get_auth()?; - let signed_params = - auth.sign_with_timestamp(&request_params, timestamp, Some(self.options().recv_window))?; let url = if market.linear.unwrap_or(true) { format!("{}/positionMargin", self.urls().fapi_private) @@ -241,22 +139,17 @@ impl Binance { format!("{}/positionMargin", self.urls().dapi_private) }; - let mut headers = HeaderMap::new(); - auth.add_auth_headers_reqwest(&mut headers); + let margin_type = if amount > Decimal::ZERO { "1" } else { "2" }; - let body = serde_json::to_value(&signed_params).map_err(|e| { - Error::from(ParseError::invalid_format( - "data", - format!("Failed to serialize params: {}", e), - )) - })?; - - let data = self - .base() - .http_client - .post(&url, Some(headers), Some(body)) - .await?; + let builder = self + .signed_request(url) + .method(HttpMethod::Post) + .param("symbol", &market.id) + .param("amount", amount.abs()) + .param("type", margin_type) + .merge_json_params(params); + let data = builder.execute().await?; Ok(data) } } diff --git a/ccxt-exchanges/src/binance/rest/market_data.rs b/ccxt-exchanges/src/binance/rest/market_data.rs index 8c9d2cd..2d8f07d 100644 --- a/ccxt-exchanges/src/binance/rest/market_data.rs +++ b/ccxt-exchanges/src/binance/rest/market_data.rs @@ -45,7 +45,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.get_rest_url_public(), endpoints::TIME); - let response = self.base().http_client.get(&url, None).await?; + let response = self.public_get(&url, None).await?; response["serverTime"] .as_i64() @@ -69,7 +69,7 @@ impl Binance { pub async fn fetch_status(&self) -> Result { // 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?; + let response = self.public_get(&url, None).await?; // Response format: { "status": 0, "msg": "normal" } // Status codes: 0 = normal, 1 = system maintenance @@ -124,7 +124,7 @@ impl Binance { &self, ) -> Result>>> { let url = format!("{}{}", self.get_rest_url_public(), endpoints::EXCHANGE_INFO); - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; let symbols = data["symbols"] .as_array() @@ -264,7 +264,7 @@ impl Binance { } } - let data = self.base().http_client.get(url.as_str(), None).await?; + let data = self.public_get(url.as_str(), None).await?; parser::parse_ticker(&data, Some(&market)) } @@ -300,7 +300,7 @@ impl Binance { drop(cache); let url = format!("{}{}", self.get_rest_url_public(), endpoints::TICKER_24HR); - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; let tickers_array = data.as_array().ok_or_else(|| { Error::from(ParseError::invalid_format( @@ -387,7 +387,7 @@ impl Binance { } } - let data = self.base().http_client.get(url.as_str(), None).await?; + let data = self.public_get(url.as_str(), None).await?; parser::parse_orderbook(&data, market.symbol.clone()) } @@ -426,7 +426,7 @@ impl Binance { ) }; - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; let trades_array = data.as_array().ok_or_else(|| { Error::from(ParseError::invalid_format( @@ -528,7 +528,7 @@ impl Binance { } } - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; let agg_trades_array = data.as_array().ok_or_else(|| { Error::from(ParseError::invalid_format( @@ -604,7 +604,7 @@ impl Binance { let auth = self.get_auth()?; auth.add_auth_headers_reqwest(&mut headers); - let data = self.base().http_client.get(&url, Some(headers)).await?; + let data = self.public_get(&url, Some(headers)).await?; let trades_array = data.as_array().ok_or_else(|| { Error::from(ParseError::invalid_format( @@ -652,7 +652,7 @@ impl Binance { format!("{}{}", self.get_rest_url_public(), endpoints::TICKER_24HR) }; - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; // Single symbol returns object, all symbols return array let stats_vec = if data.is_array() { @@ -702,7 +702,7 @@ impl Binance { endpoints::EXCHANGE_INFO, market.id ); - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; let symbols_array = data["symbols"].as_array().ok_or_else(|| { Error::from(ParseError::invalid_format("data", "Expected symbols array")) @@ -958,7 +958,7 @@ impl Binance { let _ = write!(url, "&endTime={}", end_time); } - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; parser::parse_ohlcvs(&data) } @@ -1054,7 +1054,7 @@ impl Binance { let _ = write!(url, "&endTime={}", end_time); } - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; parser::parse_ohlcvs(&data) } @@ -1142,7 +1142,7 @@ impl Binance { format!("{}/ticker/bookTicker", self.get_rest_url_public()) }; - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; parser::parse_bids_asks(&data) } @@ -1201,7 +1201,7 @@ impl Binance { format!("{}/ticker/price", self.get_rest_url_public()) }; - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; parser::parse_last_prices(&data) } @@ -1271,7 +1271,7 @@ impl Binance { ) }; - let data = self.base().http_client.get(&url, None).await?; + let data = self.public_get(&url, None).await?; parser::parse_mark_prices(&data) } diff --git a/ccxt-exchanges/src/binance/signed_request.rs b/ccxt-exchanges/src/binance/signed_request.rs index 77b0838..2af8329 100644 --- a/ccxt-exchanges/src/binance/signed_request.rs +++ b/ccxt-exchanges/src/binance/signed_request.rs @@ -296,6 +296,7 @@ impl<'a> SignedRequestBuilder<'a> { /// 3. Signs the parameters with HMAC-SHA256 /// 4. Adds authentication headers /// 5. Executes the HTTP request + /// 6. If a timestamp error is detected, resyncs time and retries once /// /// # Returns /// @@ -330,10 +331,28 @@ impl<'a> SignedRequestBuilder<'a> { // Step 2: Validate credentials self.binance.check_required_credentials()?; - // Step 3: Get signing timestamp + // Step 3: Execute the request + match self.execute_inner().await { + Ok(result) => Ok(result), + Err(err) => { + // Step 4: If timestamp error, resync time and retry once + if self.binance.is_timestamp_error(&err) { + tracing::warn!("Timestamp error detected, resyncing time and retrying"); + if self.binance.sync_time().await.is_ok() { + return self.execute_inner().await; + } + } + Err(err) + } + } + } + + /// Internal execution logic that can be called multiple times for retry. + async fn execute_inner(&self) -> Result { + // Get signing timestamp let timestamp = self.binance.get_signing_timestamp().await?; - // Step 4: Get auth and sign parameters + // Get auth and sign parameters let auth = self.binance.get_auth()?; let signed_params = auth.sign_with_timestamp( &self.params, @@ -341,16 +360,16 @@ impl<'a> SignedRequestBuilder<'a> { Some(self.binance.options().recv_window), )?; - // Step 5: Add authentication headers + // Add authentication headers let mut headers = HeaderMap::new(); auth.add_auth_headers_reqwest(&mut headers); - // Step 6: Execute HTTP request based on method + // Execute HTTP request based on method // Note: Binance API requires all signed requests to use query string parameters, // not JSON body, even for POST/PUT/DELETE requests. let query_string = build_query_string(&signed_params); let url = if query_string.is_empty() { - self.endpoint + self.endpoint.clone() } else { format!("{}?{}", self.endpoint, query_string) }; @@ -386,15 +405,15 @@ impl<'a> SignedRequestBuilder<'a> { } }?; - // Step 7: Update rate limiter from response headers - if let Some(headers) = result.get("responseHeaders") { - let rate_info = RateLimitInfo::from_headers(headers); + // Update rate limiter from response headers + if let Some(resp_headers) = result.get("responseHeaders") { + let rate_info = RateLimitInfo::from_headers(resp_headers); if rate_info.has_data() { self.binance.rate_limiter().update(rate_info); } } - // Step 8: Check for Binance API error in response body + // Check for Binance API error in response body // Binance may return HTTP 200 with an error in the JSON body: {"code": -1121, "msg": "..."} if let Some(api_error) = BinanceApiError::from_json(&result) { // Check for IP ban (HTTP 418) diff --git a/ccxt-exchanges/src/binance/ws/connection_manager.rs b/ccxt-exchanges/src/binance/ws/connection_manager.rs index 2c3c4e8..983d677 100644 --- a/ccxt-exchanges/src/binance/ws/connection_manager.rs +++ b/ccxt-exchanges/src/binance/ws/connection_manager.rs @@ -48,6 +48,17 @@ impl BinanceConnectionManager { /// Returns the WebSocket URL for the given market type. /// + /// # URL Routing + /// + /// - `MarketType::Spot` → `stream.binance.com` (spot and margin trading) + /// - `MarketType::Futures` → `fstream.binance.com` (Linear/USDT-margined contracts) + /// - `MarketType::Swap` → `dstream.binance.com` (Inverse/Coin-margined contracts) + /// - `MarketType::Option` → `nbstream.binance.com` (European-style options) + /// + /// Callers should select the `MarketType` based on the market's `linear`/`inverse` flags: + /// - `market.linear == true` → use `MarketType::Futures` + /// - `market.inverse == true` → use `MarketType::Swap` + /// /// # Arguments /// /// * `market_type` - The market type to get the URL for diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index 4a602e5..1c57093 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/ccxt-exchanges/src/binance/ws/mod.rs @@ -146,14 +146,6 @@ impl BinanceWs { let subscription_manager = Arc::new(SubscriptionManager::new()); let message_router = Arc::new(MessageRouter::new(url, subscription_manager.clone(), None)); - // 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 { message_router, subscription_manager, @@ -203,14 +195,6 @@ impl BinanceWs { None, )); - // 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 { message_router, subscription_manager, @@ -248,14 +232,6 @@ impl BinanceWs { Some(listen_key_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 { message_router, subscription_manager, diff --git a/ccxt-exchanges/src/binance/ws/subscriptions.rs b/ccxt-exchanges/src/binance/ws/subscriptions.rs index 5ad4c37..7cca675 100644 --- a/ccxt-exchanges/src/binance/ws/subscriptions.rs +++ b/ccxt-exchanges/src/binance/ws/subscriptions.rs @@ -69,8 +69,8 @@ pub struct Subscription { pub sub_type: SubscriptionType, /// Timestamp when the subscription was created pub subscribed_at: Instant, - /// Sender for forwarding WebSocket messages to consumers - pub sender: tokio::sync::mpsc::Sender, + /// Senders for forwarding WebSocket messages to consumers (supports multiple subscribers) + senders: Arc>>>, /// Reference count for this subscription (how many handles are active) ref_count: Arc, } @@ -88,15 +88,44 @@ impl Subscription { symbol, sub_type, subscribed_at: Instant::now(), - sender, + senders: Arc::new(std::sync::Mutex::new(vec![sender])), ref_count: Arc::new(AtomicUsize::new(1)), } } - /// Sends a message to the subscriber + /// Adds a new sender to this subscription for multi-subscriber support. + pub fn add_sender(&self, sender: tokio::sync::mpsc::Sender) { + if let Ok(mut senders) = self.senders.lock() { + senders.push(sender); + } + } + + /// Sends a message to all subscribers, removing closed senders. + /// + /// Returns `true` if at least one sender successfully received the message. pub fn send(&self, message: Value) -> bool { - // Use try_send to avoid blocking if channel is full (drop strategy) - self.sender.try_send(message).is_ok() + if let Ok(mut senders) = self.senders.lock() { + let mut any_sent = false; + senders.retain(|sender| { + match sender.try_send(message.clone()) { + Ok(()) => { + any_sent = true; + true // keep this sender + } + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + // Channel full - keep sender but count as not sent for backpressure + true + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + // Receiver dropped - remove this sender + false + } + } + }); + any_sent || !senders.is_empty() + } else { + false + } } /// Increments the reference count and returns the new value @@ -149,8 +178,9 @@ impl SubscriptionManager { ) -> ccxt_core::error::Result<()> { let mut subs = self.subscriptions.write().await; - // Check if subscription already exists - increment ref count + // Check if subscription already exists - add sender and increment ref count if let Some(existing) = subs.get(&stream) { + existing.add_sender(sender); existing.add_ref(); return Ok(()); }