feat(exchange): add Binance balance parsing and WebSocket implementation

ccxt-core/src/ws_client/config.rs

@@ -0,0 +1,135 @@
+//! WebSocket configuration types.
+
+use rand::Rng;
+use std::time::Duration;
+
+/// Exponential backoff configuration for reconnection.
+#[derive(Debug, Clone)]
+pub struct BackoffConfig {
+    /// Base delay for first retry (default: 1 second)
+    pub base_delay: Duration,
+    /// Maximum delay cap (default: 60 seconds)
+    pub max_delay: Duration,
+    /// Jitter factor (0.0 - 1.0, default: 0.25 for 25%)
+    pub jitter_factor: f64,
+    /// Multiplier for exponential growth (default: 2.0)
+    pub multiplier: f64,
+}
+
+impl Default for BackoffConfig {
+    fn default() -> Self {
+        Self {
+            base_delay: Duration::from_secs(1),
+            max_delay: Duration::from_secs(60),
+            jitter_factor: 0.25,
+            multiplier: 2.0,
+        }
+    }
+}
+
+/// Calculates retry delay with exponential backoff and jitter.
+#[derive(Debug, Clone)]
+pub struct BackoffStrategy {
+    config: BackoffConfig,
+}
+
+impl BackoffStrategy {
+    /// Creates a new backoff strategy with the given configuration.
+    pub fn new(config: BackoffConfig) -> Self {
+        Self { config }
+    }
+
+    /// Creates a new backoff strategy with default configuration.
+    pub fn with_defaults() -> Self {
+        Self::new(BackoffConfig::default())
+    }
+
+    /// Returns a reference to the underlying configuration.
+    pub fn config(&self) -> &BackoffConfig {
+        &self.config
+    }
+
+    /// Calculates delay for the given attempt number.
+    #[allow(clippy::cast_precision_loss, clippy::cast_possible_truncation)]
+    pub fn calculate_delay(&self, attempt: u32) -> Duration {
+        let base_ms = self.config.base_delay.as_millis() as f64;
+        let multiplier = self.config.multiplier;
+        let max_ms = self.config.max_delay.as_millis() as f64;
+
+        let exponential_delay_ms = base_ms * multiplier.powi(attempt as i32);
+        let capped_delay_ms = exponential_delay_ms.min(max_ms);
+
+        let jitter_ms = if self.config.jitter_factor > 0.0 {
+            let jitter_range = capped_delay_ms * self.config.jitter_factor;
+            rand::rng().random::<f64>() * jitter_range
+        } else {
+            0.0
+        };
+
+        Duration::from_millis((capped_delay_ms + jitter_ms) as u64)
+    }
+
+    /// Calculates the base delay (without jitter) for the given attempt number.
+    #[allow(clippy::cast_precision_loss, clippy::cast_possible_truncation)]
+    pub fn calculate_delay_without_jitter(&self, attempt: u32) -> Duration {
+        let base_ms = self.config.base_delay.as_millis() as f64;
+        let multiplier = self.config.multiplier;
+        let max_ms = self.config.max_delay.as_millis() as f64;
+
+        let exponential_delay_ms = base_ms * multiplier.powi(attempt as i32);
+        let capped_delay_ms = exponential_delay_ms.min(max_ms);
+
+        Duration::from_millis(capped_delay_ms as u64)
+    }
+}
+
+/// Default maximum number of subscriptions.
+pub const DEFAULT_MAX_SUBSCRIPTIONS: usize = 100;
+
+/// Default shutdown timeout in milliseconds.
+pub const DEFAULT_SHUTDOWN_TIMEOUT: u64 = 5000;
+
+/// WebSocket connection configuration.
+#[derive(Debug, Clone)]
+pub struct WsConfig {
+    /// WebSocket server URL
+    pub url: String,
+    /// Connection timeout in milliseconds
+    pub connect_timeout: u64,
+    /// Ping interval in milliseconds
+    pub ping_interval: u64,
+    /// Reconnection delay in milliseconds (legacy, use `backoff_config` for exponential backoff)
+    pub reconnect_interval: u64,
+    /// Maximum reconnection attempts before giving up
+    pub max_reconnect_attempts: u32,
+    /// Enable automatic reconnection on disconnect
+    pub auto_reconnect: bool,
+    /// Enable message compression
+    pub enable_compression: bool,
+    /// Pong timeout in milliseconds
+    pub pong_timeout: u64,
+    /// Exponential backoff configuration for reconnection.
+    pub backoff_config: BackoffConfig,
+    /// Maximum number of subscriptions allowed.
+    pub max_subscriptions: usize,
+    /// Shutdown timeout in milliseconds.
+    pub shutdown_timeout: u64,
+}
+
+impl Default for WsConfig {
+    fn default() -> Self {
+        Self {
+            url: String::new(),
+            connect_timeout: 10000,
+            ping_interval: 30000,
+            reconnect_interval: 5000,
+            max_reconnect_attempts: 5,
+            auto_reconnect: true,
+            enable_compression: false,
+            pong_timeout: 90000,
+            backoff_config: BackoffConfig::default(),
+            max_subscriptions: DEFAULT_MAX_SUBSCRIPTIONS,
+            shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT,
+        }
+    }
+}

