rusty_ems/exchanges/

binance_websocket_trading.rs

//! Binance WebSocket Trading Implementation
//!
//! This module provides a high-performance WebSocket trading client for Binance exchange
//! with robust connection management, automatic reconnection, and comprehensive health monitoring.
//!
//! # Features
//!
//! - **Dual-level ping/pong mechanism**: Both WebSocket protocol-level and JSON API-level
//! - **Automatic reconnection**: Configurable backoff strategy with exponential delays
//! - **Connection health monitoring**: Real-time metrics and health status
//! - **Ed25519 authentication**: Support for both raw and DER-encoded private keys
//! - **Zero-copy message processing**: Optimized for low-latency trading
//!
//! # Connection Management
//!
//! The client maintains connection health through:
//! - WebSocket ping frames sent every 30 seconds
//! - JSON ping requests for API-level health checks
//! - Automatic reconnection on connection loss
//! - Configurable pong timeout (default: 10 seconds)
//!
//! # Reconnection Behavior
//!
//! When connection is lost, the client will:
//! 1. Detect timeout through missing pong responses
//! 2. Kill the existing socket connection
//! 3. Wait for backoff period (starts at 1s, doubles each attempt)
//! 4. Attempt to reconnect and re-authenticate
//! 5. Reset backoff on successful connection
//!
//! Configuration:
//! - Initial backoff: 1 second
//! - Max backoff: 60 seconds
//! - Max attempts: 10 (configurable via `MAX_RECONNECTION_ATTEMPTS`)
//!
//! # Usage Example
//!
//! ```rust,no_run
//! use rusty_common::auth::exchanges::binance::BinanceAuth;
//! use rusty_ems::exchanges::binance_websocket_trading::BinanceWebSocketTrader;
//! use std::sync::Arc;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
//!     let auth = Arc::new(BinanceAuth::new_ed25519(
//!         "api_key".into(),
//!         "private_key_base64".into()
//!     )?);
//!
//!     let trader = BinanceWebSocketTrader::new(auth);
//!     let (report_tx, report_rx) = flume::bounded(100);
//!
//!     // Connect and authenticate
//!     trader.connect(report_tx).await?;
//!
//!     // Check connection health
//!     let health = trader.get_connection_health();
//!     println!("Connection healthy: {}", health.is_healthy);
//!
//!     // Send JSON ping
//!     trader.send_ping().await?;
//!
//!     Ok(())
//! }
//! ```

use rusty_common::collections::FxHashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering};

use anyhow::{Result, anyhow, bail};
use async_trait::async_trait;
use flume::Sender;
use futures::{SinkExt, StreamExt};
use log::{debug, error, info, warn};
use parking_lot::RwLock;
use quanta::Clock;
use rand::Rng;
use rust_decimal::Decimal;
use rusty_common::SmartString;
use rusty_common::auth::exchanges::binance::BinanceAuth;
use rusty_common::id_generation;
use rusty_common::websocket::connector::{WebSocketSink, WebSocketStream};
use rusty_common::websocket::{Message, WebSocketConfig, WebSocketConnector};
use rusty_model::{
    enums::{OrderSide, OrderStatus, OrderType, TimeInForce},
    instruments::InstrumentId,
    position::{MarginType, PositionSide, PositionUpdate},
    trading_order::Order,
    venues::Venue,
};
use simd_json::prelude::*;
use simd_json::value::owned::{Object, Value as JsonValue};
use smallvec::SmallVec;
use std::time::Duration;
use tokio::sync::RwLock as AsyncRwLock;
use tokio::task::JoinHandle;
use tokio::time::interval;
use uuid::Uuid;
use yawc::frame::OpCode;

use crate::error::batch_errors::{BatchResult, BatchStatus, OrderResult, OrderResultMap};
use crate::execution_engine::ExecutionReport;
use crate::position_manager::PositionManager;

/// WebSocket API URLs for different environments
const BINANCE_WS_API_URL: &str = "wss://ws-api.binance.com/ws-api/v3";
const BINANCE_WS_API_TESTNET_URL: &str = "wss://ws-api.testnet.binance.vision/ws-api/v3";

/// Ping interval for keeping WebSocket connection alive (30 seconds)
const PING_INTERVAL_SECONDS: u64 = 30;

/// Pong timeout (10 seconds - if no pong received within this time, consider connection dead)
const PONG_TIMEOUT_SECONDS: u64 = 10;

/// Maximum message size allowed (10MB)
const MAX_MESSAGE_SIZE: usize = 10 * 1024 * 1024;

/// Maximum pending requests before cleanup
const MAX_PENDING_REQUESTS: usize = 10000;

/// Rate limiting constants for batch orders
const MAX_ORDERS_PER_10_SECONDS: u32 = 300;
const RATE_LIMIT_WINDOW_MS: u64 = 10_000; // 10 seconds in milliseconds
const MAX_BATCH_SIZE: usize = 50; // Maximum orders in a single batch (Binance WebSocket rate limit is 300/10s)

/// Order rate limiter for batch orders
struct OrderRateLimiter {
    order_times: std::collections::VecDeque<u64>,
    clock: Clock,
}

impl OrderRateLimiter {
    const fn new(clock: Clock) -> Self {
        Self {
            order_times: std::collections::VecDeque::new(),
            clock,
        }
    }

    /// Remove expired timestamps from the queue
    fn cleanup_expired(&mut self) {
        let now = self.clock.raw() / 1_000_000; // Convert to milliseconds
        let window_start = now.saturating_sub(RATE_LIMIT_WINDOW_MS);

        while let Some(&front) = self.order_times.front() {
            if front < window_start {
                self.order_times.pop_front();
            } else {
                break;
            }
        }
    }

    fn can_place_orders(&mut self, count: usize) -> bool {
        self.cleanup_expired();
        // Check if we can place the requested number of orders
        self.order_times.len() + count <= MAX_ORDERS_PER_10_SECONDS as usize
    }

    fn record_orders(&mut self, count: usize) {
        let now = self.clock.raw() / 1_000_000; // Convert to milliseconds
        for _ in 0..count {
            self.order_times.push_back(now);
        }
    }

    fn get_current_usage(&mut self) -> (usize, usize) {
        self.cleanup_expired();
        (self.order_times.len(), MAX_ORDERS_PER_10_SECONDS as usize)
    }

    fn current_order_count(&mut self) -> usize {
        self.cleanup_expired();
        self.order_times.len()
    }
}

/// Connection state for proper state machine tracking
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum ConnectionState {
    /// Not connected to the WebSocket
    Disconnected = 0,
    /// Establishing WebSocket connection
    Connecting = 1,
    /// WebSocket connection established but not authenticated
    Connected = 2,
    /// Authentication process in progress
    Authenticating = 3,
    /// WebSocket connected and authenticated for trading
    Authenticated = 4,
    /// Disconnection process in progress
    Disconnecting = 5,
}

impl From<u8> for ConnectionState {
    fn from(value: u8) -> Self {
        match value {
            0 => Self::Disconnected,
            1 => Self::Connecting,
            2 => Self::Connected,
            3 => Self::Authenticating,
            4 => Self::Authenticated,
            5 => Self::Disconnecting,
            _ => Self::Disconnected,
        }
    }
}

/// Request ID generator with overflow protection
#[derive(Debug, Clone)]
enum RequestId {
    Sequential(u64),
    Uuid(SmartString),
}

impl RequestId {
    fn as_json_value(&self) -> JsonValue {
        match self {
            Self::Sequential(id) => JsonValue::from(*id),
            Self::Uuid(id) => JsonValue::from(id.as_str()),
        }
    }

    /// Create a lookup key without string allocation for sequential IDs
    fn to_lookup_key(&self) -> RequestKey {
        match self {
            Self::Sequential(id) => RequestKey::Sequential(*id),
            Self::Uuid(id) => RequestKey::Uuid(id.clone()),
        }
    }
}

/// Optimized key type for pending requests lookup
/// Avoids string allocations for numeric IDs in hot paths
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum RequestKey {
    Sequential(u64),
    Uuid(SmartString),
}

impl From<&RequestId> for RequestKey {
    fn from(req_id: &RequestId) -> Self {
        req_id.to_lookup_key()
    }
}

/// Request ID generator for WebSocket API requests
#[derive(Debug)]
struct RequestIdGenerator {
    counter: AtomicU64,
    overflow_detected: AtomicBool,
}

impl RequestIdGenerator {
    const fn new() -> Self {
        Self {
            counter: AtomicU64::new(1),
            overflow_detected: AtomicBool::new(false),
        }
    }

    fn next_id(&self) -> RequestId {
        if self.overflow_detected.load(Ordering::Acquire) {
            // Already in UUID mode
            return RequestId::Uuid(SmartString::from(Uuid::new_v4().to_string()));
        }

        let id = self.counter.fetch_add(1, Ordering::SeqCst);
        if id == u64::MAX {
            // Overflow detected, switch to UUID mode
            self.overflow_detected.store(true, Ordering::Release);
            RequestId::Uuid(SmartString::from(Uuid::new_v4().to_string()))
        } else {
            RequestId::Sequential(id)
        }
    }
}

/// Parameters for modifying an order
#[derive(Debug, Clone)]
pub struct ModifyOrderParams {
    /// Unique identifier of the order to modify
    pub order_id: SmartString,
    /// Trading pair symbol (e.g., "BTCUSDT")
    pub symbol: SmartString,
    /// Order side (buy or sell)
    pub side: OrderSide,
    /// Type of order (limit, market, etc.)
    pub order_type: OrderType,
    /// New quantity for the order
    pub new_quantity: Decimal,
    /// New price for the order (None for market orders)
    pub new_price: Option<Decimal>,
}

/// Pending request tracking with proper ID type
#[derive(Debug, Clone)]
struct PendingRequest {
    pub method: SmartString,
    pub timestamp: u64,
    pub report_tx: Option<Sender<ExecutionReport>>,
    pub request_id: RequestId,
}

/// Pending requests storage with overflow protection
/// Optimized to use `RequestKey` instead of string allocations
type PendingRequestsMap = FxHashMap<RequestKey, PendingRequest>;

/// High-performance WebSocket-only trading client for Binance
///
/// This client provides ultra-low latency trading by using WebSocket API exclusively.
/// No REST API dependencies for maximum performance in HFT scenarios.
///
/// # Usage
///
/// ```rust,no_run
/// use std::sync::Arc;
/// use rusty_ems::exchanges::BinanceWebSocketTrader;
/// use rusty_common::auth::exchanges::binance::BinanceAuth;
/// use flume;
///
/// # async fn example() -> anyhow::Result<()> {
/// let auth = Arc::new(BinanceAuth::new_hmac("APKEY".into(), "SECRET".into()));
/// let trader = Arc::new(BinanceWebSocketTrader::new(auth));
/// let (report_tx, _report_rx) = flume::bounded(100);
///
/// // Connect and use trading methods
/// trader.connect(report_tx.clone()).await?;
/// // trader.place_order(order, report_tx.clone()).await?;
/// // trader.cancel_order("order123".into(), report_tx.clone()).await?;
/// # Ok(())
/// # }
/// ```
/// Task handles for proper lifecycle management
#[derive(Default)]
struct TaskHandles {
    response_handler: Option<JoinHandle<()>>,
    ping_handler: Option<JoinHandle<()>>,
    reconnection_monitor: Option<JoinHandle<()>>,
}

/// Connection metrics for observability
/// Cache-line aligned to prevent false sharing in multi-threaded scenarios
#[repr(align(64))] // Align to cache line boundary (64 bytes on x86-64)
#[derive(Debug, Default)]
pub struct ConnectionMetrics {
    /// Number of WebSocket protocol ping frames sent
    pub websocket_pings_sent: AtomicU64,
    /// Number of WebSocket protocol pong frames received
    pub websocket_pongs_received: AtomicU64,
    /// Number of JSON ping messages sent for application-level keepalive
    pub json_pings_sent: AtomicU64,
    /// Number of JSON pong responses received from server
    pub json_pongs_received: AtomicU64,
    /// Total number of messages sent to the server
    pub messages_sent: AtomicU64,
    /// Total number of messages received from the server
    pub messages_received: AtomicU64,
    /// Number of authentication attempts made
    pub authentication_attempts: AtomicU64,
    /// Number of reconnection attempts initiated
    pub reconnection_attempts: AtomicU64,
    /// Number of successful reconnection attempts
    pub successful_reconnections: AtomicU64,
    /// Number of failed reconnection attempts
    pub failed_reconnections: AtomicU64,
}

/// Detailed connection health information
#[derive(Debug, Clone)]
pub struct ConnectionHealth {
    /// Current connection state
    pub state: ConnectionState,
    /// Whether the WebSocket connection is established
    pub is_connected: bool,
    /// Whether the connection is authenticated and ready for trading
    pub is_authenticated: bool,
    /// Time elapsed since last ping was sent
    pub time_since_last_ping: Option<Duration>,
    /// Time elapsed since last pong was received
    pub time_since_last_pong: Option<Duration>,
    /// Time elapsed since authentication was completed
    pub time_since_auth: Option<Duration>,
    /// Total WebSocket protocol pings sent
    pub websocket_pings_sent: u64,
    /// Total WebSocket protocol pongs received
    pub websocket_pongs_received: u64,
    /// Total JSON ping messages sent
    pub json_pings_sent: u64,
    /// Total JSON pong responses received
    pub json_pongs_received: u64,
    /// Total messages sent to server
    pub messages_sent: u64,
    /// Total messages received from server
    pub messages_received: u64,
    /// Total reconnection attempts initiated
    pub reconnection_attempts: u64,
    /// Total successful reconnections completed
    pub successful_reconnections: u64,
    /// Total failed reconnection attempts
    pub failed_reconnections: u64,
}

