nautilus-bybit 0.63.0

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! WebSocket message handler for Bybit.

use std::{
    collections::VecDeque,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicU64, Ordering},
    },
};

use dashmap::DashMap;
use nautilus_network::{
    error::SendError,
    retry::{RetryManager, create_websocket_retry_manager},
    websocket::{AuthTracker, SubscriptionState, WebSocketClient},
};
use serde::Serialize;
use serde_json::Value;
use tokio_tungstenite::tungstenite::Message;
use ustr::Ustr;

use super::{
    enums::BybitWsOperation,
    error::{BybitWsError, create_bybit_timeout_error, should_retry_bybit_error},
    messages::{
        BybitWebSocketError, BybitWsFrame, BybitWsMessage, BybitWsOrderResponse, BybitWsResponse,
        BybitWsSubscriptionMsg,
    },
    parse::parse_bybit_ws_frame,
};
use crate::common::{
    enums::{BybitProductType, BybitWsOrderRequestOp},
    rate_limit::{
        BYBIT_RATE_LIMIT_HEADER, BYBIT_RATE_LIMIT_RESET_HEADER, BYBIT_RATE_LIMIT_STATUS_HEADER,
        BybitRateLimiter,
    },
};

/// Semantic order command whose time-sensitive header is built at send time.
#[derive(Debug)]
pub struct BybitWsOrderCommand {
    pub(crate) req_id: String,
    pub(crate) op: BybitWsOrderRequestOp,
    pub(crate) category: BybitProductType,
    pub(crate) weight: u32,
    pub(crate) referer: Option<String>,
    pub(crate) args: Vec<Value>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct BybitWsSignedOrderRequest {
    req_id: String,
    op: BybitWsOrderRequestOp,
    header: BybitWsSignedOrderHeader,
    args: Vec<Value>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "SCREAMING-KEBAB-CASE")]
struct BybitWsSignedOrderHeader {
    x_bapi_timestamp: String,
    x_bapi_recv_window: String,
    #[serde(rename = "Referer", skip_serializing_if = "Option::is_none")]
    referer: Option<String>,
}

#[derive(Clone, Copy, Debug)]
struct PendingRate {
    endpoint: &'static str,
    category: BybitProductType,
}

#[derive(Debug)]
enum OrderSendFailure {
    NotSent(BybitWsError),
    Ambiguous(BybitWsError),
}

/// Commands sent from the outer client to the inner message handler.
#[derive(Debug)]
pub enum HandlerCommand {
    SetClient(WebSocketClient),
    Disconnect,
    Authenticate { payload: String },
    Subscribe { topics: Vec<String> },
    Unsubscribe { topics: Vec<String> },
    SendOrder { command: BybitWsOrderCommand },
    SendOrders { commands: Vec<BybitWsOrderCommand> },
}

pub(super) struct BybitWsFeedHandler {
    signal: Arc<AtomicBool>,
    inner: Option<WebSocketClient>,
    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
    auth_tracker: AuthTracker,
    subscriptions: SubscriptionState,
    rate_limiter: BybitRateLimiter,
    recv_window_ms: Arc<AtomicU64>,
    pending_rates: DashMap<String, PendingRate>,
    pending_orders: VecDeque<BybitWsOrderCommand>,
    retry_manager: RetryManager<BybitWsError>,
}

impl BybitWsFeedHandler {
    /// Creates a new [`BybitWsFeedHandler`] instance.
    pub(super) fn new(
        signal: Arc<AtomicBool>,
        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
        auth_tracker: AuthTracker,
        subscriptions: SubscriptionState,
        rate_limiter: BybitRateLimiter,
        recv_window_ms: Arc<AtomicU64>,
    ) -> Self {
        Self {
            signal,
            inner: None,
            cmd_rx,
            raw_rx,
            auth_tracker,
            subscriptions,
            rate_limiter,
            recv_window_ms,
            pending_rates: DashMap::new(),
            pending_orders: VecDeque::new(),
            retry_manager: create_websocket_retry_manager(),
        }
    }

    pub(super) fn is_stopped(&self) -> bool {
        self.signal.load(Ordering::Relaxed)
    }

