Dev (#45)

* fix: watch mark price

* fix(binance): correct markPrice stream name and enhance futures parsing

- Fixed markPrice stream subscription from "@markprice" to "@markprice" to match Binance API
- Added futures contract expiry timestamp parsing from deliveryDate field
- Enhanced symbol parsing to include expiry suffix for futures contracts
- Updated unified symbol format to include expiry information for futures
- Improved mark price subscription logic with proper channel management
- Added timestamp to datetime conversion for futures expiry dates

Author: Praying

ccxt-core/src/base_exchange/mod.rs

@@ -93,10 +93,10 @@ impl BaseExchange {
                     strategy_type: crate::retry_strategy::RetryStrategyType::Fixed,
                     ..crate::retry_strategy::RetryConfig::default()
                 }),
-            max_response_size: 10 * 1024 * 1024, // 10MB default
-            max_request_size: 10 * 1024 * 1024,  // 10MB default
-            circuit_breaker: None,               // Disabled by default for backward compatibility
-            pool_max_idle_per_host: 10,          // Default: 10 idle connections per host
+            max_response_size: 128 * 1024 * 1024, // 128MB default
+            max_request_size: 10 * 1024 * 1024,   // 10MB default
+            circuit_breaker: None,                // Disabled by default for backward compatibility
+            pool_max_idle_per_host: 10,           // Default: 10 idle connections per host
             pool_idle_timeout: Duration::from_secs(90), // Default: 90 seconds
         };
 

ccxt-core/src/http_client/config.rs

@@ -78,8 +78,8 @@ impl Default for HttpConfig {
             proxy: None,
             enable_rate_limit: true,
             retry_config: None,
-            max_response_size: 10 * 1024 * 1024,
-            max_request_size: 10 * 1024 * 1024,
+            max_response_size: 128 * 1024 * 1024, // 128MB default
+            max_request_size: 10 * 1024 * 1024,   // 10MB default
             circuit_breaker: None,
             pool_max_idle_per_host: 10,
             pool_idle_timeout: Duration::from_secs(90),

ccxt-core/src/ws_client/config.rs

@@ -156,6 +156,14 @@ pub struct WsConfig {
     ///
     /// Default: `DropOldest` to ensure most recent data is available.
     pub backpressure_strategy: BackpressureStrategy,
+    /// Maximum message size in bytes.
+    ///
+    /// Default: 128 MiB (134,217,728 bytes) if None.
+    pub max_message_size: Option<usize>,
+    /// Maximum frame size in bytes.
+    ///
+    /// Default: 32 MiB (33,554,432 bytes) if None.
+    pub max_frame_size: Option<usize>,
 }
 
 impl Default for WsConfig {
@@ -175,6 +183,8 @@ impl Default for WsConfig {
             message_channel_capacity: DEFAULT_MESSAGE_CHANNEL_CAPACITY,
             write_channel_capacity: DEFAULT_WRITE_CHANNEL_CAPACITY,
             backpressure_strategy: BackpressureStrategy::default(),
+            max_message_size: Some(128 * 1024 * 1024),
+            max_frame_size: Some(32 * 1024 * 1024),
         }
     }
 }

ccxt-core/src/ws_client/mod.rs

