refactor(exchanges): update test configurations and enhance ws client…

ccxt-core/src/ws_client/mod.rs

@@ -43,6 +43,15 @@ use tracing::{debug, error, info, instrument, warn};
 #[allow(dead_code)]
 type WsWriter = SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>;
 
+/// Builder for custom subscribe messages.
+pub type SubscribeMessageBuilder = Arc<
+    dyn Fn(String, Option<String>, Option<HashMap<String, Value>>) -> Result<Value> + Send + Sync,
+>;
+
+/// Builder for custom unsubscribe messages.
+pub type UnsubscribeMessageBuilder =
+    Arc<dyn Fn(String, Option<String>) -> Result<Value> + Send + Sync>;
+
 /// Async WebSocket client for exchange streaming APIs.
 ///
 /// # Backpressure Handling
@@ -68,6 +77,10 @@ pub struct WsClient {
     cancel_token: Arc<Mutex<Option<CancellationToken>>>,
     #[debug(skip)]
     event_callback: Arc<Mutex<Option<WsEventCallback>>>,
+    #[debug(skip)]
+    subscribe_message_builder: Option<SubscribeMessageBuilder>,
+    #[debug(skip)]
+    unsubscribe_message_builder: Option<UnsubscribeMessageBuilder>,
     /// Counter for dropped messages due to backpressure
     dropped_messages: Arc<AtomicU32>,
 }
@@ -93,6 +106,8 @@ impl WsClient {
             stats: Arc::new(WsStats::new()),
             cancel_token: Arc::new(Mutex::new(None)),
             event_callback: Arc::new(Mutex::new(None)),
+            subscribe_message_builder: None,
+            unsubscribe_message_builder: None,
             dropped_messages: Arc::new(AtomicU32::new(0)),
         }
     }
@@ -109,6 +124,26 @@ impl WsClient {
         debug!("Event callback cleared");
     }
 
