rusty_ems/exchanges/bybit/

websocket_trading.rs

use anyhow::{Result, anyhow, bail};
use async_trait::async_trait;
use flume::Sender;
use rust_decimal::Decimal;
use simd_json::prelude::*;
use simd_json::{json, value::owned::Value as JsonValue};
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use futures::{SinkExt, StreamExt};
use log::{debug, error, info, warn};
use parking_lot::RwLock;
use quanta::Clock;
use rusty_common::collections::FxHashMap;
use rusty_common::websocket::Message;
use rusty_common::websocket::connector::{WebSocketSink, WebSocketStream};
use rusty_common::{SmartString, time};
use rusty_model::{
    enums::{OrderSide, OrderStatus, OrderType, TimeInForce},
    instruments::InstrumentId,
    trading_order::Order,
    venues::Venue,
};
use smallvec::SmallVec;
use tokio::sync::RwLock as AsyncRwLock;
use tokio::task::JoinHandle;
use tokio::time::interval;
use uuid::Uuid;

// Removed unused import
use crate::exchanges::bybit_rest::{BybitAccountInfo, BybitAccountType, BybitRestClient};
use crate::execution_engine::{Exchange, ExecutionReport};
use rusty_common::auth::exchanges::bybit::BybitAuth;

// Bybit WebSocket Trading Implementation
//
// Provides WebSocket trading functionality for Bybit exchange with support for:
// - V5 unified WebSocket API
// - Order placement, amendment, and cancellation via WebSocket
// - Batch order operations
// - TP/SL order management
// - Real-time order status updates
// - Automatic reconnection with exponential backoff
// - Connection health monitoring
//
// # Features
//
// - **Unified Trading**: Support for spot, linear, inverse, and options
// - **Batch Operations**: Up to 20 orders per batch request
// - **Order Amendment**: Modify price and quantity of open orders
// - **Account Types**: Handles both UNIFIED and CONTRACT accounts
// - **Rate Limiting**: Built-in rate limit handling (3000 reqs/sec per IP)
// - **Automatic Reconnection**: Configurable backoff strategy

/// WebSocket API URLs for different environments
const BYBIT_WS_TRADE_URL: &str = "wss://stream.bybit.com/v5/trade";
const BYBIT_WS_TRADE_TESTNET_URL: &str = "wss://stream-testnet.bybit.com/v5/trade";
const BYBIT_WS_PRIVATE_URL: &str = "wss://stream.bybit.com/v5/private";
const BYBIT_WS_PRIVATE_TESTNET_URL: &str = "wss://stream-testnet.bybit.com/v5/private";

/// Connection management constants
const PING_INTERVAL_SECONDS: u64 = 30;
const PONG_TIMEOUT_SECONDS: u64 = 10;
const MAX_MESSAGE_SIZE: usize = 10 * 1024 * 1024; // 10MB
const MAX_PENDING_REQUESTS: usize = 10000;
const MAX_RECONNECTION_ATTEMPTS: u32 = 10;

/// Rate limiting constants
const MAX_REQUESTS_PER_SECOND: u32 = 3000; // IP rate limit
const RATE_LIMIT_WINDOW_MS: u64 = 1_000; // 1 second

/// Bybit product categories
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BybitCategory {
    /// Spot trading category
    Spot,
    /// Linear futures/perpetuals category
    Linear,
    /// Inverse futures/perpetuals category
    Inverse,
    /// Options trading category
    Option,
}

impl BybitCategory {
    /// Parse from string representation
    #[must_use]
    pub fn parse_category(s: &str) -> Option<Self> {
        match s {
            "spot" => Some(Self::Spot),
            "linear" => Some(Self::Linear),
            "inverse" => Some(Self::Inverse),
            "option" => Some(Self::Option),
            _ => None,
        }
    }

    /// Convert to string representation
    #[must_use]
    pub const fn as_str(&self) -> &'static str {
        match self {
            Self::Spot => "spot",
            Self::Linear => "linear",
            Self::Inverse => "inverse",
            Self::Option => "option",
        }
    }

    /// Get maximum batch size for this category
    #[must_use]
    pub const fn max_batch_size(&self) -> usize {
        match self {
            Self::Spot => 10,
            Self::Linear => 20,
            Self::Inverse => 20,
            Self::Option => 20,
        }
    }
}

/// Instrument information cache entry
#[derive(Debug, Clone)]
pub struct InstrumentInfo {
    /// Symbol identifier
    pub symbol: SmartString,
    /// Trading category (spot, linear, inverse, option)
    pub category: BybitCategory,
    /// Trading status (Trading, Settled, etc.)
    pub status: SmartString,
    /// Base currency or coin
    pub base_coin: SmartString,
    /// Quote currency or coin
    pub quote_coin: SmartString,
    /// Contract type for derivatives (LinearPerpetual, etc.)
    pub contract_type: Option<SmartString>,
    /// Timestamp when cached (nanoseconds)
    pub cached_at: u64, // Timestamp when cached
}

impl InstrumentInfo {
    /// Create new instrument info
    #[must_use]
    pub fn new(
        symbol: SmartString,
        category: BybitCategory,
        status: SmartString,
        base_coin: SmartString,
        quote_coin: SmartString,
        contract_type: Option<SmartString>,
    ) -> Self {
        Self {
            symbol,
            category,
            status,
            base_coin,
            quote_coin,
            contract_type,
            cached_at: rusty_common::time::get_epoch_timestamp_ns(),
        }
    }

    /// Check if cached entry is expired (older than 1 hour)
    #[must_use]
    pub fn is_expired(&self) -> bool {
        let now = rusty_common::time::get_epoch_timestamp_ns();
        let one_hour_ns = 3600 * 1_000_000_000;
        now.saturating_sub(self.cached_at) > one_hour_ns
    }
}

/// Batch order limits per category
const MAX_BATCH_SPOT: usize = 10;
const MAX_BATCH_LINEAR: usize = 20;
const MAX_BATCH_INVERSE: usize = 20;
const MAX_BATCH_OPTION: usize = 20;

/// Order categories - backward compatibility constants
const CATEGORY_SPOT: &str = "spot";
const CATEGORY_LINEAR: &str = "linear";
const CATEGORY_INVERSE: &str = "inverse";
const CATEGORY_OPTION: &str = "option";

/// Connection health tracking
#[derive(Debug, Clone)]
pub struct ConnectionHealth {
    /// Whether the connection is healthy
    pub is_healthy: bool,
    /// Timestamp of last ping sent (nanoseconds)
    pub last_ping_sent: u64,
    /// Timestamp of last pong received (nanoseconds)
    pub last_pong_received: u64,
    /// Number of pending ping requests
    pub pending_pings: u32,
    /// Number of reconnection attempts made
    pub reconnection_attempts: u32,
    /// Total messages sent
    pub messages_sent: u64,
    /// Total messages received
    pub messages_received: u64,
}

impl ConnectionHealth {
    const fn new() -> Self {
        Self {
            is_healthy: false,
            last_ping_sent: 0,
            last_pong_received: 0,
            pending_pings: 0,
            reconnection_attempts: 0,
            messages_sent: 0,
            messages_received: 0,
        }
    }
}

/// Rate limiter for request throttling
struct RateLimiter {
    request_times: VecDeque<u64>,
    clock: Clock,
}

