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

nautilus_data/
client.rs