/// High-performance WebSocket-only trading client for Binance
///
/// Provides ultra-low latency order execution using WebSocket API exclusively.
/// Includes connection management, authentication, rate limiting, and comprehensive metrics.
pub struct BinanceWebSocketTrader {
    /// Authentication handler
    auth: Arc<BinanceAuth>,

    /// WebSocket connection for trading API
    ws_sink: Arc<AsyncRwLock<Option<WebSocketSink>>>,
    ws_stream: Arc<AsyncRwLock<Option<WebSocketStream>>>,

    /// Connection state machine
    state: Arc<AtomicU8>,

    /// High-precision clock (using monotonic time)
    clock: Clock,

    /// Request ID generator with overflow protection
    request_id_gen: Arc<RequestIdGenerator>,

    /// Pending requests tracking with string keys for UUID support
    pending_requests: Arc<RwLock<PendingRequestsMap>>,

    /// WebSocket API URL (production or testnet)
    ws_url: &'static str,

    /// Last ping sent timestamp (nanoseconds, monotonic)
    last_ping_time: Arc<AtomicU64>,

    /// Last pong received timestamp (nanoseconds, monotonic)
    last_pong_time: Arc<AtomicU64>,

    /// Authentication completed timestamp (nanoseconds, monotonic)
    auth_completed_time: Arc<AtomicU64>,

    /// Task handles for lifecycle management
    task_handles: Arc<AsyncRwLock<TaskHandles>>,

    /// Connection metrics
    metrics: Arc<ConnectionMetrics>,

    /// Reconnection backoff state
    reconnection_backoff_ms: Arc<AtomicU64>,

    /// Control flag for reconnection monitor
    reconnection_control: Arc<ReconnectionMonitorControl>,

    /// Order rate limiter for batch orders
    order_rate_limiter: Arc<RwLock<OrderRateLimiter>>,

    /// Position manager for tracking futures positions
    position_manager: Arc<dyn PositionManager>,
}
/// Atomic flag to signal reconnection monitor to stop
/// This prevents race conditions during socket cleanup
#[derive(Clone)]
struct ReconnectionMonitorControl {
    should_stop: Arc<AtomicBool>,
}

impl Clone for BinanceWebSocketTrader {
    fn clone(&self) -> Self {
        Self {
            auth: self.auth.clone(),
            ws_sink: self.ws_sink.clone(),
            ws_stream: self.ws_stream.clone(),
            state: self.state.clone(),
            clock: self.clock.clone(),
            request_id_gen: self.request_id_gen.clone(), // Shared ID generator
            pending_requests: self.pending_requests.clone(),
            ws_url: self.ws_url,
            last_ping_time: self.last_ping_time.clone(),
            last_pong_time: self.last_pong_time.clone(),
            auth_completed_time: self.auth_completed_time.clone(),
            task_handles: self.task_handles.clone(),
            metrics: self.metrics.clone(),
            reconnection_backoff_ms: self.reconnection_backoff_ms.clone(),
            reconnection_control: self.reconnection_control.clone(),
            order_rate_limiter: self.order_rate_limiter.clone(),
            position_manager: self.position_manager.clone(),
        }
    }
}

impl BinanceWebSocketTrader {
    /// Get current connection state
    fn get_state(&self) -> ConnectionState {
        ConnectionState::from(self.state.load(Ordering::Acquire))
    }

    /// Set connection state
    fn set_state(&self, new_state: ConnectionState) {
        let old_state = self.get_state();
        self.state.store(new_state as u8, Ordering::Release);

        if old_state != new_state {
            debug!("Connection state transition: {old_state:?} -> {new_state:?}");
        }
    }

    /// Check if authenticated and ready for operations
    #[must_use]
    pub fn is_authenticated(&self) -> bool {
        self.get_state() == ConnectionState::Authenticated
    }

    /// Get connection metrics
    #[must_use]
    pub const fn metrics(&self) -> &Arc<ConnectionMetrics> {
        &self.metrics
    }

    /// Get detailed connection health information
    /// Returns a struct with current connection status, timing information, and metrics
    #[must_use]
    pub fn get_connection_health(&self) -> ConnectionHealth {
        let now = self.clock.raw();
        let last_ping = self.last_ping_time.load(Ordering::Acquire);
        let last_pong = self.last_pong_time.load(Ordering::Acquire);
        let auth_time = self.auth_completed_time.load(Ordering::Acquire);

        let time_since_last_ping = if last_ping > 0 {
            Some(Duration::from_nanos(now.saturating_sub(last_ping)))
        } else {
            None
        };

        let time_since_last_pong = if last_pong > 0 {
            Some(Duration::from_nanos(now.saturating_sub(last_pong)))
        } else {
            None
        };

        let time_since_auth = if auth_time > 0 {
            Some(Duration::from_nanos(now.saturating_sub(auth_time)))
        } else {
            None
        };

        ConnectionHealth {
            state: self.get_state(),
            is_connected: self.is_connected(),
            is_authenticated: self.is_authenticated(),
            time_since_last_ping,
            time_since_last_pong,
            time_since_auth,
            websocket_pings_sent: self.metrics.websocket_pings_sent.load(Ordering::Relaxed),
            websocket_pongs_received: self
                .metrics
                .websocket_pongs_received
                .load(Ordering::Relaxed),
            json_pings_sent: self.metrics.json_pings_sent.load(Ordering::Relaxed),
            json_pongs_received: self.metrics.json_pongs_received.load(Ordering::Relaxed),
            messages_sent: self.metrics.messages_sent.load(Ordering::Relaxed),
            messages_received: self.metrics.messages_received.load(Ordering::Relaxed),
            reconnection_attempts: self.metrics.reconnection_attempts.load(Ordering::Relaxed),
            successful_reconnections: self
                .metrics
                .successful_reconnections
                .load(Ordering::Relaxed),
            failed_reconnections: self.metrics.failed_reconnections.load(Ordering::Relaxed),
        }
    }

    /// Calculate exponential backoff with jitter
    fn calculate_backoff_ms(&self) -> u64 {
        let current = self.reconnection_backoff_ms.load(Ordering::Acquire);
        let max_backoff = 60_000; // 60 seconds max

        // Add jitter (±25%)
        let jitter = (current / 4) as i64;
        let mut rng = rand::rng();
        let random_jitter = rng.random_range(-jitter..=jitter);

        let next_backoff = ((current * 2).min(max_backoff) as i64 + random_jitter).max(1000) as u64;
        self.reconnection_backoff_ms
            .store(next_backoff, Ordering::Release);

        current
    }

    /// Reset backoff on successful connection
    fn reset_backoff(&self) {
        self.reconnection_backoff_ms.store(1000, Ordering::Release);
    }

    /// Create a new WebSocket-only trading client
    #[must_use]
    pub fn new(auth: Arc<BinanceAuth>, position_manager: Arc<dyn PositionManager>) -> Self {
        let clock = Clock::new();
        Self {
            auth,
            ws_sink: Arc::new(AsyncRwLock::new(None)),
            ws_stream: Arc::new(AsyncRwLock::new(None)),
            state: Arc::new(AtomicU8::new(ConnectionState::Disconnected as u8)),
            clock: clock.clone(),
            request_id_gen: Arc::new(RequestIdGenerator::new()),
            pending_requests: Arc::new(RwLock::new(FxHashMap::default())),
            ws_url: BINANCE_WS_API_URL,
            last_ping_time: Arc::new(AtomicU64::new(0)),
            last_pong_time: Arc::new(AtomicU64::new(0)),
            auth_completed_time: Arc::new(AtomicU64::new(0)),
            task_handles: Arc::new(AsyncRwLock::new(TaskHandles::default())),
            metrics: Arc::new(ConnectionMetrics::default()),
            reconnection_backoff_ms: Arc::new(AtomicU64::new(1000)), // Start with 1 second
            reconnection_control: Arc::new(ReconnectionMonitorControl {
                should_stop: Arc::new(AtomicBool::new(false)),
            }),
            order_rate_limiter: Arc::new(RwLock::new(OrderRateLimiter::new(clock))),
            position_manager,
        }
    }

    /// Create a new WebSocket-only trading client for testnet
    #[must_use]
    pub fn new_testnet(auth: Arc<BinanceAuth>, position_manager: Arc<dyn PositionManager>) -> Self {
        let clock = Clock::new();
        Self {
            auth,
            ws_sink: Arc::new(AsyncRwLock::new(None)),
            ws_stream: Arc::new(AsyncRwLock::new(None)),
            state: Arc::new(AtomicU8::new(ConnectionState::Disconnected as u8)),
            clock: clock.clone(),
            request_id_gen: Arc::new(RequestIdGenerator::new()),
            pending_requests: Arc::new(RwLock::new(FxHashMap::default())),
            ws_url: BINANCE_WS_API_TESTNET_URL,
            last_ping_time: Arc::new(AtomicU64::new(0)),
            last_pong_time: Arc::new(AtomicU64::new(0)),
            auth_completed_time: Arc::new(AtomicU64::new(0)),
            task_handles: Arc::new(AsyncRwLock::new(TaskHandles::default())),
            metrics: Arc::new(ConnectionMetrics::default()),
            reconnection_backoff_ms: Arc::new(AtomicU64::new(1000)), // Start with 1 second
            reconnection_control: Arc::new(ReconnectionMonitorControl {
                should_stop: Arc::new(AtomicBool::new(false)),
            }),
            order_rate_limiter: Arc::new(RwLock::new(OrderRateLimiter::new(clock))),
            position_manager,
        }
    }

    /// Connect to Binance WebSocket API
    pub async fn connect(&self, report_tx: Sender<ExecutionReport>) -> Result<()> {
        // Check if already connected or connecting
        match self.get_state() {
            ConnectionState::Authenticated => {
                debug!("Already authenticated, skipping connection");
                return Ok(());
            }
            ConnectionState::Connecting
            | ConnectionState::Connected
            | ConnectionState::Authenticating => {
                warn!("Connection already in progress");
                return Err(anyhow!("Connection already in progress"));
            }
            _ => {}
        }

        // Transition to connecting state
        self.set_state(ConnectionState::Connecting);

        let ws_url = self.ws_url;
        debug!("Connecting to Binance WebSocket API: {ws_url}");

        // Create WebSocket configuration
        let ws_config =
            WebSocketConfig::new(rusty_common::types::Exchange::Binance, ws_url.to_string());

        let mut connector = WebSocketConnector::new(
            ws_config,
            Arc::new(parking_lot::RwLock::new(
                rusty_common::websocket::ConnectionStats::default(),
            )),
            Arc::new(parking_lot::RwLock::new(
                rusty_common::websocket::ConnectionState::Disconnected,
            )),
        );

        let (ws_sink, ws_stream) = match connector.connect_with_retry(ws_url).await {
            Ok(result) => result,
            Err(e) => {
                self.set_state(ConnectionState::Disconnected);
                return Err(anyhow!("WebSocket connection failed: {}", e));
            }
        };

        *self.ws_sink.write().await = Some(ws_sink);
        *self.ws_stream.write().await = Some(ws_stream);

        // Transition to connected state
        self.set_state(ConnectionState::Connected);

        // Start response handler and ping handler with proper task management
        let response_handle = self.start_response_handler(report_tx.clone()).await?;
        let ping_handle = self.start_ping_handler().await?;
        let reconnect_handle = self.start_reconnection_monitor(report_tx.clone()).await?;

        // Store task handles
        {
            let mut handles = self.task_handles.write().await;
            handles.response_handler = Some(response_handle);
            handles.ping_handler = Some(ping_handle);
            handles.reconnection_monitor = Some(reconnect_handle);
        }

        // Authenticate the session
        self.set_state(ConnectionState::Authenticating);
        match self.authenticate_session().await {
            Ok(()) => {
                self.set_state(ConnectionState::Authenticated);
                self.auth_completed_time
                    .store(self.clock.raw(), Ordering::Release);
                self.reset_backoff();
                debug!("Successfully authenticated to Binance WebSocket API");
                Ok(())
            }
            Err(e) => {
                error!("Authentication failed: {e}");
                self.set_state(ConnectionState::Connected); // Revert to connected state
                Err(e)
            }
        }
    }