impl RateLimiter {
    const fn new(clock: Clock) -> Self {
        Self {
            request_times: VecDeque::new(),
            clock,
        }
    }

    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.request_times.front() {
            if front < window_start {
                self.request_times.pop_front();
            } else {
                break;
            }
        }
    }

    fn can_send_request(&mut self) -> bool {
        self.cleanup_expired();
        self.request_times.len() < MAX_REQUESTS_PER_SECOND as usize
    }

    fn record_request(&mut self) {
        let now = self.clock.raw() / 1_000_000;
        self.request_times.push_back(now);
    }
}

/// Pending request tracking
struct PendingRequest {
    op_type: SmartString,
    timestamp: u64,
    report_sender: Sender<ExecutionReport>,
}

/// Bybit WebSocket trading client
///
/// Provides high-performance WebSocket trading functionality for Bybit exchange
/// with support for V5 unified API, batch operations, and connection health monitoring.
pub struct BybitWebsocketTrading {
    /// Authentication handler
    auth: Arc<BybitAuth>,

    /// Optional REST client for account info and validation (V5 features)
    rest_client: Option<Arc<BybitRestClient>>,

    /// WebSocket connections
    trade_sink: Arc<AsyncRwLock<Option<WebSocketSink>>>,
    trade_stream: Arc<AsyncRwLock<Option<WebSocketStream>>>,
    private_sink: Arc<AsyncRwLock<Option<WebSocketSink>>>,
    private_stream: Arc<AsyncRwLock<Option<WebSocketStream>>>,

    /// Configuration
    account_type: SmartString,
    testnet: bool,

    /// V5 Account info cache
    account_info: Arc<RwLock<Option<BybitAccountInfo>>>,

    /// Connection state
    is_connected: Arc<AtomicBool>,
    is_authenticated: Arc<AtomicBool>,

    /// Monitoring
    clock: Clock,
    health: Arc<RwLock<ConnectionHealth>>,
    rate_limiter: Arc<RwLock<RateLimiter>>,

    /// Request tracking
    pending_requests: Arc<RwLock<FxHashMap<SmartString, PendingRequest>>>,

    /// Order tracking: maps `order_id` -> symbol
    order_symbol_map: Arc<RwLock<FxHashMap<SmartString, SmartString>>>,

    /// Instrument info cache: maps symbol -> instrument info
    instrument_cache: Arc<RwLock<FxHashMap<SmartString, InstrumentInfo>>>,

    /// Task handles
    ping_handle: Arc<RwLock<Option<JoinHandle<()>>>>,
    message_handle: Arc<RwLock<Option<JoinHandle<()>>>>,

    /// Execution report channel
    report_sender: Arc<RwLock<Option<Sender<ExecutionReport>>>>,
}

impl BybitWebsocketTrading {
    /// Create a new Bybit WebSocket trading instance
    pub fn new(auth: Arc<BybitAuth>, account_type: SmartString, testnet: bool) -> Self {
        let clock = Clock::new();

        Self {
            auth,
            rest_client: None,
            trade_sink: Arc::new(AsyncRwLock::new(None)),
            trade_stream: Arc::new(AsyncRwLock::new(None)),
            private_sink: Arc::new(AsyncRwLock::new(None)),
            private_stream: Arc::new(AsyncRwLock::new(None)),
            account_type,
            testnet,
            account_info: Arc::new(RwLock::new(None)),
            is_connected: Arc::new(AtomicBool::new(false)),
            is_authenticated: Arc::new(AtomicBool::new(false)),
            clock: clock.clone(),
            health: Arc::new(RwLock::new(ConnectionHealth::new())),
            rate_limiter: Arc::new(RwLock::new(RateLimiter::new(clock))),
            pending_requests: Arc::new(RwLock::new(FxHashMap::default())),
            order_symbol_map: Arc::new(RwLock::new(FxHashMap::default())),
            instrument_cache: Arc::new(RwLock::new(FxHashMap::default())),
            ping_handle: Arc::new(RwLock::new(None)),
            message_handle: Arc::new(RwLock::new(None)),
            report_sender: Arc::new(RwLock::new(None)),
        }
    }

    /// Get connection health status
    #[must_use]
    pub fn get_connection_health(&self) -> ConnectionHealth {
        self.health.read().clone()
    }

    /// Set REST client for V5 account features (optional)
    pub fn set_rest_client(&mut self, rest_client: Arc<BybitRestClient>) {
        self.rest_client = Some(rest_client);
    }

    /// Get and cache account information for V5 compliance
    pub async fn get_account_info(&self) -> Result<BybitAccountInfo> {
        // Check cache first
        if let Some(cached_info) = self.account_info.read().as_ref() {
            return Ok(cached_info.clone());
        }

        // Fetch from REST API if available
        let Some(rest_client) = &self.rest_client else {
            bail!("REST client not available - cannot fetch account info");
        };

        let account_info = rest_client
            .get_account_info()
            .await
            .map_err(|e| anyhow!("Failed to get account info: {e}"))?;

        // Cache the result
        *self.account_info.write() = Some(account_info.clone());

        Ok(account_info)
    }

    /// Check if account supports V5 unified features
    pub async fn supports_unified_features(&self) -> Result<bool> {
        let account_info = self.get_account_info().await?;
        Ok(account_info.supports_unified_features())
    }

    /// Check if account supports UTA 2.0 advanced features
    pub async fn supports_uta2_features(&self) -> Result<bool> {
        let account_info = self.get_account_info().await?;
        Ok(account_info.supports_uta2_features())
    }

    /// Validate order parameters for account type (if REST client available)
    async fn validate_order_for_account(&self, order: &Order, category: &str) -> Result<()> {
        // Skip validation if REST client not available (graceful degradation)
        if self.rest_client.is_none() {
            debug!("REST client not available - skipping V5 account validation");
            return Ok(());
        }

        let account_info = match self.get_account_info().await {
            Ok(info) => info,
            Err(e) => {
                warn!("Failed to get account info for validation: {e} - proceeding with order");
                return Ok(());
            }
        };

        let account_type = account_info.get_account_type();

        if let Some(account_type) = account_type {
            // UTA 2.0: Inverse futures only support one-way mode
            if account_type.supports_uta2_features() && category == "inverse" {
                // For WebSocket, we assume one-way mode (positionIdx = 0)
                // This is automatically handled in order creation
                debug!("UTA 2.0 account: Using one-way mode for inverse futures");
            }

            // Classic account: No unified margin features
            if account_type == BybitAccountType::Classic {
                // WebSocket doesn't directly handle margin params, but we can log info
                if category == "spot" {
                    debug!("Classic account spot trading via WebSocket");
                }
            }

            debug!("V5 Account validation passed for account type: {account_type:?}");
        }

        Ok(())
    }