    /// Sends a WebSocket message with retry logic.
    async fn send_with_retry(&self, payload: String) -> Result<(), BybitWsError> {
        if let Some(client) = &self.inner {
            self.retry_manager
                .execute_with_retry(
                    "websocket_send",
                    || {
                        let payload = payload.clone();
                        async move {
                            client
                                .send_text(payload, None)
                                .await
                                .map_err(|e| BybitWsError::Transport(format!("Send failed: {e}")))
                        }
                    },
                    should_retry_bybit_error,
                    |e| create_bybit_timeout_error(e.to_string()),
                )
                .await
        } else {
            Err(BybitWsError::ClientError(
                "No active WebSocket client".to_string(),
            ))
        }
    }

    async fn send_order(&self, command: &BybitWsOrderCommand) -> Result<bool, OrderSendFailure> {
        if !self.auth_tracker.is_authenticated() {
            return Ok(false);
        }

        let client = self.inner.as_ref().ok_or_else(|| {
            OrderSendFailure::NotSent(BybitWsError::ClientError(
                "No active WebSocket client".to_string(),
            ))
        })?;
        let (endpoint, _) = order_operation(command.op);

        self.rate_limiter
            .acquire_ws_order(endpoint, command.category, command.weight)
            .await
            .map_err(|e| OrderSendFailure::NotSent(BybitWsError::ClientError(e)))?;

        if !self.auth_tracker.is_authenticated() {
            return Ok(false);
        }

        let connection_epoch = client.connection_epoch();
        let recv_window_ms = self.recv_window_ms.load(Ordering::Acquire);
        let request = BybitWsSignedOrderRequest {
            req_id: command.req_id.clone(),
            op: command.op,
            header: BybitWsSignedOrderHeader {
                x_bapi_timestamp: nautilus_core::time::get_atomic_clock_realtime()
                    .get_time_ms()
                    .to_string(),
                x_bapi_recv_window: recv_window_ms.to_string(),
                referer: command.referer.clone(),
            },
            args: command.args.clone(),
        };
        let payload = serde_json::to_string(&request)
            .map_err(|e| OrderSendFailure::NotSent(BybitWsError::Json(e.to_string())))?;
        self.pending_rates.insert(
            command.req_id.clone(),
            PendingRate {
                endpoint,
                category: command.category,
            },
        );

        match client
            .send_text_on_connection(payload, None, connection_epoch)
            .await
        {
            Ok(()) => Ok(true),
            Err(e) => self.classify_order_send_error(&command.req_id, e),
        }
    }

    fn classify_order_send_error(
        &self,
        req_id: &str,
        error: SendError,
    ) -> Result<bool, OrderSendFailure> {
        match error {
            SendError::ConnectionChanged => {
                self.pending_rates.remove(req_id);
                self.auth_tracker.invalidate();
                Ok(false)
            }
            SendError::Timeout => {
                self.pending_rates.remove(req_id);
                self.auth_tracker.invalidate();
                Err(OrderSendFailure::NotSent(BybitWsError::ClientError(
                    "Order command was not written".to_string(),
                )))
            }
            SendError::InvalidInput(_) | SendError::Closed => {
                self.pending_rates.remove(req_id);
                Err(OrderSendFailure::NotSent(BybitWsError::ClientError(
                    "Order command was not written".to_string(),
                )))
            }
            error => Err(OrderSendFailure::Ambiguous(BybitWsError::Send(
                error.to_string(),
            ))),
        }
    }