ccxt-core/src/ws_client/error.rs

@@ -0,0 +1,204 @@
+//! WebSocket error types and classification.
+
+use crate::error::Error;
+
+/// WebSocket error classification.
+///
+/// This enum categorizes WebSocket errors into two types:
+/// - `Transient`: Temporary errors that may recover with retry (network issues, server unavailable)
+/// - `Permanent`: Errors that should not be retried (authentication failures, protocol errors)
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum WsErrorKind {
+    /// Transient errors that may recover with retry.
+    Transient,
+    /// Permanent errors that should not be retried.
+    Permanent,
+}
+
+impl WsErrorKind {
+    /// Returns `true` if this is a transient error that may recover with retry.
+    #[inline]
+    #[must_use]
+    pub fn is_transient(self) -> bool {
+        matches!(self, Self::Transient)
+    }
+
+    /// Returns `true` if this is a permanent error that should not be retried.
+    #[inline]
+    #[must_use]
+    pub fn is_permanent(self) -> bool {
+        matches!(self, Self::Permanent)
+    }
+}
+
+impl std::fmt::Display for WsErrorKind {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Transient => write!(f, "Transient"),
+            Self::Permanent => write!(f, "Permanent"),
+        }
+    }
+}
+
+/// Extended WebSocket error with classification.
+#[derive(Debug)]
+pub struct WsError {
+    kind: WsErrorKind,
+    message: String,
+    source: Option<Box<dyn std::error::Error + Send + Sync + 'static>>,
+}
+
+impl WsError {
+    /// Creates a new `WsError` with the specified kind and message.
+    pub fn new(kind: WsErrorKind, message: impl Into<String>) -> Self {
+        Self {
+            kind,
+            message: message.into(),
+            source: None,
+        }
+    }
+
+    /// Creates a new `WsError` with a source error.
+    pub fn with_source<E>(kind: WsErrorKind, message: impl Into<String>, source: E) -> Self
+    where
+        E: std::error::Error + Send + Sync + 'static,
+    {
+        Self {
+            kind,
+            message: message.into(),
+            source: Some(Box::new(source)),
+        }
+    }
+
+    /// Creates a transient error.
+    pub fn transient(message: impl Into<String>) -> Self {
+        Self::new(WsErrorKind::Transient, message)
+    }
+
+    /// Creates a transient error with a source.
+    pub fn transient_with_source<E>(message: impl Into<String>, source: E) -> Self
+    where
+        E: std::error::Error + Send + Sync + 'static,
+    {
+        Self::with_source(WsErrorKind::Transient, message, source)
+    }
+
+    /// Creates a permanent error.
+    pub fn permanent(message: impl Into<String>) -> Self {
+        Self::new(WsErrorKind::Permanent, message)
+    }
+
+    /// Creates a permanent error with a source.
+    pub fn permanent_with_source<E>(message: impl Into<String>, source: E) -> Self
+    where
+        E: std::error::Error + Send + Sync + 'static,
+    {
+        Self::with_source(WsErrorKind::Permanent, message, source)
+    }
+
+    /// Returns the error kind.
+    #[inline]
+    #[must_use]
+    pub fn kind(&self) -> WsErrorKind {
+        self.kind
+    }
+
+    /// Returns the error message.
+    #[inline]
+    #[must_use]
+    pub fn message(&self) -> &str {
+        &self.message
+    }
+
+    /// Returns `true` if this is a transient error.
+    #[inline]
+    #[must_use]
+    pub fn is_transient(&self) -> bool {
+        self.kind.is_transient()
+    }
+
+    /// Returns `true` if this is a permanent error.
+    #[inline]
+    #[must_use]
+    pub fn is_permanent(&self) -> bool {
+        self.kind.is_permanent()
+    }
+
+    /// Returns the source error, if any.
+    #[must_use]
+    pub fn source(&self) -> Option<&(dyn std::error::Error + Send + Sync + 'static)> {
+        self.source.as_deref()
+    }
+
+    /// Classifies a tungstenite WebSocket error.
+    pub fn from_tungstenite(err: &tokio_tungstenite::tungstenite::Error) -> Self {
+        use tokio_tungstenite::tungstenite::Error as TungError;
+
+        match err {
+            TungError::Io(io_err) => {
+                let message = format!("IO error: {io_err}");
+                Self::transient_with_source(
+                    message,
+                    std::io::Error::new(io_err.kind(), io_err.to_string()),
+                )
+            }
+            TungError::ConnectionClosed => Self::transient("Connection closed by server"),
+            TungError::AlreadyClosed => Self::transient("Connection already closed"),
+            TungError::Protocol(protocol_err) => {
+                Self::permanent(format!("Protocol error: {protocol_err}"))
+            }
+            TungError::Utf8(_) => Self::permanent("UTF-8 encoding error in WebSocket message"),
+            TungError::Http(response) => {
+                let status = response.status();
+                let status_code = status.as_u16();
+                if status_code == 401 || status_code == 403 {
+                    Self::permanent(format!("Authentication error: HTTP {status}"))
+                } else if status.is_server_error() {
+                    Self::transient(format!("Server error: HTTP {status}"))
+                } else {
+                    Self::permanent(format!("HTTP error: {status}"))
+                }
+            }
+            TungError::HttpFormat(http_err) => {
+                Self::permanent(format!("HTTP format error: {http_err}"))
+            }
+            TungError::Url(url_err) => Self::permanent(format!("Invalid URL: {url_err}")),
+            TungError::Tls(tls_err) => Self::transient(format!("TLS error: {tls_err}")),
+            TungError::Capacity(capacity_err) => {
+                Self::permanent(format!("Capacity error: {capacity_err}"))
+            }
+            TungError::WriteBufferFull(msg) => {
+                Self::transient(format!("Write buffer full: {msg:?}"))
+            }
+            TungError::AttackAttempt => Self::permanent("Potential attack detected"),
+        }
+    }
+
+    /// Classifies a generic error and wraps it in a `WsError`.
+    pub fn from_error(err: &Error) -> Self {
+        if err.as_authentication().is_some() {
+            return Self::permanent(format!("Authentication error: {err}"));
+        }
+        if err.as_cancelled().is_some() {
+            return Self::permanent(format!("Operation cancelled: {err}"));
+        }
+        if err.as_resource_exhausted().is_some() {
+            return Self::permanent(format!("Resource exhausted: {err}"));
+        }
+        Self::transient(format!("Error: {err}"))
+    }
+}
+
+impl std::fmt::Display for WsError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "[{}] {}", self.kind, self.message)
+    }
+}
+
+impl std::error::Error for WsError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        self.source
+            .as_ref()
+            .map(|e| e.as_ref() as &(dyn std::error::Error + 'static))
+    }
+}