From 8111c36dda4ae6923a6591fda1933cbe0b735896 Mon Sep 17 00:00:00 2001 From: quran Date: Sun, 25 Jan 2026 21:46:36 +0800 Subject: [PATCH] refactor(exchanges): update test configurations and enhance ws client functionality - Remove unused Decimal import from binance advanced market data tests - Add deprecated attribute allowances to OHLCV test functions across multiple test files - Remove unused create_test_binance function from binance my trades test - Remove unused SecretString imports from property tests - Add optional fields to Market struct initialization in property tests - Replace direct market cache access with getter methods in bitget property tests - Update deprecated method calls to use new getter methods in bitget property tests - Add #[allow(deprecated)] to various test functions to suppress warnings - Remove unused HyperLiquid import from hyperliquid integration test - Conditionally apply deprecated attribute to exchange module outside test builds - Add custom message builder functionality to ws client for flexible subscription handling - Implement Binance-specific message builders with atomic request ID tracking - Extract Binance subscribe/unsubscribe payload construction into dedicated functions - Add unit tests for Binance payload construction functions - Add deprecated attribute allowance to timestamp test functions - Remove unused Amount import from OKX/Bybit shared property test --- ccxt-core/src/ws_client/mod.rs | 57 ++++++++++-- .../src/binance/signing_strategy.rs | 2 +- ccxt-exchanges/src/binance/ws/handlers.rs | 14 +-- ccxt-exchanges/src/binance/ws/mod.rs | 86 ++++++++++++++++++- ccxt-exchanges/src/bitget/rest/mod.rs | 1 + ccxt-exchanges/src/bybit/rest.rs | 1 + ccxt-exchanges/src/lib.rs | 9 +- ccxt-exchanges/src/okx/rest/mod.rs | 1 + .../binance_advanced_market_data_test.rs | 4 +- .../tests/binance_conditional_orders_test.rs | 2 + .../tests/binance_integration_test.rs | 2 + .../tests/binance_my_trades_test.rs | 12 --- .../tests/bitget_integration_test.rs | 1 + ccxt-exchanges/tests/bitget_property_test.rs | 21 +++-- .../tests/bybit_integration_test.rs | 1 + .../tests/hyperliquid_integration_test.rs | 2 +- .../tests/okx_bybit_shared_property_test.rs | 9 +- ccxt-exchanges/tests/okx_integration_test.rs | 1 + 18 files changed, 176 insertions(+), 50 deletions(-) diff --git a/ccxt-core/src/ws_client/mod.rs b/ccxt-core/src/ws_client/mod.rs index fd2a068..7551861 100644 --- a/ccxt-core/src/ws_client/mod.rs +++ b/ccxt-core/src/ws_client/mod.rs @@ -43,6 +43,15 @@ use tracing::{debug, error, info, instrument, warn}; #[allow(dead_code)] type WsWriter = SplitSink>, Message>; +/// Builder for custom subscribe messages. +pub type SubscribeMessageBuilder = Arc< + dyn Fn(String, Option, Option>) -> Result + Send + Sync, +>; + +/// Builder for custom unsubscribe messages. +pub type UnsubscribeMessageBuilder = + Arc) -> Result + Send + Sync>; + /// Async WebSocket client for exchange streaming APIs. /// /// # Backpressure Handling @@ -68,6 +77,10 @@ pub struct WsClient { cancel_token: Arc>>, #[debug(skip)] event_callback: Arc>>, + #[debug(skip)] + subscribe_message_builder: Option, + #[debug(skip)] + unsubscribe_message_builder: Option, /// Counter for dropped messages due to backpressure dropped_messages: Arc, } @@ -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, params: Option>, ) -> 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, ) -> 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 } diff --git a/ccxt-exchanges/src/binance/signing_strategy.rs b/ccxt-exchanges/src/binance/signing_strategy.rs index cf6be49..d706bba 100644 --- a/ccxt-exchanges/src/binance/signing_strategy.rs +++ b/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); } diff --git a/ccxt-exchanges/src/binance/ws/handlers.rs b/ccxt-exchanges/src/binance/ws/handlers.rs index 45b6aac..955645f 100644 --- a/ccxt-exchanges/src/binance/ws/handlers.rs +++ b/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( diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index eb9715e..024756d 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/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, + _params: Option>| { + 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| { + 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, id: u64) -> Value { + serde_json::json!({ + "method": "SUBSCRIBE", + "params": streams, + "id": id + }) +} + +#[allow(clippy::disallowed_methods)] +fn build_binance_unsubscribe_payload(streams: Vec, 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(); diff --git a/ccxt-exchanges/src/bitget/rest/mod.rs b/ccxt-exchanges/src/bitget/rest/mod.rs index 5874251..5c34ea3 100644 --- a/ccxt-exchanges/src/bitget/rest/mod.rs +++ b/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(); diff --git a/ccxt-exchanges/src/bybit/rest.rs b/ccxt-exchanges/src/bybit/rest.rs index 2366ea7..161db06 100644 --- a/ccxt-exchanges/src/bybit/rest.rs +++ b/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(); diff --git a/ccxt-exchanges/src/lib.rs b/ccxt-exchanges/src/lib.rs index 99378aa..ebdbafd 100644 --- a/ccxt-exchanges/src/lib.rs +++ b/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; diff --git a/ccxt-exchanges/src/okx/rest/mod.rs b/ccxt-exchanges/src/okx/rest/mod.rs index aef892c..0cd24a6 100644 --- a/ccxt-exchanges/src/okx/rest/mod.rs +++ b/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(); diff --git a/ccxt-exchanges/tests/binance_advanced_market_data_test.rs b/ccxt-exchanges/tests/binance_advanced_market_data_test.rs index 728a82e..60a4d8e 100644 --- a/ccxt-exchanges/tests/binance_advanced_market_data_test.rs +++ b/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() { diff --git a/ccxt-exchanges/tests/binance_conditional_orders_test.rs b/ccxt-exchanges/tests/binance_conditional_orders_test.rs index e7fb2fb..3c9b8eb 100644 --- a/ccxt-exchanges/tests/binance_conditional_orders_test.rs +++ b/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; diff --git a/ccxt-exchanges/tests/binance_integration_test.rs b/ccxt-exchanges/tests/binance_integration_test.rs index 6b91355..1ba6234 100644 --- a/ccxt-exchanges/tests/binance_integration_test.rs +++ b/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() { diff --git a/ccxt-exchanges/tests/binance_my_trades_test.rs b/ccxt-exchanges/tests/binance_my_trades_test.rs index e385e22..bbebce7 100644 --- a/ccxt-exchanges/tests/binance_my_trades_test.rs +++ b/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> { - 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<()> { diff --git a/ccxt-exchanges/tests/bitget_integration_test.rs b/ccxt-exchanges/tests/bitget_integration_test.rs index 1d74fcf..67e3c14 100644 --- a/ccxt-exchanges/tests/bitget_integration_test.rs +++ b/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() { diff --git a/ccxt-exchanges/tests/bitget_property_test.rs b/ccxt-exchanges/tests/bitget_property_test.rs index ab3ea6b..6e2e985 100644 --- a/ccxt-exchanges/tests/bitget_property_test.rs +++ b/ccxt-exchanges/tests/bitget_property_test.rs @@ -2,7 +2,6 @@ //! //! These tests verify correctness properties using proptest. -use ccxt_core::SecretString; use proptest::prelude::*; /// **Feature: bitget-exchange, Property 1: Builder Configuration Preservation** @@ -1369,6 +1368,10 @@ mod data_roundtrip_consistency { average: None, base_volume: None, quote_volume: None, + funding_rate: None, + open_interest: None, + index_price: None, + mark_price: None, info: std::collections::HashMap::new(), }; @@ -1668,11 +1671,11 @@ mod market_cache_consistency { // Verify markets_by_id is consistent with markets for market in &markets { // Check markets index - let by_symbol = cache.markets.get(&market.symbol); + let by_symbol = cache.get_market(&market.symbol); assert!(by_symbol.is_some(), "Market should be indexed by symbol"); // Check markets_by_id index - let by_id = cache.markets_by_id.get(&market.id); + let by_id = cache.get_market_by_id(&market.id); assert!(by_id.is_some(), "Market should be indexed by ID"); // Both should point to the same market data @@ -1699,10 +1702,12 @@ mod market_cache_consistency { // Read from cache let cache = bitget.base().market_cache.read().await; + let symbols = cache.symbols(); + // Verify symbols list contains all market symbols for market in &markets { assert!( - cache.symbols.contains(&market.symbol), + symbols.contains(&market.symbol), "Symbols list should contain {}", market.symbol ); @@ -1714,7 +1719,7 @@ mod market_cache_consistency { .map(|m| m.symbol.clone()) .collect(); assert_eq!( - cache.symbols.len(), + symbols.len(), unique_symbols.len(), "Symbols list length should match unique market count" ); @@ -1737,7 +1742,7 @@ mod market_cache_consistency { // Verify first batch is loaded { let cache = bitget.base().market_cache.read().await; - assert!(cache.loaded, "Cache should be loaded after first set"); + assert!(cache.is_loaded(), "Cache should be loaded after first set"); } // Set second batch of markets (simulating reload) @@ -1754,7 +1759,7 @@ mod market_cache_consistency { for symbol in &unique_symbols2 { assert!( - cache.markets.contains_key(symbol), + cache.contains_market(symbol), "New market {} should be in cache", symbol ); @@ -1768,7 +1773,7 @@ mod market_cache_consistency { for symbol in &unique_symbols1 { if !unique_symbols2.contains(symbol) { assert!( - !cache.markets.contains_key(symbol), + !cache.contains_market(symbol), "Old market {} should NOT be in cache after reload", symbol ); diff --git a/ccxt-exchanges/tests/bybit_integration_test.rs b/ccxt-exchanges/tests/bybit_integration_test.rs index 782788f..b9cee8d 100644 --- a/ccxt-exchanges/tests/bybit_integration_test.rs +++ b/ccxt-exchanges/tests/bybit_integration_test.rs @@ -258,6 +258,7 @@ async fn test_fetch_trades() { } /// Test fetching OHLCV (candlestick) data from the real API. +#[allow(deprecated)] #[tokio::test] #[ignore] // Requires network access async fn test_fetch_ohlcv() { diff --git a/ccxt-exchanges/tests/hyperliquid_integration_test.rs b/ccxt-exchanges/tests/hyperliquid_integration_test.rs index e9ddbf2..3a23434 100644 --- a/ccxt-exchanges/tests/hyperliquid_integration_test.rs +++ b/ccxt-exchanges/tests/hyperliquid_integration_test.rs @@ -4,7 +4,7 @@ //! Run with: cargo test --test hyperliquid_integration_test -- --ignored use ccxt_core::exchange::Exchange; -use ccxt_exchanges::hyperliquid::{HyperLiquid, HyperLiquidBuilder}; +use ccxt_exchanges::hyperliquid::HyperLiquidBuilder; /// Test creating a HyperLiquid instance without authentication. #[test] diff --git a/ccxt-exchanges/tests/okx_bybit_shared_property_test.rs b/ccxt-exchanges/tests/okx_bybit_shared_property_test.rs index 24fe4c1..1d5b981 100644 --- a/ccxt-exchanges/tests/okx_bybit_shared_property_test.rs +++ b/ccxt-exchanges/tests/okx_bybit_shared_property_test.rs @@ -3,7 +3,6 @@ //! These tests verify correctness properties that apply to both exchanges using proptest. use base64::Engine; -use ccxt_core::SecretString; use proptest::prelude::*; use rust_decimal::Decimal; use rust_decimal::prelude::FromStr; @@ -1558,7 +1557,7 @@ mod order_status_mapping { /// SHALL produce an equivalent Market struct. mod market_data_roundtrip { use super::*; - use ccxt_core::types::{Market, MarketLimits, MarketPrecision, MarketType, MinMax}; + use ccxt_core::types::{Market, MarketLimits, MarketPrecision, MarketType}; use rust_decimal::prelude::FromPrimitive; // Strategy for generating valid market types @@ -1658,7 +1657,7 @@ mod market_data_roundtrip { mod ticker_data_roundtrip { use super::*; use ccxt_core::types::Ticker; - use ccxt_core::types::financial::{Amount, Price}; + use ccxt_core::types::financial::Price; use rust_decimal::prelude::FromPrimitive; // Strategy for generating valid symbols @@ -1711,6 +1710,10 @@ mod ticker_data_roundtrip { average: None, base_volume: None, quote_volume: None, + funding_rate: None, + open_interest: None, + index_price: None, + mark_price: None, info: std::collections::HashMap::new(), }; diff --git a/ccxt-exchanges/tests/okx_integration_test.rs b/ccxt-exchanges/tests/okx_integration_test.rs index dc46dfe..f4d72cd 100644 --- a/ccxt-exchanges/tests/okx_integration_test.rs +++ b/ccxt-exchanges/tests/okx_integration_test.rs @@ -231,6 +231,7 @@ async fn test_fetch_trades() { } /// Test fetching OHLCV (candlestick) data from the real API. +#[allow(deprecated)] #[tokio::test] #[ignore] // Requires network access async fn test_fetch_ohlcv() {