    pub(super) async fn next(&mut self) -> Option<BybitWsMessage> {
        loop {
            if self.auth_tracker.is_authenticated()
                && let Some(command) = self.pending_orders.pop_front()
            {
                let req_id = command.req_id.clone();
                match self.send_order(&command).await {
                    Ok(true) => {}
                    Ok(false) => self.pending_orders.push_front(command),
                    Err(OrderSendFailure::NotSent(error)) => {
                        return Some(order_not_sent_message(&command, &error));
                    }
                    Err(OrderSendFailure::Ambiguous(error)) => {
                        log::error!("Ambiguous order send failure: req_id={req_id}, error={error}");
                    }
                }
                continue;
            }

            tokio::select! {
                Some(cmd) = self.cmd_rx.recv() => {
                    match cmd {
                        HandlerCommand::SetClient(client) => {
                            log::debug!("WebSocketClient received by handler");
                            self.inner = Some(client);
                        }
                        HandlerCommand::Disconnect => {
                            log::debug!("Disconnect command received");

                            if let Some(client) = self.inner.take() {
                                client.disconnect().await;
                            }
                        }
                        HandlerCommand::Authenticate { payload } => {
                            log::debug!("Authenticate command received");

                            if let Err(e) = self.send_with_retry(payload).await {
                                log::error!("Failed to send authentication after retries: {e}");
                            }
                        }
                        HandlerCommand::Subscribe { topics } => {
                            for topic in topics {
                                log::debug!("Subscribing to topic: topic={topic}");
                                if let Err(e) = self.send_with_retry(topic.clone()).await {
                                    log::error!("Failed to send subscription after retries: topic={topic}, error={e}");
                                }
                            }
                        }
                        HandlerCommand::Unsubscribe { topics } => {
                            for topic in topics {
                                log::debug!("Unsubscribing from topic: topic={topic}");
                                if let Err(e) = self.send_with_retry(topic.clone()).await {
                                    log::error!("Failed to send unsubscription after retries: topic={topic}, error={e}");
                                }
                            }
                        }
                        HandlerCommand::SendOrder { command } => {
                            let req_id = command.req_id.clone();
                            match self.send_order(&command).await {
                                Ok(true) => {}
                                Ok(false) => self.pending_orders.push_back(command),
                                Err(OrderSendFailure::NotSent(error)) => {
                                    return Some(order_not_sent_message(&command, &error));
                                }
                                Err(OrderSendFailure::Ambiguous(error)) => {
                                    log::error!(
                                        "Ambiguous order send failure: req_id={req_id}, error={error}"
                                    );
                                }
                            }
                        }
                        HandlerCommand::SendOrders { commands } => {
                            self.pending_orders.extend(commands);
                        }
                    }
                }

                () = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
                    if self.signal.load(Ordering::Relaxed) {
                        log::debug!("Stop signal received during idle period");
                        return None;
                    }
                }

                msg = self.raw_rx.recv() => {
                    let msg = match msg {
                        Some(msg) => msg,
                        None => {
                            log::debug!("WebSocket stream closed");
                            return None;
                        }
                    };

                    if let Message::Ping(data) = &msg {
                        log::trace!("Received ping frame with {} bytes", data.len());

                        if let Some(client) = &self.inner
                            && let Err(e) = client.send_pong(data.to_vec()).await
                        {
                            log::warn!("Failed to send pong frame: error={e}");
                        }
                        continue;
                    }

                    let frame = match Self::parse_raw_frame(msg) {
                        Some(frame) => frame,
                        None => continue,
                    };

                    if self.signal.load(Ordering::Relaxed) {
                        log::debug!("Stop signal received");
                        return None;
                    }

                    match frame {
                        BybitWsFrame::Subscription(ref sub_msg) => {
                            self.handle_subscription_ack(sub_msg);
                        }
                        BybitWsFrame::Auth(auth_response) => {
                            let is_success = auth_response.success.unwrap_or(false)
                                || (auth_response.ret_code == Some(0));

                            if is_success {
                                self.auth_tracker.succeed();
                                log::debug!("WebSocket authenticated");
                            } else {
                                let error_msg = auth_response
                                    .ret_msg
                                    .as_deref()
                                    .unwrap_or("Authentication rejected");
                                self.auth_tracker.fail(error_msg);
                                log::error!("WebSocket authentication failed: error={error_msg}");
                            }
                            return Some(BybitWsMessage::Auth(auth_response));
                        }
                        BybitWsFrame::ErrorResponse(ref resp) => {
                            // Failed subscription/unsubscription ACKs arrive as
                            // ErrorResponse when success=false. Route them through
                            // the subscription state machine when the op field is
                            // present, otherwise forward as a generic error.
                            if let Some(op) = &resp.op {
                                if *op == BybitWsOperation::Subscribe
                                    || *op == BybitWsOperation::Unsubscribe
                                {
                                    self.handle_subscription_error(resp);
                                } else {
                                    let error = BybitWebSocketError::from_response(resp);
                                    return Some(BybitWsMessage::Error(error));
                                }
                            } else {
                                let error = BybitWebSocketError::from_response(resp);
                                return Some(BybitWsMessage::Error(error));
                            }
                        }
                        BybitWsFrame::OrderResponse(resp) => {
                            self.observe_order_rate_limit(&resp);
                            return Some(BybitWsMessage::OrderResponse(resp));
                        }
                        BybitWsFrame::Orderbook(msg) => {
                            return Some(BybitWsMessage::Orderbook(msg));
                        }
                        BybitWsFrame::Trade(msg) => {
                            return Some(BybitWsMessage::Trade(msg));
                        }
                        BybitWsFrame::Kline(msg) => {
                            return Some(BybitWsMessage::Kline(msg));
                        }
                        BybitWsFrame::TickerLinear(msg) => {
                            return Some(BybitWsMessage::TickerLinear(msg));
                        }
                        BybitWsFrame::TickerOption(msg) => {
                            return Some(BybitWsMessage::TickerOption(msg));
                        }
                        BybitWsFrame::AccountOrder(msg) => {
                            return Some(BybitWsMessage::AccountOrder(msg));
                        }
                        BybitWsFrame::AccountExecution(msg) => {
                            return Some(BybitWsMessage::AccountExecution(msg));
                        }
                        BybitWsFrame::AccountExecutionFast(msg) => {
                            return Some(BybitWsMessage::AccountExecutionFast(msg));
                        }
                        BybitWsFrame::AccountWallet(msg) => {
                            return Some(BybitWsMessage::AccountWallet(msg));
                        }
                        BybitWsFrame::AccountPosition(msg) => {
                            return Some(BybitWsMessage::AccountPosition(msg));
                        }
                        BybitWsFrame::Reconnected => {
                            self.auth_tracker.invalidate();
                            return Some(BybitWsMessage::Reconnected);
                        }
                        BybitWsFrame::Unknown(value) => {
                            log::debug!("Unknown WebSocket frame: {value}");
                        }
                    }
                }
            }
        }
    }

    fn observe_order_rate_limit(&self, response: &super::messages::BybitWsOrderResponse) {
        let Some(req_id) = response.req_id.as_deref() else {
            return;
        };
        let Some((_, pending)) = self.pending_rates.remove(req_id) else {
            return;
        };
        let Some(header) = response.header.as_ref() else {
            return;
        };
        let parse_u32 = |key: &str| {
            header.get(key).and_then(|value| {
                value
                    .as_u64()
                    .and_then(|value| u32::try_from(value).ok())
                    .or_else(|| value.as_str()?.parse::<u32>().ok())
            })
        };
        let Some(limit) = parse_u32(BYBIT_RATE_LIMIT_HEADER) else {
            return;
        };
        let Some(remaining) = parse_u32(BYBIT_RATE_LIMIT_STATUS_HEADER) else {
            return;
        };
        let reset_timestamp_ms = header.get(BYBIT_RATE_LIMIT_RESET_HEADER).and_then(|value| {
            value
                .as_i64()
                .or_else(|| value.as_str()?.parse::<i64>().ok())
        });
        self.rate_limiter.observe_account(
            pending.endpoint,
            Some(pending.category),
            limit,
            remaining,
            reset_timestamp_ms,
        );
    }

    fn handle_subscription_ack(&self, sub_msg: &BybitWsSubscriptionMsg) {
        match sub_msg.op {
            BybitWsOperation::Subscribe => {
                if sub_msg.success {
                    if let Some(topic) = &sub_msg.req_id {
                        self.subscriptions.confirm_subscribe(topic);
                        log::debug!("Subscription confirmed: topic={topic}");
                    } else {
                        // No req_id, fall back to confirming all pending
                        for topic in self.subscriptions.pending_subscribe_topics() {
                            self.subscriptions.confirm_subscribe(&topic);
                            log::debug!("Subscription confirmed (bulk): topic={topic}");
                        }
                    }
                } else if let Some(topic) = &sub_msg.req_id {
                    self.subscriptions.mark_failure(topic);
                    log::warn!(
                        "Subscription failed: topic={topic}, error={:?}",
                        sub_msg.ret_msg
                    );
                } else {
                    for topic in self.subscriptions.pending_subscribe_topics() {
                        self.subscriptions.mark_failure(&topic);
                        log::warn!(
                            "Subscription failed (bulk): topic={topic}, error={:?}",
                            sub_msg.ret_msg
                        );
                    }
                }
            }
            BybitWsOperation::Unsubscribe => {
                if sub_msg.success {
                    if let Some(topic) = &sub_msg.req_id {
                        self.subscriptions.confirm_unsubscribe(topic);
                        log::debug!("Unsubscription confirmed: topic={topic}");
                    } else {
                        for topic in self.subscriptions.pending_unsubscribe_topics() {
                            self.subscriptions.confirm_unsubscribe(&topic);
                            log::debug!("Unsubscription confirmed (bulk): topic={topic}");
                        }
                    }
                } else {
                    let topic_desc = sub_msg.req_id.as_deref().unwrap_or("unknown");
                    log::warn!(
                        "Unsubscription failed: topic={topic_desc}, error={:?}",
                        sub_msg.ret_msg
                    );
                }
            }
            _ => {}
        }
    }

    fn handle_subscription_error(&self, resp: &BybitWsResponse) {
        let topic = resp.req_id.as_deref().unwrap_or("unknown");
        let error_msg = resp.ret_msg.as_deref().unwrap_or("unknown error");

        match resp.op {
            Some(BybitWsOperation::Subscribe) => {
                // Duplicate subscribe is harmless: the topic is active on the
                // venue, so confirm it instead of looping retries every reconnect.
                if is_already_subscribed_error(error_msg)
                    && let Some(ref req_id) = resp.req_id
                {
                    self.subscriptions.confirm_subscribe(req_id);
                    log::debug!("Subscription duplicate ignored: topic={topic}, error={error_msg}");
                    return;
                }

                if let Some(ref req_id) = resp.req_id {
                    self.subscriptions.mark_failure(req_id);
                } else {
                    for t in self.subscriptions.pending_subscribe_topics() {
                        self.subscriptions.mark_failure(&t);
                    }
                }
                log::warn!("Subscription error: topic={topic}, error={error_msg}");
            }
            Some(BybitWsOperation::Unsubscribe) => {
                log::warn!("Unsubscription error: topic={topic}, error={error_msg}");
            }
            _ => {}
        }
    }

    fn parse_raw_frame(msg: Message) -> Option<BybitWsFrame> {
        match msg {
            Message::Text(text) => {
                if text == nautilus_network::RECONNECTED {
                    log::debug!("Received WebSocket reconnected signal");
                    return Some(BybitWsFrame::Reconnected);
                }

                if text.trim().eq_ignore_ascii_case("pong") {
                    return None;
                }

                log::trace!("Raw websocket message: {text}");

                let value: serde_json::Value = match serde_json::from_str(&text) {
                    Ok(v) => v,
                    Err(e) => {
                        log::error!("Failed to parse WebSocket message: {e}: {text}");
                        return None;
                    }
                };

                if value
                    .get("op")
                    .and_then(serde_json::Value::as_str)
                    .is_some_and(|op| op == BybitWsOperation::Pong.as_ref())
                {
                    return None;
                }

                Some(parse_bybit_ws_frame(value))
            }
            Message::Binary(msg) => {
                log::debug!("Raw binary frame ({} bytes)", msg.len());
                log::trace!("Raw binary: {msg:?}");
                None
            }
            Message::Close(_) => {
                log::debug!("Received close message, waiting for reconnection");
                None
            }
            _ => None,
        }
    }
}

