rusty_ems/exchanges/

bybit_websocket_trading.rs

//! Bybit V5 WebSocket Trading Implementation
//!
//! This module provides a high-performance WebSocket trading client for Bybit V5 exchange
//! with robust connection management, automatic reconnection, and comprehensive health monitoring.
//!
//! # Features
//!
//! - **V5 unified authentication**: Support for all product types (spot, linear, inverse, option)
//! - **Batch operations**: Create, cancel, and amend up to 20 orders in a single request
//! - **Position updates**: Real-time position change notifications
//! - **Connection health monitoring**: Real-time metrics and health status
//! - **Automatic reconnection**: Configurable backoff strategy with exponential delays
//! - **Zero-copy message processing**: Optimized for low-latency trading
//!
//! # Connection Management
//!
//! The client maintains connection health through:
//! - WebSocket ping frames sent every 20 seconds (as recommended by Bybit)
//! - JSON ping requests for API-level health checks
//! - Automatic reconnection on connection loss
//! - Configurable pong timeout (default: 10 seconds)
//!
//! # Batch Operations
//!
//! Supports Bybit V5 batch operations:
//! - `order.create-batch`: Create up to 20 orders (10 for spot) in one request
//! - `order.cancel-batch`: Cancel multiple orders by IDs
//! - `order.amend-batch`: Modify multiple orders
//!
//! # Usage Example
//!
//! ```rust,no_run
//! use rusty_common::auth::exchanges::bybit::BybitAuth;
//! use rusty_ems::exchanges::bybit_websocket_trading::BybitWebSocketTrader;
//! use std::sync::Arc;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
//!     let auth = Arc::new(BybitAuth::new(
//!         "api_key".into(),
//!         "secret_key".into()
//!     ));
//!
//!     let trader = BybitWebSocketTrader::new(auth, false); // false = mainnet
//!     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);
//!
//!     Ok(())
//! }
//! ```

use crate::instrument_registry::InstrumentRegistry;
use rusty_common::auth::exchanges::bybit::{BybitAuth, BybitWsTradingMessage};
use rusty_common::collections::FxHashMap;
use rusty_common::utils::id_generation;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering};

use crate::execution_engine::ExecutionReport;
use anyhow::{Result, 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 rust_decimal::Decimal;
use rusty_common::types::Exchange;
use rusty_common::websocket::{
    Message, WebSocketConfig,
    client::ConnectionState as WsConnectionState,
    connector::{WebSocketConnector, WebSocketSink, WebSocketStream},
    stats::new_shared_stats,
};
use rusty_model::{
    enums::{OrderSide, OrderStatus, OrderType, TimeInForce},
    instruments::InstrumentId,
    position::{MarginType, PositionSide, PositionUpdate},
    trading_order::Order,
    types::PositionId,
    venues::Venue,
};
use simd_json::prelude::*;
use simd_json::value::owned::Value as JsonValue;
use smallvec::SmallVec;
use smartstring::alias::String as SmartString;
use std::str::FromStr;
use std::time::Duration;
use tokio::sync::RwLock as AsyncRwLock;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tokio::time::interval;
use uuid::Uuid;

/// Default WebSocket URLs for Bybit V5
const WS_MAINNET_URL: &str = "wss://stream.bybit.com/v5/trade";
const WS_TESTNET_URL: &str = "wss://stream-testnet.bybit.com/v5/trade";

/// Connection configuration constants
const PING_INTERVAL_SECONDS: u64 = 20; // Bybit recommends 20 seconds
const PONG_TIMEOUT_SECONDS: u64 = 10;
const MAX_RECONNECTION_ATTEMPTS: u8 = 10;
const INITIAL_BACKOFF_MS: u64 = 1000;
const MAX_BACKOFF_MS: u64 = 60000;

/// Bybit V5 product categories
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BybitCategory {
    /// Spot trading category for spot markets
    Spot,
    /// Linear futures category for USDT-margined contracts
    Linear,
    /// Inverse futures category for coin-margined contracts
    Inverse,
    /// Options category for options contracts
    Options,
}

/// Bybit account type
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BybitAccountType {
    /// Unified margin account (cross-margin across all products)
    Unified,
    /// Classic account (isolated margin per product)
    Contract,
}

impl BybitCategory {
    const fn as_str(&self) -> &'static str {
        match self {
            Self::Spot => "spot",
            Self::Linear => "linear",
            Self::Inverse => "inverse",
            Self::Options => "option",
        }
    }

    /// Parse from API string
    #[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::Options),
            _ => None,
        }
    }
}

/// Instrument info for symbol categorization
#[derive(Debug, Clone)]
struct InstrumentInfo {
    /// Trading symbol (e.g., "BTCUSDT")
    symbol: SmartString,
    /// Bybit product category (spot, linear, inverse, options)
    category: BybitCategory,
    /// Base coin of the trading pair (e.g., "BTC")
    base_coin: SmartString,
    /// Quote coin of the trading pair (e.g., "USDT")
    quote_coin: SmartString,
}

/// Connection health metrics
#[derive(Debug, Clone, Default)]
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 reconnection attempts made
    pub reconnection_attempts: u8,
    /// Total number of messages sent
    pub messages_sent: u64,
    /// Total number of messages received
    pub messages_received: u64,
}

/// Bybit WebSocket order request for individual orders
#[derive(Debug, Clone)]
pub struct BybitOrderRequest {
    /// Trading symbol (e.g., "BTCUSDT")
    pub symbol: SmartString,
    /// Order side (Buy or Sell)
    pub side: OrderSide,
    /// Order type (Market, Limit, etc.)
    pub order_type: OrderType,
    /// Order quantity
    pub quantity: Decimal,
    /// Order price (required for limit orders)
    pub price: Option<Decimal>,
    /// Time in force (GTC, IOC, FOK, etc.)
    pub time_in_force: TimeInForce,
    /// Bybit product category (spot, linear, inverse, options)
    pub category: BybitCategory,
    /// Client-provided order link ID for tracking
    pub order_link_id: Option<SmartString>,
    /// Whether this is a reduce-only order
    pub reduce_only: Option<bool>,
    /// Whether to close position on trigger
    pub close_on_trigger: Option<bool>,
    /// Position index for hedge mode (0=one-way, 1=buy-side, 2=sell-side)
    pub position_idx: Option<u32>,
    /// Take profit price
    pub take_profit: Option<Decimal>,
    /// Stop loss price
    pub stop_loss: Option<Decimal>,
    /// Take profit trigger price type (LastPrice, MarkPrice, IndexPrice)
    pub tp_trigger_by: Option<SmartString>,
    /// Stop loss trigger price type (LastPrice, MarkPrice, IndexPrice)
    pub sl_trigger_by: Option<SmartString>,
    /// Take profit limit price for TP limit orders
    pub tp_limit_price: Option<Decimal>,
    /// Stop loss limit price for SL limit orders
    pub sl_limit_price: Option<Decimal>,
}

/// Bybit WebSocket batch order request
#[derive(Debug, Clone)]
pub struct BybitBatchOrderRequest {
    /// Bybit product category for all orders in the batch
    pub category: BybitCategory,
    /// List of orders to be placed in the batch (max 20 orders, 10 for spot)
    pub orders: Vec<BybitOrderRequest>,
}

/// Order amendment request
#[derive(Debug, Clone)]
pub struct BybitAmendOrderRequest {
    /// Bybit product category (spot, linear, inverse, options)
    pub category: BybitCategory,
    /// Trading symbol (e.g., "BTCUSDT")
    pub symbol: SmartString,
    /// Bybit order ID (either order_id or order_link_id must be provided)
    pub order_id: Option<SmartString>,
    /// Client-provided order link ID (either order_id or order_link_id must be provided)
    pub order_link_id: Option<SmartString>,
    /// New order quantity
    pub qty: Option<Decimal>,
    /// New order price
    pub price: Option<Decimal>,
    /// New take profit price
    pub take_profit: Option<Decimal>,
    /// New stop loss price
    pub stop_loss: Option<Decimal>,
    /// Take profit trigger price type (LastPrice, MarkPrice, IndexPrice)
    pub tp_trigger_by: Option<SmartString>,
    /// Stop loss trigger price type (LastPrice, MarkPrice, IndexPrice)
    pub sl_trigger_by: Option<SmartString>,
    /// New trigger price for conditional orders
    pub trigger_price: Option<Decimal>,
    /// New take profit limit price for TP limit orders
    pub tp_limit_price: Option<Decimal>,
    /// New stop loss limit price for SL limit orders
    pub sl_limit_price: Option<Decimal>,
}