    /// Map Bybit V5 order status strings to internal OrderStatus enum
    /// Comprehensive mapping covering all V5 status values
    fn map_v5_order_status(status_str: &str) -> OrderStatus {
        match status_str {
            // V5 Standard Statuses
            "New" => OrderStatus::New,
            "PartiallyFilled" => OrderStatus::PartiallyFilled,
            "Filled" => OrderStatus::Filled,
            "Cancelled" => OrderStatus::Cancelled,
            "Rejected" => OrderStatus::Rejected,

            // V5 Additional Statuses
            "PendingCancel" => OrderStatus::PendingCancel,
            "Untriggered" => OrderStatus::Pending, // Conditional orders waiting to trigger
            "Triggered" => OrderStatus::New,       // Conditional order just triggered
            "Deactivated" => OrderStatus::Cancelled, // Conditional order deactivated
            "Active" => OrderStatus::Open,         // Order is actively working

            // V5 Spot-specific Statuses
            "PENDING" => OrderStatus::Pending, // Order pending execution
            "OPEN" => OrderStatus::Open,       // Order is open
            "PARTIALLY_FILLED" => OrderStatus::PartiallyFilled, // Alternative format
            "ORDER_FILLED" => OrderStatus::Filled, // Alternative format
            "CANCELED" => OrderStatus::Cancelled, // Alternative spelling
            "ORDER_FAILED" => OrderStatus::Rejected, // Order failed

            // V5 Futures/Perps-specific Statuses
            "Created" => OrderStatus::New,     // Order just created
            "Expired" => OrderStatus::Expired, // Order expired (GTD orders)

            // V5 Error/Unknown Statuses
            "Unknown" => OrderStatus::Unknown,
            "" => OrderStatus::Unknown, // Empty status

            // Fallback for any unmapped statuses
            _ => {
                warn!("Unknown Bybit V5 order status: '{status_str}' - mapping to Unknown");
                OrderStatus::Unknown
            }
        }
    }