fn order_operation(op: BybitWsOrderRequestOp) -> (&'static str, &'static str) {
    match op {
        BybitWsOrderRequestOp::Create => ("/v5/order/create", "order.create"),
        BybitWsOrderRequestOp::Amend => ("/v5/order/amend", "order.amend"),
        BybitWsOrderRequestOp::Cancel => ("/v5/order/cancel", "order.cancel"),
        BybitWsOrderRequestOp::CreateBatch => ("/v5/order/create-batch", "order.create-batch"),
        BybitWsOrderRequestOp::AmendBatch => ("/v5/order/amend-batch", "order.amend-batch"),
        BybitWsOrderRequestOp::CancelBatch => ("/v5/order/cancel-batch", "order.cancel-batch"),
    }
}

fn order_not_sent_message(command: &BybitWsOrderCommand, error: &BybitWsError) -> BybitWsMessage {
    let (_, op) = order_operation(command.op);
    BybitWsMessage::OrderResponse(BybitWsOrderResponse {
        op: Ustr::from(op),
        conn_id: None,
        ret_code: -1,
        ret_msg: error.to_string(),
        data: Value::Object(serde_json::Map::new()),
        req_id: Some(command.req_id.clone()),
        header: None,
        ret_ext_info: None,
    })
}

fn is_already_subscribed_error(error_msg: &str) -> bool {
    error_msg
        .to_ascii_lowercase()
        .contains("already subscribed")
}