/// Cancel all orders request
#[derive(Debug, Clone)]
pub struct BybitCancelAllRequest {
    /// Bybit product category (spot, linear, inverse, options)
    pub category: BybitCategory,
    /// Optional symbol filter (cancels only orders for this symbol)
    pub symbol: Option<SmartString>,
    /// Optional base coin filter (cancels only orders for this base coin)
    pub base_coin: Option<SmartString>,
    /// Optional settle coin filter (cancels only orders for this settle coin)
    pub settle_coin: Option<SmartString>,
}

/// Request tracking information for pending orders
#[derive(Debug, Clone)]
struct PendingRequest {
    /// Instrument ID associated with the request
    pub instrument_id: InstrumentId,
    /// Bybit product category (spot, linear, inverse, options)
    pub category: BybitCategory,
    /// Type of request (create, cancel, amend, batch operations)
    pub request_type: RequestType,
    /// Timestamp when the request was created (nanoseconds)
    pub timestamp: u64,
}

/// Type of request being tracked
#[derive(Debug, Clone, PartialEq)]
enum RequestType {
    /// Single order creation request
    CreateOrder,
    /// Single order cancellation request
    CancelOrder,
    /// Single order amendment request
    AmendOrder,
    /// Batch order creation request
    BatchCreate,
    /// Batch order cancellation request
    BatchCancel,
    /// Batch order amendment request
    BatchAmend,
}

/// Bybit WebSocket Trading Client
pub struct BybitWebSocketTrader {
    auth: Arc<BybitAuth>,
    testnet: bool,
    connection_health: Arc<RwLock<ConnectionHealth>>,
    ws_sink: Arc<AsyncRwLock<Option<WebSocketSink>>>,
    ws_stream: Arc<AsyncRwLock<Option<WebSocketStream>>>,
    is_connected: Arc<AtomicBool>,
    is_authenticated: Arc<AtomicBool>,
    reconnection_attempts: Arc<AtomicU8>,
    clock: Clock,
    ping_task: Arc<AsyncRwLock<Option<JoinHandle<()>>>>,
    message_task: Arc<AsyncRwLock<Option<JoinHandle<()>>>>,
    cleanup_task: Arc<AsyncRwLock<Option<JoinHandle<()>>>>,
    request_counter: Arc<AtomicU64>,
    /// Channel for authentication completion signaling
    auth_completion_tx: Arc<AsyncRwLock<Option<oneshot::Sender<Result<()>>>>>,
    /// Track pending requests to map responses to original instrument IDs
    pending_requests: Arc<AsyncRwLock<FxHashMap<SmartString, PendingRequest>>>,
    /// Track position state for position updates
    position_state: Arc<RwLock<FxHashMap<PositionId, PositionUpdate>>>,

    /// Account type (UNIFIED or CONTRACT)
    account_type: RwLock<Option<BybitAccountType>>,

    /// Instrument info cache for symbol categorization
    instrument_cache: Arc<RwLock<FxHashMap<SmartString, InstrumentInfo>>>,

    /// Instrument registry for order/instrument mapping
    instrument_registry: Arc<dyn InstrumentRegistry>,
}

impl BybitWebSocketTrader {
    /// Create a new Bybit WebSocket trader
    pub fn new(
        auth: Arc<BybitAuth>,
        testnet: bool,
        instrument_registry: Arc<dyn InstrumentRegistry>,
    ) -> Self {
        Self {
            auth,
            testnet,
            connection_health: Arc::new(RwLock::new(ConnectionHealth::default())),
            ws_sink: Arc::new(AsyncRwLock::new(None)),
            ws_stream: Arc::new(AsyncRwLock::new(None)),
            is_connected: Arc::new(AtomicBool::new(false)),
            is_authenticated: Arc::new(AtomicBool::new(false)),
            reconnection_attempts: Arc::new(AtomicU8::new(0)),
            clock: Clock::new(),
            ping_task: Arc::new(AsyncRwLock::new(None)),
            message_task: Arc::new(AsyncRwLock::new(None)),
            cleanup_task: Arc::new(AsyncRwLock::new(None)),
            request_counter: Arc::new(AtomicU64::new(0)),
            auth_completion_tx: Arc::new(AsyncRwLock::new(None)),
            pending_requests: Arc::new(AsyncRwLock::new(FxHashMap::default())),
            position_state: Arc::new(RwLock::new(FxHashMap::default())),
            account_type: RwLock::new(None),
            instrument_cache: Arc::new(RwLock::new(FxHashMap::default())),
            instrument_registry,
        }
    }

