nautilus_bybit/

data.rs

// -------------------------------------------------------------------------------------------------
//  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.
// -------------------------------------------------------------------------------------------------

//! Live market data client implementation for the Bybit adapter.

use std::{
    future::Future,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use ahash::{AHashMap, AHashSet};
use anyhow::Context;
use futures_util::{StreamExt, pin_mut};
use nautilus_common::{
    cache::quote::QuoteCache,
    clients::DataClient,
    live::{runner::get_data_event_sender, sender::EventSender},
    messages::{
        DataEvent,
        data::{
            BarsResponse, BookResponse, DataResponse, FundingRatesResponse, InstrumentResponse,
            InstrumentsResponse, OptionChainReferencePriceResponse, RequestBars,
            RequestBookSnapshot, RequestFundingRates, RequestInstrument, RequestInstruments,
            RequestOptionChainReferencePrice, RequestTrades, SubscribeBars, SubscribeBookDeltas,
            SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument,
            SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices,
            SubscribeOptionGreeks, SubscribeQuotes, SubscribeTrades, TradesResponse,
            UnsubscribeBars, UnsubscribeBookDeltas, UnsubscribeFundingRates,
            UnsubscribeIndexPrices, UnsubscribeInstrument, UnsubscribeInstrumentStatus,
            UnsubscribeInstruments, UnsubscribeMarkPrices, UnsubscribeOptionGreeks,
            UnsubscribeQuotes, UnsubscribeTrades,
        },
    },
};
use nautilus_core::{
    AtomicMap, AtomicSet,
    datetime::datetime_to_unix_nanos,
    time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_live::{
    SocketControlFactory,
    task::{TaskGroup, TaskGroupGuard},
};
use nautilus_model::{
    data::{BarType, Data},
    enums::{BookType, MarketStatusAction},
    identifiers::{ClientId, InstrumentId, Venue},
    instruments::{Instrument, InstrumentAny},
    orderbook::book::OrderBook,
    types::Price,
};
use rust_decimal::Decimal;
use tokio_util::sync::CancellationToken;
use ustr::Ustr;

use crate::{
    common::{
        consts::{
            BYBIT_BOOK_DEPTHS, BYBIT_DEFAULT_ORDERBOOK_DEPTH, BYBIT_QUOTE_DEPTH, BYBIT_VENUE,
        },
        enums::BybitProductType,
        instruments::diff_and_emit_instruments,
        parse::{extract_raw_symbol, make_bybit_symbol},
        status::{diff_and_emit_statuses, emit_status},
    },
    config::BybitDataClientConfig,
    http::client::BybitHttpClient,
    websocket::{
        client::BybitWebSocketClient,
        messages::BybitWsMessage,
        parse::{
            parse_kline_topic, parse_millis_i64, parse_orderbook_deltas, parse_orderbook_quote,
            parse_orderbook_topic, parse_ticker_linear_funding, parse_ticker_linear_index_price,
            parse_ticker_linear_mark_price, parse_ticker_option_greeks,
            parse_ticker_option_index_price, parse_ticker_option_mark_price,
            parse_ticker_option_quote, parse_ws_kline_bar, parse_ws_trade_tick,
        },
    },
};

/// Live market data client for Bybit.
#[derive(Debug)]
pub struct BybitDataClient {
    client_id: ClientId,
    config: BybitDataClientConfig,
    http_client: BybitHttpClient,
    ws_clients: Vec<BybitWebSocketClient>,
    is_connected: AtomicBool,
    cancellation_token: CancellationToken,
    session_tasks: TaskGroup,
    command_tasks: TaskGroup,
    shutdown_errors: Vec<String>,
    data_sender: EventSender<DataEvent>,
    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
    book_depths: Arc<AtomicMap<InstrumentId, u32>>,
    quote_subs: Arc<AtomicSet<InstrumentId>>,
    ticker_subs: Arc<AtomicMap<InstrumentId, AHashSet<&'static str>>>,
    trade_subs: Arc<AtomicSet<InstrumentId>>,
    option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
    instrument_status_subs: Arc<AtomicSet<InstrumentId>>,
    status_cache: Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
    instrument_subs: Arc<AtomicSet<InstrumentId>>,
    subscribe_all_instruments: Arc<AtomicBool>,
    clock: &'static AtomicTime,
}

impl BybitDataClient {
    /// Creates a new [`BybitDataClient`] instance.
    ///
    /// # Errors
    ///
    /// Returns an error if the client fails to initialize.
    pub fn new(client_id: ClientId, config: BybitDataClientConfig) -> anyhow::Result<Self> {
        let clock = get_atomic_clock_realtime();
        let data_sender = get_data_event_sender();
        let socket_factory = SocketControlFactory::new(client_id, Some(*BYBIT_VENUE));
        let proxy_url = config
            .proxy_url
            .as_ref()
            .map(|value| value.expose_secret().to_owned());

        let http_client = if let (Some(api_key), Some(api_secret)) =
            (config.api_key.clone(), config.api_secret.clone())
        {
            BybitHttpClient::with_credentials(
                api_key.into_inner(),
                api_secret.into_inner(),
                Some(config.http_base_url()),
                config.http_timeout_secs,
                config.max_retries,
                config.retry_delay_initial_ms,
                config.retry_delay_max_ms,
                config.recv_window_ms,
                proxy_url.clone(),
            )?
        } else {
            BybitHttpClient::new(
                Some(config.http_base_url()),
                config.http_timeout_secs,
                config.max_retries,
                config.retry_delay_initial_ms,
                config.retry_delay_max_ms,
                config.recv_window_ms,
                proxy_url.clone(),
            )?
        };

        // Create a WebSocket client for each product type (default to Linear if empty)
        let product_types = if config.product_types.is_empty() {
            vec![BybitProductType::Linear]
        } else {
            config.product_types.clone()
        };

        let ws_clients: Vec<BybitWebSocketClient> = product_types
            .iter()
            .map(|product_type| {
                BybitWebSocketClient::new_public_with(
                    *product_type,
                    config.environment,
                    Some(config.ws_public_url_for(*product_type)),
                    config.heartbeat_interval_secs,
                    config.transport_backend,
                    proxy_url.clone(),
                )
                .with_socket_control(
                    socket_factory.control(format!("bybit-{}-data-streams", product_type.as_str())),
                )
            })
            .collect();

        let session_tasks = TaskGroup::new();
        let command_tasks = TaskGroup::new();

        Ok(Self {
            client_id,
            config,
            http_client,
            ws_clients,
            is_connected: AtomicBool::new(false),
            cancellation_token: session_tasks.cancellation_token(),
            session_tasks,
            command_tasks,
            shutdown_errors: Vec::new(),
            data_sender,
            instruments: Arc::new(AtomicMap::new()),
            book_depths: Arc::new(AtomicMap::new()),
            quote_subs: Arc::new(AtomicSet::new()),
            ticker_subs: Arc::new(AtomicMap::new()),
            trade_subs: Arc::new(AtomicSet::new()),
            option_greeks_subs: Arc::new(AtomicSet::new()),
            instrument_status_subs: Arc::new(AtomicSet::new()),
            status_cache: Arc::new(AtomicMap::new()),
            instrument_subs: Arc::new(AtomicSet::new()),
            subscribe_all_instruments: Arc::new(AtomicBool::new(false)),
            clock,
        })
    }

    fn venue(&self) -> Venue {
        *BYBIT_VENUE
    }

    fn get_ws_client_for_product(
        &self,
        product_type: BybitProductType,
    ) -> Option<&BybitWebSocketClient> {
        self.ws_clients
            .iter()
            .find(|ws| ws.product_type() == Some(product_type))
    }

    fn get_product_type_for_instrument(
        &self,
        instrument_id: InstrumentId,
    ) -> Option<BybitProductType> {
        let guard = self.instruments.load();
        guard
            .get(&instrument_id)
            .and_then(|_| BybitProductType::from_suffix(instrument_id.symbol.as_str()))
    }

    fn spawn_command<F>(&self, future: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        if let Err(e) = self.command_tasks.spawn(future) {
            log::warn!("Skipping Bybit data command after shutdown began: {e}");
        }
    }

    async fn finish_tasks(&self) -> anyhow::Result<()> {
        let (session_result, command_result) = tokio::join!(
            self.session_tasks
                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
            self.command_tasks
                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
        );
        let mut errors = Vec::new();
        if let Err(e) = session_result {
            errors.push(format!("failed to finish Bybit data session tasks: {e}"));
        }

        if let Err(e) = command_result {
            errors.push(format!("failed to finish Bybit data command tasks: {e}"));
        }

        if errors.is_empty() {
            Ok(())
        } else {
            anyhow::bail!(errors.join("; "))
        }
    }

    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
        if !self.session_tasks.is_open() || !self.command_tasks.is_open() {
            self.teardown_partial_connect().await?;
            self.session_tasks
                .start_generation()
                .context("failed to start Bybit data session task generation")?;
            self.command_tasks
                .start_generation()
                .context("failed to start Bybit data command task generation")?;
            self.cancellation_token = self.session_tasks.cancellation_token();
        }
        Ok(())
    }

    fn spawn_ws<F>(&self, fut: F, context: &'static str)
    where
        F: Future<Output = anyhow::Result<()>> + Send + 'static,
    {
        let future = async move {
            if let Err(e) = fut.await {
                log::error!("{context}: {e:?}");
            }
        };

        if let Err(e) = self.command_tasks.spawn(future) {
            log::warn!("Skipping Bybit {context} after shutdown began: {e}");
        }
    }

    fn spawn_instrument_polling(
        &self,
        product_types: &[BybitProductType],
        poll_secs: u64,
    ) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instruments = self.instruments.clone();
        let status_cache = self.status_cache.clone();
        let status_subs = self.instrument_status_subs.clone();
        let instrument_subs = self.instrument_subs.clone();
        let subscribe_all_instruments = self.subscribe_all_instruments.clone();
        let cancel = self.cancellation_token.clone();
        let clock = self.clock;
        let product_types = product_types.to_vec();

        let future = async move {
            let mut interval = tokio::time::interval(Duration::from_secs(poll_secs));
            interval.tick().await; // Skip first immediate tick

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        let all_flag = subscribe_all_instruments.load(Ordering::Relaxed);
                        let want_instruments = all_flag || !instrument_subs.is_empty();
                        let want_statuses = !status_subs.is_empty();
                        if !want_instruments && !want_statuses {
                            continue;
                        }

                        let mut all_statuses = AHashMap::new();

                        if want_instruments {
                            let subs: Option<AHashSet<InstrumentId>> = if all_flag {
                                None
                            } else {
                                Some((**instrument_subs.load()).clone())
                            };
                            let mut inst_cache = (**instruments.load()).clone();

                            for &pt in &product_types {
                                match http.request_instruments_with_statuses(pt).await {
                                    Ok((fetched, statuses)) => {
                                        diff_and_emit_instruments(
                                            &fetched, &mut inst_cache, subs.as_ref(), &sender,
                                        );

                                        for (id, action) in statuses {
                                            if inst_cache.contains_key(&id) {
                                                all_statuses.insert(id, action);
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        log::warn!("Bybit instrument poll failed for {pt:?}: {e}");
                                    }
                                }
                            }

                            instruments.store(inst_cache);
                        } else {
                            for &pt in &product_types {
                                match http.request_instrument_statuses(pt).await {
                                    Ok(new_statuses) => {
                                        let inst_guard = instruments.load();
                                        for (id, action) in new_statuses {
                                            if inst_guard.contains_key(&id) {
                                                all_statuses.insert(id, action);
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        log::warn!("Bybit instrument status poll failed for {pt:?}: {e}");
                                    }
                                }
                            }
                        }

                        if want_statuses {
                            let ts = clock.get_time_ns();
                            let mut cache = (**status_cache.load()).clone();
                            let subs_guard = status_subs.load();
                            diff_and_emit_statuses(
                                &all_statuses, &mut cache, Some(&subs_guard), &sender, ts, ts,
                            );
                            status_cache.store(cache);
                        }
                    }
                    () = cancel.cancelled() => {
                        log::debug!("Bybit instrument polling task cancelled");
                        break;
                    }
                }
            }
        };

        self.session_tasks
            .spawn(future)
            .context("failed to register Bybit instrument polling task")?;
        log::debug!("Instrument polling started: interval={poll_secs}s");
        Ok(())
    }

    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        for ws_client in &self.ws_clients {
            ws_client.begin_shutdown();
        }

        for ws_client in &mut self.ws_clients {
            if let Err(e) = ws_client.close().await {
                self.shutdown_errors.push(e.to_string());
            }
        }

        if let Err(e) = self.finish_tasks().await {
            self.shutdown_errors.push(e.to_string());
        }
        self.is_connected.store(false, Ordering::Release);

        if self.shutdown_errors.is_empty() {
            Ok(())
        } else {
            let errors = std::mem::take(&mut self.shutdown_errors);
            anyhow::bail!("Bybit data shutdown failed: {}", errors.join("; "))
        }
    }
}

fn send_data(sender: &EventSender<DataEvent>, data: Data) {
    if let Err(e) = sender.send(DataEvent::Data(data)) {
        log::error!("Failed to emit data event: {e}");
    }
}

fn validate_orderbook_depth(depth: u32) -> anyhow::Result<()> {
    if !BYBIT_BOOK_DEPTHS.contains(&depth) {
        anyhow::bail!("invalid depth {depth}; valid values are {BYBIT_BOOK_DEPTHS:?}");
    }

    Ok(())
}

/// Cached funding state per symbol: (funding_rate, next_funding_time, funding_interval_hour).
type FundingCacheEntry = (Option<String>, Option<String>, Option<String>);

#[expect(clippy::too_many_arguments)]
fn handle_ws_message(
    message: &BybitWsMessage,
    data_sender: &EventSender<DataEvent>,
    instruments: &AHashMap<Ustr, InstrumentAny>,
    product_type: Option<BybitProductType>,
    trade_subs: &Arc<AtomicSet<InstrumentId>>,
    ticker_subs: &Arc<AtomicMap<InstrumentId, AHashSet<&'static str>>>,
    quote_subs: &Arc<AtomicSet<InstrumentId>>,
    book_depths: &Arc<AtomicMap<InstrumentId, u32>>,
    option_greeks_subs: &Arc<AtomicSet<InstrumentId>>,
    bar_types_cache: &Arc<AtomicMap<String, BarType>>,
    quote_cache: &mut QuoteCache,
    funding_cache: &mut AHashMap<Ustr, FundingCacheEntry>,
    clock: &AtomicTime,
) {
    let ts_init = clock.get_time_ns();
    let resolve = |raw_symbol: &Ustr| -> Option<&InstrumentAny> {
        let key = product_type.map_or(*raw_symbol, |pt| make_bybit_symbol(raw_symbol, pt));
        instruments.get(&key)
    };

    match message {
        BybitWsMessage::Orderbook(msg) => {
            let Some(instrument) = resolve(&msg.data.s) else {
                log::warn!("Unknown symbol in orderbook update: {}", msg.data.s);
                return;
            };
            let instrument_id = instrument.id();

            let Ok((depth, symbol)) = parse_orderbook_topic(msg.topic.as_str()) else {
                log::warn!("Invalid orderbook topic: {}", msg.topic);
                return;
            };

            if symbol != msg.data.s.as_str() {
                log::warn!(
                    "Orderbook topic symbol does not match payload: {}",
                    msg.topic
                );
                return;
            }

            if book_depths.load().get(&instrument_id) == Some(&depth) {
                match parse_orderbook_deltas(msg, instrument, ts_init) {
                    Ok(deltas) => {
                        send_data(data_sender, Data::BookDeltas(Box::new(deltas)));
                    }
                    Err(e) => log::error!("Failed to parse orderbook deltas: {e}"),
                }
            }

            if depth == BYBIT_QUOTE_DEPTH && quote_subs.contains(&instrument_id) {
                match parse_orderbook_quote(msg, instrument, ts_init) {
                    Ok(quote) => {
                        quote_cache.insert(instrument_id, quote);
                        send_data(data_sender, Data::Quote(quote));
                    }
                    Err(e) => log::error!("Failed to parse orderbook quote: {e}"),
                }
            }
        }
        BybitWsMessage::Trade(msg) => {
            for trade in &msg.data {
                let Some(instrument) = resolve(&trade.s) else {
                    continue;
                };
                let instrument_id = instrument.id();
                if !trade_subs.contains(&instrument_id) {
                    continue;
                }

                match parse_ws_trade_tick(trade, instrument, ts_init) {
                    Ok(tick) => send_data(data_sender, Data::Trade(tick)),
                    Err(e) => log::error!("Failed to parse trade tick: {e}"),
                }
            }
        }
        BybitWsMessage::Kline(msg) => {
            let Ok((_, raw_symbol)) = parse_kline_topic(msg.topic.as_str()) else {
                log::warn!("Invalid kline topic: {}", msg.topic);
                return;
            };
            let ustr_symbol = Ustr::from(raw_symbol);
            let Some(instrument) = resolve(&ustr_symbol) else {
                log::warn!("Unknown symbol in kline update: {raw_symbol}");
                return;
            };
            let topic_key = msg.topic.as_str();
            let Some(bar_type) = bar_types_cache.load().get(topic_key).copied() else {
                log::warn!("No bar type cached for kline topic: {topic_key}");
                return;
            };

            for kline in &msg.data {
                if !kline.confirm {
                    continue;
                }

                match parse_ws_kline_bar(kline, instrument, bar_type, true, ts_init) {
                    Ok(bar) => send_data(data_sender, Data::Bar(bar)),
                    Err(e) => log::error!("Failed to parse kline bar: {e}"),
                }
            }
        }
        BybitWsMessage::TickerLinear(msg) => {
            let Some(instrument) = resolve(&msg.data.symbol) else {
                log::warn!("Unknown symbol in ticker update: {}", msg.data.symbol);
                return;
            };
            let instrument_id = instrument.id();
            let subs = ticker_subs.load();
            let sub_set = subs.get(&instrument_id);

            let ts_event = match parse_millis_i64(msg.ts, "ticker.ts") {
                Ok(ts) => ts,
                Err(e) => {
                    log::error!("Failed to parse ticker timestamp: {e}");
                    return;
                }
            };

            if sub_set.is_some_and(|s| s.contains("funding"))
                && matches!(instrument, InstrumentAny::CryptoPerpetual(_))
            {
                let cache_entry = funding_cache
                    .entry(msg.data.symbol)
                    .or_insert((None, None, None));
                let mut changed = false;

                if let Some(rate) = &msg.data.funding_rate
                    && cache_entry.0.as_ref() != Some(rate)
                {
                    cache_entry.0 = Some(rate.clone());
                    changed = true;
                }

                if let Some(next_time) = &msg.data.next_funding_time
                    && cache_entry.1.as_ref() != Some(next_time)
                {
                    cache_entry.1 = Some(next_time.clone());
                    changed = true;
                }

                if let Some(interval) = &msg.data.funding_interval_hour {
                    cache_entry.2 = Some(interval.clone());
                }

                if changed && cache_entry.0.is_some() {
                    let mut merged = msg.data.clone();

                    if merged.funding_rate.is_none() {
                        merged.funding_rate.clone_from(&cache_entry.0);
                    }

                    if merged.next_funding_time.is_none() {
                        merged.next_funding_time.clone_from(&cache_entry.1);
                    }

                    if merged.funding_interval_hour.is_none() {
                        merged.funding_interval_hour.clone_from(&cache_entry.2);
                    }

                    match parse_ticker_linear_funding(&merged, instrument_id, ts_event, ts_init) {
                        Ok(update) => {
                            if let Err(e) = data_sender.send(DataEvent::FundingRate(update)) {
                                log::error!("Failed to emit funding rate event: {e}");
                            }
                        }
                        Err(e) => log::error!("Failed to parse ticker linear funding: {e}"),
                    }
                }
            }

            if sub_set.is_some_and(|s| s.contains("mark_prices")) && msg.data.mark_price.is_some() {
                match parse_ticker_linear_mark_price(&msg.data, instrument, ts_event, ts_init) {
                    Ok(update) => send_data(data_sender, Data::MarkPrice(update)),
                    Err(e) => log::debug!("Skipping mark price update: {e}"),
                }
            }

            if sub_set.is_some_and(|s| s.contains("index_prices")) && msg.data.index_price.is_some()
            {
                match parse_ticker_linear_index_price(&msg.data, instrument, ts_event, ts_init) {
                    Ok(update) => send_data(data_sender, Data::IndexPrice(update)),
                    Err(e) => log::debug!("Skipping index price update: {e}"),
                }
            }
        }
        BybitWsMessage::TickerOption(msg) => {
            let Some(instrument) = resolve(&msg.data.symbol) else {
                log::warn!(
                    "Unknown symbol in option ticker update: {}",
                    msg.data.symbol
                );
                return;
            };
            let instrument_id = instrument.id();
            let subs = ticker_subs.load();
            let sub_set = subs.get(&instrument_id);

            if sub_set.is_some_and(|s| s.contains("quotes")) {
                match parse_ticker_option_quote(msg, instrument, ts_init) {
                    Ok(quote) => {
                        let last = quote_cache.get(&instrument_id);
                        if last.is_none_or(|q| *q != quote) {
                            quote_cache.insert(instrument_id, quote);
                            send_data(data_sender, Data::Quote(quote));
                        }
                    }
                    Err(e) => log::error!("Failed to parse ticker option quote: {e}"),
                }
            }

            if sub_set.is_some_and(|s| s.contains("mark_prices")) {
                match parse_ticker_option_mark_price(msg, instrument, ts_init) {
                    Ok(update) => send_data(data_sender, Data::MarkPrice(update)),
                    Err(e) => log::error!("Failed to parse ticker option mark price: {e}"),
                }
            }

            if sub_set.is_some_and(|s| s.contains("index_prices")) {
                match parse_ticker_option_index_price(msg, instrument, ts_init) {
                    Ok(update) => send_data(data_sender, Data::IndexPrice(update)),
                    Err(e) => log::error!("Failed to parse ticker option index price: {e}"),
                }
            }

            if option_greeks_subs.contains(&instrument_id) {
                match parse_ticker_option_greeks(msg, instrument, ts_init) {
                    Ok(greeks) => {
                        if let Err(e) = data_sender.send(DataEvent::OptionGreeks(greeks)) {
                            log::error!("Failed to send option greeks: {e}");
                        }
                    }
                    Err(e) => log::error!("Failed to parse option greeks: {e}"),
                }
            }
        }
        BybitWsMessage::Reconnected => {
            quote_cache.clear();
            funding_cache.clear();
            log::info!("WebSocket reconnected, cleared caches");
        }
        BybitWsMessage::Error(e) => {
            log::warn!(
                "Bybit WebSocket error: code={} message={}",
                e.code,
                e.message
            );
        }
        BybitWsMessage::Auth(_)
        | BybitWsMessage::OrderResponse(_)
        | BybitWsMessage::AccountOrder(_)
        | BybitWsMessage::AccountExecution(_)
        | BybitWsMessage::AccountExecutionFast(_)
        | BybitWsMessage::AccountWallet(_)
        | BybitWsMessage::AccountPosition(_) => {}
    }
}

fn upsert_instrument(
    cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
    instrument: InstrumentAny,
) {
    cache.insert(instrument.id(), instrument);
}

#[async_trait::async_trait(?Send)]
impl DataClient for BybitDataClient {
    fn client_id(&self) -> ClientId {
        self.client_id
    }

    fn venue(&self) -> Option<Venue> {
        Some(self.venue())
    }

    fn start(&mut self) -> anyhow::Result<()> {
        log::info!(
            "Started: client_id={}, product_types={:?}, environment={:?}, proxy_url={:?}",
            self.client_id,
            self.config.product_types,
            self.config.environment,
            self.config.proxy_url,
        );
        Ok(())
    }

    fn stop(&mut self) -> anyhow::Result<()> {
        log::info!("Stopping {id}", id = self.client_id);
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        for ws_client in &self.ws_clients {
            ws_client.begin_shutdown();
        }
        self.is_connected.store(false, Ordering::Relaxed);
        Ok(())
    }

    fn reset(&mut self) -> anyhow::Result<()> {
        log::debug!("Resetting {id}", id = self.client_id);
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        for ws_client in &self.ws_clients {
            ws_client.begin_shutdown();
        }
        self.is_connected.store(false, Ordering::Relaxed);
        self.book_depths.store(AHashMap::new());
        self.quote_subs.store(AHashSet::new());
        self.ticker_subs.store(AHashMap::new());
        self.option_greeks_subs.store(AHashSet::new());
        self.instrument_status_subs.store(AHashSet::new());
        self.status_cache.store(AHashMap::new());
        self.instrument_subs.store(AHashSet::new());
        self.subscribe_all_instruments
            .store(false, Ordering::Relaxed);
        Ok(())
    }

    fn dispose(&mut self) -> anyhow::Result<()> {
        log::debug!("Disposing {id}", id = self.client_id);
        self.stop()
    }

    async fn connect(&mut self) -> anyhow::Result<()> {
        if self.is_connected() && self.session_tasks.is_open() && self.command_tasks.is_open() {
            return Ok(());
        }

        self.prepare_task_groups().await?;
        let ws_clients = self.ws_clients.clone();
        let setup_guard =
            TaskGroupGuard::new(&[&self.session_tasks, &self.command_tasks], move || {
                for ws_client in ws_clients {
                    ws_client.begin_shutdown();
                }
            });

        let product_types = if self.config.product_types.is_empty() {
            vec![BybitProductType::Linear]
        } else {
            self.config.product_types.clone()
        };

        let mut all_instruments = Vec::new();

        for product_type in &product_types {
            let fetched = self
                .http_client
                .request_instruments(*product_type, None, None)
                .await
                .with_context(|| {
                    format!("failed to request Bybit instruments for {product_type:?}")
                })?;

            self.http_client.cache_instruments(&fetched);

            self.instruments.rcu(|m| {
                for instrument in &fetched {
                    m.insert(instrument.id(), instrument.clone());
                }
            });

            all_instruments.extend(fetched);
        }

        // Seed instrument status cache from initial fetch
        if self
            .config
            .instrument_poll_interval_secs
            .is_some_and(|s| s > 0)
        {
            // Collect all statuses first (without holding the lock across await)
            let mut collected_statuses = Vec::new();

            for product_type in &product_types {
                match self
                    .http_client
                    .request_instrument_statuses(*product_type)
                    .await
                {
                    Ok(statuses) => collected_statuses.push(statuses),
                    Err(e) => {
                        log::warn!(
                            "Failed to seed instrument status cache for {product_type:?}: {e}"
                        );
                    }
                }
            }

            let inst_guard = self.instruments.load();
            let mut status_map = AHashMap::new();

            for statuses in collected_statuses {
                for (id, action) in statuses {
                    if inst_guard.contains_key(&id) {
                        status_map.insert(id, action);
                    }
                }
            }
            log::debug!(
                "Seeded instrument status cache with {} entries",
                status_map.len()
            );
            self.status_cache.store(status_map);
        }

        for instrument in all_instruments {
            if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
                log::warn!("Failed to send instrument: {e}");
            }
        }

        // Build instruments map keyed by full Nautilus symbol for parsing
        let instruments_by_symbol: Arc<AHashMap<Ustr, InstrumentAny>> = {
            let guard = self.instruments.load();
            let mut map = AHashMap::new();
            for instrument in guard.values() {
                map.insert(instrument.id().symbol.inner(), instrument.clone());
            }
            Arc::new(map)
        };

        let session_result = async {
            for ws_client in &mut self.ws_clients {
                ws_client
                    .connect()
                    .await
                    .context("failed to connect Bybit WebSocket")?;
                ws_client
                    .wait_until_active(10.0)
                    .await
                    .context("WebSocket did not become active")?;

                let stream = ws_client.stream();
                let product_type = ws_client.product_type();
                let sender = self.data_sender.clone();
                let trade_subs = self.trade_subs.clone();
                let ticker_subs = self.ticker_subs.clone();
                let quote_subs = self.quote_subs.clone();
                let book_depths = self.book_depths.clone();
                let option_greeks_subs = self.option_greeks_subs.clone();
                let bar_types_cache = ws_client.bar_types_cache().clone();
                let instruments = Arc::clone(&instruments_by_symbol);
                let clock = self.clock;
                let cancel = self.cancellation_token.clone();

                let future = async move {
                    let mut quote_cache = QuoteCache::new();
                    let mut funding_cache: AHashMap<Ustr, FundingCacheEntry> = AHashMap::new();

                    pin_mut!(stream);

                    loop {
                        tokio::select! {
                            Some(message) = stream.next() => {
                                handle_ws_message(
                                    &message,
                                    &sender,
                                    &instruments,
                                    product_type,
                                    &trade_subs,
                                    &ticker_subs,
                                    &quote_subs,
                                    &book_depths,
                                    &option_greeks_subs,
                                    &bar_types_cache,
                                    &mut quote_cache,
                                    &mut funding_cache,
                                    clock,
                                );
                            }
                            () = cancel.cancelled() => {
                                log::debug!("WebSocket stream task cancelled");
                                break;
                            }
                        }
                    }
                };
                self.session_tasks
                    .spawn(future)
                    .context("failed to register Bybit WebSocket stream task")?;
            }

            if let Some(poll_secs) = self.config.instrument_poll_interval_secs
                && poll_secs > 0
            {
                self.spawn_instrument_polling(&product_types, poll_secs)?;
            }

            Ok::<(), anyhow::Error>(())
        }
        .await;

        if let Err(e) = session_result {
            if let Err(teardown_error) = self.teardown_partial_connect().await {
                return Err(e.context(format!(
                    "Bybit data startup teardown failed: {teardown_error}"
                )));
            }
            return Err(e);
        }

        setup_guard.disarm();
        self.is_connected.store(true, Ordering::Release);
        log::info!("Connected: client_id={}", self.client_id);
        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        for ws_client in &self.ws_clients {
            ws_client.begin_shutdown();
        }

        for ws_client in &mut self.ws_clients {
            if let Err(e) = ws_client.close().await {
                self.shutdown_errors.push(e.to_string());
            }
        }

        // Allow time for unsubscribe confirmations
        tokio::time::sleep(Duration::from_millis(500)).await;

        if let Err(e) = self.finish_tasks().await {
            self.shutdown_errors.push(e.to_string());
        }

        self.book_depths.store(AHashMap::new());
        self.quote_subs.store(AHashSet::new());
        self.ticker_subs.store(AHashMap::new());
        self.trade_subs.store(AHashSet::new());
        self.option_greeks_subs.store(AHashSet::new());
        self.instrument_status_subs.store(AHashSet::new());
        self.status_cache.store(AHashMap::new());
        self.instrument_subs.store(AHashSet::new());
        self.subscribe_all_instruments
            .store(false, Ordering::Relaxed);
        self.is_connected.store(false, Ordering::Release);
        log::info!("Disconnected: client_id={}", self.client_id);

        if self.shutdown_errors.is_empty() {
            Ok(())
        } else {
            let errors = std::mem::take(&mut self.shutdown_errors);
            anyhow::bail!("Bybit data shutdown failed: {}", errors.join("; "))
        }
    }

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

    fn is_disconnected(&self) -> bool {
        !self.is_connected()
    }

    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
        if cmd.book_type != BookType::L2_MBP {
            anyhow::bail!("Bybit only supports L2_MBP order book deltas");
        }

        let depth = cmd
            .depth
            .map_or(BYBIT_DEFAULT_ORDERBOOK_DEPTH, |d| d.get() as u32);

        validate_orderbook_depth(depth)?;

        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        let ws = self
            .get_ws_client_for_product(product_type)
            .context("no WebSocket client for product type")?
            .clone();

        if let Some(subscribed_depth) = self.book_depths.load().get(&instrument_id) {
            anyhow::ensure!(
                *subscribed_depth == depth,
                "Already subscribed to book depth {subscribed_depth} for {instrument_id}"
            );
            return Ok(());
        }

        self.book_depths.insert(instrument_id, depth);
        let book_depths = Arc::clone(&self.book_depths);

        self.spawn_ws(
            async move {
                if let Err(e) = ws.subscribe_orderbook(instrument_id, depth).await {
                    if let Err(e) = ws.unsubscribe_orderbook(instrument_id, depth).await {
                        log::warn!("Failed to unsubscribe after orderbook subscription error: {e}");
                    }
                    book_depths.rcu(|depths| {
                        if depths.get(&instrument_id) == Some(&depth) {
                            depths.remove(&instrument_id);
                        }
                    });
                    return Err(e).context("orderbook subscription");
                }
                Ok(())
            },
            "order book delta subscription",
        );

        Ok(())
    }

    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        let ws = self
            .get_ws_client_for_product(product_type)
            .context("no WebSocket client for product type")?
            .clone();

        if product_type == BybitProductType::Option {
            let mut should_subscribe = false;
            self.ticker_subs.rcu(|m| {
                let entry = m.entry(instrument_id).or_default();
                should_subscribe = entry.is_empty();
                entry.insert("quotes");
            });

            if should_subscribe {
                self.spawn_ws(
                    async move {
                        ws.subscribe_ticker(instrument_id)
                            .await
                            .context("ticker subscription")
                    },
                    "quote subscription",
                );
            }
        } else {
            if self.quote_subs.contains(&instrument_id) {
                return Ok(());
            }

            self.quote_subs.insert(instrument_id);

            self.spawn_ws(
                async move {
                    ws.subscribe_orderbook(instrument_id, BYBIT_QUOTE_DEPTH)
                        .await
                        .context("orderbook subscription for quotes")
                },
                "quote subscription (orderbook)",
            );
        }
        Ok(())
    }

    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        self.trade_subs.insert(instrument_id);

        let ws = self
            .get_ws_client_for_product(product_type)
            .context("no WebSocket client for product type")?
            .clone();

        self.spawn_ws(
            async move {
                ws.subscribe_trades(instrument_id)
                    .await
                    .context("trades subscription")
            },
            "trade subscription",
        );
        Ok(())
    }

    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        if product_type == BybitProductType::Spot || product_type == BybitProductType::Option {
            anyhow::bail!("Funding rates not available for {product_type:?} instruments");
        }

        let guard = self.instruments.load();
        if let Some(instrument) = guard.get(&instrument_id)
            && !matches!(instrument, InstrumentAny::CryptoPerpetual(_))
        {
            anyhow::bail!("Funding rates only available for perpetuals, not {instrument_id}");
        }

        let mut should_subscribe = false;
        self.ticker_subs.rcu(|m| {
            let entry = m.entry(instrument_id).or_default();
            should_subscribe = entry.is_empty();
            entry.insert("funding");
        });

        if should_subscribe {
            let ws = self
                .get_ws_client_for_product(product_type)
                .context("no WebSocket client for product type")?
                .clone();

            self.spawn_ws(
                async move {
                    ws.subscribe_ticker(instrument_id)
                        .await
                        .context("ticker subscription for funding rates")
                },
                "funding rate subscription",
            );
        }
        Ok(())
    }

    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        if product_type == BybitProductType::Spot {
            anyhow::bail!("Mark prices not available for Spot instruments");
        }

        let mut should_subscribe = false;
        self.ticker_subs.rcu(|m| {
            let entry = m.entry(instrument_id).or_default();
            should_subscribe = entry.is_empty();
            entry.insert("mark_prices");
        });

        if should_subscribe {
            let ws = self
                .get_ws_client_for_product(product_type)
                .context("no WebSocket client for product type")?
                .clone();

            self.spawn_ws(
                async move {
                    ws.subscribe_ticker(instrument_id)
                        .await
                        .context("ticker subscription for mark prices")
                },
                "mark price subscription",
            );
        }
        Ok(())
    }

    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        if product_type == BybitProductType::Spot {
            anyhow::bail!("Index prices not available for Spot instruments");
        }

        let mut should_subscribe = false;
        self.ticker_subs.rcu(|m| {
            let entry = m.entry(instrument_id).or_default();
            should_subscribe = entry.is_empty();
            entry.insert("index_prices");
        });

        if should_subscribe {
            let ws = self
                .get_ws_client_for_product(product_type)
                .context("no WebSocket client for product type")?
                .clone();

            self.spawn_ws(
                async move {
                    ws.subscribe_ticker(instrument_id)
                        .await
                        .context("ticker subscription for index prices")
                },
                "index price subscription",
            );
        }
        Ok(())
    }

    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
        let bar_type = cmd.bar_type;
        let instrument_id = bar_type.instrument_id();
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        if product_type == BybitProductType::Option {
            anyhow::bail!("Bybit does not support kline/bar data for options");
        }

        let ws = self
            .get_ws_client_for_product(product_type)
            .context("no WebSocket client for product type")?
            .clone();

        self.spawn_ws(
            async move {
                ws.subscribe_bars(bar_type)
                    .await
                    .context("bars subscription")
            },
            "bar subscription",
        );
        Ok(())
    }

    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let Some(depth) = self.book_depths.load().get(&instrument_id).copied() else {
            return Ok(());
        };

        self.book_depths.remove(&instrument_id);

        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        let ws = self
            .get_ws_client_for_product(product_type)
            .context("no WebSocket client for product type")?
            .clone();

        self.spawn_ws(
            async move {
                ws.unsubscribe_orderbook(instrument_id, depth)
                    .await
                    .context("orderbook unsubscribe")
            },
            "order book unsubscribe",
        );
        Ok(())
    }

    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        let ws = self
            .get_ws_client_for_product(product_type)
            .context("no WebSocket client for product type")?
            .clone();

        if product_type == BybitProductType::Option {
            let mut should_unsubscribe = false;
            self.ticker_subs.rcu(|m| {
                if let Some(entry) = m.get_mut(&instrument_id) {
                    entry.remove("quotes");
                    if entry.is_empty() {
                        m.remove(&instrument_id);
                        should_unsubscribe = true;
                    } else {
                        should_unsubscribe = false;
                    }
                } else {
                    should_unsubscribe = false;
                }
            });

            if should_unsubscribe {
                self.spawn_ws(
                    async move {
                        ws.unsubscribe_ticker(instrument_id)
                            .await
                            .context("ticker unsubscribe")
                    },
                    "quote unsubscribe",
                );
            }
        } else {
            if !self.quote_subs.contains(&instrument_id) {
                return Ok(());
            }

            self.quote_subs.remove(&instrument_id);

            self.spawn_ws(
                async move {
                    ws.unsubscribe_orderbook(instrument_id, BYBIT_QUOTE_DEPTH)
                        .await
                        .context("orderbook unsubscribe for quotes")
                },
                "quote unsubscribe (orderbook)",
            );
        }
        Ok(())
    }

    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        self.trade_subs.remove(&instrument_id);

        let ws = self
            .get_ws_client_for_product(product_type)
            .context("no WebSocket client for product type")?
            .clone();

        self.spawn_ws(
            async move {
                ws.unsubscribe_trades(instrument_id)
                    .await
                    .context("trades unsubscribe")
            },
            "trade unsubscribe",
        );
        Ok(())
    }

    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        let mut should_unsubscribe = false;
        self.ticker_subs.rcu(|m| {
            if let Some(entry) = m.get_mut(&instrument_id) {
                entry.remove("funding");
                if entry.is_empty() {
                    m.remove(&instrument_id);
                    should_unsubscribe = true;
                } else {
                    should_unsubscribe = false;
                }
            } else {
                should_unsubscribe = false;
            }
        });

        if should_unsubscribe {
            let ws = self
                .get_ws_client_for_product(product_type)
                .context("no WebSocket client for product type")?
                .clone();

            self.spawn_ws(
                async move {
                    ws.unsubscribe_ticker(instrument_id)
                        .await
                        .context("ticker unsubscribe for funding rates")
                },
                "funding rate unsubscribe",
            );
        }
        Ok(())
    }

    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        let mut should_unsubscribe = false;
        self.ticker_subs.rcu(|m| {
            if let Some(entry) = m.get_mut(&instrument_id) {
                entry.remove("mark_prices");
                if entry.is_empty() {
                    m.remove(&instrument_id);
                    should_unsubscribe = true;
                } else {
                    should_unsubscribe = false;
                }
            } else {
                should_unsubscribe = false;
            }
        });

        if should_unsubscribe {
            let ws = self
                .get_ws_client_for_product(product_type)
                .context("no WebSocket client for product type")?
                .clone();

            self.spawn_ws(
                async move {
                    ws.unsubscribe_ticker(instrument_id)
                        .await
                        .context("ticker unsubscribe for mark prices")
                },
                "mark price unsubscribe",
            );
        }
        Ok(())
    }

    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        let mut should_unsubscribe = false;
        self.ticker_subs.rcu(|m| {
            if let Some(entry) = m.get_mut(&instrument_id) {
                entry.remove("index_prices");
                if entry.is_empty() {
                    m.remove(&instrument_id);
                    should_unsubscribe = true;
                } else {
                    should_unsubscribe = false;
                }
            } else {
                should_unsubscribe = false;
            }
        });

        if should_unsubscribe {
            let ws = self
                .get_ws_client_for_product(product_type)
                .context("no WebSocket client for product type")?
                .clone();

            self.spawn_ws(
                async move {
                    ws.unsubscribe_ticker(instrument_id)
                        .await
                        .context("ticker unsubscribe for index prices")
                },
                "index price unsubscribe",
            );
        }
        Ok(())
    }

    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
        let bar_type = cmd.bar_type;
        let instrument_id = bar_type.instrument_id();
        let product_type = self
            .get_product_type_for_instrument(instrument_id)
            .unwrap_or(BybitProductType::Linear);

        let ws = self
            .get_ws_client_for_product(product_type)
            .context("no WebSocket client for product type")?
            .clone();

        self.spawn_ws(
            async move {
                ws.unsubscribe_bars(bar_type)
                    .await
                    .context("bars unsubscribe")
            },
            "bar unsubscribe",
        );
        Ok(())
    }

    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        self.option_greeks_subs.insert(instrument_id);

        let mut should_subscribe = false;
        self.ticker_subs.rcu(|m| {
            let entry = m.entry(instrument_id).or_default();
            should_subscribe = entry.is_empty();
            entry.insert("option_greeks");
        });

        if should_subscribe {
            let product_type = self
                .get_product_type_for_instrument(instrument_id)
                .unwrap_or(BybitProductType::Option);

            let ws = self
                .get_ws_client_for_product(product_type)
                .context("no WebSocket client for product type")?
                .clone();

            self.spawn_ws(
                async move {
                    ws.subscribe_ticker(instrument_id)
                        .await
                        .context("ticker subscription for option greeks")
                },
                "option greeks subscription",
            );
        }
        Ok(())
    }

    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        self.option_greeks_subs.remove(&instrument_id);

        let mut should_unsubscribe = false;
        self.ticker_subs.rcu(|m| {
            if let Some(entry) = m.get_mut(&instrument_id) {
                entry.remove("option_greeks");
                if entry.is_empty() {
                    m.remove(&instrument_id);
                    should_unsubscribe = true;
                } else {
                    should_unsubscribe = false;
                }
            } else {
                should_unsubscribe = false;
            }
        });

        if should_unsubscribe {
            let product_type = self
                .get_product_type_for_instrument(instrument_id)
                .unwrap_or(BybitProductType::Option);

            let ws = self
                .get_ws_client_for_product(product_type)
                .context("no WebSocket client for product type")?
                .clone();

            self.spawn_ws(
                async move {
                    ws.unsubscribe_ticker(instrument_id)
                        .await
                        .context("ticker unsubscribe for option greeks")
                },
                "option greeks unsubscribe",
            );
        }
        Ok(())
    }

    fn subscribe_instruments(&mut self, cmd: SubscribeInstruments) -> anyhow::Result<()> {
        log::debug!(
            "subscribe_instruments: {venue} (definition updates detected via periodic instrument info polling)",
            venue = cmd.venue,
        );
        self.subscribe_all_instruments
            .store(true, Ordering::Relaxed);
        Ok(())
    }

    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
        log::debug!(
            "subscribe_instrument: {id} (definition updates detected via periodic instrument info polling)",
            id = cmd.instrument_id,
        );
        self.instrument_subs.insert(cmd.instrument_id);
        Ok(())
    }

    fn unsubscribe_instruments(&mut self, cmd: &UnsubscribeInstruments) -> anyhow::Result<()> {
        log::debug!("unsubscribe_instruments: {venue}", venue = cmd.venue);
        self.subscribe_all_instruments
            .store(false, Ordering::Relaxed);
        Ok(())
    }

    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
        log::debug!("unsubscribe_instrument: {id}", id = cmd.instrument_id);
        self.instrument_subs.remove(&cmd.instrument_id);
        Ok(())
    }

    fn subscribe_instrument_status(
        &mut self,
        cmd: SubscribeInstrumentStatus,
    ) -> anyhow::Result<()> {
        log::debug!(
            "subscribe_instrument_status: {id} (status changes detected via periodic instrument info polling)",
            id = cmd.instrument_id,
        );
        self.instrument_status_subs.insert(cmd.instrument_id);

        if let Some(action) = self.status_cache.load().get(&cmd.instrument_id).copied() {
            let ts = self.clock.get_time_ns();
            emit_status(&self.data_sender, cmd.instrument_id, action, ts, ts);
        }

        Ok(())
    }

    fn unsubscribe_instrument_status(
        &mut self,
        cmd: &UnsubscribeInstrumentStatus,
    ) -> anyhow::Result<()> {
        log::debug!(
            "unsubscribe_instrument_status: {id}",
            id = cmd.instrument_id,
        );
        self.instrument_status_subs.remove(&cmd.instrument_id);
        Ok(())
    }

    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instruments_cache = self.instruments.clone();
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let venue = self.venue();
        let start = request.start;
        let end = request.end;
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);
        let product_types = if self.config.product_types.is_empty() {
            vec![BybitProductType::Linear]
        } else {
            self.config.product_types.clone()
        };

        self.spawn_command(async move {
            let mut all_instruments = Vec::new();

            for product_type in product_types {
                match http.request_instruments(product_type, None, None).await {
                    Ok(instruments) => {
                        for instrument in instruments {
                            upsert_instrument(&instruments_cache, instrument.clone());
                            all_instruments.push(instrument);
                        }
                    }
                    Err(e) => {
                        log::error!("Failed to fetch instruments for {product_type:?}: {e:?}");
                    }
                }
            }

            let response = DataResponse::Instruments(InstrumentsResponse::new(
                request_id,
                client_id,
                venue,
                all_instruments,
                start_nanos,
                end_nanos,
                clock.get_time_ns(),
                params,
            ));

            if let Err(e) = sender.send(DataEvent::Response(response)) {
                log::error!("Failed to send instruments response: {e}");
            }
        });

        Ok(())
    }

    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instruments = self.instruments.clone();
        let instrument_id = request.instrument_id;
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let start = request.start;
        let end = request.end;
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);

        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
            .unwrap_or(BybitProductType::Linear);
        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str()).to_string();

        self.spawn_command(async move {
            match http
                .request_instruments(product_type, Some(raw_symbol), None)
                .await
                .context("fetch instrument from API")
            {
                Ok(fetched) => {
                    if let Some(instrument) = fetched.into_iter().find(|i| i.id() == instrument_id)
                    {
                        upsert_instrument(&instruments, instrument.clone());

                        let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
                            request_id,
                            client_id,
                            instrument.id(),
                            instrument,
                            start_nanos,
                            end_nanos,
                            clock.get_time_ns(),
                            params,
                        )));

                        if let Err(e) = sender.send(DataEvent::Response(response)) {
                            log::error!("Failed to send instrument response: {e}");
                        }
                    } else {
                        log::error!("Instrument not found: {instrument_id}");
                    }
                }
                Err(e) => log::error!("Instrument request failed: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instrument_id = request.instrument_id;
        let depth = request.depth.map(|n| n.get() as u32);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;

        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
            .unwrap_or(BybitProductType::Linear);

        self.spawn_command(async move {
            match http
                .request_orderbook_snapshot(product_type, instrument_id, depth)
                .await
                .context("failed to request book snapshot from Bybit")
            {
                Ok(deltas) => {
                    let mut book = OrderBook::new(instrument_id, BookType::L2_MBP);
                    if let Err(e) = book.apply_deltas(&deltas) {
                        log::error!("Failed to apply book deltas for {instrument_id}: {e}");
                        return;
                    }

                    let response = DataResponse::Book(BookResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        book,
                        None,
                        None,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send book snapshot response: {e}");
                    }
                }
                Err(e) => log::error!("Book snapshot request failed for {instrument_id}: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instrument_id = request.instrument_id;
        let start = request.start;
        let end = request.end;
        let limit = request.limit.map(|n| n.get() as u32);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);

        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
            .unwrap_or(BybitProductType::Linear);

        self.spawn_command(async move {
            match http
                .request_trades(product_type, instrument_id, limit)
                .await
                .context("failed to request trades from Bybit")
            {
                Ok(trades) => {
                    let response = DataResponse::Trades(TradesResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        trades,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send trades response: {e}");
                    }
                }
                Err(e) => log::error!("Trade request failed: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let bar_type = request.bar_type;
        let start = request.start;
        let end = request.end;
        let limit = request.limit.map(|n| n.get() as u32);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);

        let instrument_id = bar_type.instrument_id();
        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
            .unwrap_or(BybitProductType::Linear);

        self.spawn_command(async move {
            match http
                .request_bars(product_type, bar_type, start, end, limit, true)
                .await
                .context("failed to request bars from Bybit")
            {
                Ok(bars) => {
                    let response = DataResponse::Bars(BarsResponse::new(
                        request_id,
                        client_id,
                        bar_type,
                        bars,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send bars response: {e}");
                    }
                }
                Err(e) => log::error!("Bar request failed: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
        let http = self.http_client.clone();
        let sender = self.data_sender.clone();
        let instrument_id = request.instrument_id;
        let start = request.start;
        let end = request.end;
        let limit = request.limit.map(|n| n.get() as u32);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);

        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
            .unwrap_or(BybitProductType::Linear);

        if product_type == BybitProductType::Spot || product_type == BybitProductType::Option {
            anyhow::bail!("Funding rates not available for {product_type} instruments");
        }

        self.spawn_command(async move {
            match http
                .request_funding_rates(product_type, instrument_id, start, end, limit)
                .await
                .context("failed to request funding rates from Bybit")
            {
                Ok(funding_rates) => {
                    let response = DataResponse::FundingRates(FundingRatesResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        funding_rates,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send funding rates response: {e}");
                    }
                }
                Err(e) => log::error!("Funding rates request failed for {instrument_id}: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_option_chain_reference_price(
        &self,
        request: RequestOptionChainReferencePrice,
    ) -> anyhow::Result<()> {
        let series_id = request.series_id;
        let instrument_id = request.instrument_id;
        let http_client = self.http_client.clone();
        let sender = self.data_sender.clone();
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id());
        let params = request.params;
        let clock = self.clock;

        self.spawn_command(async move {
            let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str()).to_string();
            let query = crate::http::query::BybitTickersParams {
                category: BybitProductType::Option,
                symbol: Some(raw_symbol),
                base_coin: None,
                exp_date: None,
            };
            let price = match http_client.request_option_tickers_raw_with_params(&query).await {
                Ok(tickers) => tickers.into_iter().find_map(|ticker| {
                    let decimal = ticker.underlying_price.parse::<Decimal>().ok()?;
                    if decimal <= Decimal::ZERO {
                        return None;
                    }

                    match Price::from_decimal(decimal) {
                        Ok(price) => Some(price),
                        Err(e) => {
                            log::warn!(
                                "Invalid Bybit option-chain reference price for {instrument_id}: {e}"
                            );
                            None
                        }
                    }
                }),
                Err(e) => {
                    log::error!(
                        "Option-chain reference price request failed for {series_id}: {e:?}"
                    );
                    None
                }
            };
            let response = DataResponse::OptionChainReferencePrice(
                OptionChainReferencePriceResponse::new(
                    request_id,
                    client_id,
                    series_id,
                    price,
                    clock.get_time_ns(),
                    params,
                ),
            );

            if let Err(e) = sender.send(DataEvent::Response(response)) {
                log::error!("Failed to send option-chain reference price response: {e}");
            }
        });

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::{sync::Arc, time::Duration};

    use ahash::{AHashMap, AHashSet};
    use nautilus_common::{
        cache::quote::QuoteCache,
        clients::DataClient,
        live::runner::set_data_event_sender,
        messages::{DataEvent, data::SubscribeBookDeltas},
        testing::wait_until_async,
    };
    use nautilus_core::{
        AtomicMap, AtomicSet, UUID4, UnixNanos,
        time::{AtomicTime, get_atomic_clock_realtime},
    };
    use nautilus_model::{
        data::{BarType, Data, QuoteTick},
        enums::{AggressorSide, BookAction, BookType},
        identifiers::InstrumentId,
        instruments::{Instrument, InstrumentAny},
        types::{Price, Quantity},
    };
    use rstest::rstest;
    use ustr::Ustr;

    use super::{BybitDataClient, handle_ws_message, validate_orderbook_depth};
    use crate::{
        common::{
            consts::BYBIT_CLIENT_ID,
            enums::BybitProductType,
            parse::{parse_linear_instrument, parse_option_instrument},
            rate_limit::BYBIT_OPTION_SUBSCRIPTION_LIMIT,
            testing::load_test_json,
        },
        config::BybitDataClientConfig,
        http::models::{
            BybitFeeRate, BybitInstrumentLinearResponse, BybitInstrumentOptionResponse,
        },
        websocket::messages::{
            BybitWsMessage, BybitWsOrderbookDepthMsg, BybitWsTickerLinearMsg,
            BybitWsTickerOptionMsg, BybitWsTradeMsg,
        },
    };

    fn sample_fee_rate(
        symbol: &str,
        taker: &str,
        maker: &str,
        base_coin: Option<&str>,
    ) -> BybitFeeRate {
        BybitFeeRate {
            symbol: Ustr::from(symbol),
            taker_fee_rate: taker.to_string(),
            maker_fee_rate: maker.to_string(),
            base_coin: base_coin.map(Ustr::from),
        }
    }

    fn linear_instrument() -> InstrumentAny {
        let json = load_test_json("http_get_instruments_linear.json");
        let response: BybitInstrumentLinearResponse = serde_json::from_str(&json).unwrap();
        let instrument = &response.result.list[0];
        let fee_rate = sample_fee_rate("BTCUSDT", "0.00055", "0.0001", Some("BTC"));
        let ts = UnixNanos::new(1_700_000_000_000_000_000);
        parse_linear_instrument(instrument, &fee_rate, ts, ts).unwrap()
    }

    fn option_instrument() -> InstrumentAny {
        let json = load_test_json("http_get_instruments_option.json");
        let response: BybitInstrumentOptionResponse = serde_json::from_str(&json).unwrap();
        let instrument = &response.result.list[0];
        let ts = UnixNanos::new(1_700_000_000_000_000_000);
        parse_option_instrument(instrument, None, ts, ts).unwrap()
    }

    fn build_instruments(instruments: &[InstrumentAny]) -> AHashMap<Ustr, InstrumentAny> {
        let mut map = AHashMap::new();
        for inst in instruments {
            map.insert(inst.id().symbol.inner(), inst.clone());
        }
        map
    }

    #[expect(clippy::type_complexity)]
    fn empty_subs() -> (
        Arc<AtomicSet<InstrumentId>>,
        Arc<AtomicMap<InstrumentId, AHashSet<&'static str>>>,
        Arc<AtomicSet<InstrumentId>>,
        Arc<AtomicMap<InstrumentId, u32>>,
        Arc<AtomicSet<InstrumentId>>,
        Arc<AtomicMap<String, BarType>>,
    ) {
        (
            Arc::new(AtomicSet::new()),
            Arc::new(AtomicMap::new()),
            Arc::new(AtomicSet::new()),
            Arc::new(AtomicMap::new()),
            Arc::new(AtomicSet::new()),
            Arc::new(AtomicMap::new()),
        )
    }

    #[tokio::test]
    async fn test_failed_book_subscription_releases_registration() {
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        set_data_event_sender(tx);
        let config = BybitDataClientConfig {
            product_types: vec![BybitProductType::Option],
            ..Default::default()
        };
        let mut client = BybitDataClient::new(*BYBIT_CLIENT_ID, config).unwrap();
        let ws = client.ws_clients[0].clone();
        let topics = (0..BYBIT_OPTION_SUBSCRIPTION_LIMIT)
            .map(|index| format!("tickers.OPTION-{index}"))
            .collect();

        // The disconnected command channel rejects the send after registering these topics
        ws.subscribe(topics).await.unwrap_err();
        let instrument_id = InstrumentId::from("ETH-26JUN26-16000-P-OPTION.BYBIT");
        let expected_error = format!(
            "Client error: Option WebSocket subscription limit is {BYBIT_OPTION_SUBSCRIPTION_LIMIT} arguments per connection, requested {}",
            BYBIT_OPTION_SUBSCRIPTION_LIMIT + 1,
        );
        assert_eq!(
            ws.subscribe_orderbook(instrument_id, 50)
                .await
                .unwrap_err()
                .to_string(),
            expected_error,
        );
        client
            .instruments
            .insert(instrument_id, option_instrument());
        let command = SubscribeBookDeltas::new(
            instrument_id,
            BookType::L2_MBP,
            Some(*BYBIT_CLIENT_ID),
            None,
            UUID4::new(),
            UnixNanos::default(),
            None,
            false,
            None,
            None,
        );

        for _ in 0..2 {
            client.subscribe_book_deltas(command.clone()).unwrap();
            assert_eq!(client.book_depths.load().get(&instrument_id), Some(&50));
            wait_until_async(
                || async { !client.book_depths.contains_key(&instrument_id) },
                Duration::from_secs(2),
            )
            .await;
            assert!(!client.book_depths.contains_key(&instrument_id));
            assert_eq!(
                ws.subscribe_orderbook(instrument_id, 50)
                    .await
                    .unwrap_err()
                    .to_string(),
                expected_error,
            );
        }
    }

    #[tokio::test]
    async fn test_failed_book_subscription_releases_transport_reference() {
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        set_data_event_sender(tx);
        let mut client =
            BybitDataClient::new(*BYBIT_CLIENT_ID, BybitDataClientConfig::default()).unwrap();
        let ws = client.ws_clients[0].clone();
        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
        client
            .subscribe_book_deltas(SubscribeBookDeltas::new(
                instrument_id,
                BookType::L2_MBP,
                Some(*BYBIT_CLIENT_ID),
                None,
                UUID4::new(),
                UnixNanos::default(),
                None,
                false,
                None,
                None,
            ))
            .unwrap();
        wait_until_async(
            || async { !client.book_depths.contains_key(&instrument_id) },
            Duration::from_secs(2),
        )
        .await;

        let error = ws.subscribe_orderbook(instrument_id, 50).await.unwrap_err();

        assert_eq!(
            error.to_string(),
            "WebSocket send error: Failed to send subscribe command: channel closed",
        );
        assert!(!client.book_depths.contains_key(&instrument_id));
    }

    #[rstest]
    fn test_validate_orderbook_depth_accepts_1000() {
        assert!(validate_orderbook_depth(1000).is_ok());
    }

    #[rstest]
    fn test_validate_orderbook_depth_rejects_500() {
        let e = validate_orderbook_depth(500).unwrap_err();

        assert_eq!(
            e.to_string(),
            "invalid depth 500; valid values are [1, 50, 200, 1000]"
        );
    }

    #[rstest]
    fn test_handle_trade_message_emits_trade_tick() {
        let instrument = linear_instrument();
        let instruments = build_instruments(std::slice::from_ref(&instrument));
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();
        trade_subs.insert(instrument.id());
        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_public_trade.json");
        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
        let ws_msg = BybitWsMessage::Trade(msg);

        handle_ws_message(
            &ws_msg,
            &tx.into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        let event = rx.try_recv().unwrap();
        match event {
            DataEvent::Data(Data::Trade(tick)) => {
                assert_eq!(tick.instrument_id, instrument.id());
                assert_eq!(tick.price, instrument.make_price(27451.00));
                assert_eq!(tick.size, instrument.make_qty(0.010, None));
                assert_eq!(tick.aggressor_side, AggressorSide::Buy);
            }
            other => panic!("Expected Trade data event, found {other:?}"),
        }
    }

    #[rstest]
    fn test_handle_trade_message_unknown_symbol_no_event() {
        let instruments = AHashMap::new();
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();
        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_public_trade.json");
        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
        let ws_msg = BybitWsMessage::Trade(msg);

        handle_ws_message(
            &ws_msg,
            &tx.into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        rx.try_recv().unwrap_err();
    }

    #[rstest]
    #[case::shared_depth_one("orderbook.1.BTCUSDT", 1, "snapshot", true, true)]
    #[case::depth_one_with_deeper_book("orderbook.1.BTCUSDT", 50, "snapshot", false, true)]
    #[case::deeper_snapshot("orderbook.50.BTCUSDT", 50, "snapshot", true, false)]
    #[case::deeper_deletion("orderbook.50.BTCUSDT", 50, "delta", true, false)]
    #[case::unsubscribed_depth("orderbook.200.BTCUSDT", 50, "delta", false, false)]
    #[case::invalid_topic("orderbook.invalid.BTCUSDT", 1, "snapshot", false, false)]
    #[case::wrong_symbol("orderbook.1.ETHUSDT", 1, "snapshot", false, false)]
    fn test_handle_orderbook_routes_by_depth(
        #[case] topic: &str,
        #[case] book_depth: u32,
        #[case] msg_type: &str,
        #[case] expect_book: bool,
        #[case] expect_quote: bool,
    ) {
        let instrument = linear_instrument();
        let instrument_id = instrument.id();
        let instruments = build_instruments(&[instrument]);
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();
        book_depths.insert(instrument_id, book_depth);
        quote_subs.insert(instrument_id);
        let previous = QuoteTick::new(
            instrument_id,
            Price::from("27000.00"),
            Price::from("27001.00"),
            Quantity::from("2.000"),
            Quantity::from("3.000"),
            UnixNanos::new(1),
            UnixNanos::new(2),
        );
        let mut quote_cache = QuoteCache::new();
        quote_cache.insert(instrument_id, previous);
        let mut funding_cache = AHashMap::new();
        let ts_init = UnixNanos::new(1_709_891_680_000_000_000);
        let clock = AtomicTime::new(false, ts_init);
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let fixture = if msg_type == "snapshot" {
            "ws_orderbook_snapshot.json"
        } else {
            "ws_orderbook_delta.json"
        };
        let mut msg: BybitWsOrderbookDepthMsg =
            serde_json::from_str(&load_test_json(fixture)).unwrap();
        msg.topic = topic.into();

        handle_ws_message(
            &BybitWsMessage::Orderbook(msg),
            &tx.into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            &clock,
        );

        if expect_book {
            let DataEvent::Data(Data::BookDeltas(deltas)) = rx.try_recv().unwrap() else {
                panic!("Expected book deltas");
            };
            assert_eq!(deltas.instrument_id, instrument_id);
            assert_eq!(
                deltas.deltas[0].action == BookAction::Clear,
                msg_type == "snapshot"
            );

            if msg_type == "delta" {
                assert_eq!(deltas.deltas.len(), 2);
                assert_eq!(deltas.deltas[1].action, BookAction::Delete);
                assert_eq!(deltas.deltas[1].order.size, Quantity::from("0.000"));
            }
        }

        if expect_quote {
            let expected = QuoteTick::new(
                instrument_id,
                Price::from("27450.00"),
                Price::from("27451.00"),
                Quantity::from("0.500"),
                Quantity::from("0.750"),
                UnixNanos::new(1_709_891_679_000_000_000),
                ts_init,
            );
            let DataEvent::Data(Data::Quote(quote)) = rx.try_recv().unwrap() else {
                panic!("Expected quote");
            };
            assert_eq!(quote, expected);
            assert_eq!(quote_cache.get(&instrument_id), Some(&expected));
        } else {
            assert_eq!(quote_cache.get(&instrument_id), Some(&previous));
        }
        rx.try_recv().unwrap_err();
    }

    #[rstest]
    fn test_handle_orderbook_message_no_sub_no_event() {
        let instrument = linear_instrument();
        let instruments = build_instruments(&[instrument]);
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();
        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_orderbook_snapshot.json");
        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
        let ws_msg = BybitWsMessage::Orderbook(msg);

        handle_ws_message(
            &ws_msg,
            &tx.into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        rx.try_recv().unwrap_err();
    }

    #[rstest]
    fn test_handle_ticker_linear_does_not_emit_quote() {
        let instrument = linear_instrument();
        let instrument_id = instrument.id();
        let instruments = build_instruments(&[instrument]);
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();

        let mut subs = AHashSet::new();
        subs.insert("quotes");
        ticker_subs.insert(instrument_id, subs);

        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_ticker_linear.json");
        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
        let ws_msg = BybitWsMessage::TickerLinear(msg);

        handle_ws_message(
            &ws_msg,
            &tx.into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        rx.try_recv().unwrap_err();
        assert!(quote_cache.is_empty());
    }

    #[rstest]
    fn test_handle_ticker_linear_funding_dedup() {
        let instrument = linear_instrument();
        let instrument_id = instrument.id();
        let instruments = build_instruments(&[instrument]);
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();

        let mut subs = AHashSet::new();
        subs.insert("funding");
        ticker_subs.insert(instrument_id, subs);

        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_ticker_linear.json");
        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
        let ws_msg = BybitWsMessage::TickerLinear(msg.clone());

        handle_ws_message(
            &ws_msg,
            &tx.clone().into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        let event = rx.try_recv().unwrap();
        assert!(matches!(event, DataEvent::FundingRate(_)));

        // Send same message again, funding unchanged so should be deduped
        let ws_msg2 = BybitWsMessage::TickerLinear(msg);
        handle_ws_message(
            &ws_msg2,
            &tx.into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        rx.try_recv().unwrap_err();
    }

    #[rstest]
    fn test_handle_ticker_linear_mark_and_index_prices() {
        let instrument = linear_instrument();
        let instrument_id = instrument.id();
        let instruments = build_instruments(&[instrument]);
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();

        let mut subs = AHashSet::new();
        subs.insert("mark_prices");
        subs.insert("index_prices");
        ticker_subs.insert(instrument_id, subs);

        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_ticker_linear.json");
        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
        let ws_msg = BybitWsMessage::TickerLinear(msg);

        handle_ws_message(
            &ws_msg,
            &tx.into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        let event1 = rx.try_recv().unwrap();
        assert!(matches!(event1, DataEvent::Data(Data::MarkPrice(_))));

        let event2 = rx.try_recv().unwrap();
        assert!(matches!(event2, DataEvent::Data(Data::IndexPrice(_))));
    }

    #[rstest]
    fn test_handle_reconnected_clears_caches() {
        let instruments = AHashMap::new();
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();
        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();

        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
        quote_cache.insert(
            instrument_id,
            QuoteTick::new(
                instrument_id,
                Price::from("100.00"),
                Price::from("101.00"),
                Quantity::from("1.0"),
                Quantity::from("1.0"),
                UnixNanos::default(),
                UnixNanos::default(),
            ),
        );
        funding_cache.insert(
            Ustr::from("BTCUSDT"),
            (
                Some("-0.001".to_string()),
                Some("1000".to_string()),
                Some("8".to_string()),
            ),
        );

        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();

        handle_ws_message(
            &BybitWsMessage::Reconnected,
            &tx.into(),
            &instruments,
            None,
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        assert!(quote_cache.is_empty());
        assert!(funding_cache.is_empty());
    }

    #[rstest]
    fn test_handle_ticker_option_greeks() {
        // Use the option instrument but key it by the ticker fixture symbol
        // (fixture instrument is ETH-26JUN26-16000-P, ticker fixture is BTC-6JAN23-17500-C)
        let instrument = option_instrument();
        let instrument_id = instrument.id();

        // Key the instrument by the fixture ticker symbol with OPTION suffix
        let ticker_key = Ustr::from("BTC-6JAN23-17500-C-OPTION");
        let mut instruments = AHashMap::new();
        instruments.insert(ticker_key, instrument);

        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();
        greeks_subs.insert(instrument_id);

        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_ticker_option.json");
        let msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
        let ws_msg = BybitWsMessage::TickerOption(msg);

        handle_ws_message(
            &ws_msg,
            &tx.into(),
            &instruments,
            Some(BybitProductType::Option),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        let event = rx.try_recv().unwrap();
        assert!(matches!(event, DataEvent::OptionGreeks(_)));
    }

    #[rstest]
    fn test_handle_execution_message_ignored_by_data() {
        let instruments = AHashMap::new();
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();
        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_account_order.json");
        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
            serde_json::from_str(&json).unwrap();
        let ws_msg = BybitWsMessage::AccountOrder(msg);

        handle_ws_message(
            &ws_msg,
            &tx.into(),
            &instruments,
            None,
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        rx.try_recv().unwrap_err();
    }

    #[rstest]
    fn test_instrument_resolution_with_product_type() {
        let instrument = linear_instrument();

        let mut map = AHashMap::new();
        map.insert(instrument.id().symbol.inner(), instrument.clone());

        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();
        trade_subs.insert(instrument.id());
        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_public_trade.json");
        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();

        // With None product_type, raw symbol "BTCUSDT" does not match "BTCUSDT-LINEAR"
        handle_ws_message(
            &BybitWsMessage::Trade(msg.clone()),
            &tx.clone().into(),
            &map,
            None,
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );
        rx.try_recv().unwrap_err();

        // With product_type=Linear, "BTCUSDT" -> "BTCUSDT-LINEAR" matches
        handle_ws_message(
            &BybitWsMessage::Trade(msg),
            &tx.into(),
            &map,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );

        let event = rx.try_recv().unwrap();
        assert!(matches!(event, DataEvent::Data(Data::Trade(_))));
    }

    #[rstest]
    fn test_handle_trade_filters_by_subscription() {
        let instrument = linear_instrument();
        let instruments = build_instruments(std::slice::from_ref(&instrument));
        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
            empty_subs();
        let mut quote_cache = QuoteCache::new();
        let mut funding_cache = AHashMap::new();
        let clock = get_atomic_clock_realtime();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let json = load_test_json("ws_public_trade.json");
        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();

        // Without subscription, trade should be filtered out
        handle_ws_message(
            &BybitWsMessage::Trade(msg.clone()),
            &tx.clone().into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );
        rx.try_recv().unwrap_err();

        // With subscription, trade should be emitted
        trade_subs.insert(instrument.id());
        handle_ws_message(
            &BybitWsMessage::Trade(msg),
            &tx.into(),
            &instruments,
            Some(BybitProductType::Linear),
            &trade_subs,
            &ticker_subs,
            &quote_subs,
            &book_depths,
            &greeks_subs,
            &bar_types,
            &mut quote_cache,
            &mut funding_cache,
            clock,
        );
        let event = rx.try_recv().unwrap();
        assert!(matches!(event, DataEvent::Data(Data::Trade(_))));
    }
}