    /// Authenticate the WebSocket session using logon request
    async fn authenticate_session(&self) -> Result<()> {
        debug!("🔐 Generating WebSocket authentication message...");

        // Generate WebSocket authentication message
        let auth_message = self.auth.generate_ws_auth().map_err(|e| {
            error!("🔐 Failed to generate WebSocket auth: {e}");
            anyhow!("Failed to generate WebSocket auth: {}", e)
        })?;

        debug!("🔐 Authentication message generated, sending to server...");

        // Send authentication message with timeout
        let send_result = tokio::time::timeout(
            Duration::from_secs(10), // 10 second timeout for auth send
            async {
                if let Some(ref mut ws_sink) = self.ws_sink.write().await.as_mut() {
                    ws_sink
                        .send(Message::Text(auth_message).to_frame_view())
                        .await
                        .map_err(|e| anyhow!("Failed to send auth message: {}", e))
                } else {
                    Err(anyhow!("WebSocket sink not available"))
                }
            },
        )
        .await;

        match send_result {
            Ok(Ok(())) => {
                debug!("🔐 WebSocket authentication sent successfully");
                Ok(())
            }
            Ok(Err(e)) => {
                error!("🔐 Failed to send authentication message: {e}");
                Err(e)
            }
            Err(_) => {
                error!("🔐 Authentication send timed out after 10 seconds");
                Err(anyhow!("Authentication send timed out"))
            }
        }
    }