+    /// Sets a custom builder for subscribe messages.
+    pub fn set_subscribe_message_builder(&mut self, builder: SubscribeMessageBuilder) {
+        self.subscribe_message_builder = Some(builder);
+    }
+
+    /// Clears the custom subscribe message builder.
+    pub fn clear_subscribe_message_builder(&mut self) {
+        self.subscribe_message_builder = None;
+    }
+
+    /// Sets a custom builder for unsubscribe messages.
+    pub fn set_unsubscribe_message_builder(&mut self, builder: UnsubscribeMessageBuilder) {
+        self.unsubscribe_message_builder = Some(builder);
+    }
+
+    /// Clears the custom unsubscribe message builder.
+    pub fn clear_unsubscribe_message_builder(&mut self) {
+        self.unsubscribe_message_builder = None;
+    }
+
     async fn emit_event(&self, event: WsEvent) {
         let callback = self.event_callback.lock().await;
         if let Some(ref cb) = *callback {
@@ -823,12 +858,16 @@ impl WsClient {
         symbol: Option<String>,
         params: Option<HashMap<String, Value>>,
     ) -> Result<()> {
-        let msg = WsMessage::Subscribe {
-            channel,
-            symbol,
-            params,
+        let json = if let Some(builder) = self.subscribe_message_builder.as_ref() {
+            builder(channel, symbol, params)?
+        } else {
+            let msg = WsMessage::Subscribe {
+                channel,
+                symbol,
+                params,
+            };
+            serde_json::to_value(&msg).map_err(Error::from)?
         };
-        let json = serde_json::to_value(&msg).map_err(Error::from)?;
         self.send_json(&json).await
     }
 
@@ -837,8 +876,12 @@ impl WsClient {
         channel: String,
         symbol: Option<String>,
     ) -> Result<()> {
-        let msg = WsMessage::Unsubscribe { channel, symbol };
-        let json = serde_json::to_value(&msg).map_err(Error::from)?;
+        let json = if let Some(builder) = self.unsubscribe_message_builder.as_ref() {
+            builder(channel, symbol)?
+        } else {
+            let msg = WsMessage::Unsubscribe { channel, symbol };
+            serde_json::to_value(&msg).map_err(Error::from)?
+        };
         self.send_json(&json).await
     }
 

ccxt-exchanges/src/binance/signing_strategy.rs

@@ -94,7 +94,7 @@ mod tests {
     fn test_signing_strategy_creation() {
         let config = ExchangeConfig::default();
         let binance = Arc::new(Binance::new(config).unwrap());
-        let strategy = BinanceSigningStrategy::new(binance);
+        let _strategy = BinanceSigningStrategy::new(binance);
         // Strategy created successfully
         assert!(true);
     }

ccxt-exchanges/src/binance/ws/handlers.rs

@@ -152,12 +152,7 @@ impl MessageRouter {
             .request_id
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
 
-        #[allow(clippy::disallowed_methods)]
-        let request = serde_json::json!({
-            "method": "SUBSCRIBE",
-            "params": streams,
-            "id": id
-        });
+        let request = super::build_binance_subscribe_payload(streams, id);
 
         client
             .send(tokio_tungstenite::tungstenite::protocol::Message::Text(
@@ -183,12 +178,7 @@ impl MessageRouter {
             .request_id
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
 
-        #[allow(clippy::disallowed_methods)]
-        let request = serde_json::json!({
-            "method": "UNSUBSCRIBE",
-            "params": streams,
-            "id": id
-        });
+        let request = super::build_binance_unsubscribe_payload(streams, id);
 
         client
             .send(tokio_tungstenite::tungstenite::protocol::Message::Text(

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

@@ -23,6 +23,7 @@ use ccxt_core::ws_client::{WsClient, WsConfig};
 use serde_json::Value;
 use std::collections::{HashMap, VecDeque};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 use tokio::sync::{Mutex, RwLock};
 
@@ -55,6 +56,33 @@ impl std::fmt::Debug for BinanceWs {
 }
 
 impl BinanceWs {
+    fn configure_client(client: &mut WsClient) {
+        let request_id = Arc::new(AtomicU64::new(1));
+
+        let subscribe_builder = {
+            let request_id = Arc::clone(&request_id);
+            Arc::new(
+                move |channel: String,
+                      _symbol: Option<String>,
+                      _params: Option<HashMap<String, Value>>| {
+                    let id = request_id.fetch_add(1, Ordering::SeqCst);
+                    Ok(build_binance_subscribe_payload(vec![channel], id))
+                },
+            )
+        };
+
+        let unsubscribe_builder = {
+            let request_id = Arc::clone(&request_id);
+            Arc::new(move |channel: String, _symbol: Option<String>| {
+                let id = request_id.fetch_add(1, Ordering::SeqCst);
+                Ok(build_binance_unsubscribe_payload(vec![channel], id))
+            })
+        };
+
+        client.set_subscribe_message_builder(subscribe_builder);
+        client.set_unsubscribe_message_builder(unsubscribe_builder);
+    }
+
     /// Creates a new Binance WebSocket client
     pub fn new(url: String) -> Self {
         let config = WsConfig {
@@ -69,8 +97,11 @@ impl BinanceWs {
             ..Default::default()
         };
 
+        let mut client = WsClient::new(config);
+        Self::configure_client(&mut client);
+
         Self {
-            client: Arc::new(WsClient::new(config)),
+            client: Arc::new(client),
             listen_key: Arc::new(RwLock::new(None)),
             listen_key_manager: None,
             auto_reconnect_coordinator: Arc::new(Mutex::new(None)),
@@ -101,8 +132,11 @@ impl BinanceWs {
             ..Default::default()
         };
 
+        let mut client = WsClient::new(config);
+        Self::configure_client(&mut client);
+
         Self {
-            client: Arc::new(WsClient::new(config)),
+            client: Arc::new(client),
             listen_key: Arc::new(RwLock::new(None)),
             listen_key_manager: Some(Arc::new(ListenKeyManager::new(binance))),
             auto_reconnect_coordinator: Arc::new(Mutex::new(None)),
@@ -713,6 +747,24 @@ impl BinanceWs {
 // Include Binance impl methods in a separate file to keep mod.rs manageable
 include!("binance_impl.rs");
 
+#[allow(clippy::disallowed_methods)]
+fn build_binance_subscribe_payload(streams: Vec<String>, id: u64) -> Value {
+    serde_json::json!({
+        "method": "SUBSCRIBE",
+        "params": streams,
+        "id": id
+    })
+}
+
+#[allow(clippy::disallowed_methods)]
+fn build_binance_unsubscribe_payload(streams: Vec<String>, id: u64) -> Value {
+    serde_json::json!({
+        "method": "UNSUBSCRIBE",
+        "params": streams,
+        "id": id
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -740,6 +792,36 @@ mod tests {
         assert_eq!(kline_stream, "btcusdt@kline_1m");
     }
 
+    #[test]
+    fn test_binance_subscribe_payload() {
+        let payload =
+            build_binance_subscribe_payload(vec!["btcusdt@fundingRate".to_string()], 12345);
+
+        #[allow(clippy::disallowed_methods)]
+        let expected = serde_json::json!({
+            "method": "SUBSCRIBE",
+            "params": ["btcusdt@fundingRate"],
+            "id": 12345
+        });
+
+        assert_eq!(payload, expected);
+    }
+
+    #[test]
+    fn test_binance_unsubscribe_payload() {
+        let payload =
+            build_binance_unsubscribe_payload(vec!["btcusdt@fundingRate".to_string()], 67890);
+
+        #[allow(clippy::disallowed_methods)]
+        let expected = serde_json::json!({
+            "method": "UNSUBSCRIBE",
+            "params": ["btcusdt@fundingRate"],
+            "id": 67890
+        });
+
+        assert_eq!(payload, expected);
+    }
+
     #[tokio::test]
     async fn test_subscription_manager_basic() {
         let manager = SubscriptionManager::new();

ccxt-exchanges/src/bitget/rest/mod.rs

@@ -269,6 +269,7 @@ mod tests {
     }
 
     #[test]
+    #[allow(deprecated)]
     fn test_get_timestamp() {
         let _bitget = Bitget::builder().build().unwrap();
         let ts = Bitget::get_timestamp();

ccxt-exchanges/src/bybit/rest.rs

@@ -1285,6 +1285,7 @@ mod tests {
     }
 
     #[test]
+    #[allow(deprecated)]
     fn test_get_timestamp() {
         let _bybit = Bybit::builder().build().unwrap();
         let ts = Bybit::get_timestamp();

ccxt-exchanges/src/lib.rs

@@ -82,9 +82,12 @@ pub use ccxt_core;
 ///
 /// This module is deprecated. Use `ccxt_core::Exchange` and
 /// `ccxt_core::ExchangeCapabilities` directly instead.
-#[deprecated(
-    since = "0.2.0",
-    note = "Use `ccxt_core::Exchange` and `ccxt_core::ExchangeCapabilities` directly instead."
+#[cfg_attr(
+    not(test),
+    deprecated(
+        since = "0.2.0",
+        note = "Use `ccxt_core::Exchange` and `ccxt_core::ExchangeCapabilities` directly instead."
+    )
 )]
 pub mod exchange;
 

ccxt-exchanges/src/okx/rest/mod.rs

@@ -306,6 +306,7 @@ mod tests {
     }
 
     #[test]
+    #[allow(deprecated)]
     fn test_get_timestamp() {
         let _okx = Okx::builder().build().unwrap();
         let ts = Okx::get_timestamp();

ccxt-exchanges/tests/binance_advanced_market_data_test.rs

@@ -7,7 +7,6 @@
 
 use ccxt_core::ExchangeConfig;
 use ccxt_exchanges::binance::Binance;
-use rust_decimal::Decimal;
 use rust_decimal_macros::dec;
 
 /// Create Binance client for testing.
@@ -44,6 +43,7 @@ fn create_authenticated_binance_client() -> Binance {
 mod ohlcv_tests {
     use super::*;
 
+    #[allow(deprecated)]
     #[tokio::test]
     #[ignore] // Requires network connection
     async fn test_fetch_ohlcv_basic() {
@@ -76,6 +76,7 @@ mod ohlcv_tests {
         assert!(first.low <= first.close, "Low should be <= close");
     }
 
+    #[allow(deprecated)]
     #[tokio::test]
     #[ignore] // Requires network connection
     async fn test_fetch_ohlcv_different_timeframes() {
@@ -98,6 +99,7 @@ mod ohlcv_tests {
         }
     }
 
+    #[allow(deprecated)]
     #[tokio::test]
     #[ignore] // Requires network connection
     async fn test_fetch_ohlcv_with_since() {

ccxt-exchanges/tests/binance_conditional_orders_test.rs

@@ -1,3 +1,5 @@
+#![allow(deprecated)]
+
 use ccxt_core::types::financial::{Amount, Price};
 use ccxt_core::{ExchangeConfig, Order, OrderSide, OrderStatus, OrderType};
 use ccxt_exchanges::binance::Binance;

ccxt-exchanges/tests/binance_integration_test.rs

@@ -157,6 +157,7 @@ async fn test_fetch_trades_real() {
 }
 
 /// Test fetching OHLCV (candlestick) data from the real API.
+#[allow(deprecated)]
 #[tokio::test]
 #[ignore]
 async fn test_fetch_ohlcv_real() {
@@ -195,6 +196,7 @@ async fn test_invalid_symbol() {
 }
 
 /// Test handling of unsupported timeframe (2h is not supported by Binance).
+#[allow(deprecated)]
 #[tokio::test]
 #[ignore]
 async fn test_unsupported_timeframe() {

ccxt-exchanges/tests/binance_my_trades_test.rs

@@ -17,18 +17,6 @@ mod tests {
     use super::*;
     use std::collections::HashMap;
 
-    /// Create a Binance instance for testing.
-    fn create_test_binance() -> Result<Arc<Binance>> {
-        let config = ExchangeConfig {
-            id: "binance".to_string(),
-            name: "Binance".to_string(),
-            api_key: Some(ccxt_core::SecretString::new("test_api_key")),
-            secret: Some(ccxt_core::SecretString::new("test_api_secret")),
-            ..Default::default()
-        };
-        Ok(Arc::new(Binance::new(config)?))
-    }
-
     #[tokio::test]
     #[ignore]
     async fn test_watch_my_trades_all_symbols() -> Result<()> {

ccxt-exchanges/tests/bitget_integration_test.rs

@@ -267,6 +267,7 @@ async fn test_fetch_trades() {
 
 /// Test fetching OHLCV (candlestick) data from real API.
 /// _Requirements: 5.1_
+#[allow(deprecated)]
 #[tokio::test]
 #[ignore] // Requires network access
 async fn test_fetch_ohlcv() {