@@ -34,7 +34,8 @@ use tokio::net::TcpStream;
 use tokio::sync::{Mutex, RwLock, mpsc};
 use tokio::time::{Duration, interval};
 use tokio_tungstenite::{
-    MaybeTlsStream, WebSocketStream, connect_async, tungstenite::protocol::Message,
+    MaybeTlsStream, WebSocketStream, connect_async_with_config,
+    tungstenite::protocol::{Message, WebSocketConfig},
 };
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, instrument, warn};
@@ -154,9 +155,17 @@ impl WsClient {
         let url = self.config.url.clone();
         info!("Initiating WebSocket connection");
 
+        let mut ws_config = WebSocketConfig::default();
+        if let Some(limit) = self.config.max_message_size {
+            ws_config.max_message_size = Some(limit);
+        }
+        if let Some(limit) = self.config.max_frame_size {
+            ws_config.max_frame_size = Some(limit);
+        }
+
         match tokio::time::timeout(
             Duration::from_millis(self.config.connect_timeout),
-            connect_async(&url),
+            connect_async_with_config(&url, Some(ws_config), false),
         )
         .await
         {
@@ -214,6 +223,14 @@ impl WsClient {
         self.set_state(WsConnectionState::Connecting);
         let url = self.config.url.clone();
 
+        let mut ws_config = WebSocketConfig::default();
+        if let Some(limit) = self.config.max_message_size {
+            ws_config.max_message_size = Some(limit);
+        }
+        if let Some(limit) = self.config.max_frame_size {
+            ws_config.max_frame_size = Some(limit);
+        }
+
         tokio::select! {
             biased;
             () = token.cancelled() => {
@@ -223,7 +240,7 @@ impl WsClient {
             }
             result = tokio::time::timeout(
                 Duration::from_millis(self.config.connect_timeout),
-                connect_async(&url),
+                connect_async_with_config(&url, Some(ws_config), false),
             ) => {
                 match result {
                     Ok(Ok((ws_stream, response))) => {

ccxt-exchanges/src/binance/parser/market.rs

@@ -2,6 +2,7 @@ use super::value_to_hashmap;
 use ccxt_core::{
     Result,
     error::{Error, ParseError},
+    parser_utils::timestamp_to_datetime,
     types::{Market, MarketLimits, MarketPrecision, MarketType, MinMax},
 };
 use rust_decimal::Decimal;
@@ -32,6 +33,61 @@ pub fn parse_market(data: &Value) -> Result<Market> {
     let active = status == "TRADING";
     let margin = data["isMarginTradingAllowed"].as_bool().unwrap_or(false);
 
+    // Check for contract type to determine if this is a futures/swap market
+    let contract_type = data["contractType"].as_str();
+    let is_contract = contract_type.is_some();
+
+    let market_type = if let Some(ct) = contract_type {
+        if ct == "PERPETUAL" {
+            MarketType::Swap
+        } else {
+            MarketType::Futures
+        }
+    } else {
+        MarketType::Spot
+    };
+
+    let mut linear = None;
+    let mut inverse = None;
+    let mut settle_id = None;
+    let mut settle = None;
+
+    if is_contract {
+        if let Some(margin_asset) = data["marginAsset"].as_str() {
+            settle_id = Some(margin_asset.to_string());
+            settle = Some(margin_asset.to_string()); // In Binance, asset ID is usually the code
+
+            if margin_asset == quote_asset {
+                linear = Some(true);
+                inverse = Some(false);
+            } else if margin_asset == base_asset {
+                linear = Some(false);
+                inverse = Some(true);
+            }
+        }
+    }
+
+    let expiry_timestamp = if market_type == MarketType::Futures {
+        data["deliveryDate"].as_i64().filter(|ts| *ts > 0)
+    } else {
+        None
+    };
+
+    let expiry_datetime = expiry_timestamp.and_then(timestamp_to_datetime);
+
+    let expiry_suffix = if market_type == MarketType::Futures {
+        symbol.rfind('_').and_then(|pos| {
+            let suffix = &symbol[pos + 1..];
+            if suffix.len() == 6 && suffix.chars().all(|c| c.is_ascii_digit()) {
+                Some(suffix.to_string())
+            } else {
+                None
+            }
+        })
+    } else {
+        None
+    };
+
     let mut price_precision: Option<Decimal> = None;
     let mut amount_precision: Option<Decimal> = None;
     let mut min_amount: Option<Decimal> = None;
@@ -81,7 +137,24 @@ pub fn parse_market(data: &Value) -> Result<Market> {
         }
     }
 
-    let unified_symbol = format!("{}/{}", base_asset, quote_asset);
+    let unified_symbol = if is_contract {
+        if let Some(s) = &settle {
+            if market_type == MarketType::Futures {
+                if let Some(expiry) = &expiry_suffix {
+                    format!("{}/{}:{}-{}", base_asset, quote_asset, s, expiry)
+                } else {
+                    format!("{}/{}:{}", base_asset, quote_asset, s)
+                }
+            } else {
+                format!("{}/{}:{}", base_asset, quote_asset, s)
+            }
+        } else {
+            format!("{}/{}", base_asset, quote_asset)
+        }
+    } else {
+        format!("{}/{}", base_asset, quote_asset)
+    };
+
     let parsed_symbol = ccxt_core::symbol::SymbolParser::parse(&unified_symbol).ok();
 
     Ok(Market {
@@ -90,21 +163,21 @@ pub fn parse_market(data: &Value) -> Result<Market> {
         parsed_symbol,
         base: base_asset.clone(),
         quote: quote_asset.clone(),
-        settle: None,
         base_id: Some(base_asset),
         quote_id: Some(quote_asset),
-        settle_id: None,
-        market_type: MarketType::Spot,
+        settle_id,
+        settle,
+        market_type,
         active,
         margin,
-        contract: Some(false),
-        linear: None,
-        inverse: None,
+        contract: Some(is_contract),
+        linear,
+        inverse,
         taker: Decimal::from_str("0.001").ok(),
         maker: Decimal::from_str("0.001").ok(),
         contract_size: None,
-        expiry: None,
-        expiry_datetime: None,
+        expiry: expiry_timestamp,
+        expiry_datetime,
         strike: None,
         option_type: None,
         percentage: Some(true),

ccxt-exchanges/src/binance/parser/mod.rs

@@ -12,7 +12,10 @@ pub use balance::{parse_balance, parse_balance_with_type};
 pub use funding::{parse_funding_rate, parse_funding_rate_history};
 pub use margin::{parse_margin_adjustment, parse_margin_loan};
 pub use market::{parse_currencies, parse_market};
-pub use misc::{parse_mark_prices, parse_stats_24hr, parse_trading_fee, parse_trading_limits};
+pub use misc::{
+    parse_mark_prices, parse_stats_24hr, parse_trading_fee, parse_trading_limits,
+    parse_ws_mark_price,
+};
 pub use ohlcv::{parse_ohlcvs, parse_ws_ohlcv};
 pub use order::parse_order;
 pub use orderbook::{parse_orderbook, parse_ws_orderbook};