From 8758577c369d114b62d268a542eca9cfdcb1a43e Mon Sep 17 00:00:00 2001 From: quran Date: Sat, 7 Feb 2026 02:06:16 +0800 Subject: [PATCH] feat(binance): implement market type aware websocket connection management - Add BinanceEndpointRouter import and EndpointType usage for REST endpoints - Implement market-type-aware endpoint routing for myTrades API calls - Create comprehensive listen key management functions with market type routing - Add specialized functions for linear (USDT-M) and inverse (COIN-M) futures - Enhance connection manager to handle multiple market types separately - Add URL encoding for special characters in query parameters and signatures - Update WebSocket connection methods to use market type specific endpoints - Implement proper URL routing for spot, futures, swap, and options markets - Add comprehensive test coverage for URL encoding and connection management --- ccxt-exchanges/src/binance/auth.rs | 112 +++- ccxt-exchanges/src/binance/error.rs | 325 +++++++++- ccxt-exchanges/src/binance/mod.rs | 208 +++++- ccxt-exchanges/src/binance/rate_limiter.rs | 556 +++++++++++++++++ ccxt-exchanges/src/binance/rest/account.rs | 368 ++++++++++- .../src/binance/rest/market_data.rs | 11 + ccxt-exchanges/src/binance/rest/spot.rs | 46 +- ccxt-exchanges/src/binance/signed_request.rs | 95 +-- .../src/binance/signing_strategy.rs | 6 +- ccxt-exchanges/src/binance/ws/binance_impl.rs | 47 +- .../src/binance/ws/connection_manager.rs | 406 ++++++++++-- ccxt-exchanges/src/binance/ws/listen_key.rs | 83 ++- ccxt-exchanges/src/binance/ws/mod.rs | 25 +- .../src/binance/ws/subscriptions.rs | 154 ++++- ccxt-exchanges/src/binance/ws/user_data.rs | 590 ++++++++++++++++++ .../src/binance/ws_exchange_impl.rs | 163 +++-- 16 files changed, 2954 insertions(+), 241 deletions(-) create mode 100644 ccxt-exchanges/src/binance/rate_limiter.rs diff --git a/ccxt-exchanges/src/binance/auth.rs b/ccxt-exchanges/src/binance/auth.rs index 0693dde..13b8bf8 100644 --- a/ccxt-exchanges/src/binance/auth.rs +++ b/ccxt-exchanges/src/binance/auth.rs @@ -140,6 +140,7 @@ impl BinanceAuth { /// # Returns /// /// Returns a URL-encoded query string with parameters sorted by key. + /// Both keys and values are URL-encoded to handle special characters. #[allow(clippy::unused_self)] pub(crate) fn build_query_string(&self, params: &BTreeMap) -> String { let mut pairs: Vec<_> = params.iter().collect(); @@ -147,7 +148,7 @@ impl BinanceAuth { pairs .iter() - .map(|(k, v)| format!("{k}={v}")) + .map(|(k, v)| format!("{}={}", urlencoding::encode(k), urlencoding::encode(v))) .collect::>() .join("&") } @@ -188,7 +189,7 @@ impl BinanceAuth { /// /// # Returns /// -/// Returns the complete URL with query parameters. +/// Returns the complete URL with URL-encoded query parameters. pub fn build_signed_url( base_url: &str, endpoint: &str, @@ -196,7 +197,7 @@ pub fn build_signed_url( ) -> String { let query_string = params .iter() - .map(|(k, v)| format!("{k}={v}")) + .map(|(k, v)| format!("{}={}", urlencoding::encode(k), urlencoding::encode(v))) .collect::>() .join("&"); @@ -310,4 +311,109 @@ mod tests { assert_eq!(url, "https://api.binance.com/api/v3/time"); } + + #[test] + fn test_build_query_string_url_encoding() { + let auth = BinanceAuth::new("test_key", "test_secret"); + let mut params = BTreeMap::new(); + // Test special characters that need URL encoding + params.insert("clientOrderId".to_string(), "order+123&test".to_string()); + params.insert("symbol".to_string(), "BTC/USDT".to_string()); + + let query = auth.build_query_string(¶ms); + + // Verify special characters are URL encoded + assert!(query.contains("clientOrderId=order%2B123%26test")); + assert!(query.contains("symbol=BTC%2FUSDT")); + // Verify & is used as separator (not encoded) + assert!(query.contains("&")); + } + + #[test] + fn test_build_query_string_space_encoding() { + let auth = BinanceAuth::new("test_key", "test_secret"); + let mut params = BTreeMap::new(); + params.insert("note".to_string(), "hello world".to_string()); + + let query = auth.build_query_string(¶ms); + + // Space should be encoded as %20 + assert!(query.contains("note=hello%20world")); + } + + #[test] + fn test_build_query_string_equals_encoding() { + let auth = BinanceAuth::new("test_key", "test_secret"); + let mut params = BTreeMap::new(); + params.insert("filter".to_string(), "price=100".to_string()); + + let query = auth.build_query_string(¶ms); + + // = in value should be encoded as %3D + assert!(query.contains("filter=price%3D100")); + } + + #[test] + fn test_signature_consistency_with_encoding() { + let auth = BinanceAuth::new("test_key", "test_secret"); + let mut params = BTreeMap::new(); + params.insert("symbol".to_string(), "BTC/USDT".to_string()); + params.insert("side".to_string(), "BUY".to_string()); + + // Sign the same params twice + let signed1 = auth.sign_params(¶ms).expect("Sign failed"); + let signed2 = auth.sign_params(¶ms).expect("Sign failed"); + + // Signatures should be identical for same input + assert_eq!( + signed1.get("signature").expect("Missing sig1"), + signed2.get("signature").expect("Missing sig2") + ); + } + + #[test] + fn test_signature_changes_with_different_params() { + let auth = BinanceAuth::new("test_key", "test_secret"); + + let mut params1 = BTreeMap::new(); + params1.insert("symbol".to_string(), "BTCUSDT".to_string()); + + let mut params2 = BTreeMap::new(); + params2.insert("symbol".to_string(), "ETHUSDT".to_string()); + + let signed1 = auth.sign_params(¶ms1).expect("Sign failed"); + let signed2 = auth.sign_params(¶ms2).expect("Sign failed"); + + // Signatures should be different for different input + assert_ne!( + signed1.get("signature").expect("Missing sig1"), + signed2.get("signature").expect("Missing sig2") + ); + } + + #[test] + fn test_build_signed_url_with_special_chars() { + let mut params = HashMap::new(); + params.insert("clientOrderId".to_string(), "test+order".to_string()); + params.insert("signature".to_string(), "abc123".to_string()); + + let url = build_signed_url("https://api.binance.com", "/api/v3/order", ¶ms); + + // Special chars in values should be URL encoded + assert!(url.contains("clientOrderId=test%2Border")); + } + + #[test] + fn test_alphabetical_sorting() { + let auth = BinanceAuth::new("test_key", "test_secret"); + let mut params = BTreeMap::new(); + params.insert("zebra".to_string(), "z".to_string()); + params.insert("apple".to_string(), "a".to_string()); + params.insert("mango".to_string(), "m".to_string()); + + let query = auth.build_query_string(¶ms); + + // BTreeMap ensures alphabetical order + assert_eq!(query, "apple=a&mango=m&zebra=z"); + } } diff --git a/ccxt-exchanges/src/binance/error.rs b/ccxt-exchanges/src/binance/error.rs index bbde0e1..114b254 100644 --- a/ccxt-exchanges/src/binance/error.rs +++ b/ccxt-exchanges/src/binance/error.rs @@ -1,27 +1,27 @@ //! Binance-specific error types. //! -//! This module provides module-specific error types for Binance WebSocket operations. +//! This module provides module-specific error types for Binance operations. //! These errors can be converted to the core `Error` type while preserving //! structured information for downcast capability. //! +//! # Error Types +//! +//! - [`BinanceApiError`]: REST API errors with Binance error codes +//! - [`BinanceWsError`]: WebSocket-specific errors +//! //! # Example //! //! ```rust -//! use ccxt_exchanges::binance::error::BinanceWsError; +//! use ccxt_exchanges::binance::error::{BinanceApiError, BinanceWsError}; //! use ccxt_core::error::Error as CoreError; //! -//! // Create a Binance-specific error -//! let ws_err = BinanceWsError::subscription_failed("btcusdt@ticker", "Invalid stream"); +//! // Create a Binance API error +//! let api_err = BinanceApiError::new(-1121, "Invalid symbol."); +//! let core_err: CoreError = api_err.into(); //! -//! // Convert to core error for public API +//! // Create a Binance WebSocket error +//! let ws_err = BinanceWsError::subscription_failed("btcusdt@ticker", "Invalid stream"); //! let core_err: CoreError = ws_err.into(); -//! -//! // Downcast back to access structured data -//! if let Some(binance_err) = core_err.downcast_websocket::() { -//! if let Some(stream) = binance_err.stream_name() { -//! println!("Failed stream: {}", stream); -//! } -//! } //! ``` use ccxt_core::error::Error as CoreError; @@ -30,6 +30,155 @@ use thiserror::Error; #[cfg(test)] use std::error::Error as StdError; +/// Binance REST API error. +/// +/// Represents errors returned by the Binance REST API in the format: +/// ```json +/// {"code": -1121, "msg": "Invalid symbol."} +/// ``` +/// +/// This error type preserves the original Binance error code and message, +/// and can be converted to the appropriate `CoreError` variant based on +/// the error code. +#[derive(Error, Debug, Clone)] +#[error("Binance API error {code}: {msg}")] +pub struct BinanceApiError { + /// Binance error code (negative integer) + pub code: i32, + /// Error message from Binance + pub msg: String, +} + +impl BinanceApiError { + /// Creates a new Binance API error. + /// + /// # Arguments + /// + /// * `code` - The Binance error code (typically negative) + /// * `msg` - The error message from Binance + /// + /// # Example + /// + /// ```rust + /// use ccxt_exchanges::binance::error::BinanceApiError; + /// + /// let err = BinanceApiError::new(-1121, "Invalid symbol."); + /// assert_eq!(err.code, -1121); + /// ``` + pub fn new(code: i32, msg: impl Into) -> Self { + Self { + code, + msg: msg.into(), + } + } + + /// Parses a Binance API error from JSON response. + /// + /// # Arguments + /// + /// * `json` - The JSON response from Binance API + /// + /// # Returns + /// + /// Returns `Some(BinanceApiError)` if the JSON contains `code` and `msg` fields, + /// `None` otherwise. + /// + /// # Example + /// + /// ```rust + /// use ccxt_exchanges::binance::error::BinanceApiError; + /// use serde_json::json; + /// + /// let response = json!({"code": -1121, "msg": "Invalid symbol."}); + /// let err = BinanceApiError::from_json(&response); + /// assert!(err.is_some()); + /// assert_eq!(err.unwrap().code, -1121); + /// ``` + pub fn from_json(json: &serde_json::Value) -> Option { + let code = json.get("code")?.as_i64()? as i32; + let msg = json.get("msg")?.as_str()?.to_string(); + Some(Self { code, msg }) + } + + /// Returns true if this is a rate limit error. + /// + /// Rate limit errors have codes: + /// - -1003: Too many requests + /// - -1015: Too many orders + pub fn is_rate_limit(&self) -> bool { + matches!(self.code, -1003 | -1015) + } + + /// Returns true if this is an IP ban error. + /// + /// IP ban errors have code -1003 with specific messages or HTTP 418 status. + pub fn is_ip_banned(&self) -> bool { + self.code == -1003 && self.msg.contains("banned") + } + + /// Returns true if this is an authentication error. + pub fn is_auth_error(&self) -> bool { + matches!(self.code, -2014 | -2015 | -1022) + } + + /// Returns true if this is an invalid symbol error. + pub fn is_invalid_symbol(&self) -> bool { + self.code == -1121 + } + + /// Returns true if this is an insufficient balance error. + pub fn is_insufficient_balance(&self) -> bool { + matches!(self.code, -2010 | -2011) + } + + /// Returns true if this is an order not found error. + pub fn is_order_not_found(&self) -> bool { + self.code == -2013 + } + + /// Converts this error to the appropriate `CoreError` variant. + /// + /// The mapping follows these rules: + /// - Authentication errors (-2014, -2015, -1022) → `CoreError::Authentication` + /// - Rate limit errors (-1003, -1015) → `CoreError::RateLimit` + /// - Invalid symbol (-1121) → `CoreError::BadSymbol` + /// - Insufficient balance (-2010, -2011) → `CoreError::InsufficientBalance` + /// - Order not found (-2013) → `CoreError::OrderNotFound` + /// - Invalid order errors → `CoreError::InvalidOrder` + /// - Network errors (-1001) → `CoreError::Network` + /// - Other errors → `CoreError::Exchange` + pub fn to_core_error(&self) -> CoreError { + use std::borrow::Cow; + + match self.code { + // Authentication errors + -2014 | -2015 | -1022 => CoreError::authentication(self.msg.clone()), + // Rate limit errors + -1003 | -1015 => CoreError::rate_limit(self.msg.clone(), None), + // Invalid symbol + -1121 => CoreError::bad_symbol(&self.msg), + // Insufficient balance + -2010 | -2011 => CoreError::insufficient_balance(self.msg.clone()), + // Order not found + -2013 => CoreError::OrderNotFound(Cow::Owned(self.msg.clone())), + // Invalid order (various) + -1102 | -1106 | -1111 | -1112 | -1114 | -1115 | -1116 | -1117 | -1118 => { + CoreError::InvalidOrder(Cow::Owned(self.msg.clone())) + } + // Network/timeout errors + -1001 => CoreError::network(&self.msg), + // All other errors + _ => CoreError::exchange(self.code.to_string(), &self.msg), + } + } +} + +impl From for CoreError { + fn from(e: BinanceApiError) -> Self { + e.to_core_error() + } +} + /// Binance WebSocket specific errors. /// /// Implements `StdError + Send + Sync` for use with `CoreError::WebSocket`. @@ -337,11 +486,163 @@ mod tests { fn test_error_is_send_sync() { fn assert_send_sync() {} assert_send_sync::(); + assert_send_sync::(); } #[test] fn test_error_is_std_error() { fn assert_std_error() {} assert_std_error::(); + assert_std_error::(); + } + + // ==================== BinanceApiError Tests ==================== + + #[test] + fn test_binance_api_error_new() { + let err = BinanceApiError::new(-1121, "Invalid symbol."); + assert_eq!(err.code, -1121); + assert_eq!(err.msg, "Invalid symbol."); + assert!(err.to_string().contains("-1121")); + assert!(err.to_string().contains("Invalid symbol.")); + } + + #[test] + fn test_binance_api_error_from_json() { + let json = serde_json::json!({"code": -1121, "msg": "Invalid symbol."}); + let err = BinanceApiError::from_json(&json); + assert!(err.is_some()); + let err = err.unwrap(); + assert_eq!(err.code, -1121); + assert_eq!(err.msg, "Invalid symbol."); + } + + #[test] + fn test_binance_api_error_from_json_missing_fields() { + // Missing code + let json = serde_json::json!({"msg": "Invalid symbol."}); + assert!(BinanceApiError::from_json(&json).is_none()); + + // Missing msg + let json = serde_json::json!({"code": -1121}); + assert!(BinanceApiError::from_json(&json).is_none()); + + // Empty object + let json = serde_json::json!({}); + assert!(BinanceApiError::from_json(&json).is_none()); + } + + #[test] + fn test_binance_api_error_is_rate_limit() { + let err = BinanceApiError::new(-1003, "Too many requests"); + assert!(err.is_rate_limit()); + + let err = BinanceApiError::new(-1015, "Too many orders"); + assert!(err.is_rate_limit()); + + let err = BinanceApiError::new(-1121, "Invalid symbol"); + assert!(!err.is_rate_limit()); + } + + #[test] + fn test_binance_api_error_is_auth_error() { + let err = BinanceApiError::new(-2014, "API-key format invalid"); + assert!(err.is_auth_error()); + + let err = BinanceApiError::new(-2015, "Invalid API-key, IP, or permissions"); + assert!(err.is_auth_error()); + + let err = BinanceApiError::new(-1022, "Signature for this request is not valid"); + assert!(err.is_auth_error()); + + let err = BinanceApiError::new(-1121, "Invalid symbol"); + assert!(!err.is_auth_error()); + } + + #[test] + fn test_binance_api_error_is_invalid_symbol() { + let err = BinanceApiError::new(-1121, "Invalid symbol."); + assert!(err.is_invalid_symbol()); + + let err = BinanceApiError::new(-1003, "Too many requests"); + assert!(!err.is_invalid_symbol()); + } + + #[test] + fn test_binance_api_error_is_insufficient_balance() { + let err = BinanceApiError::new(-2010, "Account has insufficient balance"); + assert!(err.is_insufficient_balance()); + + let err = BinanceApiError::new(-2011, "Unknown order sent"); + assert!(err.is_insufficient_balance()); + + let err = BinanceApiError::new(-1121, "Invalid symbol"); + assert!(!err.is_insufficient_balance()); + } + + #[test] + fn test_binance_api_error_is_order_not_found() { + let err = BinanceApiError::new(-2013, "Order does not exist"); + assert!(err.is_order_not_found()); + + let err = BinanceApiError::new(-1121, "Invalid symbol"); + assert!(!err.is_order_not_found()); + } + + #[test] + fn test_binance_api_error_to_core_error_auth() { + let err = BinanceApiError::new(-2015, "Invalid API-key"); + let core_err: CoreError = err.into(); + assert!(matches!(core_err, CoreError::Authentication(_))); + } + + #[test] + fn test_binance_api_error_to_core_error_rate_limit() { + let err = BinanceApiError::new(-1003, "Too many requests"); + let core_err: CoreError = err.into(); + assert!(matches!(core_err, CoreError::RateLimit { .. })); + } + + #[test] + fn test_binance_api_error_to_core_error_invalid_symbol() { + let err = BinanceApiError::new(-1121, "Invalid symbol."); + let core_err: CoreError = err.into(); + // bad_symbol returns InvalidRequest variant + assert!(matches!(core_err, CoreError::InvalidRequest(_))); + } + + #[test] + fn test_binance_api_error_to_core_error_insufficient_funds() { + let err = BinanceApiError::new(-2010, "Account has insufficient balance"); + let core_err: CoreError = err.into(); + assert!(matches!(core_err, CoreError::InsufficientBalance(_))); + } + + #[test] + fn test_binance_api_error_to_core_error_order_not_found() { + let err = BinanceApiError::new(-2013, "Order does not exist"); + let core_err: CoreError = err.into(); + assert!(matches!(core_err, CoreError::OrderNotFound(_))); + } + + #[test] + fn test_binance_api_error_to_core_error_invalid_order() { + let err = BinanceApiError::new(-1102, "Mandatory parameter was not sent"); + let core_err: CoreError = err.into(); + assert!(matches!(core_err, CoreError::InvalidOrder(_))); + } + + #[test] + fn test_binance_api_error_to_core_error_network() { + let err = BinanceApiError::new(-1001, "Internal error; unable to process your request"); + let core_err: CoreError = err.into(); + assert!(matches!(core_err, CoreError::Network(_))); + } + + #[test] + fn test_binance_api_error_to_core_error_exchange() { + let err = BinanceApiError::new(-9999, "Unknown error"); + let core_err: CoreError = err.into(); + assert!(matches!(core_err, CoreError::Exchange(_))); } } diff --git a/ccxt-exchanges/src/binance/mod.rs b/ccxt-exchanges/src/binance/mod.rs index e861bdb..8cdace9 100644 --- a/ccxt-exchanges/src/binance/mod.rs +++ b/ccxt-exchanges/src/binance/mod.rs @@ -3,6 +3,7 @@ //! Supports spot trading, futures trading, and options trading with complete REST API and WebSocket support. use ccxt_core::types::EndpointType; +use ccxt_core::types::MarketType; use ccxt_core::types::default_type::{DefaultSubType, DefaultType}; use ccxt_core::{BaseExchange, ExchangeConfig, Result}; use std::sync::Arc; @@ -17,6 +18,7 @@ pub mod error; mod exchange_impl; pub mod options; pub mod parser; +pub mod rate_limiter; pub mod rest; pub mod signed_request; pub mod signing_strategy; @@ -28,10 +30,14 @@ mod ws_exchange_impl; pub use builder::BinanceBuilder; pub use endpoint_router::BinanceEndpointRouter; +pub use error::BinanceApiError; pub use options::BinanceOptions; pub use signed_request::{HttpMethod, SignedRequestBuilder}; pub use time_sync::{TimeSyncConfig, TimeSyncManager}; pub use urls::BinanceUrls; + +use rate_limiter::WeightRateLimiter; + /// Binance exchange structure. #[derive(Debug, Clone)] pub struct Binance { @@ -46,6 +52,8 @@ pub struct Binance { ws_connection: Arc>>, /// Centralized WebSocket connection manager pub connection_manager: Arc, + /// Rate limiter for API requests. + rate_limiter: Arc, } #[allow(deprecated)] @@ -97,20 +105,29 @@ impl Binance { let options = BinanceOptions::default(); let time_sync = Arc::new(TimeSyncManager::new()); - // Determine WS URL for ConnectionManager - let urls = if base.config.sandbox { + // Determine URLs for ConnectionManager + let mut urls = if base.config.sandbox { BinanceUrls::testnet() } else { BinanceUrls::production() }; - // Apply overrides logic similar to self.urls() - let ws_url = if let Some(url) = base.config.url_overrides.get("ws") { - url.clone() - } else { - urls.ws.clone() - }; + // Apply URL overrides + if let Some(ws_url) = base.config.url_overrides.get("ws") { + urls.ws.clone_from(ws_url); + } + if let Some(ws_fapi_url) = base.config.url_overrides.get("wsFapi") { + urls.ws_fapi.clone_from(ws_fapi_url); + } + if let Some(ws_dapi_url) = base.config.url_overrides.get("wsDapi") { + urls.ws_dapi.clone_from(ws_dapi_url); + } + if let Some(ws_eapi_url) = base.config.url_overrides.get("wsEapi") { + urls.ws_eapi.clone_from(ws_eapi_url); + } - let connection_manager = Arc::new(ws::BinanceConnectionManager::new(ws_url)); + let connection_manager = + Arc::new(ws::BinanceConnectionManager::new(urls, base.config.sandbox)); + let rate_limiter = Arc::new(WeightRateLimiter::new()); Ok(Self { base, @@ -118,6 +135,7 @@ impl Binance { time_sync, ws_connection: Arc::new(RwLock::new(None)), connection_manager, + rate_limiter, }) } @@ -156,7 +174,7 @@ impl Binance { }; let time_sync = Arc::new(TimeSyncManager::with_config(time_sync_config)); - // Determine WS URL + // Determine WS URLs let mut urls = if base.config.sandbox { BinanceUrls::testnet() } else { @@ -177,16 +195,9 @@ impl Binance { urls.ws_eapi.clone_from(ws_eapi_url); } - let ws_url = match options.default_type { - DefaultType::Swap | DefaultType::Futures => match options.default_sub_type { - Some(DefaultSubType::Inverse) => urls.ws_dapi.clone(), - _ => urls.ws_fapi.clone(), - }, - DefaultType::Option => urls.ws_eapi.clone(), - _ => urls.ws.clone(), - }; - - let connection_manager = Arc::new(ws::BinanceConnectionManager::new(ws_url)); + let connection_manager = + Arc::new(ws::BinanceConnectionManager::new(urls, base.config.sandbox)); + let rate_limiter = Arc::new(WeightRateLimiter::new()); Ok(Self { base, @@ -194,6 +205,7 @@ impl Binance { time_sync, ws_connection: Arc::new(RwLock::new(None)), connection_manager, + rate_limiter, }) } @@ -227,7 +239,7 @@ impl Binance { }; let time_sync = Arc::new(TimeSyncManager::with_config(time_sync_config)); - // Determine WS URL (Futures/Swap) + // Determine WS URLs (Futures/Swap) let mut urls = if base.config.sandbox { BinanceUrls::testnet() } else { @@ -238,10 +250,9 @@ impl Binance { urls.ws_fapi.clone_from(ws_fapi_url); } - // Default to FAPI/Linear for new_swap if not configured otherwise - let ws_url = urls.ws_fapi.clone(); - - let connection_manager = Arc::new(ws::BinanceConnectionManager::new(ws_url)); + let connection_manager = + Arc::new(ws::BinanceConnectionManager::new(urls, base.config.sandbox)); + let rate_limiter = Arc::new(WeightRateLimiter::new()); Ok(Self { base, @@ -249,6 +260,7 @@ impl Binance { time_sync, ws_connection: Arc::new(RwLock::new(None)), connection_manager, + rate_limiter, }) } @@ -317,6 +329,25 @@ impl Binance { &self.time_sync } + /// Returns a reference to the rate limiter. + /// + /// The `WeightRateLimiter` tracks API usage based on response headers + /// and provides throttling recommendations to avoid hitting rate limits. + /// + /// # Example + /// + /// ```no_run + /// use ccxt_exchanges::binance::Binance; + /// use ccxt_core::ExchangeConfig; + /// + /// let binance = Binance::new(ExchangeConfig::default()).unwrap(); + /// let rate_limiter = binance.rate_limiter(); + /// println!("Current weight usage: {}%", rate_limiter.weight_usage_ratio() * 100.0); + /// ``` + pub fn rate_limiter(&self) -> &Arc { + &self.rate_limiter + } + /// Creates a new signed request builder for the given endpoint. /// /// This is the recommended way to make authenticated API requests. @@ -646,7 +677,45 @@ impl Binance { /// ``` pub fn create_authenticated_ws(self: &std::sync::Arc) -> ws::BinanceWs { let ws_url = self.get_ws_url(); - ws::BinanceWs::new_with_auth(ws_url, self.clone()) + let market_type = MarketType::from(self.options.default_type); + ws::BinanceWs::new_with_auth(ws_url, self.clone(), market_type) + } + + /// Checks a JSON response for Binance API errors and converts them to `Result`. + /// + /// Binance API may return errors in the response body even with HTTP 200 status. + /// The error format is: `{"code": -1121, "msg": "Invalid symbol."}` + /// + /// This method should be called after receiving a response from public API endpoints + /// to ensure proper error handling. + /// + /// # Arguments + /// + /// * `response` - The JSON response from Binance API + /// + /// # Returns + /// + /// Returns `Ok(response)` if no error is found, or `Err(CoreError)` if the response + /// contains a Binance API error. + /// + /// # Example + /// + /// ```no_run + /// use ccxt_exchanges::binance::Binance; + /// use ccxt_core::ExchangeConfig; + /// + /// # async fn example() -> ccxt_core::Result<()> { + /// let binance = Binance::new(ExchangeConfig::default())?; + /// let response = binance.base().http_client.get("https://api.binance.com/api/v3/ticker/price", None).await?; + /// let validated = binance.check_response(response)?; + /// # Ok(()) + /// # } + /// ``` + pub fn check_response(&self, response: serde_json::Value) -> Result { + if let Some(api_error) = BinanceApiError::from_json(&response) { + return Err(api_error.into()); + } + Ok(response) } } @@ -1438,4 +1507,91 @@ mod tests { urls.ws_eapi ); } + + // ============================================================ + // check_response Tests + // ============================================================ + + #[test] + fn test_check_response_success() { + let config = ExchangeConfig::default(); + let binance = Binance::new(config).unwrap(); + + // Valid response without error + let response = serde_json::json!({ + "symbol": "BTCUSDT", + "price": "50000.00" + }); + + let result = binance.check_response(response.clone()); + assert!(result.is_ok()); + assert_eq!(result.unwrap(), response); + } + + #[test] + fn test_check_response_with_binance_error() { + let config = ExchangeConfig::default(); + let binance = Binance::new(config).unwrap(); + + // Response with Binance API error + let response = serde_json::json!({ + "code": -1121, + "msg": "Invalid symbol." + }); + + let result = binance.check_response(response); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert!(err.to_string().contains("Invalid symbol")); + } + + #[test] + fn test_check_response_with_rate_limit_error() { + let config = ExchangeConfig::default(); + let binance = Binance::new(config).unwrap(); + + // Response with rate limit error + let response = serde_json::json!({ + "code": -1003, + "msg": "Too many requests; please use the websocket for live updates." + }); + + let result = binance.check_response(response); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert!(matches!(err, ccxt_core::error::Error::RateLimit { .. })); + } + + #[test] + fn test_check_response_with_auth_error() { + let config = ExchangeConfig::default(); + let binance = Binance::new(config).unwrap(); + + // Response with authentication error + let response = serde_json::json!({ + "code": -2015, + "msg": "Invalid API-key, IP, or permissions for action." + }); + + let result = binance.check_response(response); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert!(matches!(err, ccxt_core::error::Error::Authentication(_))); + } + + #[test] + fn test_check_response_array_response() { + let config = ExchangeConfig::default(); + let binance = Binance::new(config).unwrap(); + + // Array response (common for list endpoints) + let response = serde_json::json!([ + {"symbol": "BTCUSDT", "price": "50000.00"}, + {"symbol": "ETHUSDT", "price": "3000.00"} + ]); + + let result = binance.check_response(response.clone()); + assert!(result.is_ok()); + assert_eq!(result.unwrap(), response); + } } diff --git a/ccxt-exchanges/src/binance/rate_limiter.rs b/ccxt-exchanges/src/binance/rate_limiter.rs new file mode 100644 index 0000000..48fc8ce --- /dev/null +++ b/ccxt-exchanges/src/binance/rate_limiter.rs @@ -0,0 +1,556 @@ +//! Binance rate limiter implementation. +//! +//! This module provides weight-based rate limiting for Binance API requests. +//! Binance uses a weight system where different endpoints consume different +//! amounts of weight, and there are separate limits for different categories. +//! +//! # Rate Limit Categories +//! +//! - **Request Weight**: General API request limit (default: 1200/minute for spot) +//! - **Order Count**: Order placement limit (default: 10 orders/second, 100000/day) +//! - **Raw Requests**: Raw request count limit +//! +//! # Response Headers +//! +//! Binance returns rate limit information in response headers: +//! - `X-MBX-USED-WEIGHT-*`: Current used weight +//! - `X-MBX-ORDER-COUNT-*`: Current order count +//! - `Retry-After`: Seconds to wait when rate limited +//! +//! # Example +//! +//! ```rust +//! use ccxt_exchanges::binance::rate_limiter::{WeightRateLimiter, RateLimitInfo}; +//! +//! let limiter = WeightRateLimiter::new(); +//! +//! // Update from response headers +//! let info = RateLimitInfo { +//! used_weight_1m: Some(500), +//! order_count_10s: Some(5), +//! order_count_1d: Some(1000), +//! retry_after: None, +//! }; +//! limiter.update(info); +//! +//! // Check if we should throttle +//! if limiter.should_throttle() { +//! // Wait before making more requests +//! } +//! ``` + +use std::sync::RwLock; +use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +/// Default weight limit per minute for spot API. +pub const DEFAULT_WEIGHT_LIMIT_1M: u64 = 1200; + +/// Default order count limit per 10 seconds. +pub const DEFAULT_ORDER_LIMIT_10S: u64 = 100; + +/// Default order count limit per day. +pub const DEFAULT_ORDER_LIMIT_1D: u64 = 200000; + +/// Threshold percentage at which to start throttling (80%). +pub const THROTTLE_THRESHOLD: f64 = 0.80; + +/// Rate limit information extracted from response headers. +#[derive(Debug, Clone, Default)] +pub struct RateLimitInfo { + /// Used weight in the current 1-minute window. + pub used_weight_1m: Option, + /// Order count in the current 10-second window. + pub order_count_10s: Option, + /// Order count in the current day. + pub order_count_1d: Option, + /// Retry-After header value in seconds (when rate limited). + pub retry_after: Option, +} + +impl RateLimitInfo { + /// Creates a new `RateLimitInfo` from response headers. + /// + /// # Arguments + /// + /// * `headers` - JSON object containing response headers + /// + /// # Returns + /// + /// Returns a `RateLimitInfo` with values extracted from headers. + pub fn from_headers(headers: &serde_json::Value) -> Self { + let mut info = Self::default(); + + if let Some(obj) = headers.as_object() { + // Extract X-MBX-USED-WEIGHT-1M + if let Some(weight) = obj + .get("x-mbx-used-weight-1m") + .or_else(|| obj.get("X-MBX-USED-WEIGHT-1M")) + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + { + info.used_weight_1m = Some(weight); + } + + // Extract X-MBX-ORDER-COUNT-10S + if let Some(count) = obj + .get("x-mbx-order-count-10s") + .or_else(|| obj.get("X-MBX-ORDER-COUNT-10S")) + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + { + info.order_count_10s = Some(count); + } + + // Extract X-MBX-ORDER-COUNT-1D + if let Some(count) = obj + .get("x-mbx-order-count-1d") + .or_else(|| obj.get("X-MBX-ORDER-COUNT-1D")) + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + { + info.order_count_1d = Some(count); + } + + // Extract Retry-After + if let Some(retry) = obj + .get("retry-after") + .or_else(|| obj.get("Retry-After")) + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + { + info.retry_after = Some(retry); + } + } + + info + } + + /// Returns true if any rate limit information is present. + pub fn has_data(&self) -> bool { + self.used_weight_1m.is_some() + || self.order_count_10s.is_some() + || self.order_count_1d.is_some() + || self.retry_after.is_some() + } +} + +/// Weight-based rate limiter for Binance API. +/// +/// Tracks API usage based on response headers and provides throttling +/// recommendations to avoid hitting rate limits. +#[derive(Debug)] +pub struct WeightRateLimiter { + /// Current used weight (1-minute window). + used_weight_1m: AtomicU64, + /// Current order count (10-second window). + order_count_10s: AtomicU64, + /// Current order count (daily). + order_count_1d: AtomicU64, + /// Weight limit per minute. + weight_limit_1m: AtomicU64, + /// Order limit per 10 seconds. + order_limit_10s: AtomicU64, + /// Order limit per day. + order_limit_1d: AtomicU64, + /// Retry-After timestamp (when we can resume requests). + retry_after_until: RwLock>, + /// Last update timestamp. + last_update: RwLock>, + /// IP ban detection timestamp. + ip_banned_until: AtomicI64, +} + +impl Default for WeightRateLimiter { + fn default() -> Self { + Self::new() + } +} + +impl WeightRateLimiter { + /// Creates a new rate limiter with default limits. + pub fn new() -> Self { + Self { + used_weight_1m: AtomicU64::new(0), + order_count_10s: AtomicU64::new(0), + order_count_1d: AtomicU64::new(0), + weight_limit_1m: AtomicU64::new(DEFAULT_WEIGHT_LIMIT_1M), + order_limit_10s: AtomicU64::new(DEFAULT_ORDER_LIMIT_10S), + order_limit_1d: AtomicU64::new(DEFAULT_ORDER_LIMIT_1D), + retry_after_until: RwLock::new(None), + last_update: RwLock::new(None), + ip_banned_until: AtomicI64::new(0), + } + } + + /// Creates a new rate limiter with custom limits. + /// + /// # Arguments + /// + /// * `weight_limit_1m` - Weight limit per minute + /// * `order_limit_10s` - Order limit per 10 seconds + /// * `order_limit_1d` - Order limit per day + pub fn with_limits(weight_limit_1m: u64, order_limit_10s: u64, order_limit_1d: u64) -> Self { + Self { + used_weight_1m: AtomicU64::new(0), + order_count_10s: AtomicU64::new(0), + order_count_1d: AtomicU64::new(0), + weight_limit_1m: AtomicU64::new(weight_limit_1m), + order_limit_10s: AtomicU64::new(order_limit_10s), + order_limit_1d: AtomicU64::new(order_limit_1d), + retry_after_until: RwLock::new(None), + last_update: RwLock::new(None), + ip_banned_until: AtomicI64::new(0), + } + } + + /// Updates the rate limiter with information from response headers. + /// + /// # Arguments + /// + /// * `info` - Rate limit information extracted from headers + pub fn update(&self, info: RateLimitInfo) { + if let Some(weight) = info.used_weight_1m { + self.used_weight_1m.store(weight, Ordering::SeqCst); + } + + if let Some(count) = info.order_count_10s { + self.order_count_10s.store(count, Ordering::SeqCst); + } + + if let Some(count) = info.order_count_1d { + self.order_count_1d.store(count, Ordering::SeqCst); + } + + if let Some(retry_secs) = info.retry_after { + let until = Instant::now() + Duration::from_secs(retry_secs); + if let Ok(mut guard) = self.retry_after_until.write() { + *guard = Some(until); + } + } + + if let Ok(mut guard) = self.last_update.write() { + *guard = Some(Instant::now()); + } + } + + /// Sets the IP ban duration. + /// + /// # Arguments + /// + /// * `duration` - Duration of the IP ban + pub fn set_ip_banned(&self, duration: Duration) { + let until = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64 + + duration.as_secs() as i64; + self.ip_banned_until.store(until, Ordering::SeqCst); + } + + /// Returns true if the IP is currently banned. + pub fn is_ip_banned(&self) -> bool { + let banned_until = self.ip_banned_until.load(Ordering::SeqCst); + if banned_until == 0 { + return false; + } + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + + now < banned_until + } + + /// Returns true if we should throttle requests. + /// + /// Throttling is recommended when: + /// - We're in a Retry-After period + /// - Weight usage exceeds 80% of the limit + /// - Order count exceeds 80% of the limit + /// - IP is banned + pub fn should_throttle(&self) -> bool { + // Check IP ban + if self.is_ip_banned() { + return true; + } + + // Check Retry-After + if let Ok(guard) = self.retry_after_until.read() { + if let Some(until) = *guard { + if Instant::now() < until { + return true; + } + } + } + + // Check weight threshold (80%) + let weight = self.used_weight_1m.load(Ordering::SeqCst); + let weight_limit = self.weight_limit_1m.load(Ordering::SeqCst); + #[allow(clippy::cast_precision_loss)] + if (weight as f64) >= (weight_limit as f64) * THROTTLE_THRESHOLD { + return true; + } + + // Check order count threshold (80%) + let order_count = self.order_count_10s.load(Ordering::SeqCst); + let order_limit = self.order_limit_10s.load(Ordering::SeqCst); + if order_count as f64 >= order_limit as f64 * THROTTLE_THRESHOLD { + return true; + } + + false + } + + /// Returns the recommended wait duration before making the next request. + /// + /// Returns `None` if no waiting is needed. + pub fn wait_duration(&self) -> Option { + // Check IP ban first + if self.is_ip_banned() { + let banned_until = self.ip_banned_until.load(Ordering::SeqCst); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + if banned_until > now { + return Some(Duration::from_secs((banned_until - now) as u64)); + } + } + + // Check Retry-After + if let Ok(guard) = self.retry_after_until.read() { + if let Some(until) = *guard { + let now = Instant::now(); + if until > now { + return Some(until - now); + } + } + } + + // If we're at the threshold, suggest a small wait + if self.should_throttle() { + // Wait until the next minute window resets (conservative estimate) + return Some(Duration::from_secs(1)); + } + + None + } + + /// Returns the current used weight. + pub fn used_weight(&self) -> u64 { + self.used_weight_1m.load(Ordering::SeqCst) + } + + /// Returns the weight limit. + pub fn weight_limit(&self) -> u64 { + self.weight_limit_1m.load(Ordering::SeqCst) + } + + /// Returns the current order count (10s window). + pub fn order_count_10s(&self) -> u64 { + self.order_count_10s.load(Ordering::SeqCst) + } + + /// Returns the current order count (daily). + pub fn order_count_1d(&self) -> u64 { + self.order_count_1d.load(Ordering::SeqCst) + } + + /// Returns the order limit per day. + pub fn order_limit_1d(&self) -> u64 { + self.order_limit_1d.load(Ordering::SeqCst) + } + + /// Returns the weight usage percentage (0.0 to 1.0). + pub fn weight_usage_ratio(&self) -> f64 { + let weight = self.used_weight_1m.load(Ordering::SeqCst) as f64; + let limit = self.weight_limit_1m.load(Ordering::SeqCst) as f64; + if limit > 0.0 { weight / limit } else { 0.0 } + } + + /// Resets all counters. + /// + /// This should be called when the rate limit window resets. + pub fn reset(&self) { + self.used_weight_1m.store(0, Ordering::SeqCst); + self.order_count_10s.store(0, Ordering::SeqCst); + self.order_count_1d.store(0, Ordering::SeqCst); + if let Ok(mut guard) = self.retry_after_until.write() { + *guard = None; + } + self.ip_banned_until.store(0, Ordering::SeqCst); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_rate_limiter_new() { + let limiter = WeightRateLimiter::new(); + assert_eq!(limiter.used_weight(), 0); + assert_eq!(limiter.weight_limit(), DEFAULT_WEIGHT_LIMIT_1M); + assert!(!limiter.should_throttle()); + } + + #[test] + fn test_rate_limiter_with_limits() { + let limiter = WeightRateLimiter::with_limits(2400, 200, 400000); + assert_eq!(limiter.weight_limit(), 2400); + } + + #[test] + fn test_rate_limiter_update() { + let limiter = WeightRateLimiter::new(); + + let info = RateLimitInfo { + used_weight_1m: Some(500), + order_count_10s: Some(5), + order_count_1d: Some(1000), + retry_after: None, + }; + + limiter.update(info); + + assert_eq!(limiter.used_weight(), 500); + assert_eq!(limiter.order_count_10s(), 5); + assert_eq!(limiter.order_count_1d(), 1000); + } + + #[test] + fn test_rate_limiter_throttle_at_threshold() { + let limiter = WeightRateLimiter::new(); + + // Set weight to 80% of limit + let threshold_weight = (DEFAULT_WEIGHT_LIMIT_1M as f64 * THROTTLE_THRESHOLD) as u64; + let info = RateLimitInfo { + used_weight_1m: Some(threshold_weight), + ..Default::default() + }; + + limiter.update(info); + assert!(limiter.should_throttle()); + } + + #[test] + fn test_rate_limiter_no_throttle_below_threshold() { + let limiter = WeightRateLimiter::new(); + + // Set weight to 50% of limit + let info = RateLimitInfo { + used_weight_1m: Some(DEFAULT_WEIGHT_LIMIT_1M / 2), + ..Default::default() + }; + + limiter.update(info); + assert!(!limiter.should_throttle()); + } + + #[test] + fn test_rate_limiter_retry_after() { + let limiter = WeightRateLimiter::new(); + + let info = RateLimitInfo { + retry_after: Some(5), + ..Default::default() + }; + + limiter.update(info); + assert!(limiter.should_throttle()); + assert!(limiter.wait_duration().is_some()); + } + + #[test] + fn test_rate_limiter_ip_banned() { + let limiter = WeightRateLimiter::new(); + + limiter.set_ip_banned(Duration::from_secs(60)); + assert!(limiter.is_ip_banned()); + assert!(limiter.should_throttle()); + } + + #[test] + fn test_rate_limiter_reset() { + let limiter = WeightRateLimiter::new(); + + let info = RateLimitInfo { + used_weight_1m: Some(1000), + order_count_10s: Some(50), + order_count_1d: Some(5000), + retry_after: Some(10), + }; + + limiter.update(info); + limiter.set_ip_banned(Duration::from_secs(60)); + + limiter.reset(); + + assert_eq!(limiter.used_weight(), 0); + assert_eq!(limiter.order_count_10s(), 0); + assert_eq!(limiter.order_count_1d(), 0); + assert!(!limiter.is_ip_banned()); + assert!(!limiter.should_throttle()); + } + + #[test] + fn test_rate_limit_info_from_headers() { + let headers = serde_json::json!({ + "x-mbx-used-weight-1m": "500", + "x-mbx-order-count-10s": "5", + "x-mbx-order-count-1d": "1000", + "retry-after": "30" + }); + + let info = RateLimitInfo::from_headers(&headers); + + assert_eq!(info.used_weight_1m, Some(500)); + assert_eq!(info.order_count_10s, Some(5)); + assert_eq!(info.order_count_1d, Some(1000)); + assert_eq!(info.retry_after, Some(30)); + } + + #[test] + fn test_rate_limit_info_from_headers_uppercase() { + let headers = serde_json::json!({ + "X-MBX-USED-WEIGHT-1M": "600", + "X-MBX-ORDER-COUNT-10S": "10", + "Retry-After": "60" + }); + + let info = RateLimitInfo::from_headers(&headers); + + assert_eq!(info.used_weight_1m, Some(600)); + assert_eq!(info.order_count_10s, Some(10)); + assert_eq!(info.retry_after, Some(60)); + } + + #[test] + fn test_rate_limit_info_has_data() { + let empty = RateLimitInfo::default(); + assert!(!empty.has_data()); + + let with_weight = RateLimitInfo { + used_weight_1m: Some(100), + ..Default::default() + }; + assert!(with_weight.has_data()); + } + + #[test] + fn test_weight_usage_ratio() { + let limiter = WeightRateLimiter::new(); + + // 50% usage + let info = RateLimitInfo { + used_weight_1m: Some(DEFAULT_WEIGHT_LIMIT_1M / 2), + ..Default::default() + }; + limiter.update(info); + + let ratio = limiter.weight_usage_ratio(); + assert!((ratio - 0.5).abs() < 0.01); + } +} diff --git a/ccxt-exchanges/src/binance/rest/account.rs b/ccxt-exchanges/src/binance/rest/account.rs index af702dc..e77c092 100644 --- a/ccxt-exchanges/src/binance/rest/account.rs +++ b/ccxt-exchanges/src/binance/rest/account.rs @@ -3,11 +3,11 @@ //! This module contains all account-related methods including balance, //! trade history, account configuration, and user data stream management. -use super::super::{Binance, parser}; +use super::super::{Binance, BinanceEndpointRouter, parser}; use ccxt_core::types::AccountType; use ccxt_core::{ Error, ParseError, Result, - types::{Balance, Currency, FeeTradingFee, MarketType, Trade}, + types::{Balance, Currency, EndpointType, FeeTradingFee, MarketType, Trade}, }; use reqwest::header::HeaderMap; use std::collections::HashMap; @@ -391,7 +391,9 @@ impl Binance { limit: Option, ) -> Result> { let market = self.base().market(symbol).await?; - let url = format!("{}/myTrades", self.urls().private); + // Use market-type-aware endpoint routing + let base_url = self.rest_endpoint(&market, EndpointType::Private); + let url = format!("{}/myTrades", base_url); let data = self .signed_request(url) @@ -470,7 +472,9 @@ impl Binance { params: Option>, ) -> Result> { let market = self.base().market(symbol).await?; - let url = format!("{}/myTrades", self.urls().private); + // Use market-type-aware endpoint routing + let base_url = self.rest_endpoint(&market, EndpointType::Private); + let url = format!("{}/myTrades", base_url); // Convert HashMap to serde_json::Value for merge_json_params let json_params = params.map(|p| { @@ -714,7 +718,7 @@ impl Binance { Ok(fees) } - /// Create a listen key for user data stream. + /// Create a listen key for user data stream (spot market). /// /// Creates a new listen key that can be used to subscribe to user data streams /// via WebSocket. The listen key is valid for 60 minutes. @@ -746,9 +750,57 @@ impl Binance { /// # } /// ``` pub async fn create_listen_key(&self) -> Result { + self.create_listen_key_for_market(None).await + } + + /// Create a listen key for user data stream with market type routing. + /// + /// Creates a new listen key for the specified market type. Different market types + /// use different API endpoints: + /// - Spot: `/api/v3/userDataStream` + /// - USDT-M Futures: `/fapi/v1/listenKey` + /// - COIN-M Futures: `/dapi/v1/listenKey` + /// - Options: `/eapi/v1/listenKey` + /// + /// # Arguments + /// + /// * `market_type` - Optional market type. If None, defaults to Spot. + /// + /// # Returns + /// + /// Returns the listen key string. + /// + /// # Errors + /// + /// Returns an error if: + /// - API credentials are not configured + /// - API request fails + /// + /// # Example + /// + /// ```no_run + /// # use ccxt_exchanges::binance::Binance; + /// # use ccxt_core::ExchangeConfig; + /// # use ccxt_core::types::MarketType; + /// # async fn example() -> ccxt_core::Result<()> { + /// let mut config = ExchangeConfig::default(); + /// config.api_key = Some("your_api_key".to_string()); + /// config.secret = Some("your_secret".to_string()); + /// let binance = Binance::new(config)?; + /// + /// // Create listen key for futures + /// let listen_key = binance.create_listen_key_for_market(Some(MarketType::Swap)).await?; + /// println!("Futures Listen Key: {}", listen_key); + /// # Ok(()) + /// # } + /// ``` + pub async fn create_listen_key_for_market( + &self, + market_type: Option, + ) -> Result { self.check_required_credentials()?; - let url = format!("{}/userDataStream", self.urls().public); + let url = self.get_listen_key_url(market_type, None); let mut headers = HeaderMap::new(); let auth = self.get_auth()?; @@ -766,7 +818,7 @@ impl Binance { .ok_or_else(|| Error::from(ParseError::missing_field("listenKey"))) } - /// Refresh listen key to extend validity. + /// Refresh listen key to extend validity (spot market). /// /// Extends the listen key validity by 60 minutes. Recommended to call /// every 30 minutes to maintain the connection. @@ -786,13 +838,36 @@ impl Binance { /// - Listen key is invalid or expired /// - API request fails pub async fn refresh_listen_key(&self, listen_key: &str) -> Result<()> { + self.refresh_listen_key_for_market(listen_key, None).await + } + + /// Refresh listen key with market type routing. + /// + /// Extends the listen key validity by 60 minutes for the specified market type. + /// + /// # Arguments + /// + /// * `listen_key` - The listen key to refresh. + /// * `market_type` - Optional market type. If None, defaults to Spot. + /// + /// # Returns + /// + /// Returns `Ok(())` on success. + /// + /// # Errors + /// + /// Returns an error if: + /// - API credentials are not configured + /// - Listen key is invalid or expired + /// - API request fails + pub async fn refresh_listen_key_for_market( + &self, + listen_key: &str, + market_type: Option, + ) -> Result<()> { self.check_required_credentials()?; - let url = format!( - "{}/userDataStream?listenKey={}", - self.urls().public, - listen_key - ); + let url = self.get_listen_key_url(market_type, Some(listen_key)); let mut headers = HeaderMap::new(); let auth = self.get_auth()?; @@ -807,7 +882,7 @@ impl Binance { Ok(()) } - /// Delete listen key to close user data stream. + /// Delete listen key to close user data stream (spot market). /// /// Closes the user data stream connection and invalidates the listen key. /// @@ -825,13 +900,270 @@ impl Binance { /// - API credentials are not configured /// - API request fails pub async fn delete_listen_key(&self, listen_key: &str) -> Result<()> { + self.delete_listen_key_for_market(listen_key, None).await + } + + /// Delete listen key with market type routing. + /// + /// Closes the user data stream connection for the specified market type. + /// + /// # Arguments + /// + /// * `listen_key` - The listen key to delete. + /// * `market_type` - Optional market type. If None, defaults to Spot. + /// + /// # Returns + /// + /// Returns `Ok(())` on success. + /// + /// # Errors + /// + /// Returns an error if: + /// - API credentials are not configured + /// - API request fails + pub async fn delete_listen_key_for_market( + &self, + listen_key: &str, + market_type: Option, + ) -> Result<()> { + self.check_required_credentials()?; + + let url = self.get_listen_key_url(market_type, Some(listen_key)); + let mut headers = HeaderMap::new(); + + let auth = self.get_auth()?; + auth.add_auth_headers_reqwest(&mut headers); + + let _response = self + .base() + .http_client + .delete(&url, Some(headers), None) + .await?; + + Ok(()) + } + + /// Get the listen key URL for the specified market type. + /// + /// # Arguments + /// + /// * `market_type` - Optional market type. If None, defaults to Spot. + /// * `listen_key` - Optional listen key to include in the URL (for refresh/delete). + /// + /// # Returns + /// + /// Returns the appropriate URL for the listen key operation. + fn get_listen_key_url( + &self, + market_type: Option, + listen_key: Option<&str>, + ) -> String { + let urls = self.urls(); + let (base_url, endpoint) = match market_type { + Some(MarketType::Swap | MarketType::Futures) => { + // Check if it's linear (USDT-M) or inverse (COIN-M) based on default_type + // For now, default to linear (fapi) for Swap/Futures + (urls.fapi_public.clone(), "listenKey") + } + Some(MarketType::Option) => (urls.eapi_public.clone(), "listenKey"), + // Spot is the default + None | Some(MarketType::Spot) => (urls.public.clone(), "userDataStream"), + }; + + match listen_key { + Some(key) => format!("{}/{}?listenKey={}", base_url, endpoint, key), + None => format!("{}/{}", base_url, endpoint), + } + } + + /// Get the listen key URL for linear (USDT-M) futures. + /// + /// # Arguments + /// + /// * `listen_key` - Optional listen key to include in the URL. + /// + /// # Returns + /// + /// Returns the URL for linear futures listen key operations. + pub fn get_linear_listen_key_url(&self, listen_key: Option<&str>) -> String { + let base_url = &self.urls().fapi_public; + match listen_key { + Some(key) => format!("{}/listenKey?listenKey={}", base_url, key), + None => format!("{}/listenKey", base_url), + } + } + + /// Get the listen key URL for inverse (COIN-M) futures. + /// + /// # Arguments + /// + /// * `listen_key` - Optional listen key to include in the URL. + /// + /// # Returns + /// + /// Returns the URL for inverse futures listen key operations. + pub fn get_inverse_listen_key_url(&self, listen_key: Option<&str>) -> String { + let base_url = &self.urls().dapi_public; + match listen_key { + Some(key) => format!("{}/listenKey?listenKey={}", base_url, key), + None => format!("{}/listenKey", base_url), + } + } + + /// Create a listen key for linear (USDT-M) futures. + /// + /// # Returns + /// + /// Returns the listen key string. + /// + /// # Errors + /// + /// Returns an error if API credentials are not configured or the request fails. + pub async fn create_linear_listen_key(&self) -> Result { + self.check_required_credentials()?; + + let url = self.get_linear_listen_key_url(None); + let mut headers = HeaderMap::new(); + + let auth = self.get_auth()?; + auth.add_auth_headers_reqwest(&mut headers); + + let response = self + .base() + .http_client + .post(&url, Some(headers), None) + .await?; + + response["listenKey"] + .as_str() + .map(ToString::to_string) + .ok_or_else(|| Error::from(ParseError::missing_field("listenKey"))) + } + + /// Create a listen key for inverse (COIN-M) futures. + /// + /// # Returns + /// + /// Returns the listen key string. + /// + /// # Errors + /// + /// Returns an error if API credentials are not configured or the request fails. + pub async fn create_inverse_listen_key(&self) -> Result { + self.check_required_credentials()?; + + let url = self.get_inverse_listen_key_url(None); + let mut headers = HeaderMap::new(); + + let auth = self.get_auth()?; + auth.add_auth_headers_reqwest(&mut headers); + + let response = self + .base() + .http_client + .post(&url, Some(headers), None) + .await?; + + response["listenKey"] + .as_str() + .map(ToString::to_string) + .ok_or_else(|| Error::from(ParseError::missing_field("listenKey"))) + } + + /// Refresh a linear (USDT-M) futures listen key. + /// + /// # Arguments + /// + /// * `listen_key` - The listen key to refresh. + /// + /// # Returns + /// + /// Returns `Ok(())` on success. + pub async fn refresh_linear_listen_key(&self, listen_key: &str) -> Result<()> { + self.check_required_credentials()?; + + let url = self.get_linear_listen_key_url(Some(listen_key)); + let mut headers = HeaderMap::new(); + + let auth = self.get_auth()?; + auth.add_auth_headers_reqwest(&mut headers); + + let _response = self + .base() + .http_client + .put(&url, Some(headers), None) + .await?; + + Ok(()) + } + + /// Refresh an inverse (COIN-M) futures listen key. + /// + /// # Arguments + /// + /// * `listen_key` - The listen key to refresh. + /// + /// # Returns + /// + /// Returns `Ok(())` on success. + pub async fn refresh_inverse_listen_key(&self, listen_key: &str) -> Result<()> { + self.check_required_credentials()?; + + let url = self.get_inverse_listen_key_url(Some(listen_key)); + let mut headers = HeaderMap::new(); + + let auth = self.get_auth()?; + auth.add_auth_headers_reqwest(&mut headers); + + let _response = self + .base() + .http_client + .put(&url, Some(headers), None) + .await?; + + Ok(()) + } + + /// Delete a linear (USDT-M) futures listen key. + /// + /// # Arguments + /// + /// * `listen_key` - The listen key to delete. + /// + /// # Returns + /// + /// Returns `Ok(())` on success. + pub async fn delete_linear_listen_key(&self, listen_key: &str) -> Result<()> { + self.check_required_credentials()?; + + let url = self.get_linear_listen_key_url(Some(listen_key)); + let mut headers = HeaderMap::new(); + + let auth = self.get_auth()?; + auth.add_auth_headers_reqwest(&mut headers); + + let _response = self + .base() + .http_client + .delete(&url, Some(headers), None) + .await?; + + Ok(()) + } + + /// Delete an inverse (COIN-M) futures listen key. + /// + /// # Arguments + /// + /// * `listen_key` - The listen key to delete. + /// + /// # Returns + /// + /// Returns `Ok(())` on success. + pub async fn delete_inverse_listen_key(&self, listen_key: &str) -> Result<()> { self.check_required_credentials()?; - let url = format!( - "{}/userDataStream?listenKey={}", - self.urls().public, - listen_key - ); + let url = self.get_inverse_listen_key_url(Some(listen_key)); let mut headers = HeaderMap::new(); let auth = self.get_auth()?; diff --git a/ccxt-exchanges/src/binance/rest/market_data.rs b/ccxt-exchanges/src/binance/rest/market_data.rs index 6613562..8c9d2cd 100644 --- a/ccxt-exchanges/src/binance/rest/market_data.rs +++ b/ccxt-exchanges/src/binance/rest/market_data.rs @@ -358,6 +358,17 @@ impl Binance { symbol: &str, limit: Option, ) -> Result { + // Validate depth limit if provided + const VALID_DEPTH_LIMITS: &[u32] = &[5, 10, 20, 50, 100, 500, 1000, 5000]; + if let Some(l) = limit { + if !VALID_DEPTH_LIMITS.contains(&l) { + return Err(Error::invalid_request(format!( + "Invalid depth limit: {}. Valid values: {:?}", + l, VALID_DEPTH_LIMITS + ))); + } + } + let market = self.base().market(symbol).await?; let full_url = format!( diff --git a/ccxt-exchanges/src/binance/rest/spot.rs b/ccxt-exchanges/src/binance/rest/spot.rs index 8d0d85f..48ec6fb 100644 --- a/ccxt-exchanges/src/binance/rest/spot.rs +++ b/ccxt-exchanges/src/binance/rest/spot.rs @@ -3,10 +3,12 @@ //! This module contains all spot trading methods including order creation, //! cancellation, and order management. -use super::super::{Binance, constants::endpoints, parser, signed_request::HttpMethod}; +use super::super::{ + Binance, BinanceEndpointRouter, constants::endpoints, parser, signed_request::HttpMethod, +}; use ccxt_core::{ Error, ParseError, Result, - types::{Amount, Order, OrderRequest, OrderSide, OrderType, Price, TimeInForce}, + types::{Amount, EndpointType, Order, OrderRequest, OrderSide, OrderType, Price, TimeInForce}, }; use rust_decimal::Decimal; use std::collections::{BTreeMap, HashMap}; @@ -191,7 +193,9 @@ impl Binance { request_params.insert("positionSide".to_string(), position_side); } - let url = format!("{}{}", self.urls().private, endpoints::ORDER); + // Use market-type-aware endpoint routing for correct API base URL + let base_url = self.rest_endpoint(&market, EndpointType::Private); + let url = format!("{}{}", base_url, endpoints::ORDER); let data = self .signed_request(url) .method(HttpMethod::Post) @@ -333,7 +337,9 @@ impl Binance { } } - let url = format!("{}{}", self.urls().private, endpoints::ORDER); + // Use market-type-aware endpoint routing for correct API base URL + let base_url = self.rest_endpoint(&market, EndpointType::Private); + let url = format!("{}{}", base_url, endpoints::ORDER); let data = self .signed_request(url) .method(HttpMethod::Post) @@ -360,7 +366,9 @@ impl Binance { /// Returns an error if authentication fails, market is not found, or the API request fails. pub async fn cancel_order(&self, id: &str, symbol: &str) -> Result { let market = self.base().market(symbol).await?; - let url = format!("{}{}", self.urls().private, endpoints::ORDER); + // Use market-type-aware endpoint routing + let base_url = self.rest_endpoint(&market, EndpointType::Private); + let url = format!("{}{}", base_url, endpoints::ORDER); let data = self .signed_request(url) @@ -389,7 +397,9 @@ impl Binance { /// Returns an error if authentication fails, market is not found, or the API request fails. pub async fn fetch_order(&self, id: &str, symbol: &str) -> Result { let market = self.base().market(symbol).await?; - let url = format!("{}{}", self.urls().private, endpoints::ORDER); + // Use market-type-aware endpoint routing + let base_url = self.rest_endpoint(&market, EndpointType::Private); + let url = format!("{}{}", base_url, endpoints::ORDER); let data = self .signed_request(url) @@ -421,7 +431,13 @@ impl Binance { None }; - let url = format!("{}{}", self.urls().private, endpoints::OPEN_ORDERS); + // Use market-type-aware endpoint routing if market is available, + // otherwise fall back to spot endpoint + let base_url = match &market { + Some(m) => self.rest_endpoint(m, EndpointType::Private), + None => self.urls().private.clone(), + }; + let url = format!("{}{}", base_url, endpoints::OPEN_ORDERS); let data = self .signed_request(url) @@ -499,7 +515,9 @@ impl Binance { /// Returns an error if authentication fails, market is not found, or the API request fails. pub async fn cancel_all_orders(&self, symbol: &str) -> Result> { let market = self.base().market(symbol).await?; - let url = format!("{}{}", self.urls().private, endpoints::OPEN_ORDERS); + // Use market-type-aware endpoint routing + let base_url = self.rest_endpoint(&market, EndpointType::Private); + let url = format!("{}{}", base_url, endpoints::OPEN_ORDERS); let data = self .signed_request(url) @@ -552,7 +570,9 @@ impl Binance { )) })?; - let url = format!("{}{}", self.urls().private, endpoints::OPEN_ORDERS); + // Use market-type-aware endpoint routing + let base_url = self.rest_endpoint(&market, EndpointType::Private); + let url = format!("{}{}", base_url, endpoints::OPEN_ORDERS); let data = self .signed_request(url) @@ -609,7 +629,13 @@ impl Binance { None }; - let url = format!("{}{}", self.urls().private, endpoints::ALL_ORDERS); + // Use market-type-aware endpoint routing if market is available, + // otherwise fall back to spot endpoint + let base_url = match &market { + Some(m) => self.rest_endpoint(m, EndpointType::Private), + None => self.urls().private.clone(), + }; + let url = format!("{}{}", base_url, endpoints::ALL_ORDERS); let data = self .signed_request(url) diff --git a/ccxt-exchanges/src/binance/signed_request.rs b/ccxt-exchanges/src/binance/signed_request.rs index bee1ab9..77b0838 100644 --- a/ccxt-exchanges/src/binance/signed_request.rs +++ b/ccxt-exchanges/src/binance/signed_request.rs @@ -40,7 +40,9 @@ //! ``` use super::Binance; -use ccxt_core::{Error, ParseError, Result}; +use super::error::BinanceApiError; +use super::rate_limiter::RateLimitInfo; +use ccxt_core::Result; use reqwest::header::HeaderMap; use serde_json::Value; use std::collections::BTreeMap; @@ -51,11 +53,11 @@ pub enum HttpMethod { /// GET request - parameters in query string #[default] Get, - /// POST request - parameters in JSON body + /// POST request - parameters in query string (Binance requires query string, not JSON body) Post, - /// PUT request - parameters in JSON body + /// PUT request - parameters in query string (Binance requires query string, not JSON body) Put, - /// DELETE request - parameters in JSON body + /// DELETE request - parameters in query string (Binance requires query string, not JSON body) Delete, } @@ -320,13 +322,18 @@ impl<'a> SignedRequestBuilder<'a> { /// # } /// ``` pub async fn execute(self) -> Result { - // Step 1: Validate credentials + // Step 1: Check rate limiter - wait if needed + if let Some(wait_duration) = self.binance.rate_limiter().wait_duration() { + tokio::time::sleep(wait_duration).await; + } + + // Step 2: Validate credentials self.binance.check_required_credentials()?; - // Step 2: Get signing timestamp + // Step 3: Get signing timestamp let timestamp = self.binance.get_signing_timestamp().await?; - // Step 3: Get auth and sign parameters + // Step 4: Get auth and sign parameters let auth = self.binance.get_auth()?; let signed_params = auth.sign_with_timestamp( &self.params, @@ -334,20 +341,22 @@ impl<'a> SignedRequestBuilder<'a> { Some(self.binance.options().recv_window), )?; - // Step 4: Add authentication headers + // Step 5: Add authentication headers let mut headers = HeaderMap::new(); auth.add_auth_headers_reqwest(&mut headers); - // Step 5: Execute HTTP request based on method - match self.method { + // Step 6: 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 + } else { + format!("{}?{}", self.endpoint, query_string) + }; + + let result = match self.method { HttpMethod::Get => { - // Build query string for GET requests - let query_string = build_query_string(&signed_params); - let url = if query_string.is_empty() { - self.endpoint - } else { - format!("{}?{}", self.endpoint, query_string) - }; self.binance .base() .http_client @@ -355,58 +364,60 @@ impl<'a> SignedRequestBuilder<'a> { .await } HttpMethod::Post => { - // Serialize parameters as JSON body for POST requests - let body = serde_json::to_value(&signed_params).map_err(|e| { - Error::from(ParseError::invalid_format( - "data", - format!("Failed to serialize request body: {}", e), - )) - })?; self.binance .base() .http_client - .post(&self.endpoint, Some(headers), Some(body)) + .post(&url, Some(headers), None) .await } HttpMethod::Put => { - // Serialize parameters as JSON body for PUT requests - let body = serde_json::to_value(&signed_params).map_err(|e| { - Error::from(ParseError::invalid_format( - "data", - format!("Failed to serialize request body: {}", e), - )) - })?; self.binance .base() .http_client - .put(&self.endpoint, Some(headers), Some(body)) + .put(&url, Some(headers), None) .await } HttpMethod::Delete => { - // Serialize parameters as JSON body for DELETE requests - let body = serde_json::to_value(&signed_params).map_err(|e| { - Error::from(ParseError::invalid_format( - "data", - format!("Failed to serialize request body: {}", e), - )) - })?; self.binance .base() .http_client - .delete(&self.endpoint, Some(headers), Some(body)) + .delete(&url, Some(headers), None) .await } + }?; + + // Step 7: Update rate limiter from response headers + if let Some(headers) = result.get("responseHeaders") { + let rate_info = RateLimitInfo::from_headers(headers); + if rate_info.has_data() { + self.binance.rate_limiter().update(rate_info); + } } + + // Step 8: 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) + if api_error.is_ip_banned() { + self.binance + .rate_limiter() + .set_ip_banned(std::time::Duration::from_secs(7200)); // 2 hours default + } + return Err(api_error.into()); + } + + Ok(result) } } /// Builds a URL-encoded query string from parameters. /// /// Parameters are sorted alphabetically by key (due to BTreeMap ordering). +/// Both keys and values are URL-encoded to handle special characters. fn build_query_string(params: &BTreeMap) -> String { params .iter() - .map(|(k, v)| format!("{}={}", k, v)) + .map(|(k, v)| format!("{}={}", urlencoding::encode(k), urlencoding::encode(v))) .collect::>() .join("&") } diff --git a/ccxt-exchanges/src/binance/signing_strategy.rs b/ccxt-exchanges/src/binance/signing_strategy.rs index 1e1c2b5..a7f6d2a 100644 --- a/ccxt-exchanges/src/binance/signing_strategy.rs +++ b/ccxt-exchanges/src/binance/signing_strategy.rs @@ -76,11 +76,13 @@ impl SigningStrategy for BinanceSigningStrategy { } } -/// Build query string for signing (without URL encoding). +/// Build query string for signing with URL encoding. +/// +/// Both keys and values are URL-encoded to handle special characters correctly. fn build_signing_string(params: &std::collections::BTreeMap) -> String { params .iter() - .map(|(k, v)| format!("{}={}", k, v)) + .map(|(k, v)| format!("{}={}", urlencoding::encode(k), urlencoding::encode(v))) .collect::>() .join("&") } diff --git a/ccxt-exchanges/src/binance/ws/binance_impl.rs b/ccxt-exchanges/src/binance/ws/binance_impl.rs index 8e84f77..808511f 100644 --- a/ccxt-exchanges/src/binance/ws/binance_impl.rs +++ b/ccxt-exchanges/src/binance/ws/binance_impl.rs @@ -9,7 +9,8 @@ impl Binance { &self, symbol: &str, ) -> Result> { - let ws = self.connection_manager.get_public_connection().await?; + let default_market_type = MarketType::from(self.options.default_type); + let ws = self.connection_manager.get_public_connection(default_market_type).await?; let binance_symbol = symbol.replace('/', "").to_lowercase(); ws.subscribe_ticker(&binance_symbol).await } @@ -21,7 +22,8 @@ impl Binance { &self, symbol: &str, ) -> Result> { - let ws = self.connection_manager.get_public_connection().await?; + let default_market_type = MarketType::from(self.options.default_type); + let ws = self.connection_manager.get_public_connection(default_market_type).await?; let binance_symbol = symbol.replace('/', "").to_lowercase(); ws.subscribe_trades(&binance_symbol).await } @@ -36,7 +38,8 @@ impl Binance { ) -> Result> { use super::ws::{DepthLevel, UpdateSpeed}; - let ws = self.connection_manager.get_public_connection().await?; + let default_market_type = MarketType::from(self.options.default_type); + let ws = self.connection_manager.get_public_connection(default_market_type).await?; let binance_symbol = symbol.replace('/', "").to_lowercase(); let depth_level = match levels.unwrap_or(20) { 5 => DepthLevel::L5, @@ -55,7 +58,8 @@ impl Binance { symbol: &str, interval: &str, ) -> Result> { - let ws = self.connection_manager.get_public_connection().await?; + let default_market_type = MarketType::from(self.options.default_type); + let ws = self.connection_manager.get_public_connection(default_market_type).await?; let binance_symbol = symbol.replace('/', "").to_lowercase(); ws.subscribe_kline(&binance_symbol, interval).await } @@ -78,7 +82,7 @@ impl Binance { "ticker" }; - let ws = self.connection_manager.get_public_connection().await?; + let ws = self.connection_manager.get_public_connection(market.market_type).await?; ws.watch_ticker_internal(&binance_symbol, channel_name) .await } @@ -116,7 +120,8 @@ impl Binance { None }; - let ws = self.connection_manager.get_public_connection().await?; + let default_market_type = MarketType::from(self.options.default_type); + let ws = self.connection_manager.get_public_connection(default_market_type).await?; ws.watch_tickers_internal(binance_symbols, channel_name) .await } @@ -153,7 +158,7 @@ impl Binance { "markPrice" }; - let ws = self.connection_manager.get_public_connection().await?; + let ws = self.connection_manager.get_public_connection(market.market_type).await?; ws.watch_mark_price_internal(&binance_symbol, channel_name) .await } @@ -184,7 +189,7 @@ impl Binance { UpdateSpeed::Ms100 }; - let ws = self.connection_manager.get_public_connection().await?; + let ws = self.connection_manager.get_public_connection(market.market_type).await?; ws.watch_orderbook_internal(self, &binance_symbol, limit, update_speed, is_futures) .await } @@ -212,10 +217,12 @@ impl Binance { let mut binance_symbols = Vec::new(); let mut is_futures = false; + let mut market_type = MarketType::from(self.options.default_type); for symbol in &symbols { let market = self.base.market(symbol).await?; binance_symbols.push(market.id.to_lowercase()); + market_type = market.market_type; let current_is_futures = market.market_type == MarketType::Swap || market.market_type == MarketType::Futures; @@ -236,7 +243,7 @@ impl Binance { UpdateSpeed::Ms100 }; - let ws = self.connection_manager.get_public_connection().await?; + let ws = self.connection_manager.get_public_connection(market_type).await?; ws.watch_orderbooks_internal(self, binance_symbols, limit, update_speed, is_futures) .await } @@ -263,6 +270,7 @@ impl Binance { "markPrice" }; + let mut market_type = MarketType::from(self.options.default_type); let binance_symbols = if let Some(syms) = symbols { let mut result = Vec::new(); for symbol in syms { @@ -275,6 +283,7 @@ impl Binance { market.market_type ))); } + market_type = market.market_type; result.push(market.id.to_lowercase()); } Some(result) @@ -282,7 +291,7 @@ impl Binance { None }; - let ws = self.connection_manager.get_public_connection().await?; + let ws = self.connection_manager.get_public_connection(market_type).await?; ws.watch_mark_prices_internal(binance_symbols, channel_name) .await } @@ -299,7 +308,7 @@ impl Binance { let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); - let ws = self.connection_manager.get_public_connection().await?; + let ws = self.connection_manager.get_public_connection(market.market_type).await?; ws.watch_trades_internal(symbol, &binance_symbol, since, limit, Some(&market)) .await @@ -318,7 +327,7 @@ impl Binance { let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); - let ws = self.connection_manager.get_public_connection().await?; + let ws = self.connection_manager.get_public_connection(market.market_type).await?; ws.watch_ohlcv_internal(symbol, &binance_symbol, timeframe, since, limit) .await @@ -331,7 +340,7 @@ impl Binance { let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); - let ws = self.connection_manager.get_public_connection().await?; + let ws = self.connection_manager.get_public_connection(market.market_type).await?; ws.watch_bids_asks_internal(symbol, &binance_symbol).await } @@ -367,9 +376,10 @@ impl Binance { true }; + let default_market_type = MarketType::from(self.options.default_type); let ws = self .connection_manager - .get_private_connection(&self) + .get_private_connection(default_market_type, &self) .await?; if fetch_snapshot { @@ -397,9 +407,10 @@ impl Binance { ) -> Result> { self.base.load_markets(false).await?; + let default_market_type = MarketType::from(self.options.default_type); let ws = self .connection_manager - .get_private_connection(&self) + .get_private_connection(default_market_type, &self) .await?; ws.watch_orders_internal(symbol, since, limit).await } @@ -412,9 +423,10 @@ impl Binance { limit: Option, _params: Option>, ) -> Result> { + let default_market_type = MarketType::from(self.options.default_type); let ws = self .connection_manager - .get_private_connection(&self) + .get_private_connection(default_market_type, &self) .await?; ws.watch_my_trades_internal(symbol, since, limit).await } @@ -427,9 +439,10 @@ impl Binance { limit: Option, _params: Option>, ) -> Result> { + let default_market_type = MarketType::from(self.options.default_type); let ws = self .connection_manager - .get_private_connection(&self) + .get_private_connection(default_market_type, &self) .await?; ws.watch_positions_internal(symbols, since, limit).await } diff --git a/ccxt-exchanges/src/binance/ws/connection_manager.rs b/ccxt-exchanges/src/binance/ws/connection_manager.rs index 557f6e4..2c3c4e8 100644 --- a/ccxt-exchanges/src/binance/ws/connection_manager.rs +++ b/ccxt-exchanges/src/binance/ws/connection_manager.rs @@ -1,103 +1,201 @@ use crate::binance::Binance; +use crate::binance::BinanceUrls; use crate::binance::ws::BinanceWs; use ccxt_core::error::Result; +use ccxt_core::types::MarketType; +use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; const MAX_STREAMS_PER_CONNECTION: usize = 200; -/// Manages multiple WebSocket connections for Binance to handle sharding and isolation +/// Manages multiple WebSocket connections for Binance to handle sharding and isolation. +/// +/// Connections are organized by market type (Spot, Swap, Futures, Option) to ensure +/// that each market type connects to the correct WebSocket endpoint. #[derive(Debug)] pub struct BinanceConnectionManager { - /// Pool of public data connections (shards) - public_shards: RwLock>>, - /// Dedicated connection for private user data - private_client: RwLock>>, - /// Base WebSocket URL - base_ws_url: String, + /// Pool of public data connections organized by market type + public_shards: RwLock>>>, + /// Dedicated connections for private user data organized by market type + private_clients: RwLock>>, + /// WebSocket URLs for different market types + urls: BinanceUrls, + /// Whether to use sandbox/testnet URLs + is_sandbox: bool, } impl BinanceConnectionManager { - /// Creates a new connection manager - pub fn new(base_ws_url: String) -> Self { + /// Creates a new connection manager with the given URLs. + pub fn new(urls: BinanceUrls, is_sandbox: bool) -> Self { Self { - public_shards: RwLock::new(Vec::new()), - private_client: RwLock::new(None), - base_ws_url, + public_shards: RwLock::new(HashMap::new()), + private_clients: RwLock::new(HashMap::new()), + urls, + is_sandbox, } } - /// Gets or creates a public connection suitable for a new subscription - pub async fn get_public_connection(&self) -> Result> { + /// Creates a new connection manager for production environment. + pub fn production() -> Self { + Self::new(BinanceUrls::production(), false) + } + + /// Creates a new connection manager for testnet environment. + pub fn testnet() -> Self { + Self::new(BinanceUrls::testnet(), true) + } + + /// Returns the WebSocket URL for the given market type. + /// + /// # Arguments + /// + /// * `market_type` - The market type to get the URL for + /// + /// # Returns + /// + /// The WebSocket URL for the specified market type. + pub fn get_ws_url(&self, market_type: MarketType) -> &str { + match market_type { + MarketType::Spot => &self.urls.ws, + MarketType::Futures => &self.urls.ws_fapi, // Linear/USDT-margined futures + MarketType::Swap => &self.urls.ws_dapi, // Inverse/Coin-margined perpetuals + MarketType::Option => &self.urls.ws_eapi, + } + } + + /// Gets or creates a public connection suitable for a new subscription. + /// + /// Connections are lazily initialized on first subscription for each market type. + /// + /// # Arguments + /// + /// * `market_type` - The market type for the connection + /// + /// # Returns + /// + /// An Arc to the WebSocket client for the specified market type. + pub async fn get_public_connection(&self, market_type: MarketType) -> Result> { let mut shards = self.public_shards.write().await; + // Get or create the shard list for this market type + let market_shards = shards.entry(market_type).or_insert_with(Vec::new); + // Try to find an existing shard with capacity - for shard in shards.iter() { + for shard in market_shards.iter() { if shard.subscription_count() < MAX_STREAMS_PER_CONNECTION { return Ok(shard.clone()); } } // If no shard available or all full, create a new one - let new_shard = Arc::new(BinanceWs::new(self.base_ws_url.clone())); + let ws_url = self.get_ws_url(market_type).to_string(); + let new_shard = Arc::new(BinanceWs::new(ws_url)); // Connect the new shard immediately new_shard.connect().await?; - shards.push(new_shard.clone()); + market_shards.push(new_shard.clone()); Ok(new_shard) } - /// Gets or creates the dedicated private connection - pub async fn get_private_connection(&self, binance: &Arc) -> Result> { - let mut client_lock = self.private_client.write().await; + /// Gets or creates the dedicated private connection for the given market type. + /// + /// # Arguments + /// + /// * `market_type` - The market type for the connection + /// * `binance` - Reference to the Binance instance for authentication + /// + /// # Returns + /// + /// An Arc to the authenticated WebSocket client. + pub async fn get_private_connection( + &self, + market_type: MarketType, + binance: &Arc, + ) -> Result> { + let mut clients = self.private_clients.write().await; - if let Some(client) = client_lock.as_ref() { + // Check if we have an existing connected client for this market type + if let Some(client) = clients.get(&market_type) { if client.is_connected() { return Ok(client.clone()); } } - // Create new authenticated client + // Create new authenticated client with the correct URL for this market type + let ws_url = self.get_ws_url(market_type).to_string(); let new_client = Arc::new(BinanceWs::new_with_auth( - self.base_ws_url.clone(), + ws_url, binance.clone(), + market_type, )); // Connect user stream (this handles listenKey generation and management) new_client.connect_user_stream().await?; - *client_lock = Some(new_client.clone()); + clients.insert(market_type, new_client.clone()); Ok(new_client) } - /// Returns the number of active public shards + /// Returns the number of active public shards across all market types. pub async fn active_shards_count(&self) -> usize { - self.public_shards.read().await.len() + self.public_shards.read().await.values().map(Vec::len).sum() + } + + /// Returns the number of active public shards for a specific market type. + pub async fn active_shards_count_for(&self, market_type: MarketType) -> usize { + self.public_shards + .read() + .await + .get(&market_type) + .map_or(0, Vec::len) } - /// Disconnects all active connections + /// Disconnects all active connections across all market types. pub async fn disconnect_all(&self) -> Result<()> { + // Disconnect all public shards let mut shards = self.public_shards.write().await; - for shard in shards.iter() { - let _ = shard.disconnect().await; + for market_shards in shards.values() { + for shard in market_shards { + let _ = shard.disconnect().await; + } } shards.clear(); - let mut private = self.private_client.write().await; - if let Some(client) = private.take() { + // Disconnect all private clients + let mut private = self.private_clients.write().await; + for (_, client) in private.drain() { let _ = client.disconnect().await; } Ok(()) } - /// Checks if any connection is active (non-blocking) + /// Disconnects all connections for a specific market type. + pub async fn disconnect_market(&self, market_type: MarketType) -> Result<()> { + // Disconnect public shards for this market type + let mut shards = self.public_shards.write().await; + if let Some(market_shards) = shards.remove(&market_type) { + for shard in &market_shards { + let _ = shard.disconnect().await; + } + } + + // Disconnect private client for this market type + let mut private = self.private_clients.write().await; + if let Some(client) = private.remove(&market_type) { + let _ = client.disconnect().await; + } + + Ok(()) + } + + /// Checks if any connection is active (non-blocking). pub fn is_connected(&self) -> bool { if let Ok(shards) = self.public_shards.try_read() { - if !shards.is_empty() { - // Check if at least one shard is connected? - for shard in shards.iter() { + for market_shards in shards.values() { + for shard in market_shards { if shard.is_connected() { return true; } @@ -105,8 +203,8 @@ impl BinanceConnectionManager { } } - if let Ok(private) = self.private_client.try_read() { - if let Some(client) = private.as_ref() { + if let Ok(private) = self.private_clients.try_read() { + for (_, client) in private.iter() { if client.is_connected() { return true; } @@ -116,19 +214,243 @@ impl BinanceConnectionManager { false } - /// Returns a list of all active subscriptions across all connections + /// Checks if a connection is active for a specific market type (non-blocking). + pub fn is_connected_for(&self, market_type: MarketType) -> bool { + if let Ok(shards) = self.public_shards.try_read() { + if let Some(market_shards) = shards.get(&market_type) { + for shard in market_shards { + if shard.is_connected() { + return true; + } + } + } + } + + if let Ok(private) = self.private_clients.try_read() { + if let Some(client) = private.get(&market_type) { + if client.is_connected() { + return true; + } + } + } + + false + } + + /// Returns a list of all active subscriptions across all connections. pub fn get_all_subscriptions(&self) -> Vec { let mut subs = Vec::new(); if let Ok(shards) = self.public_shards.try_read() { - for shard in shards.iter() { - subs.extend(shard.subscriptions()); + for market_shards in shards.values() { + for shard in market_shards { + subs.extend(shard.subscriptions()); + } + } + } + if let Ok(private) = self.private_clients.try_read() { + for (_, client) in private.iter() { + subs.extend(client.subscriptions()); + } + } + subs + } + + /// Returns a list of active subscriptions for a specific market type. + pub fn get_subscriptions_for(&self, market_type: MarketType) -> Vec { + let mut subs = Vec::new(); + if let Ok(shards) = self.public_shards.try_read() { + if let Some(market_shards) = shards.get(&market_type) { + for shard in market_shards { + subs.extend(shard.subscriptions()); + } } } - if let Ok(private) = self.private_client.try_read() { - if let Some(client) = private.as_ref() { + if let Ok(private) = self.private_clients.try_read() { + if let Some(client) = private.get(&market_type) { subs.extend(client.subscriptions()); } } subs } + + /// Returns a reference to the URLs configuration. + pub fn urls(&self) -> &BinanceUrls { + &self.urls + } + + /// Returns whether this manager is configured for sandbox/testnet. + pub fn is_sandbox(&self) -> bool { + self.is_sandbox + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_production_urls() { + let manager = BinanceConnectionManager::production(); + assert!(!manager.is_sandbox()); + + assert_eq!( + manager.get_ws_url(MarketType::Spot), + "wss://stream.binance.com:9443/ws" + ); + assert_eq!( + manager.get_ws_url(MarketType::Futures), + "wss://fstream.binance.com/ws" + ); + assert_eq!( + manager.get_ws_url(MarketType::Swap), + "wss://dstream.binance.com/ws" + ); + assert_eq!( + manager.get_ws_url(MarketType::Option), + "wss://nbstream.binance.com/eoptions/ws" + ); + } + + #[test] + fn test_testnet_urls() { + let manager = BinanceConnectionManager::testnet(); + assert!(manager.is_sandbox()); + + assert_eq!( + manager.get_ws_url(MarketType::Spot), + "wss://testnet.binance.vision/ws" + ); + assert_eq!( + manager.get_ws_url(MarketType::Futures), + "wss://stream.binancefuture.com/ws" + ); + assert_eq!( + manager.get_ws_url(MarketType::Swap), + "wss://dstream.binancefuture.com/ws" + ); + } + + #[test] + fn test_custom_urls() { + let mut urls = BinanceUrls::production(); + urls.ws = "wss://custom.example.com/ws".to_string(); + urls.ws_fapi = "wss://custom-fapi.example.com/ws".to_string(); + + let manager = BinanceConnectionManager::new(urls, false); + + assert_eq!( + manager.get_ws_url(MarketType::Spot), + "wss://custom.example.com/ws" + ); + assert_eq!( + manager.get_ws_url(MarketType::Futures), + "wss://custom-fapi.example.com/ws" + ); + // Unchanged URLs should still be production defaults + assert_eq!( + manager.get_ws_url(MarketType::Swap), + "wss://dstream.binance.com/ws" + ); + } + + #[test] + fn test_initial_state() { + let manager = BinanceConnectionManager::production(); + + // No connections initially + assert!(!manager.is_connected()); + assert!(!manager.is_connected_for(MarketType::Spot)); + assert!(!manager.is_connected_for(MarketType::Futures)); + assert!(!manager.is_connected_for(MarketType::Swap)); + assert!(!manager.is_connected_for(MarketType::Option)); + + // No subscriptions initially + assert!(manager.get_all_subscriptions().is_empty()); + assert!(manager.get_subscriptions_for(MarketType::Spot).is_empty()); + } + + #[tokio::test] + async fn test_active_shards_count_initial() { + let manager = BinanceConnectionManager::production(); + + assert_eq!(manager.active_shards_count().await, 0); + assert_eq!(manager.active_shards_count_for(MarketType::Spot).await, 0); + assert_eq!( + manager.active_shards_count_for(MarketType::Futures).await, + 0 + ); + } + + #[tokio::test] + async fn test_disconnect_all_empty() { + let manager = BinanceConnectionManager::production(); + + // Should not error when disconnecting with no connections + let result = manager.disconnect_all().await; + assert!(result.is_ok()); + } + + #[tokio::test] + async fn test_disconnect_market_empty() { + let manager = BinanceConnectionManager::production(); + + // Should not error when disconnecting a market type with no connections + let result = manager.disconnect_market(MarketType::Spot).await; + assert!(result.is_ok()); + } + + #[test] + fn test_urls_accessor() { + let manager = BinanceConnectionManager::production(); + let urls = manager.urls(); + + assert!(urls.ws.contains("stream.binance.com")); + assert!(urls.ws_fapi.contains("fstream.binance.com")); + assert!(urls.ws_dapi.contains("dstream.binance.com")); + assert!(urls.ws_eapi.contains("nbstream.binance.com")); + } + + #[test] + fn test_market_type_url_routing_completeness() { + let manager = BinanceConnectionManager::production(); + + // Ensure all market types return non-empty URLs + for market_type in &[ + MarketType::Spot, + MarketType::Futures, + MarketType::Swap, + MarketType::Option, + ] { + let url = manager.get_ws_url(*market_type); + assert!( + !url.is_empty(), + "URL for {:?} should not be empty", + market_type + ); + assert!( + url.starts_with("wss://"), + "URL for {:?} should start with wss://, got: {}", + market_type, + url + ); + } + } + + #[test] + fn test_different_market_types_get_different_urls() { + let manager = BinanceConnectionManager::production(); + + let spot_url = manager.get_ws_url(MarketType::Spot); + let futures_url = manager.get_ws_url(MarketType::Futures); + let swap_url = manager.get_ws_url(MarketType::Swap); + let option_url = manager.get_ws_url(MarketType::Option); + + // All URLs should be different + assert_ne!(spot_url, futures_url); + assert_ne!(spot_url, swap_url); + assert_ne!(spot_url, option_url); + assert_ne!(futures_url, swap_url); + assert_ne!(futures_url, option_url); + assert_ne!(swap_url, option_url); + } } diff --git a/ccxt-exchanges/src/binance/ws/listen_key.rs b/ccxt-exchanges/src/binance/ws/listen_key.rs index 9fb2fd9..3e8b0c3 100644 --- a/ccxt-exchanges/src/binance/ws/listen_key.rs +++ b/ccxt-exchanges/src/binance/ws/listen_key.rs @@ -2,6 +2,7 @@ use crate::binance::Binance; use ccxt_core::error::{Error, Result}; +use ccxt_core::types::MarketType; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::{Mutex, RwLock}; @@ -26,13 +27,16 @@ const MAX_REFRESH_RETRIES: u32 = 3; /// Delay between refresh retry attempts const REFRESH_RETRY_DELAY: Duration = Duration::from_secs(5); -/// Listen key manager +/// Listen key manager with multi-market type support /// /// Automatically manages Binance user data stream listen keys by: -/// - Creating and caching listen keys -/// - Refreshing them every 30 minutes +/// - Creating and caching listen keys per market type +/// - Refreshing them every 25 minutes (configurable) /// - Detecting expiration and rebuilding /// - Tracking connection state +/// +/// Each market type (Spot, Futures, Swap, Option) uses a different API endpoint +/// for listen key management, so this manager routes requests accordingly. pub struct ListenKeyManager { /// Reference to the Binance instance binance: Arc, @@ -40,6 +44,8 @@ pub struct ListenKeyManager { listen_key: Arc>>, /// Listen key creation timestamp created_at: Arc>>, + /// Market type this manager is responsible for + market_type: MarketType, /// Configured refresh interval refresh_interval: Duration, /// Handle to the auto-refresh task @@ -47,17 +53,28 @@ pub struct ListenKeyManager { } impl ListenKeyManager { - /// Creates a new `ListenKeyManager` + /// Creates a new `ListenKeyManager` for the default (Spot) market type pub fn new(binance: Arc) -> Self { + Self::new_for_market(binance, MarketType::Spot) + } + + /// Creates a new `ListenKeyManager` for a specific market type + pub fn new_for_market(binance: Arc, market_type: MarketType) -> Self { Self { binance, listen_key: Arc::new(RwLock::new(None)), created_at: Arc::new(RwLock::new(None)), + market_type, refresh_interval: LISTEN_KEY_REFRESH_INTERVAL, refresh_task: Arc::new(Mutex::new(None)), } } + /// Returns the market type this manager is responsible for + pub fn market_type(&self) -> MarketType { + self.market_type + } + /// Retrieves or creates a listen key pub async fn get_or_create(&self) -> Result { let key_opt = self.listen_key.read().await.clone(); @@ -70,6 +87,7 @@ impl ListenKeyManager { if elapsed > LISTEN_KEY_MAX_AGE { drop(created); tracing::info!( + market_type = %self.market_type, "Listen key is {} minutes old, creating new one", elapsed.as_secs() / 60 ); @@ -82,13 +100,21 @@ impl ListenKeyManager { self.create_new().await } - /// Creates a new listen key + /// Creates a new listen key using the market-type-specific endpoint async fn create_new(&self) -> Result { - let key = self.binance.create_listen_key().await?; + let key = self + .binance + .create_listen_key_for_market(Some(self.market_type)) + .await?; *self.listen_key.write().await = Some(key.clone()); *self.created_at.write().await = Some(Instant::now()); + tracing::debug!( + market_type = %self.market_type, + "Created new listen key" + ); + Ok(key) } @@ -97,8 +123,14 @@ impl ListenKeyManager { let key_opt = self.listen_key.read().await.clone(); if let Some(key) = key_opt { - self.binance.refresh_listen_key(&key).await?; + self.binance + .refresh_listen_key_for_market(&key, Some(self.market_type)) + .await?; *self.created_at.write().await = Some(Instant::now()); + tracing::debug!( + market_type = %self.market_type, + "Refreshed listen key" + ); Ok(()) } else { Err(Error::invalid_request("No listen key to refresh")) @@ -119,6 +151,7 @@ impl ListenKeyManager { let created_at = self.created_at.clone(); let binance = self.binance.clone(); let interval = self.refresh_interval; + let market_type = self.market_type; let handle = tokio::spawn(async move { loop { @@ -129,15 +162,22 @@ impl ListenKeyManager { // Try to refresh with retries let mut refresh_succeeded = false; for attempt in 1..=MAX_REFRESH_RETRIES { - match binance.refresh_listen_key(&key).await { + match binance + .refresh_listen_key_for_market(&key, Some(market_type)) + .await + { Ok(()) => { *created_at.write().await = Some(Instant::now()); - tracing::debug!("Listen key refreshed successfully"); + tracing::debug!( + %market_type, + "Listen key refreshed successfully" + ); refresh_succeeded = true; break; } Err(e) => { tracing::warn!( + %market_type, "Failed to refresh listen key (attempt {}/{}): {}", attempt, MAX_REFRESH_RETRIES, @@ -153,16 +193,26 @@ impl ListenKeyManager { // If all refresh attempts failed, try to regenerate if !refresh_succeeded { tracing::warn!( + %market_type, "All refresh attempts failed, attempting to regenerate listen key" ); - match binance.create_listen_key().await { + match binance + .create_listen_key_for_market(Some(market_type)) + .await + { Ok(new_key) => { - tracing::info!("Regenerated listen key successfully"); + tracing::info!( + %market_type, + "Regenerated listen key successfully" + ); *listen_key.write().await = Some(new_key); *created_at.write().await = Some(Instant::now()); } Err(create_err) => { - tracing::error!("Failed to regenerate listen key: {}", create_err); + tracing::error!( + %market_type, + "Failed to regenerate listen key: {}", create_err + ); *listen_key.write().await = None; *created_at.write().await = None; break; @@ -193,11 +243,18 @@ impl ListenKeyManager { let key_opt = self.listen_key.read().await.clone(); if let Some(key) = key_opt { - self.binance.delete_listen_key(&key).await?; + self.binance + .delete_listen_key_for_market(&key, Some(self.market_type)) + .await?; *self.listen_key.write().await = None; *self.created_at.write().await = None; + tracing::debug!( + market_type = %self.market_type, + "Deleted listen key" + ); + Ok(()) } else { Ok(()) diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index 205e6a4..4a602e5 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/ccxt-exchanges/src/binance/ws/mod.rs @@ -64,7 +64,9 @@ pub use parsers::{ BidAskParser, MarkPriceParser, OhlcvParser, StreamParser, TickerParser, TradeParser, }; pub use streams::normalize_symbol; -pub use subscriptions::{ReconnectConfig, Subscription, SubscriptionManager, SubscriptionType}; +pub use subscriptions::{ + ReconnectConfig, Subscription, SubscriptionHandle, SubscriptionManager, SubscriptionType, +}; pub use types::{ BinanceWsConfig, DepthLevel, UpdateSpeed, WsChannelConfig, WsErrorRecovery, WsHealthSnapshot, }; @@ -236,10 +238,10 @@ impl BinanceWs { } } - /// Creates a WebSocket client with a listen key manager - pub fn new_with_auth(url: String, binance: Arc) -> Self { + /// Creates a WebSocket client with a listen key manager for a specific market type + pub fn new_with_auth(url: String, binance: Arc, market_type: MarketType) -> Self { let subscription_manager = Arc::new(SubscriptionManager::new()); - let listen_key_manager = Arc::new(ListenKeyManager::new(binance)); + let listen_key_manager = Arc::new(ListenKeyManager::new_for_market(binance, market_type)); let message_router = Arc::new(MessageRouter::new( url, subscription_manager.clone(), @@ -690,12 +692,21 @@ impl BinanceWs { Ok(rx) } - /// Cancels an existing subscription + /// Cancels an existing subscription. + /// + /// Only sends the UNSUBSCRIBE command to the server when the reference count + /// reaches zero (no more active handles for this stream). pub async fn unsubscribe(&self, stream: String) -> Result<()> { - self.subscription_manager + let fully_removed = self + .subscription_manager .remove_subscription(&stream) .await?; - self.message_router.unsubscribe(vec![stream]).await + + // Only send UNSUBSCRIBE to the server if the subscription was fully removed + if fully_removed { + self.message_router.unsubscribe(vec![stream]).await?; + } + Ok(()) } /// Receives the next available message diff --git a/ccxt-exchanges/src/binance/ws/subscriptions.rs b/ccxt-exchanges/src/binance/ws/subscriptions.rs index 9cdc1ce..5ad4c37 100644 --- a/ccxt-exchanges/src/binance/ws/subscriptions.rs +++ b/ccxt-exchanges/src/binance/ws/subscriptions.rs @@ -3,6 +3,7 @@ use serde_json::Value; use std::collections::HashMap; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Instant; use tokio::sync::RwLock; @@ -70,6 +71,8 @@ pub struct Subscription { pub subscribed_at: Instant, /// Sender for forwarding WebSocket messages to consumers pub sender: tokio::sync::mpsc::Sender, + /// Reference count for this subscription (how many handles are active) + ref_count: Arc, } impl Subscription { @@ -86,6 +89,7 @@ impl Subscription { sub_type, subscribed_at: Instant::now(), sender, + ref_count: Arc::new(AtomicUsize::new(1)), } } @@ -94,6 +98,22 @@ impl Subscription { // Use try_send to avoid blocking if channel is full (drop strategy) self.sender.try_send(message).is_ok() } + + /// Increments the reference count and returns the new value + pub fn add_ref(&self) -> usize { + self.ref_count.fetch_add(1, Ordering::SeqCst) + 1 + } + + /// Decrements the reference count and returns the new value + pub fn remove_ref(&self) -> usize { + let prev = self.ref_count.fetch_sub(1, Ordering::SeqCst); + prev.saturating_sub(1) + } + + /// Returns the current reference count + pub fn ref_count(&self) -> usize { + self.ref_count.load(Ordering::SeqCst) + } } /// Subscription manager @@ -116,7 +136,10 @@ impl SubscriptionManager { } } - /// Adds a subscription to the manager + /// Adds a subscription to the manager. + /// + /// If a subscription for the same stream already exists, increments the reference count + /// instead of creating a duplicate subscription. pub async fn add_subscription( &self, stream: String, @@ -124,9 +147,16 @@ impl SubscriptionManager { sub_type: SubscriptionType, sender: tokio::sync::mpsc::Sender, ) -> ccxt_core::error::Result<()> { + let mut subs = self.subscriptions.write().await; + + // Check if subscription already exists - increment ref count + if let Some(existing) = subs.get(&stream) { + existing.add_ref(); + return Ok(()); + } + let subscription = Subscription::new(stream.clone(), symbol.clone(), sub_type, sender); - let mut subs = self.subscriptions.write().await; subs.insert(stream.clone(), subscription); let mut index = self.symbol_index.write().await; @@ -138,11 +168,24 @@ impl SubscriptionManager { Ok(()) } - /// Removes a subscription by stream name - pub async fn remove_subscription(&self, stream: &str) -> ccxt_core::error::Result<()> { + /// Removes a subscription by stream name. + /// + /// Only actually removes the subscription when the reference count reaches zero. + /// Returns `true` if the subscription was fully removed (ref count hit zero). + pub async fn remove_subscription(&self, stream: &str) -> ccxt_core::error::Result { let mut subs = self.subscriptions.write().await; - if let Some(subscription) = subs.remove(stream) { + if let Some(subscription) = subs.get(stream) { + let remaining = subscription.remove_ref(); + if remaining > 0 { + // Still has active references, don't remove + return Ok(false); + } + + // Ref count is zero, remove the subscription + let Some(subscription) = subs.remove(stream) else { + return Ok(false); + }; let mut index = self.symbol_index.write().await; if let Some(streams) = index.get_mut(&subscription.symbol) { streams.retain(|s| s != stream); @@ -153,9 +196,10 @@ impl SubscriptionManager { self.active_count .fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + Ok(true) + } else { + Ok(false) } - - Ok(()) } /// Retrieves a subscription by stream name @@ -317,3 +361,99 @@ impl ReconnectConfig { self.enabled && (self.max_attempts == 0 || attempt < self.max_attempts) } } + +/// A handle to an active subscription that automatically unsubscribes when dropped. +/// +/// `SubscriptionHandle` implements a RAII pattern for WebSocket subscriptions. +/// When the handle is dropped, it decrements the reference count for the stream +/// and triggers an UNSUBSCRIBE command if no more handles reference the stream. +/// +/// # Example +/// +/// ```rust,ignore +/// let handle = subscription_manager.subscribe("btcusdt@ticker", ...).await?; +/// // ... use the subscription ... +/// drop(handle); // Automatically unsubscribes when dropped +/// ``` +pub struct SubscriptionHandle { + /// The stream name this handle is associated with + stream: String, + /// Reference to the subscription manager for cleanup + subscription_manager: Arc, + /// Reference to the message router for sending UNSUBSCRIBE + message_router: Option>, + /// Whether this handle has already been released + released: bool, +} + +impl SubscriptionHandle { + /// Creates a new subscription handle. + pub fn new( + stream: String, + subscription_manager: Arc, + message_router: Option>, + ) -> Self { + Self { + stream, + subscription_manager, + message_router, + released: false, + } + } + + /// Returns the stream name associated with this handle. + pub fn stream(&self) -> &str { + &self.stream + } + + /// Manually releases the subscription handle. + /// + /// This is equivalent to dropping the handle, but allows for async cleanup. + /// After calling this method, the Drop implementation will be a no-op. + pub async fn release(mut self) -> ccxt_core::error::Result<()> { + self.released = true; + self.do_release().await + } + + /// Internal release logic + async fn do_release(&self) -> ccxt_core::error::Result<()> { + let fully_removed = self + .subscription_manager + .remove_subscription(&self.stream) + .await?; + + if fully_removed { + if let Some(router) = &self.message_router { + router.unsubscribe(vec![self.stream.clone()]).await?; + } + } + + Ok(()) + } +} + +impl Drop for SubscriptionHandle { + fn drop(&mut self) { + if self.released { + return; + } + + // We can't do async work in Drop, so we spawn a task to handle cleanup + let stream = self.stream.clone(); + let subscription_manager = self.subscription_manager.clone(); + let message_router = self.message_router.clone(); + + tokio::spawn(async move { + let fully_removed = subscription_manager + .remove_subscription(&stream) + .await + .unwrap_or(false); + + if fully_removed { + if let Some(router) = &message_router { + let _ = router.unsubscribe(vec![stream]).await; + } + } + }); + } +} diff --git a/ccxt-exchanges/src/binance/ws/user_data.rs b/ccxt-exchanges/src/binance/ws/user_data.rs index efddd44..ae99942 100644 --- a/ccxt-exchanges/src/binance/ws/user_data.rs +++ b/ccxt-exchanges/src/binance/ws/user_data.rs @@ -371,6 +371,60 @@ pub fn parse_ws_trade(data: &Value) -> Result { }) } +/// Parses an ORDER_TRADE_UPDATE event into an Order. +/// +/// This event is sent by Binance Futures (USDT-M and COIN-M) when an order +/// is created, updated, or filled. The order data is nested under the "o" field. +/// +/// # Arguments +/// +/// * `data` - The full ORDER_TRADE_UPDATE event payload +/// +/// # Returns +/// +/// An `Order` parsed from the nested order object. +pub fn parse_order_trade_update_to_order(data: &Value) -> Result { + let order_data = data + .get("o") + .ok_or_else(|| Error::invalid_request("Missing order data 'o' in ORDER_TRADE_UPDATE"))?; + + let order_map = order_data + .as_object() + .ok_or_else(|| Error::invalid_request("ORDER_TRADE_UPDATE 'o' field is not an object"))?; + + Ok(parse_ws_order(order_map)) +} + +/// Parses an ORDER_TRADE_UPDATE event into a Trade. +/// +/// This event is sent by Binance Futures when a trade execution occurs. +/// Only produces a trade when the execution type (field "x") is "TRADE". +/// +/// # Arguments +/// +/// * `data` - The full ORDER_TRADE_UPDATE event payload +/// +/// # Returns +/// +/// A `Trade` if the execution type is TRADE, or an error otherwise. +pub fn parse_order_trade_update_to_trade(data: &Value) -> Result { + let order_data = data + .get("o") + .ok_or_else(|| Error::invalid_request("Missing order data 'o' in ORDER_TRADE_UPDATE"))?; + + // Only produce a trade when execution type is TRADE + let exec_type = order_data.get("x").and_then(|x| x.as_str()).unwrap_or(""); + + if exec_type != "TRADE" { + return Err(Error::invalid_request(format!( + "ORDER_TRADE_UPDATE execution type is '{}', not 'TRADE'", + exec_type + ))); + } + + parse_ws_trade(order_data) +} + /// Parses a WebSocket position payload pub fn parse_ws_position(data: &Value) -> Result { let symbol = data["s"] @@ -437,3 +491,539 @@ pub fn parse_ws_position(data: &Value) -> Result { datetime: Some(chrono::Utc::now().to_rfc3339()), }) } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + // ==================== parse_ws_order tests ==================== + + #[test] + fn test_parse_ws_order_spot_execution_report() { + let data = json!({ + "e": "executionReport", + "s": "BTCUSDT", + "i": 123456789, + "c": "myOrder1", + "S": "BUY", + "o": "LIMIT", + "X": "NEW", + "q": "0.5", + "p": "50000.00", + "z": "0.0", + "Z": "0.0", + "T": 1700000000000_i64, + "f": "GTC", + "P": "0.0" + }); + + let map = data.as_object().unwrap(); + let order = parse_ws_order(map); + + assert_eq!(order.symbol, "BTCUSDT"); + assert_eq!(order.id, "123456789"); + assert_eq!(order.client_order_id, Some("myOrder1".to_string())); + assert_eq!(order.side, OrderSide::Buy); + assert_eq!(order.order_type, OrderType::Limit); + assert_eq!(order.status, OrderStatus::Open); + assert_eq!(order.amount, Decimal::from_str("0.5").unwrap()); + assert_eq!(order.price, Some(Decimal::from_str("50000.00").unwrap())); + assert_eq!(order.timestamp, Some(1700000000000)); + assert_eq!(order.time_in_force, Some("GTC".to_string())); + } + + #[test] + fn test_parse_ws_order_filled() { + let data = json!({ + "e": "executionReport", + "s": "ETHUSDT", + "i": 987654321, + "S": "SELL", + "o": "MARKET", + "X": "FILLED", + "q": "10.0", + "p": "0.0", + "z": "10.0", + "Z": "30000.0", + "T": 1700000001000_i64 + }); + + let map = data.as_object().unwrap(); + let order = parse_ws_order(map); + + assert_eq!(order.symbol, "ETHUSDT"); + assert_eq!(order.side, OrderSide::Sell); + assert_eq!(order.order_type, OrderType::Market); + assert_eq!(order.status, OrderStatus::Closed); + assert_eq!(order.filled, Some(Decimal::from_str("10.0").unwrap())); + assert_eq!(order.cost, Some(Decimal::from_str("30000.0").unwrap())); + // average = cost / filled = 30000 / 10 = 3000 + assert_eq!(order.average, Some(Decimal::from_str("3000").unwrap())); + assert_eq!(order.remaining, Some(Decimal::ZERO)); + } + + #[test] + fn test_parse_ws_order_cancelled() { + let data = json!({ + "e": "executionReport", + "s": "BTCUSDT", + "i": 111, + "S": "BUY", + "o": "LIMIT", + "X": "CANCELED", + "q": "1.0", + "p": "40000.0", + "z": "0.0", + "Z": "0.0", + "T": 1700000002000_i64 + }); + + let map = data.as_object().unwrap(); + let order = parse_ws_order(map); + + assert_eq!(order.status, OrderStatus::Cancelled); + } + + #[test] + fn test_parse_ws_order_all_statuses() { + let statuses = vec![ + ("NEW", OrderStatus::Open), + ("PARTIALLY_FILLED", OrderStatus::Open), + ("FILLED", OrderStatus::Closed), + ("CANCELED", OrderStatus::Cancelled), + ("REJECTED", OrderStatus::Rejected), + ("EXPIRED", OrderStatus::Expired), + ]; + + for (binance_status, expected) in statuses { + let data = json!({ + "s": "BTCUSDT", "i": 1, "S": "BUY", "o": "LIMIT", + "X": binance_status, "q": "1.0", "p": "50000.0", + "z": "0.0", "Z": "0.0", "T": 1700000000000_i64 + }); + let map = data.as_object().unwrap(); + let order = parse_ws_order(map); + assert_eq!( + order.status, expected, + "Status mismatch for {}", + binance_status + ); + } + } + + #[test] + fn test_parse_ws_order_all_types() { + let types = vec![ + ("LIMIT", OrderType::Limit), + ("MARKET", OrderType::Market), + ("STOP_LOSS", OrderType::StopLoss), + ("STOP_LOSS_LIMIT", OrderType::StopLossLimit), + ("TAKE_PROFIT", OrderType::TakeProfit), + ("TAKE_PROFIT_LIMIT", OrderType::TakeProfitLimit), + ("LIMIT_MAKER", OrderType::LimitMaker), + ]; + + for (binance_type, expected) in types { + let data = json!({ + "s": "BTCUSDT", "i": 1, "S": "BUY", "o": binance_type, + "X": "NEW", "q": "1.0", "p": "50000.0", + "z": "0.0", "Z": "0.0", "T": 1700000000000_i64 + }); + let map = data.as_object().unwrap(); + let order = parse_ws_order(map); + assert_eq!( + order.order_type, expected, + "Type mismatch for {}", + binance_type + ); + } + } + + // ==================== parse_ws_trade tests ==================== + + #[test] + fn test_parse_ws_trade_spot() { + let data = json!({ + "e": "executionReport", + "s": "BTCUSDT", + "t": 100001, + "T": 1700000000000_i64, + "L": "50000.00", + "l": "0.1", + "Y": "5000.0", + "S": "BUY", + "o": "LIMIT", + "i": 123456, + "m": false, + "n": "0.005", + "N": "BNB" + }); + + let trade = parse_ws_trade(&data).unwrap(); + + assert_eq!(trade.symbol, "BTCUSDT"); + assert_eq!(trade.id, Some("100001".to_string())); + assert_eq!(trade.timestamp, 1700000000000); + assert_eq!( + trade.price, + ccxt_core::types::financial::Price::from(Decimal::from_str("50000.00").unwrap()) + ); + assert_eq!( + trade.amount, + ccxt_core::types::financial::Amount::from(Decimal::from_str("0.1").unwrap()) + ); + assert_eq!(trade.side, OrderSide::Buy); + assert_eq!(trade.taker_or_maker, Some(TakerOrMaker::Taker)); + assert_eq!(trade.order, Some("123456".to_string())); + + let fee = trade.fee.unwrap(); + assert_eq!(fee.cost, Decimal::from_str("0.005").unwrap()); + assert_eq!(fee.currency, "BNB"); + } + + #[test] + fn test_parse_ws_trade_maker() { + let data = json!({ + "s": "ETHUSDT", + "t": 200001, + "T": 1700000001000_i64, + "L": "3000.00", + "l": "1.0", + "S": "SELL", + "o": "LIMIT", + "i": 789, + "m": true + }); + + let trade = parse_ws_trade(&data).unwrap(); + + assert_eq!(trade.side, OrderSide::Sell); + assert_eq!(trade.taker_or_maker, Some(TakerOrMaker::Maker)); + } + + #[test] + fn test_parse_ws_trade_cost_calculated() { + // When "Y" (quote quantity) is missing, cost should be calculated as price * amount + let data = json!({ + "s": "BTCUSDT", + "T": 1700000000000_i64, + "L": "50000.00", + "l": "0.1", + "S": "BUY", + "i": 1 + }); + + let trade = parse_ws_trade(&data).unwrap(); + // cost = 50000 * 0.1 = 5000 + assert_eq!( + trade.cost, + Some(ccxt_core::types::financial::Cost::from( + Decimal::from_str("5000.0000").unwrap() + )) + ); + } + + #[test] + fn test_parse_ws_trade_missing_symbol() { + let data = json!({ + "T": 1700000000000_i64, + "L": "50000.00", + "l": "0.1" + }); + + let result = parse_ws_trade(&data); + assert!(result.is_err()); + } + + // ==================== ORDER_TRADE_UPDATE tests ==================== + + #[test] + fn test_parse_order_trade_update_to_order() { + let data = json!({ + "e": "ORDER_TRADE_UPDATE", + "T": 1700000000000_i64, + "o": { + "s": "BTCUSDT", + "i": 555666777, + "c": "futuresOrder1", + "S": "BUY", + "o": "LIMIT", + "X": "NEW", + "q": "0.01", + "p": "50000.00", + "z": "0.0", + "Z": "0.0", + "T": 1700000000000_i64, + "f": "GTC", + "ps": "LONG", + "wt": "CONTRACT_PRICE" + } + }); + + let order = parse_order_trade_update_to_order(&data).unwrap(); + + assert_eq!(order.symbol, "BTCUSDT"); + assert_eq!(order.id, "555666777"); + assert_eq!(order.client_order_id, Some("futuresOrder1".to_string())); + assert_eq!(order.side, OrderSide::Buy); + assert_eq!(order.order_type, OrderType::Limit); + assert_eq!(order.status, OrderStatus::Open); + assert_eq!(order.amount, Decimal::from_str("0.01").unwrap()); + assert_eq!(order.working_type, Some("CONTRACT_PRICE".to_string())); + } + + #[test] + fn test_parse_order_trade_update_missing_o_field() { + let data = json!({ + "e": "ORDER_TRADE_UPDATE", + "T": 1700000000000_i64 + }); + + let result = parse_order_trade_update_to_order(&data); + assert!(result.is_err()); + } + + #[test] + fn test_parse_order_trade_update_to_trade_execution() { + let data = json!({ + "e": "ORDER_TRADE_UPDATE", + "T": 1700000000000_i64, + "o": { + "s": "BTCUSDT", + "t": 300001, + "T": 1700000000000_i64, + "x": "TRADE", + "X": "PARTIALLY_FILLED", + "L": "50000.00", + "l": "0.005", + "Y": "250.0", + "S": "BUY", + "o": "LIMIT", + "i": 555666777, + "m": false, + "n": "0.001", + "N": "USDT" + } + }); + + let trade = parse_order_trade_update_to_trade(&data).unwrap(); + + assert_eq!(trade.symbol, "BTCUSDT"); + assert_eq!(trade.id, Some("300001".to_string())); + assert_eq!( + trade.price, + ccxt_core::types::financial::Price::from(Decimal::from_str("50000.00").unwrap()) + ); + assert_eq!( + trade.amount, + ccxt_core::types::financial::Amount::from(Decimal::from_str("0.005").unwrap()) + ); + assert_eq!(trade.side, OrderSide::Buy); + assert_eq!(trade.taker_or_maker, Some(TakerOrMaker::Taker)); + + let fee = trade.fee.unwrap(); + assert_eq!(fee.cost, Decimal::from_str("0.001").unwrap()); + assert_eq!(fee.currency, "USDT"); + } + + #[test] + fn test_parse_order_trade_update_to_trade_non_trade_execution() { + // When execution type is not "TRADE", should return error + let data = json!({ + "e": "ORDER_TRADE_UPDATE", + "T": 1700000000000_i64, + "o": { + "s": "BTCUSDT", + "x": "NEW", + "X": "NEW", + "S": "BUY", + "o": "LIMIT", + "i": 1 + } + }); + + let result = parse_order_trade_update_to_trade(&data); + assert!(result.is_err()); + } + + // ==================== parse_ws_position tests ==================== + + #[test] + fn test_parse_ws_position_long() { + let data = json!({ + "s": "BTCUSDT", + "pa": "0.5", + "ps": "LONG", + "ep": "50000.00", + "up": "1000.00", + "cr": "500.00", + "mt": "cross", + "iw": "5000.00" + }); + + let position = parse_ws_position(&data).unwrap(); + + assert_eq!(position.symbol, "BTCUSDT"); + assert_eq!(position.side, Some("long".to_string())); + assert_eq!(position.contracts, Some(0.5)); + assert_eq!(position.hedged, Some(true)); + assert_eq!(position.entry_price, Some(50000.00)); + assert_eq!(position.unrealized_pnl, Some(1000.00)); + assert_eq!(position.realized_pnl, Some(500.00)); + assert_eq!(position.margin_mode, Some("cross".to_string())); + assert_eq!(position.initial_margin, Some(5000.00)); + } + + #[test] + fn test_parse_ws_position_short() { + let data = json!({ + "s": "ETHUSDT", + "pa": "-10.0", + "ps": "SHORT", + "ep": "3000.00", + "up": "-200.00" + }); + + let position = parse_ws_position(&data).unwrap(); + + assert_eq!(position.symbol, "ETHUSDT"); + assert_eq!(position.side, Some("short".to_string())); + assert_eq!(position.contracts, Some(10.0)); + assert_eq!(position.hedged, Some(true)); + } + + #[test] + fn test_parse_ws_position_both_positive() { + // BOTH mode with positive amount -> long + let data = json!({ + "s": "BTCUSDT", + "pa": "1.0", + "ps": "BOTH" + }); + + let position = parse_ws_position(&data).unwrap(); + + assert_eq!(position.side, Some("long".to_string())); + assert_eq!(position.hedged, Some(false)); + } + + #[test] + fn test_parse_ws_position_both_negative() { + // BOTH mode with negative amount -> short + let data = json!({ + "s": "BTCUSDT", + "pa": "-1.0", + "ps": "BOTH" + }); + + let position = parse_ws_position(&data).unwrap(); + + assert_eq!(position.side, Some("short".to_string())); + assert_eq!(position.hedged, Some(false)); + } + + #[test] + fn test_parse_ws_position_missing_symbol() { + let data = json!({ + "pa": "1.0", + "ps": "BOTH" + }); + + let result = parse_ws_position(&data); + assert!(result.is_err()); + } + + #[test] + fn test_parse_ws_position_missing_position_amount() { + let data = json!({ + "s": "BTCUSDT", + "ps": "BOTH" + }); + + let result = parse_ws_position(&data); + assert!(result.is_err()); + } + + // ==================== handle_balance_message tests ==================== + + #[tokio::test] + async fn test_handle_balance_update() { + let balances = tokio::sync::RwLock::new(HashMap::new()); + + let msg = json!({ + "e": "balanceUpdate", + "a": "BTC", + "d": "0.5" + }); + + let result = handle_balance_message(&msg, "spot", &balances).await; + assert!(result.is_ok()); + + let balances_map = balances.read().await; + assert!(balances_map.contains_key("spot")); + } + + #[tokio::test] + async fn test_handle_outbound_account_position() { + let balances = tokio::sync::RwLock::new(HashMap::new()); + + let msg = json!({ + "e": "outboundAccountPosition", + "B": [ + {"a": "BTC", "f": "1.5", "l": "0.5"}, + {"a": "ETH", "f": "10.0", "l": "2.0"} + ] + }); + + let result = handle_balance_message(&msg, "spot", &balances).await; + assert!(result.is_ok()); + + let balances_map = balances.read().await; + assert!(balances_map.contains_key("spot")); + } + + #[tokio::test] + async fn test_handle_account_update_futures() { + let balances = tokio::sync::RwLock::new(HashMap::new()); + + let msg = json!({ + "e": "ACCOUNT_UPDATE", + "a": { + "B": [ + {"a": "USDT", "wb": "10000.00", "cw": "9500.00"} + ] + } + }); + + let result = handle_balance_message(&msg, "future", &balances).await; + assert!(result.is_ok()); + + let balances_map = balances.read().await; + assert!(balances_map.contains_key("future")); + } + + #[tokio::test] + async fn test_handle_unknown_event_type() { + let balances = tokio::sync::RwLock::new(HashMap::new()); + + let msg = json!({ + "e": "unknownEvent" + }); + + let result = handle_balance_message(&msg, "spot", &balances).await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn test_handle_missing_event_type() { + let balances = tokio::sync::RwLock::new(HashMap::new()); + + let msg = json!({ + "data": "something" + }); + + let result = handle_balance_message(&msg, "spot", &balances).await; + assert!(result.is_err()); + } +} diff --git a/ccxt-exchanges/src/binance/ws_exchange_impl.rs b/ccxt-exchanges/src/binance/ws_exchange_impl.rs index 52da4f2..83ade64 100644 --- a/ccxt-exchanges/src/binance/ws_exchange_impl.rs +++ b/ccxt-exchanges/src/binance/ws_exchange_impl.rs @@ -7,7 +7,7 @@ use async_trait::async_trait; use ccxt_core::{ error::{Error, Result}, types::{ - Balance, Ohlcv, Order, OrderBook, Ticker, Timeframe, Trade, financial::Amount, + Balance, MarketType, Ohlcv, Order, OrderBook, Ticker, Timeframe, Trade, financial::Amount, financial::Price, }, ws_client::WsConnectionState, @@ -48,8 +48,12 @@ impl WsExchange for Binance { // ==================== Connection Management ==================== async fn ws_connect(&self) -> Result<()> { - // Ensure at least one public connection is active - let _ = self.connection_manager.get_public_connection().await?; + // Ensure at least one public connection is active for the default market type + let default_market_type = MarketType::from(self.options.default_type); + let _ = self + .connection_manager + .get_public_connection(default_market_type) + .await?; Ok(()) } @@ -80,8 +84,11 @@ impl WsExchange for Binance { let binance_symbol = market.id.to_lowercase(); let stream = format!("{}@ticker", binance_symbol); - // Get shared public connection - let ws = self.connection_manager.get_public_connection().await?; + // Get shared public connection for this market type + let ws = self + .connection_manager + .get_public_connection(market.market_type) + .await?; // Create subscription channel let (tx, mut rx) = mpsc::channel(1024); @@ -141,7 +148,13 @@ impl WsExchange for Binance { markets.insert(binance_symbol.clone(), market); // Get connection (might be different for each symbol if sharding active) - let ws = self.connection_manager.get_public_connection().await?; + let Some(market_ref) = markets.get(&binance_symbol) else { + continue; + }; + let ws = self + .connection_manager + .get_public_connection(market_ref.market_type) + .await?; let (tx, mut rx) = mpsc::channel(1024); ws.subscription_manager @@ -196,6 +209,17 @@ impl WsExchange for Binance { symbol: &str, limit: Option, ) -> Result> { + // Validate depth limit if provided + const VALID_WS_DEPTH_LIMITS: &[u32] = &[5, 10, 20]; + if let Some(l) = limit { + if !VALID_WS_DEPTH_LIMITS.contains(&l) { + return Err(Error::invalid_request(format!( + "Invalid WebSocket depth limit: {}. Valid values: {:?}", + l, VALID_WS_DEPTH_LIMITS + ))); + } + } + // Load markets self.load_markets(false).await?; @@ -203,12 +227,20 @@ impl WsExchange for Binance { let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); - // Subscribe to orderbook stream - let levels = limit.unwrap_or(20); - let stream = format!("{}@depth{}@100ms", binance_symbol, levels); + // Choose stream based on limit: + // - With limit (5/10/20): use @depth{N}@100ms for partial book snapshots + // - Without limit: use @depth@100ms for incremental diff updates + let stream = if let Some(l) = limit { + format!("{}@depth{}@100ms", binance_symbol, l) + } else { + format!("{}@depth@100ms", binance_symbol) + }; - // Get shared public connection - let ws = self.connection_manager.get_public_connection().await?; + // Get shared public connection for this market type + let ws = self + .connection_manager + .get_public_connection(market.market_type) + .await?; // Create subscription channel let (tx, mut rx) = mpsc::channel(1024); @@ -259,8 +291,11 @@ impl WsExchange for Binance { let binance_symbol = market.id.to_lowercase(); let stream = format!("{}@trade", binance_symbol); - // Get shared public connection - let ws = self.connection_manager.get_public_connection().await?; + // Get shared public connection for this market type + let ws = self + .connection_manager + .get_public_connection(market.market_type) + .await?; // Create subscription channel let (tx, mut rx) = mpsc::channel(1024); @@ -319,8 +354,11 @@ impl WsExchange for Binance { let interval = timeframe.to_string(); let stream = format!("{}@kline_{}", binance_symbol, interval); - // Get shared public connection - let ws = self.connection_manager.get_public_connection().await?; + // Get shared public connection for this market type + let ws = self + .connection_manager + .get_public_connection(market.market_type) + .await?; // Create subscription channel let (tx, mut rx) = mpsc::channel(1024); @@ -387,9 +425,10 @@ impl WsExchange for Binance { .map_err(|_| Error::authentication("API credentials required for watch_balance"))?; let binance_arc = Arc::new(self.clone()); + let default_market_type = MarketType::from(self.options.default_type); let ws = self .connection_manager - .get_private_connection(&binance_arc) + .get_private_connection(default_market_type, &binance_arc) .await?; let (tx, mut rx) = mpsc::channel(1024); @@ -443,9 +482,10 @@ impl WsExchange for Binance { .map_err(|_| Error::authentication("API credentials required for watch_orders"))?; let binance_arc = Arc::new(self.clone()); + let default_market_type = MarketType::from(self.options.default_type); let ws = self .connection_manager - .get_private_connection(&binance_arc) + .get_private_connection(default_market_type, &binance_arc) .await?; let (tx, mut rx) = mpsc::channel(1024); @@ -467,9 +507,17 @@ impl WsExchange for Binance { while let Some(msg) = rx.recv().await { if let Some(data) = msg.as_object() { if let Some(event_type) = data.get("e").and_then(|e| e.as_str()) { - if event_type == "executionReport" { - let order = super::ws::user_data::parse_ws_order(data); + let order = match event_type { + // Spot market order updates + "executionReport" => Some(super::ws::user_data::parse_ws_order(data)), + // Futures market order updates + "ORDER_TRADE_UPDATE" => { + super::ws::user_data::parse_order_trade_update_to_order(&msg).ok() + } + _ => None, + }; + if let Some(order) = order { { let mut orders = orders_cache.write().await; let symbol_orders = orders @@ -504,9 +552,10 @@ impl WsExchange for Binance { .map_err(|_| Error::authentication("API credentials required for watch_my_trades"))?; let binance_arc = Arc::new(self.clone()); + let default_market_type = MarketType::from(self.options.default_type); let ws = self .connection_manager - .get_private_connection(&binance_arc) + .get_private_connection(default_market_type, &binance_arc) .await?; let (tx, mut rx) = mpsc::channel(1024); @@ -527,31 +576,39 @@ impl WsExchange for Binance { tokio::spawn(async move { while let Some(msg) = rx.recv().await { if let Some(event_type) = msg.get("e").and_then(|e| e.as_str()) { - if event_type == "executionReport" { - if let Ok(trade) = super::ws::user_data::parse_ws_trade(&msg) { - { - let mut trades = trades_cache.write().await; - - let symbol_trades = trades - .entry(trade.symbol.clone()) - .or_insert_with(std::collections::VecDeque::new); - symbol_trades.push_front(trade.clone()); - if symbol_trades.len() > 1000 { - symbol_trades.pop_back(); - } - } + let trade_result = match event_type { + // Spot market trade executions + "executionReport" => super::ws::user_data::parse_ws_trade(&msg).ok(), + // Futures market trade executions + "ORDER_TRADE_UPDATE" => { + super::ws::user_data::parse_order_trade_update_to_trade(&msg).ok() + } + _ => None, + }; - // Filter and send - if let Some(s) = &symbol_filter { - if &trade.symbol != s { - continue; - } + if let Some(trade) = trade_result { + { + let mut trades = trades_cache.write().await; + + let symbol_trades = trades + .entry(trade.symbol.clone()) + .or_insert_with(std::collections::VecDeque::new); + symbol_trades.push_front(trade.clone()); + if symbol_trades.len() > 1000 { + symbol_trades.pop_back(); } + } - if user_tx.send(Ok(trade)).await.is_err() { - break; + // Filter and send + if let Some(s) = &symbol_filter { + if &trade.symbol != s { + continue; } } + + if user_tx.send(Ok(trade)).await.is_err() { + break; + } } } } @@ -564,8 +621,19 @@ impl WsExchange for Binance { // ==================== Subscription Management ==================== async fn subscribe(&self, channel: &str, symbol: Option<&str>) -> Result<()> { + // Determine market type from symbol or use default + let market_type = if let Some(sym) = symbol { + let market = self.base.market(sym).await?; + market.market_type + } else { + MarketType::from(self.options.default_type) + }; + // Create WebSocket and connect - let ws = self.connection_manager.get_public_connection().await?; + let ws = self + .connection_manager + .get_public_connection(market_type) + .await?; // Use the appropriate subscription method based on channel // Note: The receiver is intentionally dropped here as this is a fire-and-forget subscription. @@ -615,8 +683,19 @@ impl WsExchange for Binance { channel.to_string() }; + // Determine market type from symbol or use default + let market_type = if let Some(sym) = symbol { + let market = self.base.market(sym).await?; + market.market_type + } else { + MarketType::from(self.options.default_type) + }; + // Create WS to unsubscribe - let ws = self.connection_manager.get_public_connection().await?; + let ws = self + .connection_manager + .get_public_connection(market_type) + .await?; ws.unsubscribe(stream_name).await }