    /// Map internal OrderStatus to Bybit V5 execution type for reporting
    const fn map_order_status_to_exec_type(status: OrderStatus) -> &'static str {
        match status {
            OrderStatus::New => "New",
            OrderStatus::Open => "New", // Open orders reported as New in execution
            OrderStatus::PartiallyFilled => "Trade",
            OrderStatus::Filled => "Trade",
            OrderStatus::Cancelled => "Canceled",
            OrderStatus::Rejected => "Rejected",
            OrderStatus::Expired => "Expired",
            OrderStatus::Pending => "PendingNew",
            OrderStatus::PendingCancel => "PendingCancel",
            OrderStatus::Unknown => "Unknown",
        }
    }

    /// Send WebSocket ping
    pub async fn send_ping(&self) -> Result<()> {
        let req_id = Uuid::new_v4().to_string();
        let ping_msg = json!({
            "req_id": req_id,
            "op": "ping"
        })
        .to_string();

        if let Some(sink) = &mut *self.trade_sink.write().await {
            sink.send(Message::text(ping_msg).to_frame_view()).await?;

            let mut health = self.health.write();
            health.last_ping_sent = self.clock.raw();
            health.pending_pings += 1;
            health.messages_sent += 1;
        } else {
            bail!("WebSocket trade connection not established");
        }

        Ok(())
    }

    /// Authenticate WebSocket connections
    async fn authenticate(&self) -> Result<()> {
        let expires = time::get_timestamp_ms() + 5000;
        let signature = self.auth.generate_ws_signature(expires)?;

        let auth_msg = json!({
            "op": "auth",
            "args": [
                self.auth.api_key(),
                expires,
                signature
            ]
        })
        .to_string();

        // Authenticate trade connection
        if let Some(sink) = &mut *self.trade_sink.write().await {
            sink.send(Message::text(auth_msg.clone()).to_frame_view())
                .await?;
            debug!("Sent auth message to trade connection");
        }

        // Authenticate private connection and subscribe to order updates
        if let Some(sink) = &mut *self.private_sink.write().await {
            sink.send(Message::text(auth_msg).to_frame_view()).await?;
            debug!("Sent auth message to private connection");

            // Subscribe to order stream
            let subscribe_msg = json!({
                "op": "subscribe",
                "args": ["order"]
            })
            .to_string();

            sink.send(Message::text(subscribe_msg).to_frame_view())
                .await?;
            debug!("Subscribed to order stream");
        }

        Ok(())
    }

    /// Connect and start message processing
    async fn connect_internal(&self, report_sender: Sender<ExecutionReport>) -> Result<()> {
        // Store the report sender
        *self.report_sender.write() = Some(report_sender.clone());

        let trade_url = if self.testnet {
            BYBIT_WS_TRADE_TESTNET_URL
        } else {
            BYBIT_WS_TRADE_URL
        };

        let private_url = if self.testnet {
            BYBIT_WS_PRIVATE_TESTNET_URL
        } else {
            BYBIT_WS_PRIVATE_URL
        };

        // Connect to trade WebSocket using yawc directly
        use yawc::WebSocket;
        let trade_parsed_url = trade_url
            .parse()
            .map_err(|e| anyhow::anyhow!("Invalid trade URL: {}", e))?;
        let trade_ws = WebSocket::connect(trade_parsed_url)
            .await
            .map_err(|e| anyhow::anyhow!("Trade connection failed: {}", e))?;
        let (trade_sink, trade_stream) = trade_ws.split();
        *self.trade_sink.write().await = Some(trade_sink);
        *self.trade_stream.write().await = Some(trade_stream);

        // Connect to private WebSocket
        let private_parsed_url = private_url
            .parse()
            .map_err(|e| anyhow::anyhow!("Invalid private URL: {}", e))?;
        let private_ws = WebSocket::connect(private_parsed_url)
            .await
            .map_err(|e| anyhow::anyhow!("Private connection failed: {}", e))?;
        let (private_sink, private_stream) = private_ws.split();
        *self.private_sink.write().await = Some(private_sink);
        *self.private_stream.write().await = Some(private_stream);

        self.is_connected.store(true, Ordering::Relaxed);

        // Authenticate
        self.authenticate().await?;

        // Start ping task
        self.start_ping_task();

        // Start message processing
        self.start_message_processing();

        info!("Bybit WebSocket trading connected and authenticated");
        Ok(())
    }

    /// Start ping task to keep connection alive
    fn start_ping_task(&self) {
        let trade_sink = self.trade_sink.clone();
        let health = self.health.clone();
        let clock = self.clock.clone();
        let is_connected = self.is_connected.clone();

        let handle = tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(PING_INTERVAL_SECONDS));

            loop {
                interval.tick().await;

                if !is_connected.load(Ordering::Relaxed) {
                    break;
                }

                // Check if we need to send ping
                let should_send = {
                    let h = health.read();
                    let now = clock.raw();
                    let time_since_last_ping = (now - h.last_ping_sent) / 1_000_000_000; // Convert to seconds

                    time_since_last_ping >= PING_INTERVAL_SECONDS && h.pending_pings < 3
                };

                if should_send {
                    let req_id = Uuid::new_v4().to_string();
                    let ping_msg = json!({
                        "req_id": req_id,
                        "op": "ping"
                    })
                    .to_string();

                    if let Some(sink) = &mut *trade_sink.write().await {
                        if let Err(e) = sink.send(Message::text(ping_msg).to_frame_view()).await {
                            error!("Failed to send ping: {e}");
                            is_connected.store(false, Ordering::Relaxed);
                            break;
                        }

                        let mut h = health.write();
                        h.last_ping_sent = clock.raw();
                        h.pending_pings += 1;
                        h.messages_sent += 1;
                    }
                }

                // Check for pong timeout
                {
                    let h = health.read();
                    let now = clock.raw();
                    let time_since_last_pong = (now - h.last_pong_received) / 1_000_000_000;

                    if h.pending_pings > 0 && time_since_last_pong > PONG_TIMEOUT_SECONDS {
                        warn!("Pong timeout detected, connection may be dead");
                        is_connected.store(false, Ordering::Relaxed);
                        break;
                    }
                }
            }

            debug!("Ping task terminated");
        });

        *self.ping_handle.write() = Some(handle);
    }

    /// Start processing incoming messages
    fn start_message_processing(&self) {
        let trade_stream = self.trade_stream.clone();
        let private_stream = self.private_stream.clone();
        let health = self.health.clone();
        let clock = self.clock.clone();
        let is_authenticated = self.is_authenticated.clone();
        let is_connected = self.is_connected.clone();
        let pending_requests = self.pending_requests.clone();
        let report_sender = self.report_sender.clone();
        let order_symbol_map = self.order_symbol_map.clone();

        let handle = tokio::spawn(async move {
            loop {
                if !is_connected.load(Ordering::Relaxed) {
                    break;
                }

                // Process trade stream messages
                if let Some(stream) = &mut *trade_stream.write().await {
                    if let Some(frame) = stream.next().await {
                        let msg = Message::from_frame_view(frame);
                        match msg {
                            Message::Text(text) => {
                                health.write().messages_received += 1;

                                if let Err(e) = Self::process_trade_message(
                                    &text,
                                    &health,
                                    &clock,
                                    &is_authenticated,
                                    &pending_requests,
                                    &report_sender,
                                    &order_symbol_map,
                                )
                                .await
                                {
                                    error!("Failed to process trade message: {e}");
                                }
                            }
                            Message::Binary(_) => {
                                warn!("Received unexpected binary message");
                            }
                            _ => {}
                        }
                    } else {
                        warn!("Trade stream closed");
                        is_connected.store(false, Ordering::Relaxed);
                        break;
                    }
                }

                // Process private stream messages
                if let Some(stream) = &mut *private_stream.write().await {
                    match stream.next().await {
                        Some(frame) => {
                            let msg = Message::from_frame_view(frame);
                            match msg {
                                Message::Text(text) => {
                                    health.write().messages_received += 1;

                                    if let Err(e) =
                                        Self::process_private_message(&text, &report_sender).await
                                    {
                                        error!("Failed to process private message: {e}");
                                    }
                                }
                                Message::Binary(_) => {
                                    warn!("Received unexpected binary message on private stream");
                                }
                                _ => {}
                            }
                        }
                        None => {
                            warn!("Private stream closed");
                        }
                    }
                }
            }

            debug!("Message processing task terminated");
        });

        *self.message_handle.write() = Some(handle);
    }

    /// Process messages from trade stream
    async fn process_trade_message(
        text: &str,
        health: &Arc<RwLock<ConnectionHealth>>,
        clock: &Clock,
        is_authenticated: &Arc<AtomicBool>,
        pending_requests: &Arc<RwLock<FxHashMap<SmartString, PendingRequest>>>,
        report_sender: &Arc<RwLock<Option<Sender<ExecutionReport>>>>,
        order_symbol_map: &Arc<RwLock<FxHashMap<SmartString, SmartString>>>,
    ) -> Result<()> {
        let mut text_bytes = text.as_bytes().to_vec();
        let msg = simd_json::from_slice::<JsonValue>(&mut text_bytes)?;

        if let Some(op) = msg.get("op").and_then(|v| v.as_str()) {
            match op {
                "auth" => {
                    let ret_code = msg
                        .get("retCode")
                        .and_then(simd_json::prelude::ValueAsScalar::as_i64)
                        .unwrap_or(-1);

                    if ret_code == 0 {
                        is_authenticated.store(true, Ordering::Relaxed);
                        info!("WebSocket authentication successful");
                    } else {
                        let ret_msg = msg
                            .get("retMsg")
                            .and_then(|v| v.as_str())
                            .unwrap_or("Unknown error");
                        error!("WebSocket authentication failed: {ret_msg}");
                    }
                }
                "ping" => {
                    if let Some(ret_msg) = msg.get("ret_msg").and_then(|v| v.as_str())
                        && ret_msg == "pong"
                    {
                        let mut h = health.write();
                        h.last_pong_received = clock.raw();
                        h.pending_pings = h.pending_pings.saturating_sub(1);
                        h.is_healthy = true;
                    }
                }
                "order.create" | "order.amend" | "order.cancel" => {
                    Self::process_order_response(
                        msg.clone(),
                        op,
                        pending_requests,
                        report_sender,
                        order_symbol_map,
                    )
                    .await?;
                }
                "order.create-batch" | "order.amend-batch" | "order.cancel-batch" => {
                    Self::process_batch_order_response(
                        msg.clone(),
                        op,
                        pending_requests,
                        report_sender,
                        order_symbol_map,
                    )
                    .await?;
                }
                "order.cancel-all" => {
                    Self::process_cancel_all_response(msg, pending_requests, report_sender).await?;
                }
                _ => {
                    debug!("Unhandled op type: {op}");
                }
            }
        }

        Ok(())
    }

    /// Process messages from private stream (order updates)
    async fn process_private_message(
        text: &str,
        report_sender: &Arc<RwLock<Option<Sender<ExecutionReport>>>>,
    ) -> Result<()> {
        let mut text_bytes = text.as_bytes().to_vec();
        let msg = simd_json::from_slice::<JsonValue>(&mut text_bytes)?;

        if let Some(topic) = msg.get("topic").and_then(|v| v.as_str())
            && (topic == "order" || topic.starts_with("order."))
        {
            Self::process_order_update(msg, report_sender).await?;
        }

        Ok(())
    }

    /// Process order response from trade stream
    async fn process_order_response(
        msg: JsonValue,
        op: &str,
        pending_requests: &Arc<RwLock<FxHashMap<SmartString, PendingRequest>>>,
        report_sender: &Arc<RwLock<Option<Sender<ExecutionReport>>>>,
        order_symbol_map: &Arc<RwLock<FxHashMap<SmartString, SmartString>>>,
    ) -> Result<()> {
        let req_id: Option<SmartString> = msg
            .get("reqId")
            .and_then(|v| v.as_str())
            .map(std::convert::Into::into);

        let ret_code = msg
            .get("retCode")
            .and_then(simd_json::prelude::ValueAsScalar::as_i64)
            .unwrap_or(-1);

        if let Some(req_id) = req_id
            && let Some(pending) = pending_requests.write().remove(&req_id)
        {
            if ret_code == 0 {
                // Success - order accepted
                if let Some(data) = msg.get("data") {
                    let order_id = data.get("orderId").and_then(|v| v.as_str()).unwrap_or("");
                    let client_order_id = data
                        .get("orderLinkId")
                        .and_then(|v| v.as_str())
                        .unwrap_or("");

                    // Update our order symbol mapping with the exchange order ID
                    if !order_id.is_empty()
                        && !client_order_id.is_empty()
                        && let Some(symbol) = order_symbol_map.read().get(client_order_id)
                    {
                        order_symbol_map
                            .write()
                            .insert(order_id.into(), symbol.clone());
                    }

                    let report = ExecutionReport {
                        id: Uuid::new_v4().to_string().into(),
                        order_id: order_id.into(),
                        exchange_timestamp: time::get_epoch_timestamp_ns(),
                        system_timestamp: time::get_epoch_timestamp_ns(),
                        instrument_id: InstrumentId::new(
                            "UNKNOWN",
                            rusty_model::venues::Venue::Bybit,
                        ),
                        status: OrderStatus::New,
                        filled_quantity: Decimal::ZERO,
                        remaining_quantity: Decimal::ZERO,
                        execution_price: None,
                        reject_reason: None,
                        exchange_execution_id: None,
                        is_final: false,
                    };

                    if let Err(e) = pending.report_sender.send(report) {
                        error!("Failed to send execution report: {e}");
                    }
                }
            } else {
                // Error - order rejected
                let ret_msg = msg
                    .get("retMsg")
                    .and_then(|v| v.as_str())
                    .unwrap_or("Unknown error");

                let report = ExecutionReport {
                    id: Uuid::new_v4().to_string().into(),
                    order_id: SmartString::default(),
                    exchange_timestamp: time::get_epoch_timestamp_ns(),
                    system_timestamp: time::get_epoch_timestamp_ns(),
                    instrument_id: InstrumentId::new("UNKNOWN", rusty_model::venues::Venue::Bybit),
                    status: OrderStatus::Rejected,
                    filled_quantity: Decimal::ZERO,
                    remaining_quantity: Decimal::ZERO,
                    execution_price: None,
                    reject_reason: Some(ret_msg.into()),
                    exchange_execution_id: None,
                    is_final: true,
                };

                if let Err(e) = pending.report_sender.send(report) {
                    error!("Failed to send rejection report: {e}");
                }
            }
        }

        Ok(())
    }

    /// Process batch order response
    async fn process_batch_order_response(
        msg: JsonValue,
        op: &str,
        pending_requests: &Arc<RwLock<FxHashMap<SmartString, PendingRequest>>>,
        report_sender: &Arc<RwLock<Option<Sender<ExecutionReport>>>>,
        order_symbol_map: &Arc<RwLock<FxHashMap<SmartString, SmartString>>>,
    ) -> Result<()> {
        let req_id: Option<SmartString> = msg
            .get("reqId")
            .and_then(|v| v.as_str())
            .map(std::convert::Into::into);

        let ret_code = msg
            .get("retCode")
            .and_then(simd_json::prelude::ValueAsScalar::as_i64)
            .unwrap_or(-1);

        if let Some(req_id) = req_id
            && let Some(pending) = pending_requests.write().remove(&req_id)
        {
            if ret_code == 0 {
                // Process batch results
                if let Some(data) = msg.get("data")
                    && let Some(list) = data.get("list").and_then(|v| v.as_array())
                {
                    for item in list {
                        let order_id = item.get("orderId").and_then(|v| v.as_str()).unwrap_or("");

                        let order_link_id = item
                            .get("orderLinkId")
                            .and_then(|v| v.as_str())
                            .unwrap_or("");

                        let report = ExecutionReport {
                            id: Uuid::new_v4().to_string().into(),
                            order_id: order_id.into(),
                            exchange_timestamp: time::get_epoch_timestamp_ns(),
                            system_timestamp: time::get_epoch_timestamp_ns(),
                            instrument_id: InstrumentId::new(
                                "UNKNOWN",
                                rusty_model::venues::Venue::Bybit,
                            ),
                            status: OrderStatus::New,
                            filled_quantity: Decimal::ZERO,
                            remaining_quantity: Decimal::ZERO,
                            execution_price: None,
                            reject_reason: None,
                            exchange_execution_id: Some(order_link_id.into()),
                            is_final: false,
                        };

                        if let Err(e) = pending.report_sender.send(report) {
                            error!("Failed to send batch execution report: {e}");
                        }
                    }
                }
            } else {
                // Batch request failed
                let ret_msg = msg
                    .get("retMsg")
                    .and_then(|v| v.as_str())
                    .unwrap_or("Batch order failed");

                error!("Batch order failed: {ret_msg}");
            }
        }

        Ok(())
    }

    /// Process cancel all response
    async fn process_cancel_all_response(
        msg: JsonValue,
        pending_requests: &Arc<RwLock<FxHashMap<SmartString, PendingRequest>>>,
        report_sender: &Arc<RwLock<Option<Sender<ExecutionReport>>>>,
    ) -> Result<()> {
        let req_id: Option<SmartString> = msg
            .get("reqId")
            .and_then(|v| v.as_str())
            .map(std::convert::Into::into);

        let ret_code = msg
            .get("retCode")
            .and_then(simd_json::prelude::ValueAsScalar::as_i64)
            .unwrap_or(-1);

        if let Some(req_id) = req_id
            && let Some(pending) = pending_requests.write().remove(&req_id)
        {
            if ret_code == 0 {
                // Cancel all successful
                let report = ExecutionReport {
                    id: Uuid::new_v4().to_string().into(),
                    order_id: req_id.clone(),
                    exchange_timestamp: time::get_epoch_timestamp_ns(),
                    system_timestamp: time::get_epoch_timestamp_ns(),
                    instrument_id: InstrumentId::new("UNKNOWN", rusty_model::venues::Venue::Bybit),
                    status: OrderStatus::New,
                    filled_quantity: Decimal::ZERO,
                    remaining_quantity: Decimal::ZERO,
                    execution_price: None,
                    reject_reason: None,
                    exchange_execution_id: None,
                    is_final: false,
                };

                if let Err(e) = pending.report_sender.send(report) {
                    error!("Failed to send cancel all report: {e}");
                }
            } else {
                let ret_msg = msg
                    .get("retMsg")
                    .and_then(|v| v.as_str())
                    .unwrap_or("Cancel all failed");

                error!("Cancel all failed: {ret_msg}");
            }
        }

        Ok(())
    }

    /// Process order update from private stream
    async fn process_order_update(
        msg: JsonValue,
        report_sender: &Arc<RwLock<Option<Sender<ExecutionReport>>>>,
    ) -> Result<()> {
        if let Some(data) = msg.get("data").and_then(|v| v.as_array()) {
            for order_data in data {
                let order_id = order_data
                    .get("orderId")
                    .and_then(|v| v.as_str())
                    .unwrap_or("");

                let order_link_id = order_data
                    .get("orderLinkId")
                    .and_then(|v| v.as_str())
                    .unwrap_or("");

                let symbol = order_data
                    .get("symbol")
                    .and_then(|v| v.as_str())
                    .unwrap_or("");

                let side_str = order_data
                    .get("side")
                    .and_then(|v| v.as_str())
                    .unwrap_or("Buy");

                let side = match side_str {
                    "Sell" => OrderSide::Sell,
                    _ => OrderSide::Buy,
                };

                let order_status_str = order_data
                    .get("orderStatus")
                    .and_then(|v| v.as_str())
                    .unwrap_or("New");

                let order_status = Self::map_v5_order_status(order_status_str);

                let qty = order_data
                    .get("qty")
                    .and_then(|v| v.as_str())
                    .and_then(|s| Decimal::from_str_exact(s).ok())
                    .unwrap_or(Decimal::ZERO);

                let price = order_data
                    .get("price")
                    .and_then(|v| v.as_str())
                    .and_then(|s| Decimal::from_str_exact(s).ok())
                    .unwrap_or(Decimal::ZERO);

                let cum_exec_qty = order_data
                    .get("cumExecQty")
                    .and_then(|v| v.as_str())
                    .and_then(|s| Decimal::from_str_exact(s).ok())
                    .unwrap_or(Decimal::ZERO);

                let avg_price = order_data
                    .get("avgPrice")
                    .and_then(|v| v.as_str())
                    .and_then(|s| Decimal::from_str_exact(s).ok())
                    .unwrap_or(Decimal::ZERO);

                let leaves_qty = order_data
                    .get("leavesQty")
                    .and_then(|v| v.as_str())
                    .and_then(|s| Decimal::from_str_exact(s).ok())
                    .unwrap_or(qty - cum_exec_qty);

                let exec_type = Self::map_order_status_to_exec_type(order_status);

                let report = ExecutionReport {
                    id: Uuid::new_v4().to_string().into(),
                    order_id: order_id.into(),
                    exchange_timestamp: time::get_epoch_timestamp_ns(),
                    system_timestamp: time::get_epoch_timestamp_ns(),
                    instrument_id: InstrumentId::new("UNKNOWN", rusty_model::venues::Venue::Bybit),
                    status: order_status,
                    filled_quantity: Decimal::ZERO,
                    remaining_quantity: Decimal::ZERO,
                    execution_price: None,
                    reject_reason: None,
                    exchange_execution_id: Some(order_link_id.into()),
                    is_final: false,
                };

                if let Some(sender) = &*report_sender.read()
                    && let Err(e) = sender.send(report)
                {
                    error!("Failed to send order update report: {e}");
                }
            }
        }

        Ok(())
    }

    /// Place an order via WebSocket
    async fn place_order_internal(&self, order: &Order) -> Result<SmartString> {
        // Check rate limit
        if !self.rate_limiter.write().can_send_request() {
            bail!("Rate limit exceeded");
        }

        let req_id = Uuid::new_v4().to_string();
        let timestamp = time::get_timestamp_ms();

        // Determine category based on symbol
        let category = self.get_category_from_symbol(&order.symbol);

        // Build V5-compliant order args
        let mut order_args = json!({
            "symbol": order.symbol.as_str(),
            "side": match order.side {
                OrderSide::Buy => "Buy",
                OrderSide::Sell => "Sell",
            },
            "orderType": match order.order_type {
                OrderType::Market => "Market",
                OrderType::Limit => "Limit",
                OrderType::PostOnly => "Limit", // PostOnly is handled via timeInForce
                _ => "Limit",
            },
            "qty": order.quantity.to_string(),
            "category": category,
            "timeInForce": match order.time_in_force {
                TimeInForce::GTC => "GTC",
                TimeInForce::IOC => "IOC",
                TimeInForce::FOK => "FOK",
                TimeInForce::GTX => "PostOnly", // V5: GTX maps to PostOnly
                TimeInForce::GTD => {
                    warn!("GTD (Good Till Date) not supported by Bybit, using GTC instead for order {}", order.id);
                    "GTC"  // Use GTC as closest alternative - orders remain active until cancelled
                },
            }
        });

        // V5: Add positionIdx for futures/perps (default to one-way mode)
        if category == "linear" || category == "inverse" {
            order_args["positionIdx"] = json!(0); // One-way mode
        }

        // Add price for limit orders
        if order.order_type == OrderType::Limit
            && let Some(price) = order.price
        {
            order_args["price"] = json!(price.to_string());
        }

        // Use our internal order ID as orderLinkId for tracking
        // This allows us to correlate responses with our orders
        order_args["orderLinkId"] = json!(order.id.to_string());

        // Add stop loss if provided
        if let Some(stop_price) = order.stop_price
            && stop_price > Decimal::ZERO
        {
            order_args["stopLoss"] = json!(stop_price.to_string());
            order_args["slTriggerBy"] = json!("LastPrice");
        }

        // Note: take_profit_price field doesn't exist in Order struct
        // This feature would need to be added to the Order model if needed

        // Create the request
        let request = json!({
            "reqId": &req_id,
            "header": {
                "X-BAPI-TIMESTAMP": timestamp.to_string(),
                "X-BAPI-RECV-WINDOW": "5000"
            },
            "op": "order.create",
            "args": [order_args]
        })
        .to_string();

        // Send the request
        if let Some(sink) = &mut *self.trade_sink.write().await {
            sink.send(Message::text(request).to_frame_view()).await?;
            self.rate_limiter.write().record_request();
        } else {
            bail!("WebSocket trade connection not established");
        }

        Ok(req_id.into())
    }

    /// Amend an order via WebSocket
    async fn amend_order_internal(
        &self,
        order_id: &str,
        symbol: &str,
        new_price: Option<Decimal>,
        new_quantity: Option<Decimal>,
    ) -> Result<SmartString> {
        // Check rate limit
        if !self.rate_limiter.write().can_send_request() {
            bail!("Rate limit exceeded");
        }

        let req_id = Uuid::new_v4().to_string();
        let timestamp = time::get_timestamp_ms();

        // Determine category
        let category = self.get_category_from_symbol(symbol);

        // Build amend args
        let mut amend_args = json!({
            "orderId": order_id,
            "symbol": symbol,
            "category": category,
        });

        // Add new price if provided
        if let Some(price) = new_price {
            amend_args["price"] = json!(price.to_string());
        }

        // Add new quantity if provided
        if let Some(qty) = new_quantity {
            amend_args["qty"] = json!(qty.to_string());
        }

        // Create the request
        let request = json!({
            "reqId": &req_id,
            "header": {
                "X-BAPI-TIMESTAMP": timestamp.to_string(),
                "X-BAPI-RECV-WINDOW": "5000"
            },
            "op": "order.amend",
            "args": [amend_args]
        })
        .to_string();

        // Send the request
        if let Some(sink) = &mut *self.trade_sink.write().await {
            sink.send(Message::text(request).to_frame_view()).await?;
            self.rate_limiter.write().record_request();
        } else {
            bail!("WebSocket trade connection not established");
        }

        Ok(req_id.into())
    }

    /// Cancel an order via WebSocket
    async fn cancel_order_internal(&self, order_id: &str, symbol: &str) -> Result<SmartString> {
        // Check rate limit
        if !self.rate_limiter.write().can_send_request() {
            bail!("Rate limit exceeded");
        }

        let req_id = Uuid::new_v4().to_string();
        let timestamp = time::get_timestamp_ms();

        // Determine category
        let category = self.get_category_from_symbol(symbol);

        // Build cancel args
        let cancel_args = json!({
            "orderId": order_id,
            "symbol": symbol,
            "category": category,
        });

        // Create the request
        let request = json!({
            "reqId": &req_id,
            "header": {
                "X-BAPI-TIMESTAMP": timestamp.to_string(),
                "X-BAPI-RECV-WINDOW": "5000"
            },
            "op": "order.cancel",
            "args": [cancel_args]
        })
        .to_string();

        // Send the request
        if let Some(sink) = &mut *self.trade_sink.write().await {
            sink.send(Message::text(request).to_frame_view()).await?;
            self.rate_limiter.write().record_request();
        } else {
            bail!("WebSocket trade connection not established");
        }

        Ok(req_id.into())
    }

    /// Cancel all orders via WebSocket
    async fn cancel_all_orders_internal(
        &self,
        category: &str,
        symbol: Option<&str>,
    ) -> Result<SmartString> {
        // Check rate limit
        if !self.rate_limiter.write().can_send_request() {
            bail!("Rate limit exceeded");
        }

        let req_id = Uuid::new_v4().to_string();
        let timestamp = time::get_timestamp_ms();

        // Build cancel all args
        let mut cancel_args = json!({
            "category": category,
        });

        // Add symbol if provided
        if let Some(sym) = symbol {
            cancel_args["symbol"] = json!(sym);
        }

        // Create the request
        let request = json!({
            "reqId": &req_id,
            "header": {
                "X-BAPI-TIMESTAMP": timestamp.to_string(),
                "X-BAPI-RECV-WINDOW": "5000"
            },
            "op": "order.cancel-all",
            "args": [cancel_args]
        })
        .to_string();

        // Send the request
        if let Some(sink) = &mut *self.trade_sink.write().await {
            sink.send(Message::text(request).to_frame_view()).await?;
            self.rate_limiter.write().record_request();
        } else {
            bail!("WebSocket trade connection not established");
        }

        Ok(req_id.into())
    }

    /// Determine category from symbol using cached instrument info
    fn get_category_from_symbol(&self, symbol: &str) -> &'static str {
        // First, try to get from cache
        if let Some(info) = self.instrument_cache.read().get(symbol)
            && !info.is_expired()
        {
            return info.category.as_str();
        }

        // If not in cache or expired, use fallback heuristic and potentially trigger cache refresh
        let fallback_category = self.get_category_from_symbol_heuristic(symbol);

        // Trigger async cache refresh for this symbol (fire and forget)
        let symbol_owned = symbol.to_owned();
        let cache = self.instrument_cache.clone();
        let auth = self.auth.clone();
        let testnet = self.testnet;
        tokio::spawn(async move {
            if let Err(e) =
                Self::refresh_instrument_info_for_symbol(&cache, &auth, &symbol_owned, testnet)
                    .await
            {
                debug!("Failed to refresh instrument info for {symbol_owned}: {e}");
            }
        });

        fallback_category
    }

    /// Fallback heuristic-based category determination
    fn get_category_from_symbol_heuristic(&self, symbol: &str) -> &'static str {
        if symbol.ends_with("USDT") || symbol.ends_with("USDC") {
            if symbol.contains('-') {
                CATEGORY_OPTION
            } else {
                CATEGORY_LINEAR
            }
        } else if symbol.ends_with("USD") {
            CATEGORY_INVERSE
        } else {
            CATEGORY_SPOT
        }
    }

    /// Refresh instrument info for a specific symbol
    async fn refresh_instrument_info_for_symbol(
        cache: &Arc<RwLock<FxHashMap<SmartString, InstrumentInfo>>>,
        auth: &BybitAuth,
        symbol: &str,
        testnet: bool,
    ) -> Result<()> {
        // Try each category until we find the symbol
        let categories = ["spot", "linear", "inverse", "option"];

        for category in &categories {
            if let Ok(info) = Self::fetch_instrument_info(auth, symbol, category, testnet).await {
                cache.write().insert(symbol.into(), info);
                debug!("Cached instrument info for symbol: {symbol} in category: {category}");
                return Ok(());
            }
        }

        // If not found in any category, cache with fallback heuristic
        let fallback_category = if symbol.ends_with("USDT") || symbol.ends_with("USDC") {
            if symbol.contains('-') {
                BybitCategory::Option
            } else {
                BybitCategory::Linear
            }
        } else if symbol.ends_with("USD") {
            BybitCategory::Inverse
        } else {
            BybitCategory::Spot
        };

        let info = InstrumentInfo::new(
            symbol.into(),
            fallback_category,
            "Unknown".into(),
            "Unknown".into(),
            "Unknown".into(),
            None,
        );

        cache.write().insert(symbol.into(), info);
        warn!(
            "Symbol {} not found via API, cached with heuristic category: {}",
            symbol,
            fallback_category.as_str()
        );

        Ok(())
    }

    /// Fetch instrument info from Bybit REST API
    async fn fetch_instrument_info(
        auth: &BybitAuth,
        symbol: &str,
        category: &str,
        testnet: bool,
    ) -> Result<InstrumentInfo> {
        let base_url = if testnet {
            "https://api-testnet.bybit.com"
        } else {
            "https://api.bybit.com"
        };

        let url =
            format!("{base_url}/v5/market/instruments-info?category={category}&symbol={symbol}");

        // Create HTTP client with timeout
        let client = reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(10))
            .build()?;

        let response = client
            .get(&url)
            .header("Content-Type", "application/json")
            .send()
            .await?;

        if !response.status().is_success() {
            bail!("HTTP request failed: {}", response.status());
        }

        let text = response.text().await?;
        let mut text_bytes = text.as_bytes().to_vec();
        let data = simd_json::from_slice::<JsonValue>(&mut text_bytes)?;

        // Parse response
        let ret_code = data
            .get("retCode")
            .and_then(simd_json::prelude::ValueAsScalar::as_i64)
            .unwrap_or(-1);
        if ret_code != 0 {
            let ret_msg = data
                .get("retMsg")
                .and_then(|v| v.as_str())
                .unwrap_or("Unknown error");
            bail!("API error: {}", ret_msg);
        }

        let result = data
            .get("result")
            .ok_or_else(|| anyhow!("Missing result field"))?;
        let list = result
            .get("list")
            .and_then(|v| v.as_array())
            .ok_or_else(|| anyhow!("Missing or invalid list field"))?;

        if list.is_empty() {
            bail!("Symbol not found in category {}", category);
        }

        let instrument = &list[0];

        let symbol_str: SmartString = instrument
            .get("symbol")
            .and_then(|v| v.as_str())
            .unwrap_or(symbol)
            .into();

        let category_enum = BybitCategory::parse_category(category)
            .ok_or_else(|| anyhow!("Invalid category: {}", category))?;

        let status: SmartString = instrument
            .get("status")
            .and_then(|v| v.as_str())
            .unwrap_or("Unknown")
            .into();

        let base_coin: SmartString = instrument
            .get("baseCoin")
            .and_then(|v| v.as_str())
            .unwrap_or("Unknown")
            .into();

        let quote_coin: SmartString = instrument
            .get("quoteCoin")
            .and_then(|v| v.as_str())
            .unwrap_or("Unknown")
            .into();

        let contract_type: Option<SmartString> = instrument
            .get("contractType")
            .and_then(|v| v.as_str())
            .map(std::convert::Into::into);

        Ok(InstrumentInfo::new(
            symbol_str,
            category_enum,
            status,
            base_coin,
            quote_coin,
            contract_type,
        ))
    }

    /// Refresh instrument cache for all symbols in use
    pub async fn refresh_instrument_cache(&self) -> Result<()> {
        let symbols: Vec<SmartString> = self.order_symbol_map.read().values().cloned().collect();

        if symbols.is_empty() {
            debug!("No symbols to refresh in instrument cache");
            return Ok(());
        }

        info!("Refreshing instrument cache for {} symbols", symbols.len());

        // Refresh in batches to avoid overwhelming the API
        for chunk in symbols.chunks(10) {
            let mut tasks = Vec::new();

            for symbol in chunk {
                let cache = self.instrument_cache.clone();
                let auth = self.auth.clone();
                let symbol_owned = symbol.clone();
                let testnet = self.testnet;

                let task = tokio::spawn(async move {
                    Self::refresh_instrument_info_for_symbol(&cache, &auth, &symbol_owned, testnet)
                        .await
                });

                tasks.push(task);
            }

            // Wait for this batch to complete
            for task in tasks {
                if let Err(e) = task.await {
                    error!("Failed to refresh instrument info: {e}");
                }
            }

            // Rate limit: wait 100ms between batches
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }

        info!("Instrument cache refresh completed");
        Ok(())
    }

    /// Get cached instrument info for a symbol
    #[must_use]
    pub fn get_instrument_info(&self, symbol: &str) -> Option<InstrumentInfo> {
        self.instrument_cache.read().get(symbol).cloned()
    }

    /// Clear expired entries from instrument cache
    pub fn cleanup_instrument_cache(&self) {
        let mut cache = self.instrument_cache.write();
        let initial_count = cache.len();

        cache.retain(|_, info| !info.is_expired());

        let removed_count = initial_count - cache.len();
        if removed_count > 0 {
            debug!("Removed {removed_count} expired entries from instrument cache");
        }
    }

    /// Get maximum batch size for category (string version for backward compatibility)
    fn get_max_batch_size(&self, category: &str) -> usize {
        match category {
            CATEGORY_SPOT => MAX_BATCH_SPOT,
            CATEGORY_LINEAR => MAX_BATCH_LINEAR,
            CATEGORY_INVERSE => MAX_BATCH_INVERSE,
            CATEGORY_OPTION => MAX_BATCH_OPTION,
            _ => MAX_BATCH_SPOT,
        }
    }

    /// Get maximum batch size for category (enum version)
    const fn get_max_batch_size_for_category(&self, category: BybitCategory) -> usize {
        category.max_batch_size()
    }

    /// Get category enum from symbol
    fn get_category_enum_from_symbol(&self, symbol: &str) -> BybitCategory {
        // First, try to get from cache
        if let Some(info) = self.instrument_cache.read().get(symbol)
            && !info.is_expired()
        {
            return info.category;
        }

        // Fallback to heuristic
        if symbol.ends_with("USDT") || symbol.ends_with("USDC") {
            if symbol.contains('-') {
                BybitCategory::Option
            } else {
                BybitCategory::Linear
            }
        } else if symbol.ends_with("USD") {
            BybitCategory::Inverse
        } else {
            BybitCategory::Spot
        }
    }
}
#[async_trait]
impl Exchange for BybitWebsocketTrading {
    fn venue(&self) -> Venue {
        Venue::Bybit
    }

