nautilus_data/
client.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//! Base data client functionality.
17//!
18//! Provides the `DataClientAdapter` for managing subscriptions and requests,
19//! and utilities for constructing data responses.
20
21use std::{
22    fmt::{Debug, Display},
23    hash::Hash,
24    ops::{Deref, DerefMut},
25};
26
27use ahash::AHashSet;
28use nautilus_common::{
29    clients::{DataClient, log_command_error},
30    enums::LogColor,
31    log_info,
32    messages::data::{
33        RequestBars, RequestBookDepth, RequestBookSnapshot, RequestCustomData,
34        RequestForwardPrices, RequestFundingRates, RequestInstrument, RequestInstruments,
35        RequestQuotes, RequestTrades, SubscribeBars, SubscribeBookDeltas, SubscribeBookDepth10,
36        SubscribeCommand, SubscribeCustomData, SubscribeFundingRates, SubscribeIndexPrices,
37        SubscribeInstrument, SubscribeInstrumentClose, SubscribeInstrumentStatus,
38        SubscribeInstruments, SubscribeMarkPrices, SubscribeOptionGreeks, SubscribeQuotes,
39        SubscribeTrades, UnsubscribeBars, UnsubscribeBookDeltas, UnsubscribeBookDepth10,
40        UnsubscribeCommand, UnsubscribeCustomData, UnsubscribeFundingRates, UnsubscribeIndexPrices,
41        UnsubscribeInstrument, UnsubscribeInstrumentClose, UnsubscribeInstrumentStatus,
42        UnsubscribeInstruments, UnsubscribeMarkPrices, UnsubscribeOptionGreeks, UnsubscribeQuotes,
43        UnsubscribeTrades,
44    },
45    msgbus::{self, switchboard::get_custom_topic},
46};
47#[cfg(feature = "defi")]
48use nautilus_model::defi::Blockchain;
49use nautilus_model::{
50    data::{BarType, DataType},
51    identifiers::{ClientId, InstrumentId, Venue},
52};
53
54#[cfg(feature = "defi")]
55#[allow(unused_imports)] // Brings DeFi impl blocks into scope
56use crate::defi::client as _;
57
58/// Wraps a [`DataClient`], managing subscription state and forwarding commands.
59pub struct DataClientAdapter {
60    pub(crate) client: Box<dyn DataClient>,
61    pub client_id: ClientId,
62    pub venue: Option<Venue>,
63    pub handles_book_deltas: bool,
64    pub handles_book_snapshots: bool,
65    pub subscriptions_custom: AHashSet<DataType>,
66    pub subscriptions_book_deltas: AHashSet<InstrumentId>,
67    pub subscriptions_book_depth10: AHashSet<InstrumentId>,
68    pub subscriptions_quotes: AHashSet<InstrumentId>,
69    pub subscriptions_trades: AHashSet<InstrumentId>,
70    pub subscriptions_bars: AHashSet<BarType>,
71    pub subscriptions_instrument_status: AHashSet<InstrumentId>,
72    pub subscriptions_instrument_close: AHashSet<InstrumentId>,
73    pub subscriptions_instrument: AHashSet<InstrumentId>,
74    pub subscriptions_instrument_venue: AHashSet<Venue>,
75    pub subscriptions_mark_prices: AHashSet<InstrumentId>,
76    pub subscriptions_index_prices: AHashSet<InstrumentId>,
77    pub subscriptions_funding_rates: AHashSet<InstrumentId>,
78    pub subscriptions_option_greeks: AHashSet<InstrumentId>,
79    #[cfg(feature = "defi")]
80    pub subscriptions_blocks: AHashSet<Blockchain>,
81    #[cfg(feature = "defi")]
82    pub subscriptions_pools: AHashSet<InstrumentId>,
83    #[cfg(feature = "defi")]
84    pub subscriptions_pool_swaps: AHashSet<InstrumentId>,
85    #[cfg(feature = "defi")]
86    pub subscriptions_pool_liquidity_updates: AHashSet<InstrumentId>,
87    #[cfg(feature = "defi")]
88    pub subscriptions_pool_fee_collects: AHashSet<InstrumentId>,
89    #[cfg(feature = "defi")]
90    pub subscriptions_pool_flash: AHashSet<InstrumentId>,
91}
92
93impl Deref for DataClientAdapter {
94    type Target = Box<dyn DataClient>;
95
96    fn deref(&self) -> &Self::Target {
97        &self.client
98    }
99}
100
101impl DerefMut for DataClientAdapter {
102    fn deref_mut(&mut self) -> &mut Self::Target {
103        &mut self.client
104    }
105}
106
107impl Debug for DataClientAdapter {
108    #[rustfmt::skip]
109    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
110        f.debug_struct(stringify!(DataClientAdapter))
111            .field("client_id", &self.client_id)
112            .field("venue", &self.venue)
113            .field("handles_book_deltas", &self.handles_book_deltas)
114            .field("handles_book_snapshots", &self.handles_book_snapshots)
115            .field("subscriptions_custom", &self.subscriptions_custom)
116            .field("subscriptions_book_deltas", &self.subscriptions_book_deltas)
117            .field("subscriptions_book_depth10", &self.subscriptions_book_depth10)
118            .field("subscriptions_quotes", &self.subscriptions_quotes)
119            .field("subscriptions_trades", &self.subscriptions_trades)
120            .field("subscriptions_bars", &self.subscriptions_bars)
121            .field("subscriptions_mark_prices", &self.subscriptions_mark_prices)
122            .field("subscriptions_index_prices", &self.subscriptions_index_prices)
123            .field("subscriptions_instrument_status", &self.subscriptions_instrument_status)
124            .field("subscriptions_instrument_close", &self.subscriptions_instrument_close)
125            .field("subscriptions_instrument", &self.subscriptions_instrument)
126            .field("subscriptions_instrument_venue", &self.subscriptions_instrument_venue)
127            .finish()
128    }
129}
130
131impl DataClientAdapter {
132    /// Creates a new [`DataClientAdapter`] with the given client and clock.
133    #[must_use]
134    pub fn new(
135        client_id: ClientId,
136        venue: Option<Venue>,
137        handles_order_book_deltas: bool,
138        handles_order_book_snapshots: bool,
139        client: Box<dyn DataClient>,
140    ) -> Self {
141        Self {
142            client,
143            client_id,
144            venue,
145            handles_book_deltas: handles_order_book_deltas,
146            handles_book_snapshots: handles_order_book_snapshots,
147            subscriptions_custom: AHashSet::new(),
148            subscriptions_book_deltas: AHashSet::new(),
149            subscriptions_book_depth10: AHashSet::new(),
150            subscriptions_quotes: AHashSet::new(),
151            subscriptions_trades: AHashSet::new(),
152            subscriptions_mark_prices: AHashSet::new(),
153            subscriptions_index_prices: AHashSet::new(),
154            subscriptions_funding_rates: AHashSet::new(),
155            subscriptions_option_greeks: AHashSet::new(),
156            subscriptions_bars: AHashSet::new(),
157            subscriptions_instrument_status: AHashSet::new(),
158            subscriptions_instrument_close: AHashSet::new(),
159            subscriptions_instrument: AHashSet::new(),
160            subscriptions_instrument_venue: AHashSet::new(),
161            #[cfg(feature = "defi")]
162            subscriptions_blocks: AHashSet::new(),
163            #[cfg(feature = "defi")]
164            subscriptions_pools: AHashSet::new(),
165            #[cfg(feature = "defi")]
166            subscriptions_pool_swaps: AHashSet::new(),
167            #[cfg(feature = "defi")]
168            subscriptions_pool_liquidity_updates: AHashSet::new(),
169            #[cfg(feature = "defi")]
170            subscriptions_pool_fee_collects: AHashSet::new(),
171            #[cfg(feature = "defi")]
172            subscriptions_pool_flash: AHashSet::new(),
173        }
174    }
175
176    #[expect(clippy::borrowed_box)]
177    #[must_use]
178    pub fn get_client(&self) -> &Box<dyn DataClient> {
179        &self.client
180    }
181
182    /// Connects the underlying client to the data provider.
183    ///
184    /// # Errors
185    ///
186    /// Returns an error if the connection fails.
187    pub async fn connect(&mut self) -> anyhow::Result<()> {
188        self.client.connect().await
189    }
190
191    /// Disconnects the underlying client from the data provider.
192    ///
193    /// # Errors
194    ///
195    /// Returns an error if the disconnection fails.
196    pub async fn disconnect(&mut self) -> anyhow::Result<()> {
197        self.client.disconnect().await
198    }
199
200    #[inline]
201    pub fn execute_subscribe(&mut self, cmd: SubscribeCommand) {
202        let cmd_debug = format!("{cmd:?}");
203        if let Err(e) = match cmd {
204            SubscribeCommand::Data(cmd) => self.subscribe(cmd),
205            SubscribeCommand::Instrument(cmd) => self.subscribe_instrument(cmd),
206            SubscribeCommand::Instruments(cmd) => self.subscribe_instruments(cmd),
207            SubscribeCommand::BookDeltas(cmd) => self.subscribe_book_deltas(cmd),
208            SubscribeCommand::BookDepth10(cmd) => self.subscribe_book_depth10(cmd),
209            SubscribeCommand::BookSnapshots(_) => Ok(()), // Handled internally by engine
210            SubscribeCommand::Quotes(cmd) => self.subscribe_quotes(cmd),
211            SubscribeCommand::Trades(cmd) => self.subscribe_trades(cmd),
212            SubscribeCommand::MarkPrices(cmd) => self.subscribe_mark_prices(cmd),
213            SubscribeCommand::IndexPrices(cmd) => self.subscribe_index_prices(cmd),
214            SubscribeCommand::FundingRates(cmd) => self.subscribe_funding_rates(cmd),
215            SubscribeCommand::Bars(cmd) => self.subscribe_bars(cmd),
216            SubscribeCommand::InstrumentStatus(cmd) => self.subscribe_instrument_status(cmd),
217            SubscribeCommand::InstrumentClose(cmd) => self.subscribe_instrument_close(cmd),
218            SubscribeCommand::OptionGreeks(cmd) => self.subscribe_option_greeks(cmd),
219            SubscribeCommand::OptionChain(_) => Ok(()), // Handled internally by engine
220        } {
221            log_command_error(&cmd_debug, &e);
222        }
223    }
224
225    #[inline]
226    pub fn execute_unsubscribe(&mut self, cmd: &UnsubscribeCommand) {
227        if let Err(e) = match cmd {
228            UnsubscribeCommand::Data(cmd) => self.unsubscribe(cmd),
229            UnsubscribeCommand::Instrument(cmd) => self.unsubscribe_instrument(cmd),
230            UnsubscribeCommand::Instruments(cmd) => self.unsubscribe_instruments(cmd),
231            UnsubscribeCommand::BookDeltas(cmd) => self.unsubscribe_book_deltas(cmd),
232            UnsubscribeCommand::BookDepth10(cmd) => self.unsubscribe_book_depth10(cmd),
233            UnsubscribeCommand::BookSnapshots(_) => Ok(()), // Handled internally by engine
234            UnsubscribeCommand::Quotes(cmd) => self.unsubscribe_quotes(cmd),
235            UnsubscribeCommand::Trades(cmd) => self.unsubscribe_trades(cmd),
236            UnsubscribeCommand::Bars(cmd) => self.unsubscribe_bars(cmd),
237            UnsubscribeCommand::MarkPrices(cmd) => self.unsubscribe_mark_prices(cmd),
238            UnsubscribeCommand::IndexPrices(cmd) => self.unsubscribe_index_prices(cmd),
239            UnsubscribeCommand::FundingRates(cmd) => self.unsubscribe_funding_rates(cmd),
240            UnsubscribeCommand::InstrumentStatus(cmd) => self.unsubscribe_instrument_status(cmd),
241            UnsubscribeCommand::InstrumentClose(cmd) => self.unsubscribe_instrument_close(cmd),
242            UnsubscribeCommand::OptionGreeks(cmd) => self.unsubscribe_option_greeks(cmd),
243            UnsubscribeCommand::OptionChain(_) => Ok(()), // Handled internally by engine
244        } {
245            log_command_error(&cmd, &e);
246        }
247    }
248
249    /// Subscribes to a custom data type, updating internal state and forwarding to the client.
250    ///
251    /// # Errors
252    ///
253    /// Returns an error if the underlying client subscribe operation fails.
254    pub fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
255        if !self.subscriptions_custom.contains(&cmd.data_type) {
256            self.subscriptions_custom.insert(cmd.data_type.clone());
257            log_info!("Subscribed {}", cmd.data_type, color = LogColor::Blue);
258            self.client.subscribe(cmd)?;
259        }
260        Ok(())
261    }
262
263    /// Unsubscribes from a custom data type, updating internal state and forwarding to the client.
264    ///
265    /// # Errors
266    ///
267    /// Returns an error if the underlying client unsubscribe operation fails.
268    pub fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
269        if self.subscriptions_custom.contains(&cmd.data_type) {
270            if msgbus::subscriptions_count_any(get_custom_topic(&cmd.data_type))? > 0 {
271                return Ok(());
272            }
273
274            self.subscriptions_custom.remove(&cmd.data_type);
275            self.client.unsubscribe(cmd)?;
276            log_info!("Unsubscribed {}", cmd.data_type, color = LogColor::Blue);
277        }
278        Ok(())
279    }
280
281    /// Subscribes to instrument definitions for a venue, updating internal state and forwarding to the client.
282    ///
283    /// # Errors
284    ///
285    /// Returns an error if the underlying client subscribe operation fails.
286    fn subscribe_instruments(&mut self, cmd: SubscribeInstruments) -> anyhow::Result<()> {
287        if Self::track_subscribe(
288            &mut self.subscriptions_instrument_venue,
289            cmd.venue,
290            "instruments",
291        ) {
292            self.client.subscribe_instruments(cmd)?;
293        }
294
295        Ok(())
296    }
297
298    /// Unsubscribes from instrument definition updates for a venue, updating internal state and forwarding to the client.
299    ///
300    /// # Errors
301    ///
302    /// Returns an error if the underlying client unsubscribe operation fails.
303    fn unsubscribe_instruments(&mut self, cmd: &UnsubscribeInstruments) -> anyhow::Result<()> {
304        if Self::track_unsubscribe(
305            &mut self.subscriptions_instrument_venue,
306            cmd.venue,
307            "instruments",
308        ) {
309            self.client.unsubscribe_instruments(cmd)?;
310        }
311
312        Ok(())
313    }
314
315    /// Subscribes to instrument definitions for a single instrument, updating internal state and forwarding to the client.
316    ///
317    /// # Errors
318    ///
319    /// Returns an error if the underlying client subscribe operation fails.
320    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
321        if Self::track_subscribe(
322            &mut self.subscriptions_instrument,
323            cmd.instrument_id,
324            "instrument",
325        ) {
326            self.client.subscribe_instrument(cmd)?;
327        }
328
329        Ok(())
330    }
331
332    /// Unsubscribes from instrument definition updates for a single instrument, updating internal state and forwarding to the client.
333    ///
334    /// # Errors
335    ///
336    /// Returns an error if the underlying client unsubscribe operation fails.
337    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
338        if Self::track_unsubscribe(
339            &mut self.subscriptions_instrument,
340            cmd.instrument_id,
341            "instrument",
342        ) {
343            self.client.unsubscribe_instrument(cmd)?;
344        }
345
346        Ok(())
347    }
348
349    /// Subscribes to book deltas updates for an instrument, updating internal state and forwarding to the client.
350    ///
351    /// # Errors
352    ///
353    /// Returns an error if the underlying client subscribe operation fails.
354    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
355        if Self::track_subscribe(
356            &mut self.subscriptions_book_deltas,
357            cmd.instrument_id,
358            "order book deltas",
359        ) {
360            self.client.subscribe_book_deltas(cmd)?;
361        }
362
363        Ok(())
364    }
365
366    /// Unsubscribes from book deltas for an instrument, updating internal state and forwarding to the client.
367    ///
368    /// # Errors
369    ///
370    /// Returns an error if the underlying client unsubscribe operation fails.
371    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
372        if Self::track_unsubscribe(
373            &mut self.subscriptions_book_deltas,
374            cmd.instrument_id,
375            "order book deltas",
376        ) {
377            self.client.unsubscribe_book_deltas(cmd)?;
378        }
379
380        Ok(())
381    }
382
383    /// Subscribes to book depth updates for an instrument, updating internal state and forwarding to the client.
384    ///
385    /// # Errors
386    ///
387    /// Returns an error if the underlying client subscribe operation fails.
388    fn subscribe_book_depth10(&mut self, cmd: SubscribeBookDepth10) -> anyhow::Result<()> {
389        if Self::track_subscribe(
390            &mut self.subscriptions_book_depth10,
391            cmd.instrument_id,
392            "order book depth",
393        ) {
394            self.client.subscribe_book_depth10(cmd)?;
395        }
396
397        Ok(())
398    }
399
400    /// Unsubscribes from book depth updates for an instrument, updating internal state and forwarding to the client.
401    ///
402    /// # Errors
403    ///
404    /// Returns an error if the underlying client unsubscribe operation fails.
405    fn unsubscribe_book_depth10(&mut self, cmd: &UnsubscribeBookDepth10) -> anyhow::Result<()> {
406        if Self::track_unsubscribe(
407            &mut self.subscriptions_book_depth10,
408            cmd.instrument_id,
409            "order book depth",
410        ) {
411            self.client.unsubscribe_book_depth10(cmd)?;
412        }
413
414        Ok(())
415    }
416
417    /// Subscribes to quotes for an instrument, updating internal state and forwarding to the client.
418    ///
419    /// # Errors
420    ///
421    /// Returns an error if the underlying client subscribe operation fails.
422    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
423        if Self::track_subscribe(&mut self.subscriptions_quotes, cmd.instrument_id, "quotes") {
424            self.client.subscribe_quotes(cmd)?;
425        }
426        Ok(())
427    }
428
429    /// Unsubscribes from quotes for an instrument, updating internal state and forwarding to the client.
430    ///
431    /// # Errors
432    ///
433    /// Returns an error if the underlying client unsubscribe operation fails.
434    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
435        if Self::track_unsubscribe(&mut self.subscriptions_quotes, cmd.instrument_id, "quotes") {
436            self.client.unsubscribe_quotes(cmd)?;
437        }
438        Ok(())
439    }
440
441    /// Subscribes to trades for an instrument, updating internal state and forwarding to the client.
442    ///
443    /// # Errors
444    ///
445    /// Returns an error if the underlying client subscribe operation fails.
446    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
447        if Self::track_subscribe(&mut self.subscriptions_trades, cmd.instrument_id, "trades") {
448            self.client.subscribe_trades(cmd)?;
449        }
450        Ok(())
451    }
452
453    /// Unsubscribes from trades for an instrument, updating internal state and forwarding to the client.
454    ///
455    /// # Errors
456    ///
457    /// Returns an error if the underlying client unsubscribe operation fails.
458    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
459        if Self::track_unsubscribe(&mut self.subscriptions_trades, cmd.instrument_id, "trades") {
460            self.client.unsubscribe_trades(cmd)?;
461        }
462        Ok(())
463    }
464
465    /// Subscribes to bars for a bar type, updating internal state and forwarding to the client.
466    ///
467    /// # Errors
468    ///
469    /// Returns an error if the underlying client subscribe operation fails.
470    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
471        if Self::track_subscribe(&mut self.subscriptions_bars, cmd.bar_type, "bars") {
472            self.client.subscribe_bars(cmd)?;
473        }
474        Ok(())
475    }
476
477    /// Unsubscribes from bars for a bar type, updating internal state and forwarding to the client.
478    ///
479    /// # Errors
480    ///
481    /// Returns an error if the underlying client unsubscribe operation fails.
482    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
483        if Self::track_unsubscribe(&mut self.subscriptions_bars, cmd.bar_type, "bars") {
484            self.client.unsubscribe_bars(cmd)?;
485        }
486        Ok(())
487    }
488
489    /// Subscribes to mark price updates for an instrument, updating internal state and forwarding to the client.
490    ///
491    /// # Errors
492    ///
493    /// Returns an error if the underlying client subscribe operation fails.
494    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
495        if Self::track_subscribe(
496            &mut self.subscriptions_mark_prices,
497            cmd.instrument_id,
498            "mark prices",
499        ) {
500            self.client.subscribe_mark_prices(cmd)?;
501        }
502        Ok(())
503    }
504
505    /// Unsubscribes from mark price updates for an instrument, updating internal state and forwarding to the client.
506    ///
507    /// # Errors
508    ///
509    /// Returns an error if the underlying client unsubscribe operation fails.
510    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
511        if Self::track_unsubscribe(
512            &mut self.subscriptions_mark_prices,
513            cmd.instrument_id,
514            "mark prices",
515        ) {
516            self.client.unsubscribe_mark_prices(cmd)?;
517        }
518        Ok(())
519    }
520
521    /// Subscribes to index price updates for an instrument, updating internal state and forwarding to the client.
522    ///
523    /// # Errors
524    ///
525    /// Returns an error if the underlying client subscribe operation fails.
526    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
527        if Self::track_subscribe(
528            &mut self.subscriptions_index_prices,
529            cmd.instrument_id,
530            "index prices",
531        ) {
532            self.client.subscribe_index_prices(cmd)?;
533        }
534        Ok(())
535    }
536
537    /// Unsubscribes from index price updates for an instrument, updating internal state and forwarding to the client.
538    ///
539    /// # Errors
540    ///
541    /// Returns an error if the underlying client unsubscribe operation fails.
542    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
543        if Self::track_unsubscribe(
544            &mut self.subscriptions_index_prices,
545            cmd.instrument_id,
546            "index prices",
547        ) {
548            self.client.unsubscribe_index_prices(cmd)?;
549        }
550        Ok(())
551    }
552
553    /// Subscribes to funding rate updates for an instrument, updating internal state and forwarding to the client.
554    ///
555    /// # Errors
556    ///
557    /// Returns an error if the underlying client subscribe operation fails.
558    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
559        if Self::track_subscribe(
560            &mut self.subscriptions_funding_rates,
561            cmd.instrument_id,
562            "funding rates",
563        ) {
564            self.client.subscribe_funding_rates(cmd)?;
565        }
566        Ok(())
567    }
568
569    /// Unsubscribes from funding rate updates for an instrument, updating internal state and forwarding to the client.
570    ///
571    /// # Errors
572    ///
573    /// Returns an error if the underlying client unsubscribe operation fails.
574    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
575        if Self::track_unsubscribe(
576            &mut self.subscriptions_funding_rates,
577            cmd.instrument_id,
578            "funding rates",
579        ) {
580            self.client.unsubscribe_funding_rates(cmd)?;
581        }
582        Ok(())
583    }
584
585    /// Subscribes to instrument status updates for the specified instrument.
586    ///
587    /// # Errors
588    ///
589    /// Returns an error if the underlying client subscribe operation fails.
590    fn subscribe_instrument_status(
591        &mut self,
592        cmd: SubscribeInstrumentStatus,
593    ) -> anyhow::Result<()> {
594        if Self::track_subscribe(
595            &mut self.subscriptions_instrument_status,
596            cmd.instrument_id,
597            "instrument status",
598        ) {
599            self.client.subscribe_instrument_status(cmd)?;
600        }
601        Ok(())
602    }
603
604    /// Unsubscribes from instrument status updates for the specified instrument.
605    ///
606    /// # Errors
607    ///
608    /// Returns an error if the underlying client unsubscribe operation fails.
609    fn unsubscribe_instrument_status(
610        &mut self,
611        cmd: &UnsubscribeInstrumentStatus,
612    ) -> anyhow::Result<()> {
613        if Self::track_unsubscribe(
614            &mut self.subscriptions_instrument_status,
615            cmd.instrument_id,
616            "instrument status",
617        ) {
618            self.client.unsubscribe_instrument_status(cmd)?;
619        }
620        Ok(())
621    }
622
623    /// Subscribes to instrument close events for the specified instrument.
624    ///
625    /// # Errors
626    ///
627    /// Returns an error if the underlying client subscribe operation fails.
628    fn subscribe_instrument_close(&mut self, cmd: SubscribeInstrumentClose) -> anyhow::Result<()> {
629        if Self::track_subscribe(
630            &mut self.subscriptions_instrument_close,
631            cmd.instrument_id,
632            "instrument close",
633        ) {
634            self.client.subscribe_instrument_close(cmd)?;
635        }
636        Ok(())
637    }
638
639    /// Unsubscribes from instrument close events for the specified instrument.
640    ///
641    /// # Errors
642    ///
643    /// Returns an error if the underlying client unsubscribe operation fails.
644    fn unsubscribe_instrument_close(
645        &mut self,
646        cmd: &UnsubscribeInstrumentClose,
647    ) -> anyhow::Result<()> {
648        if Self::track_unsubscribe(
649            &mut self.subscriptions_instrument_close,
650            cmd.instrument_id,
651            "instrument close",
652        ) {
653            self.client.unsubscribe_instrument_close(cmd)?;
654        }
655        Ok(())
656    }
657
658    /// Subscribes to option greeks for an instrument, updating internal state and forwarding to the client.
659    ///
660    /// # Errors
661    ///
662    /// Returns an error if the underlying client subscribe operation fails.
663    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
664        if Self::track_subscribe(
665            &mut self.subscriptions_option_greeks,
666            cmd.instrument_id,
667            "option greeks",
668        ) {
669            self.client.subscribe_option_greeks(cmd)?;
670        }
671        Ok(())
672    }
673
674    /// Unsubscribes from option greeks for an instrument, updating internal state and forwarding to the client.
675    ///
676    /// # Errors
677    ///
678    /// Returns an error if the underlying client unsubscribe operation fails.
679    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
680        if Self::track_unsubscribe(
681            &mut self.subscriptions_option_greeks,
682            cmd.instrument_id,
683            "option greeks",
684        ) {
685            self.client.unsubscribe_option_greeks(cmd)?;
686        }
687        Ok(())
688    }
689
690    /// Inserts `key` into the tracking `set` and logs the subscription confirmation.
691    ///
692    /// Returns `true` when the subscription is new (the caller should forward the command to the
693    /// client), or `false` when already subscribed.
694    pub(crate) fn track_subscribe<T>(set: &mut AHashSet<T>, key: T, data_type: &str) -> bool
695    where
696        T: Eq + Hash + Copy + Display,
697    {
698        if set.contains(&key) {
699            return false;
700        }
701        set.insert(key);
702        log_info!("Subscribed {key} {data_type}", color = LogColor::Blue);
703        true
704    }
705
706    /// Removes `key` from the tracking `set` and logs the unsubscription confirmation.
707    ///
708    /// Returns `true` when the subscription existed (the caller should forward the command to the
709    /// client), or `false` when not subscribed.
710    pub(crate) fn track_unsubscribe<T>(set: &mut AHashSet<T>, key: T, data_type: &str) -> bool
711    where
712        T: Eq + Hash + Copy + Display,
713    {
714        if !set.contains(&key) {
715            return false;
716        }
717        set.remove(&key);
718        log_info!("Unsubscribed {key} {data_type}", color = LogColor::Blue);
719        true
720    }
721
722    /// Sends a data request to the underlying client.
723    ///
724    /// # Errors
725    ///
726    /// Returns an error if the client request fails.
727    pub fn request_data(&self, req: RequestCustomData) -> anyhow::Result<()> {
728        self.client.request_data(req)
729    }
730
731    /// Sends a single instrument request to the client.
732    ///
733    /// # Errors
734    ///
735    /// Returns an error if the client fails to process the request.
736    pub fn request_instrument(&self, req: RequestInstrument) -> anyhow::Result<()> {
737        self.client.request_instrument(req)
738    }
739
740    /// Sends a batch instruments request to the client.
741    ///
742    /// # Errors
743    ///
744    /// Returns an error if the client fails to process the request.
745    pub fn request_instruments(&self, req: RequestInstruments) -> anyhow::Result<()> {
746        self.client.request_instruments(req)
747    }
748
749    /// Sends a book snapshot request for a given instrument.
750    ///
751    /// # Errors
752    ///
753    /// Returns an error if the client fails to process the book snapshot request.
754    pub fn request_book_snapshot(&self, req: RequestBookSnapshot) -> anyhow::Result<()> {
755        self.client.request_book_snapshot(req)
756    }
757
758    /// Sends a quotes request for a given instrument.
759    ///
760    /// # Errors
761    ///
762    /// Returns an error if the client fails to process the quotes request.
763    pub fn request_quotes(&self, req: RequestQuotes) -> anyhow::Result<()> {
764        self.client.request_quotes(req)
765    }
766
767    /// Sends a trades request for a given instrument.
768    ///
769    /// # Errors
770    ///
771    /// Returns an error if the client fails to process the trades request.
772    pub fn request_trades(&self, req: RequestTrades) -> anyhow::Result<()> {
773        self.client.request_trades(req)
774    }
775
776    /// Sends a funding rates request for a given instrument.
777    ///
778    /// # Errors
779    ///
780    /// Returns an error if the client fails to process the trades request.
781    pub fn request_funding_rates(&self, req: RequestFundingRates) -> anyhow::Result<()> {
782        self.client.request_funding_rates(req)
783    }
784
785    /// Sends a forward prices request for derivatives instruments.
786    ///
787    /// # Errors
788    ///
789    /// Returns an error if the client fails to process the forward prices request.
790    pub fn request_forward_prices(&self, req: RequestForwardPrices) -> anyhow::Result<()> {
791        self.client.request_forward_prices(req)
792    }
793
794    /// Sends a bars request for a given instrument and bar type.
795    ///
796    /// # Errors
797    ///
798    /// Returns an error if the client fails to process the bars request.
799    pub fn request_bars(&self, req: RequestBars) -> anyhow::Result<()> {
800        self.client.request_bars(req)
801    }
802
803    /// Sends an order book depths request for a given instrument.
804    ///
805    /// # Errors
806    ///
807    /// Returns an error if the client fails to process the order book depths request.
808    pub fn request_book_depth(&self, req: RequestBookDepth) -> anyhow::Result<()> {
809        self.client.request_book_depth(req)
810    }
811}
nautilus_data/client.rs

nautilus_data/
client.rs