#[cfg(test)]
mod tests {
    use rstest::rstest;
    use ustr::Ustr;

    use super::*;
    use crate::common::{
        consts::BYBIT_WS_TOPIC_DELIMITER, rate_limit::BybitRateLimiter, testing::load_test_json,
    };

    fn create_test_handler() -> BybitWsFeedHandler {
        let signal = Arc::new(AtomicBool::new(false));
        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
        let auth_tracker = AuthTracker::new();
        let subscriptions = SubscriptionState::new(BYBIT_WS_TOPIC_DELIMITER);

        BybitWsFeedHandler::new(
            signal,
            cmd_rx,
            raw_rx,
            auth_tracker,
            subscriptions,
            BybitRateLimiter::for_websocket(
                "wss://bybit-handler-test.invalid",
                Some("test-key"),
                None,
            ),
            Arc::new(AtomicU64::new(5_000)),
        )
    }

    fn load_value(fixture: &str) -> serde_json::Value {
        let json = load_test_json(fixture);
        serde_json::from_str(&json).unwrap()
    }

    #[rstest]
    fn test_handler_initializes() {
        let _handler = create_test_handler();
    }

    #[tokio::test]
    async fn order_without_writer_produces_correlated_terminal_response() {
        let handler = create_test_handler();
        handler.auth_tracker.succeed();
        let command = BybitWsOrderCommand {
            req_id: "not-sent-request".to_string(),
            op: BybitWsOrderRequestOp::Create,
            category: BybitProductType::Linear,
            weight: 1,
            referer: None,
            args: vec![serde_json::json!({"category": "linear"})],
        };

        let failure = handler.send_order(&command).await.unwrap_err();
        let OrderSendFailure::NotSent(error) = failure else {
            panic!("Expected definitive not-sent failure, was {failure:?}");
        };
        let BybitWsMessage::OrderResponse(response) = order_not_sent_message(&command, &error)
        else {
            panic!("Expected order response");
        };

        assert_eq!(response.op.as_str(), "order.create");
        assert_eq!(response.ret_code, -1);
        assert_eq!(response.req_id.as_deref(), Some("not-sent-request"));
        assert_eq!(response.ret_msg, "Client error: No active WebSocket client");
    }