    /// Start the response handler for incoming WebSocket messages
    async fn start_response_handler(
        &self,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<JoinHandle<()>> {
        let ws_stream = self.ws_stream.clone();
        let pending_requests = self.pending_requests.clone();
        let clock = self.clock.clone();
        let last_pong_time = self.last_pong_time.clone();
        let state = self.state.clone();
        let metrics = self.metrics.clone();
        let position_manager = self.position_manager.clone();
        let self_clone = Self {
            auth: self.auth.clone(),
            ws_sink: self.ws_sink.clone(),
            ws_stream: self.ws_stream.clone(),
            state: self.state.clone(),
            clock: self.clock.clone(),
            request_id_gen: self.request_id_gen.clone(),
            pending_requests: self.pending_requests.clone(),
            ws_url: self.ws_url,
            last_ping_time: self.last_ping_time.clone(),
            last_pong_time: self.last_pong_time.clone(),
            auth_completed_time: self.auth_completed_time.clone(),
            task_handles: self.task_handles.clone(),
            metrics: self.metrics.clone(),
            reconnection_backoff_ms: self.reconnection_backoff_ms.clone(),
            reconnection_control: self.reconnection_control.clone(),
            order_rate_limiter: self.order_rate_limiter.clone(),
            position_manager: self.position_manager.clone(),
        };

        let handle = tokio::spawn(async move {
            let mut ws_guard = ws_stream.write().await;
            if let Some(mut ws_stream) = ws_guard.take() {
                loop {
                    // Check if we should still be running
                    let current_state = ConnectionState::from(state.load(Ordering::Acquire));
                    if current_state == ConnectionState::Disconnected
                        || current_state == ConnectionState::Disconnecting
                    {
                        debug!("Response handler stopping - state: {current_state:?}");
                        break;
                    }

                    if let Some(frame) = ws_stream.next().await {
                        metrics.messages_received.fetch_add(1, Ordering::Relaxed);

                        // Validate frame size
                        if frame.payload.len() > MAX_MESSAGE_SIZE {
                            error!(
                                "Frame too large: {} bytes (max: {})",
                                frame.payload.len(),
                                MAX_MESSAGE_SIZE
                            );
                            self_clone.kill_socket("Frame size exceeded maximum").await;
                            break;
                        }

                        // Process frames manually according to YAWC guidelines
                        match frame.opcode {
                            OpCode::Text => {
                                // Handle text data frames (JSON messages)
                                let text = match std::str::from_utf8(&frame.payload) {
                                    Ok(text) => text,
                                    Err(e) => {
                                        error!("Invalid UTF-8 in text frame: {e}");
                                        self_clone.kill_socket("Invalid UTF-8 in text frame").await;
                                        break;
                                    }
                                };

                                let mut text_copy = text.to_string();
                                match unsafe { simd_json::from_str::<JsonValue>(&mut text_copy) } {
                                    Ok(json) => {
                                        debug!("response is {:?}", json.get("id"));
                                        Self::handle_websocket_message(
                                            &json,
                                            &pending_requests,
                                            &report_tx,
                                            &clock,
                                            &position_manager,
                                        )
                                        .await;
                                    }
                                    Err(e) => {
                                        error!("Failed to parse WebSocket message: {e}");
                                        self_clone.kill_socket("JSON parsing error").await;
                                        break;
                                    }
                                }
                            }
                            OpCode::Ping => {
                                // YAWC handles automatic pong responses
                                debug!(
                                    "Received ping - automatic pong response will be sent by YAWC"
                                );
                            }
                            OpCode::Pong => {
                                // Handle pong responses to our pings
                                let now = clock.raw();
                                last_pong_time.store(now, Ordering::SeqCst);
                                debug!("Received pong from server at {now}");
                            }
                            OpCode::Close => {
                                // Handle close frames with proper code extraction
                                let close_info = if frame.payload.len() >= 2 {
                                    let code =
                                        u16::from_be_bytes([frame.payload[0], frame.payload[1]]);
                                    let reason = if frame.payload.len() > 2 {
                                        std::str::from_utf8(&frame.payload[2..])
                                            .unwrap_or("Invalid UTF-8 in close reason")
                                    } else {
                                        "No reason provided"
                                    };
                                    format!("code: {code}, reason: {reason}")
                                } else {
                                    "No close code provided".to_string()
                                };

                                warn!("WebSocket connection closing: {close_info}");
                                self_clone
                                    .kill_socket(&format!(
                                        "Connection closed by server: {close_info}"
                                    ))
                                    .await;
                                break;
                            }
                            OpCode::Binary => {
                                // Handle binary frames if needed (currently not expected for Binance)
                                debug!("Received binary frame with {} bytes", frame.payload.len());
                            }
                            _ => {
                                // Handle unknown opcodes
                                debug!("Received frame with unknown opcode: {:?}", frame.opcode);
                            }
                        }
                    } else {
                        // Stream ended - connection lost
                        error!("WebSocket stream ended unexpectedly");
                        self_clone.kill_socket("Stream ended unexpectedly").await;
                        break;
                    }
                }
            } else {
                error!("No WebSocket stream available in response handler");
                self_clone
                    .kill_socket("No WebSocket stream available")
                    .await;
            }

            debug!("Response handler task exiting");
        });

        Ok(handle)
    }

    /// Start the ping handler for keeping WebSocket connection alive
    async fn start_ping_handler(&self) -> Result<JoinHandle<()>> {
        let ws_sink = self.ws_sink.clone();
        let clock = self.clock.clone();
        let last_ping_time = self.last_ping_time.clone();
        let last_pong_time = self.last_pong_time.clone();
        let state = self.state.clone();
        let auth_completed_time = self.auth_completed_time.clone();
        let metrics = self.metrics.clone();
        let self_clone = Self {
            auth: self.auth.clone(),
            ws_sink: self.ws_sink.clone(),
            ws_stream: self.ws_stream.clone(),
            state: self.state.clone(),
            clock: self.clock.clone(),
            request_id_gen: self.request_id_gen.clone(),
            pending_requests: self.pending_requests.clone(),
            ws_url: self.ws_url,
            last_ping_time: self.last_ping_time.clone(),
            last_pong_time: self.last_pong_time.clone(),
            auth_completed_time: self.auth_completed_time.clone(),
            task_handles: self.task_handles.clone(),
            metrics: self.metrics.clone(),
            reconnection_backoff_ms: self.reconnection_backoff_ms.clone(),
            reconnection_control: self.reconnection_control.clone(),
            order_rate_limiter: self.order_rate_limiter.clone(),
            position_manager: self.position_manager.clone(),
        };

        let handle = tokio::spawn(async move {
            let mut ping_interval = interval(Duration::from_secs(PING_INTERVAL_SECONDS));
            ping_interval.tick().await; // Skip first immediate tick

            loop {
                ping_interval.tick().await;

                // Check if we should still be running
                let current_state = ConnectionState::from(state.load(Ordering::Acquire));
                if current_state == ConnectionState::Disconnected
                    || current_state == ConnectionState::Disconnecting
                {
                    debug!("Ping handler stopping - state: {current_state:?}");
                    break;
                }

                let now = clock.raw();
                let last_pong = last_pong_time.load(Ordering::Acquire);
                let auth_time = auth_completed_time.load(Ordering::Acquire);

                // Only check for pong timeout if we're authenticated
                if current_state == ConnectionState::Authenticated && auth_time > 0 {
                    // Check if we've been waiting too long for any pong after authentication
                    let waiting_for_first_pong = last_pong == 0
                        && (now - auth_time) > (PONG_TIMEOUT_SECONDS * 1_000_000_000);

                    let pong_timeout =
                        last_pong > 0 && (now - last_pong) > (PONG_TIMEOUT_SECONDS * 1_000_000_000);

                    if waiting_for_first_pong || pong_timeout {
                        warn!("Pong timeout detected, connection is dead");
                        self_clone.kill_socket("Pong timeout").await;
                        break;
                    }
                }

                // Send ping with exception handling
                if let Some(ref mut sink) = ws_sink.write().await.as_mut() {
                    let ping_data = b"ping".to_vec();
                    match sink.send(Message::Ping(ping_data).to_frame_view()).await {
                        Ok(()) => {
                            last_ping_time.store(now, Ordering::SeqCst);
                            metrics.websocket_pings_sent.fetch_add(1, Ordering::Relaxed);
                            debug!("Sent ping at {now}");
                        }
                        Err(e) => {
                            error!("Failed to send ping: {e}");
                            self_clone
                                .kill_socket(&format!("Failed to send ping: {e}"))
                                .await;
                            break;
                        }
                    }
                } else {
                    warn!("WebSocket sink not available for ping");
                    self_clone.kill_socket("WebSocket sink not available").await;
                    break;
                }
            }

            debug!("Ping handler task exiting");
        });

        Ok(handle)
    }

    /// Start the auto-reconnection monitor for automatic recovery
    ///
    /// This monitor continuously watches the connection state and performs automatic
    /// reconnection when the connection is lost. The reconnection behavior includes:
    ///
    /// - Exponential backoff starting at 1 second, doubling up to 60 seconds max
    /// - Jitter of ±25% added to prevent thundering herd
    /// - Automatic authentication after successful reconnection
    /// - Preservation of pending requests across reconnections
    /// - Health metrics tracking for monitoring and alerting
    ///
    /// The monitor will stop when:
    /// - The connection state becomes Disconnected or Disconnecting
    /// - The connection has been stable for the configured check interval
    /// - A manual disconnect is initiated
    async fn start_reconnection_monitor(
        &self,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<JoinHandle<()>> {
        let state = self.state.clone();
        let metrics = self.metrics.clone();
        let control = self.reconnection_control.clone();

        // Reset the stop flag when starting a new monitor
        control.should_stop.store(false, Ordering::Release);

        let self_clone = Self {
            auth: self.auth.clone(),
            ws_sink: self.ws_sink.clone(),
            ws_stream: self.ws_stream.clone(),
            state: self.state.clone(),
            clock: self.clock.clone(),
            request_id_gen: self.request_id_gen.clone(),
            pending_requests: self.pending_requests.clone(),
            ws_url: self.ws_url,
            last_ping_time: self.last_ping_time.clone(),
            last_pong_time: self.last_pong_time.clone(),
            auth_completed_time: self.auth_completed_time.clone(),
            task_handles: self.task_handles.clone(),
            metrics: self.metrics.clone(),
            reconnection_backoff_ms: self.reconnection_backoff_ms.clone(),
            reconnection_control: self.reconnection_control.clone(),
            order_rate_limiter: self.order_rate_limiter.clone(),
            position_manager: self.position_manager.clone(),
        };

        let handle = tokio::spawn(async move {
            let mut check_interval = interval(Duration::from_secs(3)); // Check every 3 seconds
            check_interval.tick().await; // Skip first immediate tick

            let mut consecutive_disconnections = 0;
            const MAX_RECONNECTION_ATTEMPTS: u32 = 10; // Allow more attempts with backoff

            info!(
                "🤖 Auto-reconnection monitor started (check every 3s, max {MAX_RECONNECTION_ATTEMPTS} attempts)"
            );

            loop {
                check_interval.tick().await;

                // Check if we should stop the monitor
                if control.should_stop.load(Ordering::Acquire) {
                    info!("🛑 Reconnection monitor received stop signal - exiting gracefully");
                    break;
                }

                // Check if connection is healthy based on state
                let current_state = ConnectionState::from(state.load(Ordering::Acquire));
                let is_healthy = current_state == ConnectionState::Authenticated;

                if !is_healthy
                    && current_state != ConnectionState::Connecting
                    && current_state != ConnectionState::Authenticating
                {
                    consecutive_disconnections += 1;
                    metrics
                        .reconnection_attempts
                        .fetch_add(1, Ordering::Relaxed);

                    if consecutive_disconnections > MAX_RECONNECTION_ATTEMPTS {
                        error!(
                            "🔴 Maximum reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded. Stopping auto-reconnection."
                        );
                        break;
                    }

                    warn!(
                        "🔄 Connection unhealthy detected (state: {current_state:?})! Attempting auto-reconnection #{consecutive_disconnections}/{MAX_RECONNECTION_ATTEMPTS}"
                    );

                    // Wait with exponential backoff
                    let backoff_ms = self_clone.calculate_backoff_ms();
                    info!("⏳ Waiting {backoff_ms}ms before reconnection attempt");
                    tokio::time::sleep(Duration::from_millis(backoff_ms)).await;

                    // Attempt reconnection with timeout
                    info!("🔄 Starting internal reconnection attempt...");
                    let reconnect_start = std::time::Instant::now();

                    match tokio::time::timeout(
                        Duration::from_secs(45), // Extended timeout for full reconnection
                        self_clone.reconnect_internal(report_tx.clone()),
                    )
                    .await
                    {
                        Ok(Ok(())) => {
                            info!(
                                "✅ Auto-reconnection successful in {:.2}s!",
                                reconnect_start.elapsed().as_secs_f32()
                            );
                            consecutive_disconnections = 0; // Reset counter on success
                            metrics
                                .successful_reconnections
                                .fetch_add(1, Ordering::Relaxed);
                            self_clone.reset_backoff(); // Reset exponential backoff
                        }
                        Ok(Err(e)) => {
                            error!(
                                "❌ Auto-reconnection failed after {:.2}s: {}",
                                reconnect_start.elapsed().as_secs_f32(),
                                e
                            );
                            // Will try again in next iteration
                        }
                        Err(_) => {
                            error!(
                                "❌ Auto-reconnection timed out after {:.2}s (45s limit) - will retry",
                                reconnect_start.elapsed().as_secs_f32()
                            );
                            // Force kill any hanging connections
                            self_clone
                                .kill_socket("Reconnection timeout - forcing cleanup")
                                .await;
                        }
                    }
                } else {
                    // Connection is healthy, reset counter
                    if consecutive_disconnections > 0 {
                        consecutive_disconnections = 0;
                        debug!("Connection restored and stable");
                    }
                }
            }

            warn!("Auto-reconnection monitor exiting");
        });

        Ok(handle)
    }

    /// Internal reconnection method that doesn't cause infinite loops
    async fn reconnect_internal(&self, report_tx: Sender<ExecutionReport>) -> Result<()> {
        info!("🔄 RECONNECT-STEP 1/3: Internal reconnection starting...");

        // Kill existing connection first (but don't start new monitor)
        info!("🔄 RECONNECT-STEP 2/3: Killing existing connection...");

        // Add timeout to kill_socket in case it hangs
        match tokio::time::timeout(
            Duration::from_secs(5), // 5 second timeout for killing socket
            self.kill_socket("Internal reconnection requested"),
        )
        .await
        {
            Ok(()) => {
                info!("🔄 RECONNECT-STEP 2/3: Socket killed successfully");
            }
            Err(_) => {
                error!("🔄 RECONNECT-STEP 2/3: Socket kill timed out after 5s - continuing anyway");
                // Force reset connection state
            }
        }

        // Small delay to allow cleanup
        info!("🔄 RECONNECT-STEP 2/3: Waiting 100ms for cleanup...");
        tokio::time::sleep(Duration::from_millis(100)).await;
        info!("🔄 RECONNECT-STEP 2/3: Cleanup delay completed");

        // Attempt new connection (this will NOT start a new monitor since we're already in one)
        info!("🔄 RECONNECT-STEP 3/3: Attempting new connection...");
        match self.connect_without_monitor(report_tx).await {
            Ok(()) => {
                info!("🎉 Internal reconnection completed successfully");
                Ok(())
            }
            Err(e) => {
                error!("💥 Internal reconnection failed: {e}");
                Err(e)
            }
        }
    }

    /// Connect without starting a new reconnection monitor (used internally)
    async fn connect_without_monitor(&self, report_tx: Sender<ExecutionReport>) -> Result<()> {
        let ws_url = self.ws_url;

        info!("🔌 STEP 1/6: Connecting to Binance WebSocket API (internal): {ws_url}");

        // Create WebSocket configuration
        let ws_config =
            WebSocketConfig::new(rusty_common::types::Exchange::Binance, ws_url.to_string());

        let mut connector = WebSocketConnector::new(
            ws_config,
            Arc::new(parking_lot::RwLock::new(
                rusty_common::websocket::ConnectionStats::default(),
            )),
            Arc::new(parking_lot::RwLock::new(
                rusty_common::websocket::ConnectionState::Disconnected,
            )),
        );

        info!("📡 STEP 2/6: Attempting WebSocket connection with timeout...");
        let connection_start = std::time::Instant::now();

        let (ws_sink, ws_stream) = match tokio::time::timeout(
            Duration::from_secs(15), // 15 second timeout for connection
            connector.connect_with_retry(ws_url),
        )
        .await
        {
            Ok(Ok((sink, stream))) => {
                info!(
                    "🔗 STEP 2/6: WebSocket connection established in {:.2}s",
                    connection_start.elapsed().as_secs_f32()
                );
                (sink, stream)
            }
            Ok(Err(e)) => {
                error!(
                    "📡 STEP 2/6: WebSocket connection failed after {:.2}s: {}",
                    connection_start.elapsed().as_secs_f32(),
                    e
                );
                return Err(anyhow!("WebSocket connection failed: {}", e));
            }
            Err(_) => {
                error!("📡 STEP 2/6: WebSocket connection timed out after 15 seconds");
                return Err(anyhow!("WebSocket connection timed out"));
            }
        };

        info!("🔗 STEP 3/6: Setting up WebSocket streams...");
        *self.ws_sink.write().await = Some(ws_sink);
        *self.ws_stream.write().await = Some(ws_stream);

        info!("🚀 STEP 4/6: Starting response handler...");
        // Start response handler and ping handler (but NOT reconnection monitor)
        self.start_response_handler(report_tx).await.map_err(|e| {
            error!("🚀 STEP 4/6: Failed to start response handler: {e}");
            anyhow!("Failed to start response handler: {}", e)
        })?;

        debug!("🏓 STEP 5/6: Starting ping handler...");
        self.start_ping_handler().await.map_err(|e| {
            error!("🏓 STEP 5/6: Failed to start ping handler: {e}");
            anyhow!("Failed to start ping handler: {}", e)
        })?;

        info!("🔐 STEP 6/6: Authenticating session...");
        // Authenticate the session
        self.authenticate_session().await.map_err(|e| {
            error!("🔐 STEP 6/6: Authentication failed: {e}");
            anyhow!("Authentication failed: {}", e)
        })?;

        info!("✅ Successfully reconnected to Binance WebSocket API");
        Ok(())
    }

    /// Handle incoming WebSocket messages
    async fn handle_websocket_message(
        json: &JsonValue,
        pending_requests: &Arc<RwLock<PendingRequestsMap>>,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
        position_manager: &Arc<dyn PositionManager>,
    ) {
        // Check if this is a response to our request
        let pending_request = if let Some(request_id) = json.get("id") {
            let mut pending = pending_requests.write();

            // Try to extract ID as either number or string - optimized to avoid string allocation
            if let Some(id_num) = request_id.as_u64() {
                let key = RequestKey::Sequential(id_num);
                pending.remove(&key)
            } else if let Some(id_str) = request_id.as_str() {
                let key = RequestKey::Uuid(SmartString::from(id_str));
                pending.remove(&key)
            } else {
                None
            }
        } else {
            None
        };

        if let Some(pending_req) = pending_request {
            Self::handle_response(json, &pending_req, report_tx, clock).await;
        }
        // Check if this is an execution report from user data stream
        else if let Some(event_type) = json.get("e").and_then(|v| v.as_str()) {
            if event_type == "executionReport" {
                Self::handle_execution_report(json, report_tx, clock).await;
            } else if event_type == "ACCOUNT_UPDATE" {
                Self::handle_account_update(json, position_manager).await;
            }
        }
    }

    /// Handle response to our trading requests
    async fn handle_response(
        json: &JsonValue,
        pending_req: &PendingRequest,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) {
        // Check for errors
        if let Some(error) = json.get("error") {
            let error_code = error
                .get("code")
                .and_then(simd_json::prelude::ValueAsScalar::as_i64)
                .unwrap_or(0);
            let error_msg = error
                .get("msg")
                .and_then(|v| v.as_str())
                .unwrap_or("Unknown error");

            error!("WebSocket API error {error_code}: {error_msg}");

            // Send rejection report if we have a report channel
            if let Some(ref tx) = pending_req.report_tx {
                let report = ExecutionReport {
                    id: id_generation::generate_ws_timestamp_id("error", clock.raw()),
                    order_id: "unknown".into(),
                    exchange_timestamp: 0,
                    system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
                    instrument_id: InstrumentId {
                        symbol: "UNKNOWN".into(),
                        venue: Venue::Binance,
                    },
                    status: OrderStatus::Rejected,
                    filled_quantity: Decimal::ZERO,
                    remaining_quantity: Decimal::ZERO,
                    execution_price: None,
                    reject_reason: Some(format!("{error_msg} ({error_code})").into()),
                    exchange_execution_id: None,
                    is_final: true,
                };

                if let Err(e) = tx.send_async(report).await {
                    error!("Failed to send error report: {e}");
                }
            }
            return;
        }

        // Handle successful responses based on method
        match pending_req.method.as_str() {
            "order.place" => {
                Self::handle_order_place_response(json, pending_req, report_tx, clock).await;
            }
            "order.cancel" => {
                Self::handle_order_cancel_response(json, pending_req, report_tx, clock).await;
            }
            "order.cancelReplace" => {
                Self::handle_order_modify_response(json, pending_req, report_tx, clock).await;
            }
            "ping" => {
                Self::handle_ping_response(json, pending_req).await;
            }
            "openOrders.cancelAll" => {
                Self::handle_cancel_all_orders_response(json, pending_req, report_tx, clock).await;
            }
            _ => {
                info!("Received response for method: {}", pending_req.method);
            }
        }
    }

    /// Handle order placement response
    async fn handle_order_place_response(
        json: &JsonValue,
        pending_req: &PendingRequest,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) {
        if let Some(ref tx) = pending_req.report_tx
            && let Some(result) = json.get("result")
        {
            let symbol: SmartString = result
                .get("symbol")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .into();

            let order_id: SmartString = result
                .get("orderId")
                .and_then(simd_json::prelude::ValueAsScalar::as_u64)
                .unwrap_or(0)
                .to_string()
                .into();

            let client_order_id: SmartString = result
                .get("clientOrderId")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .into();

            let status = result.get("status").and_then(|v| v.as_str()).unwrap_or("");

            let original_qty: SmartString = result
                .get("origQty")
                .and_then(|v| v.as_str())
                .unwrap_or("0")
                .into();

            let executed_qty: SmartString = result
                .get("executedQty")
                .and_then(|v| v.as_str())
                .unwrap_or("0")
                .into();

            let price: SmartString = result
                .get("price")
                .and_then(|v| v.as_str())
                .unwrap_or("0")
                .into();

            let transaction_time = result
                .get("transactTime")
                .and_then(simd_json::prelude::ValueAsScalar::as_u64)
                .unwrap_or(0);

            let report = ExecutionReport {
                id: id_generation::generate_ws_report_id("place", &client_order_id),
                order_id: client_order_id,
                exchange_timestamp: transaction_time * 1_000_000, // Convert ms to ns
                system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
                instrument_id: InstrumentId {
                    symbol,
                    venue: Venue::Binance,
                },
                status: Self::map_order_status(status),
                filled_quantity: Decimal::from_str_exact(&executed_qty).unwrap_or(Decimal::ZERO),
                remaining_quantity: {
                    let original = Decimal::from_str_exact(&original_qty).unwrap_or(Decimal::ZERO);
                    let executed = Decimal::from_str_exact(&executed_qty).unwrap_or(Decimal::ZERO);
                    original - executed
                },
                execution_price: Decimal::from_str_exact(&price).ok(),
                reject_reason: None,
                exchange_execution_id: Some(order_id),
                is_final: false,
            };

            // Handle backpressure by using try_send with timeout
            match tokio::time::timeout(
                Duration::from_millis(100), // 100ms timeout
                tx.send_async(report),
            )
            .await
            {
                Ok(Ok(())) => debug!("Order place report sent successfully"),
                Ok(Err(e)) => {
                    error!("Failed to send order place report: {e}");
                    // Channel closed, nothing we can do
                }
                Err(_) => {
                    warn!("Channel backpressure detected - order place report send timed out");
                    // Could implement a fallback strategy here if needed
                }
            }
        }
    }

    /// Handle order cancellation response
    async fn handle_order_cancel_response(
        json: &JsonValue,
        pending_req: &PendingRequest,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) {
        // Similar to place response but for cancellation
        if let Some(ref tx) = pending_req.report_tx
            && let Some(result) = json.get("result")
        {
            let symbol: SmartString = result
                .get("symbol")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .into();

            let client_order_id: SmartString = result
                .get("clientOrderId")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .into();

            let report = ExecutionReport {
                id: id_generation::generate_ws_report_id("cancel", &client_order_id),
                order_id: client_order_id,
                exchange_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
                system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
                instrument_id: InstrumentId {
                    symbol,
                    venue: Venue::Binance,
                },
                status: OrderStatus::Cancelled,
                filled_quantity: Decimal::ZERO,
                remaining_quantity: Decimal::ZERO,
                execution_price: None,
                reject_reason: None,
                exchange_execution_id: None,
                is_final: true,
            };

            // Handle backpressure by using try_send with timeout
            match tokio::time::timeout(
                Duration::from_millis(100), // 100ms timeout
                tx.send_async(report),
            )
            .await
            {
                Ok(Ok(())) => debug!("Cancel report sent successfully"),
                Ok(Err(e)) => {
                    error!("Failed to send cancel report: {e}");
                    // Channel closed, nothing we can do
                }
                Err(_) => {
                    warn!("Channel backpressure detected - cancel report send timed out");
                    // Could implement a fallback strategy here if needed
                }
            }
        }
    }

    /// Handle order modification response
    async fn handle_order_modify_response(
        json: &JsonValue,
        pending_req: &PendingRequest,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) {
        // Similar handling for order modification
        Self::handle_order_place_response(json, pending_req, report_tx, clock).await;
    }

    /// Handle cancel all orders response
    async fn handle_cancel_all_orders_response(
        json: &JsonValue,
        pending_req: &PendingRequest,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) {
        if let Some(ref tx) = pending_req.report_tx {
            if let Some(result) = json.get("result") {
                // Cancel all orders response contains an array of cancelled orders
                if let Some(orders_array) = result.as_array() {
                    for order_data in orders_array {
                        let symbol: SmartString = order_data
                            .get("symbol")
                            .and_then(|v| v.as_str())
                            .unwrap_or("")
                            .into();

                        let client_order_id: SmartString = order_data
                            .get("clientOrderId")
                            .and_then(|v| v.as_str())
                            .unwrap_or("")
                            .into();

                        let report = ExecutionReport {
                            id: id_generation::generate_ws_report_id(
                                "cancel_all",
                                &client_order_id,
                            ),
                            order_id: client_order_id,
                            exchange_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
                            system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
                            instrument_id: InstrumentId {
                                symbol,
                                venue: Venue::Binance,
                            },
                            status: OrderStatus::Cancelled,
                            filled_quantity: Decimal::ZERO,
                            remaining_quantity: Decimal::ZERO,
                            execution_price: None,
                            reject_reason: None,
                            exchange_execution_id: None,
                            is_final: true,
                        };

                        // Send execution report for each cancelled order
                        match tokio::time::timeout(
                            Duration::from_millis(100), // 100ms timeout
                            tx.send_async(report),
                        )
                        .await
                        {
                            Ok(Ok(())) => debug!(
                                "Cancel all order report sent successfully for order: {}",
                                order_data
                                    .get("clientOrderId")
                                    .and_then(|v| v.as_str())
                                    .unwrap_or("unknown")
                            ),
                            Ok(Err(e)) => {
                                error!("Failed to send cancel all order report: {e}");
                                // Channel closed, nothing we can do
                            }
                            Err(_) => {
                                warn!(
                                    "Channel backpressure detected - cancel all order report send timed out"
                                );
                                // Could implement a fallback strategy here if needed
                            }
                        }
                    }
                } else {
                    // Handle case where result is not an array (error or empty result)
                    debug!("Cancel all orders result is not an array: {result:?}");
                }
            } else {
                debug!("No result field in cancel all orders response");
            }
        }
    }

    /// Handle ping response from JSON API
    async fn handle_ping_response(json: &JsonValue, pending_req: &PendingRequest) {
        if let Some(result) = json.get("result") {
            debug!("✅ JSON ping successful - Server responded: {result:?}");
            debug!("   Request timestamp: {}", pending_req.timestamp);
            debug!("   Method: {}", pending_req.method);
        } else {
            debug!("✅ JSON ping successful - Empty result");
        }
    }

    /// Handle execution reports from user data stream
    async fn handle_execution_report(
        json: &JsonValue,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) {
        // Handle execution reports similar to the existing WebSocket client
        let symbol: SmartString = json.get("s").and_then(|v| v.as_str()).unwrap_or("").into();

        let order_id: SmartString = json
            .get("i")
            .and_then(simd_json::prelude::ValueAsScalar::as_u64)
            .unwrap_or(0)
            .to_string()
            .into();

        let client_order_id: SmartString =
            json.get("c").and_then(|v| v.as_str()).unwrap_or("").into();

        let status = json.get("X").and_then(|v| v.as_str()).unwrap_or("");

        let executed_qty: SmartString =
            json.get("z").and_then(|v| v.as_str()).unwrap_or("0").into();

        let original_qty: SmartString =
            json.get("q").and_then(|v| v.as_str()).unwrap_or("0").into();

        let price: SmartString = json.get("p").and_then(|v| v.as_str()).unwrap_or("0").into();

        let transaction_time = json
            .get("T")
            .and_then(simd_json::prelude::ValueAsScalar::as_u64)
            .unwrap_or(0);

        let report = ExecutionReport {
            id: id_generation::generate_ws_report_id("exec", &client_order_id),
            order_id: client_order_id,
            exchange_timestamp: transaction_time * 1_000_000, // Convert ms to ns
            system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            instrument_id: InstrumentId {
                symbol,
                venue: Venue::Binance,
            },
            status: Self::map_order_status(status),
            filled_quantity: Decimal::from_str_exact(&executed_qty).unwrap_or(Decimal::ZERO),
            remaining_quantity: {
                let original = Decimal::from_str_exact(&original_qty).unwrap_or(Decimal::ZERO);
                let executed = Decimal::from_str_exact(&executed_qty).unwrap_or(Decimal::ZERO);
                original - executed
            },
            execution_price: Decimal::from_str_exact(&price).ok(),
            reject_reason: None,
            exchange_execution_id: Some(order_id),
            is_final: matches!(
                Self::map_order_status(status),
                OrderStatus::Filled
                    | OrderStatus::Cancelled
                    | OrderStatus::Rejected
                    | OrderStatus::Expired
            ),
        };

        // Handle backpressure by using try_send with timeout
        match tokio::time::timeout(
            Duration::from_millis(100), // 100ms timeout
            report_tx.send_async(report),
        )
        .await
        {
            Ok(Ok(())) => debug!("Execution report sent successfully"),
            Ok(Err(e)) => {
                error!("Failed to send execution report: {e}");
                // Channel closed, nothing we can do
            }
            Err(_) => {
                warn!("Channel backpressure detected - execution report send timed out");
                // Could implement a fallback strategy here if needed
            }
        }
    }

    /// Handle account update events from user data stream
    pub async fn handle_account_update(
        json: &JsonValue,
        position_manager: &Arc<dyn PositionManager>,
    ) {
        // Extract account update information from ACCOUNT_UPDATE event
        let event_time = json
            .get("E")
            .and_then(simd_json::prelude::ValueAsScalar::as_u64)
            .unwrap_or(0);
        let transaction_time = json
            .get("T")
            .and_then(simd_json::prelude::ValueAsScalar::as_u64)
            .unwrap_or(0);

        // Handle position updates (futures)
        if let Some(positions) = json
            .get("a")
            .and_then(|a| a.get("P"))
            .and_then(|p| p.as_array())
        {
            for position_json in positions {
                if let Some(position_update) =
                    Self::parse_position_update(position_json, event_time, transaction_time)
                {
                    // Log position update
                    info!(
                        "Position update: {} {} {} @ {}",
                        position_update.symbol,
                        position_update.side,
                        position_update.amount,
                        position_update.entry_price
                    );

                    // Send position update to position manager
                    if let Err(e) = position_manager
                        .update_position(position_update.clone())
                        .await
                    {
                        error!("Failed to update position in position manager: {e}");
                    }
                }
            }
        }

        // Handle balance updates
        if let Some(balances) = json
            .get("a")
            .and_then(|a| a.get("B"))
            .and_then(|b| b.as_array())
        {
            for balance_json in balances {
                if let Some(asset) = balance_json.get("a").and_then(|v| v.as_str())
                    && let Some(wallet_balance) = balance_json.get("wb").and_then(|v| v.as_str())
                    && let Some(cross_wallet_balance) =
                        balance_json.get("cw").and_then(|v| v.as_str())
                {
                    debug!(
                        "Balance update: {asset} wallet={wallet_balance} cross_wallet={cross_wallet_balance}"
                    );
                }
            }
        }
    }

    /// Generate stable position ID based on position characteristics
    /// Uses a single hash for optimal performance in HFT scenarios
    fn generate_stable_position_id(
        symbol: &str,
        side: PositionSide,
        venue: Venue,
    ) -> rusty_model::types::PositionId {
        // Create a deterministic UUID based on position characteristics
        // This ensures the same position always gets the same ID

        use std::collections::hash_map::DefaultHasher;
        use std::hash::{Hash, Hasher};

        // Create a hasher and hash all components in order
        let mut hasher = DefaultHasher::new();

        // Hash a version prefix for future compatibility
        "v1-position".hash(&mut hasher);
        symbol.hash(&mut hasher);
        (side as u8).hash(&mut hasher);
        (venue as u8).hash(&mut hasher);

        let hash = hasher.finish();

        // Create deterministic UUID bytes from the hash
        // Use a simple mixing function to expand 64-bit hash to 128-bit UUID
        let mut uuid_bytes = [0u8; 16];

        // Lower 64 bits: original hash
        uuid_bytes[0..8].copy_from_slice(&hash.to_le_bytes());

        // Upper 64 bits: rotated and XORed version for better distribution
        let mixed = hash.rotate_left(32) ^ hash;
        uuid_bytes[8..16].copy_from_slice(&mixed.to_le_bytes());

        // Set version (4) and variant bits for valid UUID format
        uuid_bytes[6] = (uuid_bytes[6] & 0x0f) | 0x40; // Version 4
        uuid_bytes[8] = (uuid_bytes[8] & 0x3f) | 0x80; // Variant 10

        let stable_uuid = Uuid::from_bytes(uuid_bytes);
        rusty_model::types::PositionId::from_uuid(stable_uuid)
    }

    /// Parse position update from `ACCOUNT_UPDATE` event
    #[must_use]
    pub fn parse_position_update(
        position_json: &JsonValue,
        event_time: u64,
        _transaction_time: u64,
    ) -> Option<PositionUpdate> {
        // Extract position fields from JSON
        let symbol = position_json.get("s").and_then(|v| v.as_str())?;
        let position_amount_str = position_json.get("pa").and_then(|v| v.as_str())?;
        let entry_price_str = position_json.get("ep").and_then(|v| v.as_str())?;
        let breakeven_price_str = position_json.get("bep").and_then(|v| v.as_str())?;
        let unrealized_pnl_str = position_json.get("up").and_then(|v| v.as_str())?;
        let realized_pnl_str = position_json.get("cr").and_then(|v| v.as_str())?;
        let margin_type_str = position_json.get("mt").and_then(|v| v.as_str())?;
        let isolated_wallet_str = position_json.get("iw").and_then(|v| v.as_str())?;
        let position_side_str = position_json.get("ps").and_then(|v| v.as_str())?;

        // Parse numeric values
        let amount = Decimal::from_str_exact(position_amount_str).ok()?;
        let entry_price = Decimal::from_str_exact(entry_price_str).ok()?;
        let breakeven_price = Decimal::from_str_exact(breakeven_price_str).ok()?;
        let unrealized_pnl = Decimal::from_str_exact(unrealized_pnl_str).ok()?;
        let realized_pnl = Decimal::from_str_exact(realized_pnl_str).ok()?;
        let isolated_wallet = Decimal::from_str_exact(isolated_wallet_str).ok()?;

        // Parse enums
        let position_side = position_side_str.parse::<PositionSide>().ok()?;
        let margin_type = margin_type_str.parse::<MarginType>().ok()?;

        // Skip positions with zero amount (closed positions)
        if amount.is_zero() {
            debug!("Skipping closed position for {symbol}");
            return None;
        }

        // Generate stable position ID based on position characteristics
        let position_id = Self::generate_stable_position_id(symbol, position_side, Venue::Binance);

        Some(PositionUpdate {
            position_id,
            venue: Venue::Binance,
            symbol: SmartString::from(symbol),
            side: position_side,
            amount,
            entry_price,
            breakeven_price,
            unrealized_pnl,
            realized_pnl,
            margin_type,
            isolated_wallet,
            timestamp_ns: event_time * 1_000_000, // Convert ms to ns
        })
    }

    /// Map Binance order status to internal status
    fn map_order_status(status: &str) -> OrderStatus {
        match status {
            "NEW" => OrderStatus::New,
            "PARTIALLY_FILLED" => OrderStatus::PartiallyFilled,
            "FILLED" => OrderStatus::Filled,
            "CANCELED" | "CANCELLED" => OrderStatus::Cancelled,
            "REJECTED" => OrderStatus::Rejected,
            "EXPIRED" => OrderStatus::Expired,
            "PENDING_CANCEL" => OrderStatus::Pending,
            "PENDING_NEW" => OrderStatus::Pending,
            _ => OrderStatus::Unknown,
        }
    }

    /// Place batch orders via WebSocket API (up to 50 orders)
    ///
    /// This method ensures rate limiting compliance (300 orders/10 seconds)
    /// and uses concurrent execution for optimal performance.
    ///
    /// Returns Ok(()) if all orders succeed, or an error with details about failed orders
    pub async fn place_batch_orders(
        &self,
        orders: Vec<Order>,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()> {
        if orders.is_empty() {
            return Ok(());
        }

        if orders.len() > MAX_BATCH_SIZE {
            bail!("Batch size exceeds maximum of {} orders", MAX_BATCH_SIZE);
        }

        // Check rate limit
        self.check_and_record_rate_limit(orders.len())?;

        // Place orders concurrently with proper error handling
        let batch_result = self
            .place_batch_orders_concurrent(orders, report_tx)
            .await?;

        // Check if any orders failed
        if batch_result.has_failures() {
            // Log details about failures
            let failures_by_type = batch_result.get_failures_by_error_type();
            for (error_type, failed_orders) in failures_by_type {
                warn!(
                    "{} orders failed due to {}: {:?}",
                    failed_orders.len(),
                    error_type,
                    failed_orders.iter().map(|o| &o.id).collect::<Vec<_>>()
                );
            }

            // Return error if all orders failed
            if batch_result.status == BatchStatus::AllFailed {
                bail!(
                    "All {} orders in batch failed (success rate: {:.1}%)",
                    batch_result.summary.total_orders,
                    batch_result.success_rate()
                );
            }
        }

        info!(
            "Batch order placement completed: {} succeeded, {} failed (success rate: {:.1}%)",
            batch_result.summary.successful_orders,
            batch_result.summary.failed_orders,
            batch_result.success_rate()
        );

        Ok(())
    }

    /// Internal concurrent batch order placement with detailed error reporting
    async fn place_batch_orders_concurrent(
        &self,
        orders: Vec<Order>,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<BatchResult<()>> {
        use crate::error::EMSError;
        use futures::future::join_all;

        let start_time = self.clock.raw();
        let total_orders = orders.len();

        // Create concurrent tasks for each order
        let order_tasks: Vec<_> = orders
            .into_iter()
            .map(|order| {
                let report_tx = report_tx.clone();
                let client_order_id = order.id;
                let order_clone = order.clone();
                async move {
                    let result = self.place_order_internal(order_clone, report_tx).await;
                    (client_order_id, order, result)
                }
            })
            .collect();

        // Execute all orders concurrently
        let results = join_all(order_tasks).await;

        // Process results and build BatchResult
        let mut order_results: OrderResultMap<()> = FxHashMap::default();
        let mut success_count = 0;
        let mut failed_count = 0;

        for (client_order_id, order, result) in results {
            match result {
                Ok(()) => {
                    order_results
                        .insert(client_order_id.to_string().into(), OrderResult::success(()));
                    success_count += 1;
                    debug!("Order {client_order_id} placed successfully");
                }
                Err(e) => {
                    // Convert anyhow::Error to EMSError
                    let ems_error = if let Some(ems_err) = e.downcast_ref::<EMSError>() {
                        ems_err.clone()
                    } else {
                        EMSError::order_submission(e.to_string())
                    };

                    order_results.insert(
                        client_order_id.to_string().into(),
                        OrderResult::failed(ems_error, order),
                    );
                    failed_count += 1;
                    warn!("Order {client_order_id} failed: {e}");
                }
            }
        }

        let processing_time_ns = self.clock.raw() - start_time;

        // Build appropriate BatchResult based on results
        let batch_result = if failed_count == 0 {
            // All succeeded
            BatchResult::success(order_results, processing_time_ns)
        } else if success_count > 0 {
            // Partial success
            BatchResult::partial_success(order_results, processing_time_ns)
        } else {
            // All failed
            BatchResult::all_failed(order_results, processing_time_ns)
        };

        // Log summary
        match batch_result.status {
            BatchStatus::AllSucceeded => {
                info!("All {total_orders} orders placed successfully");
            }
            BatchStatus::PartialSuccess => {
                warn!("Batch partially succeeded: {success_count}/{total_orders} orders placed");
            }
            BatchStatus::AllFailed => {
                error!("All {total_orders} orders in batch failed");
            }
            BatchStatus::TransportFailure => {
                // This shouldn't happen in this implementation
                error!("Unexpected transport failure status");
            }
        }

        Ok(batch_result)
    }

    /// Check and record rate limit for order placement
    fn check_and_record_rate_limit(&self, num_orders: usize) -> Result<()> {
        let mut rate_limiter = self.order_rate_limiter.write();
        if !rate_limiter.can_place_orders(num_orders) {
            let current_count = rate_limiter.current_order_count();
            if num_orders == 1 {
                bail!(
                    "Rate limit would be exceeded: {} orders in window, limit is {}",
                    current_count,
                    MAX_ORDERS_PER_10_SECONDS
                );
            }
            bail!(
                "Rate limit would be exceeded: {} orders in window, {} requested, limit is {}",
                current_count,
                num_orders,
                MAX_ORDERS_PER_10_SECONDS
            );
        }
        rate_limiter.record_orders(num_orders);
        Ok(())
    }

    /// Place order via WebSocket API
    pub async fn place_order(
        &self,
        order: Order,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Check rate limit for single order
        self.check_and_record_rate_limit(1)?;

        self.place_order_internal(order, report_tx).await
    }

    /// Internal order placement logic without rate limiting
    async fn place_order_internal(
        &self,
        order: Order,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()> {
        let request_id = self.request_id_gen.next_id();
        let timestamp = self.clock.raw();

        // Build order request params directly as an Object for clarity and performance.
        let mut params = Object::with_capacity(8);
        params.insert("symbol".into(), order.symbol.as_str().into());
        params.insert("side".into(), Self::map_order_side(order.side).into());
        params.insert("type".into(), Self::map_order_type(order.order_type).into());
        params.insert("quantity".into(), order.quantity.to_string().into());
        params.insert("newClientOrderId".into(), order.id.to_string().into());
        params.insert("timestamp".into(), (timestamp / 1_000_000).into());

        // Add price if it's a limit order
        if let Some(price) = order.price {
            params.insert("price".into(), price.to_string().into());
        }

        // Add time in force
        let tif = Self::get_time_in_force(order.order_type);
        params.insert("timeInForce".into(), Self::map_time_in_force(tif).into());

        // Build complete request
        let request = simd_json::json!({
            "id": request_id.as_json_value(),
            "method": "order.place",
            "params": JsonValue::from(params)
        });

        // Track the pending request with string key
        {
            let mut pending = self.pending_requests.write();
            let key = request_id.to_lookup_key();
            pending.insert(
                key,
                PendingRequest {
                    method: "order.place".into(),
                    timestamp,
                    report_tx: Some(report_tx),
                    request_id: request_id.clone(),
                },
            );
        }

        // Send the request
        self.send_request(request).await?;

        Ok(())
    }

    /// Get current rate limit status
    #[must_use]
    pub fn get_rate_limit_status(&self) -> (usize, usize) {
        let mut rate_limiter = self.order_rate_limiter.write();
        let current = rate_limiter.current_order_count();
        (current, MAX_ORDERS_PER_10_SECONDS as usize)
    }

    /// Cancel order via WebSocket API
    pub async fn cancel_order(
        &self,
        order_id: SmartString,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()> {
        let request_id = self.request_id_gen.next_id();
        let timestamp = self.clock.raw();

        let request = simd_json::json!({
            "id": request_id.as_json_value(),
            "method": "order.cancel",
            "params": {
                "origClientOrderId": order_id,
                "timestamp": timestamp / 1_000_000 // Convert ns to ms
            }
        });

        // Track the pending request with string key
        {
            let mut pending = self.pending_requests.write();
            let key = request_id.to_lookup_key();
            pending.insert(
                key,
                PendingRequest {
                    method: "order.cancel".into(),
                    timestamp,
                    report_tx: Some(report_tx),
                    request_id: request_id.clone(),
                },
            );
        }

        self.send_request(request).await?;

        Ok(())
    }

    /// Modify order via WebSocket API (cancel and replace)
    pub async fn modify_order(
        &self,
        params: ModifyOrderParams,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()> {
        let request_id = self.request_id_gen.next_id();
        let timestamp = self.clock.raw();

        let mut request = simd_json::json!({
            "id": request_id.as_json_value(),
            "method": "order.cancelReplace",
            "params": {
                "cancelOrigClientOrderId": params.order_id,
                "symbol": params.symbol,
                "side": Self::map_order_side(params.side),
                "type": Self::map_order_type(params.order_type),
                "quantity": params.new_quantity.to_string(),
                "timeInForce": Self::map_time_in_force(Self::get_time_in_force(params.order_type)),
                "timestamp": timestamp / 1_000_000 // Convert ns to ms
            }
        });

        if let Some(price) = params.new_price {
            request["params"]["price"] = price.to_string().into();
        }

        // Track the pending request with string key
        {
            let mut pending = self.pending_requests.write();
            let key = request_id.to_lookup_key();
            pending.insert(
                key,
                PendingRequest {
                    method: "order.cancelReplace".into(),
                    timestamp,
                    report_tx: Some(report_tx),
                    request_id: request_id.clone(),
                },
            );
        }

        self.send_request(request).await?;

        Ok(())
    }

    /// Send request via WebSocket
    async fn send_request(&self, request: JsonValue) -> Result<()> {
        let request_str = simd_json::to_string(&request)
            .map_err(|e| anyhow!("Failed to serialize JSON request: {}", e))?;
        debug!("Sending request: {request_str}");
        let request_object = Message::Text(request_str.into()).to_frame_view();

        if let Some(ref mut ws_sink) = self.ws_sink.write().await.as_mut() {
            match ws_sink.send(request_object).await {
                Ok(()) => Ok(()),
                Err(e) => {
                    // Kill socket on send error
                    let error_msg = format!("Failed to send WebSocket request: {e}");
                    self.kill_socket(&error_msg).await;
                    Err(anyhow!(error_msg))
                }
            }
        } else {
            self.kill_socket("WebSocket not connected").await;
            Err(anyhow!("WebSocket not connected"))
        }
    }

    /// Convert internal `OrderType` to Binance order type
    const fn map_order_type(order_type: OrderType) -> &'static str {
        match order_type {
            OrderType::Market => "MARKET",
            OrderType::Limit => "LIMIT",
            OrderType::Stop => "STOP_LOSS",
            OrderType::StopLimit => "STOP_LOSS_LIMIT",
            OrderType::FillOrKill => "LIMIT",
            OrderType::ImmediateOrCancel => "LIMIT",
            OrderType::PostOnly => "LIMIT",
        }
    }

    /// Convert internal `OrderSide` to Binance side
    const fn map_order_side(side: OrderSide) -> &'static str {
        match side {
            OrderSide::Buy => "BUY",
            OrderSide::Sell => "SELL",
        }
    }

    /// Convert internal `TimeInForce` to Binance TIF
    const fn map_time_in_force(tif: TimeInForce) -> &'static str {
        match tif {
            TimeInForce::GTC => "GTC",
            TimeInForce::IOC => "IOC",
            TimeInForce::FOK => "FOK",
            _ => "GTC",
        }
    }

    /// Get appropriate time in force for order type
    const fn get_time_in_force(order_type: OrderType) -> TimeInForce {
        match order_type {
            OrderType::Market => TimeInForce::IOC,
            OrderType::Limit => TimeInForce::GTC,
            OrderType::Stop => TimeInForce::GTC,
            OrderType::StopLimit => TimeInForce::GTC,
            OrderType::FillOrKill => TimeInForce::FOK,
            OrderType::ImmediateOrCancel => TimeInForce::IOC,
            OrderType::PostOnly => TimeInForce::GTC,
        }
    }

    /// Send JSON ping to check connection
    pub async fn send_ping(&self) -> Result<()> {
        // Check if connected first
        if !self.is_connected() {
            return Err(anyhow!("WebSocket not connected - cannot send JSON ping"));
        }

        let request_id = self.request_id_gen.next_id();
        let timestamp = self.clock.raw();

        let request = simd_json::json!({
            "id": request_id.as_json_value(),
            "method": "ping",
            "params": {}
        });

        // Track the pending request with optimized key (no string allocation for sequential IDs)
        let key = request_id.to_lookup_key();
        {
            let mut pending = self.pending_requests.write();
            pending.insert(
                key.clone(),
                PendingRequest {
                    method: "ping".into(),
                    timestamp,
                    report_tx: None, // Ping doesn't need execution report
                    request_id: request_id.clone(),
                },
            );
        }

        let display_id = match &request_id {
            RequestId::Sequential(id) => id.to_string(),
            RequestId::Uuid(id) => id.to_string(),
        };
        debug!("📤 Sending JSON ping request (id: {display_id})");

        // Send with additional error handling
        match self.send_request(request).await {
            Ok(()) => {
                debug!("✅ JSON ping #{display_id} sent successfully");
                Ok(())
            }
            Err(e) => {
                // Remove from pending requests if send failed
                {
                    let mut pending = self.pending_requests.write();
                    pending.remove(&key);
                }

                // Mark as disconnected if we get a broken pipe
                if e.to_string().contains("Broken pipe") {
                    warn!("🔌 WebSocket connection broken, marking as disconnected");
                }

                Err(e)
            }
        }
    }

    /// Check if connected
    #[must_use]
    pub fn is_connected(&self) -> bool {
        let state = self.get_state();
        matches!(state, ConnectionState::Authenticated)
    }

    /// Kill the socket connection and clean up resources
    /// This allows background tasks to detect the disconnection and spawn a new socket
    async fn kill_socket(&self, reason: &str) {
        warn!("🗡️  KILL-STEP 1/9: Killing WebSocket connection: {reason}");

        // Transition to disconnecting state
        self.set_state(ConnectionState::Disconnecting);
        info!("🗡️  KILL-STEP 2/9: State set to Disconnecting");

        // Step 1: Signal reconnection monitor to stop gracefully
        info!("🗡️  KILL-STEP 3/9: Signaling reconnection monitor to stop...");
        self.reconnection_control
            .should_stop
            .store(true, Ordering::Release);

        // Step 2: Cancel other background tasks
        info!("🗡️  KILL-STEP 4/9: Cancelling background tasks...");
        let handles = {
            let mut handles = self.task_handles.write().await;
            TaskHandles {
                response_handler: handles.response_handler.take(),
                ping_handler: handles.ping_handler.take(),
                reconnection_monitor: handles.reconnection_monitor.take(),
            }
        };

        // Cancel tasks by aborting them
        if let Some(handle) = handles.response_handler {
            handle.abort();
            debug!("Response handler task aborted");
        }
        if let Some(handle) = handles.ping_handler {
            handle.abort();
            debug!("Ping handler task aborted");
        }

        // Wait a bit for reconnection monitor to exit gracefully
        if let Some(handle) = handles.reconnection_monitor {
            match tokio::time::timeout(Duration::from_millis(100), handle).await {
                Ok(_) => debug!("Reconnection monitor exited gracefully"),
                Err(_) => {
                    debug!("Reconnection monitor didn't exit in time, continuing anyway");
                    // The monitor will exit on its next check interval
                }
            }
        }

        info!("🗡️  KILL-STEP 5/9: Tasks cancelled");

        // Step 3: Close WebSocket sink first (sends close frame)
        info!("🗡️  KILL-STEP 6/9: Closing WebSocket sink...");
        if let Some(mut ws_sink) = self.ws_sink.write().await.take() {
            // Add timeout to sink close to prevent hanging
            match tokio::time::timeout(Duration::from_secs(2), ws_sink.close()).await {
                Ok(Ok(())) => {
                    info!("🗡️  KILL-STEP 6/9: WebSocket sink closed successfully");
                }
                Ok(Err(e)) => {
                    warn!("🗡️  KILL-STEP 6/9: WebSocket sink close failed: {e}");
                }
                Err(_) => {
                    warn!("🗡️  KILL-STEP 6/9: WebSocket sink close timed out - continuing anyway");
                }
            }
        } else {
            info!("🗡️  KILL-STEP 6/9: No WebSocket sink to close");
        }

        // Step 4: Clean up WebSocket stream
        info!("🗡️  KILL-STEP 7/9: Cleaning up WebSocket stream...");
        match tokio::time::timeout(Duration::from_secs(1), async {
            *self.ws_stream.write().await = None;
        })
        .await
        {
            Ok(()) => {
                info!("🗡️  KILL-STEP 7/9: WebSocket stream cleaned up successfully");
            }
            Err(_) => {
                warn!("🗡️  KILL-STEP 7/9: WebSocket stream cleanup timed out - continuing anyway");
            }
        }

        // Step 5: Clear any pending requests to prevent memory leaks
        info!("🗡️  KILL-STEP 8/9: Clearing pending requests...");
        match tokio::time::timeout(Duration::from_secs(1), async {
            let mut pending = self.pending_requests.write();
            // Check if we have too many pending requests (possible memory leak)
            if pending.len() > MAX_PENDING_REQUESTS {
                warn!(
                    "Detected {} pending requests - possible memory leak",
                    pending.len()
                );
            }
            pending.clear();
        })
        .await
        {
            Ok(()) => {
                info!("🗡️  KILL-STEP 8/9: Pending requests cleared successfully");
            }
            Err(_) => {
                warn!("🗡️  KILL-STEP 8/9: Pending requests cleanup timed out - continuing anyway");
            }
        }

        // Step 6: Reset state
        info!("🗡️  KILL-STEP 9/9: Resetting state...");
        self.last_ping_time.store(0, Ordering::Release);
        self.last_pong_time.store(0, Ordering::Release);
        self.auth_completed_time.store(0, Ordering::Release);
        self.set_state(ConnectionState::Disconnected);

        info!(
            "🗡️  ✅ WebSocket connection killed and resources cleaned up - ready for reconnection"
        );
    }

    /// Attempt to reconnect WebSocket (can be called by external monitoring tasks)
    /// Note: This will start a new auto-reconnection monitor, so use sparingly
    pub async fn reconnect(&self, report_tx: Sender<ExecutionReport>) -> Result<()> {
        info!("External reconnection requested...");

        // Kill existing connection first
        self.kill_socket("External reconnection requested").await;

        // Small delay to allow cleanup
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Attempt new connection (this will start a new auto-reconnection monitor)
        self.connect(report_tx).await
    }

    /// Gracefully disconnect from WebSocket
    pub async fn disconnect(&self) -> Result<()> {
        info!("Gracefully disconnecting from Binance WebSocket...");

        // Use kill_socket for proper cleanup
        self.kill_socket("Graceful disconnect requested").await;

        // Cancel the reconnection monitor to prevent auto-reconnection
        if let Some(handle) = self.task_handles.write().await.reconnection_monitor.take() {
            handle.abort();
            info!("Reconnection monitor cancelled");
        }

        info!("Disconnected from Binance WebSocket");
        Ok(())
    }
}

#[async_trait]
impl crate::execution_engine::Exchange for BinanceWebSocketTrader {
    fn venue(&self) -> Venue {
        Venue::Binance
    }

    async fn place_order(&self, order: Order, report_tx: Sender<ExecutionReport>) -> Result<()> {
        // Check rate limit for single order
        self.check_and_record_rate_limit(1)?;

        self.place_order_internal(order, report_tx).await
    }

    async fn cancel_order(
        &self,
        order_id: SmartString,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()> {
        let request_id = self.request_id_gen.next_id();
        let timestamp = self.clock.raw();

        let mut params = Object::with_capacity(2);
        params.insert("origClientOrderId".into(), order_id.as_str().into());
        params.insert("timestamp".into(), (timestamp / 1_000_000).into());

        // Build complete request
        let request = simd_json::json!({
            "id": request_id.as_json_value(),
            "method": "order.cancel",
            "params": JsonValue::from(params)
        });

        // Track the pending request
        {
            let mut pending = self.pending_requests.write();
            let key = request_id.to_lookup_key();
            pending.insert(
                key,
                PendingRequest {
                    method: "order.cancel".into(),
                    timestamp,
                    report_tx: Some(report_tx),
                    request_id: request_id.clone(),
                },
            );
        }

        self.send_request(request).await
    }

    async fn modify_order(
        &self,
        order_id: SmartString,
        new_price: Option<Decimal>,
        new_quantity: Option<Decimal>,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()> {
        let request_id = self.request_id_gen.next_id();
        let timestamp = self.clock.raw();

        let mut params = Object::with_capacity(4);
        params.insert("origClientOrderId".into(), order_id.as_str().into());
        params.insert("timestamp".into(), (timestamp / 1_000_000).into());

        if let Some(price) = new_price {
            params.insert("price".into(), price.to_string().into());
        }
        if let Some(quantity) = new_quantity {
            params.insert("quantity".into(), quantity.to_string().into());
        }

        // Build complete request
        let request = simd_json::json!({
            "id": request_id.as_json_value(),
            "method": "order.cancelReplace",
            "params": JsonValue::from(params)
        });

        // Track the pending request
        {
            let mut pending = self.pending_requests.write();
            let key = request_id.to_lookup_key();
            pending.insert(
                key,
                PendingRequest {
                    method: "order.cancelReplace".into(),
                    timestamp,
                    report_tx: Some(report_tx),
                    request_id: request_id.clone(),
                },
            );
        }

        self.send_request(request).await
    }

    async fn cancel_all_orders(
        &self,
        instrument_id: Option<InstrumentId>,
        report_tx: Sender<ExecutionReport>,
    ) -> Result<()> {
        let request_id = self.request_id_gen.next_id();
        let timestamp = self.clock.raw();

        // Binance WebSocket API requires symbol parameter for openOrders.cancelAll
        let instrument = instrument_id.ok_or_else(|| {
            anyhow!("Binance openOrders.cancelAll requires a symbol parameter - instrument_id cannot be None")
        })?;

        let mut params = Object::with_capacity(2);
        params.insert("timestamp".into(), (timestamp / 1_000_000).into());
        params.insert("symbol".into(), instrument.symbol.as_str().into());

        // Build complete request
        let request = simd_json::json!({
            "id": request_id.as_json_value(),
            "method": "openOrders.cancelAll",
            "params": JsonValue::from(params)
        });

        // Track the pending request
        {
            let mut pending = self.pending_requests.write();
            let key = request_id.to_lookup_key();
            pending.insert(
                key,
                PendingRequest {
                    method: "openOrders.cancelAll".into(),
                    timestamp,
                    report_tx: Some(report_tx),
                    request_id: request_id.clone(),
                },
            );
        }

        self.send_request(request).await
    }

    async fn get_order_status(&self, order_id: &str) -> Result<OrderStatus> {
        // This method is typically implemented via REST API for Binance
        // For WebSocket-only, we'd need to subscribe to order updates or query via a different method
        // For now, return an error indicating not supported
        Err(anyhow!(
            "get_order_status not supported via WebSocket for Binance"
        ))
    }

    async fn connect(&self, report_sender: Sender<ExecutionReport>) -> Result<()> {
        self.connect_without_monitor(report_sender).await
    }

    async fn disconnect(&self) -> Result<()> {
        self.disconnect().await
    }

    async fn is_connected(&self) -> bool {
        self.get_state() == ConnectionState::Authenticated
    }

    async fn get_instruments(&self) -> Result<SmallVec<[InstrumentId; 32]>> {
        // This is typically fetched via REST API. For WebSocket-only, it's not directly available.
        // Return an empty list or fetch from a cached source if available.
        Ok(SmallVec::new())
    }

    async fn send_fix_message(&self, message: Vec<u8>) -> Result<()> {
        log::warn!(
            "BinanceWebSocketTrader: send_fix_message not implemented. Message: {:?}",
            message
        );
        Err(anyhow!(
            "FIX message sending not implemented for Binance WebSocket"
        ))
    }

    async fn receive_fix_message(&self) -> Result<Vec<u8>> {
        log::warn!("BinanceWebSocketTrader: receive_fix_message not implemented.");
        Err(anyhow!(
            "FIX message receiving not implemented for Binance WebSocket"
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::position_manager::{DefaultPositionManager, PositionManager};
    use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};

    /// Create a mock position manager for testing
    fn create_mock_position_manager() -> Arc<dyn PositionManager> {
        Arc::new(DefaultPositionManager::new(None))
    }

    #[tokio::test]
    async fn test_websocket_trader_creation() {
        let api_key = SmartString::from("test_api_key");
        let private_key = BASE64.encode([1u8; 32]); // Test key

        let auth = Arc::new(BinanceAuth::new_ed25519(api_key, private_key.into()).unwrap());
        let trader = BinanceWebSocketTrader::new(auth, create_mock_position_manager());

        assert!(!trader.is_connected());

        // Test request ID generation
        let first_id = trader.request_id_gen.next_id();
        assert!(
            matches!(first_id, RequestId::Sequential(1)),
            "Expected sequential ID 1, got {first_id:?}"
        );

        let second_id = trader.request_id_gen.next_id();
        assert!(
            matches!(second_id, RequestId::Sequential(2)),
            "Expected sequential ID 2, got {second_id:?}"
        );
    }

    #[test]
    fn test_stable_position_id_generation() {
        // Test that the same position characteristics generate the same ID
        let symbol = "BTCUSDT";
        let side = PositionSide::Long;
        let venue = Venue::Binance;

        let id1 = BinanceWebSocketTrader::generate_stable_position_id(symbol, side, venue);
        let id2 = BinanceWebSocketTrader::generate_stable_position_id(symbol, side, venue);

        assert_eq!(
            id1, id2,
            "Same position characteristics should generate same ID"
        );

        // Test that different characteristics generate different IDs
        let different_symbol_id =
            BinanceWebSocketTrader::generate_stable_position_id("ETHUSDT", side, venue);
        let different_side_id =
            BinanceWebSocketTrader::generate_stable_position_id(symbol, PositionSide::Short, venue);

        assert_ne!(
            id1, different_symbol_id,
            "Different symbols should generate different IDs"
        );
        assert_ne!(
            id1, different_side_id,
            "Different sides should generate different IDs"
        );

        // Test that IDs are deterministic across runs
        let id3 = BinanceWebSocketTrader::generate_stable_position_id(symbol, side, venue);
        assert_eq!(id1, id3, "ID generation should be deterministic");
    }

    #[test]
    fn test_order_type_mapping() {
        assert_eq!(
            BinanceWebSocketTrader::map_order_type(OrderType::Market),
            "MARKET"
        );
        assert_eq!(
            BinanceWebSocketTrader::map_order_type(OrderType::Limit),
            "LIMIT"
        );
        assert_eq!(
            BinanceWebSocketTrader::map_order_side(OrderSide::Buy),
            "BUY"
        );
        assert_eq!(
            BinanceWebSocketTrader::map_order_side(OrderSide::Sell),
            "SELL"
        );
    }

    #[test]
    fn test_order_status_mapping() {
        assert_eq!(
            BinanceWebSocketTrader::map_order_status("NEW"),
            OrderStatus::New
        );
        assert_eq!(
            BinanceWebSocketTrader::map_order_status("FILLED"),
            OrderStatus::Filled
        );
        assert_eq!(
            BinanceWebSocketTrader::map_order_status("CANCELLED"),
            OrderStatus::Cancelled
        );
    }

    #[tokio::test]
    async fn test_ping_handler_lifecycle() {
        let auth = Arc::new(BinanceAuth::new_hmac(
            "test_key".into(),
            "test_secret".into(),
        ));
        let (report_tx, _report_rx) = flume::bounded::<ExecutionReport>(100);
        let trader = BinanceWebSocketTrader::new_testnet(auth, create_mock_position_manager());

        // Check initial state
        assert_eq!(trader.get_state(), ConnectionState::Disconnected);
        assert_eq!(
            trader.metrics.websocket_pings_sent.load(Ordering::Relaxed),
            0
        );
        assert_eq!(
            trader
                .metrics
                .websocket_pongs_received
                .load(Ordering::Relaxed),
            0
        );

        // Verify ping handler starts properly (without actual connection)
        let result = trader.start_ping_handler().await;
        assert!(result.is_ok());
        let handle = result.unwrap();

        // Let it run briefly
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Abort the handler
        handle.abort();
    }

    #[tokio::test]
    async fn test_json_ping_not_connected() {
        let auth = Arc::new(BinanceAuth::new_hmac(
            "test_key".into(),
            "test_secret".into(),
        ));
        let (report_tx, _report_rx) = flume::bounded::<ExecutionReport>(100);
        let trader = BinanceWebSocketTrader::new_testnet(auth, create_mock_position_manager());

        // Try to send ping when not connected
        let result = trader.send_ping().await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("WebSocket not connected")
        );
    }

    #[test]
    fn test_connection_metrics() {
        let auth = Arc::new(BinanceAuth::new_hmac(
            "test_key".into(),
            "test_secret".into(),
        ));
        let (report_tx, _report_rx) = flume::bounded::<ExecutionReport>(100);
        let trader = BinanceWebSocketTrader::new_testnet(auth, create_mock_position_manager());

        // Test initial metrics
        let metrics = &trader.metrics;
        assert_eq!(metrics.websocket_pings_sent.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.websocket_pongs_received.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.json_pings_sent.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.json_pongs_received.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.messages_sent.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.messages_received.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.reconnection_attempts.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.successful_reconnections.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.failed_reconnections.load(Ordering::Relaxed), 0);

        // Test metric updates
        metrics.websocket_pings_sent.fetch_add(1, Ordering::Relaxed);
        assert_eq!(metrics.websocket_pings_sent.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn test_ping_pong_timeout_calculation() {
        // Verify ping/pong constants
        assert_eq!(PING_INTERVAL_SECONDS, 30);
        assert_eq!(PONG_TIMEOUT_SECONDS, 10);

        // Ensure timeout is reasonable compared to interval
        // This is a compile-time check - the constants are validated at compile time
        const _: () = assert!(PONG_TIMEOUT_SECONDS < PING_INTERVAL_SECONDS);
    }

    #[test]
    fn test_batch_size_constants() {
        // Verify new batch size limits
        assert_eq!(MAX_BATCH_SIZE, 50);
        assert_eq!(MAX_ORDERS_PER_10_SECONDS, 300);

        // Ensure batch size is reasonable compared to rate limit
        assert!(MAX_BATCH_SIZE <= MAX_ORDERS_PER_10_SECONDS as usize);

        // Ensure we can fit multiple batches within rate limit window
        assert!(MAX_BATCH_SIZE * 5 < MAX_ORDERS_PER_10_SECONDS as usize);
    }

    #[tokio::test]
    async fn test_batch_orders_empty_list() {
        let auth = Arc::new(BinanceAuth::new_hmac(
            "test_key".into(),
            "test_secret".into(),
        ));
        let trader = BinanceWebSocketTrader::new(auth, create_mock_position_manager());
        let (report_tx, _report_rx) = flume::bounded(100);

        // Empty batch should succeed without error
        let result = trader.place_batch_orders(vec![], report_tx).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_batch_orders_exceeds_limit() {
        let auth = Arc::new(BinanceAuth::new_hmac(
            "test_key".into(),
            "test_secret".into(),
        ));
        let trader = BinanceWebSocketTrader::new(auth, create_mock_position_manager());
        let (report_tx, _report_rx) = flume::bounded(100);

        // Create orders exceeding MAX_BATCH_SIZE
        let mut orders = Vec::new();
        for i in 0..=MAX_BATCH_SIZE {
            let order = Order::new(
                Venue::Binance,
                "BTCUSDT",
                OrderSide::Buy,
                rusty_model::enums::OrderType::Limit,
                Decimal::ONE,
                Some(Decimal::from(50000)),
                rusty_model::types::ClientId::new(format!("test_client_{i}")),
            );
            orders.push(order);
        }

        // Should fail with batch size exceeded error
        let result = trader.place_batch_orders(orders, report_tx).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("exceeds maximum"));
    }

    #[test]
    fn test_rate_limiter_functionality() {
        let auth = Arc::new(BinanceAuth::new_hmac(
            "test_key".into(),
            "test_secret".into(),
        ));
        let trader = BinanceWebSocketTrader::new(auth, create_mock_position_manager());

        // Test rate limit status through trader interface
        let (current, limit) = trader.get_rate_limit_status();
        assert_eq!(current, 0); // Initially empty
        assert_eq!(limit, MAX_ORDERS_PER_10_SECONDS as usize);

        // Test direct rate limiter functionality
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock);

        // Should allow orders within limit
        assert!(rate_limiter.can_place_orders(100));
        rate_limiter.record_orders(100);

        // Should allow more orders up to limit
        assert!(rate_limiter.can_place_orders(200));
        rate_limiter.record_orders(200);

        // Should deny orders exceeding limit
        assert!(!rate_limiter.can_place_orders(1));

        // Test current count reporting
        let current = rate_limiter.current_order_count();
        assert_eq!(current, 300);
    }

    #[test]
    fn test_order_rate_limiter_new() {
        let clock = Clock::new();
        let rate_limiter = OrderRateLimiter::new(clock.clone());

        // Should start with empty order times
        assert_eq!(rate_limiter.order_times.len(), 0);

        // Clock should be set correctly
        assert_eq!(
            rate_limiter.clock.raw() / 1_000_000,
            clock.raw() / 1_000_000
        );
    }

    #[test]
    fn test_order_rate_limiter_can_place_orders_basic() {
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock);

        // Should be able to place orders when empty
        assert!(rate_limiter.can_place_orders(1));
        assert!(rate_limiter.can_place_orders(10));
        assert!(rate_limiter.can_place_orders(MAX_ORDERS_PER_10_SECONDS as usize));

        // Should not be able to place more than max orders
        assert!(!rate_limiter.can_place_orders(MAX_ORDERS_PER_10_SECONDS as usize + 1));
    }

    #[test]
    fn test_order_rate_limiter_can_place_orders_with_existing() {
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock.clone());

        // Add some existing orders
        let now = clock.raw() / 1_000_000;
        rate_limiter.order_times.push_back(now);
        rate_limiter.order_times.push_back(now);

        // Should account for existing orders
        assert!(rate_limiter.can_place_orders(1));
        assert!(rate_limiter.can_place_orders(MAX_ORDERS_PER_10_SECONDS as usize - 2));
        assert!(!rate_limiter.can_place_orders(MAX_ORDERS_PER_10_SECONDS as usize - 1));
    }

    #[test]
    fn test_order_rate_limiter_record_orders() {
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock);

        // Record single order
        rate_limiter.record_orders(1);
        assert_eq!(rate_limiter.order_times.len(), 1);

        // Record multiple orders
        rate_limiter.record_orders(5);
        assert_eq!(rate_limiter.order_times.len(), 6);

        // All recorded orders should have similar timestamps
        let first_time = rate_limiter.order_times[0];
        let last_time = rate_limiter.order_times[5];
        assert_eq!(first_time, last_time); // Should be same millisecond
    }

    #[test]
    fn test_order_rate_limiter_get_current_usage() {
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock);

        // Initial usage should be 0
        let (current, limit) = rate_limiter.get_current_usage();
        assert_eq!(current, 0);
        assert_eq!(limit, MAX_ORDERS_PER_10_SECONDS as usize);

        // Record some orders
        rate_limiter.record_orders(10);
        let (current, limit) = rate_limiter.get_current_usage();
        assert_eq!(current, 10);
        assert_eq!(limit, MAX_ORDERS_PER_10_SECONDS as usize);
    }

    #[test]
    fn test_order_rate_limiter_current_order_count() {
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock);

        // Initial count should be 0
        assert_eq!(rate_limiter.current_order_count(), 0);

        // Record some orders
        rate_limiter.record_orders(7);
        assert_eq!(rate_limiter.current_order_count(), 7);

        // Record more orders
        rate_limiter.record_orders(3);
        assert_eq!(rate_limiter.current_order_count(), 10);
    }

    #[test]
    fn test_rate_limiter_window_cleanup() {
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock.clone());

        // Add orders that should expire
        rate_limiter.order_times.push_back(0); // Very old timestamp
        rate_limiter.order_times.push_back(1000); // Old timestamp

        // Add current orders
        let now = clock.raw() / 1_000_000;
        rate_limiter.order_times.push_back(now);
        rate_limiter.order_times.push_back(now);

        // Check should clean up old orders
        assert!(rate_limiter.can_place_orders(1));

        // Should only have current orders remaining
        assert_eq!(rate_limiter.order_times.len(), 2);
    }

    #[test]
    fn test_order_rate_limiter_edge_cases() {
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock);

        // Test with zero count
        assert!(rate_limiter.can_place_orders(0));
        rate_limiter.record_orders(0);
        assert_eq!(rate_limiter.current_order_count(), 0);

        // Test at exact limit
        let max_orders = MAX_ORDERS_PER_10_SECONDS as usize;
        assert!(rate_limiter.can_place_orders(max_orders));
        rate_limiter.record_orders(max_orders);
        assert_eq!(rate_limiter.current_order_count(), max_orders);

        // Should not be able to place any more
        assert!(!rate_limiter.can_place_orders(1));
    }

    #[test]
    fn test_order_rate_limiter_window_sliding() {
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock.clone());

        // Add orders at different times within the window
        let now = clock.raw() / 1_000_000;
        let window_start = now.saturating_sub(RATE_LIMIT_WINDOW_MS);

        // Add orders at start of window (should still be valid)
        rate_limiter.order_times.push_back(window_start + 1000); // 1 second into window
        rate_limiter.order_times.push_back(window_start + 5000); // 5 seconds into window

        // Add orders at end of window (current time)
        rate_limiter.order_times.push_back(now);

        // All orders should be counted
        assert_eq!(rate_limiter.current_order_count(), 3);

        // Should be able to place more orders up to limit
        assert!(rate_limiter.can_place_orders(MAX_ORDERS_PER_10_SECONDS as usize - 3));
    }

