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

nautilus_bybit/
data.rs