    #[rstest]
    fn connection_change_invalidates_authentication_before_requeue() {
        let handler = create_test_handler();
        handler.auth_tracker.succeed();
        handler.pending_rates.insert(
            "reconnect-request".to_string(),
            PendingRate {
                endpoint: "/v5/order/create",
                category: BybitProductType::Linear,
            },
        );

        let result = handler
            .classify_order_send_error("reconnect-request", SendError::ConnectionChanged)
            .unwrap();

        assert!(!result);
        assert!(!handler.auth_tracker.is_authenticated());
        assert!(!handler.pending_rates.contains_key("reconnect-request"));
    }

    #[rstest]
    fn pre_active_timeout_is_terminal_not_sent() {
        let handler = create_test_handler();
        handler.auth_tracker.succeed();
        handler.pending_rates.insert(
            "timeout-request".to_string(),
            PendingRate {
                endpoint: "/v5/order/create",
                category: BybitProductType::Linear,
            },
        );

        let failure = handler
            .classify_order_send_error("timeout-request", SendError::Timeout)
            .unwrap_err();

        assert!(matches!(failure, OrderSendFailure::NotSent(_)));
        assert!(!handler.auth_tracker.is_authenticated());
        assert!(!handler.pending_rates.contains_key("timeout-request"));
    }

