Skip to main content

nautilus_bybit/
data.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Live market data client implementation for the Bybit adapter.
17
18use std::{
19    future::Future,
20    sync::{
21        Arc,
22        atomic::{AtomicBool, Ordering},
23    },
24};
25
26use ahash::{AHashMap, AHashSet};
27use anyhow::Context;
28use futures_util::{StreamExt, pin_mut};
29use nautilus_common::{
30    clients::DataClient,
31    live::{runner::get_data_event_sender, runtime::get_runtime},
32    messages::{
33        DataEvent,
34        data::{
35            BarsResponse, BookResponse, DataResponse, ForwardPricesResponse, FundingRatesResponse,
36            InstrumentResponse, InstrumentsResponse, RequestBars, RequestBookSnapshot,
37            RequestForwardPrices, RequestFundingRates, RequestInstrument, RequestInstruments,
38            RequestTrades, SubscribeBars, SubscribeBookDeltas, SubscribeFundingRates,
39            SubscribeIndexPrices, SubscribeInstrument, SubscribeInstrumentStatus,
40            SubscribeInstruments, SubscribeMarkPrices, SubscribeOptionGreeks, SubscribeQuotes,
41            SubscribeTrades, TradesResponse, UnsubscribeBars, UnsubscribeBookDeltas,
42            UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrument,
43            UnsubscribeInstrumentStatus, UnsubscribeInstruments, UnsubscribeMarkPrices,
44            UnsubscribeOptionGreeks, UnsubscribeQuotes, UnsubscribeTrades,
45        },
46    },
47};
48use nautilus_core::{
49    AtomicMap, AtomicSet,
50    datetime::datetime_to_unix_nanos,
51    time::{AtomicTime, get_atomic_clock_realtime},
52};
53use nautilus_model::{
54    data::{BarType, Data, ForwardPrice, OrderBookDeltas_API, QuoteTick},
55    enums::{BookType, MarketStatusAction},
56    identifiers::{ClientId, InstrumentId, Venue},
57    instruments::{Instrument, InstrumentAny},
58    orderbook::book::OrderBook,
59};
60use rust_decimal::Decimal;
61use tokio::{task::JoinHandle, time::Duration};
62use tokio_util::sync::CancellationToken;
63use ustr::Ustr;
64
65use crate::{
66    common::{
67        consts::{BYBIT_DEFAULT_ORDERBOOK_DEPTH, BYBIT_VENUE},
68        enums::BybitProductType,
69        instruments::diff_and_emit_instruments,
70        parse::{extract_raw_symbol, make_bybit_symbol},
71        status::{diff_and_emit_statuses, emit_status},
72        symbol::BybitSymbol,
73    },
74    config::BybitDataClientConfig,
75    http::client::BybitHttpClient,
76    websocket::{
77        client::BybitWebSocketClient,
78        messages::BybitWsMessage,
79        parse::{
80            parse_kline_topic, parse_millis_i64, parse_orderbook_deltas, parse_orderbook_quote,
81            parse_ticker_linear_funding, parse_ticker_linear_index_price,
82            parse_ticker_linear_mark_price, parse_ticker_linear_quote, parse_ticker_option_greeks,
83            parse_ticker_option_index_price, parse_ticker_option_mark_price,
84            parse_ticker_option_quote, parse_ws_kline_bar, parse_ws_trade_tick,
85        },
86    },
87};
88
89/// Live market data client for Bybit.
90#[derive(Debug)]
91pub struct BybitDataClient {
92    client_id: ClientId,
93    config: BybitDataClientConfig,
94    http_client: BybitHttpClient,
95    ws_clients: Vec<BybitWebSocketClient>,
96    is_connected: AtomicBool,
97    cancellation_token: CancellationToken,
98    tasks: Vec<JoinHandle<()>>,
99    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
100    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
101    book_depths: Arc<AtomicMap<InstrumentId, u32>>,
102    quote_depths: Arc<AtomicMap<InstrumentId, u32>>,
103    ticker_subs: Arc<AtomicMap<InstrumentId, AHashSet<&'static str>>>,
104    trade_subs: Arc<AtomicSet<InstrumentId>>,
105    option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
106    instrument_status_subs: Arc<AtomicSet<InstrumentId>>,
107    status_cache: Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
108    instrument_subs: Arc<AtomicSet<InstrumentId>>,
109    /// Venue-wide subscription: when set, definition updates are emitted for all instruments.
110    subscribe_all_instruments: Arc<AtomicBool>,
111    clock: &'static AtomicTime,
112}
113
114impl BybitDataClient {
115    /// Creates a new [`BybitDataClient`] instance.
116    ///
117    /// # Errors
118    ///
119    /// Returns an error if the client fails to initialize.
120    pub fn new(client_id: ClientId, config: BybitDataClientConfig) -> anyhow::Result<Self> {
121        let clock = get_atomic_clock_realtime();
122        let data_sender = get_data_event_sender();
123
124        let http_client = if let (Some(api_key), Some(api_secret)) =
125            (config.api_key.clone(), config.api_secret.clone())
126        {
127            BybitHttpClient::with_credentials(
128                api_key,
129                api_secret,
130                Some(config.http_base_url()),
131                config.http_timeout_secs,
132                config.max_retries,
133                config.retry_delay_initial_ms,
134                config.retry_delay_max_ms,
135                config.recv_window_ms,
136                config.proxy_url.clone(),
137            )?
138        } else {
139            BybitHttpClient::new(
140                Some(config.http_base_url()),
141                config.http_timeout_secs,
142                config.max_retries,
143                config.retry_delay_initial_ms,
144                config.retry_delay_max_ms,
145                config.recv_window_ms,
146                config.proxy_url.clone(),
147            )?
148        };
149
150        // Create a WebSocket client for each product type (default to Linear if empty)
151        let product_types = if config.product_types.is_empty() {
152            vec![BybitProductType::Linear]
153        } else {
154            config.product_types.clone()
155        };
156
157        let ws_clients: Vec<BybitWebSocketClient> = product_types
158            .iter()
159            .map(|product_type| {
160                BybitWebSocketClient::new_public_with(
161                    *product_type,
162                    config.environment,
163                    Some(config.ws_public_url_for(*product_type)),
164                    config.heartbeat_interval_secs,
165                    config.transport_backend,
166                    config.proxy_url.clone(),
167                )
168            })
169            .collect();
170
171        Ok(Self {
172            client_id,
173            config,
174            http_client,
175            ws_clients,
176            is_connected: AtomicBool::new(false),
177            cancellation_token: CancellationToken::new(),
178            tasks: Vec::new(),
179            data_sender,
180            instruments: Arc::new(AtomicMap::new()),
181            book_depths: Arc::new(AtomicMap::new()),
182            quote_depths: Arc::new(AtomicMap::new()),
183            ticker_subs: Arc::new(AtomicMap::new()),
184            trade_subs: Arc::new(AtomicSet::new()),
185            option_greeks_subs: Arc::new(AtomicSet::new()),
186            instrument_status_subs: Arc::new(AtomicSet::new()),
187            status_cache: Arc::new(AtomicMap::new()),
188            instrument_subs: Arc::new(AtomicSet::new()),
189            subscribe_all_instruments: Arc::new(AtomicBool::new(false)),
190            clock,
191        })
192    }
193
194    fn venue(&self) -> Venue {
195        *BYBIT_VENUE
196    }
197
198    fn get_ws_client_for_product(
199        &self,
200        product_type: BybitProductType,
201    ) -> Option<&BybitWebSocketClient> {
202        self.ws_clients
203            .iter()
204            .find(|ws| ws.product_type() == Some(product_type))
205    }
206
207    fn get_product_type_for_instrument(
208        &self,
209        instrument_id: InstrumentId,
210    ) -> Option<BybitProductType> {
211        let guard = self.instruments.load();
212        guard
213            .get(&instrument_id)
214            .and_then(|_| BybitProductType::from_suffix(instrument_id.symbol.as_str()))
215    }
216
217    fn spawn_ws<F>(&self, fut: F, context: &'static str)
218    where
219        F: Future<Output = anyhow::Result<()>> + Send + 'static,
220    {
221        get_runtime().spawn(async move {
222            if let Err(e) = fut.await {
223                log::error!("{context}: {e:?}");
224            }
225        });
226    }
227
228    fn spawn_instrument_polling(&mut self, product_types: &[BybitProductType], poll_secs: u64) {
229        let http = self.http_client.clone();
230        let sender = self.data_sender.clone();
231        let instruments = self.instruments.clone();
232        let status_cache = self.status_cache.clone();
233        let status_subs = self.instrument_status_subs.clone();
234        let instrument_subs = self.instrument_subs.clone();
235        let subscribe_all_instruments = self.subscribe_all_instruments.clone();
236        let cancel = self.cancellation_token.clone();
237        let clock = self.clock;
238        let product_types = product_types.to_vec();
239
240        let handle = get_runtime().spawn(async move {
241            let mut interval = tokio::time::interval(Duration::from_secs(poll_secs));
242            interval.tick().await; // Skip first immediate tick
243
244            loop {
245                tokio::select! {
246                    _ = interval.tick() => {
247                        let all_flag = subscribe_all_instruments.load(Ordering::Relaxed);
248                        let want_instruments = all_flag || !instrument_subs.is_empty();
249                        let want_statuses = !status_subs.is_empty();
250                        if !want_instruments && !want_statuses {
251                            continue;
252                        }
253
254                        let mut all_statuses = AHashMap::new();
255
256                        if want_instruments {
257                            let subs: Option<AHashSet<InstrumentId>> = if all_flag {
258                                None
259                            } else {
260                                Some((**instrument_subs.load()).clone())
261                            };
262                            let mut inst_cache = (**instruments.load()).clone();
263
264                            for &pt in &product_types {
265                                match http.request_instruments_with_statuses(pt).await {
266                                    Ok((fetched, statuses)) => {
267                                        diff_and_emit_instruments(
268                                            &fetched, &mut inst_cache, subs.as_ref(), &sender,
269                                        );
270
271                                        for (id, action) in statuses {
272                                            if inst_cache.contains_key(&id) {
273                                                all_statuses.insert(id, action);
274                                            }
275                                        }
276                                    }
277                                    Err(e) => {
278                                        log::warn!("Bybit instrument poll failed for {pt:?}: {e}");
279                                    }
280                                }
281                            }
282
283                            instruments.store(inst_cache);
284                        } else {
285                            for &pt in &product_types {
286                                match http.request_instrument_statuses(pt).await {
287                                    Ok(new_statuses) => {
288                                        let inst_guard = instruments.load();
289                                        for (id, action) in new_statuses {
290                                            if inst_guard.contains_key(&id) {
291                                                all_statuses.insert(id, action);
292                                            }
293                                        }
294                                    }
295                                    Err(e) => {
296                                        log::warn!("Bybit instrument status poll failed for {pt:?}: {e}");
297                                    }
298                                }
299                            }
300                        }
301
302                        if want_statuses {
303                            let ts = clock.get_time_ns();
304                            let mut cache = (**status_cache.load()).clone();
305                            let subs_guard = status_subs.load();
306                            diff_and_emit_statuses(
307                                &all_statuses, &mut cache, Some(&subs_guard), &sender, ts, ts,
308                            );
309                            status_cache.store(cache);
310                        }
311                    }
312                    () = cancel.cancelled() => {
313                        log::debug!("Bybit instrument polling task cancelled");
314                        break;
315                    }
316                }
317            }
318        });
319        self.tasks.push(handle);
320        log::debug!("Instrument polling started: interval={poll_secs}s");
321    }
322}
323
324fn send_data(sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>, data: Data) {
325    if let Err(e) = sender.send(DataEvent::Data(data)) {
326        log::error!("Failed to emit data event: {e}");
327    }
328}
329
330/// Cached funding state per symbol: (funding_rate, next_funding_time, funding_interval_hour).
331type FundingCacheEntry = (Option<String>, Option<String>, Option<String>);
332
333#[expect(clippy::too_many_arguments)]
334fn handle_ws_message(
335    message: &BybitWsMessage,
336    data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
337    instruments: &AHashMap<Ustr, InstrumentAny>,
338    product_type: Option<BybitProductType>,
339    trade_subs: &Arc<AtomicSet<InstrumentId>>,
340    ticker_subs: &Arc<AtomicMap<InstrumentId, AHashSet<&'static str>>>,
341    quote_depths: &Arc<AtomicMap<InstrumentId, u32>>,
342    book_depths: &Arc<AtomicMap<InstrumentId, u32>>,
343    option_greeks_subs: &Arc<AtomicSet<InstrumentId>>,
344    bar_types_cache: &Arc<AtomicMap<String, BarType>>,
345    quote_cache: &mut AHashMap<InstrumentId, QuoteTick>,
346    funding_cache: &mut AHashMap<Ustr, FundingCacheEntry>,
347    clock: &AtomicTime,
348) {
349    let ts_init = clock.get_time_ns();
350    let resolve = |raw_symbol: &Ustr| -> Option<&InstrumentAny> {
351        let key = product_type.map_or(*raw_symbol, |pt| make_bybit_symbol(raw_symbol, pt));
352        instruments.get(&key)
353    };
354
355    match message {
356        BybitWsMessage::Orderbook(msg) => {
357            let Some(instrument) = resolve(&msg.data.s) else {
358                log::warn!("Unknown symbol in orderbook update: {}", msg.data.s);
359                return;
360            };
361            let instrument_id = instrument.id();
362
363            // Emit deltas if subscribed to book
364            let has_book_sub = book_depths.contains_key(&instrument_id);
365
366            if has_book_sub {
367                match parse_orderbook_deltas(msg, instrument, ts_init) {
368                    Ok(deltas) => {
369                        send_data(data_sender, Data::Deltas(OrderBookDeltas_API::new(deltas)));
370                    }
371                    Err(e) => log::error!("Failed to parse orderbook deltas: {e}"),
372                }
373            }
374
375            // Emit quote from best bid/ask if subscribed
376            let has_quote_sub = quote_depths.contains_key(&instrument_id);
377            let has_ticker_quote_sub = ticker_subs
378                .load()
379                .get(&instrument_id)
380                .is_some_and(|s| s.contains("quotes"));
381
382            if has_quote_sub || has_ticker_quote_sub {
383                let last_quote = quote_cache.get(&instrument_id);
384                match parse_orderbook_quote(msg, instrument, last_quote, ts_init) {
385                    Ok(quote) => {
386                        quote_cache.insert(instrument_id, quote);
387                        send_data(data_sender, Data::Quote(quote));
388                    }
389                    Err(e) => log::error!("Failed to parse orderbook quote: {e}"),
390                }
391            }
392        }
393        BybitWsMessage::Trade(msg) => {
394            for trade in &msg.data {
395                let Some(instrument) = resolve(&trade.s) else {
396                    continue;
397                };
398                let instrument_id = instrument.id();
399                if !trade_subs.contains(&instrument_id) {
400                    continue;
401                }
402
403                match parse_ws_trade_tick(trade, instrument, ts_init) {
404                    Ok(tick) => send_data(data_sender, Data::Trade(tick)),
405                    Err(e) => log::error!("Failed to parse trade tick: {e}"),
406                }
407            }
408        }
409        BybitWsMessage::Kline(msg) => {
410            let Ok((_, raw_symbol)) = parse_kline_topic(msg.topic.as_str()) else {
411                log::warn!("Invalid kline topic: {}", msg.topic);
412                return;
413            };
414            let ustr_symbol = Ustr::from(raw_symbol);
415            let Some(instrument) = resolve(&ustr_symbol) else {
416                log::warn!("Unknown symbol in kline update: {raw_symbol}");
417                return;
418            };
419            let topic_key = msg.topic.as_str();
420            let Some(bar_type) = bar_types_cache.load().get(topic_key).copied() else {
421                log::warn!("No bar type cached for kline topic: {topic_key}");
422                return;
423            };
424
425            for kline in &msg.data {
426                if !kline.confirm {
427                    continue;
428                }
429
430                match parse_ws_kline_bar(kline, instrument, bar_type, true, ts_init) {
431                    Ok(bar) => send_data(data_sender, Data::Bar(bar)),
432                    Err(e) => log::error!("Failed to parse kline bar: {e}"),
433                }
434            }
435        }
436        BybitWsMessage::TickerLinear(msg) => {
437            let Some(instrument) = resolve(&msg.data.symbol) else {
438                log::warn!("Unknown symbol in ticker update: {}", msg.data.symbol);
439                return;
440            };
441            let instrument_id = instrument.id();
442            let subs = ticker_subs.load();
443            let sub_set = subs.get(&instrument_id);
444
445            if sub_set.is_some_and(|s| s.contains("quotes")) && msg.data.bid1_price.is_some() {
446                match parse_ticker_linear_quote(msg, instrument, ts_init) {
447                    Ok(quote) => {
448                        let last = quote_cache.get(&instrument_id);
449                        if last.is_none_or(|q| *q != quote) {
450                            quote_cache.insert(instrument_id, quote);
451                            send_data(data_sender, Data::Quote(quote));
452                        }
453                    }
454                    Err(e) => log::debug!("Skipping partial ticker update: {e}"),
455                }
456            }
457
458            let ts_event = match parse_millis_i64(msg.ts, "ticker.ts") {
459                Ok(ts) => ts,
460                Err(e) => {
461                    log::error!("Failed to parse ticker timestamp: {e}");
462                    return;
463                }
464            };
465
466            if sub_set.is_some_and(|s| s.contains("funding"))
467                && matches!(instrument, InstrumentAny::CryptoPerpetual(_))
468            {
469                let cache_entry = funding_cache
470                    .entry(msg.data.symbol)
471                    .or_insert((None, None, None));
472                let mut changed = false;
473
474                if let Some(rate) = &msg.data.funding_rate
475                    && cache_entry.0.as_ref() != Some(rate)
476                {
477                    cache_entry.0 = Some(rate.clone());
478                    changed = true;
479                }
480
481                if let Some(next_time) = &msg.data.next_funding_time
482                    && cache_entry.1.as_ref() != Some(next_time)
483                {
484                    cache_entry.1 = Some(next_time.clone());
485                    changed = true;
486                }
487
488                if let Some(interval) = &msg.data.funding_interval_hour {
489                    cache_entry.2 = Some(interval.clone());
490                }
491
492                if changed && cache_entry.0.is_some() {
493                    let mut merged = msg.data.clone();
494
495                    if merged.funding_rate.is_none() {
496                        merged.funding_rate.clone_from(&cache_entry.0);
497                    }
498
499                    if merged.next_funding_time.is_none() {
500                        merged.next_funding_time.clone_from(&cache_entry.1);
501                    }
502
503                    if merged.funding_interval_hour.is_none() {
504                        merged.funding_interval_hour.clone_from(&cache_entry.2);
505                    }
506
507                    match parse_ticker_linear_funding(&merged, instrument_id, ts_event, ts_init) {
508                        Ok(update) => {
509                            if let Err(e) = data_sender.send(DataEvent::FundingRate(update)) {
510                                log::error!("Failed to emit funding rate event: {e}");
511                            }
512                        }
513                        Err(e) => log::error!("Failed to parse ticker linear funding: {e}"),
514                    }
515                }
516            }
517
518            if sub_set.is_some_and(|s| s.contains("mark_prices")) && msg.data.mark_price.is_some() {
519                match parse_ticker_linear_mark_price(&msg.data, instrument, ts_event, ts_init) {
520                    Ok(update) => send_data(data_sender, Data::MarkPriceUpdate(update)),
521                    Err(e) => log::debug!("Skipping mark price update: {e}"),
522                }
523            }
524
525            if sub_set.is_some_and(|s| s.contains("index_prices")) && msg.data.index_price.is_some()
526            {
527                match parse_ticker_linear_index_price(&msg.data, instrument, ts_event, ts_init) {
528                    Ok(update) => send_data(data_sender, Data::IndexPriceUpdate(update)),
529                    Err(e) => log::debug!("Skipping index price update: {e}"),
530                }
531            }
532        }
533        BybitWsMessage::TickerOption(msg) => {
534            let Some(instrument) = resolve(&msg.data.symbol) else {
535                log::warn!(
536                    "Unknown symbol in option ticker update: {}",
537                    msg.data.symbol
538                );
539                return;
540            };
541            let instrument_id = instrument.id();
542            let subs = ticker_subs.load();
543            let sub_set = subs.get(&instrument_id);
544
545            if sub_set.is_some_and(|s| s.contains("quotes")) {
546                match parse_ticker_option_quote(msg, instrument, ts_init) {
547                    Ok(quote) => {
548                        let last = quote_cache.get(&instrument_id);
549                        if last.is_none_or(|q| *q != quote) {
550                            quote_cache.insert(instrument_id, quote);
551                            send_data(data_sender, Data::Quote(quote));
552                        }
553                    }
554                    Err(e) => log::error!("Failed to parse ticker option quote: {e}"),
555                }
556            }
557
558            if sub_set.is_some_and(|s| s.contains("mark_prices")) {
559                match parse_ticker_option_mark_price(msg, instrument, ts_init) {
560                    Ok(update) => send_data(data_sender, Data::MarkPriceUpdate(update)),
561                    Err(e) => log::error!("Failed to parse ticker option mark price: {e}"),
562                }
563            }
564
565            if sub_set.is_some_and(|s| s.contains("index_prices")) {
566                match parse_ticker_option_index_price(msg, instrument, ts_init) {
567                    Ok(update) => send_data(data_sender, Data::IndexPriceUpdate(update)),
568                    Err(e) => log::error!("Failed to parse ticker option index price: {e}"),
569                }
570            }
571
572            if option_greeks_subs.contains(&instrument_id) {
573                match parse_ticker_option_greeks(msg, instrument, ts_init) {
574                    Ok(greeks) => {
575                        if let Err(e) = data_sender.send(DataEvent::OptionGreeks(greeks)) {
576                            log::error!("Failed to send option greeks: {e}");
577                        }
578                    }
579                    Err(e) => log::error!("Failed to parse option greeks: {e}"),
580                }
581            }
582        }
583        BybitWsMessage::Reconnected => {
584            quote_cache.clear();
585            funding_cache.clear();
586            log::info!("WebSocket reconnected, cleared caches");
587        }
588        BybitWsMessage::Error(e) => {
589            log::warn!(
590                "Bybit WebSocket error: code={} message={}",
591                e.code,
592                e.message
593            );
594        }
595        BybitWsMessage::Auth(_)
596        | BybitWsMessage::OrderResponse(_)
597        | BybitWsMessage::AccountOrder(_)
598        | BybitWsMessage::AccountExecution(_)
599        | BybitWsMessage::AccountExecutionFast(_)
600        | BybitWsMessage::AccountWallet(_)
601        | BybitWsMessage::AccountPosition(_) => {}
602    }
603}
604
605fn upsert_instrument(
606    cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
607    instrument: InstrumentAny,
608) {
609    cache.insert(instrument.id(), instrument);
610}
611
612#[async_trait::async_trait(?Send)]
613impl DataClient for BybitDataClient {
614    fn client_id(&self) -> ClientId {
615        self.client_id
616    }
617
618    fn venue(&self) -> Option<Venue> {
619        Some(self.venue())
620    }
621
622    fn start(&mut self) -> anyhow::Result<()> {
623        log::info!(
624            "Started: client_id={}, product_types={:?}, environment={:?}, proxy_url={:?}",
625            self.client_id,
626            self.config.product_types,
627            self.config.environment,
628            self.config.proxy_url,
629        );
630        Ok(())
631    }
632
633    fn stop(&mut self) -> anyhow::Result<()> {
634        log::info!("Stopping {id}", id = self.client_id);
635        self.cancellation_token.cancel();
636        self.is_connected.store(false, Ordering::Relaxed);
637        Ok(())
638    }
639
640    fn reset(&mut self) -> anyhow::Result<()> {
641        log::debug!("Resetting {id}", id = self.client_id);
642        self.is_connected.store(false, Ordering::Relaxed);
643        self.cancellation_token = CancellationToken::new();
644        self.tasks.clear();
645        self.book_depths.store(AHashMap::new());
646        self.quote_depths.store(AHashMap::new());
647        self.ticker_subs.store(AHashMap::new());
648        self.option_greeks_subs.store(AHashSet::new());
649        self.instrument_status_subs.store(AHashSet::new());
650        self.status_cache.store(AHashMap::new());
651        self.instrument_subs.store(AHashSet::new());
652        self.subscribe_all_instruments
653            .store(false, Ordering::Relaxed);
654        Ok(())
655    }
656
657    fn dispose(&mut self) -> anyhow::Result<()> {
658        log::debug!("Disposing {id}", id = self.client_id);
659        self.stop()
660    }
661
662    async fn connect(&mut self) -> anyhow::Result<()> {
663        if self.is_connected() {
664            return Ok(());
665        }
666
667        let product_types = if self.config.product_types.is_empty() {
668            vec![BybitProductType::Linear]
669        } else {
670            self.config.product_types.clone()
671        };
672
673        let mut all_instruments = Vec::new();
674
675        for product_type in &product_types {
676            let fetched = self
677                .http_client
678                .request_instruments(*product_type, None, None)
679                .await
680                .with_context(|| {
681                    format!("failed to request Bybit instruments for {product_type:?}")
682                })?;
683
684            self.http_client.cache_instruments(&fetched);
685
686            self.instruments.rcu(|m| {
687                for instrument in &fetched {
688                    m.insert(instrument.id(), instrument.clone());
689                }
690            });
691
692            all_instruments.extend(fetched);
693        }
694
695        // Seed instrument status cache from initial fetch
696        if self
697            .config
698            .instrument_poll_interval_secs
699            .is_some_and(|s| s > 0)
700        {
701            // Collect all statuses first (without holding the lock across await)
702            let mut collected_statuses = Vec::new();
703
704            for product_type in &product_types {
705                match self
706                    .http_client
707                    .request_instrument_statuses(*product_type)
708                    .await
709                {
710                    Ok(statuses) => collected_statuses.push(statuses),
711                    Err(e) => {
712                        log::warn!(
713                            "Failed to seed instrument status cache for {product_type:?}: {e}"
714                        );
715                    }
716                }
717            }
718
719            let inst_guard = self.instruments.load();
720            let mut status_map = AHashMap::new();
721
722            for statuses in collected_statuses {
723                for (id, action) in statuses {
724                    if inst_guard.contains_key(&id) {
725                        status_map.insert(id, action);
726                    }
727                }
728            }
729            log::debug!(
730                "Seeded instrument status cache with {} entries",
731                status_map.len()
732            );
733            self.status_cache.store(status_map);
734        }
735
736        for instrument in all_instruments {
737            if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
738                log::warn!("Failed to send instrument: {e}");
739            }
740        }
741
742        // Build instruments map keyed by full Nautilus symbol for parsing
743        let instruments_by_symbol: Arc<AHashMap<Ustr, InstrumentAny>> = {
744            let guard = self.instruments.load();
745            let mut map = AHashMap::new();
746            for instrument in guard.values() {
747                map.insert(instrument.id().symbol.inner(), instrument.clone());
748            }
749            Arc::new(map)
750        };
751
752        for ws_client in &mut self.ws_clients {
753            ws_client
754                .connect()
755                .await
756                .context("failed to connect Bybit WebSocket")?;
757            ws_client
758                .wait_until_active(10.0)
759                .await
760                .context("WebSocket did not become active")?;
761
762            let stream = ws_client.stream();
763            let product_type = ws_client.product_type();
764            let sender = self.data_sender.clone();
765            let trade_subs = self.trade_subs.clone();
766            let ticker_subs = self.ticker_subs.clone();
767            let quote_depths = self.quote_depths.clone();
768            let book_depths = self.book_depths.clone();
769            let option_greeks_subs = self.option_greeks_subs.clone();
770            let bar_types_cache = ws_client.bar_types_cache().clone();
771            let instruments = Arc::clone(&instruments_by_symbol);
772            let clock = self.clock;
773            let cancel = self.cancellation_token.clone();
774
775            let handle = get_runtime().spawn(async move {
776                let mut quote_cache: AHashMap<InstrumentId, QuoteTick> = AHashMap::new();
777                let mut funding_cache: AHashMap<Ustr, FundingCacheEntry> = AHashMap::new();
778
779                pin_mut!(stream);
780
781                loop {
782                    tokio::select! {
783                        Some(message) = stream.next() => {
784                            handle_ws_message(
785                                &message,
786                                &sender,
787                                &instruments,
788                                product_type,
789                                &trade_subs,
790                                &ticker_subs,
791                                &quote_depths,
792                                &book_depths,
793                                &option_greeks_subs,
794                                &bar_types_cache,
795                                &mut quote_cache,
796                                &mut funding_cache,
797                                clock,
798                            );
799                        }
800                        () = cancel.cancelled() => {
801                            log::debug!("WebSocket stream task cancelled");
802                            break;
803                        }
804                    }
805                }
806            });
807            self.tasks.push(handle);
808        }
809
810        // Spawn the instrument/status polling task (serves both definition and status subs).
811        if let Some(poll_secs) = self.config.instrument_poll_interval_secs
812            && poll_secs > 0
813        {
814            self.spawn_instrument_polling(&product_types, poll_secs);
815        }
816
817        self.is_connected.store(true, Ordering::Release);
818        log::info!("Connected: client_id={}", self.client_id);
819        Ok(())
820    }
821
822    async fn disconnect(&mut self) -> anyhow::Result<()> {
823        if self.is_disconnected() {
824            return Ok(());
825        }
826
827        self.cancellation_token.cancel();
828
829        // Reinitialize token so reconnect can spawn new stream tasks
830        self.cancellation_token = CancellationToken::new();
831
832        for ws_client in &mut self.ws_clients {
833            if let Err(e) = ws_client.close().await {
834                log::warn!("Error closing WebSocket: {e:?}");
835            }
836        }
837
838        // Allow time for unsubscribe confirmations
839        tokio::time::sleep(Duration::from_millis(500)).await;
840
841        let handles: Vec<_> = self.tasks.drain(..).collect();
842        for handle in handles {
843            if let Err(e) = handle.await {
844                log::error!("Error joining WebSocket task: {e}");
845            }
846        }
847
848        self.book_depths.store(AHashMap::new());
849        self.quote_depths.store(AHashMap::new());
850        self.ticker_subs.store(AHashMap::new());
851        self.trade_subs.store(AHashSet::new());
852        self.option_greeks_subs.store(AHashSet::new());
853        self.instrument_status_subs.store(AHashSet::new());
854        self.status_cache.store(AHashMap::new());
855        self.instrument_subs.store(AHashSet::new());
856        self.subscribe_all_instruments
857            .store(false, Ordering::Relaxed);
858        self.is_connected.store(false, Ordering::Release);
859        log::info!("Disconnected: client_id={}", self.client_id);
860        Ok(())
861    }
862
863    fn is_connected(&self) -> bool {
864        self.is_connected.load(Ordering::Relaxed)
865    }
866
867    fn is_disconnected(&self) -> bool {
868        !self.is_connected()
869    }
870
871    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
872        if cmd.book_type != BookType::L2_MBP {
873            anyhow::bail!("Bybit only supports L2_MBP order book deltas");
874        }
875
876        let depth = cmd
877            .depth
878            .map_or(BYBIT_DEFAULT_ORDERBOOK_DEPTH, |d| d.get() as u32);
879
880        if !matches!(depth, 1 | 50 | 200 | 500) {
881            anyhow::bail!("invalid depth {depth}; valid values are 1, 50, 200, or 500");
882        }
883
884        let instrument_id = cmd.instrument_id;
885        let product_type = self
886            .get_product_type_for_instrument(instrument_id)
887            .unwrap_or(BybitProductType::Linear);
888
889        let ws = self
890            .get_ws_client_for_product(product_type)
891            .context("no WebSocket client for product type")?
892            .clone();
893
894        let book_depths = Arc::clone(&self.book_depths);
895
896        self.spawn_ws(
897            async move {
898                ws.subscribe_orderbook(instrument_id, depth)
899                    .await
900                    .context("orderbook subscription")?;
901                book_depths.insert(instrument_id, depth);
902                Ok(())
903            },
904            "order book delta subscription",
905        );
906
907        Ok(())
908    }
909
910    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
911        let instrument_id = cmd.instrument_id;
912        let product_type = self
913            .get_product_type_for_instrument(instrument_id)
914            .unwrap_or(BybitProductType::Linear);
915
916        let ws = self
917            .get_ws_client_for_product(product_type)
918            .context("no WebSocket client for product type")?
919            .clone();
920
921        // SPOT ticker channel doesn't include bid/ask, use orderbook depth=1
922        if product_type == BybitProductType::Spot {
923            let depth = 1;
924            self.quote_depths.insert(instrument_id, depth);
925
926            self.spawn_ws(
927                async move {
928                    ws.subscribe_orderbook(instrument_id, depth)
929                        .await
930                        .context("orderbook subscription for quotes")
931                },
932                "quote subscription (spot orderbook)",
933            );
934        } else {
935            let mut should_subscribe = false;
936            self.ticker_subs.rcu(|m| {
937                let entry = m.entry(instrument_id).or_default();
938                should_subscribe = entry.is_empty();
939                entry.insert("quotes");
940            });
941
942            if should_subscribe {
943                self.spawn_ws(
944                    async move {
945                        ws.subscribe_ticker(instrument_id)
946                            .await
947                            .context("ticker subscription")
948                    },
949                    "quote subscription",
950                );
951            }
952        }
953        Ok(())
954    }
955
956    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
957        let instrument_id = cmd.instrument_id;
958        let product_type = self
959            .get_product_type_for_instrument(instrument_id)
960            .unwrap_or(BybitProductType::Linear);
961
962        self.trade_subs.insert(instrument_id);
963
964        let ws = self
965            .get_ws_client_for_product(product_type)
966            .context("no WebSocket client for product type")?
967            .clone();
968
969        self.spawn_ws(
970            async move {
971                ws.subscribe_trades(instrument_id)
972                    .await
973                    .context("trades subscription")
974            },
975            "trade subscription",
976        );
977        Ok(())
978    }
979
980    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
981        let instrument_id = cmd.instrument_id;
982        let product_type = self
983            .get_product_type_for_instrument(instrument_id)
984            .unwrap_or(BybitProductType::Linear);
985
986        if product_type == BybitProductType::Spot || product_type == BybitProductType::Option {
987            anyhow::bail!("Funding rates not available for {product_type:?} instruments");
988        }
989
990        let guard = self.instruments.load();
991        if let Some(instrument) = guard.get(&instrument_id)
992            && !matches!(instrument, InstrumentAny::CryptoPerpetual(_))
993        {
994            anyhow::bail!("Funding rates only available for perpetuals, not {instrument_id}");
995        }
996
997        let mut should_subscribe = false;
998        self.ticker_subs.rcu(|m| {
999            let entry = m.entry(instrument_id).or_default();
1000            should_subscribe = entry.is_empty();
1001            entry.insert("funding");
1002        });
1003
1004        if should_subscribe {
1005            let ws = self
1006                .get_ws_client_for_product(product_type)
1007                .context("no WebSocket client for product type")?
1008                .clone();
1009
1010            self.spawn_ws(
1011                async move {
1012                    ws.subscribe_ticker(instrument_id)
1013                        .await
1014                        .context("ticker subscription for funding rates")
1015                },
1016                "funding rate subscription",
1017            );
1018        }
1019        Ok(())
1020    }
1021
1022    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
1023        let instrument_id = cmd.instrument_id;
1024        let product_type = self
1025            .get_product_type_for_instrument(instrument_id)
1026            .unwrap_or(BybitProductType::Linear);
1027
1028        if product_type == BybitProductType::Spot {
1029            anyhow::bail!("Mark prices not available for Spot instruments");
1030        }
1031
1032        let mut should_subscribe = false;
1033        self.ticker_subs.rcu(|m| {
1034            let entry = m.entry(instrument_id).or_default();
1035            should_subscribe = entry.is_empty();
1036            entry.insert("mark_prices");
1037        });
1038
1039        if should_subscribe {
1040            let ws = self
1041                .get_ws_client_for_product(product_type)
1042                .context("no WebSocket client for product type")?
1043                .clone();
1044
1045            self.spawn_ws(
1046                async move {
1047                    ws.subscribe_ticker(instrument_id)
1048                        .await
1049                        .context("ticker subscription for mark prices")
1050                },
1051                "mark price subscription",
1052            );
1053        }
1054        Ok(())
1055    }
1056
1057    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
1058        let instrument_id = cmd.instrument_id;
1059        let product_type = self
1060            .get_product_type_for_instrument(instrument_id)
1061            .unwrap_or(BybitProductType::Linear);
1062
1063        if product_type == BybitProductType::Spot {
1064            anyhow::bail!("Index prices not available for Spot instruments");
1065        }
1066
1067        let mut should_subscribe = false;
1068        self.ticker_subs.rcu(|m| {
1069            let entry = m.entry(instrument_id).or_default();
1070            should_subscribe = entry.is_empty();
1071            entry.insert("index_prices");
1072        });
1073
1074        if should_subscribe {
1075            let ws = self
1076                .get_ws_client_for_product(product_type)
1077                .context("no WebSocket client for product type")?
1078                .clone();
1079
1080            self.spawn_ws(
1081                async move {
1082                    ws.subscribe_ticker(instrument_id)
1083                        .await
1084                        .context("ticker subscription for index prices")
1085                },
1086                "index price subscription",
1087            );
1088        }
1089        Ok(())
1090    }
1091
1092    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
1093        let bar_type = cmd.bar_type;
1094        let instrument_id = bar_type.instrument_id();
1095        let product_type = self
1096            .get_product_type_for_instrument(instrument_id)
1097            .unwrap_or(BybitProductType::Linear);
1098
1099        if product_type == BybitProductType::Option {
1100            anyhow::bail!("Bybit does not support kline/bar data for options");
1101        }
1102
1103        let ws = self
1104            .get_ws_client_for_product(product_type)
1105            .context("no WebSocket client for product type")?
1106            .clone();
1107
1108        self.spawn_ws(
1109            async move {
1110                ws.subscribe_bars(bar_type)
1111                    .await
1112                    .context("bars subscription")
1113            },
1114            "bar subscription",
1115        );
1116        Ok(())
1117    }
1118
1119    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
1120        let instrument_id = cmd.instrument_id;
1121        let depth = self
1122            .book_depths
1123            .load()
1124            .get(&instrument_id)
1125            .copied()
1126            .unwrap_or(BYBIT_DEFAULT_ORDERBOOK_DEPTH);
1127        self.book_depths.remove(&instrument_id);
1128
1129        let product_type = self
1130            .get_product_type_for_instrument(instrument_id)
1131            .unwrap_or(BybitProductType::Linear);
1132
1133        // Check if spot quote subscription is using the same depth
1134        let quote_using_same_depth = self
1135            .quote_depths
1136            .load()
1137            .get(&instrument_id)
1138            .is_some_and(|&d| d == depth);
1139
1140        if quote_using_same_depth {
1141            return Ok(());
1142        }
1143
1144        let ws = self
1145            .get_ws_client_for_product(product_type)
1146            .context("no WebSocket client for product type")?
1147            .clone();
1148
1149        self.spawn_ws(
1150            async move {
1151                ws.unsubscribe_orderbook(instrument_id, depth)
1152                    .await
1153                    .context("orderbook unsubscribe")
1154            },
1155            "order book unsubscribe",
1156        );
1157        Ok(())
1158    }
1159
1160    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
1161        let instrument_id = cmd.instrument_id;
1162        let product_type = self
1163            .get_product_type_for_instrument(instrument_id)
1164            .unwrap_or(BybitProductType::Linear);
1165
1166        let ws = self
1167            .get_ws_client_for_product(product_type)
1168            .context("no WebSocket client for product type")?
1169            .clone();
1170
1171        if product_type == BybitProductType::Spot {
1172            let depth = self
1173                .quote_depths
1174                .load()
1175                .get(&instrument_id)
1176                .copied()
1177                .unwrap_or(1);
1178            self.quote_depths.remove(&instrument_id);
1179
1180            // Check if book deltas subscription is using the same depth
1181            let book_using_same_depth = self
1182                .book_depths
1183                .load()
1184                .get(&instrument_id)
1185                .is_some_and(|&d| d == depth);
1186
1187            if !book_using_same_depth {
1188                self.spawn_ws(
1189                    async move {
1190                        ws.unsubscribe_orderbook(instrument_id, depth)
1191                            .await
1192                            .context("orderbook unsubscribe for quotes")
1193                    },
1194                    "quote unsubscribe (spot orderbook)",
1195                );
1196            }
1197        } else {
1198            let mut should_unsubscribe = false;
1199            self.ticker_subs.rcu(|m| {
1200                if let Some(entry) = m.get_mut(&instrument_id) {
1201                    entry.remove("quotes");
1202                    if entry.is_empty() {
1203                        m.remove(&instrument_id);
1204                        should_unsubscribe = true;
1205                    } else {
1206                        should_unsubscribe = false;
1207                    }
1208                } else {
1209                    should_unsubscribe = false;
1210                }
1211            });
1212
1213            if should_unsubscribe {
1214                self.spawn_ws(
1215                    async move {
1216                        ws.unsubscribe_ticker(instrument_id)
1217                            .await
1218                            .context("ticker unsubscribe")
1219                    },
1220                    "quote unsubscribe",
1221                );
1222            }
1223        }
1224        Ok(())
1225    }
1226
1227    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
1228        let instrument_id = cmd.instrument_id;
1229        let product_type = self
1230            .get_product_type_for_instrument(instrument_id)
1231            .unwrap_or(BybitProductType::Linear);
1232
1233        self.trade_subs.remove(&instrument_id);
1234
1235        let ws = self
1236            .get_ws_client_for_product(product_type)
1237            .context("no WebSocket client for product type")?
1238            .clone();
1239
1240        self.spawn_ws(
1241            async move {
1242                ws.unsubscribe_trades(instrument_id)
1243                    .await
1244                    .context("trades unsubscribe")
1245            },
1246            "trade unsubscribe",
1247        );
1248        Ok(())
1249    }
1250
1251    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
1252        let instrument_id = cmd.instrument_id;
1253        let product_type = self
1254            .get_product_type_for_instrument(instrument_id)
1255            .unwrap_or(BybitProductType::Linear);
1256
1257        let mut should_unsubscribe = false;
1258        self.ticker_subs.rcu(|m| {
1259            if let Some(entry) = m.get_mut(&instrument_id) {
1260                entry.remove("funding");
1261                if entry.is_empty() {
1262                    m.remove(&instrument_id);
1263                    should_unsubscribe = true;
1264                } else {
1265                    should_unsubscribe = false;
1266                }
1267            } else {
1268                should_unsubscribe = false;
1269            }
1270        });
1271
1272        if should_unsubscribe {
1273            let ws = self
1274                .get_ws_client_for_product(product_type)
1275                .context("no WebSocket client for product type")?
1276                .clone();
1277
1278            self.spawn_ws(
1279                async move {
1280                    ws.unsubscribe_ticker(instrument_id)
1281                        .await
1282                        .context("ticker unsubscribe for funding rates")
1283                },
1284                "funding rate unsubscribe",
1285            );
1286        }
1287        Ok(())
1288    }
1289
1290    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
1291        let instrument_id = cmd.instrument_id;
1292        let product_type = self
1293            .get_product_type_for_instrument(instrument_id)
1294            .unwrap_or(BybitProductType::Linear);
1295
1296        let mut should_unsubscribe = false;
1297        self.ticker_subs.rcu(|m| {
1298            if let Some(entry) = m.get_mut(&instrument_id) {
1299                entry.remove("mark_prices");
1300                if entry.is_empty() {
1301                    m.remove(&instrument_id);
1302                    should_unsubscribe = true;
1303                } else {
1304                    should_unsubscribe = false;
1305                }
1306            } else {
1307                should_unsubscribe = false;
1308            }
1309        });
1310
1311        if should_unsubscribe {
1312            let ws = self
1313                .get_ws_client_for_product(product_type)
1314                .context("no WebSocket client for product type")?
1315                .clone();
1316
1317            self.spawn_ws(
1318                async move {
1319                    ws.unsubscribe_ticker(instrument_id)
1320                        .await
1321                        .context("ticker unsubscribe for mark prices")
1322                },
1323                "mark price unsubscribe",
1324            );
1325        }
1326        Ok(())
1327    }
1328
1329    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
1330        let instrument_id = cmd.instrument_id;
1331        let product_type = self
1332            .get_product_type_for_instrument(instrument_id)
1333            .unwrap_or(BybitProductType::Linear);
1334
1335        let mut should_unsubscribe = false;
1336        self.ticker_subs.rcu(|m| {
1337            if let Some(entry) = m.get_mut(&instrument_id) {
1338                entry.remove("index_prices");
1339                if entry.is_empty() {
1340                    m.remove(&instrument_id);
1341                    should_unsubscribe = true;
1342                } else {
1343                    should_unsubscribe = false;
1344                }
1345            } else {
1346                should_unsubscribe = false;
1347            }
1348        });
1349
1350        if should_unsubscribe {
1351            let ws = self
1352                .get_ws_client_for_product(product_type)
1353                .context("no WebSocket client for product type")?
1354                .clone();
1355
1356            self.spawn_ws(
1357                async move {
1358                    ws.unsubscribe_ticker(instrument_id)
1359                        .await
1360                        .context("ticker unsubscribe for index prices")
1361                },
1362                "index price unsubscribe",
1363            );
1364        }
1365        Ok(())
1366    }
1367
1368    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
1369        let bar_type = cmd.bar_type;
1370        let instrument_id = bar_type.instrument_id();
1371        let product_type = self
1372            .get_product_type_for_instrument(instrument_id)
1373            .unwrap_or(BybitProductType::Linear);
1374
1375        let ws = self
1376            .get_ws_client_for_product(product_type)
1377            .context("no WebSocket client for product type")?
1378            .clone();
1379
1380        self.spawn_ws(
1381            async move {
1382                ws.unsubscribe_bars(bar_type)
1383                    .await
1384                    .context("bars unsubscribe")
1385            },
1386            "bar unsubscribe",
1387        );
1388        Ok(())
1389    }
1390
1391    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
1392        let instrument_id = cmd.instrument_id;
1393        self.option_greeks_subs.insert(instrument_id);
1394
1395        let mut should_subscribe = false;
1396        self.ticker_subs.rcu(|m| {
1397            let entry = m.entry(instrument_id).or_default();
1398            should_subscribe = entry.is_empty();
1399            entry.insert("option_greeks");
1400        });
1401
1402        if should_subscribe {
1403            let product_type = self
1404                .get_product_type_for_instrument(instrument_id)
1405                .unwrap_or(BybitProductType::Option);
1406
1407            let ws = self
1408                .get_ws_client_for_product(product_type)
1409                .context("no WebSocket client for product type")?
1410                .clone();
1411
1412            self.spawn_ws(
1413                async move {
1414                    ws.subscribe_ticker(instrument_id)
1415                        .await
1416                        .context("ticker subscription for option greeks")
1417                },
1418                "option greeks subscription",
1419            );
1420        }
1421        Ok(())
1422    }
1423
1424    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
1425        let instrument_id = cmd.instrument_id;
1426        self.option_greeks_subs.remove(&instrument_id);
1427
1428        let mut should_unsubscribe = false;
1429        self.ticker_subs.rcu(|m| {
1430            if let Some(entry) = m.get_mut(&instrument_id) {
1431                entry.remove("option_greeks");
1432                if entry.is_empty() {
1433                    m.remove(&instrument_id);
1434                    should_unsubscribe = true;
1435                } else {
1436                    should_unsubscribe = false;
1437                }
1438            } else {
1439                should_unsubscribe = false;
1440            }
1441        });
1442
1443        if should_unsubscribe {
1444            let product_type = self
1445                .get_product_type_for_instrument(instrument_id)
1446                .unwrap_or(BybitProductType::Option);
1447
1448            let ws = self
1449                .get_ws_client_for_product(product_type)
1450                .context("no WebSocket client for product type")?
1451                .clone();
1452
1453            self.spawn_ws(
1454                async move {
1455                    ws.unsubscribe_ticker(instrument_id)
1456                        .await
1457                        .context("ticker unsubscribe for option greeks")
1458                },
1459                "option greeks unsubscribe",
1460            );
1461        }
1462        Ok(())
1463    }
1464
1465    fn subscribe_instruments(&mut self, cmd: SubscribeInstruments) -> anyhow::Result<()> {
1466        log::debug!(
1467            "subscribe_instruments: {venue} (definition updates detected via periodic instrument info polling)",
1468            venue = cmd.venue,
1469        );
1470        self.subscribe_all_instruments
1471            .store(true, Ordering::Relaxed);
1472        Ok(())
1473    }
1474
1475    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
1476        log::debug!(
1477            "subscribe_instrument: {id} (definition updates detected via periodic instrument info polling)",
1478            id = cmd.instrument_id,
1479        );
1480        self.instrument_subs.insert(cmd.instrument_id);
1481        Ok(())
1482    }
1483
1484    fn unsubscribe_instruments(&mut self, cmd: &UnsubscribeInstruments) -> anyhow::Result<()> {
1485        log::debug!("unsubscribe_instruments: {venue}", venue = cmd.venue);
1486        self.subscribe_all_instruments
1487            .store(false, Ordering::Relaxed);
1488        Ok(())
1489    }
1490
1491    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
1492        log::debug!("unsubscribe_instrument: {id}", id = cmd.instrument_id);
1493        self.instrument_subs.remove(&cmd.instrument_id);
1494        Ok(())
1495    }
1496
1497    fn subscribe_instrument_status(
1498        &mut self,
1499        cmd: SubscribeInstrumentStatus,
1500    ) -> anyhow::Result<()> {
1501        log::debug!(
1502            "subscribe_instrument_status: {id} (status changes detected via periodic instrument info polling)",
1503            id = cmd.instrument_id,
1504        );
1505        self.instrument_status_subs.insert(cmd.instrument_id);
1506
1507        if let Some(action) = self.status_cache.load().get(&cmd.instrument_id).copied() {
1508            let ts = self.clock.get_time_ns();
1509            emit_status(&self.data_sender, cmd.instrument_id, action, ts, ts);
1510        }
1511
1512        Ok(())
1513    }
1514
1515    fn unsubscribe_instrument_status(
1516        &mut self,
1517        cmd: &UnsubscribeInstrumentStatus,
1518    ) -> anyhow::Result<()> {
1519        log::debug!(
1520            "unsubscribe_instrument_status: {id}",
1521            id = cmd.instrument_id,
1522        );
1523        self.instrument_status_subs.remove(&cmd.instrument_id);
1524        Ok(())
1525    }
1526
1527    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
1528        let http = self.http_client.clone();
1529        let sender = self.data_sender.clone();
1530        let instruments_cache = self.instruments.clone();
1531        let request_id = request.request_id;
1532        let client_id = request.client_id.unwrap_or(self.client_id);
1533        let venue = self.venue();
1534        let start = request.start;
1535        let end = request.end;
1536        let params = request.params;
1537        let clock = self.clock;
1538        let start_nanos = datetime_to_unix_nanos(start);
1539        let end_nanos = datetime_to_unix_nanos(end);
1540        let product_types = if self.config.product_types.is_empty() {
1541            vec![BybitProductType::Linear]
1542        } else {
1543            self.config.product_types.clone()
1544        };
1545
1546        get_runtime().spawn(async move {
1547            let mut all_instruments = Vec::new();
1548
1549            for product_type in product_types {
1550                match http.request_instruments(product_type, None, None).await {
1551                    Ok(instruments) => {
1552                        for instrument in instruments {
1553                            upsert_instrument(&instruments_cache, instrument.clone());
1554                            all_instruments.push(instrument);
1555                        }
1556                    }
1557                    Err(e) => {
1558                        log::error!("Failed to fetch instruments for {product_type:?}: {e:?}");
1559                    }
1560                }
1561            }
1562
1563            let response = DataResponse::Instruments(InstrumentsResponse::new(
1564                request_id,
1565                client_id,
1566                venue,
1567                all_instruments,
1568                start_nanos,
1569                end_nanos,
1570                clock.get_time_ns(),
1571                params,
1572            ));
1573
1574            if let Err(e) = sender.send(DataEvent::Response(response)) {
1575                log::error!("Failed to send instruments response: {e}");
1576            }
1577        });
1578
1579        Ok(())
1580    }
1581
1582    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
1583        let http = self.http_client.clone();
1584        let sender = self.data_sender.clone();
1585        let instruments = self.instruments.clone();
1586        let instrument_id = request.instrument_id;
1587        let request_id = request.request_id;
1588        let client_id = request.client_id.unwrap_or(self.client_id);
1589        let start = request.start;
1590        let end = request.end;
1591        let params = request.params;
1592        let clock = self.clock;
1593        let start_nanos = datetime_to_unix_nanos(start);
1594        let end_nanos = datetime_to_unix_nanos(end);
1595
1596        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
1597            .unwrap_or(BybitProductType::Linear);
1598        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str()).to_string();
1599
1600        get_runtime().spawn(async move {
1601            match http
1602                .request_instruments(product_type, Some(raw_symbol), None)
1603                .await
1604                .context("fetch instrument from API")
1605            {
1606                Ok(fetched) => {
1607                    if let Some(instrument) = fetched.into_iter().find(|i| i.id() == instrument_id)
1608                    {
1609                        upsert_instrument(&instruments, instrument.clone());
1610
1611                        let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
1612                            request_id,
1613                            client_id,
1614                            instrument.id(),
1615                            instrument,
1616                            start_nanos,
1617                            end_nanos,
1618                            clock.get_time_ns(),
1619                            params,
1620                        )));
1621
1622                        if let Err(e) = sender.send(DataEvent::Response(response)) {
1623                            log::error!("Failed to send instrument response: {e}");
1624                        }
1625                    } else {
1626                        log::error!("Instrument not found: {instrument_id}");
1627                    }
1628                }
1629                Err(e) => log::error!("Instrument request failed: {e:?}"),
1630            }
1631        });
1632
1633        Ok(())
1634    }
1635
1636    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
1637        let http = self.http_client.clone();
1638        let sender = self.data_sender.clone();
1639        let instrument_id = request.instrument_id;
1640        let depth = request.depth.map(|n| n.get() as u32);
1641        let request_id = request.request_id;
1642        let client_id = request.client_id.unwrap_or(self.client_id);
1643        let params = request.params;
1644        let clock = self.clock;
1645
1646        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
1647            .unwrap_or(BybitProductType::Linear);
1648
1649        get_runtime().spawn(async move {
1650            match http
1651                .request_orderbook_snapshot(product_type, instrument_id, depth)
1652                .await
1653                .context("failed to request book snapshot from Bybit")
1654            {
1655                Ok(deltas) => {
1656                    let mut book = OrderBook::new(instrument_id, BookType::L2_MBP);
1657                    if let Err(e) = book.apply_deltas(&deltas) {
1658                        log::error!("Failed to apply book deltas for {instrument_id}: {e}");
1659                        return;
1660                    }
1661
1662                    let response = DataResponse::Book(BookResponse::new(
1663                        request_id,
1664                        client_id,
1665                        instrument_id,
1666                        book,
1667                        None,
1668                        None,
1669                        clock.get_time_ns(),
1670                        params,
1671                    ));
1672
1673                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1674                        log::error!("Failed to send book snapshot response: {e}");
1675                    }
1676                }
1677                Err(e) => log::error!("Book snapshot request failed for {instrument_id}: {e:?}"),
1678            }
1679        });
1680
1681        Ok(())
1682    }
1683
1684    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
1685        let http = self.http_client.clone();
1686        let sender = self.data_sender.clone();
1687        let instrument_id = request.instrument_id;
1688        let start = request.start;
1689        let end = request.end;
1690        let limit = request.limit.map(|n| n.get() as u32);
1691        let request_id = request.request_id;
1692        let client_id = request.client_id.unwrap_or(self.client_id);
1693        let params = request.params;
1694        let clock = self.clock;
1695        let start_nanos = datetime_to_unix_nanos(start);
1696        let end_nanos = datetime_to_unix_nanos(end);
1697
1698        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
1699            .unwrap_or(BybitProductType::Linear);
1700
1701        get_runtime().spawn(async move {
1702            match http
1703                .request_trades(product_type, instrument_id, limit)
1704                .await
1705                .context("failed to request trades from Bybit")
1706            {
1707                Ok(trades) => {
1708                    let response = DataResponse::Trades(TradesResponse::new(
1709                        request_id,
1710                        client_id,
1711                        instrument_id,
1712                        trades,
1713                        start_nanos,
1714                        end_nanos,
1715                        clock.get_time_ns(),
1716                        params,
1717                    ));
1718
1719                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1720                        log::error!("Failed to send trades response: {e}");
1721                    }
1722                }
1723                Err(e) => log::error!("Trade request failed: {e:?}"),
1724            }
1725        });
1726
1727        Ok(())
1728    }
1729
1730    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
1731        let http = self.http_client.clone();
1732        let sender = self.data_sender.clone();
1733        let bar_type = request.bar_type;
1734        let start = request.start;
1735        let end = request.end;
1736        let limit = request.limit.map(|n| n.get() as u32);
1737        let request_id = request.request_id;
1738        let client_id = request.client_id.unwrap_or(self.client_id);
1739        let params = request.params;
1740        let clock = self.clock;
1741        let start_nanos = datetime_to_unix_nanos(start);
1742        let end_nanos = datetime_to_unix_nanos(end);
1743
1744        let instrument_id = bar_type.instrument_id();
1745        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
1746            .unwrap_or(BybitProductType::Linear);
1747
1748        get_runtime().spawn(async move {
1749            match http
1750                .request_bars(product_type, bar_type, start, end, limit, true)
1751                .await
1752                .context("failed to request bars from Bybit")
1753            {
1754                Ok(bars) => {
1755                    let response = DataResponse::Bars(BarsResponse::new(
1756                        request_id,
1757                        client_id,
1758                        bar_type,
1759                        bars,
1760                        start_nanos,
1761                        end_nanos,
1762                        clock.get_time_ns(),
1763                        params,
1764                    ));
1765
1766                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1767                        log::error!("Failed to send bars response: {e}");
1768                    }
1769                }
1770                Err(e) => log::error!("Bar request failed: {e:?}"),
1771            }
1772        });
1773
1774        Ok(())
1775    }
1776
1777    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
1778        let http = self.http_client.clone();
1779        let sender = self.data_sender.clone();
1780        let instrument_id = request.instrument_id;
1781        let start = request.start;
1782        let end = request.end;
1783        let limit = request.limit.map(|n| n.get() as u32);
1784        let request_id = request.request_id;
1785        let client_id = request.client_id.unwrap_or(self.client_id);
1786        let params = request.params;
1787        let clock = self.clock;
1788        let start_nanos = datetime_to_unix_nanos(start);
1789        let end_nanos = datetime_to_unix_nanos(end);
1790
1791        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
1792            .unwrap_or(BybitProductType::Linear);
1793
1794        if product_type == BybitProductType::Spot || product_type == BybitProductType::Option {
1795            anyhow::bail!("Funding rates not available for {product_type} instruments");
1796        }
1797
1798        get_runtime().spawn(async move {
1799            match http
1800                .request_funding_rates(product_type, instrument_id, start, end, limit)
1801                .await
1802                .context("failed to request funding rates from Bybit")
1803            {
1804                Ok(funding_rates) => {
1805                    let response = DataResponse::FundingRates(FundingRatesResponse::new(
1806                        request_id,
1807                        client_id,
1808                        instrument_id,
1809                        funding_rates,
1810                        start_nanos,
1811                        end_nanos,
1812                        clock.get_time_ns(),
1813                        params,
1814                    ));
1815
1816                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1817                        log::error!("Failed to send funding rates response: {e}");
1818                    }
1819                }
1820                Err(e) => log::error!("Funding rates request failed for {instrument_id}: {e:?}"),
1821            }
1822        });
1823
1824        Ok(())
1825    }
1826
1827    fn request_forward_prices(&self, request: RequestForwardPrices) -> anyhow::Result<()> {
1828        let underlying = request.underlying.to_string();
1829        let instrument_id = request.instrument_id;
1830        let http_client = self.http_client.clone();
1831        let sender = self.data_sender.clone();
1832        let request_id = request.request_id;
1833        let client_id = self.client_id();
1834        let params = request.params;
1835        let clock = self.clock;
1836        let venue = *BYBIT_VENUE;
1837
1838        get_runtime().spawn(async move {
1839            let result = if let Some(inst_id) = instrument_id {
1840                // Single-instrument path: fetch ticker for one symbol
1841                let raw_symbol = extract_raw_symbol(inst_id.symbol.as_str()).to_string();
1842                log::debug!(
1843                    "Requesting forward price for {underlying} (single instrument: {raw_symbol})"
1844                );
1845
1846                let params = crate::http::query::BybitTickersParams {
1847                    category: BybitProductType::Option,
1848                    symbol: Some(raw_symbol.clone()),
1849                    base_coin: None,
1850                    exp_date: None,
1851                };
1852
1853                match http_client.request_option_tickers_raw_with_params(&params).await {
1854                    Ok(tickers) => {
1855                        let ts = clock.get_time_ns();
1856                        let forward_prices: Vec<ForwardPrice> = tickers
1857                            .into_iter()
1858                            .filter_map(|t| {
1859                                let up: Decimal = t.underlying_price.parse().ok()?;
1860                                if up.is_zero() {
1861                                    return None;
1862                                }
1863                                Some(ForwardPrice::new(inst_id, up, None, ts, ts))
1864                            })
1865                            .collect();
1866
1867                        log::debug!(
1868                            "Fetched {} forward price for {underlying} (single instrument: {raw_symbol})",
1869                            forward_prices.len(),
1870                        );
1871                        Ok((forward_prices, ts))
1872                    }
1873                    Err(e) => Err(e),
1874                }
1875            } else {
1876                // Bulk path: fetch all option tickers
1877                log::debug!("Requesting option forward prices for base_coin={underlying} (bulk)");
1878
1879                match http_client.request_option_tickers_raw(&underlying).await {
1880                    Ok(tickers) => {
1881                        let ts = clock.get_time_ns();
1882
1883                        // Deduplicate: all options at the same expiry share the same
1884                        // forward price. Extract expiry prefix (e.g. "BTC-28FEB26" from
1885                        // "BTC-28FEB26-65000-C") and keep only one entry per expiry.
1886                        let mut seen_expiries = std::collections::HashSet::new();
1887                        let forward_prices: Vec<ForwardPrice> = tickers
1888                            .into_iter()
1889                            .filter_map(|t| {
1890                                let up: Decimal = t.underlying_price.parse().ok()?;
1891                                if up.is_zero() {
1892                                    return None;
1893                                }
1894                                let parts: Vec<&str> = t.symbol.splitn(3, '-').collect();
1895                                let expiry_key = if parts.len() >= 2 {
1896                                    format!("{}-{}", parts[0], parts[1])
1897                                } else {
1898                                    t.symbol.to_string()
1899                                };
1900
1901                                if !seen_expiries.insert(expiry_key) {
1902                                    return None;
1903                                }
1904                                Some(ForwardPrice::new(
1905                                    BybitSymbol::new(format!("{}-OPTION", t.symbol))
1906                                        .map(|s| s.to_instrument_id())
1907                                        .ok()?,
1908                                    up,
1909                                    None,
1910                                    ts,
1911                                    ts,
1912                                ))
1913                            })
1914                            .collect();
1915
1916                        log::debug!(
1917                            "Fetched {} forward prices (per-expiry) for {underlying}",
1918                            forward_prices.len(),
1919                        );
1920                        Ok((forward_prices, ts))
1921                    }
1922                    Err(e) => Err(e),
1923                }
1924            };
1925
1926            match result {
1927                Ok((forward_prices, ts)) => {
1928                    let response = DataResponse::ForwardPrices(ForwardPricesResponse::new(
1929                        request_id,
1930                        client_id,
1931                        venue,
1932                        forward_prices,
1933                        ts,
1934                        params,
1935                    ));
1936
1937                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1938                        log::error!("Failed to send forward prices response: {e}");
1939                    }
1940                }
1941                Err(e) => {
1942                    log::error!("Forward prices request failed for {underlying}: {e:?}");
1943                }
1944            }
1945        });
1946
1947        Ok(())
1948    }
1949}
1950
1951#[cfg(test)]
1952mod tests {
1953    use std::sync::Arc;
1954
1955    use ahash::{AHashMap, AHashSet};
1956    use nautilus_common::messages::DataEvent;
1957    use nautilus_core::{AtomicMap, AtomicSet, UnixNanos, time::get_atomic_clock_realtime};
1958    use nautilus_model::{
1959        data::{BarType, Data, QuoteTick},
1960        enums::AggressorSide,
1961        identifiers::InstrumentId,
1962        instruments::{Instrument, InstrumentAny},
1963        types::{Price, Quantity},
1964    };
1965    use rstest::rstest;
1966    use ustr::Ustr;
1967
1968    use super::handle_ws_message;
1969    use crate::{
1970        common::{
1971            enums::BybitProductType,
1972            parse::{parse_linear_instrument, parse_option_instrument},
1973            testing::load_test_json,
1974        },
1975        http::models::{
1976            BybitFeeRate, BybitInstrumentLinearResponse, BybitInstrumentOptionResponse,
1977        },
1978        websocket::messages::{
1979            BybitWsMessage, BybitWsOrderbookDepthMsg, BybitWsTickerLinearMsg,
1980            BybitWsTickerOptionMsg, BybitWsTradeMsg,
1981        },
1982    };
1983
1984    fn sample_fee_rate(
1985        symbol: &str,
1986        taker: &str,
1987        maker: &str,
1988        base_coin: Option<&str>,
1989    ) -> BybitFeeRate {
1990        BybitFeeRate {
1991            symbol: Ustr::from(symbol),
1992            taker_fee_rate: taker.to_string(),
1993            maker_fee_rate: maker.to_string(),
1994            base_coin: base_coin.map(Ustr::from),
1995        }
1996    }
1997
1998    fn linear_instrument() -> InstrumentAny {
1999        let json = load_test_json("http_get_instruments_linear.json");
2000        let response: BybitInstrumentLinearResponse = serde_json::from_str(&json).unwrap();
2001        let instrument = &response.result.list[0];
2002        let fee_rate = sample_fee_rate("BTCUSDT", "0.00055", "0.0001", Some("BTC"));
2003        let ts = UnixNanos::new(1_700_000_000_000_000_000);
2004        parse_linear_instrument(instrument, &fee_rate, ts, ts).unwrap()
2005    }
2006
2007    fn option_instrument() -> InstrumentAny {
2008        let json = load_test_json("http_get_instruments_option.json");
2009        let response: BybitInstrumentOptionResponse = serde_json::from_str(&json).unwrap();
2010        let instrument = &response.result.list[0];
2011        let ts = UnixNanos::new(1_700_000_000_000_000_000);
2012        parse_option_instrument(instrument, None, ts, ts).unwrap()
2013    }
2014
2015    fn build_instruments(instruments: &[InstrumentAny]) -> AHashMap<Ustr, InstrumentAny> {
2016        let mut map = AHashMap::new();
2017        for inst in instruments {
2018            map.insert(inst.id().symbol.inner(), inst.clone());
2019        }
2020        map
2021    }
2022
2023    #[expect(clippy::type_complexity)]
2024    fn empty_subs() -> (
2025        Arc<AtomicSet<InstrumentId>>,
2026        Arc<AtomicMap<InstrumentId, AHashSet<&'static str>>>,
2027        Arc<AtomicMap<InstrumentId, u32>>,
2028        Arc<AtomicMap<InstrumentId, u32>>,
2029        Arc<AtomicSet<InstrumentId>>,
2030        Arc<AtomicMap<String, BarType>>,
2031    ) {
2032        (
2033            Arc::new(AtomicSet::new()),
2034            Arc::new(AtomicMap::new()),
2035            Arc::new(AtomicMap::new()),
2036            Arc::new(AtomicMap::new()),
2037            Arc::new(AtomicSet::new()),
2038            Arc::new(AtomicMap::new()),
2039        )
2040    }
2041
2042    #[rstest]
2043    fn test_handle_trade_message_emits_trade_tick() {
2044        let instrument = linear_instrument();
2045        let instruments = build_instruments(std::slice::from_ref(&instrument));
2046        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2047            empty_subs();
2048        trade_subs.insert(instrument.id());
2049        let mut quote_cache = AHashMap::new();
2050        let mut funding_cache = AHashMap::new();
2051        let clock = get_atomic_clock_realtime();
2052
2053        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2054
2055        let json = load_test_json("ws_public_trade.json");
2056        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
2057        let ws_msg = BybitWsMessage::Trade(msg);
2058
2059        handle_ws_message(
2060            &ws_msg,
2061            &tx,
2062            &instruments,
2063            Some(BybitProductType::Linear),
2064            &trade_subs,
2065            &ticker_subs,
2066            &quote_depths,
2067            &book_depths,
2068            &greeks_subs,
2069            &bar_types,
2070            &mut quote_cache,
2071            &mut funding_cache,
2072            clock,
2073        );
2074
2075        let event = rx.try_recv().unwrap();
2076        match event {
2077            DataEvent::Data(Data::Trade(tick)) => {
2078                assert_eq!(tick.instrument_id, instrument.id());
2079                assert_eq!(tick.price, instrument.make_price(27451.00));
2080                assert_eq!(tick.size, instrument.make_qty(0.010, None));
2081                assert_eq!(tick.aggressor_side, AggressorSide::Buyer);
2082            }
2083            other => panic!("Expected Trade data event, found {other:?}"),
2084        }
2085    }
2086
2087    #[rstest]
2088    fn test_handle_trade_message_unknown_symbol_no_event() {
2089        let instruments = AHashMap::new();
2090        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2091            empty_subs();
2092        let mut quote_cache = AHashMap::new();
2093        let mut funding_cache = AHashMap::new();
2094        let clock = get_atomic_clock_realtime();
2095
2096        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2097
2098        let json = load_test_json("ws_public_trade.json");
2099        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
2100        let ws_msg = BybitWsMessage::Trade(msg);
2101
2102        handle_ws_message(
2103            &ws_msg,
2104            &tx,
2105            &instruments,
2106            Some(BybitProductType::Linear),
2107            &trade_subs,
2108            &ticker_subs,
2109            &quote_depths,
2110            &book_depths,
2111            &greeks_subs,
2112            &bar_types,
2113            &mut quote_cache,
2114            &mut funding_cache,
2115            clock,
2116        );
2117
2118        rx.try_recv().unwrap_err();
2119    }
2120
2121    #[rstest]
2122    fn test_handle_orderbook_message_emits_deltas_and_quote() {
2123        let instrument = linear_instrument();
2124        let instrument_id = instrument.id();
2125        let instruments = build_instruments(&[instrument]);
2126        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2127            empty_subs();
2128
2129        book_depths.insert(instrument_id, 1);
2130        quote_depths.insert(instrument_id, 1);
2131
2132        let mut quote_cache = AHashMap::new();
2133        let mut funding_cache = AHashMap::new();
2134        let clock = get_atomic_clock_realtime();
2135
2136        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2137
2138        let json = load_test_json("ws_orderbook_snapshot.json");
2139        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
2140        let ws_msg = BybitWsMessage::Orderbook(msg);
2141
2142        handle_ws_message(
2143            &ws_msg,
2144            &tx,
2145            &instruments,
2146            Some(BybitProductType::Linear),
2147            &trade_subs,
2148            &ticker_subs,
2149            &quote_depths,
2150            &book_depths,
2151            &greeks_subs,
2152            &bar_types,
2153            &mut quote_cache,
2154            &mut funding_cache,
2155            clock,
2156        );
2157
2158        let event1 = rx.try_recv().unwrap();
2159        assert!(matches!(event1, DataEvent::Data(Data::Deltas(_))));
2160
2161        let event2 = rx.try_recv().unwrap();
2162        assert!(matches!(event2, DataEvent::Data(Data::Quote(_))));
2163    }
2164
2165    #[rstest]
2166    fn test_handle_orderbook_message_no_sub_no_event() {
2167        let instrument = linear_instrument();
2168        let instruments = build_instruments(&[instrument]);
2169        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2170            empty_subs();
2171        let mut quote_cache = AHashMap::new();
2172        let mut funding_cache = AHashMap::new();
2173        let clock = get_atomic_clock_realtime();
2174
2175        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2176
2177        let json = load_test_json("ws_orderbook_snapshot.json");
2178        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
2179        let ws_msg = BybitWsMessage::Orderbook(msg);
2180
2181        handle_ws_message(
2182            &ws_msg,
2183            &tx,
2184            &instruments,
2185            Some(BybitProductType::Linear),
2186            &trade_subs,
2187            &ticker_subs,
2188            &quote_depths,
2189            &book_depths,
2190            &greeks_subs,
2191            &bar_types,
2192            &mut quote_cache,
2193            &mut funding_cache,
2194            clock,
2195        );
2196
2197        rx.try_recv().unwrap_err();
2198    }
2199
2200    #[rstest]
2201    fn test_handle_ticker_linear_emits_quote() {
2202        let instrument = linear_instrument();
2203        let instrument_id = instrument.id();
2204        let instruments = build_instruments(&[instrument]);
2205        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2206            empty_subs();
2207
2208        let mut subs = AHashSet::new();
2209        subs.insert("quotes");
2210        ticker_subs.insert(instrument_id, subs);
2211
2212        let mut quote_cache = AHashMap::new();
2213        let mut funding_cache = AHashMap::new();
2214        let clock = get_atomic_clock_realtime();
2215
2216        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2217
2218        let json = load_test_json("ws_ticker_linear.json");
2219        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
2220        let ws_msg = BybitWsMessage::TickerLinear(msg);
2221
2222        handle_ws_message(
2223            &ws_msg,
2224            &tx,
2225            &instruments,
2226            Some(BybitProductType::Linear),
2227            &trade_subs,
2228            &ticker_subs,
2229            &quote_depths,
2230            &book_depths,
2231            &greeks_subs,
2232            &bar_types,
2233            &mut quote_cache,
2234            &mut funding_cache,
2235            clock,
2236        );
2237
2238        let event = rx.try_recv().unwrap();
2239        assert!(matches!(event, DataEvent::Data(Data::Quote(_))));
2240        assert!(quote_cache.contains_key(&instrument_id));
2241    }
2242
2243    #[rstest]
2244    fn test_handle_ticker_linear_funding_dedup() {
2245        let instrument = linear_instrument();
2246        let instrument_id = instrument.id();
2247        let instruments = build_instruments(&[instrument]);
2248        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2249            empty_subs();
2250
2251        let mut subs = AHashSet::new();
2252        subs.insert("funding");
2253        ticker_subs.insert(instrument_id, subs);
2254
2255        let mut quote_cache = AHashMap::new();
2256        let mut funding_cache = AHashMap::new();
2257        let clock = get_atomic_clock_realtime();
2258
2259        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2260
2261        let json = load_test_json("ws_ticker_linear.json");
2262        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
2263        let ws_msg = BybitWsMessage::TickerLinear(msg.clone());
2264
2265        handle_ws_message(
2266            &ws_msg,
2267            &tx,
2268            &instruments,
2269            Some(BybitProductType::Linear),
2270            &trade_subs,
2271            &ticker_subs,
2272            &quote_depths,
2273            &book_depths,
2274            &greeks_subs,
2275            &bar_types,
2276            &mut quote_cache,
2277            &mut funding_cache,
2278            clock,
2279        );
2280
2281        let event = rx.try_recv().unwrap();
2282        assert!(matches!(event, DataEvent::FundingRate(_)));
2283
2284        // Send same message again, funding unchanged so should be deduped
2285        let ws_msg2 = BybitWsMessage::TickerLinear(msg);
2286        handle_ws_message(
2287            &ws_msg2,
2288            &tx,
2289            &instruments,
2290            Some(BybitProductType::Linear),
2291            &trade_subs,
2292            &ticker_subs,
2293            &quote_depths,
2294            &book_depths,
2295            &greeks_subs,
2296            &bar_types,
2297            &mut quote_cache,
2298            &mut funding_cache,
2299            clock,
2300        );
2301
2302        rx.try_recv().unwrap_err();
2303    }
2304
2305    #[rstest]
2306    fn test_handle_ticker_linear_mark_and_index_prices() {
2307        let instrument = linear_instrument();
2308        let instrument_id = instrument.id();
2309        let instruments = build_instruments(&[instrument]);
2310        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2311            empty_subs();
2312
2313        let mut subs = AHashSet::new();
2314        subs.insert("mark_prices");
2315        subs.insert("index_prices");
2316        ticker_subs.insert(instrument_id, subs);
2317
2318        let mut quote_cache = AHashMap::new();
2319        let mut funding_cache = AHashMap::new();
2320        let clock = get_atomic_clock_realtime();
2321
2322        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2323
2324        let json = load_test_json("ws_ticker_linear.json");
2325        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
2326        let ws_msg = BybitWsMessage::TickerLinear(msg);
2327
2328        handle_ws_message(
2329            &ws_msg,
2330            &tx,
2331            &instruments,
2332            Some(BybitProductType::Linear),
2333            &trade_subs,
2334            &ticker_subs,
2335            &quote_depths,
2336            &book_depths,
2337            &greeks_subs,
2338            &bar_types,
2339            &mut quote_cache,
2340            &mut funding_cache,
2341            clock,
2342        );
2343
2344        let event1 = rx.try_recv().unwrap();
2345        assert!(matches!(event1, DataEvent::Data(Data::MarkPriceUpdate(_))));
2346
2347        let event2 = rx.try_recv().unwrap();
2348        assert!(matches!(event2, DataEvent::Data(Data::IndexPriceUpdate(_))));
2349    }
2350
2351    #[rstest]
2352    fn test_handle_reconnected_clears_caches() {
2353        let instruments = AHashMap::new();
2354        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2355            empty_subs();
2356        let mut quote_cache = AHashMap::new();
2357        let mut funding_cache = AHashMap::new();
2358        let clock = get_atomic_clock_realtime();
2359
2360        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
2361        quote_cache.insert(
2362            instrument_id,
2363            QuoteTick::new(
2364                instrument_id,
2365                Price::from("100.00"),
2366                Price::from("101.00"),
2367                Quantity::from("1.0"),
2368                Quantity::from("1.0"),
2369                UnixNanos::default(),
2370                UnixNanos::default(),
2371            ),
2372        );
2373        funding_cache.insert(
2374            Ustr::from("BTCUSDT"),
2375            (
2376                Some("-0.001".to_string()),
2377                Some("1000".to_string()),
2378                Some("8".to_string()),
2379            ),
2380        );
2381
2382        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
2383
2384        handle_ws_message(
2385            &BybitWsMessage::Reconnected,
2386            &tx,
2387            &instruments,
2388            None,
2389            &trade_subs,
2390            &ticker_subs,
2391            &quote_depths,
2392            &book_depths,
2393            &greeks_subs,
2394            &bar_types,
2395            &mut quote_cache,
2396            &mut funding_cache,
2397            clock,
2398        );
2399
2400        assert!(quote_cache.is_empty());
2401        assert!(funding_cache.is_empty());
2402    }
2403
2404    #[rstest]
2405    fn test_handle_ticker_option_greeks() {
2406        // Use the option instrument but key it by the ticker fixture symbol
2407        // (fixture instrument is ETH-26JUN26-16000-P, ticker fixture is BTC-6JAN23-17500-C)
2408        let instrument = option_instrument();
2409        let instrument_id = instrument.id();
2410
2411        // Key the instrument by the fixture ticker symbol with OPTION suffix
2412        let ticker_key = Ustr::from("BTC-6JAN23-17500-C-OPTION");
2413        let mut instruments = AHashMap::new();
2414        instruments.insert(ticker_key, instrument);
2415
2416        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2417            empty_subs();
2418        greeks_subs.insert(instrument_id);
2419
2420        let mut quote_cache = AHashMap::new();
2421        let mut funding_cache = AHashMap::new();
2422        let clock = get_atomic_clock_realtime();
2423
2424        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2425
2426        let json = load_test_json("ws_ticker_option.json");
2427        let msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
2428        let ws_msg = BybitWsMessage::TickerOption(msg);
2429
2430        handle_ws_message(
2431            &ws_msg,
2432            &tx,
2433            &instruments,
2434            Some(BybitProductType::Option),
2435            &trade_subs,
2436            &ticker_subs,
2437            &quote_depths,
2438            &book_depths,
2439            &greeks_subs,
2440            &bar_types,
2441            &mut quote_cache,
2442            &mut funding_cache,
2443            clock,
2444        );
2445
2446        let event = rx.try_recv().unwrap();
2447        assert!(matches!(event, DataEvent::OptionGreeks(_)));
2448    }
2449
2450    #[rstest]
2451    fn test_handle_execution_message_ignored_by_data() {
2452        let instruments = AHashMap::new();
2453        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2454            empty_subs();
2455        let mut quote_cache = AHashMap::new();
2456        let mut funding_cache = AHashMap::new();
2457        let clock = get_atomic_clock_realtime();
2458
2459        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2460
2461        let json = load_test_json("ws_account_order.json");
2462        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
2463            serde_json::from_str(&json).unwrap();
2464        let ws_msg = BybitWsMessage::AccountOrder(msg);
2465
2466        handle_ws_message(
2467            &ws_msg,
2468            &tx,
2469            &instruments,
2470            None,
2471            &trade_subs,
2472            &ticker_subs,
2473            &quote_depths,
2474            &book_depths,
2475            &greeks_subs,
2476            &bar_types,
2477            &mut quote_cache,
2478            &mut funding_cache,
2479            clock,
2480        );
2481
2482        rx.try_recv().unwrap_err();
2483    }
2484
2485    #[rstest]
2486    fn test_instrument_resolution_with_product_type() {
2487        let instrument = linear_instrument();
2488
2489        let mut map = AHashMap::new();
2490        map.insert(instrument.id().symbol.inner(), instrument.clone());
2491
2492        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2493            empty_subs();
2494        trade_subs.insert(instrument.id());
2495        let mut quote_cache = AHashMap::new();
2496        let mut funding_cache = AHashMap::new();
2497        let clock = get_atomic_clock_realtime();
2498        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2499
2500        let json = load_test_json("ws_public_trade.json");
2501        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
2502
2503        // With None product_type, raw symbol "BTCUSDT" does not match "BTCUSDT-LINEAR"
2504        handle_ws_message(
2505            &BybitWsMessage::Trade(msg.clone()),
2506            &tx,
2507            &map,
2508            None,
2509            &trade_subs,
2510            &ticker_subs,
2511            &quote_depths,
2512            &book_depths,
2513            &greeks_subs,
2514            &bar_types,
2515            &mut quote_cache,
2516            &mut funding_cache,
2517            clock,
2518        );
2519        rx.try_recv().unwrap_err();
2520
2521        // With product_type=Linear, "BTCUSDT" -> "BTCUSDT-LINEAR" matches
2522        handle_ws_message(
2523            &BybitWsMessage::Trade(msg),
2524            &tx,
2525            &map,
2526            Some(BybitProductType::Linear),
2527            &trade_subs,
2528            &ticker_subs,
2529            &quote_depths,
2530            &book_depths,
2531            &greeks_subs,
2532            &bar_types,
2533            &mut quote_cache,
2534            &mut funding_cache,
2535            clock,
2536        );
2537
2538        let event = rx.try_recv().unwrap();
2539        assert!(matches!(event, DataEvent::Data(Data::Trade(_))));
2540    }
2541
2542    #[rstest]
2543    fn test_handle_trade_filters_by_subscription() {
2544        let instrument = linear_instrument();
2545        let instruments = build_instruments(std::slice::from_ref(&instrument));
2546        let (trade_subs, ticker_subs, quote_depths, book_depths, greeks_subs, bar_types) =
2547            empty_subs();
2548        let mut quote_cache = AHashMap::new();
2549        let mut funding_cache = AHashMap::new();
2550        let clock = get_atomic_clock_realtime();
2551        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2552
2553        let json = load_test_json("ws_public_trade.json");
2554        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
2555
2556        // Without subscription, trade should be filtered out
2557        handle_ws_message(
2558            &BybitWsMessage::Trade(msg.clone()),
2559            &tx,
2560            &instruments,
2561            Some(BybitProductType::Linear),
2562            &trade_subs,
2563            &ticker_subs,
2564            &quote_depths,
2565            &book_depths,
2566            &greeks_subs,
2567            &bar_types,
2568            &mut quote_cache,
2569            &mut funding_cache,
2570            clock,
2571        );
2572        rx.try_recv().unwrap_err();
2573
2574        // With subscription, trade should be emitted
2575        trade_subs.insert(instrument.id());
2576        handle_ws_message(
2577            &BybitWsMessage::Trade(msg),
2578            &tx,
2579            &instruments,
2580            Some(BybitProductType::Linear),
2581            &trade_subs,
2582            &ticker_subs,
2583            &quote_depths,
2584            &book_depths,
2585            &greeks_subs,
2586            &bar_types,
2587            &mut quote_cache,
2588            &mut funding_cache,
2589            clock,
2590        );
2591        let event = rx.try_recv().unwrap();
2592        assert!(matches!(event, DataEvent::Data(Data::Trade(_))));
2593    }
2594}