Skip to content
Merged

Dev #45

Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions ccxt-core/src/base_exchange/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
};

Expand Down
4 changes: 2 additions & 2 deletions ccxt-core/src/http_client/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
10 changes: 10 additions & 0 deletions ccxt-core/src/ws_client/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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),
}
}
}
23 changes: 20 additions & 3 deletions ccxt-core/src/ws_client/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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
{
Expand Down Expand Up @@ -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() => {
Expand All @@ -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))) => {
Expand Down
91 changes: 82 additions & 9 deletions ccxt-exchanges/src/binance/parser/market.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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 {
Expand All @@ -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),
Expand Down
5 changes: 4 additions & 1 deletion ccxt-exchanges/src/binance/parser/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down
Loading