    #[test]
    fn test_order_rate_limiter_cleanup_all_methods() {
        let clock = Clock::new();
        let mut rate_limiter = OrderRateLimiter::new(clock);

        // Add very old orders that should be cleaned up
        rate_limiter.order_times.push_back(0);
        rate_limiter.order_times.push_back(1000);
        rate_limiter.order_times.push_back(2000);

        // Test that all methods clean up old orders
        assert!(rate_limiter.can_place_orders(1));
        assert_eq!(rate_limiter.order_times.len(), 0); // Should be cleaned up

        // Add old orders again
        rate_limiter.order_times.push_back(0);
        rate_limiter.order_times.push_back(1000);

        let count = rate_limiter.current_order_count();
        assert_eq!(count, 0); // Should be cleaned up
        assert_eq!(rate_limiter.order_times.len(), 0);

        // Add old orders again
        rate_limiter.order_times.push_back(0);
        rate_limiter.order_times.push_back(1000);

        let (current, _) = rate_limiter.get_current_usage();
        assert_eq!(current, 0); // Should be cleaned up
        assert_eq!(rate_limiter.order_times.len(), 0);
    }

    #[tokio::test]
    async fn test_concurrent_batch_error_handling() {
        let auth = Arc::new(BinanceAuth::new_hmac(
            "test_key".into(),
            "test_secret".into(),
        ));
        let trader = BinanceWebSocketTrader::new(auth, create_mock_position_manager());
        let (report_tx, _report_rx) = flume::bounded(100);

        // Create a small batch that won't trigger rate limits
        let mut orders = Vec::new();
        for i in 0..3 {
            let order = Order::new(
                Venue::Binance,
                "BTCUSDT",
                OrderSide::Buy,
                rusty_model::enums::OrderType::Limit,
                Decimal::ONE,
                Some(Decimal::from(50000)),
                rusty_model::types::ClientId::new(format!("test_client_{i}")),
            );
            orders.push(order);
        }

        // This will fail because we're not connected, but should test the concurrent logic
        let result = trader.place_batch_orders(orders, report_tx).await;
        assert!(result.is_err());

        // Error should mention batch failure rather than individual order failure
        let error_msg = result.unwrap_err().to_string();
        assert!(error_msg.contains("batch") || error_msg.contains("All orders"));
    }

    #[test]
    fn test_batch_size_increase() {
        // Verify that we increased batch size from 5 to 50
        assert_eq!(MAX_BATCH_SIZE, 50);

        // Should fit well within rate limits
        assert!(MAX_BATCH_SIZE < MAX_ORDERS_PER_10_SECONDS as usize / 2);
    }
}