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

nautilus_bybit/
data.rs