    /// Get the WebSocket URL based on testnet flag
    const fn get_ws_url(&self) -> &'static str {
        if self.testnet {
            WS_TESTNET_URL
        } else {
            WS_MAINNET_URL
        }
    }

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

    /// Check if connected and authenticated
    pub fn is_ready(&self) -> bool {
        self.is_connected.load(Ordering::Relaxed) && self.is_authenticated.load(Ordering::Relaxed)
    }

    /// Connect to Bybit WebSocket trading endpoint
    pub async fn connect(&self, report_tx: Sender<ExecutionReport>) -> Result<()> {
        let url = self.get_ws_url();
        info!("Connecting to Bybit WebSocket: {url}");

        // Create WebSocket connection
        let config = WebSocketConfig::builder(Exchange::Bybit, url.to_string())
            .connect_timeout(Duration::from_secs(30))
            .timeout(Duration::from_secs(10))
            .heartbeat(30000, 60000, 3)
            .build();

        let stats = new_shared_stats();
        let connection_state = Arc::new(RwLock::new(WsConnectionState::Disconnected));
        let mut connector =
            WebSocketConnector::new(config, stats.clone(), connection_state.clone());
        let (sink, stream) = connector.connect_with_retry(url).await?;

        // Store connection references
        *self.ws_sink.write().await = Some(sink);
        *self.ws_stream.write().await = Some(stream);

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

        // Update connection health
        {
            let mut health = self.connection_health.write();
            health.is_healthy = true;
            health.last_pong_received = self.clock.raw();
        }

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

        // Start ping task
        self.start_ping_task().await;

        // Start message processing task
        self.start_message_task(report_tx).await;

        // Start periodic cleanup task for old requests
        self.start_cleanup_task().await;

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

    /// Authenticate with Bybit WebSocket
    async fn authenticate(&self) -> Result<()> {
        let req_id = format!("auth_{}", Uuid::new_v4());
        let auth_message = self
            .auth
            .create_ws_auth_message(Some(req_id.clone().into()))
            .map_err(|e| anyhow::anyhow!("Failed to create WebSocket auth message: {}", e))?;

        // Create authentication completion channel
        let (auth_tx, auth_rx) = oneshot::channel();

        // Store the sender for use in message handling
        {
            let mut completion_tx = self.auth_completion_tx.write().await;
            *completion_tx = Some(auth_tx);
        }

        let json_str = simd_json::to_string(&auth_message)?;
        self.send_raw_message(json_str).await?;

        debug!("Sent authentication message with req_id: {req_id}");

        // Wait for authentication response with timeout
        match tokio::time::timeout(Duration::from_secs(10), auth_rx).await {
            Ok(Ok(Ok(()))) => {
                self.is_authenticated.store(true, Ordering::Relaxed);
                info!("WebSocket authentication successful");
                Ok(())
            }
            Ok(Ok(Err(e))) => {
                error!("WebSocket authentication failed: {e}");
                Err(e)
            }
            Ok(Err(_)) => {
                error!("Authentication channel was closed unexpectedly");
                bail!("Authentication channel closed")
            }
            Err(_) => {
                error!("Authentication timed out after 10 seconds");
                bail!("Authentication timeout")
            }
        }
    }

    /// Send raw message to WebSocket
    async fn send_raw_message(&self, message: String) -> Result<()> {
        if let Some(sink) = self.ws_sink.write().await.as_mut() {
            sink.send(Message::Text(message.into()).to_frame_view())
                .await?;

            // Update health metrics
            {
                let mut health = self.connection_health.write();
                health.messages_sent += 1;
            }

            Ok(())
        } else {
            bail!("WebSocket sink not available")
        }
    }

    /// Start ping task for connection health
    async fn start_ping_task(&self) {
        let sink = self.ws_sink.clone();
        let health = self.connection_health.clone();
        let clock = self.clock.clone();
        let is_connected = self.is_connected.clone();

        let task = 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;
                }

                // Send WebSocket ping
                if let Some(sink) = sink.write().await.as_mut() {
                    let ping_message = simd_json::json!({
                        "req_id": format!("ping_{}", Uuid::new_v4()),
                        "op": "ping"
                    });

                    let json_str = simd_json::to_string(&ping_message).unwrap_or_default();
                    if let Err(e) = sink
                        .send(Message::Text(json_str.into()).to_frame_view())
                        .await
                    {
                        error!("Failed to send ping: {e}");
                        break;
                    }

                    // Update health
                    {
                        let mut health_guard = health.write();
                        health_guard.last_ping_sent = clock.raw();
                    }
                }
            }
        });

        *self.ping_task.write().await = Some(task);
    }

    /// Start message processing task
    async fn start_message_task(&self, report_tx: Sender<ExecutionReport>) {
        let stream = self.ws_stream.clone();
        let health = self.connection_health.clone();
        let clock = self.clock.clone();
        let is_connected = self.is_connected.clone();
        let auth_completion_tx = self.auth_completion_tx.clone();
        let pending_requests = self.pending_requests.clone();
        let position_state = self.position_state.clone();
        let instrument_registry = self.instrument_registry.clone();

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

                if let Some(stream) = stream.write().await.as_mut() {
                    if let Some(frame) = stream.next().await {
                        let message = Message::from_frame_view(frame);

                        // Update health
                        {
                            let mut health_guard = health.write();
                            health_guard.messages_received += 1;
                            health_guard.last_pong_received = clock.raw();
                        }

                        // Process message
                        if let Err(e) = Self::process_message(
                            message,
                            &report_tx,
                            &auth_completion_tx,
                            &pending_requests,
                            &position_state,
                            &clock,
                            &instrument_registry,
                        )
                        .await
                        {
                            error!("Failed to process message: {e}");
                        }
                    } else {
                        debug!("WebSocket stream ended");
                        break;
                    }
                }
            }
        });

        *self.message_task.write().await = Some(task);
    }

    /// Process incoming WebSocket messages
    async fn process_message(
        message: Message,
        report_tx: &Sender<ExecutionReport>,
        auth_completion_tx: &Arc<AsyncRwLock<Option<oneshot::Sender<Result<()>>>>>,
        pending_requests: &Arc<AsyncRwLock<FxHashMap<SmartString, PendingRequest>>>,
        position_state: &Arc<RwLock<FxHashMap<PositionId, PositionUpdate>>>,
        clock: &Clock,
        instrument_registry: &Arc<dyn InstrumentRegistry>,
    ) -> Result<()> {
        match message {
            Message::Text(text) => {
                debug!("Received message: {text}");

                // Parse JSON - text is already UTF-8 validated by WebSocket protocol
                let mut json_str = text.clone();
                // SAFETY: json_str comes from WebSocket text frame which is UTF-8 validated
                let json: JsonValue = unsafe { simd_json::from_str(&mut json_str)? };

                // Handle different message types
                if let Some(op) = json.get("op").and_then(|v| v.as_str()) {
                    match op {
                        "auth" => {
                            Self::handle_auth_response(&json, auth_completion_tx).await?;
                        }
                        "pong" => {
                            debug!("Received pong response");
                        }
                        "order.create" | "order.cancel" | "order.amend" => {
                            Self::handle_order_response(
                                &json,
                                report_tx,
                                pending_requests,
                                clock,
                                instrument_registry,
                            )
                            .await?;
                        }
                        "order.create-batch" | "order.cancel-batch" | "order.amend-batch" => {
                            Self::handle_batch_response(
                                &json,
                                report_tx,
                                pending_requests,
                                clock,
                                instrument_registry,
                            )
                            .await?;
                        }
                        _ => {
                            debug!("Unknown operation: {op}");
                        }
                    }
                } else if let Some(topic) = json.get("topic").and_then(|v| v.as_str()) {
                    match topic {
                        "order" => {
                            Self::handle_order_update(&json, report_tx, clock, instrument_registry)
                                .await?;
                        }
                        "position" => {
                            Self::handle_position_update(
                                &json,
                                position_state,
                                Some(report_tx),
                                clock,
                            )
                            .await?;
                        }
                        _ => {
                            debug!("Unknown topic: {topic}");
                        }
                    }
                }
            }
            Message::Binary(data) => {
                debug!("Received binary message: {} bytes", data.len());
            }
            Message::Ping(data) => {
                debug!("Received ping: {} bytes", data.len());
            }
            Message::Pong(data) => {
                debug!("Received pong: {} bytes", data.len());
            }
            Message::Close(frame) => {
                warn!("Received close frame: {frame:?}");
            }
            Message::Frame(_) => {
                debug!("Received raw frame message");
            }
        }

        Ok(())
    }

    /// Handle authentication response
    async fn handle_auth_response(
        json: &JsonValue,
        auth_completion_tx: &Arc<AsyncRwLock<Option<oneshot::Sender<Result<()>>>>>,
    ) -> Result<()> {
        let result = if let Some(success) = json
            .get("success")
            .and_then(simd_json::prelude::ValueAsScalar::as_bool)
        {
            if success {
                info!("WebSocket authentication successful");
                Ok(())
            } else {
                let msg = json
                    .get("ret_msg")
                    .and_then(|v| v.as_str())
                    .unwrap_or("Unknown error");
                error!("WebSocket authentication failed: {msg}");
                Err(anyhow::anyhow!("Authentication failed: {}", msg))
            }
        } else {
            Err(anyhow::anyhow!("Invalid authentication response format"))
        };

        // Signal authentication completion
        if let Some(tx) = auth_completion_tx.write().await.take() {
            let signal_result = match &result {
                Ok(()) => Ok(()),
                Err(e) => Err(anyhow::anyhow!("{}", e)),
            };
            let _ = tx.send(signal_result);
        }

        result
    }

    /// Handle order response with request tracking
    async fn handle_order_response(
        json: &JsonValue,
        report_tx: &Sender<ExecutionReport>,
        pending_requests: &Arc<AsyncRwLock<FxHashMap<SmartString, PendingRequest>>>,
        clock: &Clock,
        instrument_registry: &Arc<dyn InstrumentRegistry>,
    ) -> Result<()> {
        debug!("Processing order response: {json:?}");

        // Extract request ID to find tracked request
        let req_id: SmartString = json
            .get("req_id")
            .and_then(|v| v.as_str())
            .map_or_else(|| "unknown".into(), std::convert::Into::into);

        // Look up the pending request
        let pending_request = {
            let pending = pending_requests.read().await;
            pending.get(&req_id).cloned()
        };

        // Extract order information and create execution report
        if let Some(ret_code) = json
            .get("retCode")
            .and_then(simd_json::prelude::ValueAsScalar::as_i64)
        {
            let success = ret_code == 0;
            let status = if success {
                OrderStatus::New
            } else {
                OrderStatus::Rejected
            };

            // Use tracked instrument ID or fallback to parsing from response
            let instrument_id = if let Some(ref request) = pending_request {
                request.instrument_id.clone()
            } else {
                // Fallback: try to extract symbol from response, then normalize
                let symbol = json
                    .get("result")
                    .and_then(|r| r.get("symbol"))
                    .and_then(|v| v.as_str())
                    .map_or_else(
                        || "BTC/USDT".into(),
                        |s| instrument_registry.normalize_symbol(s, Venue::Bybit),
                    ); // Default fallback
                InstrumentId::new(&symbol, Venue::Bybit)
            };

            let order_id: SmartString = json
                .get("result")
                .and_then(|r| r.get("orderId"))
                .and_then(|v| v.as_str())
                .unwrap_or("unknown")
                .into();

            // Cache order mapping if order was successfully created
            if success
                && order_id != "unknown"
                && let Some(ref request) = pending_request
            {
                instrument_registry.cache_order_mapping(
                    &order_id,
                    request.instrument_id.clone(),
                    None, // No metadata for now
                );
            }

            let report = ExecutionReport {
                id: id_generation::generate_report_id_with_uuid("bybit"),
                order_id: order_id.clone(),
                exchange_timestamp: json
                    .get("time")
                    .and_then(simd_json::prelude::ValueAsScalar::as_u64)
                    .unwrap_or(0)
                    * 1_000_000,
                system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
                instrument_id,
                status,
                filled_quantity: Decimal::ZERO,
                remaining_quantity: Decimal::ZERO,
                execution_price: None,
                reject_reason: if success {
                    None
                } else {
                    Some(
                        json.get("retMsg")
                            .and_then(|v| v.as_str())
                            .unwrap_or("Unknown error")
                            .into(),
                    )
                },
                exchange_execution_id: None,
                is_final: !success,
            };

            if let Err(e) = report_tx.try_send(report) {
                error!("Failed to send execution report: {e}");
            }

            // Clean up completed request
            if success || status == OrderStatus::Rejected {
                pending_requests.write().await.remove(&req_id);
            }
        }

        Ok(())
    }

    /// Handle batch response with request tracking
    async fn handle_batch_response(
        json: &JsonValue,
        report_tx: &Sender<ExecutionReport>,
        pending_requests: &Arc<AsyncRwLock<FxHashMap<SmartString, PendingRequest>>>,
        clock: &Clock,
        instrument_registry: &Arc<dyn InstrumentRegistry>,
    ) -> Result<()> {
        debug!("Processing batch response: {json:?}");

        // Extract request ID to find tracked request for default instrument info
        let req_id: SmartString = json
            .get("req_id")
            .and_then(|v| v.as_str())
            .map_or_else(|| "unknown".into(), std::convert::Into::into);

        // Look up the pending request for category information
        let pending_request = {
            let pending = pending_requests.read().await;
            pending.get(&req_id).cloned()
        };

        // Handle batch responses - each order in the batch gets its own execution report
        if let Some(result) = json.get("result") {
            if let Some(list) = result.get("list").and_then(|v| v.as_array()) {
                for (i, order) in list.iter().enumerate() {
                    // Use symbol from response if available, otherwise fallback to tracked request
                    let symbol = order
                        .get("symbol")
                        .and_then(|v| v.as_str())
                        .map(|s| instrument_registry.normalize_symbol(s, Venue::Bybit))
                        .unwrap_or_else(|| {
                            if let Some(ref request) = pending_request {
                                request.instrument_id.symbol.clone()
                            } else {
                                // Use normalized fallback instead of UNKNOWN
                                instrument_registry.normalize_symbol("BTCUSDT", Venue::Bybit)
                            }
                        });

                    let report = ExecutionReport {
                        id: id_generation::generate_batch_id("bybit", i),
                        order_id: order
                            .get("orderId")
                            .and_then(|v| v.as_str())
                            .unwrap_or("unknown")
                            .into(),
                        exchange_timestamp: order
                            .get("createAt")
                            .and_then(simd_json::prelude::ValueAsScalar::as_u64)
                            .unwrap_or(0)
                            * 1_000_000,
                        system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
                        instrument_id: InstrumentId::new(&symbol, Venue::Bybit),
                        status: OrderStatus::New,
                        filled_quantity: Decimal::ZERO,
                        remaining_quantity: Decimal::ZERO,
                        execution_price: None,
                        reject_reason: None,
                        exchange_execution_id: order
                            .get("orderId")
                            .and_then(|v| v.as_str())
                            .map(std::convert::Into::into),
                        is_final: false,
                    };

                    if let Err(e) = report_tx.try_send(report) {
                        error!("Failed to send batch execution report: {e}");
                    }
                }
            }

            // Clean up completed batch request
            pending_requests.write().await.remove(&req_id);
        }

        Ok(())
    }

    /// Handle order update
    async fn handle_order_update(
        json: &JsonValue,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
        instrument_registry: &Arc<dyn InstrumentRegistry>,
    ) -> Result<()> {
        debug!("Processing order update: {json:?}");

        if let Some(data) = json.get("data").and_then(|v| v.as_array()) {
            for order in data {
                let status = Self::map_order_status(
                    order
                        .get("orderStatus")
                        .and_then(|v| v.as_str())
                        .unwrap_or("Unknown"),
                );

                let report = ExecutionReport {
                    id: id_generation::generate_report_id_with_uuid("bybit_update"),
                    order_id: order
                        .get("orderId")
                        .and_then(|v| v.as_str())
                        .unwrap_or("unknown")
                        .into(),
                    exchange_timestamp: order
                        .get("updatedTime")
                        .and_then(simd_json::prelude::ValueAsScalar::as_u64)
                        .unwrap_or(0)
                        * 1_000_000,
                    system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
                    instrument_id: InstrumentId::new(
                        order.get("symbol").and_then(|v| v.as_str()).map_or_else(
                            || instrument_registry.normalize_symbol("BTCUSDT", Venue::Bybit),
                            |s| instrument_registry.normalize_symbol(s, Venue::Bybit),
                        ),
                        Venue::Bybit,
                    ),
                    status,
                    filled_quantity: order
                        .get("cumExecQty")
                        .and_then(|v| v.as_str())
                        .and_then(|s| s.parse().ok())
                        .unwrap_or(Decimal::ZERO),
                    remaining_quantity: order
                        .get("leavesQty")
                        .and_then(|v| v.as_str())
                        .and_then(|s| s.parse().ok())
                        .unwrap_or(Decimal::ZERO),
                    execution_price: order
                        .get("avgPrice")
                        .and_then(|v| v.as_str())
                        .and_then(|s| s.parse().ok()),
                    reject_reason: None,
                    exchange_execution_id: order
                        .get("orderId")
                        .and_then(|v| v.as_str())
                        .map(std::convert::Into::into),
                    is_final: matches!(
                        status,
                        OrderStatus::Filled | OrderStatus::Cancelled | OrderStatus::Rejected
                    ),
                };

                if let Err(e) = report_tx.try_send(report) {
                    error!("Failed to send order update report: {e}");
                }
            }
        }

        Ok(())
    }

    /// Handle position update
    async fn handle_position_update(
        json: &JsonValue,
        position_state: &Arc<RwLock<FxHashMap<PositionId, PositionUpdate>>>,
        report_tx: Option<&Sender<ExecutionReport>>,
        clock: &Clock,
    ) -> Result<()> {
        debug!("Processing position update: {json:?}");

        // Parse position updates from the data array
        if let Some(data) = json.get("data").and_then(|v| v.as_array()) {
            let timestamp_ns = clock.raw();

            for position_json in data {
                if let Some(position_update) =
                    Self::parse_position_update(position_json, timestamp_ns)
                {
                    // Update position state
                    {
                        let mut state = position_state.write();
                        state.insert(position_update.position_id, position_update.clone());
                    }

                    // Log position update
                    info!(
                        "Position update: {} {} {} @ {}",
                        position_update.symbol,
                        position_update.side,
                        position_update.amount,
                        position_update.entry_price
                    );

                    // Send position update via execution report if channel is available
                    if let Some(tx) = report_tx {
                        let report = ExecutionReport {
                            id: id_generation::generate_report_id_with_uuid("bybit_position"),
                            order_id: SmartString::default(), // Position updates don't have order IDs
                            exchange_timestamp: timestamp_ns,
                            system_timestamp: timestamp_ns,
                            instrument_id: InstrumentId::new(
                                position_update.symbol.clone(),
                                Venue::Bybit,
                            ),
                            status: OrderStatus::New, // Use a placeholder status for position updates
                            filled_quantity: position_update.amount,
                            remaining_quantity: Decimal::ZERO, // Position updates don't have remaining quantity
                            execution_price: Some(position_update.entry_price),
                            reject_reason: None,
                            exchange_execution_id: Some(
                                format!("position_{}", position_update.position_id).into(),
                            ),
                            is_final: false,
                        };

                        if let Err(e) = tx.try_send(report) {
                            error!("Failed to send position update report: {e}");
                        }
                    }
                }
            }
        }

        Ok(())
    }

    /// Parse Bybit position update from JSON
    fn parse_position_update(
        position_json: &JsonValue,
        timestamp_ns: u64,
    ) -> Option<PositionUpdate> {
        // Extract position fields from Bybit position JSON
        let symbol = position_json.get("symbol").and_then(|v| v.as_str())?;
        let side_str = position_json.get("side").and_then(|v| v.as_str())?;
        let size_str = position_json.get("size").and_then(|v| v.as_str())?;
        let position_value_str = position_json
            .get("positionValue")
            .and_then(|v| v.as_str())?;
        let entry_price_str = position_json.get("entryPrice").and_then(|v| v.as_str())?;
        let unrealized_pnl_str = position_json
            .get("unrealisedPnl")
            .and_then(|v| v.as_str())?;
        let realized_pnl_str = position_json
            .get("cumRealisedPnl")
            .and_then(|v| v.as_str())?;
        let margin_mode_str = position_json.get("tradeMode").and_then(|v| v.as_str())?;

        // Parse numeric values
        let amount = Decimal::from_str(size_str).ok()?;
        let position_value = Decimal::from_str(position_value_str).ok()?;
        let entry_price = Decimal::from_str(entry_price_str).ok()?;
        let unrealized_pnl = Decimal::from_str(unrealized_pnl_str).ok()?;
        let realized_pnl = Decimal::from_str(realized_pnl_str).ok()?;

        // Parse side (Bybit uses "Buy" and "Sell")
        let position_side = match side_str {
            "Buy" => PositionSide::Long,
            "Sell" => PositionSide::Short,
            _ => {
                debug!("Unknown position side: {side_str}");
                return None;
            }
        };

        // Parse margin type
        let margin_type = match margin_mode_str {
            "cross_margin" | "0" => MarginType::Cross,
            "isolated_margin" | "1" => MarginType::Isolated,
            _ => MarginType::Cross, // Default to cross margin
        };

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

        // Generate stable position ID
        let position_id = Self::generate_stable_position_id(symbol, position_side, Venue::Bybit);

        Some(PositionUpdate {
            position_id,
            venue: Venue::Bybit,
            symbol: SmartString::from(symbol),
            side: position_side,
            amount,
            entry_price,
            breakeven_price: entry_price, // Bybit doesn't provide separate breakeven price
            unrealized_pnl,
            realized_pnl,
            margin_type,
            isolated_wallet: position_value, // Use position value as isolated wallet
            timestamp_ns,
        })
    }

    /// Generate stable position ID based on position characteristics
    fn generate_stable_position_id(symbol: &str, side: PositionSide, venue: Venue) -> PositionId {
        // Create a deterministic UUID based on position characteristics
        // This ensures the same position always gets the same ID
        let position_key = format!("{symbol}-{side}-{venue:?}");

        // Use stable hashing algorithm to ensure consistent IDs across restarts
        // Using SHA-256 truncated to ensure deterministic results across different
        // compiler versions and architectures (unlike FxHasher)
        use sha2::{Digest, Sha256};

        // Create primary hash from position key
        let mut hasher = Sha256::new();
        hasher.update(position_key.as_bytes());
        let primary_hash = hasher.finalize();

        // Create secondary hash with additional entropy for uniqueness
        let mut secondary_hasher = Sha256::new();
        secondary_hasher.update((symbol.len() as u64).to_le_bytes());
        secondary_hasher.update((side as u8).to_le_bytes());
        secondary_hasher.update((venue as u8).to_le_bytes());
        secondary_hasher.update(b"position_id_v1"); // Version marker for future compatibility
        let secondary_hash = secondary_hasher.finalize();

        // Create 128-bit UUID from the two hashes
        // Take first 8 bytes from each hash to create a 16-byte UUID
        let mut uuid_bytes = [0u8; 16];

        // Use first 8 bytes of primary hash
        uuid_bytes[0..8].copy_from_slice(&primary_hash[0..8]);

        // Use first 8 bytes of secondary hash
        uuid_bytes[8..16].copy_from_slice(&secondary_hash[0..8]);

        // Create UUID from bytes
        let stable_uuid = Uuid::from_bytes(uuid_bytes);
        PositionId::from_uuid(stable_uuid)
    }

    /// Map Bybit order status to internal `OrderStatus`
    fn map_order_status(status: &str) -> OrderStatus {
        match status {
            "Created" | "New" => OrderStatus::New,
            "PartiallyFilled" => OrderStatus::PartiallyFilled,
            "Filled" => OrderStatus::Filled,
            "Cancelled" => OrderStatus::Cancelled,
            "Rejected" => OrderStatus::Rejected,
            "PendingCancel" => OrderStatus::Pending,
            "Deactivated" => OrderStatus::Cancelled,
            "Active" => OrderStatus::Open,
            "Untriggered" => OrderStatus::Pending,
            "Triggered" => OrderStatus::Open,
            _ => OrderStatus::Unknown,
        }
    }

    /// Map internal `OrderSide` to Bybit order side
    const fn map_order_side(side: OrderSide) -> &'static str {
        match side {
            OrderSide::Buy => "Buy",
            OrderSide::Sell => "Sell",
        }
    }

    /// Map internal `OrderType` to Bybit order type
    const fn map_order_type(order_type: OrderType) -> &'static str {
        match order_type {
            OrderType::Market => "Market",
            OrderType::Limit => "Limit",
            _ => "Limit",
        }
    }

    /// Map internal `TimeInForce` to Bybit time in force
    const fn map_time_in_force(tif: TimeInForce) -> &'static str {
        match tif {
            TimeInForce::GTC => "GTC",
            TimeInForce::IOC => "IOC",
            TimeInForce::FOK => "FOK",
            TimeInForce::GTX => "PostOnly",
            _ => "GTC",
        }
    }

    /// Track a pending request
    async fn track_request(
        &self,
        req_id: SmartString,
        instrument_id: InstrumentId,
        category: BybitCategory,
        request_type: RequestType,
    ) {
        let pending_request = PendingRequest {
            instrument_id,
            category,
            request_type,
            timestamp: self.clock.raw(),
        };

        self.pending_requests
            .write()
            .await
            .insert(req_id, pending_request);
    }

    /// Find the category for an order ID by looking through tracked requests
    async fn find_order_category(&self, order_id: &str) -> Option<BybitCategory> {
        let pending = self.pending_requests.read().await;

        // Search through pending requests to find matching order
        // This is a fallback for when we don't have the req_id directly
        for (_, request) in pending.iter() {
            if matches!(
                request.request_type,
                RequestType::CreateOrder | RequestType::BatchCreate
            ) {
                // In a production system, we'd maintain order_id -> category mapping
                // For now, return the first create request's category as a fallback
                return Some(request.category.clone());
            }
        }

        None
    }

    /// Clean up old pending requests (older than 5 minutes)
    async fn cleanup_old_requests(&self) {
        let now = self.clock.raw();
        let max_age_ns = 5 * 60 * 1_000_000_000; // 5 minutes in nanoseconds

        let mut pending = self.pending_requests.write().await;
        pending.retain(|_, request| now.saturating_sub(request.timestamp) < max_age_ns);
    }

    /// Create a single order with tracking
    pub async fn create_order(&self, order: &BybitOrderRequest) -> Result<()> {
        let req_id = id_generation::generate_exchange_request_id(
            "bybit",
            "create",
            self.request_counter.fetch_add(1, Ordering::Relaxed),
        );
        let timestamp = BybitAuth::get_timestamp();
        let header = self.auth.create_ws_trading_header(timestamp);

        // Create instrument ID from order symbol
        let instrument_id = InstrumentId::new(&order.symbol, Venue::Bybit);

        // Track the request
        self.track_request(
            req_id.clone(),
            instrument_id,
            order.category.clone(),
            RequestType::CreateOrder,
        )
        .await;

        let mut order_data = simd_json::json!({
            "category": order.category.as_str(),
            "symbol": order.symbol,
            "side": Self::map_order_side(order.side),
            "orderType": Self::map_order_type(order.order_type),
            "qty": order.quantity.to_string(),
            "price": order.price.map_or("0".to_string(), |p| p.to_string()),
            "timeInForce": Self::map_time_in_force(order.time_in_force),
            "orderLinkId": order.order_link_id.as_ref().map_or(String::new(), std::string::ToString::to_string),
            "reduceOnly": order.reduce_only.unwrap_or(false),
            "closeOnTrigger": order.close_on_trigger.unwrap_or(false),
            "positionIdx": order.position_idx.unwrap_or(0)
        });

        // Add TP/SL parameters if provided
        if let Some(tp) = order.take_profit {
            order_data["takeProfit"] = simd_json::json!(tp.to_string());
        }
        if let Some(sl) = order.stop_loss {
            order_data["stopLoss"] = simd_json::json!(sl.to_string());
        }
        if let Some(tp_trigger_by) = &order.tp_trigger_by {
            order_data["tpTriggerBy"] = simd_json::json!(tp_trigger_by);
        }
        if let Some(sl_trigger_by) = &order.sl_trigger_by {
            order_data["slTriggerBy"] = simd_json::json!(sl_trigger_by);
        }
        if let Some(tp_limit_price) = order.tp_limit_price {
            order_data["tpLimitPrice"] = simd_json::json!(tp_limit_price.to_string());
        }
        if let Some(sl_limit_price) = order.sl_limit_price {
            order_data["slLimitPrice"] = simd_json::json!(sl_limit_price.to_string());
        }

        let message = BybitWsTradingMessage {
            req_id,
            header,
            op: "order.create".into(),
            args: vec![order_data],
        };

        let json_str = simd_json::to_string(&message)?;
        self.send_raw_message(json_str).await?;

        Ok(())
    }

    /// Create multiple orders in a batch with tracking
    pub async fn create_batch_orders(&self, batch: &BybitBatchOrderRequest) -> Result<()> {
        let req_id = id_generation::generate_exchange_request_id(
            "bybit",
            "batch_create",
            self.request_counter.fetch_add(1, Ordering::Relaxed),
        );
        let timestamp = BybitAuth::get_timestamp();
        let header = self.auth.create_ws_trading_header(timestamp);

        // Track the batch request with the first order's symbol as representative
        if let Some(first_order) = batch.orders.first() {
            let instrument_id = InstrumentId::new(&first_order.symbol, Venue::Bybit);
            self.track_request(
                req_id.clone(),
                instrument_id,
                batch.category.clone(),
                RequestType::BatchCreate,
            )
            .await;
        }

        let mut batch_data = Vec::new();
        for order in &batch.orders {
            let mut order_data = simd_json::json!({
                "category": batch.category.as_str(),
                "symbol": order.symbol,
                "side": Self::map_order_side(order.side),
                "orderType": Self::map_order_type(order.order_type),
                "qty": order.quantity.to_string(),
                "price": order.price.map_or("0".to_string(), |p| p.to_string()),
                "timeInForce": Self::map_time_in_force(order.time_in_force),
                "orderLinkId": order.order_link_id.as_ref().map_or(String::new(), std::string::ToString::to_string),
                "reduceOnly": order.reduce_only.unwrap_or(false),
                "closeOnTrigger": order.close_on_trigger.unwrap_or(false),
                "positionIdx": order.position_idx.unwrap_or(0)
            });

            // Add TP/SL parameters if provided
            if let Some(tp) = order.take_profit {
                order_data["takeProfit"] = simd_json::json!(tp.to_string());
            }
            if let Some(sl) = order.stop_loss {
                order_data["stopLoss"] = simd_json::json!(sl.to_string());
            }
            if let Some(tp_trigger_by) = &order.tp_trigger_by {
                order_data["tpTriggerBy"] = simd_json::json!(tp_trigger_by);
            }
            if let Some(sl_trigger_by) = &order.sl_trigger_by {
                order_data["slTriggerBy"] = simd_json::json!(sl_trigger_by);
            }
            if let Some(tp_limit_price) = order.tp_limit_price {
                order_data["tpLimitPrice"] = simd_json::json!(tp_limit_price.to_string());
            }
            if let Some(sl_limit_price) = order.sl_limit_price {
                order_data["slLimitPrice"] = simd_json::json!(sl_limit_price.to_string());
            }

            batch_data.push(order_data);
        }

        let message = BybitWsTradingMessage {
            req_id,
            header,
            op: "order.create-batch".into(),
            args: batch_data,
        };

        let json_str = simd_json::to_string(&message)?;
        self.send_raw_message(json_str).await?;

        Ok(())
    }

    /// Cancel a single order with tracking
    pub async fn cancel_order(&self, category: BybitCategory, order_id: &str) -> Result<()> {
        let req_id = id_generation::generate_exchange_request_id(
            "bybit",
            "cancel",
            self.request_counter.fetch_add(1, Ordering::Relaxed),
        );
        let timestamp = BybitAuth::get_timestamp();
        let header = self.auth.create_ws_trading_header(timestamp);

        // Track the cancellation request - lookup instrument from registry
        let instrument_id = self
            .instrument_registry
            .lookup_by_order_id(order_id)
            .unwrap_or_else(|| {
                // Fallback: create instrument with normalized symbol from order_id
                let normalized_symbol = self
                    .instrument_registry
                    .normalize_symbol(order_id, Venue::Bybit);
                InstrumentId::new(normalized_symbol, Venue::Bybit)
            });
        self.track_request(
            req_id.clone(),
            instrument_id,
            category.clone(),
            RequestType::CancelOrder,
        )
        .await;

        let cancel_data = simd_json::json!({
            "category": category.as_str(),
            "orderId": order_id
        });

        let message = BybitWsTradingMessage {
            req_id,
            header,
            op: "order.cancel".into(),
            args: vec![cancel_data],
        };

        let json_str = simd_json::to_string(&message)?;
        self.send_raw_message(json_str).await?;

        Ok(())
    }

    /// Cancel multiple orders in a batch with tracking
    pub async fn cancel_batch_orders(
        &self,
        category: BybitCategory,
        order_ids: &[&str],
    ) -> Result<()> {
        let req_id = id_generation::generate_exchange_request_id(
            "bybit",
            "batch_cancel",
            self.request_counter.fetch_add(1, Ordering::Relaxed),
        );
        let timestamp = BybitAuth::get_timestamp();
        let header = self.auth.create_ws_trading_header(timestamp);

        // Track the batch cancellation request - use first order for tracking
        let instrument_id = if let Some(first_order_id) = order_ids.first() {
            self.instrument_registry
                .lookup_by_order_id(first_order_id)
                .unwrap_or_else(|| {
                    let normalized_symbol = self
                        .instrument_registry
                        .normalize_symbol("BTCUSDT", Venue::Bybit);
                    InstrumentId::new(&normalized_symbol, Venue::Bybit)
                })
        } else {
            // Fallback for empty batch - use default symbol
            let normalized_symbol = self
                .instrument_registry
                .normalize_symbol("BTCUSDT", Venue::Bybit);
            InstrumentId::new(&normalized_symbol, Venue::Bybit)
        };
        self.track_request(
            req_id.clone(),
            instrument_id,
            category.clone(),
            RequestType::BatchCancel,
        )
        .await;

        let mut batch_data = Vec::new();
        for order_id in order_ids {
            let cancel_data = simd_json::json!({
                "category": category.as_str(),
                "orderId": order_id
            });
            batch_data.push(cancel_data);
        }

        let message = BybitWsTradingMessage {
            req_id,
            header,
            op: "order.cancel-batch".into(),
            args: batch_data,
        };

        let json_str = simd_json::to_string(&message)?;
        self.send_raw_message(json_str).await?;

        Ok(())
    }

    /// Amend an existing order
    pub async fn amend_order(&self, request: &BybitAmendOrderRequest) -> Result<()> {
        let req_id = id_generation::generate_exchange_request_id(
            "bybit",
            "amend",
            self.request_counter.fetch_add(1, Ordering::Relaxed),
        );
        let timestamp = BybitAuth::get_timestamp();
        let header = self.auth.create_ws_trading_header(timestamp);

        // Build amend data
        let mut amend_data = simd_json::json!({
            "category": request.category.as_str(),
            "symbol": request.symbol,
        });

        // Add order identifier (either orderId or orderLinkId required)
        if let Some(order_id) = &request.order_id {
            amend_data["orderId"] = simd_json::json!(order_id);
        } else if let Some(order_link_id) = &request.order_link_id {
            amend_data["orderLinkId"] = simd_json::json!(order_link_id);
        } else {
            bail!("Either orderId or orderLinkId must be provided");
        }

        // Add optional amendment fields
        if let Some(qty) = request.qty {
            amend_data["qty"] = simd_json::json!(qty.to_string());
        }
        if let Some(price) = request.price {
            amend_data["price"] = simd_json::json!(price.to_string());
        }
        if let Some(tp) = request.take_profit {
            amend_data["takeProfit"] = simd_json::json!(tp.to_string());
        }
        if let Some(sl) = request.stop_loss {
            amend_data["stopLoss"] = simd_json::json!(sl.to_string());
        }
        if let Some(tp_trigger_by) = &request.tp_trigger_by {
            amend_data["tpTriggerBy"] = simd_json::json!(tp_trigger_by);
        }
        if let Some(sl_trigger_by) = &request.sl_trigger_by {
            amend_data["slTriggerBy"] = simd_json::json!(sl_trigger_by);
        }
        if let Some(trigger_price) = request.trigger_price {
            amend_data["triggerPrice"] = simd_json::json!(trigger_price.to_string());
        }
        if let Some(tp_limit_price) = request.tp_limit_price {
            amend_data["tpLimitPrice"] = simd_json::json!(tp_limit_price.to_string());
        }
        if let Some(sl_limit_price) = request.sl_limit_price {
            amend_data["slLimitPrice"] = simd_json::json!(sl_limit_price.to_string());
        }

        let message = BybitWsTradingMessage {
            req_id,
            header,
            op: "order.amend".into(),
            args: vec![amend_data],
        };

        let json_str = simd_json::to_string(&message)?;
        self.send_raw_message(json_str).await?;

        Ok(())
    }

    /// Cancel all orders for a category
    pub async fn cancel_all_orders_internal(&self, request: &BybitCancelAllRequest) -> Result<()> {
        let req_id = id_generation::generate_exchange_request_id(
            "bybit",
            "cancel_all",
            self.request_counter.fetch_add(1, Ordering::Relaxed),
        );
        let timestamp = BybitAuth::get_timestamp();
        let header = self.auth.create_ws_trading_header(timestamp);

        // Build cancel all data
        let mut cancel_data = simd_json::json!({
            "category": request.category.as_str(),
        });

        // Add optional filters
        if let Some(symbol) = &request.symbol {
            cancel_data["symbol"] = simd_json::json!(symbol);
        }
        if let Some(base_coin) = &request.base_coin {
            cancel_data["baseCoin"] = simd_json::json!(base_coin);
        }
        if let Some(settle_coin) = &request.settle_coin {
            cancel_data["settleCoin"] = simd_json::json!(settle_coin);
        }

        let message = BybitWsTradingMessage {
            req_id,
            header,
            op: "order.cancel-all".into(),
            args: vec![cancel_data],
        };

        let json_str = simd_json::to_string(&message)?;
        self.send_raw_message(json_str).await?;

        Ok(())
    }

    /// Find order info (category and symbol) from order ID
    async fn find_order_info(&self, order_id: &str) -> Option<(BybitCategory, SmartString)> {
        let pending = self.pending_requests.read().await;
        for (_, request) in pending.iter() {
            // Try to match based on instrument symbol or order_id
            if request.instrument_id.symbol.as_str() == order_id {
                return Some((
                    request.category.clone(),
                    request.instrument_id.symbol.clone(),
                ));
            }
        }
        None
    }

    /// Send JSON ping to test connection
    pub async fn send_ping(&self) -> Result<()> {
        let ping_message = simd_json::json!({
            "req_id": format!("ping_{}", Uuid::new_v4()),
            "op": "ping"
        });

        let json_str = simd_json::to_string(&ping_message)?;
        self.send_raw_message(json_str).await?;

        Ok(())
    }

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

        // Stop tasks
        if let Some(task) = self.ping_task.write().await.take() {
            task.abort();
        }
        if let Some(task) = self.message_task.write().await.take() {
            task.abort();
        }
        if let Some(task) = self.cleanup_task.write().await.take() {
            task.abort();
        }

        // Close WebSocket connection
        if let Some(sink) = self.ws_sink.write().await.take() {
            drop(sink);
        }
        if let Some(stream) = self.ws_stream.write().await.take() {
            drop(stream);
        }

        // Update health
        {
            let mut health = self.connection_health.write();
            health.is_healthy = false;
        }

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

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

    async fn place_order(
        &self,
        order: Order,
        report_sender: Sender<crate::execution_engine::ExecutionReport>,
    ) -> Result<()> {
        // Check if connected
        if !self.is_connected().await {
            bail!("WebSocket not connected to Bybit");
        }

        // Clone order fields before moving order to convert_order
        let order_id = order.id;
        let symbol = order.symbol.clone();
        let venue = order.venue;
        let quantity = order.quantity;

        // Convert Order to BybitOrderRequest
        let bybit_order = self.convert_order(order)?;

        // Send immediate acknowledgment for order placement
        let ack_report = ExecutionReport {
            id: Uuid::new_v4().to_string().into(),
            order_id: order_id.to_string().into(),
            exchange_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            instrument_id: InstrumentId::new(symbol, venue),
            status: OrderStatus::New,
            filled_quantity: Decimal::ZERO,
            remaining_quantity: quantity,
            execution_price: None,
            reject_reason: None,
            exchange_execution_id: None,
            is_final: false,
        };

        if let Err(e) = report_sender.send_async(ack_report).await {
            warn!("Failed to send order placement acknowledgment: {e}");
        }

        // Send order via WebSocket
        self.create_order(&bybit_order).await
    }

    async fn cancel_order(
        &self,
        order_id: SmartString,
        report_sender: Sender<crate::execution_engine::ExecutionReport>,
    ) -> Result<()> {
        // Check if connected
        if !self.is_connected().await {
            bail!("WebSocket not connected to Bybit");
        }

        // Send immediate acknowledgment for order cancellation request
        let cancel_ack_report = ExecutionReport {
            id: Uuid::new_v4().to_string().into(),
            order_id: order_id.clone(),
            exchange_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            instrument_id: self
                .instrument_registry
                .lookup_by_order_id(&order_id)
                .unwrap_or_else(|| {
                    // Fallback: create instrument with normalized symbol
                    let normalized_symbol = self
                        .instrument_registry
                        .normalize_symbol(&order_id, Venue::Bybit);
                    InstrumentId::new(normalized_symbol, Venue::Bybit)
                }),
            status: OrderStatus::Pending,
            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) = report_sender.send_async(cancel_ack_report).await {
            warn!("Failed to send order cancellation acknowledgment: {e}");
        }

        // Cancel via WebSocket
        // Try to find the category from tracked requests, fallback to Spot
        let category = self
            .find_order_category(&order_id)
            .await
            .unwrap_or(BybitCategory::Spot);

        self.cancel_order(category, &order_id).await
    }

    async fn modify_order(
        &self,
        order_id: SmartString,
        new_price: Option<Decimal>,
        new_quantity: Option<Decimal>,
        report_sender: Sender<crate::execution_engine::ExecutionReport>,
    ) -> Result<()> {
        // Check if connected
        if !self.is_connected().await {
            bail!("WebSocket not connected to Bybit");
        }

        // Send immediate acknowledgment for order modification request
        let modify_ack_report = ExecutionReport {
            id: Uuid::new_v4().to_string().into(),
            order_id: order_id.clone(),
            exchange_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            instrument_id: self
                .instrument_registry
                .lookup_by_order_id(&order_id)
                .unwrap_or_else(|| {
                    // Fallback: create instrument with normalized symbol
                    let normalized_symbol = self
                        .instrument_registry
                        .normalize_symbol(&order_id, Venue::Bybit);
                    InstrumentId::new(normalized_symbol, Venue::Bybit)
                }),
            status: OrderStatus::Pending,
            filled_quantity: Decimal::ZERO,
            remaining_quantity: new_quantity.unwrap_or(Decimal::ZERO),
            execution_price: new_price,
            reject_reason: None,
            exchange_execution_id: None,
            is_final: false,
        };

        if let Err(e) = report_sender.send_async(modify_ack_report).await {
            warn!("Failed to send order modification acknowledgment: {e}");
        }

        // Find order category and symbol
        let (category, symbol) = self
            .find_order_info(&order_id)
            .await
            .ok_or_else(|| anyhow::anyhow!("Order not found in tracking"))?;

        // Create amendment request
        let amend_request = BybitAmendOrderRequest {
            category,
            symbol,
            order_id: Some(order_id.clone()),
            order_link_id: None,
            qty: new_quantity,
            price: new_price,
            take_profit: None,
            stop_loss: None,
            tp_trigger_by: None,
            sl_trigger_by: None,
            trigger_price: None,
            tp_limit_price: None,
            sl_limit_price: None,
        };

        self.amend_order(&amend_request).await
    }

    async fn cancel_all_orders(
        &self,
        instrument_id: Option<InstrumentId>,
        report_sender: Sender<crate::execution_engine::ExecutionReport>,
    ) -> Result<()> {
        // Check if connected
        if !self.is_connected().await {
            bail!("WebSocket not connected to Bybit");
        }

        // Clone instrument_id for use in both places
        let instrument_id_for_report = instrument_id.clone();

        // Send immediate acknowledgment for cancel all request
        // Note: Individual order cancellations will come through WebSocket stream
        let cancel_all_ack_report = ExecutionReport {
            id: Uuid::new_v4().to_string().into(),
            order_id: "CANCEL_ALL".into(),
            exchange_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            instrument_id: instrument_id_for_report
                .unwrap_or_else(|| InstrumentId::new("ALL", rusty_model::venues::Venue::Bybit)),
            status: OrderStatus::New, // Indicating the cancel-all request was accepted
            filled_quantity: Decimal::ZERO,
            remaining_quantity: Decimal::ZERO,
            execution_price: None,
            reject_reason: None,
            exchange_execution_id: None,
            is_final: true, // Cancel-all is a one-time operation
        };

        if let Err(e) = report_sender.send_async(cancel_all_ack_report).await {
            warn!("Failed to send cancel-all acknowledgment: {e}");
        }

        // Create cancel all request
        let cancel_request = BybitCancelAllRequest {
            category: BybitCategory::Spot, // Default to spot, could be made configurable
            symbol: instrument_id.map(|id| id.symbol),
            base_coin: None,
            settle_coin: None,
        };

        self.cancel_all_orders_internal(&cancel_request).await
    }

    async fn get_order_status(&self, order_id: &str) -> Result<OrderStatus> {
        // Bybit WebSocket doesn't support order queries - must use REST API
        bail!("Order status query not supported via WebSocket - use REST API")
    }

    async fn connect(
        &self,
        report_sender: Sender<crate::execution_engine::ExecutionReport>,
    ) -> Result<()> {
        // Use the proper report_sender channel instead of creating a dummy one
        // Call the inherent method directly to avoid recursion
        Self::connect(self, report_sender).await
    }

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

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

    async fn get_instruments(&self) -> Result<SmallVec<[InstrumentId; 32]>> {
        // Instrument discovery not supported via WebSocket
        bail!("Instrument discovery not supported via WebSocket - use REST API")
    }

    async fn send_fix_message(&self, _message: Vec<u8>) -> Result<()> {
        anyhow::bail!("FIX protocol not supported on Bybit WebSocket")
    }

    async fn receive_fix_message(&self) -> Result<Vec<u8>> {
        anyhow::bail!("FIX protocol not supported on Bybit WebSocket")
    }
}