    async fn place_order(
        &self,
        order: Order,
        report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // V5 Compliance: Validate order for account type (if possible)
        let category = self.get_category_from_symbol(&order.symbol);
        if let Err(e) = self.validate_order_for_account(&order, category).await {
            warn!("V5 validation failed but continuing with order: {e}");
        }

        // Store the pending request
        let req_id = self.place_order_internal(&order).await?;

        // Track order symbol mapping using our internal order ID
        // We'll update this when we get the exchange order ID
        self.order_symbol_map
            .write()
            .insert(order.id.to_string().into(), order.symbol.clone());

        let pending = PendingRequest {
            op_type: SmartString::from("order.create"),
            timestamp: self.clock.raw(),
            report_sender,
        };

        self.pending_requests.write().insert(req_id, pending);

        Ok(())
    }

    async fn cancel_order(
        &self,
        order_id: SmartString,
        report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Get the symbol from our order tracking
        let symbol = self
            .order_symbol_map
            .read()
            .get(&order_id)
            .cloned()
            .ok_or_else(|| anyhow!("Symbol not found for order_id: {}", order_id))?;

        let req_id = self.cancel_order_internal(&order_id, &symbol).await?;

        let pending = PendingRequest {
            op_type: SmartString::from("order.cancel"),
            timestamp: self.clock.raw(),
            report_sender,
        };

        self.pending_requests.write().insert(req_id, pending);

        Ok(())
    }