    #[rstest]
    fn test_parse_frame_auth_success() {
        let value = load_value("ws_auth_success.json");
        let frame = parse_bybit_ws_frame(value);
        match frame {
            BybitWsFrame::Auth(auth) => {
                assert_eq!(auth.conn_id.as_deref(), Some("cejreaspqfm9se7usbrg-2xh"));
                assert_eq!(auth.ret_code, Some(0));
                assert_eq!(auth.success, Some(true));
            }
            other => panic!("Expected Auth, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_frame_auth_failure() {
        let value = load_value("ws_auth_failure.json");
        let frame = parse_bybit_ws_frame(value);
        match frame {
            BybitWsFrame::ErrorResponse(resp) => {
                assert_eq!(resp.ret_code, Some(10003));
                assert_eq!(resp.ret_msg.as_deref(), Some("Invalid apikey"));
            }
            other => panic!("Expected ErrorResponse, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_frame_subscription_ack() {
        let value = load_value("ws_subscription_ack.json");
        let frame = parse_bybit_ws_frame(value);
        match frame {
            BybitWsFrame::Subscription(sub) => {
                assert!(sub.success);
                assert_eq!(sub.op, BybitWsOperation::Subscribe);
                assert_eq!(sub.req_id.as_deref(), Some("sub-orderbook-1"));
            }
            other => panic!("Expected Subscription, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_frame_subscription_failure() {
        let value = load_value("ws_subscription_failure.json");
        let frame = parse_bybit_ws_frame(value);
        match frame {
            BybitWsFrame::ErrorResponse(resp) => {
                assert_eq!(
                    resp.ret_msg.as_deref(),
                    Some("Invalid topic: invalid.topic.BTCUSDT")
                );
            }
            other => panic!("Expected ErrorResponse, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_frame_order_response() {
        let value = load_value("ws_order_response.json");
        let frame = parse_bybit_ws_frame(value);
        match frame {
            BybitWsFrame::OrderResponse(resp) => {
                assert_eq!(resp.op.as_str(), "order.create");
                assert_eq!(resp.ret_code, 0);
                assert_eq!(resp.ret_msg, "OK");
            }
            other => panic!("Expected OrderResponse, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_frame_orderbook() {
        let value = load_value("ws_orderbook_snapshot.json");
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::Orderbook(_)),
            "Expected Orderbook, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_frame_trade() {
        let value = load_value("ws_public_trade.json");
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::Trade(_)),
            "Expected Trade, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_frame_kline() {
        let value = load_value("ws_kline.json");
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::Kline(_)),
            "Expected Kline, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_frame_ticker_linear() {
        let value = load_value("ws_ticker_linear.json");
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::TickerLinear(_)),
            "Expected TickerLinear, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_frame_ticker_option() {
        let value = load_value("ws_ticker_option.json");
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::TickerOption(_)),
            "Expected TickerOption, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_frame_account_order() {
        let value = load_value("ws_account_order.json");
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::AccountOrder(_)),
            "Expected AccountOrder, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_frame_account_execution() {
        let value = load_value("ws_account_execution.json");
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::AccountExecution(_)),
            "Expected AccountExecution, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_frame_account_wallet() {
        let value = load_value("ws_account_wallet.json");
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::AccountWallet(_)),
            "Expected AccountWallet, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_frame_account_position() {
        let value = load_value("ws_account_position.json");
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::AccountPosition(_)),
            "Expected AccountPosition, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_frame_unknown_message() {
        let value: serde_json::Value = serde_json::json!({"foo": "bar"});
        let frame = parse_bybit_ws_frame(value);
        assert!(
            matches!(frame, BybitWsFrame::Unknown(_)),
            "Expected Unknown, was {frame:?}"
        );
    }

    #[rstest]
    fn test_parse_raw_reconnected_signal() {
        let msg = Message::Text(nautilus_network::RECONNECTED.to_string().into());
        let result = BybitWsFeedHandler::parse_raw_frame(msg);
        assert!(
            matches!(result, Some(BybitWsFrame::Reconnected)),
            "Expected Some(Reconnected), was {result:?}"
        );
    }

    #[rstest]
    fn test_parse_raw_pong_text() {
        let msg = Message::Text("pong".into());
        let result = BybitWsFeedHandler::parse_raw_frame(msg);
        assert!(result.is_none(), "Expected None for pong, was {result:?}");
    }

    #[rstest]
    fn test_parse_raw_json_pong_message() {
        let msg = Message::Text(
            r#"{"args":["1777226678908"],"conn_id":"yzr7jz02gws1vh60mk5m-hxqdp","op":"pong"}"#
                .into(),
        );
        let result = BybitWsFeedHandler::parse_raw_frame(msg);
        assert!(
            result.is_none(),
            "Expected None for JSON pong, was {result:?}"
        );
    }

    #[rstest]
    fn test_parse_raw_valid_json() {
        let json = load_test_json("ws_public_trade.json");
        let msg = Message::Text(json.into());
        let result = BybitWsFeedHandler::parse_raw_frame(msg);
        assert!(
            matches!(result, Some(BybitWsFrame::Trade(_))),
            "Expected Some(Trade), was {result:?}"
        );
    }

    #[rstest]
    fn test_parse_raw_invalid_json() {
        let msg = Message::Text("not valid json".into());
        let result = BybitWsFeedHandler::parse_raw_frame(msg);
        assert!(
            result.is_none(),
            "Expected None for invalid JSON, was {result:?}"
        );
    }

    #[rstest]
    fn test_parse_raw_binary_message() {
        let msg = Message::Binary(vec![0x01, 0x02].into());
        let result = BybitWsFeedHandler::parse_raw_frame(msg);
        assert!(result.is_none(), "Expected None for binary, was {result:?}");
    }

    #[rstest]
    fn test_subscription_ack_with_req_id_confirms_only_that_topic() {
        let handler = create_test_handler();
        handler.subscriptions.mark_subscribe("orderbook.50.BTCUSDT");
        handler.subscriptions.mark_subscribe("publicTrade.BTCUSDT");

        let ack = BybitWsSubscriptionMsg {
            success: true,
            op: BybitWsOperation::Subscribe,
            conn_id: None,
            req_id: Some("orderbook.50.BTCUSDT".to_string()),
            ret_msg: None,
        };

        handler.handle_subscription_ack(&ack);

        // Only orderbook should be confirmed, trade stays pending
        assert!(
            handler
                .subscriptions
                .pending_subscribe_topics()
                .contains(&"publicTrade.BTCUSDT".to_string())
        );
        assert!(
            !handler
                .subscriptions
                .pending_subscribe_topics()
                .contains(&"orderbook.50.BTCUSDT".to_string())
        );
    }

    #[rstest]
    fn test_subscription_failure_with_req_id_marks_only_that_topic() {
        let handler = create_test_handler();
        handler.subscriptions.mark_subscribe("orderbook.50.BTCUSDT");
        handler.subscriptions.mark_subscribe("publicTrade.BTCUSDT");

        let ack = BybitWsSubscriptionMsg {
            success: false,
            op: BybitWsOperation::Subscribe,
            conn_id: None,
            req_id: Some("orderbook.50.BTCUSDT".to_string()),
            ret_msg: Some("Invalid topic".to_string()),
        };

        handler.handle_subscription_ack(&ack);

        // Orderbook should be marked as failed (back to pending for retry)
        // Trade should remain pending (unaffected)
        let pending = handler.subscriptions.pending_subscribe_topics();
        assert!(pending.contains(&"orderbook.50.BTCUSDT".to_string()));
        assert!(pending.contains(&"publicTrade.BTCUSDT".to_string()));
    }

    #[rstest]
    fn test_error_response_with_subscribe_op_triggers_mark_failure() {
        let handler = create_test_handler();
        handler
            .subscriptions
            .mark_subscribe("invalid.topic.BTCUSDT");

        let resp = BybitWsResponse {
            op: Some(BybitWsOperation::Subscribe),
            topic: None,
            success: Some(false),
            conn_id: None,
            req_id: Some("invalid.topic.BTCUSDT".to_string()),
            ret_code: Some(10001),
            ret_msg: Some("Invalid topic".to_string()),
        };

        handler.handle_subscription_error(&resp);

        // Topic should still be in pending (mark_failure moves confirmed -> pending)
        let pending = handler.subscriptions.pending_subscribe_topics();
        assert!(pending.contains(&"invalid.topic.BTCUSDT".to_string()));
    }

    #[rstest]
    fn test_already_subscribed_error_confirms_topic() {
        let handler = create_test_handler();
        handler.subscriptions.mark_subscribe("tickers.ETHUSDT");

        let resp = BybitWsResponse {
            op: Some(BybitWsOperation::Subscribe),
            topic: None,
            success: Some(false),
            conn_id: None,
            req_id: Some("tickers.ETHUSDT".to_string()),
            ret_code: Some(10001),
            ret_msg: Some("error:already subscribed,topic:tickers.ETHUSDT".to_string()),
        };

        handler.handle_subscription_error(&resp);

        let pending = handler.subscriptions.pending_subscribe_topics();
        assert!(!pending.contains(&"tickers.ETHUSDT".to_string()));
        let symbols = handler.subscriptions.confirmed();
        let entry = symbols
            .get(&Ustr::from("tickers"))
            .expect("channel present");
        assert!(entry.contains(&Ustr::from("ETHUSDT")));
    }

    #[rstest]
    fn test_subscription_ack_without_req_id_confirms_all_pending() {
        let handler = create_test_handler();
        handler.subscriptions.mark_subscribe("orderbook.50.BTCUSDT");
        handler.subscriptions.mark_subscribe("publicTrade.BTCUSDT");

        let ack = BybitWsSubscriptionMsg {
            success: true,
            op: BybitWsOperation::Subscribe,
            conn_id: None,
            req_id: None,
            ret_msg: None,
        };

        handler.handle_subscription_ack(&ack);

        // Both should be confirmed when no req_id
        assert!(handler.subscriptions.pending_subscribe_topics().is_empty());
    }
}