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    time::Duration,
25};
26
27use ahash::{AHashMap, AHashSet};
28use anyhow::Context;
29use dashmap::DashMap;
30use futures_util::{StreamExt, pin_mut};
31use nautilus_common::{
32    clients::{DataClient, execute_bar_subscriptions, execute_instrument_subscriptions},
33    live::{dst::time::Instant, runner::get_data_event_sender, sender::EventSender},
34    messages::{
35        DataEvent,
36        data::{
37            BarsResponse, BookResponse, DataResponse, FundingRatesResponse, InstrumentResponse,
38            InstrumentsResponse, OptionChainReferencePriceResponse, RequestBars,
39            RequestBookSnapshot, RequestFundingRates, RequestInstrument, RequestInstruments,
40            RequestOptionChainReferencePrice, RequestTrades, SubscribeBars, SubscribeBookDeltas,
41            SubscribeCustomData, SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument,
42            SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices,
43            SubscribeOptionGreeks, SubscribeQuotes, SubscribeTrades, TradesResponse,
44            UnsubscribeBars, UnsubscribeBookDeltas, UnsubscribeCustomData, UnsubscribeFundingRates,
45            UnsubscribeIndexPrices, UnsubscribeInstrument, UnsubscribeInstrumentStatus,
46            UnsubscribeInstruments, UnsubscribeMarkPrices, UnsubscribeOptionGreeks,
47            UnsubscribeQuotes, UnsubscribeTrades,
48        },
49    },
50};
51use nautilus_core::{
52    AtomicMap, AtomicSet, UnixNanos,
53    datetime::datetime_to_unix_nanos,
54    time::{AtomicTime, get_atomic_clock_realtime},
55};
56use nautilus_live::{
57    SocketControlFactory,
58    task::{TaskGroup, TaskGroupGuard, TaskSpawner},
59};
60use nautilus_model::{
61    data::{BarType, CustomData, Data, DataType, QuoteTick},
62    enums::{BookType, MarketStatusAction},
63    identifiers::{ClientId, InstrumentId, Venue},
64    instruments::{Instrument, InstrumentAny},
65    orderbook::book::OrderBook,
66    types::Price,
67};
68use rust_decimal::Decimal;
69use tokio_util::sync::CancellationToken;
70use ustr::Ustr;
71
72use crate::{
73    book::{
74        BookSequenceOutcome,
75        recovery::{reset_on_reconnect, spawn_subscription_task, start_recovery},
76        sync::BookSyncTracker,
77    },
78    common::{
79        consts::{
80            BYBIT_BOOK_DEPTHS, BYBIT_DEFAULT_ORDERBOOK_DEPTH, BYBIT_QUOTE_DEPTH, BYBIT_VENUE,
81        },
82        enums::BybitProductType,
83        instruments::diff_and_emit_instruments,
84        parse::{extract_raw_symbol, make_bybit_symbol},
85        status::{diff_and_emit_statuses, emit_status},
86    },
87    config::BybitDataClientConfig,
88    data_types::register_bybit_custom_data,
89    http::client::BybitHttpClient,
90    websocket::{
91        client::BybitWebSocketClient,
92        messages::{BybitWsMessage, BybitWsOrderbookDepthMsg},
93        parse::{
94            parse_kline_topic, parse_millis_i64, parse_orderbook_deltas, parse_orderbook_quote,
95            parse_orderbook_topic, parse_ticker_linear_funding, parse_ticker_linear_index_price,
96            parse_ticker_linear_mark_price, parse_ticker_option_greeks,
97            parse_ticker_option_index_price, parse_ticker_option_mark_price,
98            parse_ticker_option_quote, parse_ws_kline_bar, parse_ws_liquidation,
99            parse_ws_trade_tick,
100        },
101    },
102};
103
104/// Live market data client for Bybit.
105#[derive(Debug)]
106pub struct BybitDataClient {
107    client_id: ClientId,
108    config: BybitDataClientConfig,
109    http_client: BybitHttpClient,
110    ws_clients: Vec<BybitWebSocketClient>,
111    session_established: AtomicBool,
112    cancellation_token: CancellationToken,
113    session_tasks: TaskGroup,
114    command_tasks: TaskGroup,
115    shutdown_errors: Vec<String>,
116    data_sender: EventSender<DataEvent>,
117    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
118    book_depths: Arc<AtomicMap<InstrumentId, u32>>,
119    book_sync: BookSyncTracker,
120    quote_subs: Arc<AtomicSet<InstrumentId>>,
121    last_quotes: Arc<DashMap<InstrumentId, QuoteTick>>,
122    ticker_subs: Arc<AtomicMap<InstrumentId, AHashSet<&'static str>>>,
123    trade_subs: Arc<AtomicSet<InstrumentId>>,
124    option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
125    liquidation_subs: Arc<AtomicMap<InstrumentId, Arc<DataType>>>,
126    instrument_status_subs: Arc<AtomicSet<InstrumentId>>,
127    status_cache: Arc<AtomicMap<InstrumentId, MarketStatusAction>>,
128    instrument_subs: Arc<AtomicSet<InstrumentId>>,
129    subscribe_all_instruments: Arc<AtomicBool>,
130    clock: &'static AtomicTime,
131}
132
133impl BybitDataClient {
134    /// Creates a new [`BybitDataClient`] instance.
135    ///
136    /// # Errors
137    ///
138    /// Returns an error if the client fails to initialize.
139    pub fn new(client_id: ClientId, config: BybitDataClientConfig) -> anyhow::Result<Self> {
140        config.validate()?;
141
142        let clock = get_atomic_clock_realtime();
143        let data_sender = get_data_event_sender();
144        let socket_factory = SocketControlFactory::new(client_id, Some(*BYBIT_VENUE));
145        let proxy_url = config
146            .proxy_url
147            .as_ref()
148            .map(|value| value.expose_secret().to_owned());
149
150        let http_client = if let (Some(api_key), Some(api_secret)) =
151            (config.api_key.clone(), config.api_secret.clone())
152        {
153            BybitHttpClient::with_credentials(
154                api_key.into_inner(),
155                api_secret.into_inner(),
156                Some(config.http_base_url()),
157                config.http_timeout_secs,
158                config.max_retries,
159                config.retry_delay_initial_ms,
160                config.retry_delay_max_ms,
161                config.recv_window_ms,
162                proxy_url.clone(),
163            )?
164        } else {
165            BybitHttpClient::new(
166                Some(config.http_base_url()),
167                config.http_timeout_secs,
168                config.max_retries,
169                config.retry_delay_initial_ms,
170                config.retry_delay_max_ms,
171                config.recv_window_ms,
172                proxy_url.clone(),
173            )?
174        };
175
176        // Create a WebSocket client for each product type (default to Linear if empty)
177        let product_types = if config.product_types.is_empty() {
178            vec![BybitProductType::Linear]
179        } else {
180            config.product_types.clone()
181        };
182
183        let ws_clients: Vec<BybitWebSocketClient> = product_types
184            .iter()
185            .map(|product_type| {
186                BybitWebSocketClient::new_public_with(
187                    *product_type,
188                    config.environment,
189                    Some(config.ws_public_url_for(*product_type)),
190                    config.heartbeat_interval_secs,
191                    config.transport_backend,
192                    proxy_url.clone(),
193                )
194                .with_socket_control(
195                    socket_factory.control(format!("bybit-{}-data-streams", product_type.as_str())),
196                )
197            })
198            .collect();
199
200        let session_tasks = TaskGroup::new();
201        let command_tasks = TaskGroup::new();
202
203        Ok(Self {
204            client_id,
205            config,
206            http_client,
207            ws_clients,
208            session_established: AtomicBool::new(false),
209            cancellation_token: session_tasks.cancellation_token(),
210            session_tasks,
211            command_tasks,
212            shutdown_errors: Vec::new(),
213            data_sender,
214            instruments: Arc::new(AtomicMap::new()),
215            book_depths: Arc::new(AtomicMap::new()),
216            book_sync: BookSyncTracker::default(),
217            quote_subs: Arc::new(AtomicSet::new()),
218            last_quotes: Arc::new(DashMap::new()),
219            ticker_subs: Arc::new(AtomicMap::new()),
220            trade_subs: Arc::new(AtomicSet::new()),
221            option_greeks_subs: Arc::new(AtomicSet::new()),
222            liquidation_subs: Arc::new(AtomicMap::new()),
223            instrument_status_subs: Arc::new(AtomicSet::new()),
224            status_cache: Arc::new(AtomicMap::new()),
225            instrument_subs: Arc::new(AtomicSet::new()),
226            subscribe_all_instruments: Arc::new(AtomicBool::new(false)),
227            clock,
228        })
229    }
230
231    fn venue(&self) -> Venue {
232        *BYBIT_VENUE
233    }
234
235    fn get_ws_client_for_product(
236        &self,
237        product_type: BybitProductType,
238    ) -> Option<&BybitWebSocketClient> {
239        self.ws_clients
240            .iter()
241            .find(|ws| ws.product_type() == Some(product_type))
242    }
243
244    fn get_product_type_for_instrument(
245        &self,
246        instrument_id: InstrumentId,
247    ) -> Option<BybitProductType> {
248        let guard = self.instruments.load();
249        guard
250            .get(&instrument_id)
251            .and_then(|_| BybitProductType::from_suffix(instrument_id.symbol.as_str()))
252    }
253
254    fn spawn_command<F>(&self, future: F)
255    where
256        F: Future<Output = ()> + Send + 'static,
257    {
258        if let Err(e) = self.command_tasks.spawn(future) {
259            log::warn!("Skipping Bybit data command after shutdown began: {e}");
260        }
261    }
262
263    async fn finish_tasks(&self) -> anyhow::Result<()> {
264        let (session_result, command_result) = tokio::join!(
265            self.session_tasks
266                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
267            self.command_tasks
268                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
269        );
270        let mut errors = Vec::new();
271        if let Err(e) = session_result {
272            errors.push(format!("failed to finish Bybit data session tasks: {e}"));
273        }
274
275        if let Err(e) = command_result {
276            errors.push(format!("failed to finish Bybit data command tasks: {e}"));
277        }
278
279        if errors.is_empty() {
280            Ok(())
281        } else {
282            anyhow::bail!(errors.join("; "))
283        }
284    }
285
286    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
287        if !self.session_tasks.is_open() || !self.command_tasks.is_open() {
288            self.teardown_partial_connect().await?;
289            self.session_tasks
290                .start_generation()
291                .context("failed to start Bybit data session task generation")?;
292            self.command_tasks
293                .start_generation()
294                .context("failed to start Bybit data command task generation")?;
295            self.cancellation_token = self.session_tasks.cancellation_token();
296        }
297        Ok(())
298    }
299
300    fn spawn_ws<F>(&self, fut: F, context: &'static str)
301    where
302        F: Future<Output = anyhow::Result<()>> + Send + 'static,
303    {
304        let future = async move {
305            if let Err(e) = fut.await {
306                log::error!("{context}: {e:?}");
307            }
308        };
309
310        if let Err(e) = self.command_tasks.spawn(future) {
311            log::warn!("Skipping Bybit {context} after shutdown began: {e}");
312        }
313    }
314
315    fn spawn_instrument_polling(
316        &self,
317        product_types: &[BybitProductType],
318        poll_secs: u64,
319    ) -> anyhow::Result<()> {
320        let http = self.http_client.clone();
321        let sender = self.data_sender.clone();
322        let instruments = self.instruments.clone();
323        let status_cache = self.status_cache.clone();
324        let status_subs = self.instrument_status_subs.clone();
325        let instrument_subs = self.instrument_subs.clone();
326        let subscribe_all_instruments = self.subscribe_all_instruments.clone();
327        let cancel = self.cancellation_token.clone();
328        let clock = self.clock;
329        let product_types = product_types.to_vec();
330
331        let future = async move {
332            let mut interval = tokio::time::interval(Duration::from_secs(poll_secs));
333            interval.tick().await; // Skip first immediate tick
334
335            loop {
336                tokio::select! {
337                    _ = interval.tick() => {
338                        let all_flag = subscribe_all_instruments.load(Ordering::Relaxed);
339                        let want_instruments = all_flag || !instrument_subs.is_empty();
340                        let want_statuses = !status_subs.is_empty();
341                        if !want_instruments && !want_statuses {
342                            continue;
343                        }
344
345                        let mut all_statuses = AHashMap::new();
346
347                        if want_instruments {
348                            let subs: Option<AHashSet<InstrumentId>> = if all_flag {
349                                None
350                            } else {
351                                Some((**instrument_subs.load()).clone())
352                            };
353                            let mut inst_cache = (**instruments.load()).clone();
354
355                            for &pt in &product_types {
356                                match http.request_instruments_with_statuses(pt).await {
357                                    Ok((fetched, statuses)) => {
358                                        diff_and_emit_instruments(
359                                            &fetched, &mut inst_cache, subs.as_ref(), &sender,
360                                        );
361
362                                        for (id, action) in statuses {
363                                            if inst_cache.contains_key(&id) {
364                                                all_statuses.insert(id, action);
365                                            }
366                                        }
367                                    }
368                                    Err(e) => {
369                                        log::warn!("Bybit instrument poll failed for {pt:?}: {e}");
370                                    }
371                                }
372                            }
373
374                            instruments.store(inst_cache);
375                        } else {
376                            for &pt in &product_types {
377                                match http.request_instrument_statuses(pt).await {
378                                    Ok(new_statuses) => {
379                                        let inst_guard = instruments.load();
380                                        for (id, action) in new_statuses {
381                                            if inst_guard.contains_key(&id) {
382                                                all_statuses.insert(id, action);
383                                            }
384                                        }
385                                    }
386                                    Err(e) => {
387                                        log::warn!("Bybit instrument status poll failed for {pt:?}: {e}");
388                                    }
389                                }
390                            }
391                        }
392
393                        if want_statuses {
394                            let ts = clock.get_time_ns();
395                            let mut cache = (**status_cache.load()).clone();
396                            let subs_guard = status_subs.load();
397                            diff_and_emit_statuses(
398                                &all_statuses, &mut cache, Some(&subs_guard), &sender, ts, ts,
399                            );
400                            status_cache.store(cache);
401                        }
402                    }
403                    () = cancel.cancelled() => {
404                        log::debug!("Bybit instrument polling task cancelled");
405                        break;
406                    }
407                }
408            }
409        };
410
411        self.session_tasks
412            .spawn(future)
413            .context("failed to register Bybit instrument polling task")?;
414        log::debug!("Instrument polling started: interval={poll_secs}s");
415        Ok(())
416    }
417
418    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
419        self.session_tasks.begin_shutdown();
420        self.command_tasks.begin_shutdown();
421        for ws_client in &self.ws_clients {
422            ws_client.begin_shutdown();
423        }
424
425        for ws_client in &mut self.ws_clients {
426            if let Err(e) = ws_client.close().await {
427                self.shutdown_errors.push(e.to_string());
428            }
429        }
430
431        if let Err(e) = self.finish_tasks().await {
432            self.shutdown_errors.push(e.to_string());
433        }
434        self.session_established.store(false, Ordering::Release);
435
436        if self.shutdown_errors.is_empty() {
437            Ok(())
438        } else {
439            let errors = std::mem::take(&mut self.shutdown_errors);
440            anyhow::bail!("Bybit data shutdown failed: {}", errors.join("; "))
441        }
442    }
443}
444
445fn send_data(sender: &EventSender<DataEvent>, data: Data) {
446    if let Err(e) = sender.send(DataEvent::Data(data)) {
447        log::error!("Failed to emit data event: {e}");
448    }
449}
450
451fn validate_orderbook_depth(depth: u32) -> anyhow::Result<()> {
452    if !BYBIT_BOOK_DEPTHS.contains(&depth) {
453        anyhow::bail!("invalid depth {depth}; valid values are {BYBIT_BOOK_DEPTHS:?}");
454    }
455
456    Ok(())
457}
458
459fn liquidation_instrument_id(data_type: &DataType, venue: Venue) -> anyhow::Result<InstrumentId> {
460    let instrument_id = data_type
461        .instrument_id()?
462        .context("BybitLiquidation requires `instrument_id` metadata")?;
463
464    if instrument_id.venue != venue {
465        anyhow::bail!("BybitLiquidation requires a {venue} instrument, received {instrument_id}");
466    }
467
468    Ok(instrument_id)
469}
470
471fn liquidation_product_type(instrument_id: InstrumentId) -> anyhow::Result<BybitProductType> {
472    match BybitProductType::from_suffix(instrument_id.symbol.as_str()) {
473        Some(product_type @ (BybitProductType::Linear | BybitProductType::Inverse)) => {
474            Ok(product_type)
475        }
476        Some(product_type) => {
477            anyhow::bail!("Liquidations not available for {product_type:?} instruments")
478        }
479        None => anyhow::bail!(
480            "Liquidations require a -LINEAR or -INVERSE instrument, received {instrument_id}"
481        ),
482    }
483}
484
485// Each subscribe stores its own `Arc`, so pointer identity tells its entry apart from a later
486// subscription with an equal data type, and only the caller that removes the entry releases the
487// topic reference.
488fn remove_owned_liquidation_sub(
489    subs: &AtomicMap<InstrumentId, Arc<DataType>>,
490    instrument_id: InstrumentId,
491    owned: &Arc<DataType>,
492) -> bool {
493    let mut removed = false;
494
495    subs.rcu(|map| {
496        removed = map
497            .get(&instrument_id)
498            .is_some_and(|current| Arc::ptr_eq(current, owned));
499
500        if removed {
501            map.remove(&instrument_id);
502        }
503    });
504
505    removed
506}
507
508/// Cached funding state per symbol: (funding_rate, next_funding_time, funding_interval_hour).
509type FundingCacheEntry = (Option<String>, Option<String>, Option<String>);
510
511#[expect(clippy::too_many_arguments)]
512fn handle_ws_message(
513    message: &BybitWsMessage,
514    data_sender: &EventSender<DataEvent>,
515    instruments: &AHashMap<Ustr, InstrumentAny>,
516    product_type: Option<BybitProductType>,
517    trade_subs: &Arc<AtomicSet<InstrumentId>>,
518    ticker_subs: &Arc<AtomicMap<InstrumentId, AHashSet<&'static str>>>,
519    quote_subs: &Arc<AtomicSet<InstrumentId>>,
520    book_depths: &Arc<AtomicMap<InstrumentId, u32>>,
521    book_sync: &BookSyncTracker,
522    ws_client: &BybitWebSocketClient,
523    option_greeks_subs: &Arc<AtomicSet<InstrumentId>>,
524    liquidation_subs: &Arc<AtomicMap<InstrumentId, Arc<DataType>>>,
525    bar_types_cache: &Arc<AtomicMap<String, BarType>>,
526    bars_timestamp_on_close: bool,
527    last_quotes: &DashMap<InstrumentId, QuoteTick>,
528    funding_cache: &mut AHashMap<Ustr, FundingCacheEntry>,
529    snapshot_timeout: Duration,
530    tasks: &TaskSpawner,
531    clock: &AtomicTime,
532) {
533    let ts_init = clock.get_time_ns();
534    let resolve = |raw_symbol: &Ustr| -> Option<&InstrumentAny> {
535        let key = product_type.map_or(*raw_symbol, |pt| make_bybit_symbol(raw_symbol, pt));
536        instruments.get(&key)
537    };
538
539    match message {
540        BybitWsMessage::Orderbook(msg) => {
541            let Some(instrument) = resolve(&msg.data.s) else {
542                log::warn!("Unknown symbol in orderbook update: {}", msg.data.s);
543                return;
544            };
545            let instrument_id = instrument.id();
546
547            let Ok((depth, symbol)) = parse_orderbook_topic(msg.topic.as_str()) else {
548                log::warn!("Invalid orderbook topic: {}", msg.topic);
549                return;
550            };
551
552            if symbol != msg.data.s.as_str() {
553                log::warn!(
554                    "Orderbook topic symbol does not match payload: {}",
555                    msg.topic
556                );
557                return;
558            }
559
560            if book_depths.load().get(&instrument_id) == Some(&depth) {
561                handle_orderbook_deltas(
562                    msg,
563                    instrument,
564                    data_sender,
565                    book_depths,
566                    book_sync,
567                    ws_client,
568                    snapshot_timeout,
569                    tasks,
570                    ts_init,
571                );
572            }
573
574            if depth == BYBIT_QUOTE_DEPTH && quote_subs.contains(&instrument_id) {
575                match parse_orderbook_quote(msg, instrument, ts_init) {
576                    Ok(quote) => {
577                        send_data(data_sender, Data::Quote(quote));
578                    }
579                    Err(e) => log::error!("Failed to parse orderbook quote: {e}"),
580                }
581            }
582        }
583        BybitWsMessage::Trade(msg) => {
584            for trade in &msg.data {
585                let Some(instrument) = resolve(&trade.s) else {
586                    continue;
587                };
588                let instrument_id = instrument.id();
589                if !trade_subs.contains(&instrument_id) {
590                    continue;
591                }
592
593                match parse_ws_trade_tick(trade, instrument, ts_init) {
594                    Ok(tick) => send_data(data_sender, Data::Trade(tick)),
595                    Err(e) => log::error!("Failed to parse trade tick: {e}"),
596                }
597            }
598        }
599        BybitWsMessage::Liquidation(msg) => {
600            let subs = liquidation_subs.load();
601
602            for liquidation in &msg.data {
603                let Some(instrument) = resolve(&liquidation.s) else {
604                    continue;
605                };
606
607                // Emit on the subscribed data type so the message bus topic matches the subscriber
608                let Some(data_type) = subs.get(&instrument.id()) else {
609                    continue;
610                };
611
612                match parse_ws_liquidation(liquidation, instrument, ts_init) {
613                    Ok(liquidation) => {
614                        let custom =
615                            CustomData::new(Arc::new(liquidation), DataType::clone(data_type));
616                        send_data(data_sender, Data::Custom(custom));
617                    }
618                    Err(e) => log::error!("Failed to parse liquidation: {e}"),
619                }
620            }
621        }
622        BybitWsMessage::Kline(msg) => {
623            let Ok((_, raw_symbol)) = parse_kline_topic(msg.topic.as_str()) else {
624                log::warn!("Invalid kline topic: {}", msg.topic);
625                return;
626            };
627            let ustr_symbol = Ustr::from(raw_symbol);
628            let Some(instrument) = resolve(&ustr_symbol) else {
629                log::warn!("Unknown symbol in kline update: {raw_symbol}");
630                return;
631            };
632            let topic_key = msg.topic.as_str();
633            let Some(bar_type) = bar_types_cache.load().get(topic_key).copied() else {
634                log::debug!("No bar type cached for kline topic: {topic_key}");
635                return;
636            };
637
638            for kline in &msg.data {
639                if !kline.confirm {
640                    continue;
641                }
642
643                match parse_ws_kline_bar(
644                    kline,
645                    instrument,
646                    bar_type,
647                    bars_timestamp_on_close,
648                    ts_init,
649                ) {
650                    Ok(bar) => send_data(data_sender, Data::Bar(bar)),
651                    Err(e) => log::error!("Failed to parse kline bar: {e}"),
652                }
653            }
654        }
655        BybitWsMessage::TickerLinear(msg) => {
656            let Some(instrument) = resolve(&msg.data.symbol) else {
657                log::warn!("Unknown symbol in ticker update: {}", msg.data.symbol);
658                return;
659            };
660            let instrument_id = instrument.id();
661            let subs = ticker_subs.load();
662            let sub_set = subs.get(&instrument_id);
663
664            let ts_event = match parse_millis_i64(msg.ts, "ticker.ts") {
665                Ok(ts) => ts,
666                Err(e) => {
667                    log::error!("Failed to parse ticker timestamp: {e}");
668                    return;
669                }
670            };
671
672            if sub_set.is_some_and(|s| s.contains("funding"))
673                && matches!(
674                    instrument,
675                    InstrumentAny::CryptoPerpetual(_) | InstrumentAny::PerpetualContract(_)
676                )
677            {
678                let cache_entry = funding_cache
679                    .entry(msg.data.symbol)
680                    .or_insert((None, None, None));
681                let mut changed = false;
682
683                if let Some(rate) = &msg.data.funding_rate
684                    && cache_entry.0.as_ref() != Some(rate)
685                {
686                    cache_entry.0 = Some(rate.clone());
687                    changed = true;
688                }
689
690                if let Some(next_time) = &msg.data.next_funding_time
691                    && cache_entry.1.as_ref() != Some(next_time)
692                {
693                    cache_entry.1 = Some(next_time.clone());
694                    changed = true;
695                }
696
697                if let Some(interval) = &msg.data.funding_interval_hour {
698                    cache_entry.2 = Some(interval.clone());
699                }
700
701                if changed && cache_entry.0.is_some() {
702                    let mut merged = msg.data.clone();
703
704                    if merged.funding_rate.is_none() {
705                        merged.funding_rate.clone_from(&cache_entry.0);
706                    }
707
708                    if merged.next_funding_time.is_none() {
709                        merged.next_funding_time.clone_from(&cache_entry.1);
710                    }
711
712                    if merged.funding_interval_hour.is_none() {
713                        merged.funding_interval_hour.clone_from(&cache_entry.2);
714                    }
715
716                    match parse_ticker_linear_funding(&merged, instrument_id, ts_event, ts_init) {
717                        Ok(update) => {
718                            if let Err(e) =
719                                data_sender.send(DataEvent::Data(Data::FundingRate(update)))
720                            {
721                                log::error!("Failed to emit funding rate event: {e}");
722                            }
723                        }
724                        Err(e) => log::error!("Failed to parse ticker linear funding: {e}"),
725                    }
726                }
727            }
728
729            if sub_set.is_some_and(|s| s.contains("mark_prices")) && msg.data.mark_price.is_some() {
730                match parse_ticker_linear_mark_price(&msg.data, instrument, ts_event, ts_init) {
731                    Ok(update) => send_data(data_sender, Data::MarkPrice(update)),
732                    Err(e) => log::debug!("Skipping mark price update: {e}"),
733                }
734            }
735
736            if sub_set.is_some_and(|s| s.contains("index_prices")) && msg.data.index_price.is_some()
737            {
738                match parse_ticker_linear_index_price(&msg.data, instrument, ts_event, ts_init) {
739                    Ok(update) => send_data(data_sender, Data::IndexPrice(update)),
740                    Err(e) => log::debug!("Skipping index price update: {e}"),
741                }
742            }
743        }
744        BybitWsMessage::TickerOption(msg) => {
745            let Some(instrument) = resolve(&msg.data.symbol) else {
746                log::warn!(
747                    "Unknown symbol in option ticker update: {}",
748                    msg.data.symbol
749                );
750                return;
751            };
752            let instrument_id = instrument.id();
753            let subs = ticker_subs.load();
754            let sub_set = subs.get(&instrument_id);
755
756            if sub_set.is_some_and(|s| s.contains("quotes")) {
757                match parse_ticker_option_quote(msg, instrument, ts_init) {
758                    Ok(quote) => {
759                        let last = last_quotes.get(&instrument_id).map(|quote| *quote);
760                        if last.is_none_or(|q| !q.has_same_bid_ask(&quote)) {
761                            last_quotes.insert(instrument_id, quote);
762                            send_data(data_sender, Data::Quote(quote));
763                        }
764                    }
765                    Err(e) => log::error!("Failed to parse ticker option quote: {e}"),
766                }
767            }
768
769            if sub_set.is_some_and(|s| s.contains("mark_prices")) {
770                match parse_ticker_option_mark_price(msg, instrument, ts_init) {
771                    Ok(update) => send_data(data_sender, Data::MarkPrice(update)),
772                    Err(e) => log::error!("Failed to parse ticker option mark price: {e}"),
773                }
774            }
775
776            if sub_set.is_some_and(|s| s.contains("index_prices")) {
777                match parse_ticker_option_index_price(msg, instrument, ts_init) {
778                    Ok(update) => send_data(data_sender, Data::IndexPrice(update)),
779                    Err(e) => log::error!("Failed to parse ticker option index price: {e}"),
780                }
781            }
782
783            if option_greeks_subs.contains(&instrument_id) {
784                match parse_ticker_option_greeks(msg, instrument, ts_init) {
785                    Ok(greeks) => {
786                        if let Err(e) =
787                            data_sender.send(DataEvent::Data(Data::OptionGreeks(greeks)))
788                        {
789                            log::error!("Failed to send option greeks: {e}");
790                        }
791                    }
792                    Err(e) => log::error!("Failed to parse option greeks: {e}"),
793                }
794            }
795        }
796        BybitWsMessage::Reconnected => {
797            reset_option_quotes(last_quotes, product_type);
798            funding_cache.clear();
799            log::info!("WebSocket reconnected, cleared caches");
800
801            if let Some(product_type) = product_type {
802                reset_on_reconnect(
803                    product_type,
804                    book_depths,
805                    book_sync,
806                    ws_client,
807                    snapshot_timeout,
808                    tasks,
809                );
810            }
811        }
812        BybitWsMessage::Error(e) => {
813            log::warn!(
814                "Bybit WebSocket error: code={} message={}",
815                e.code,
816                e.message
817            );
818        }
819        BybitWsMessage::Auth(_)
820        | BybitWsMessage::OrderResponse(_)
821        | BybitWsMessage::AccountOrder(_)
822        | BybitWsMessage::AccountExecution(_)
823        | BybitWsMessage::AccountExecutionFast(_)
824        | BybitWsMessage::AccountWallet(_)
825        | BybitWsMessage::AccountPosition(_) => {}
826    }
827}
828
829// Validates the update ID before parsing, so a delta without levels still advances the book
830#[expect(clippy::too_many_arguments)]
831fn handle_orderbook_deltas(
832    msg: &BybitWsOrderbookDepthMsg,
833    instrument: &InstrumentAny,
834    data_sender: &EventSender<DataEvent>,
835    book_depths: &AtomicMap<InstrumentId, u32>,
836    book_sync: &BookSyncTracker,
837    ws_client: &BybitWebSocketClient,
838    snapshot_timeout: Duration,
839    tasks: &TaskSpawner,
840    ts_init: UnixNanos,
841) {
842    let instrument_id = instrument.id();
843    let is_snapshot = msg.msg_type.eq_ignore_ascii_case("snapshot");
844
845    let outcome = match u64::try_from(msg.data.u) {
846        Ok(update_id) => book_sync.validate(instrument_id, is_snapshot, update_id, Instant::now()),
847        Err(e) => {
848            log::error!(
849                "Invalid orderbook update ID {} for {instrument_id}: {e}",
850                msg.data.u
851            );
852            BookSequenceOutcome::Recover
853        }
854    };
855
856    match outcome {
857        BookSequenceOutcome::Accept => {}
858        BookSequenceOutcome::Suppress => return,
859        BookSequenceOutcome::Recover => {
860            start_recovery(
861                instrument_id,
862                book_depths,
863                book_sync,
864                ws_client,
865                snapshot_timeout,
866                tasks,
867            );
868            return;
869        }
870    }
871
872    if !is_snapshot && msg.data.b.is_empty() && msg.data.a.is_empty() {
873        return;
874    }
875
876    match parse_orderbook_deltas(msg, instrument, ts_init) {
877        Ok(deltas) => send_data(data_sender, Data::BookDeltas(Box::new(deltas))),
878        Err(e) => {
879            log::error!("Failed to parse orderbook deltas for {instrument_id}: {e}");
880            start_recovery(
881                instrument_id,
882                book_depths,
883                book_sync,
884                ws_client,
885                snapshot_timeout,
886                tasks,
887            );
888        }
889    }
890}
891
892fn upsert_instrument(
893    cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
894    instrument: InstrumentAny,
895) {
896    cache.insert(instrument.id(), instrument);
897}
898
899fn reset_option_quotes(
900    last_quotes: &DashMap<InstrumentId, QuoteTick>,
901    product_type: Option<BybitProductType>,
902) {
903    if product_type == Some(BybitProductType::Option) {
904        last_quotes.clear();
905    }
906}
907
908#[async_trait::async_trait(?Send)]
909impl DataClient for BybitDataClient {
910    fn client_id(&self) -> ClientId {
911        self.client_id
912    }
913
914    fn venue(&self) -> Option<Venue> {
915        Some(self.venue())
916    }
917
918    fn start(&mut self) -> anyhow::Result<()> {
919        log::info!(
920            "Started: client_id={}, product_types={:?}, environment={:?}, proxy_url={:?}",
921            self.client_id,
922            self.config.product_types,
923            self.config.environment,
924            self.config.proxy_url,
925        );
926        Ok(())
927    }
928
929    fn stop(&mut self) -> anyhow::Result<()> {
930        log::info!("Stopping {id}", id = self.client_id);
931        self.session_tasks.begin_shutdown();
932        self.command_tasks.begin_shutdown();
933        for ws_client in &self.ws_clients {
934            ws_client.begin_shutdown();
935        }
936        self.session_established.store(false, Ordering::Relaxed);
937        Ok(())
938    }
939
940    fn reset(&mut self) -> anyhow::Result<()> {
941        log::debug!("Resetting {id}", id = self.client_id);
942        self.session_tasks.begin_shutdown();
943        self.command_tasks.begin_shutdown();
944        for ws_client in &self.ws_clients {
945            ws_client.begin_shutdown();
946        }
947        self.session_established.store(false, Ordering::Relaxed);
948        self.book_depths.store(AHashMap::new());
949        self.book_sync.clear();
950        self.quote_subs.store(AHashSet::new());
951        self.last_quotes.clear();
952        self.ticker_subs.store(AHashMap::new());
953        self.option_greeks_subs.store(AHashSet::new());
954        self.liquidation_subs.store(AHashMap::new());
955        self.instrument_status_subs.store(AHashSet::new());
956        self.status_cache.store(AHashMap::new());
957        self.instrument_subs.store(AHashSet::new());
958        self.subscribe_all_instruments
959            .store(false, Ordering::Relaxed);
960        Ok(())
961    }
962
963    fn dispose(&mut self) -> anyhow::Result<()> {
964        log::debug!("Disposing {id}", id = self.client_id);
965        self.stop()
966    }
967
968    async fn connect(&mut self) -> anyhow::Result<()> {
969        if self.session_established.load(Ordering::Relaxed)
970            && self.session_tasks.is_open()
971            && self.command_tasks.is_open()
972        {
973            return Ok(());
974        }
975
976        register_bybit_custom_data();
977
978        self.prepare_task_groups().await?;
979        self.last_quotes.clear();
980        let ws_clients = self.ws_clients.clone();
981        let setup_guard =
982            TaskGroupGuard::new(&[&self.session_tasks, &self.command_tasks], move || {
983                for ws_client in ws_clients {
984                    ws_client.begin_shutdown();
985                }
986            });
987
988        let product_types = if self.config.product_types.is_empty() {
989            vec![BybitProductType::Linear]
990        } else {
991            self.config.product_types.clone()
992        };
993
994        let mut all_instruments = Vec::new();
995
996        for product_type in &product_types {
997            let fetched = self
998                .http_client
999                .request_instruments(*product_type, None, None)
1000                .await
1001                .with_context(|| {
1002                    format!("failed to request Bybit instruments for {product_type:?}")
1003                })?;
1004
1005            self.http_client.cache_instruments(&fetched);
1006
1007            self.instruments.rcu(|m| {
1008                for instrument in &fetched {
1009                    m.insert(instrument.id(), instrument.clone());
1010                }
1011            });
1012
1013            all_instruments.extend(fetched);
1014        }
1015
1016        // Seed instrument status cache from initial fetch
1017        if self
1018            .config
1019            .instrument_poll_interval_secs
1020            .is_some_and(|s| s > 0)
1021        {
1022            // Collect all statuses first (without holding the lock across await)
1023            let mut collected_statuses = Vec::new();
1024
1025            for product_type in &product_types {
1026                match self
1027                    .http_client
1028                    .request_instrument_statuses(*product_type)
1029                    .await
1030                {
1031                    Ok(statuses) => collected_statuses.push(statuses),
1032                    Err(e) => {
1033                        log::warn!(
1034                            "Failed to seed instrument status cache for {product_type:?}: {e}"
1035                        );
1036                    }
1037                }
1038            }
1039
1040            let inst_guard = self.instruments.load();
1041            let mut status_map = AHashMap::new();
1042
1043            for statuses in collected_statuses {
1044                for (id, action) in statuses {
1045                    if inst_guard.contains_key(&id) {
1046                        status_map.insert(id, action);
1047                    }
1048                }
1049            }
1050            log::debug!(
1051                "Seeded instrument status cache with {} entries",
1052                status_map.len()
1053            );
1054            self.status_cache.store(status_map);
1055        }
1056
1057        for instrument in all_instruments {
1058            if let Err(e) = self.data_sender.send(DataEvent::Data(instrument.into())) {
1059                log::warn!("Failed to send instrument: {e}");
1060            }
1061        }
1062
1063        // Build instruments map keyed by full Nautilus symbol for parsing
1064        let instruments_by_symbol: Arc<AHashMap<Ustr, InstrumentAny>> = {
1065            let guard = self.instruments.load();
1066            let mut map = AHashMap::new();
1067            for instrument in guard.values() {
1068                map.insert(instrument.id().symbol.inner(), instrument.clone());
1069            }
1070            Arc::new(map)
1071        };
1072
1073        let session_result = async {
1074            for ws_client in &mut self.ws_clients {
1075                ws_client
1076                    .connect()
1077                    .await
1078                    .context("failed to connect Bybit WebSocket")?;
1079                ws_client
1080                    .wait_until_active(10.0)
1081                    .await
1082                    .context("WebSocket did not become active")?;
1083
1084                let stream = ws_client.stream();
1085                let product_type = ws_client.product_type();
1086                let sender = self.data_sender.clone();
1087                let trade_subs = self.trade_subs.clone();
1088                let ticker_subs = self.ticker_subs.clone();
1089                let quote_subs = self.quote_subs.clone();
1090                let last_quotes = self.last_quotes.clone();
1091                let book_depths = self.book_depths.clone();
1092                let book_sync = self.book_sync.clone();
1093                let snapshot_timeout = Duration::from_secs(self.config.book_snapshot_timeout_secs);
1094                let tasks = self
1095                    .command_tasks
1096                    .spawner()
1097                    .context("Bybit data command task admission is closed")?;
1098                let option_greeks_subs = self.option_greeks_subs.clone();
1099                let liquidation_subs = self.liquidation_subs.clone();
1100                let bar_types_cache = ws_client.bar_types_cache().clone();
1101                let bars_timestamp_on_close = self.config.bars_timestamp_on_close;
1102                let instruments = Arc::clone(&instruments_by_symbol);
1103                let clock = self.clock;
1104                let cancel = self.cancellation_token.clone();
1105                let ws_client = ws_client.clone();
1106
1107                let future = async move {
1108                    let mut funding_cache: AHashMap<Ustr, FundingCacheEntry> = AHashMap::new();
1109
1110                    pin_mut!(stream);
1111
1112                    loop {
1113                        tokio::select! {
1114                            Some(message) = stream.next() => {
1115                                handle_ws_message(
1116                                    &message,
1117                                    &sender,
1118                                    &instruments,
1119                                    product_type,
1120                                    &trade_subs,
1121                                    &ticker_subs,
1122                                    &quote_subs,
1123                                    &book_depths,
1124                                    &book_sync,
1125                                    &ws_client,
1126                                    &option_greeks_subs,
1127                                    &liquidation_subs,
1128                                    &bar_types_cache,
1129                                    bars_timestamp_on_close,
1130                                    &last_quotes,
1131                                    &mut funding_cache,
1132                                    snapshot_timeout,
1133                                    &tasks,
1134                                    clock,
1135                                );
1136                            }
1137                            () = cancel.cancelled() => {
1138                                log::debug!("WebSocket stream task cancelled");
1139                                break;
1140                            }
1141                        }
1142                    }
1143                };
1144                self.session_tasks
1145                    .spawn(future)
1146                    .context("failed to register Bybit WebSocket stream task")?;
1147            }
1148
1149            if let Some(poll_secs) = self.config.instrument_poll_interval_secs
1150                && poll_secs > 0
1151            {
1152                self.spawn_instrument_polling(&product_types, poll_secs)?;
1153            }
1154
1155            Ok::<(), anyhow::Error>(())
1156        }
1157        .await;
1158
1159        if let Err(e) = session_result {
1160            if let Err(teardown_error) = self.teardown_partial_connect().await {
1161                return Err(e.context(format!(
1162                    "Bybit data startup teardown failed: {teardown_error}"
1163                )));
1164            }
1165            return Err(e);
1166        }
1167
1168        setup_guard.disarm();
1169        self.session_established.store(true, Ordering::Release);
1170        log::info!("Connected: client_id={}", self.client_id);
1171        Ok(())
1172    }
1173
1174    async fn disconnect(&mut self) -> anyhow::Result<()> {
1175        self.session_tasks.begin_shutdown();
1176        self.command_tasks.begin_shutdown();
1177        for ws_client in &self.ws_clients {
1178            ws_client.begin_shutdown();
1179        }
1180
1181        for ws_client in &mut self.ws_clients {
1182            if let Err(e) = ws_client.close().await {
1183                self.shutdown_errors.push(e.to_string());
1184            }
1185        }
1186
1187        // Allow time for unsubscribe confirmations
1188        tokio::time::sleep(Duration::from_millis(500)).await;
1189
1190        if let Err(e) = self.finish_tasks().await {
1191            self.shutdown_errors.push(e.to_string());
1192        }
1193
1194        self.book_depths.store(AHashMap::new());
1195        self.book_sync.clear();
1196        self.quote_subs.store(AHashSet::new());
1197        self.last_quotes.clear();
1198        self.ticker_subs.store(AHashMap::new());
1199        self.trade_subs.store(AHashSet::new());
1200        self.option_greeks_subs.store(AHashSet::new());
1201        self.liquidation_subs.store(AHashMap::new());
1202        self.instrument_status_subs.store(AHashSet::new());
1203        self.status_cache.store(AHashMap::new());
1204        self.instrument_subs.store(AHashSet::new());
1205        self.subscribe_all_instruments
1206            .store(false, Ordering::Relaxed);
1207        self.session_established.store(false, Ordering::Release);
1208        log::info!("Disconnected: client_id={}", self.client_id);
1209
1210        if self.shutdown_errors.is_empty() {
1211            Ok(())
1212        } else {
1213            let errors = std::mem::take(&mut self.shutdown_errors);
1214            anyhow::bail!("Bybit data shutdown failed: {}", errors.join("; "))
1215        }
1216    }
1217
1218    fn is_connected(&self) -> bool {
1219        self.session_established.load(Ordering::Relaxed)
1220            && self.ws_clients.iter().all(BybitWebSocketClient::is_active)
1221    }
1222
1223    fn is_disconnected(&self) -> bool {
1224        !self.is_connected()
1225    }
1226
1227    fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
1228        let data_type = cmd.data_type.type_name();
1229        if data_type != "BybitLiquidation" {
1230            log::warn!("Unsupported custom data subscription: {data_type}");
1231            return Ok(());
1232        }
1233
1234        let instrument_id = liquidation_instrument_id(&cmd.data_type, self.venue())?;
1235        let product_type = liquidation_product_type(instrument_id)?;
1236
1237        let ws = self
1238            .get_ws_client_for_product(product_type)
1239            .context("no WebSocket client for product type")?
1240            .clone();
1241
1242        if let Some(subscribed) = self.liquidation_subs.get_cloned(&instrument_id) {
1243            anyhow::ensure!(
1244                *subscribed == cmd.data_type,
1245                "Already subscribed to liquidations for {instrument_id} as {subscribed}"
1246            );
1247            return Ok(());
1248        }
1249
1250        let subscription = Arc::new(cmd.data_type);
1251        self.liquidation_subs
1252            .insert(instrument_id, Arc::clone(&subscription));
1253        let liquidation_subs = Arc::clone(&self.liquidation_subs);
1254
1255        self.spawn_ws(
1256            async move {
1257                if let Err(e) = ws.subscribe_liquidations(instrument_id).await {
1258                    // An unsubscribe that already removed this entry also released its reference
1259                    if remove_owned_liquidation_sub(&liquidation_subs, instrument_id, &subscription)
1260                        && let Err(e) = ws.unsubscribe_liquidations(instrument_id).await
1261                    {
1262                        log::warn!(
1263                            "Failed to unsubscribe after liquidation subscription error: {e}"
1264                        );
1265                    }
1266
1267                    return Err(e).context("liquidations subscription");
1268                }
1269
1270                Ok(())
1271            },
1272            "liquidation subscription",
1273        );
1274
1275        Ok(())
1276    }
1277
1278    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
1279        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1280            let cmd = cmd.with_instrument_ids(instrument_id);
1281            if cmd.book_type != BookType::L2_MBP {
1282                anyhow::bail!("Bybit only supports L2_MBP order book deltas");
1283            }
1284
1285            let depth = cmd
1286                .depth
1287                .map_or(BYBIT_DEFAULT_ORDERBOOK_DEPTH, |d| d.get() as u32);
1288
1289            validate_orderbook_depth(depth)?;
1290
1291            let product_type = self
1292                .get_product_type_for_instrument(instrument_id)
1293                .unwrap_or(BybitProductType::Linear);
1294
1295            let ws_client = self
1296                .get_ws_client_for_product(product_type)
1297                .context("no WebSocket client for product type")?
1298                .clone();
1299
1300            if let Some(subscribed_depth) = self.book_depths.load().get(&instrument_id) {
1301                anyhow::ensure!(
1302                    *subscribed_depth == depth,
1303                    "Already subscribed to book depth {subscribed_depth} for {instrument_id}"
1304                );
1305                return Ok(());
1306            }
1307
1308            let tasks = match self.command_tasks.spawner() {
1309                Ok(tasks) => tasks,
1310                Err(e) => {
1311                    log::warn!("Skipping Bybit data command after shutdown began: {e}");
1312                    return Ok(());
1313                }
1314            };
1315
1316            spawn_subscription_task(
1317                instrument_id,
1318                depth,
1319                Arc::clone(&self.book_depths),
1320                self.book_sync.clone(),
1321                ws_client,
1322                Duration::from_secs(self.config.book_snapshot_timeout_secs),
1323                &tasks,
1324            );
1325
1326            Ok(())
1327        })
1328    }
1329
1330    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
1331        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1332            self.subscribe_quotes_member(instrument_id)
1333        })
1334    }
1335
1336    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
1337        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1338            let product_type = self
1339                .get_product_type_for_instrument(instrument_id)
1340                .unwrap_or(BybitProductType::Linear);
1341
1342            self.trade_subs.insert(instrument_id);
1343
1344            let ws = self
1345                .get_ws_client_for_product(product_type)
1346                .context("no WebSocket client for product type")?
1347                .clone();
1348
1349            self.spawn_ws(
1350                async move {
1351                    ws.subscribe_trades(instrument_id)
1352                        .await
1353                        .context("trades subscription")
1354                },
1355                "trade subscription",
1356            );
1357
1358            Ok(())
1359        })
1360    }
1361
1362    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
1363        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1364            let product_type = self
1365                .get_product_type_for_instrument(instrument_id)
1366                .unwrap_or(BybitProductType::Linear);
1367
1368            if product_type == BybitProductType::Spot || product_type == BybitProductType::Option {
1369                anyhow::bail!("Funding rates not available for {product_type:?} instruments");
1370            }
1371
1372            let guard = self.instruments.load();
1373            if let Some(instrument) = guard.get(&instrument_id)
1374                && !matches!(
1375                    instrument,
1376                    InstrumentAny::CryptoPerpetual(_) | InstrumentAny::PerpetualContract(_)
1377                )
1378            {
1379                anyhow::bail!("Funding rates only available for perpetuals, not {instrument_id}");
1380            }
1381
1382            let mut should_subscribe = false;
1383            self.ticker_subs.rcu(|m| {
1384                let entry = m.entry(instrument_id).or_default();
1385                should_subscribe = entry.is_empty();
1386                entry.insert("funding");
1387            });
1388
1389            if should_subscribe {
1390                let ws = self
1391                    .get_ws_client_for_product(product_type)
1392                    .context("no WebSocket client for product type")?
1393                    .clone();
1394
1395                self.spawn_ws(
1396                    async move {
1397                        ws.subscribe_ticker(instrument_id)
1398                            .await
1399                            .context("ticker subscription for funding rates")
1400                    },
1401                    "funding rate subscription",
1402                );
1403            }
1404
1405            Ok(())
1406        })
1407    }
1408
1409    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
1410        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1411            let product_type = self
1412                .get_product_type_for_instrument(instrument_id)
1413                .unwrap_or(BybitProductType::Linear);
1414
1415            if product_type == BybitProductType::Spot {
1416                anyhow::bail!("Mark prices not available for Spot instruments");
1417            }
1418
1419            let mut should_subscribe = false;
1420            self.ticker_subs.rcu(|m| {
1421                let entry = m.entry(instrument_id).or_default();
1422                should_subscribe = entry.is_empty();
1423                entry.insert("mark_prices");
1424            });
1425
1426            if should_subscribe {
1427                let ws = self
1428                    .get_ws_client_for_product(product_type)
1429                    .context("no WebSocket client for product type")?
1430                    .clone();
1431
1432                self.spawn_ws(
1433                    async move {
1434                        ws.subscribe_ticker(instrument_id)
1435                            .await
1436                            .context("ticker subscription for mark prices")
1437                    },
1438                    "mark price subscription",
1439                );
1440            }
1441
1442            Ok(())
1443        })
1444    }
1445
1446    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
1447        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1448            let product_type = self
1449                .get_product_type_for_instrument(instrument_id)
1450                .unwrap_or(BybitProductType::Linear);
1451
1452            if product_type == BybitProductType::Spot {
1453                anyhow::bail!("Index prices not available for Spot instruments");
1454            }
1455
1456            let mut should_subscribe = false;
1457            self.ticker_subs.rcu(|m| {
1458                let entry = m.entry(instrument_id).or_default();
1459                should_subscribe = entry.is_empty();
1460                entry.insert("index_prices");
1461            });
1462
1463            if should_subscribe {
1464                let ws = self
1465                    .get_ws_client_for_product(product_type)
1466                    .context("no WebSocket client for product type")?
1467                    .clone();
1468
1469                self.spawn_ws(
1470                    async move {
1471                        ws.subscribe_ticker(instrument_id)
1472                            .await
1473                            .context("ticker subscription for index prices")
1474                    },
1475                    "index price subscription",
1476                );
1477            }
1478
1479            Ok(())
1480        })
1481    }
1482
1483    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
1484        execute_bar_subscriptions(&cmd.bar_types, |bar_type| {
1485            let instrument_id = bar_type.instrument_id();
1486            let product_type = self
1487                .get_product_type_for_instrument(instrument_id)
1488                .unwrap_or(BybitProductType::Linear);
1489
1490            if product_type == BybitProductType::Option {
1491                anyhow::bail!("Bybit does not support kline/bar data for options");
1492            }
1493
1494            let ws = self
1495                .get_ws_client_for_product(product_type)
1496                .context("no WebSocket client for product type")?
1497                .clone();
1498
1499            self.spawn_ws(
1500                async move {
1501                    ws.subscribe_bars(bar_type)
1502                        .await
1503                        .context("bars subscription")
1504                },
1505                "bar subscription",
1506            );
1507
1508            Ok(())
1509        })
1510    }
1511
1512    fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
1513        let data_type = cmd.data_type.type_name();
1514        if data_type != "BybitLiquidation" {
1515            log::warn!("Unsupported custom data unsubscription: {data_type}");
1516            return Ok(());
1517        }
1518
1519        let instrument_id = liquidation_instrument_id(&cmd.data_type, self.venue())?;
1520
1521        let Some(subscription) = self
1522            .liquidation_subs
1523            .get_cloned(&instrument_id)
1524            .filter(|subscribed| **subscribed == cmd.data_type)
1525        else {
1526            return Ok(());
1527        };
1528
1529        let product_type = liquidation_product_type(instrument_id)?;
1530
1531        let ws = self
1532            .get_ws_client_for_product(product_type)
1533            .context("no WebSocket client for product type")?
1534            .clone();
1535
1536        // A failed subscribe may have removed the entry and released its reference already
1537        if !remove_owned_liquidation_sub(&self.liquidation_subs, instrument_id, &subscription) {
1538            return Ok(());
1539        }
1540
1541        self.spawn_ws(
1542            async move {
1543                ws.unsubscribe_liquidations(instrument_id)
1544                    .await
1545                    .context("liquidations unsubscribe")
1546            },
1547            "liquidation unsubscribe",
1548        );
1549
1550        Ok(())
1551    }
1552
1553    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
1554        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1555            let Some(depth) = self.book_depths.load().get(&instrument_id).copied() else {
1556                return Ok(());
1557            };
1558
1559            self.book_depths.remove(&instrument_id);
1560            self.book_sync.remove(instrument_id);
1561
1562            let product_type = self
1563                .get_product_type_for_instrument(instrument_id)
1564                .unwrap_or(BybitProductType::Linear);
1565
1566            let ws = self
1567                .get_ws_client_for_product(product_type)
1568                .context("no WebSocket client for product type")?
1569                .clone();
1570
1571            self.spawn_ws(
1572                async move {
1573                    ws.unsubscribe_orderbook(instrument_id, depth)
1574                        .await
1575                        .context("orderbook unsubscribe")
1576                },
1577                "order book unsubscribe",
1578            );
1579
1580            Ok(())
1581        })
1582    }
1583
1584    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
1585        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1586            self.unsubscribe_quotes_member(instrument_id)
1587        })
1588    }
1589
1590    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
1591        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1592            let product_type = self
1593                .get_product_type_for_instrument(instrument_id)
1594                .unwrap_or(BybitProductType::Linear);
1595
1596            self.trade_subs.remove(&instrument_id);
1597
1598            let ws = self
1599                .get_ws_client_for_product(product_type)
1600                .context("no WebSocket client for product type")?
1601                .clone();
1602
1603            self.spawn_ws(
1604                async move {
1605                    ws.unsubscribe_trades(instrument_id)
1606                        .await
1607                        .context("trades unsubscribe")
1608                },
1609                "trade unsubscribe",
1610            );
1611
1612            Ok(())
1613        })
1614    }
1615
1616    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
1617        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1618            let product_type = self
1619                .get_product_type_for_instrument(instrument_id)
1620                .unwrap_or(BybitProductType::Linear);
1621
1622            let mut should_unsubscribe = false;
1623            self.ticker_subs.rcu(|m| {
1624                if let Some(entry) = m.get_mut(&instrument_id) {
1625                    entry.remove("funding");
1626                    if entry.is_empty() {
1627                        m.remove(&instrument_id);
1628                        should_unsubscribe = true;
1629                    } else {
1630                        should_unsubscribe = false;
1631                    }
1632                } else {
1633                    should_unsubscribe = false;
1634                }
1635            });
1636
1637            if should_unsubscribe {
1638                let ws = self
1639                    .get_ws_client_for_product(product_type)
1640                    .context("no WebSocket client for product type")?
1641                    .clone();
1642
1643                self.spawn_ws(
1644                    async move {
1645                        ws.unsubscribe_ticker(instrument_id)
1646                            .await
1647                            .context("ticker unsubscribe for funding rates")
1648                    },
1649                    "funding rate unsubscribe",
1650                );
1651            }
1652
1653            Ok(())
1654        })
1655    }
1656
1657    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
1658        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1659            let product_type = self
1660                .get_product_type_for_instrument(instrument_id)
1661                .unwrap_or(BybitProductType::Linear);
1662
1663            let mut should_unsubscribe = false;
1664            self.ticker_subs.rcu(|m| {
1665                if let Some(entry) = m.get_mut(&instrument_id) {
1666                    entry.remove("mark_prices");
1667                    if entry.is_empty() {
1668                        m.remove(&instrument_id);
1669                        should_unsubscribe = true;
1670                    } else {
1671                        should_unsubscribe = false;
1672                    }
1673                } else {
1674                    should_unsubscribe = false;
1675                }
1676            });
1677
1678            if should_unsubscribe {
1679                let ws = self
1680                    .get_ws_client_for_product(product_type)
1681                    .context("no WebSocket client for product type")?
1682                    .clone();
1683
1684                self.spawn_ws(
1685                    async move {
1686                        ws.unsubscribe_ticker(instrument_id)
1687                            .await
1688                            .context("ticker unsubscribe for mark prices")
1689                    },
1690                    "mark price unsubscribe",
1691                );
1692            }
1693
1694            Ok(())
1695        })
1696    }
1697
1698    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
1699        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1700            let product_type = self
1701                .get_product_type_for_instrument(instrument_id)
1702                .unwrap_or(BybitProductType::Linear);
1703
1704            let mut should_unsubscribe = false;
1705            self.ticker_subs.rcu(|m| {
1706                if let Some(entry) = m.get_mut(&instrument_id) {
1707                    entry.remove("index_prices");
1708                    if entry.is_empty() {
1709                        m.remove(&instrument_id);
1710                        should_unsubscribe = true;
1711                    } else {
1712                        should_unsubscribe = false;
1713                    }
1714                } else {
1715                    should_unsubscribe = false;
1716                }
1717            });
1718
1719            if should_unsubscribe {
1720                let ws = self
1721                    .get_ws_client_for_product(product_type)
1722                    .context("no WebSocket client for product type")?
1723                    .clone();
1724
1725                self.spawn_ws(
1726                    async move {
1727                        ws.unsubscribe_ticker(instrument_id)
1728                            .await
1729                            .context("ticker unsubscribe for index prices")
1730                    },
1731                    "index price unsubscribe",
1732                );
1733            }
1734
1735            Ok(())
1736        })
1737    }
1738
1739    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
1740        execute_bar_subscriptions(&cmd.bar_types, |bar_type| {
1741            let instrument_id = bar_type.instrument_id();
1742            let product_type = self
1743                .get_product_type_for_instrument(instrument_id)
1744                .unwrap_or(BybitProductType::Linear);
1745
1746            let ws = self
1747                .get_ws_client_for_product(product_type)
1748                .context("no WebSocket client for product type")?
1749                .clone();
1750
1751            self.spawn_ws(
1752                async move {
1753                    ws.unsubscribe_bars(bar_type)
1754                        .await
1755                        .context("bars unsubscribe")
1756                },
1757                "bar unsubscribe",
1758            );
1759
1760            Ok(())
1761        })
1762    }
1763
1764    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
1765        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1766            self.option_greeks_subs.insert(instrument_id);
1767
1768            let mut should_subscribe = false;
1769            self.ticker_subs.rcu(|m| {
1770                let entry = m.entry(instrument_id).or_default();
1771                should_subscribe = entry.is_empty();
1772                entry.insert("option_greeks");
1773            });
1774
1775            if should_subscribe {
1776                let product_type = self
1777                    .get_product_type_for_instrument(instrument_id)
1778                    .unwrap_or(BybitProductType::Option);
1779
1780                let ws = self
1781                    .get_ws_client_for_product(product_type)
1782                    .context("no WebSocket client for product type")?
1783                    .clone();
1784
1785                self.spawn_ws(
1786                    async move {
1787                        ws.subscribe_ticker(instrument_id)
1788                            .await
1789                            .context("ticker subscription for option greeks")
1790                    },
1791                    "option greeks subscription",
1792                );
1793            }
1794
1795            Ok(())
1796        })
1797    }
1798
1799    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
1800        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1801            self.option_greeks_subs.remove(&instrument_id);
1802
1803            let mut should_unsubscribe = false;
1804            self.ticker_subs.rcu(|m| {
1805                if let Some(entry) = m.get_mut(&instrument_id) {
1806                    entry.remove("option_greeks");
1807                    if entry.is_empty() {
1808                        m.remove(&instrument_id);
1809                        should_unsubscribe = true;
1810                    } else {
1811                        should_unsubscribe = false;
1812                    }
1813                } else {
1814                    should_unsubscribe = false;
1815                }
1816            });
1817
1818            if should_unsubscribe {
1819                let product_type = self
1820                    .get_product_type_for_instrument(instrument_id)
1821                    .unwrap_or(BybitProductType::Option);
1822
1823                let ws = self
1824                    .get_ws_client_for_product(product_type)
1825                    .context("no WebSocket client for product type")?
1826                    .clone();
1827
1828                self.spawn_ws(
1829                    async move {
1830                        ws.unsubscribe_ticker(instrument_id)
1831                            .await
1832                            .context("ticker unsubscribe for option greeks")
1833                    },
1834                    "option greeks unsubscribe",
1835                );
1836            }
1837
1838            Ok(())
1839        })
1840    }
1841
1842    fn subscribe_instruments(&mut self, cmd: SubscribeInstruments) -> anyhow::Result<()> {
1843        log::debug!(
1844            "subscribe_instruments: {venue} (definition updates detected via periodic instrument info polling)",
1845            venue = cmd.venue,
1846        );
1847        self.subscribe_all_instruments
1848            .store(true, Ordering::Relaxed);
1849        Ok(())
1850    }
1851
1852    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
1853        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1854            log::debug!(
1855                "subscribe_instrument: {instrument_id} (definition updates detected via periodic instrument info polling)",
1856            );
1857            self.instrument_subs.insert(instrument_id);
1858            Ok(())
1859        })
1860    }
1861
1862    fn unsubscribe_instruments(&mut self, cmd: &UnsubscribeInstruments) -> anyhow::Result<()> {
1863        log::debug!("unsubscribe_instruments: {venue}", venue = cmd.venue);
1864        self.subscribe_all_instruments
1865            .store(false, Ordering::Relaxed);
1866        Ok(())
1867    }
1868
1869    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
1870        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1871            log::debug!("unsubscribe_instrument: {instrument_id}");
1872            self.instrument_subs.remove(&instrument_id);
1873            Ok(())
1874        })
1875    }
1876
1877    fn subscribe_instrument_status(
1878        &mut self,
1879        cmd: SubscribeInstrumentStatus,
1880    ) -> anyhow::Result<()> {
1881        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1882            log::debug!(
1883                "subscribe_instrument_status: {instrument_id} (status changes detected via periodic instrument info polling)",
1884            );
1885            self.instrument_status_subs.insert(instrument_id);
1886
1887            if let Some(action) = self.status_cache.load().get(&instrument_id).copied() {
1888                let ts = self.clock.get_time_ns();
1889                emit_status(&self.data_sender, instrument_id, action, ts, ts);
1890            }
1891
1892            Ok(())
1893        })
1894    }
1895
1896    fn unsubscribe_instrument_status(
1897        &mut self,
1898        cmd: &UnsubscribeInstrumentStatus,
1899    ) -> anyhow::Result<()> {
1900        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
1901            log::debug!("unsubscribe_instrument_status: {instrument_id}",);
1902            self.instrument_status_subs.remove(&instrument_id);
1903            Ok(())
1904        })
1905    }
1906
1907    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
1908        let http = self.http_client.clone();
1909        let sender = self.data_sender.clone();
1910        let instruments_cache = self.instruments.clone();
1911        let request_id = request.request_id;
1912        let client_id = request.client_id.unwrap_or(self.client_id);
1913        let venue = self.venue();
1914        let start = request.start;
1915        let end = request.end;
1916        let params = request.params;
1917        let clock = self.clock;
1918        let start_nanos = datetime_to_unix_nanos(start);
1919        let end_nanos = datetime_to_unix_nanos(end);
1920        let product_types = if self.config.product_types.is_empty() {
1921            vec![BybitProductType::Linear]
1922        } else {
1923            self.config.product_types.clone()
1924        };
1925
1926        self.spawn_command(async move {
1927            let mut all_instruments = Vec::new();
1928
1929            for product_type in product_types {
1930                match http.request_instruments(product_type, None, None).await {
1931                    Ok(instruments) => {
1932                        instruments_cache.rcu(|cache| {
1933                            for instrument in &instruments {
1934                                cache.insert(instrument.id(), instrument.clone());
1935                            }
1936                        });
1937
1938                        all_instruments.extend(instruments);
1939                    }
1940                    Err(e) => {
1941                        log::error!("Failed to fetch instruments for {product_type:?}: {e:?}");
1942                    }
1943                }
1944            }
1945
1946            let response = DataResponse::Instruments(InstrumentsResponse::new(
1947                request_id,
1948                client_id,
1949                venue,
1950                all_instruments,
1951                start_nanos,
1952                end_nanos,
1953                clock.get_time_ns(),
1954                params,
1955            ));
1956
1957            if let Err(e) = sender.send(DataEvent::Response(response)) {
1958                log::error!("Failed to send instruments response: {e}");
1959            }
1960        });
1961
1962        Ok(())
1963    }
1964
1965    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
1966        let instrument_id = request.single_instrument_id()?;
1967
1968        let http = self.http_client.clone();
1969        let sender = self.data_sender.clone();
1970        let instruments = self.instruments.clone();
1971        let request_id = request.request_id;
1972        let client_id = request.client_id.unwrap_or(self.client_id);
1973        let start = request.start;
1974        let end = request.end;
1975        let params = request.params;
1976        let clock = self.clock;
1977        let start_nanos = datetime_to_unix_nanos(start);
1978        let end_nanos = datetime_to_unix_nanos(end);
1979
1980        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
1981            .unwrap_or(BybitProductType::Linear);
1982        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str()).to_string();
1983
1984        self.spawn_command(async move {
1985            match http
1986                .request_instruments(product_type, Some(raw_symbol), None)
1987                .await
1988                .context("fetch instrument from API")
1989            {
1990                Ok(fetched) => {
1991                    if let Some(instrument) = fetched.into_iter().find(|i| i.id() == instrument_id)
1992                    {
1993                        upsert_instrument(&instruments, instrument.clone());
1994
1995                        let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
1996                            request_id,
1997                            client_id,
1998                            instrument.id(),
1999                            instrument,
2000                            start_nanos,
2001                            end_nanos,
2002                            clock.get_time_ns(),
2003                            params,
2004                        )));
2005
2006                        if let Err(e) = sender.send(DataEvent::Response(response)) {
2007                            log::error!("Failed to send instrument response: {e}");
2008                        }
2009                    } else {
2010                        log::error!("Instrument not found: {instrument_id}");
2011                    }
2012                }
2013                Err(e) => log::error!("Instrument request failed: {e:?}"),
2014            }
2015        });
2016
2017        Ok(())
2018    }
2019
2020    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
2021        let instrument_id = request.single_instrument_id()?;
2022
2023        let http = self.http_client.clone();
2024        let sender = self.data_sender.clone();
2025        let depth = request.depth.map(|n| n.get() as u32);
2026        let request_id = request.request_id;
2027        let client_id = request.client_id.unwrap_or(self.client_id);
2028        let params = request.params;
2029        let clock = self.clock;
2030
2031        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
2032            .unwrap_or(BybitProductType::Linear);
2033
2034        self.spawn_command(async move {
2035            match http
2036                .request_orderbook_snapshot(product_type, instrument_id, depth)
2037                .await
2038                .context("failed to request book snapshot from Bybit")
2039            {
2040                Ok(deltas) => {
2041                    let mut book = OrderBook::new(instrument_id, BookType::L2_MBP);
2042                    if let Err(e) = book.apply_deltas(&deltas) {
2043                        log::error!("Failed to apply book deltas for {instrument_id}: {e}");
2044                        return;
2045                    }
2046
2047                    let response = DataResponse::Book(BookResponse::new(
2048                        request_id,
2049                        client_id,
2050                        instrument_id,
2051                        book,
2052                        None,
2053                        None,
2054                        clock.get_time_ns(),
2055                        params,
2056                    ));
2057
2058                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2059                        log::error!("Failed to send book snapshot response: {e}");
2060                    }
2061                }
2062                Err(e) => log::error!("Book snapshot request failed for {instrument_id}: {e:?}"),
2063            }
2064        });
2065
2066        Ok(())
2067    }
2068
2069    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
2070        let instrument_id = request.single_instrument_id()?;
2071
2072        let http = self.http_client.clone();
2073        let sender = self.data_sender.clone();
2074        let start = request.start;
2075        let end = request.end;
2076        let limit = request.limit.map(|n| n.get() as u32);
2077        let request_id = request.request_id;
2078        let client_id = request.client_id.unwrap_or(self.client_id);
2079        let params = request.params;
2080        let clock = self.clock;
2081        let start_nanos = datetime_to_unix_nanos(start);
2082        let end_nanos = datetime_to_unix_nanos(end);
2083
2084        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
2085            .unwrap_or(BybitProductType::Linear);
2086
2087        self.spawn_command(async move {
2088            match http
2089                .request_trades(product_type, instrument_id, limit)
2090                .await
2091                .context("failed to request trades from Bybit")
2092            {
2093                Ok(trades) => {
2094                    let response = DataResponse::Trades(TradesResponse::new(
2095                        request_id,
2096                        client_id,
2097                        instrument_id,
2098                        trades,
2099                        start_nanos,
2100                        end_nanos,
2101                        clock.get_time_ns(),
2102                        params,
2103                    ));
2104
2105                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2106                        log::error!("Failed to send trades response: {e}");
2107                    }
2108                }
2109                Err(e) => log::error!("Trade request failed: {e:?}"),
2110            }
2111        });
2112
2113        Ok(())
2114    }
2115
2116    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
2117        let bar_type = request.single_bar_type()?;
2118
2119        let http = self.http_client.clone();
2120        let bars_timestamp_on_close = self.config.bars_timestamp_on_close;
2121        let sender = self.data_sender.clone();
2122        let start = request.start;
2123        let end = request.end;
2124        let limit = request.limit.map(|n| n.get() as u32);
2125        let request_id = request.request_id;
2126        let client_id = request.client_id.unwrap_or(self.client_id);
2127        let params = request.params;
2128        let clock = self.clock;
2129        let start_nanos = datetime_to_unix_nanos(start);
2130        let end_nanos = datetime_to_unix_nanos(end);
2131
2132        let instrument_id = bar_type.instrument_id();
2133        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
2134            .unwrap_or(BybitProductType::Linear);
2135
2136        self.spawn_command(async move {
2137            match http
2138                .request_bars(
2139                    product_type,
2140                    bar_type,
2141                    start,
2142                    end,
2143                    limit,
2144                    bars_timestamp_on_close,
2145                )
2146                .await
2147                .context("failed to request bars from Bybit")
2148            {
2149                Ok(bars) => {
2150                    let response = DataResponse::Bars(BarsResponse::new(
2151                        request_id,
2152                        client_id,
2153                        bar_type,
2154                        bars,
2155                        start_nanos,
2156                        end_nanos,
2157                        clock.get_time_ns(),
2158                        params,
2159                    ));
2160
2161                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2162                        log::error!("Failed to send bars response: {e}");
2163                    }
2164                }
2165                Err(e) => log::error!("Bar request failed: {e:?}"),
2166            }
2167        });
2168
2169        Ok(())
2170    }
2171
2172    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
2173        let instrument_id = request.single_instrument_id()?;
2174
2175        let http = self.http_client.clone();
2176        let sender = self.data_sender.clone();
2177        let start = request.start;
2178        let end = request.end;
2179        let limit = request.limit.map(|n| n.get() as u32);
2180        let request_id = request.request_id;
2181        let client_id = request.client_id.unwrap_or(self.client_id);
2182        let params = request.params;
2183        let clock = self.clock;
2184        let start_nanos = datetime_to_unix_nanos(start);
2185        let end_nanos = datetime_to_unix_nanos(end);
2186
2187        let product_type = BybitProductType::from_suffix(instrument_id.symbol.as_str())
2188            .unwrap_or(BybitProductType::Linear);
2189
2190        if product_type == BybitProductType::Spot || product_type == BybitProductType::Option {
2191            anyhow::bail!("Funding rates not available for {product_type} instruments");
2192        }
2193
2194        self.spawn_command(async move {
2195            match http
2196                .request_funding_rates(product_type, instrument_id, start, end, limit)
2197                .await
2198                .context("failed to request funding rates from Bybit")
2199            {
2200                Ok(funding_rates) => {
2201                    let response = DataResponse::FundingRates(FundingRatesResponse::new(
2202                        request_id,
2203                        client_id,
2204                        instrument_id,
2205                        funding_rates,
2206                        start_nanos,
2207                        end_nanos,
2208                        clock.get_time_ns(),
2209                        params,
2210                    ));
2211
2212                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2213                        log::error!("Failed to send funding rates response: {e}");
2214                    }
2215                }
2216                Err(e) => log::error!("Funding rates request failed for {instrument_id}: {e:?}"),
2217            }
2218        });
2219
2220        Ok(())
2221    }
2222
2223    fn request_option_chain_reference_price(
2224        &self,
2225        request: RequestOptionChainReferencePrice,
2226    ) -> anyhow::Result<()> {
2227        let series_id = request.series_id;
2228        let instrument_id = request.instrument_id;
2229        let http_client = self.http_client.clone();
2230        let sender = self.data_sender.clone();
2231        let request_id = request.request_id;
2232        let client_id = request.client_id.unwrap_or(self.client_id());
2233        let params = request.params;
2234        let clock = self.clock;
2235
2236        self.spawn_command(async move {
2237            let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str()).to_string();
2238            let query = crate::http::query::BybitTickersParams {
2239                category: BybitProductType::Option,
2240                symbol: Some(raw_symbol),
2241                base_coin: None,
2242                exp_date: None,
2243            };
2244            let price = match http_client.request_option_tickers_raw_with_params(&query).await {
2245                Ok(tickers) => tickers.into_iter().find_map(|ticker| {
2246                    let decimal = ticker.underlying_price.parse::<Decimal>().ok()?;
2247                    if decimal <= Decimal::ZERO {
2248                        return None;
2249                    }
2250
2251                    match Price::from_decimal(decimal) {
2252                        Ok(price) => Some(price),
2253                        Err(e) => {
2254                            log::warn!(
2255                                "Invalid Bybit option-chain reference price for {instrument_id}: {e}"
2256                            );
2257                            None
2258                        }
2259                    }
2260                }),
2261                Err(e) => {
2262                    log::error!(
2263                        "Option-chain reference price request failed for {series_id}: {e:?}"
2264                    );
2265                    None
2266                }
2267            };
2268            let response = DataResponse::OptionChainReferencePrice(
2269                OptionChainReferencePriceResponse::new(
2270                    request_id,
2271                    client_id,
2272                    series_id,
2273                    price,
2274                    clock.get_time_ns(),
2275                    params,
2276                ),
2277            );
2278
2279            if let Err(e) = sender.send(DataEvent::Response(response)) {
2280                log::error!("Failed to send option-chain reference price response: {e}");
2281            }
2282        });
2283
2284        Ok(())
2285    }
2286}
2287
2288impl BybitDataClient {
2289    fn subscribe_quotes_member(&self, instrument_id: InstrumentId) -> anyhow::Result<()> {
2290        let product_type = self
2291            .get_product_type_for_instrument(instrument_id)
2292            .unwrap_or(BybitProductType::Linear);
2293
2294        let ws = self
2295            .get_ws_client_for_product(product_type)
2296            .context("no WebSocket client for product type")?
2297            .clone();
2298
2299        if product_type == BybitProductType::Option {
2300            self.last_quotes.remove(&instrument_id);
2301            let mut should_subscribe = false;
2302            self.ticker_subs.rcu(|m| {
2303                let entry = m.entry(instrument_id).or_default();
2304                should_subscribe = entry.is_empty();
2305                entry.insert("quotes");
2306            });
2307
2308            if should_subscribe {
2309                self.spawn_ws(
2310                    async move {
2311                        ws.subscribe_ticker(instrument_id)
2312                            .await
2313                            .context("ticker subscription")
2314                    },
2315                    "quote subscription",
2316                );
2317            }
2318        } else {
2319            if self.quote_subs.contains(&instrument_id) {
2320                return Ok(());
2321            }
2322
2323            self.quote_subs.insert(instrument_id);
2324            let quote_subs = Arc::clone(&self.quote_subs);
2325
2326            self.spawn_ws(
2327                async move {
2328                    if let Err(e) = ws
2329                        .subscribe_orderbook(instrument_id, BYBIT_QUOTE_DEPTH)
2330                        .await
2331                    {
2332                        // A stale entry, or a leftover reference that a retry would ride, lets a
2333                        // later unsubscribe drop the depth-1 book's topic, and an unsubscribe that
2334                        // removed the entry first already released the reference.
2335                        let mut removed = false;
2336                        quote_subs.rcu(|subs| removed = subs.remove(&instrument_id));
2337
2338                        if removed
2339                            && let Err(e) = ws
2340                                .unsubscribe_orderbook(instrument_id, BYBIT_QUOTE_DEPTH)
2341                                .await
2342                        {
2343                            log::warn!("Failed to unsubscribe after quote subscription error: {e}");
2344                        }
2345
2346                        return Err(e).context("orderbook subscription for quotes");
2347                    }
2348
2349                    Ok(())
2350                },
2351                "quote subscription (orderbook)",
2352            );
2353        }
2354
2355        Ok(())
2356    }
2357
2358    fn unsubscribe_quotes_member(&self, instrument_id: InstrumentId) -> anyhow::Result<()> {
2359        let product_type = self
2360            .get_product_type_for_instrument(instrument_id)
2361            .unwrap_or(BybitProductType::Linear);
2362
2363        let ws = self
2364            .get_ws_client_for_product(product_type)
2365            .context("no WebSocket client for product type")?
2366            .clone();
2367
2368        if product_type == BybitProductType::Option {
2369            self.last_quotes.remove(&instrument_id);
2370            let mut should_unsubscribe = false;
2371            self.ticker_subs.rcu(|m| {
2372                if let Some(entry) = m.get_mut(&instrument_id) {
2373                    entry.remove("quotes");
2374                    if entry.is_empty() {
2375                        m.remove(&instrument_id);
2376                        should_unsubscribe = true;
2377                    } else {
2378                        should_unsubscribe = false;
2379                    }
2380                } else {
2381                    should_unsubscribe = false;
2382                }
2383            });
2384
2385            if should_unsubscribe {
2386                self.spawn_ws(
2387                    async move {
2388                        ws.unsubscribe_ticker(instrument_id)
2389                            .await
2390                            .context("ticker unsubscribe")
2391                    },
2392                    "quote unsubscribe",
2393                );
2394            }
2395        } else {
2396            if !self.quote_subs.contains(&instrument_id) {
2397                return Ok(());
2398            }
2399
2400            self.quote_subs.remove(&instrument_id);
2401            self.spawn_ws(
2402                async move {
2403                    ws.unsubscribe_orderbook(instrument_id, BYBIT_QUOTE_DEPTH)
2404                        .await
2405                        .context("orderbook unsubscribe for quotes")
2406                },
2407                "quote unsubscribe (orderbook)",
2408            );
2409        }
2410
2411        Ok(())
2412    }
2413}
2414
2415#[cfg(test)]
2416#[path = "../benches"]
2417pub(crate) mod tests {
2418    // These routines are exercised by Criterion, not the unit-test harness
2419    #[allow(dead_code)]
2420    #[path = "quotes/mod.rs"]
2421    pub(crate) mod benchmarks;
2422
2423    use std::{sync::Arc, time::Duration};
2424
2425    use ahash::{AHashMap, AHashSet};
2426    use dashmap::DashMap;
2427    use nautilus_common::{
2428        clients::DataClient,
2429        live::{dst::time::Instant, runner::set_data_event_sender},
2430        messages::{
2431            DataEvent,
2432            data::{
2433                SubscribeBookDeltas, SubscribeCustomData, SubscribeFundingRates, SubscribeQuotes,
2434                SubscriptionError, UnsubscribeBookDeltas, UnsubscribeCustomData,
2435                UnsubscribeFundingRates, UnsubscribeQuotes,
2436            },
2437        },
2438        testing::wait_until_async,
2439    };
2440    use nautilus_core::{
2441        AtomicMap, AtomicSet, Params, UUID4, UnixNanos,
2442        time::{AtomicTime, get_atomic_clock_realtime},
2443    };
2444    use nautilus_live::{
2445        book::snapshot::SnapshotGate,
2446        task::{TaskGroup, TaskSpawner},
2447    };
2448    use nautilus_model::{
2449        data::{BarType, CustomData, Data, DataType, MarkPriceUpdate, QuoteTick},
2450        enums::{AggressorSide, BookAction, BookType, PositionSide},
2451        identifiers::InstrumentId,
2452        instruments::{Instrument, InstrumentAny},
2453        types::{Price, Quantity},
2454    };
2455    use rstest::rstest;
2456    use ustr::Ustr;
2457
2458    use super::{
2459        BybitDataClient, handle_orderbook_deltas, handle_ws_message, remove_owned_liquidation_sub,
2460        validate_orderbook_depth,
2461    };
2462    use crate::{
2463        book::{BookSequenceOutcome, sync::BookSyncTracker},
2464        common::{
2465            consts::BYBIT_CLIENT_ID,
2466            enums::{BybitProductType, BybitSymbolType},
2467            parse::{parse_linear_instrument, parse_option_instrument},
2468            rate_limit::BYBIT_OPTION_SUBSCRIPTION_LIMIT,
2469            testing::load_test_json,
2470        },
2471        config::BybitDataClientConfig,
2472        data_types::BybitLiquidation,
2473        http::models::{BybitInstrumentLinearResponse, BybitInstrumentOptionResponse},
2474        websocket::{
2475            client::BybitWebSocketClient,
2476            messages::{
2477                BybitWsKlineMsg, BybitWsLiquidationMsg, BybitWsMessage, BybitWsOrderbookDepthMsg,
2478                BybitWsTickerLinearMsg, BybitWsTickerOptionMsg, BybitWsTradeMsg,
2479            },
2480        },
2481    };
2482
2483    fn linear_instrument() -> InstrumentAny {
2484        let json = load_test_json("http_get_instruments_linear.json");
2485        let response: BybitInstrumentLinearResponse = serde_json::from_str(&json).unwrap();
2486        let instrument = &response.result.list[0];
2487        let ts = UnixNanos::new(1_700_000_000_000_000_000);
2488        parse_linear_instrument(instrument, ts, ts).unwrap()
2489    }
2490
2491    fn tradfi_linear_instrument() -> InstrumentAny {
2492        let json = load_test_json("http_get_instruments_linear.json");
2493        let response: BybitInstrumentLinearResponse = serde_json::from_str(&json).unwrap();
2494        let mut instrument = response.result.list[0].clone();
2495        instrument.symbol_type = Some(BybitSymbolType::Stock);
2496        let ts = UnixNanos::new(1_700_000_000_000_000_000);
2497        parse_linear_instrument(&instrument, ts, ts).unwrap()
2498    }
2499
2500    fn linear_instruments() -> Vec<InstrumentAny> {
2501        let json = load_test_json("http_get_instruments_linear.json");
2502        let response: BybitInstrumentLinearResponse = serde_json::from_str(&json).unwrap();
2503        let ts = UnixNanos::new(1_700_000_000_000_000_000);
2504        response
2505            .result
2506            .list
2507            .iter()
2508            .map(|instrument| parse_linear_instrument(instrument, ts, ts).unwrap())
2509            .collect()
2510    }
2511
2512    fn option_instrument() -> InstrumentAny {
2513        let json = load_test_json("http_get_instruments_option.json");
2514        let response: BybitInstrumentOptionResponse = serde_json::from_str(&json).unwrap();
2515        let instrument = &response.result.list[0];
2516        let ts = UnixNanos::new(1_700_000_000_000_000_000);
2517        parse_option_instrument(instrument, ts, ts).unwrap()
2518    }
2519
2520    fn build_instruments(instruments: &[InstrumentAny]) -> AHashMap<Ustr, InstrumentAny> {
2521        let mut map = AHashMap::new();
2522        for inst in instruments {
2523            map.insert(inst.id().symbol.inner(), inst.clone());
2524        }
2525        map
2526    }
2527
2528    #[expect(clippy::type_complexity)]
2529    fn empty_subs() -> (
2530        Arc<AtomicSet<InstrumentId>>,
2531        Arc<AtomicMap<InstrumentId, AHashSet<&'static str>>>,
2532        Arc<AtomicSet<InstrumentId>>,
2533        Arc<AtomicMap<InstrumentId, u32>>,
2534        Arc<AtomicSet<InstrumentId>>,
2535        Arc<AtomicMap<String, BarType>>,
2536    ) {
2537        (
2538            Arc::new(AtomicSet::new()),
2539            Arc::new(AtomicMap::new()),
2540            Arc::new(AtomicSet::new()),
2541            Arc::new(AtomicMap::new()),
2542            Arc::new(AtomicSet::new()),
2543            Arc::new(AtomicMap::new()),
2544        )
2545    }
2546
2547    fn book_context() -> (BookSyncTracker, BybitWebSocketClient, TaskSpawner) {
2548        let tasks = TaskGroup::new().spawner().unwrap();
2549        let ws_client = BybitWebSocketClient::new_public(None, 20);
2550        (BookSyncTracker::default(), ws_client, tasks)
2551    }
2552
2553    fn liquidation_data_type(instrument_id: InstrumentId, identifier: Option<&str>) -> DataType {
2554        let metadata = Params::new(
2555            [(
2556                "instrument_id".to_string(),
2557                serde_json::Value::String(instrument_id.to_string()),
2558            )]
2559            .into(),
2560        );
2561
2562        DataType::new(
2563            "BybitLiquidation",
2564            Some(metadata),
2565            identifier.map(ToString::to_string),
2566        )
2567    }
2568
2569    fn subscribe_custom_data(data_type: DataType) -> SubscribeCustomData {
2570        SubscribeCustomData::new(
2571            Some(*BYBIT_CLIENT_ID),
2572            None,
2573            data_type,
2574            UUID4::new(),
2575            UnixNanos::default(),
2576            None,
2577            None,
2578        )
2579    }
2580
2581    fn unsubscribe_custom_data(data_type: DataType) -> UnsubscribeCustomData {
2582        UnsubscribeCustomData::new(
2583            Some(*BYBIT_CLIENT_ID),
2584            None,
2585            data_type,
2586            UUID4::new(),
2587            UnixNanos::default(),
2588            None,
2589            None,
2590        )
2591    }
2592
2593    fn handle_liquidation_message(
2594        msg: BybitWsLiquidationMsg,
2595        instruments: &AHashMap<Ustr, InstrumentAny>,
2596        liquidation_subs: &Arc<AtomicMap<InstrumentId, Arc<DataType>>>,
2597        ts_init: UnixNanos,
2598    ) -> Vec<CustomData> {
2599        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
2600            empty_subs();
2601        let (book_sync, ws_client, tasks) = book_context();
2602
2603        let mut funding_cache = AHashMap::new();
2604        let clock = AtomicTime::new(false, ts_init);
2605        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2606
2607        handle_ws_message(
2608            &BybitWsMessage::Liquidation(msg),
2609            &tx.into(),
2610            instruments,
2611            Some(BybitProductType::Linear),
2612            &trade_subs,
2613            &ticker_subs,
2614            &quote_subs,
2615            &book_depths,
2616            &book_sync,
2617            &ws_client,
2618            &greeks_subs,
2619            liquidation_subs,
2620            &bar_types,
2621            true,
2622            &DashMap::new(),
2623            &mut funding_cache,
2624            Duration::ZERO,
2625            &tasks,
2626            &clock,
2627        );
2628
2629        let mut events = Vec::new();
2630
2631        while let Ok(event) = rx.try_recv() {
2632            match event {
2633                DataEvent::Data(Data::Custom(custom)) => events.push(custom),
2634                other => panic!("Expected custom data event, found {other:?}"),
2635            }
2636        }
2637
2638        events
2639    }
2640
2641    fn as_liquidation(custom: &CustomData) -> &BybitLiquidation {
2642        custom
2643            .data
2644            .as_any()
2645            .downcast_ref::<BybitLiquidation>()
2646            .expect("expected BybitLiquidation")
2647    }
2648
2649    #[tokio::test]
2650    async fn test_failed_book_subscription_releases_registration() {
2651        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
2652        set_data_event_sender(tx);
2653        let config = BybitDataClientConfig {
2654            product_types: vec![BybitProductType::Option],
2655            ..Default::default()
2656        };
2657        let mut client = BybitDataClient::new(*BYBIT_CLIENT_ID, config).unwrap();
2658        let ws = client.ws_clients[0].clone();
2659        let topics = (0..BYBIT_OPTION_SUBSCRIPTION_LIMIT)
2660            .map(|index| format!("tickers.OPTION-{index}"))
2661            .collect();
2662
2663        // The disconnected command channel rejects the send after registering these topics
2664        ws.subscribe(topics).await.unwrap_err();
2665        let instrument_id = InstrumentId::from("ETH-26JUN26-16000-P-OPTION.BYBIT");
2666        let expected_error = format!(
2667            "Client error: Option WebSocket subscription limit is {BYBIT_OPTION_SUBSCRIPTION_LIMIT} arguments per connection, requested {}",
2668            BYBIT_OPTION_SUBSCRIPTION_LIMIT + 1,
2669        );
2670        assert_eq!(
2671            ws.subscribe_orderbook(instrument_id, 50)
2672                .await
2673                .unwrap_err()
2674                .to_string(),
2675            expected_error,
2676        );
2677        client
2678            .instruments
2679            .insert(instrument_id, option_instrument());
2680        let command = SubscribeBookDeltas::new(
2681            instrument_id,
2682            BookType::L2_MBP,
2683            Some(*BYBIT_CLIENT_ID),
2684            None,
2685            UUID4::new(),
2686            UnixNanos::default(),
2687            None,
2688            false,
2689            None,
2690            None,
2691        );
2692
2693        for _ in 0..2 {
2694            client.subscribe_book_deltas(command.clone()).unwrap();
2695            wait_until_async(
2696                || async { !client.book_depths.contains_key(&instrument_id) },
2697                Duration::from_secs(2),
2698            )
2699            .await;
2700            assert!(!client.book_depths.contains_key(&instrument_id));
2701            assert_eq!(
2702                ws.subscribe_orderbook(instrument_id, 50)
2703                    .await
2704                    .unwrap_err()
2705                    .to_string(),
2706                expected_error,
2707            );
2708        }
2709    }
2710
2711    #[tokio::test]
2712    async fn test_failed_book_subscription_releases_transport_reference() {
2713        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
2714        set_data_event_sender(tx);
2715        let mut client =
2716            BybitDataClient::new(*BYBIT_CLIENT_ID, BybitDataClientConfig::default()).unwrap();
2717        let ws = client.ws_clients[0].clone();
2718        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
2719        client
2720            .subscribe_book_deltas(SubscribeBookDeltas::new(
2721                instrument_id,
2722                BookType::L2_MBP,
2723                Some(*BYBIT_CLIENT_ID),
2724                None,
2725                UUID4::new(),
2726                UnixNanos::default(),
2727                None,
2728                false,
2729                None,
2730                None,
2731            ))
2732            .unwrap();
2733        wait_until_async(
2734            || async { !client.book_depths.contains_key(&instrument_id) },
2735            Duration::from_secs(2),
2736        )
2737        .await;
2738
2739        let error = ws.subscribe_orderbook(instrument_id, 50).await.unwrap_err();
2740
2741        assert_eq!(
2742            error.to_string(),
2743            "WebSocket send error: Failed to send subscribe command: channel closed",
2744        );
2745        assert!(!client.book_depths.contains_key(&instrument_id));
2746    }
2747
2748    #[tokio::test]
2749    async fn test_failed_liquidation_subscription_releases_registration() {
2750        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
2751        set_data_event_sender(tx);
2752        let mut client =
2753            BybitDataClient::new(*BYBIT_CLIENT_ID, BybitDataClientConfig::default()).unwrap();
2754        let ws = client.ws_clients[0].clone();
2755        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
2756        let command = subscribe_custom_data(liquidation_data_type(instrument_id, None));
2757
2758        // The disconnected command channel rejects each send, so a retry must subscribe again
2759        for _ in 0..2 {
2760            client.subscribe(command.clone()).unwrap();
2761            wait_until_async(
2762                || async { !client.liquidation_subs.contains_key(&instrument_id) },
2763                Duration::from_secs(2),
2764            )
2765            .await;
2766        }
2767
2768        let error = ws.subscribe_liquidations(instrument_id).await.unwrap_err();
2769
2770        assert_eq!(
2771            error.to_string(),
2772            "WebSocket send error: Failed to send subscribe command: channel closed",
2773        );
2774        assert!(!client.liquidation_subs.contains_key(&instrument_id));
2775    }
2776
2777    #[rstest]
2778    #[tokio::test]
2779    async fn test_failed_quote_subscription_releases_registration() {
2780        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
2781        set_data_event_sender(tx);
2782        let mut client =
2783            BybitDataClient::new(*BYBIT_CLIENT_ID, BybitDataClientConfig::default()).unwrap();
2784        let ws = client.ws_clients[0].clone();
2785        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
2786        let command = SubscribeQuotes::new(
2787            instrument_id,
2788            Some(*BYBIT_CLIENT_ID),
2789            None,
2790            UUID4::new(),
2791            UnixNanos::default(),
2792            None,
2793            None,
2794        );
2795
2796        // The disconnected command channel rejects each send, so a retry must subscribe again
2797        for _ in 0..2 {
2798            client.subscribe_quotes(command.clone()).unwrap();
2799            wait_until_async(
2800                || async { !client.quote_subs.contains(&instrument_id) },
2801                Duration::from_secs(2),
2802            )
2803            .await;
2804        }
2805
2806        let error = ws.subscribe_orderbook(instrument_id, 1).await.unwrap_err();
2807
2808        assert_eq!(
2809            error.to_string(),
2810            "WebSocket send error: Failed to send subscribe command: channel closed",
2811        );
2812        assert!(!client.quote_subs.contains(&instrument_id));
2813    }
2814
2815    #[rstest]
2816    fn test_liquidation_subscription_keeps_first_data_type() {
2817        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
2818        set_data_event_sender(tx);
2819        let mut client =
2820            BybitDataClient::new(*BYBIT_CLIENT_ID, BybitDataClientConfig::default()).unwrap();
2821        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
2822        let data_type = liquidation_data_type(instrument_id, None);
2823        let other_data_type = liquidation_data_type(instrument_id, Some("other"));
2824        let subscription = Arc::new(data_type.clone());
2825        client
2826            .liquidation_subs
2827            .insert(instrument_id, Arc::clone(&subscription));
2828
2829        let error = client
2830            .subscribe(subscribe_custom_data(other_data_type.clone()))
2831            .unwrap_err();
2832        client
2833            .unsubscribe(&unsubscribe_custom_data(other_data_type))
2834            .unwrap();
2835        client
2836            .subscribe(subscribe_custom_data(data_type.clone()))
2837            .unwrap();
2838
2839        assert_eq!(
2840            error.to_string(),
2841            format!(
2842                "Already subscribed to liquidations for BTCUSDT-LINEAR.BYBIT as {}",
2843                data_type.topic()
2844            ),
2845        );
2846        assert!(
2847            client
2848                .liquidation_subs
2849                .get_cloned(&instrument_id)
2850                .is_some_and(|current| Arc::ptr_eq(&current, &subscription))
2851        );
2852    }
2853
2854    #[rstest]
2855    fn test_remove_owned_liquidation_sub_removes_own_entry() {
2856        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
2857        let owned = Arc::new(liquidation_data_type(instrument_id, None));
2858        let subs = AtomicMap::new();
2859        subs.insert(instrument_id, Arc::clone(&owned));
2860
2861        let removed = remove_owned_liquidation_sub(&subs, instrument_id, &owned);
2862
2863        assert!(removed);
2864        assert!(subs.is_empty());
2865    }
2866
2867    #[rstest]
2868    fn test_remove_owned_liquidation_sub_skips_entry_removed_by_unsubscribe() {
2869        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
2870        let owned = Arc::new(liquidation_data_type(instrument_id, None));
2871        let subs = AtomicMap::new();
2872
2873        let removed = remove_owned_liquidation_sub(&subs, instrument_id, &owned);
2874
2875        assert!(!removed);
2876        assert!(subs.is_empty());
2877    }
2878
2879    #[rstest]
2880    #[case::identical_data_type(None)]
2881    #[case::other_data_type(Some("other"))]
2882    fn test_remove_owned_liquidation_sub_keeps_later_subscription(
2883        #[case] identifier: Option<&str>,
2884    ) {
2885        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
2886        let owned = Arc::new(liquidation_data_type(instrument_id, None));
2887        let later = Arc::new(liquidation_data_type(instrument_id, identifier));
2888        let subs = AtomicMap::new();
2889        subs.insert(instrument_id, Arc::clone(&later));
2890
2891        let removed = remove_owned_liquidation_sub(&subs, instrument_id, &owned);
2892
2893        assert!(!removed);
2894        assert_eq!(subs.len(), 1);
2895        assert!(
2896            subs.get_cloned(&instrument_id)
2897                .is_some_and(|current| Arc::ptr_eq(&current, &later))
2898        );
2899    }
2900
2901    #[rstest]
2902    fn test_validate_orderbook_depth_accepts_1000() {
2903        assert!(validate_orderbook_depth(1000).is_ok());
2904    }
2905
2906    #[rstest]
2907    fn test_validate_orderbook_depth_rejects_500() {
2908        let e = validate_orderbook_depth(500).unwrap_err();
2909
2910        assert_eq!(
2911            e.to_string(),
2912            "invalid depth 500; valid values are [1, 50, 200, 1000]"
2913        );
2914    }
2915
2916    #[rstest]
2917    #[case::open(false, true, Some(1_672_324_800_000_000_000))]
2918    #[case::close(true, true, Some(1_672_325_100_000_000_000))]
2919    #[case::unconfirmed_open(false, false, None)]
2920    #[case::unconfirmed_close(true, false, None)]
2921    fn test_handle_kline_bar_timestamp(
2922        #[case] on_close: bool,
2923        #[case] confirmed: bool,
2924        #[case] expected: Option<u64>,
2925    ) {
2926        let instrument = linear_instrument();
2927        let instruments = build_instruments(std::slice::from_ref(&instrument));
2928        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
2929            empty_subs();
2930        let (book_sync, ws_client, tasks) = book_context();
2931        let bar_type = BarType::from("BTCUSDT-LINEAR.BYBIT-5-MINUTE-LAST-EXTERNAL");
2932        bar_types.insert("kline.5.BTCUSDT".to_string(), bar_type);
2933        let mut msg: BybitWsKlineMsg =
2934            serde_json::from_str(&load_test_json("ws_kline.json")).unwrap();
2935        msg.data[0].confirm = confirmed;
2936
2937        let mut funding_cache = AHashMap::new();
2938        let clock = get_atomic_clock_realtime();
2939        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2940        handle_ws_message(
2941            &BybitWsMessage::Kline(msg),
2942            &tx.into(),
2943            &instruments,
2944            Some(BybitProductType::Linear),
2945            &trade_subs,
2946            &ticker_subs,
2947            &quote_subs,
2948            &book_depths,
2949            &book_sync,
2950            &ws_client,
2951            &greeks_subs,
2952            &Arc::new(AtomicMap::new()),
2953            &bar_types,
2954            on_close,
2955            &DashMap::new(),
2956            &mut funding_cache,
2957            Duration::ZERO,
2958            &tasks,
2959            clock,
2960        );
2961
2962        if let Some(expected) = expected {
2963            let DataEvent::Data(Data::Bar(bar)) = rx.try_recv().unwrap() else {
2964                panic!("Expected bar event");
2965            };
2966            assert_eq!(bar.bar_type, bar_type);
2967            assert_eq!(bar.ts_event, UnixNanos::from(expected));
2968        }
2969        assert!(rx.try_recv().is_err());
2970    }
2971
2972    #[rstest]
2973    fn test_handle_trade_message_emits_trade_tick() {
2974        let instrument = linear_instrument();
2975        let instruments = build_instruments(std::slice::from_ref(&instrument));
2976        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
2977            empty_subs();
2978        let (book_sync, ws_client, tasks) = book_context();
2979        trade_subs.insert(instrument.id());
2980
2981        let mut funding_cache = AHashMap::new();
2982        let clock = get_atomic_clock_realtime();
2983
2984        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
2985
2986        let json = load_test_json("ws_public_trade.json");
2987        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
2988        let ws_msg = BybitWsMessage::Trade(msg);
2989
2990        handle_ws_message(
2991            &ws_msg,
2992            &tx.into(),
2993            &instruments,
2994            Some(BybitProductType::Linear),
2995            &trade_subs,
2996            &ticker_subs,
2997            &quote_subs,
2998            &book_depths,
2999            &book_sync,
3000            &ws_client,
3001            &greeks_subs,
3002            &Arc::new(AtomicMap::new()),
3003            &bar_types,
3004            true,
3005            &DashMap::new(),
3006            &mut funding_cache,
3007            Duration::ZERO,
3008            &tasks,
3009            clock,
3010        );
3011
3012        let event = rx.try_recv().unwrap();
3013        match event {
3014            DataEvent::Data(Data::Trade(tick)) => {
3015                assert_eq!(tick.instrument_id, instrument.id());
3016                assert_eq!(tick.price, instrument.make_price(27451.00));
3017                assert_eq!(tick.size, instrument.make_qty(0.010, None));
3018                assert_eq!(tick.aggressor_side, AggressorSide::Buy);
3019            }
3020            other => panic!("Expected Trade data event, found {other:?}"),
3021        }
3022    }
3023
3024    #[rstest]
3025    fn test_handle_trade_message_unknown_symbol_no_event() {
3026        let instruments = AHashMap::new();
3027        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3028            empty_subs();
3029        let (book_sync, ws_client, tasks) = book_context();
3030
3031        let mut funding_cache = AHashMap::new();
3032        let clock = get_atomic_clock_realtime();
3033
3034        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3035
3036        let json = load_test_json("ws_public_trade.json");
3037        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
3038        let ws_msg = BybitWsMessage::Trade(msg);
3039
3040        handle_ws_message(
3041            &ws_msg,
3042            &tx.into(),
3043            &instruments,
3044            Some(BybitProductType::Linear),
3045            &trade_subs,
3046            &ticker_subs,
3047            &quote_subs,
3048            &book_depths,
3049            &book_sync,
3050            &ws_client,
3051            &greeks_subs,
3052            &Arc::new(AtomicMap::new()),
3053            &bar_types,
3054            true,
3055            &DashMap::new(),
3056            &mut funding_cache,
3057            Duration::ZERO,
3058            &tasks,
3059            clock,
3060        );
3061
3062        rx.try_recv().unwrap_err();
3063    }
3064
3065    #[rstest]
3066    #[case::metadata_only(None)]
3067    #[case::with_identifier(Some("bybit-liquidations"))]
3068    fn test_handle_liquidation_message_emits_subscribed_data_type(
3069        #[case] identifier: Option<&str>,
3070    ) {
3071        let instrument = linear_instrument();
3072        let instrument_id = instrument.id();
3073        let instruments = build_instruments(&[instrument]);
3074        let data_type = liquidation_data_type(instrument_id, identifier);
3075        let liquidation_subs = Arc::new(AtomicMap::new());
3076        liquidation_subs.insert(instrument_id, Arc::new(data_type.clone()));
3077        let ts_init = UnixNanos::new(1_739_502_303_500_000_000);
3078        let msg: BybitWsLiquidationMsg =
3079            serde_json::from_str(&load_test_json("ws_all_liquidation.json")).unwrap();
3080
3081        let events = handle_liquidation_message(msg, &instruments, &liquidation_subs, ts_init);
3082
3083        assert_eq!(events.len(), 2);
3084        assert_eq!(events[0].data_type, data_type);
3085        assert_eq!(events[0].data_type.topic(), data_type.topic());
3086        assert_eq!(events[0].data_type.identifier(), identifier);
3087        assert_eq!(events[1].data_type, data_type);
3088        assert_eq!(
3089            as_liquidation(&events[0]),
3090            &BybitLiquidation::new(
3091                instrument_id,
3092                PositionSide::Long,
3093                Price::from("96250.5"),
3094                Quantity::from("0.015"),
3095                UnixNanos::new(1_739_502_302_929_000_000),
3096                ts_init,
3097            ),
3098        );
3099        assert_eq!(
3100            as_liquidation(&events[1]),
3101            &BybitLiquidation::new(
3102                instrument_id,
3103                PositionSide::Short,
3104                Price::from("97410.0"),
3105                Quantity::from("1.250"),
3106                UnixNanos::new(1_739_502_303_011_000_000),
3107                ts_init,
3108            ),
3109        );
3110    }
3111
3112    #[rstest]
3113    #[case::unsubscribed_instrument("ETHUSDT")]
3114    #[case::uncached_symbol("SOLUSDT")]
3115    fn test_handle_liquidation_message_skips_element(#[case] skipped_symbol: &str) {
3116        let instruments = linear_instruments();
3117        let instrument_id = instruments[0].id();
3118        let instruments = build_instruments(&instruments);
3119        let data_type = liquidation_data_type(instrument_id, None);
3120        let liquidation_subs = Arc::new(AtomicMap::new());
3121        liquidation_subs.insert(instrument_id, Arc::new(data_type.clone()));
3122        let ts_init = UnixNanos::new(1_739_502_303_500_000_000);
3123        let mut value: serde_json::Value =
3124            serde_json::from_str(&load_test_json("ws_all_liquidation.json")).unwrap();
3125        value["data"][1]["s"] = serde_json::json!(skipped_symbol);
3126        let msg: BybitWsLiquidationMsg = serde_json::from_value(value).unwrap();
3127
3128        let events = handle_liquidation_message(msg, &instruments, &liquidation_subs, ts_init);
3129
3130        assert_eq!(instruments.len(), 2);
3131        assert_eq!(events.len(), 1);
3132        assert_eq!(events[0].data_type, data_type);
3133        assert_eq!(
3134            as_liquidation(&events[0]),
3135            &BybitLiquidation::new(
3136                instrument_id,
3137                PositionSide::Long,
3138                Price::from("96250.5"),
3139                Quantity::from("0.015"),
3140                UnixNanos::new(1_739_502_302_929_000_000),
3141                ts_init,
3142            ),
3143        );
3144    }
3145
3146    #[rstest]
3147    fn test_handle_liquidation_message_skips_invalid_element() {
3148        let instrument = linear_instrument();
3149        let instrument_id = instrument.id();
3150        let instruments = build_instruments(&[instrument]);
3151        let data_type = liquidation_data_type(instrument_id, None);
3152        let liquidation_subs = Arc::new(AtomicMap::new());
3153        liquidation_subs.insert(instrument_id, Arc::new(data_type.clone()));
3154        let ts_init = UnixNanos::new(1_739_502_303_500_000_000);
3155        let mut value: serde_json::Value =
3156            serde_json::from_str(&load_test_json("ws_all_liquidation.json")).unwrap();
3157        value["data"][0]["S"] = serde_json::json!("");
3158        let msg: BybitWsLiquidationMsg = serde_json::from_value(value).unwrap();
3159
3160        let events = handle_liquidation_message(msg, &instruments, &liquidation_subs, ts_init);
3161
3162        assert_eq!(events.len(), 1);
3163        assert_eq!(events[0].data_type, data_type);
3164        assert_eq!(
3165            as_liquidation(&events[0]),
3166            &BybitLiquidation::new(
3167                instrument_id,
3168                PositionSide::Short,
3169                Price::from("97410.0"),
3170                Quantity::from("1.250"),
3171                UnixNanos::new(1_739_502_303_011_000_000),
3172                ts_init,
3173            ),
3174        );
3175    }
3176
3177    #[rstest]
3178    fn test_handle_liquidation_message_without_subscription_no_event() {
3179        let instrument = linear_instrument();
3180        let instruments = build_instruments(&[instrument]);
3181        let liquidation_subs = Arc::new(AtomicMap::new());
3182        let msg: BybitWsLiquidationMsg =
3183            serde_json::from_str(&load_test_json("ws_all_liquidation.json")).unwrap();
3184
3185        let events = handle_liquidation_message(
3186            msg,
3187            &instruments,
3188            &liquidation_subs,
3189            UnixNanos::new(1_739_502_303_500_000_000),
3190        );
3191
3192        assert!(events.is_empty());
3193    }
3194
3195    #[rstest]
3196    #[case::shared_depth_one("orderbook.1.BTCUSDT", 1, "snapshot", true, true)]
3197    #[case::depth_one_with_deeper_book("orderbook.1.BTCUSDT", 50, "snapshot", false, true)]
3198    #[case::deeper_snapshot("orderbook.50.BTCUSDT", 50, "snapshot", true, false)]
3199    #[case::deeper_deletion("orderbook.50.BTCUSDT", 50, "delta", true, false)]
3200    #[case::unsubscribed_depth("orderbook.200.BTCUSDT", 50, "delta", false, false)]
3201    #[case::invalid_topic("orderbook.invalid.BTCUSDT", 1, "snapshot", false, false)]
3202    #[case::wrong_symbol("orderbook.1.ETHUSDT", 1, "snapshot", false, false)]
3203    fn test_handle_orderbook_routes_by_depth(
3204        #[case] topic: &str,
3205        #[case] book_depth: u32,
3206        #[case] msg_type: &str,
3207        #[case] expect_book: bool,
3208        #[case] expect_quote: bool,
3209    ) {
3210        let instrument = linear_instrument();
3211        let instrument_id = instrument.id();
3212        let instruments = build_instruments(&[instrument]);
3213        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3214            empty_subs();
3215        let (book_sync, ws_client, tasks) = book_context();
3216        book_depths.insert(instrument_id, book_depth);
3217        book_sync.record_subscription(instrument_id, Instant::now(), SnapshotGate::default());
3218
3219        // The delta fixture follows the snapshot fixture's update ID
3220        let synced = book_sync.validate(instrument_id, true, 123_456_789, Instant::now());
3221        assert_eq!(synced, BookSequenceOutcome::Accept);
3222        quote_subs.insert(instrument_id);
3223
3224        let mut funding_cache = AHashMap::new();
3225        let ts_init = UnixNanos::new(1_709_891_680_000_000_000);
3226        let clock = AtomicTime::new(false, ts_init);
3227        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3228        let fixture = if msg_type == "snapshot" {
3229            "ws_orderbook_snapshot.json"
3230        } else {
3231            "ws_orderbook_delta.json"
3232        };
3233        let mut msg: BybitWsOrderbookDepthMsg =
3234            serde_json::from_str(&load_test_json(fixture)).unwrap();
3235        msg.topic = topic.into();
3236
3237        handle_ws_message(
3238            &BybitWsMessage::Orderbook(msg),
3239            &tx.into(),
3240            &instruments,
3241            Some(BybitProductType::Linear),
3242            &trade_subs,
3243            &ticker_subs,
3244            &quote_subs,
3245            &book_depths,
3246            &book_sync,
3247            &ws_client,
3248            &greeks_subs,
3249            &Arc::new(AtomicMap::new()),
3250            &bar_types,
3251            true,
3252            &DashMap::new(),
3253            &mut funding_cache,
3254            Duration::ZERO,
3255            &tasks,
3256            &clock,
3257        );
3258
3259        if expect_book {
3260            let DataEvent::Data(Data::BookDeltas(deltas)) = rx.try_recv().unwrap() else {
3261                panic!("Expected book deltas");
3262            };
3263            assert_eq!(deltas.instrument_id, instrument_id);
3264            assert_eq!(
3265                deltas.deltas[0].action == BookAction::Clear,
3266                msg_type == "snapshot"
3267            );
3268
3269            if msg_type == "delta" {
3270                assert_eq!(deltas.deltas.len(), 2);
3271                assert_eq!(deltas.deltas[1].action, BookAction::Delete);
3272                assert_eq!(deltas.deltas[1].order.size, Quantity::from("0.000"));
3273            }
3274        }
3275
3276        if expect_quote {
3277            let expected = QuoteTick::new(
3278                instrument_id,
3279                Price::from("27450.00"),
3280                Price::from("27451.00"),
3281                Quantity::from("0.500"),
3282                Quantity::from("0.750"),
3283                UnixNanos::new(1_709_891_679_000_000_000),
3284                ts_init,
3285            );
3286            let DataEvent::Data(Data::Quote(quote)) = rx.try_recv().unwrap() else {
3287                panic!("Expected quote");
3288            };
3289            assert_eq!(quote, expected);
3290        }
3291
3292        rx.try_recv().unwrap_err();
3293    }
3294
3295    fn orderbook_msg(msg_type: &str, update_id: i64, levels: bool) -> BybitWsOrderbookDepthMsg {
3296        let fixture = if msg_type == "snapshot" {
3297            "ws_orderbook_snapshot.json"
3298        } else {
3299            "ws_orderbook_delta.json"
3300        };
3301
3302        let mut msg: BybitWsOrderbookDepthMsg =
3303            serde_json::from_str(&load_test_json(fixture)).unwrap();
3304        msg.topic = "orderbook.50.BTCUSDT".into();
3305        msg.data.u = update_id;
3306
3307        if !levels {
3308            msg.data.b.clear();
3309            msg.data.a.clear();
3310        }
3311
3312        msg
3313    }
3314
3315    #[rstest]
3316    fn test_handle_orderbook_delta_without_levels_advances_book() {
3317        let instrument = linear_instrument();
3318        let instrument_id = instrument.id();
3319        let (_, _, _, book_depths, _, _) = empty_subs();
3320        let (book_sync, ws_client, tasks) = book_context();
3321        book_depths.insert(instrument_id, 50);
3322        book_sync.record_subscription(instrument_id, Instant::now(), SnapshotGate::default());
3323        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3324        let sender = tx.into();
3325        let ts_init = UnixNanos::new(1);
3326
3327        for msg in [
3328            orderbook_msg("snapshot", 100, true),
3329            orderbook_msg("delta", 101, false),
3330            orderbook_msg("delta", 102, true),
3331        ] {
3332            handle_orderbook_deltas(
3333                &msg,
3334                &instrument,
3335                &sender,
3336                &book_depths,
3337                &book_sync,
3338                &ws_client,
3339                Duration::ZERO,
3340                &tasks,
3341                ts_init,
3342            );
3343        }
3344
3345        let mut emitted = Vec::new();
3346
3347        while let Ok(DataEvent::Data(Data::BookDeltas(deltas))) = rx.try_recv() {
3348            emitted.push(deltas.deltas[0].action);
3349        }
3350
3351        assert_eq!(emitted, [BookAction::Clear, BookAction::Update]);
3352    }
3353
3354    #[rstest]
3355    #[case::gap(102, "27450.00")]
3356    #[case::malformed_level(101, "not-a-price")]
3357    #[case::negative_update_id(-1, "27450.00")]
3358    #[tokio::test]
3359    async fn test_handle_orderbook_fault_suppresses_output_and_claims_recovery(
3360        #[case] update_id: i64,
3361        #[case] bid_price: &str,
3362    ) {
3363        let instrument = linear_instrument();
3364        let instrument_id = instrument.id();
3365        let (_, _, _, book_depths, _, _) = empty_subs();
3366        let book_sync = BookSyncTracker::default();
3367        let ws_client = BybitWebSocketClient::new_public(None, 20);
3368        let tasks = TaskGroup::new();
3369        let spawner = tasks.spawner().unwrap();
3370        book_depths.insert(instrument_id, 50);
3371        book_sync.record_subscription(instrument_id, Instant::now(), SnapshotGate::default());
3372        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3373        let sender = tx.into();
3374        let mut fault = orderbook_msg("delta", update_id, true);
3375        fault.data.b[0][0] = bid_price.to_string();
3376
3377        for msg in [orderbook_msg("snapshot", 100, true), fault] {
3378            handle_orderbook_deltas(
3379                &msg,
3380                &instrument,
3381                &sender,
3382                &book_depths,
3383                &book_sync,
3384                &ws_client,
3385                Duration::from_secs(1),
3386                &spawner,
3387                UnixNanos::new(1),
3388            );
3389        }
3390
3391        let DataEvent::Data(Data::BookDeltas(snapshot)) = rx.try_recv().unwrap() else {
3392            panic!("Expected book snapshot");
3393        };
3394
3395        assert_eq!(snapshot.deltas[0].action, BookAction::Clear);
3396        rx.try_recv().unwrap_err();
3397        assert!(book_sync.claim_recovery(instrument_id).is_none());
3398        tasks.begin_shutdown();
3399    }
3400
3401    #[tokio::test]
3402    async fn test_handle_reconnected_resyncs_books_of_reconnected_socket() {
3403        let instrument = linear_instrument();
3404        let instrument_id = instrument.id();
3405        let instruments = build_instruments(std::slice::from_ref(&instrument));
3406        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3407            empty_subs();
3408        let book_sync = BookSyncTracker::default();
3409        let ws_client = BybitWebSocketClient::new_public(None, 20);
3410        let tasks = TaskGroup::new();
3411        let spawner = tasks.spawner().unwrap();
3412        book_depths.insert(instrument_id, 50);
3413        book_sync.record_subscription(instrument_id, Instant::now(), SnapshotGate::default());
3414        let synced = book_sync.validate(instrument_id, true, 100, Instant::now());
3415        assert_eq!(synced, BookSequenceOutcome::Accept);
3416
3417        let mut funding_cache = AHashMap::new();
3418        let clock = get_atomic_clock_realtime();
3419        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3420        let sender = tx.into();
3421        let snapshot_timeout = Duration::from_secs(10);
3422
3423        // The update follows the book's position, so only the reconnect can suppress it
3424        for message in [
3425            BybitWsMessage::Reconnected,
3426            BybitWsMessage::Orderbook(orderbook_msg("delta", 101, true)),
3427        ] {
3428            handle_ws_message(
3429                &message,
3430                &sender,
3431                &instruments,
3432                Some(BybitProductType::Linear),
3433                &trade_subs,
3434                &ticker_subs,
3435                &quote_subs,
3436                &book_depths,
3437                &book_sync,
3438                &ws_client,
3439                &greeks_subs,
3440                &Arc::new(AtomicMap::new()),
3441                &bar_types,
3442                true,
3443                &DashMap::new(),
3444                &mut funding_cache,
3445                snapshot_timeout,
3446                &spawner,
3447                clock,
3448            );
3449        }
3450
3451        let deadline = Instant::now() + snapshot_timeout;
3452        let expired = book_sync.take_expired_snapshots(BybitProductType::Linear, deadline);
3453
3454        rx.try_recv().unwrap_err();
3455        assert_eq!(expired.len(), 1);
3456        assert_eq!(expired[0].instrument_id, instrument_id);
3457        tasks.begin_shutdown();
3458    }
3459
3460    #[rstest]
3461    #[case::unsubscribe(false)]
3462    #[case::reset(true)]
3463    #[tokio::test]
3464    async fn test_book_removal_cancels_running_recovery(#[case] reset: bool) {
3465        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
3466        set_data_event_sender(tx);
3467        let mut client =
3468            BybitDataClient::new(*BYBIT_CLIENT_ID, BybitDataClientConfig::default()).unwrap();
3469        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
3470        client.book_depths.insert(instrument_id, 50);
3471        client.book_sync.record_subscription(
3472            instrument_id,
3473            Instant::now(),
3474            SnapshotGate::default(),
3475        );
3476        let recovery = client.book_sync.claim_recovery(instrument_id).unwrap();
3477
3478        if reset {
3479            client.reset().unwrap();
3480        } else {
3481            let command = UnsubscribeBookDeltas::new(
3482                instrument_id,
3483                Some(*BYBIT_CLIENT_ID),
3484                None,
3485                UUID4::new(),
3486                UnixNanos::default(),
3487                None,
3488                None,
3489            );
3490            client.unsubscribe_book_deltas(&command).unwrap();
3491        }
3492
3493        assert!(recovery.cancellation.is_cancelled());
3494        assert!(!client.book_depths.contains_key(&instrument_id));
3495    }
3496
3497    #[rstest]
3498    fn test_handle_orderbook_message_no_sub_no_event() {
3499        let instrument = linear_instrument();
3500        let instruments = build_instruments(&[instrument]);
3501        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3502            empty_subs();
3503        let (book_sync, ws_client, tasks) = book_context();
3504
3505        let mut funding_cache = AHashMap::new();
3506        let clock = get_atomic_clock_realtime();
3507
3508        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3509
3510        let json = load_test_json("ws_orderbook_snapshot.json");
3511        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
3512        let ws_msg = BybitWsMessage::Orderbook(msg);
3513
3514        handle_ws_message(
3515            &ws_msg,
3516            &tx.into(),
3517            &instruments,
3518            Some(BybitProductType::Linear),
3519            &trade_subs,
3520            &ticker_subs,
3521            &quote_subs,
3522            &book_depths,
3523            &book_sync,
3524            &ws_client,
3525            &greeks_subs,
3526            &Arc::new(AtomicMap::new()),
3527            &bar_types,
3528            true,
3529            &DashMap::new(),
3530            &mut funding_cache,
3531            Duration::ZERO,
3532            &tasks,
3533            clock,
3534        );
3535
3536        rx.try_recv().unwrap_err();
3537    }
3538
3539    #[rstest]
3540    fn test_handle_ticker_linear_does_not_emit_quote() {
3541        let instrument = linear_instrument();
3542        let instrument_id = instrument.id();
3543        let instruments = build_instruments(&[instrument]);
3544        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3545            empty_subs();
3546        let (book_sync, ws_client, tasks) = book_context();
3547
3548        let mut subs = AHashSet::new();
3549        subs.insert("quotes");
3550        ticker_subs.insert(instrument_id, subs);
3551
3552        let mut funding_cache = AHashMap::new();
3553        let clock = get_atomic_clock_realtime();
3554
3555        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3556
3557        let json = load_test_json("ws_ticker_linear.json");
3558        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
3559        let ws_msg = BybitWsMessage::TickerLinear(msg);
3560
3561        handle_ws_message(
3562            &ws_msg,
3563            &tx.into(),
3564            &instruments,
3565            Some(BybitProductType::Linear),
3566            &trade_subs,
3567            &ticker_subs,
3568            &quote_subs,
3569            &book_depths,
3570            &book_sync,
3571            &ws_client,
3572            &greeks_subs,
3573            &Arc::new(AtomicMap::new()),
3574            &bar_types,
3575            true,
3576            &DashMap::new(),
3577            &mut funding_cache,
3578            Duration::ZERO,
3579            &tasks,
3580            clock,
3581        );
3582
3583        rx.try_recv().unwrap_err();
3584    }
3585
3586    #[rstest]
3587    #[case::crypto_perpetual(linear_instrument())]
3588    #[case::perpetual_contract(tradfi_linear_instrument())]
3589    fn test_handle_ticker_linear_funding_dedup(#[case] instrument: InstrumentAny) {
3590        let instrument_id = instrument.id();
3591        let instruments = build_instruments(&[instrument]);
3592        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3593            empty_subs();
3594        let (book_sync, ws_client, tasks) = book_context();
3595
3596        let mut subs = AHashSet::new();
3597        subs.insert("funding");
3598        ticker_subs.insert(instrument_id, subs);
3599
3600        let mut funding_cache = AHashMap::new();
3601        let clock = get_atomic_clock_realtime();
3602
3603        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3604
3605        let json = load_test_json("ws_ticker_linear.json");
3606        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
3607        let ws_msg = BybitWsMessage::TickerLinear(msg.clone());
3608
3609        handle_ws_message(
3610            &ws_msg,
3611            &tx.clone().into(),
3612            &instruments,
3613            Some(BybitProductType::Linear),
3614            &trade_subs,
3615            &ticker_subs,
3616            &quote_subs,
3617            &book_depths,
3618            &book_sync,
3619            &ws_client,
3620            &greeks_subs,
3621            &Arc::new(AtomicMap::new()),
3622            &bar_types,
3623            true,
3624            &DashMap::new(),
3625            &mut funding_cache,
3626            Duration::ZERO,
3627            &tasks,
3628            clock,
3629        );
3630
3631        let event = rx.try_recv().unwrap();
3632        assert!(matches!(event, DataEvent::Data(Data::FundingRate(_))));
3633
3634        // Send same message again, funding unchanged so should be deduped
3635        let ws_msg2 = BybitWsMessage::TickerLinear(msg);
3636        handle_ws_message(
3637            &ws_msg2,
3638            &tx.into(),
3639            &instruments,
3640            Some(BybitProductType::Linear),
3641            &trade_subs,
3642            &ticker_subs,
3643            &quote_subs,
3644            &book_depths,
3645            &book_sync,
3646            &ws_client,
3647            &greeks_subs,
3648            &Arc::new(AtomicMap::new()),
3649            &bar_types,
3650            true,
3651            &DashMap::new(),
3652            &mut funding_cache,
3653            Duration::ZERO,
3654            &tasks,
3655            clock,
3656        );
3657
3658        rx.try_recv().unwrap_err();
3659    }
3660
3661    #[rstest]
3662    #[case::crypto_perpetual(linear_instrument())]
3663    #[case::perpetual_contract(tradfi_linear_instrument())]
3664    #[tokio::test]
3665    async fn test_subscribe_funding_rates_accepts_perpetuals(#[case] instrument: InstrumentAny) {
3666        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
3667        set_data_event_sender(tx);
3668        let mut client = BybitDataClient::new(
3669            *BYBIT_CLIENT_ID,
3670            BybitDataClientConfig {
3671                product_types: vec![BybitProductType::Linear],
3672                ..Default::default()
3673            },
3674        )
3675        .unwrap();
3676        let instrument_id = instrument.id();
3677        client.instruments.insert(instrument_id, instrument);
3678
3679        let command = SubscribeFundingRates::new(
3680            instrument_id,
3681            Some(*BYBIT_CLIENT_ID),
3682            None,
3683            UUID4::new(),
3684            UnixNanos::default(),
3685            None,
3686            None,
3687        );
3688        client.subscribe_funding_rates(command).unwrap();
3689
3690        let subs = client.ticker_subs.get_cloned(&instrument_id).unwrap();
3691        assert!(subs.contains("funding"));
3692    }
3693
3694    #[tokio::test]
3695    async fn test_funding_rate_list_continues_after_invalid_instrument() {
3696        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
3697        set_data_event_sender(tx);
3698
3699        let mut client = BybitDataClient::new(
3700            *BYBIT_CLIENT_ID,
3701            BybitDataClientConfig {
3702                product_types: vec![BybitProductType::Linear, BybitProductType::Option],
3703                ..Default::default()
3704            },
3705        )
3706        .unwrap();
3707
3708        let invalid = option_instrument();
3709        let first = linear_instruments().pop().unwrap();
3710        let last = tradfi_linear_instrument();
3711        let ids = [invalid.id(), first.id(), last.id()];
3712        for instrument in [invalid, first, last] {
3713            client.instruments.insert(instrument.id(), instrument);
3714        }
3715
3716        let command = SubscribeFundingRates::new(
3717            ids,
3718            Some(*BYBIT_CLIENT_ID),
3719            None,
3720            UUID4::new(),
3721            UnixNanos::from(17),
3722            Some(UUID4::new()),
3723            None,
3724        );
3725        let error = client.subscribe_funding_rates(command).unwrap_err();
3726        let partial = error.downcast_ref::<SubscriptionError>().unwrap();
3727        assert_eq!(partial.instrument_ids_accepted, vec![ids[1], ids[2]]);
3728        assert_eq!(partial.bar_types_accepted, Vec::<BarType>::new());
3729        assert!(!client.ticker_subs.contains_key(&ids[0]));
3730
3731        for id in &ids[1..] {
3732            assert_eq!(
3733                client.ticker_subs.get_cloned(id).unwrap(),
3734                AHashSet::from_iter(["funding"])
3735            );
3736        }
3737
3738        client
3739            .unsubscribe_funding_rates(&UnsubscribeFundingRates::new(
3740                [ids[1], ids[2]],
3741                Some(*BYBIT_CLIENT_ID),
3742                None,
3743                UUID4::new(),
3744                UnixNanos::from(23),
3745                None,
3746                None,
3747            ))
3748            .unwrap();
3749        assert!(client.ticker_subs.is_empty());
3750        client.disconnect().await.unwrap();
3751    }
3752
3753    #[rstest]
3754    fn test_handle_ticker_linear_mark_and_index_prices() {
3755        let instrument = linear_instrument();
3756        let instrument_id = instrument.id();
3757        let instruments = build_instruments(&[instrument]);
3758        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3759            empty_subs();
3760        let (book_sync, ws_client, tasks) = book_context();
3761
3762        let mut subs = AHashSet::new();
3763        subs.insert("mark_prices");
3764        subs.insert("index_prices");
3765        ticker_subs.insert(instrument_id, subs);
3766
3767        let mut funding_cache = AHashMap::new();
3768        let clock = get_atomic_clock_realtime();
3769
3770        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3771
3772        let json = load_test_json("ws_ticker_linear.json");
3773        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
3774        let ws_msg = BybitWsMessage::TickerLinear(msg);
3775
3776        handle_ws_message(
3777            &ws_msg,
3778            &tx.into(),
3779            &instruments,
3780            Some(BybitProductType::Linear),
3781            &trade_subs,
3782            &ticker_subs,
3783            &quote_subs,
3784            &book_depths,
3785            &book_sync,
3786            &ws_client,
3787            &greeks_subs,
3788            &Arc::new(AtomicMap::new()),
3789            &bar_types,
3790            true,
3791            &DashMap::new(),
3792            &mut funding_cache,
3793            Duration::ZERO,
3794            &tasks,
3795            clock,
3796        );
3797
3798        let event1 = rx.try_recv().unwrap();
3799        assert!(matches!(event1, DataEvent::Data(Data::MarkPrice(_))));
3800
3801        let event2 = rx.try_recv().unwrap();
3802        assert!(matches!(event2, DataEvent::Data(Data::IndexPrice(_))));
3803    }
3804
3805    #[rstest]
3806    #[case::option(BybitProductType::Option, true)]
3807    #[case::linear(BybitProductType::Linear, false)]
3808    fn test_handle_reconnected_clears_caches(
3809        #[case] product_type: BybitProductType,
3810        #[case] clear_option_quotes: bool,
3811    ) {
3812        let instruments = AHashMap::new();
3813        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3814            empty_subs();
3815        let (book_sync, ws_client, tasks) = book_context();
3816
3817        let mut funding_cache = AHashMap::new();
3818        let clock = get_atomic_clock_realtime();
3819
3820        let instrument_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
3821
3822        let option_quote = QuoteTick::new(
3823            instrument_id,
3824            Price::from("100.00"),
3825            Price::from("101.00"),
3826            Quantity::from("1.0"),
3827            Quantity::from("1.0"),
3828            UnixNanos::default(),
3829            UnixNanos::default(),
3830        );
3831        funding_cache.insert(
3832            Ustr::from("BTCUSDT"),
3833            (
3834                Some("-0.001".to_string()),
3835                Some("1000".to_string()),
3836                Some("8".to_string()),
3837            ),
3838        );
3839
3840        let last_quotes = DashMap::new();
3841        last_quotes.insert(instrument_id, option_quote);
3842
3843        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
3844
3845        handle_ws_message(
3846            &BybitWsMessage::Reconnected,
3847            &tx.into(),
3848            &instruments,
3849            Some(product_type),
3850            &trade_subs,
3851            &ticker_subs,
3852            &quote_subs,
3853            &book_depths,
3854            &book_sync,
3855            &ws_client,
3856            &greeks_subs,
3857            &Arc::new(AtomicMap::new()),
3858            &bar_types,
3859            true,
3860            &last_quotes,
3861            &mut funding_cache,
3862            Duration::ZERO,
3863            &tasks,
3864            clock,
3865        );
3866
3867        assert!(funding_cache.is_empty());
3868        assert_eq!(
3869            last_quotes.get(&instrument_id).map(|quote| *quote),
3870            if clear_option_quotes {
3871                None
3872            } else {
3873                Some(option_quote)
3874            },
3875        );
3876    }
3877
3878    #[rstest]
3879    fn test_handle_ticker_option_greeks() {
3880        // Use the option instrument but key it by the ticker fixture symbol
3881        // (fixture instrument is ETH-26JUN26-16000-P, ticker fixture is BTC-6JAN23-17500-C)
3882        let instrument = option_instrument();
3883        let instrument_id = instrument.id();
3884
3885        // Key the instrument by the fixture ticker symbol with OPTION suffix
3886        let ticker_key = Ustr::from("BTC-6JAN23-17500-C-OPTION");
3887        let mut instruments = AHashMap::new();
3888        instruments.insert(ticker_key, instrument);
3889
3890        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3891            empty_subs();
3892
3893        let (book_sync, ws_client, tasks) = book_context();
3894        greeks_subs.insert(instrument_id);
3895
3896        let mut funding_cache = AHashMap::new();
3897        let clock = get_atomic_clock_realtime();
3898
3899        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3900
3901        let json = load_test_json("ws_ticker_option.json");
3902        let msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
3903        let ws_msg = BybitWsMessage::TickerOption(msg);
3904
3905        handle_ws_message(
3906            &ws_msg,
3907            &tx.into(),
3908            &instruments,
3909            Some(BybitProductType::Option),
3910            &trade_subs,
3911            &ticker_subs,
3912            &quote_subs,
3913            &book_depths,
3914            &book_sync,
3915            &ws_client,
3916            &greeks_subs,
3917            &Arc::new(AtomicMap::new()),
3918            &bar_types,
3919            true,
3920            &DashMap::new(),
3921            &mut funding_cache,
3922            Duration::ZERO,
3923            &tasks,
3924            clock,
3925        );
3926
3927        let event = rx.try_recv().unwrap();
3928        assert!(matches!(event, DataEvent::Data(Data::OptionGreeks(_))));
3929    }
3930
3931    #[rstest]
3932    #[case::bid_price("9.5", "10", "2", "5")]
3933    #[case::ask_price("9", "11", "2", "5")]
3934    #[case::bid_size("9", "10", "3", "5")]
3935    #[case::ask_size("9", "10", "2", "6")]
3936    fn test_handle_ticker_option_quote_dedup(
3937        #[case] bid_price: &str,
3938        #[case] ask_price: &str,
3939        #[case] bid_size: &str,
3940        #[case] ask_size: &str,
3941    ) {
3942        // Key the option instrument by the fixture ticker symbol with OPTION suffix
3943        let instrument = option_instrument();
3944        let instrument_id = instrument.id();
3945        let mut instruments = AHashMap::new();
3946        instruments.insert(Ustr::from("BTC-6JAN23-17500-C-OPTION"), instrument.clone());
3947
3948        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
3949            empty_subs();
3950        let (book_sync, ws_client, tasks) = book_context();
3951
3952        let mut subs = AHashSet::new();
3953        subs.insert("quotes");
3954        subs.insert("mark_prices");
3955        ticker_subs.insert(instrument_id, subs);
3956
3957        let last_quotes = DashMap::new();
3958        let mut funding_cache = AHashMap::new();
3959        let ts1 = UnixNanos::new(1_672_917_511_100_000_000);
3960        let ts2 = UnixNanos::new(1_672_917_511_200_000_000);
3961        let ts3 = UnixNanos::new(1_672_917_511_300_000_000);
3962        let ts4 = UnixNanos::new(1_672_917_511_400_000_000);
3963        let clock = AtomicTime::new(false, ts1);
3964        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
3965
3966        let json = load_test_json("ws_ticker_option.json");
3967        let mut msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
3968
3969        msg.data.bid_price = "9".to_string();
3970        msg.data.bid_size = "2".to_string();
3971        msg.data.ask_size = "5".to_string();
3972        msg.data.mark_price = "7.8".to_string();
3973
3974        // The second ticker repeats the top of book; the third changes one field
3975        let mut unchanged = msg.clone();
3976        unchanged.ts += 100;
3977        let mut changed = msg.clone();
3978        changed.ts += 200;
3979        changed.data.bid_price = bid_price.to_string();
3980        changed.data.ask_price = ask_price.to_string();
3981        changed.data.bid_size = bid_size.to_string();
3982        changed.data.ask_size = ask_size.to_string();
3983        changed.data.mark_price = "8.2".to_string();
3984        let mut repeated = changed.clone();
3985        repeated.ts += 100;
3986
3987        for (ts_init, msg) in [
3988            (ts1, msg),
3989            (ts2, unchanged),
3990            (ts3, changed),
3991            (ts4, repeated),
3992        ] {
3993            clock.set_time(ts_init);
3994            handle_ws_message(
3995                &BybitWsMessage::TickerOption(msg),
3996                &tx.clone().into(),
3997                &instruments,
3998                Some(BybitProductType::Option),
3999                &trade_subs,
4000                &ticker_subs,
4001                &quote_subs,
4002                &book_depths,
4003                &book_sync,
4004                &ws_client,
4005                &greeks_subs,
4006                &Arc::new(AtomicMap::new()),
4007                &bar_types,
4008                true,
4009                &last_quotes,
4010                &mut funding_cache,
4011                Duration::ZERO,
4012                &tasks,
4013                &clock,
4014            );
4015        }
4016
4017        let mut quotes = Vec::new();
4018        let mut mark_prices = Vec::new();
4019
4020        while let Ok(event) = rx.try_recv() {
4021            match event {
4022                DataEvent::Data(Data::Quote(quote)) => quotes.push(quote),
4023                DataEvent::Data(Data::MarkPrice(mark)) => mark_prices.push(mark),
4024                other => panic!("Unexpected event: {other:?}"),
4025            }
4026        }
4027
4028        let price = |value: &str| {
4029            Price::from_decimal_dp(value.parse().unwrap(), instrument.price_precision()).unwrap()
4030        };
4031
4032        let size = |value: &str| {
4033            Quantity::from_decimal_dp(value.parse().unwrap(), instrument.size_precision()).unwrap()
4034        };
4035
4036        let initial = QuoteTick::new(
4037            instrument_id,
4038            price("9"),
4039            price("10"),
4040            size("2"),
4041            size("5"),
4042            UnixNanos::new(1_672_917_511_074_000_000),
4043            ts1,
4044        );
4045
4046        let changed = QuoteTick::new(
4047            instrument_id,
4048            price(bid_price),
4049            price(ask_price),
4050            size(bid_size),
4051            size(ask_size),
4052            UnixNanos::new(1_672_917_511_274_000_000),
4053            ts3,
4054        );
4055
4056        let expected_marks = [
4057            ("7.8", 1_672_917_511_074_000_000, ts1),
4058            ("7.8", 1_672_917_511_174_000_000, ts2),
4059            ("8.2", 1_672_917_511_274_000_000, ts3),
4060            ("8.2", 1_672_917_511_374_000_000, ts4),
4061        ]
4062        .map(|(value, ts_event, ts_init)| {
4063            MarkPriceUpdate::new(
4064                instrument_id,
4065                price(value),
4066                UnixNanos::new(ts_event),
4067                ts_init,
4068            )
4069        });
4070
4071        assert_eq!(quotes, vec![initial, changed]);
4072        assert_eq!(mark_prices, expected_marks);
4073    }
4074
4075    #[rstest]
4076    #[case::resubscribe(false, false)]
4077    #[case::resubscribe_after_late_quote(false, true)]
4078    #[case::reconnect(true, false)]
4079    #[tokio::test]
4080    async fn test_option_quote_subscription_emits_unchanged_initial_quote(
4081        #[case] reconnect: bool,
4082        #[case] late_quote: bool,
4083    ) {
4084        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
4085        set_data_event_sender(tx);
4086
4087        let mut client = BybitDataClient::new(
4088            *BYBIT_CLIENT_ID,
4089            BybitDataClientConfig {
4090                product_types: vec![BybitProductType::Option],
4091                ..Default::default()
4092            },
4093        )
4094        .unwrap();
4095
4096        let instrument = option_instrument();
4097        let instrument_id = instrument.id();
4098        client.instruments.insert(instrument_id, instrument.clone());
4099        let instruments = build_instruments(std::slice::from_ref(&instrument));
4100        let ts1 = UnixNanos::new(1_700_000_000_000_000_000);
4101        let ts2 = UnixNanos::new(1_700_000_000_100_000_000);
4102        let ts3 = UnixNanos::new(1_700_000_000_200_000_000);
4103        let clock = AtomicTime::new(false, ts1);
4104        let mut msg: BybitWsTickerOptionMsg =
4105            serde_json::from_str(&load_test_json("ws_ticker_option.json")).unwrap();
4106        msg.data.symbol = instrument.raw_symbol().inner();
4107        msg.data.bid_price = "9".to_string();
4108        msg.data.ask_price = "10".to_string();
4109        msg.data.bid_size = "2".to_string();
4110        msg.data.ask_size = "5".to_string();
4111
4112        let initial = QuoteTick::new(
4113            instrument_id,
4114            Price::from("9.0"),
4115            Price::from("10.0"),
4116            Quantity::from("2"),
4117            Quantity::from("5"),
4118            UnixNanos::new(1_672_917_511_074_000_000),
4119            ts1,
4120        );
4121
4122        let subscribe = || {
4123            SubscribeQuotes::new(
4124                instrument_id,
4125                Some(*BYBIT_CLIENT_ID),
4126                None,
4127                UUID4::new(),
4128                UnixNanos::default(),
4129                None,
4130                None,
4131            )
4132        };
4133
4134        let handle = |client: &BybitDataClient, message: &BybitWsMessage| {
4135            let ws = client
4136                .get_ws_client_for_product(BybitProductType::Option)
4137                .unwrap();
4138            handle_ws_message(
4139                message,
4140                &client.data_sender,
4141                &instruments,
4142                Some(BybitProductType::Option),
4143                &client.trade_subs,
4144                &client.ticker_subs,
4145                &client.quote_subs,
4146                &client.book_depths,
4147                &client.book_sync,
4148                ws,
4149                &client.option_greeks_subs,
4150                &client.liquidation_subs,
4151                ws.bar_types_cache(),
4152                true,
4153                &client.last_quotes,
4154                &mut AHashMap::new(),
4155                Duration::ZERO,
4156                &client.command_tasks.spawner().unwrap(),
4157                &clock,
4158            );
4159        };
4160
4161        client.subscribe_quotes(subscribe()).unwrap();
4162        handle(&client, &BybitWsMessage::TickerOption(msg.clone()));
4163        if reconnect {
4164            handle(&client, &BybitWsMessage::Reconnected);
4165        } else {
4166            client
4167                .unsubscribe_quotes(&UnsubscribeQuotes::new(
4168                    instrument_id,
4169                    Some(*BYBIT_CLIENT_ID),
4170                    None,
4171                    UUID4::new(),
4172                    UnixNanos::default(),
4173                    None,
4174                    None,
4175                ))
4176                .unwrap();
4177            clock.set_time(ts2);
4178            msg.ts += 100;
4179            handle(&client, &BybitWsMessage::TickerOption(msg.clone()));
4180            if late_quote {
4181                client.last_quotes.insert(instrument_id, initial);
4182            }
4183
4184            client.subscribe_quotes(subscribe()).unwrap();
4185        }
4186
4187        clock.set_time(ts3);
4188        msg.ts = 1_672_917_511_274;
4189        handle(&client, &BybitWsMessage::TickerOption(msg));
4190        let mut quotes = Vec::new();
4191
4192        while let Ok(event) = rx.try_recv() {
4193            match event {
4194                DataEvent::Data(Data::Quote(quote)) => quotes.push(quote),
4195                other => panic!("Unexpected event: {other:?}"),
4196            }
4197        }
4198
4199        assert_eq!(
4200            quotes,
4201            vec![
4202                initial,
4203                QuoteTick {
4204                    ts_event: UnixNanos::new(1_672_917_511_274_000_000),
4205                    ts_init: ts3,
4206                    ..initial
4207                }
4208            ]
4209        );
4210    }
4211
4212    #[rstest]
4213    fn test_handle_execution_message_ignored_by_data() {
4214        let instruments = AHashMap::new();
4215        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
4216            empty_subs();
4217        let (book_sync, ws_client, tasks) = book_context();
4218
4219        let mut funding_cache = AHashMap::new();
4220        let clock = get_atomic_clock_realtime();
4221
4222        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
4223
4224        let json = load_test_json("ws_account_order.json");
4225        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
4226            serde_json::from_str(&json).unwrap();
4227        let ws_msg = BybitWsMessage::AccountOrder(msg);
4228
4229        handle_ws_message(
4230            &ws_msg,
4231            &tx.into(),
4232            &instruments,
4233            None,
4234            &trade_subs,
4235            &ticker_subs,
4236            &quote_subs,
4237            &book_depths,
4238            &book_sync,
4239            &ws_client,
4240            &greeks_subs,
4241            &Arc::new(AtomicMap::new()),
4242            &bar_types,
4243            true,
4244            &DashMap::new(),
4245            &mut funding_cache,
4246            Duration::ZERO,
4247            &tasks,
4248            clock,
4249        );
4250
4251        rx.try_recv().unwrap_err();
4252    }
4253
4254    #[rstest]
4255    fn test_instrument_resolution_with_product_type() {
4256        let instrument = linear_instrument();
4257
4258        let mut map = AHashMap::new();
4259        map.insert(instrument.id().symbol.inner(), instrument.clone());
4260
4261        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
4262            empty_subs();
4263
4264        let (book_sync, ws_client, tasks) = book_context();
4265        trade_subs.insert(instrument.id());
4266
4267        let mut funding_cache = AHashMap::new();
4268        let clock = get_atomic_clock_realtime();
4269        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
4270
4271        let json = load_test_json("ws_public_trade.json");
4272        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
4273
4274        // With None product_type, raw symbol "BTCUSDT" does not match "BTCUSDT-LINEAR"
4275        handle_ws_message(
4276            &BybitWsMessage::Trade(msg.clone()),
4277            &tx.clone().into(),
4278            &map,
4279            None,
4280            &trade_subs,
4281            &ticker_subs,
4282            &quote_subs,
4283            &book_depths,
4284            &book_sync,
4285            &ws_client,
4286            &greeks_subs,
4287            &Arc::new(AtomicMap::new()),
4288            &bar_types,
4289            true,
4290            &DashMap::new(),
4291            &mut funding_cache,
4292            Duration::ZERO,
4293            &tasks,
4294            clock,
4295        );
4296        rx.try_recv().unwrap_err();
4297
4298        // With product_type=Linear, "BTCUSDT" -> "BTCUSDT-LINEAR" matches
4299        handle_ws_message(
4300            &BybitWsMessage::Trade(msg),
4301            &tx.into(),
4302            &map,
4303            Some(BybitProductType::Linear),
4304            &trade_subs,
4305            &ticker_subs,
4306            &quote_subs,
4307            &book_depths,
4308            &book_sync,
4309            &ws_client,
4310            &greeks_subs,
4311            &Arc::new(AtomicMap::new()),
4312            &bar_types,
4313            true,
4314            &DashMap::new(),
4315            &mut funding_cache,
4316            Duration::ZERO,
4317            &tasks,
4318            clock,
4319        );
4320
4321        let event = rx.try_recv().unwrap();
4322        assert!(matches!(event, DataEvent::Data(Data::Trade(_))));
4323    }
4324
4325    #[rstest]
4326    fn test_handle_trade_filters_by_subscription() {
4327        let instrument = linear_instrument();
4328        let instruments = build_instruments(std::slice::from_ref(&instrument));
4329        let (trade_subs, ticker_subs, quote_subs, book_depths, greeks_subs, bar_types) =
4330            empty_subs();
4331        let (book_sync, ws_client, tasks) = book_context();
4332
4333        let mut funding_cache = AHashMap::new();
4334        let clock = get_atomic_clock_realtime();
4335        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
4336
4337        let json = load_test_json("ws_public_trade.json");
4338        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
4339
4340        // Without subscription, trade should be filtered out
4341        handle_ws_message(
4342            &BybitWsMessage::Trade(msg.clone()),
4343            &tx.clone().into(),
4344            &instruments,
4345            Some(BybitProductType::Linear),
4346            &trade_subs,
4347            &ticker_subs,
4348            &quote_subs,
4349            &book_depths,
4350            &book_sync,
4351            &ws_client,
4352            &greeks_subs,
4353            &Arc::new(AtomicMap::new()),
4354            &bar_types,
4355            true,
4356            &DashMap::new(),
4357            &mut funding_cache,
4358            Duration::ZERO,
4359            &tasks,
4360            clock,
4361        );
4362        rx.try_recv().unwrap_err();
4363
4364        // With subscription, trade should be emitted
4365        trade_subs.insert(instrument.id());
4366        handle_ws_message(
4367            &BybitWsMessage::Trade(msg),
4368            &tx.into(),
4369            &instruments,
4370            Some(BybitProductType::Linear),
4371            &trade_subs,
4372            &ticker_subs,
4373            &quote_subs,
4374            &book_depths,
4375            &book_sync,
4376            &ws_client,
4377            &greeks_subs,
4378            &Arc::new(AtomicMap::new()),
4379            &bar_types,
4380            true,
4381            &DashMap::new(),
4382            &mut funding_cache,
4383            Duration::ZERO,
4384            &tasks,
4385            clock,
4386        );
4387        let event = rx.try_recv().unwrap();
4388        assert!(matches!(event, DataEvent::Data(Data::Trade(_))));
4389    }
4390}