// Helper methods for Exchange trait implementation
impl BybitWebSocketTrader {
    /// Start cleanup task for old pending requests
    async fn start_cleanup_task(&self) {
        let pending_requests = self.pending_requests.clone();
        let clock = self.clock.clone();

        let task = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(60)); // Cleanup every minute

            loop {
                interval.tick().await;

                let now = clock.raw();
                let max_age_ns = 5 * 60 * 1_000_000_000; // 5 minutes in nanoseconds

                let mut pending = pending_requests.write().await;
                let count_before = pending.len();
                pending.retain(|_, request| now.saturating_sub(request.timestamp) < max_age_ns);
                let count_after = pending.len();

                if count_before > count_after {
                    debug!(
                        "Cleaned up {} old pending requests",
                        count_before - count_after
                    );
                }
            }
        });

        *self.cleanup_task.write().await = Some(task);
    }

    /// Convert generic Order to `BybitOrderRequest`
    fn convert_order(&self, order: Order) -> Result<BybitOrderRequest> {
        Ok(BybitOrderRequest {
            symbol: order.symbol.clone(),
            side: order.side,
            order_type: order.order_type,
            quantity: order.quantity,
            price: order.price,
            time_in_force: TimeInForce::GTC, // Default to GTC
            category: BybitCategory::Spot,   // Default to spot
            order_link_id: Some(order.id.to_string().into()),
            reduce_only: None,
            close_on_trigger: None,
            position_idx: None,
            // TP/SL fields - initialize as None
            take_profit: None,
            stop_loss: None,
            tp_trigger_by: None,
            sl_trigger_by: None,
            tp_limit_price: None,
            sl_limit_price: None,
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::execution_engine::Exchange;
    use crate::instrument_registry::create_shared_registry;
    use rust_decimal::Decimal;
    use rusty_model::enums::OrderType;
    use rusty_model::trading_order::Order;
    use rusty_model::types::ClientId;
    use std::str::FromStr;
    use std::sync::Arc;

    #[tokio::test]
    async fn test_exchange_trait_implementation() {
        // Create a mock auth
        let api_key = "test_api_key".into();
        let api_secret = "test_api_secret".into();
        let auth = Arc::new(BybitAuth::new(api_key, api_secret));

        // Create trader instance (mainnet)
        let registry = create_shared_registry();
        let trader = BybitWebSocketTrader::new(auth, false, registry);

        // Test venue method
        assert_eq!(trader.venue(), Venue::Bybit);

        // Test is_connected when not connected
        assert!(!trader.is_connected().await);

        // Create test order
        let test_order = Order::new(
            Venue::Bybit,
            "BTCUSDT",
            OrderSide::Buy,
            OrderType::Limit,
            Decimal::from_str("0.001").unwrap(),
            Some(Decimal::from_str("50000").unwrap()),
            ClientId::new("test_client"),
        );

        // Test that place_order returns error when not connected
        let (report_tx, _report_rx) = flume::bounded(100);
        let result = trader.place_order(test_order, report_tx).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("not connected"));
    }

    #[test]
    fn test_order_status_mapping() {
        assert_eq!(
            BybitWebSocketTrader::map_order_status("New"),
            OrderStatus::New
        );
        assert_eq!(
            BybitWebSocketTrader::map_order_status("PartiallyFilled"),
            OrderStatus::PartiallyFilled
        );
        assert_eq!(
            BybitWebSocketTrader::map_order_status("Filled"),
            OrderStatus::Filled
        );
        assert_eq!(
            BybitWebSocketTrader::map_order_status("Cancelled"),
            OrderStatus::Cancelled
        );
        assert_eq!(
            BybitWebSocketTrader::map_order_status("Rejected"),
            OrderStatus::Rejected
        );
        assert_eq!(
            BybitWebSocketTrader::map_order_status("Unknown"),
            OrderStatus::Unknown
        );
    }

    #[test]
    fn test_order_side_mapping() {
        assert_eq!(BybitWebSocketTrader::map_order_side(OrderSide::Buy), "Buy");
        assert_eq!(
            BybitWebSocketTrader::map_order_side(OrderSide::Sell),
            "Sell"
        );
    }

    #[test]
    fn test_order_type_mapping() {
        assert_eq!(
            BybitWebSocketTrader::map_order_type(OrderType::Market),
            "Market"
        );
        assert_eq!(
            BybitWebSocketTrader::map_order_type(OrderType::Limit),
            "Limit"
        );
    }

    #[test]
    fn test_time_in_force_mapping() {
        assert_eq!(
            BybitWebSocketTrader::map_time_in_force(TimeInForce::GTC),
            "GTC"
        );
        assert_eq!(
            BybitWebSocketTrader::map_time_in_force(TimeInForce::IOC),
            "IOC"
        );
        assert_eq!(
            BybitWebSocketTrader::map_time_in_force(TimeInForce::FOK),
            "FOK"
        );
        assert_eq!(
            BybitWebSocketTrader::map_time_in_force(TimeInForce::GTX),
            "PostOnly"
        );
    }

    #[test]
    fn test_category_conversion() {
        assert_eq!(BybitCategory::Spot.as_str(), "spot");
        assert_eq!(BybitCategory::Linear.as_str(), "linear");
        assert_eq!(BybitCategory::Inverse.as_str(), "inverse");
        assert_eq!(BybitCategory::Options.as_str(), "option");
    }

    #[tokio::test]
    async fn test_request_tracking() {
        // Create a mock auth
        let api_key = "test_api_key".into();
        let api_secret = "test_api_secret".into();
        let auth = Arc::new(BybitAuth::new(api_key, api_secret));

        // Create trader instance (mainnet)
        let registry = create_shared_registry();
        let trader = BybitWebSocketTrader::new(auth, false, registry);

        // Test tracking a request
        let req_id: SmartString = "test_request_123".into();
        let instrument_id = InstrumentId::new("BTCUSDT", Venue::Bybit);
        let category = BybitCategory::Spot;

        trader
            .track_request(
                req_id.clone(),
                instrument_id.clone(),
                category.clone(),
                RequestType::CreateOrder,
            )
            .await;

        // Verify request was tracked
        let pending = trader.pending_requests.read().await;
        assert!(pending.contains_key(&req_id));

        let request = pending.get(&req_id).unwrap();
        assert_eq!(request.instrument_id, instrument_id);
        assert_eq!(request.category, category);
        assert_eq!(request.request_type, RequestType::CreateOrder);

        // Test finding order category
        drop(pending);
        let found_category = trader.find_order_category("some_order_id").await;
        assert_eq!(found_category, Some(BybitCategory::Spot));
    }

    #[tokio::test]
    async fn test_request_cleanup() {
        // Create a mock auth
        let api_key = "test_api_key".into();
        let api_secret = "test_api_secret".into();
        let auth = Arc::new(BybitAuth::new(api_key, api_secret));

        // Create trader instance (mainnet)
        let registry = create_shared_registry();
        let trader = BybitWebSocketTrader::new(auth, false, registry);

        // Add a request
        let req_id: SmartString = "test_request_456".into();
        let instrument_id = InstrumentId::new("ETHUSDT", Venue::Bybit);

        trader
            .track_request(
                req_id.clone(),
                instrument_id,
                BybitCategory::Linear,
                RequestType::CancelOrder,
            )
            .await;

        // Verify request exists
        {
            let pending = trader.pending_requests.read().await;
            assert_eq!(pending.len(), 1);
        }

        // Run cleanup (should not remove recent requests)
        trader.cleanup_old_requests().await;

        // Verify request still exists
        {
            let pending = trader.pending_requests.read().await;
            assert_eq!(pending.len(), 1);
        }
    }

    #[test]
    fn test_batch_order_request() {
        let orders = vec![
            BybitOrderRequest {
                symbol: "BTCUSDT".into(),
                side: OrderSide::Buy,
                order_type: OrderType::Limit,
                quantity: Decimal::from(1),
                price: Some(Decimal::from(50000)),
                time_in_force: TimeInForce::GTC,
                category: BybitCategory::Spot,
                order_link_id: Some("test_1".into()),
                reduce_only: None,
                close_on_trigger: None,
                position_idx: None,
                take_profit: None,
                stop_loss: None,
                tp_trigger_by: None,
                sl_trigger_by: None,
                tp_limit_price: None,
                sl_limit_price: None,
            },
            BybitOrderRequest {
                symbol: "ETHUSDT".into(),
                side: OrderSide::Sell,
                order_type: OrderType::Market,
                quantity: Decimal::from(2),
                price: None,
                time_in_force: TimeInForce::IOC,
                category: BybitCategory::Spot,
                order_link_id: Some("test_2".into()),
                reduce_only: Some(false),
                close_on_trigger: Some(false),
                position_idx: Some(0),
                take_profit: None,
                stop_loss: None,
                tp_trigger_by: None,
                sl_trigger_by: None,
                tp_limit_price: None,
                sl_limit_price: None,
            },
        ];

        let batch = BybitBatchOrderRequest {
            category: BybitCategory::Spot,
            orders,
        };

        assert_eq!(batch.orders.len(), 2);
        assert_eq!(batch.category, BybitCategory::Spot);
    }
}