    async fn modify_order(
        &self,
        order_id: SmartString,
        new_price: Option<Decimal>,
        new_quantity: Option<Decimal>,
        report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Get the symbol from our order tracking
        let symbol = self
            .order_symbol_map
            .read()
            .get(&order_id)
            .cloned()
            .ok_or_else(|| anyhow!("Symbol not found for order_id: {}", order_id))?;

        let req_id = self
            .amend_order_internal(&order_id, &symbol, new_price, new_quantity)
            .await?;

        let pending = PendingRequest {
            op_type: SmartString::from("order.amend"),
            timestamp: self.clock.raw(),
            report_sender,
        };

        self.pending_requests.write().insert(req_id, pending);

        Ok(())
    }

    async fn cancel_all_orders(
        &self,
        instrument_id: Option<InstrumentId>,
        report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        let (category, symbol_str) = if let Some(id) = &instrument_id {
            let category = self.get_category_from_symbol(&id.symbol);
            (category, Some(id.symbol.as_str()))
        } else {
            // Cancel all orders across all categories
            (CATEGORY_LINEAR, None) // TODO: Handle multiple categories
        };

        let req_id = self
            .cancel_all_orders_internal(category, symbol_str)
            .await?;

        let pending = PendingRequest {
            op_type: SmartString::from("order.cancel-all"),
            timestamp: self.clock.raw(),
            report_sender,
        };

        self.pending_requests.write().insert(req_id, pending);

        Ok(())
    }

