Skip to content

refactor(binance): update WebSocket API to return receivers and impro… #46

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Praying merged 1 commit into from
Jan 31, 2026
Merged

refactor(binance): update WebSocket API to return receivers and impro… #46

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
4 changes: 2 additions & 2 deletions ccxt-core/src/types/orderbook.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,8 @@ use super::{Amount, Price, Symbol, Timestamp};
/// Maximum number of buffered messages for orderbook sync
const MAX_BUFFERED_MESSAGES: usize = 100;

/// Minimum resync interval in milliseconds
const MIN_RESYNC_INTERVAL_MS: i64 = 1000;
/// Minimum resync interval in milliseconds (5 seconds)
const MIN_RESYNC_INTERVAL_MS: i64 = 5000;

/// Orderbook delta update message
#[derive(Debug, Clone, Serialize, Deserialize)]
Expand Down
76 changes: 52 additions & 24 deletions ccxt-exchanges/src/binance/ws/binance_impl.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,31 +2,59 @@
// It's separated to keep mod.rs under 800 lines

impl Binance {
/// Subscribes to the ticker stream for a unified symbol
pub async fn subscribe_ticker(&self, symbol: &str) -> Result<()> {
/// Subscribes to the ticker stream for a unified symbol.
///
/// Returns a receiver that will receive ticker updates as JSON values.
pub async fn subscribe_ticker(
&self,
symbol: &str,
) -> Result<tokio::sync::mpsc::Receiver<serde_json::Value>> {
let ws = self.connection_manager.get_public_connection().await?;
let binance_symbol = symbol.replace('/', "").to_lowercase();
ws.subscribe_ticker(&binance_symbol).await
}

/// Subscribes to the trade stream for a unified symbol
pub async fn subscribe_trades(&self, symbol: &str) -> Result<()> {
/// Subscribes to the trade stream for a unified symbol.
///
/// Returns a receiver that will receive trade updates as JSON values.
pub async fn subscribe_trades(
&self,
symbol: &str,
) -> Result<tokio::sync::mpsc::Receiver<serde_json::Value>> {
let ws = self.connection_manager.get_public_connection().await?;
let binance_symbol = symbol.replace('/', "").to_lowercase();
ws.subscribe_trades(&binance_symbol).await
}

/// Subscribes to the order book stream for a unified symbol
pub async fn subscribe_orderbook(&self, symbol: &str, levels: Option<u32>) -> Result<()> {
/// Subscribes to the order book stream for a unified symbol.
///
/// Returns a receiver that will receive order book updates as JSON values.
pub async fn subscribe_orderbook(
&self,
symbol: &str,
levels: Option<u32>,
) -> Result<tokio::sync::mpsc::Receiver<serde_json::Value>> {
use super::ws::{DepthLevel, UpdateSpeed};

let ws = self.connection_manager.get_public_connection().await?;
let binance_symbol = symbol.replace('/', "").to_lowercase();
let depth_levels = levels.unwrap_or(20);
ws.subscribe_orderbook(&binance_symbol, depth_levels, "1000ms")
let depth_level = match levels.unwrap_or(20) {
5 => DepthLevel::L5,
10 => DepthLevel::L10,
_ => DepthLevel::L20,
};
ws.subscribe_orderbook(&binance_symbol, depth_level, UpdateSpeed::Ms1000)
.await
}

/// Subscribes to the candlestick stream for a unified symbol
pub async fn subscribe_kline(&self, symbol: &str, interval: &str) -> Result<()> {
/// Subscribes to the candlestick stream for a unified symbol.
///
/// Returns a receiver that will receive kline updates as JSON values.
pub async fn subscribe_kline(
&self,
symbol: &str,
interval: &str,
) -> Result<tokio::sync::mpsc::Receiver<serde_json::Value>> {
let ws = self.connection_manager.get_public_connection().await?;
let binance_symbol = symbol.replace('/', "").to_lowercase();
ws.subscribe_kline(&binance_symbol, interval).await
Expand Down Expand Up @@ -137,6 +165,8 @@ impl Binance {
limit: Option<i64>,
params: Option<HashMap<String, Value>>,
) -> Result<OrderBook> {
use super::ws::UpdateSpeed;

self.load_markets(false).await?;

let market = self.base.market(symbol).await?;
Expand All @@ -146,19 +176,14 @@ impl Binance {
market.market_type == MarketType::Swap || market.market_type == MarketType::Futures;

let update_speed = if let Some(p) = &params {
p.get("speed")
.and_then(serde_json::Value::as_i64)
.unwrap_or(100) as i32
match p.get("speed").and_then(serde_json::Value::as_i64).unwrap_or(100) {
1000 => UpdateSpeed::Ms1000,
_ => UpdateSpeed::Ms100,
}
} else {
100
UpdateSpeed::Ms100
};

if update_speed != 100 && update_speed != 1000 {
return Err(Error::invalid_request(
"Update speed must be 100 or 1000 milliseconds",
));
}

let ws = self.connection_manager.get_public_connection().await?;
ws.watch_orderbook_internal(self, &binance_symbol, limit, update_speed, is_futures)
.await
Expand All @@ -171,6 +196,8 @@ impl Binance {
limit: Option<i64>,
params: Option<HashMap<String, Value>>,
) -> Result<HashMap<String, OrderBook>> {
use super::ws::UpdateSpeed;

if symbols.is_empty() {
return Err(Error::invalid_request("Symbols list cannot be empty"));
}
Expand Down Expand Up @@ -201,11 +228,12 @@ impl Binance {
}

let update_speed = if let Some(p) = &params {
p.get("speed")
.and_then(serde_json::Value::as_i64)
.unwrap_or(100) as i32
match p.get("speed").and_then(serde_json::Value::as_i64).unwrap_or(100) {
1000 => UpdateSpeed::Ms1000,
_ => UpdateSpeed::Ms100,
}
} else {
100
UpdateSpeed::Ms100
};

let ws = self.connection_manager.get_public_connection().await?;
Expand Down
53 changes: 31 additions & 22 deletions ccxt-exchanges/src/binance/ws/handlers.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,31 @@ impl MessageRouter {
self.is_connected.load(std::sync::atomic::Ordering::SeqCst)
}

/// Returns the current connection latency in milliseconds.
///
/// This is calculated from the ping/pong round-trip time.
/// Returns None if no latency data is available.
pub fn latency(&self) -> Option<i64> {
// Try to get latency from the underlying WsClient
// This requires a blocking read, so we use try_read
if let Ok(guard) = self.ws_client.try_read() {
if let Some(ref client) = *guard {
return client.latency();
}
}
None
}

/// Returns the number of reconnection attempts.
pub fn reconnect_count(&self) -> u32 {
if let Ok(guard) = self.ws_client.try_read() {
if let Some(ref client) = *guard {
return client.reconnect_count();
}
}
0
}

/// Applies a new reconnection configuration
pub async fn set_reconnect_config(&self, config: super::subscriptions::ReconnectConfig) {
*self.reconnect_config.write().await = config;
Expand Down Expand Up @@ -416,36 +441,20 @@ impl MessageRouter {
let symbol_opt = payload.get("s").and_then(|s| s.as_str());

if let Some(symbol) = symbol_opt {
let mut active_streams = subscription_manager
.get_subscriptions_by_symbol(symbol)
// Normalize symbol to lowercase since subscriptions are stored normalized
let normalized_symbol = symbol.to_lowercase();
let active_streams = subscription_manager
.get_subscriptions_by_symbol(&normalized_symbol)
.await;

tracing::debug!(
"Routing message for symbol {}: stream_name={}, active_subscriptions={}",
"Routing message for symbol {} (normalized: {}): stream_name={}, active_subscriptions={}",
symbol,
normalized_symbol,
stream_name,
active_streams.len()
);

// If no streams found for uppercase symbol (e.g. "BTCUSDT"), try lowercase (e.g. "btcusdt")
// This is common since Binance API sends uppercase symbols but we usually subscribe with lowercase
if active_streams.is_empty() {
let lower_symbol = symbol.to_lowercase();
if lower_symbol != symbol {
let lower_streams = subscription_manager
.get_subscriptions_by_symbol(&lower_symbol)
.await;
if !lower_streams.is_empty() {
active_streams = lower_streams;
tracing::debug!(
"Found subscriptions for lowercased symbol {}: count={}",
lower_symbol,
active_streams.len()
);
}
}
}

let mut fallback_sent = false;

for sub in active_streams {
Expand Down
89 changes: 63 additions & 26 deletions ccxt-exchanges/src/binance/ws/listen_key.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,24 @@ use std::time::{Duration, Instant};
use tokio::sync::{Mutex, RwLock};
use tokio::task::JoinHandle;

/// Listen key refresh interval (30 minutes)
const LISTEN_KEY_REFRESH_INTERVAL: Duration = Duration::from_secs(30 * 60);
/// Listen key refresh interval (25 minutes)
///
/// Binance listen keys expire after 60 minutes. We refresh at 25 minutes
/// to provide a safety margin and avoid expiration during network delays.
const LISTEN_KEY_REFRESH_INTERVAL: Duration = Duration::from_secs(25 * 60);

/// Maximum age before forcing a new listen key (20 minutes)
///
/// If a listen key is older than this, we create a new one instead of
/// returning the cached key. This is more conservative than the refresh
/// interval to handle cases where auto-refresh might have failed.
const LISTEN_KEY_MAX_AGE: Duration = Duration::from_secs(20 * 60);

/// Maximum number of retry attempts for refresh failures
const MAX_REFRESH_RETRIES: u32 = 3;

/// Delay between refresh retry attempts
const REFRESH_RETRY_DELAY: Duration = Duration::from_secs(5);

/// Listen key manager
///
Expand Down Expand Up @@ -50,8 +66,13 @@ impl ListenKeyManager {
let created = self.created_at.read().await;
if let Some(created_time) = *created {
let elapsed = created_time.elapsed();
if elapsed > Duration::from_secs(50 * 60) {
// Use the more conservative max age threshold
if elapsed > LISTEN_KEY_MAX_AGE {
drop(created);
tracing::info!(
"Listen key is {} minutes old, creating new one",
elapsed.as_secs() / 60
);
return self.create_new().await;
}
}
Expand Down Expand Up @@ -105,33 +126,49 @@ impl ListenKeyManager {

let key_opt = listen_key.read().await.clone();
if let Some(key) = key_opt {
match binance.refresh_listen_key(&key).await {
Ok(()) => {
*created_at.write().await = Some(Instant::now());
}
Err(e) => {
tracing::warn!(
"Failed to refresh listen key: {}. Attempting regeneration.",
e
);
match binance.create_listen_key().await {
Ok(new_key) => {
tracing::info!("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
);
*listen_key.write().await = None;
*created_at.write().await = None;
break;
// Try to refresh with retries
let mut refresh_succeeded = false;
for attempt in 1..=MAX_REFRESH_RETRIES {
match binance.refresh_listen_key(&key).await {
Ok(()) => {
*created_at.write().await = Some(Instant::now());
tracing::debug!("Listen key refreshed successfully");
refresh_succeeded = true;
break;
}
Err(e) => {
tracing::warn!(
"Failed to refresh listen key (attempt {}/{}): {}",
attempt,
MAX_REFRESH_RETRIES,
e
);
if attempt < MAX_REFRESH_RETRIES {
tokio::time::sleep(REFRESH_RETRY_DELAY).await;
}
}
}
}

// If all refresh attempts failed, try to regenerate
if !refresh_succeeded {
tracing::warn!(
"All refresh attempts failed, attempting to regenerate listen key"
);
match binance.create_listen_key().await {
Ok(new_key) => {
tracing::info!("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);
*listen_key.write().await = None;
*created_at.write().await = None;
break;
}
}
}
} else {
break;
}
Expand Down
Loading