    async fn get_order_status(&self, order_id: &str) -> Result<OrderStatus> {
        // WebSocket doesn't provide query functionality
        // You would need to track order status from the order stream
        bail!("Order status query not supported via WebSocket - track from order stream")
    }

    async fn connect(&self, report_sender: Sender<ExecutionReport>) -> Result<()> {
        // Use the proper report_sender channel instead of creating a dummy one
        self.connect_internal(report_sender).await
    }

    async fn disconnect(&self) -> Result<()> {
        self.is_connected.store(false, Ordering::Relaxed);
        self.is_authenticated.store(false, Ordering::Relaxed);

        // Close WebSocket connections
        if let Some(sink) = &mut *self.trade_sink.write().await {
            let _ = sink.close().await;
        }
        *self.trade_sink.write().await = None;
        *self.trade_stream.write().await = None;

        if let Some(sink) = &mut *self.private_sink.write().await {
            let _ = sink.close().await;
        }
        *self.private_sink.write().await = None;
        *self.private_stream.write().await = None;

        // Cancel background tasks
        if let Some(handle) = self.ping_handle.write().take() {
            handle.abort();
        }

        if let Some(handle) = self.message_handle.write().take() {
            handle.abort();
        }

        // Clear pending requests
        self.pending_requests.write().clear();

        info!("Bybit WebSocket trading disconnected");
        Ok(())
    }

    async fn is_connected(&self) -> bool {
        self.is_connected.load(Ordering::Relaxed) && self.is_authenticated.load(Ordering::Relaxed)
    }

    async fn get_instruments(&self) -> Result<SmallVec<[InstrumentId; 32]>> {
        // WebSocket doesn't provide instrument query functionality
        // You would need to use REST API or maintain a cached list
        bail!("Instrument query not supported via WebSocket")
    }

    async fn send_fix_message(&self, message: Vec<u8>) -> Result<()> {
        // Bybit WebSocket doesn't support FIX protocol
        bail!("FIX protocol not supported on Bybit WebSocket")
    }

    async fn receive_fix_message(&self) -> Result<Vec<u8>> {
        // Bybit WebSocket doesn't support FIX protocol
        bail!("FIX protocol not supported on Bybit WebSocket")
    }
}