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, SubscribeBookDepth10, SubscribeCommand,
38        SubscribeCustomData, SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument,
39        SubscribeInstrumentClose, SubscribeInstrumentStatus, SubscribeInstruments,
40        SubscribeMarkPrices, SubscribeOptionGreeks, SubscribeQuotes, SubscribeTrades,
41        UnsubscribeBars, UnsubscribeBookDeltas, UnsubscribeBookDepth10, 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_depth10: 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_depth10", &self.subscriptions_book_depth10)
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_depth10: 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 cmd_debug = format!("{cmd:?}");
234        let result = match cmd {
235            SubscribeCommand::Data(cmd) => self.subscribe(cmd),
236            SubscribeCommand::Instrument(cmd) => self.subscribe_instrument(cmd),
237            SubscribeCommand::Instruments(cmd) => self.subscribe_instruments(cmd),
238            SubscribeCommand::BookDeltas(cmd) => self.subscribe_book_deltas(cmd),
239            SubscribeCommand::BookDepth10(cmd) => self.subscribe_book_depth10(cmd),
240            SubscribeCommand::BookSnapshots(_) => Ok(()), // Handled internally by engine
241            SubscribeCommand::Quotes(cmd) => self.subscribe_quotes(cmd),
242            SubscribeCommand::Trades(cmd) => self.subscribe_trades(cmd),
243            SubscribeCommand::MarkPrices(cmd) => self.subscribe_mark_prices(cmd),
244            SubscribeCommand::IndexPrices(cmd) => self.subscribe_index_prices(cmd),
245            SubscribeCommand::FundingRates(cmd) => self.subscribe_funding_rates(cmd),
246            SubscribeCommand::Bars(cmd) => self.subscribe_bars(cmd),
247            SubscribeCommand::InstrumentStatus(cmd) => self.subscribe_instrument_status(cmd),
248            SubscribeCommand::InstrumentClose(cmd) => self.subscribe_instrument_close(cmd),
249            SubscribeCommand::OptionGreeks(cmd) => self.subscribe_option_greeks(cmd),
250            SubscribeCommand::OptionChain(_) => Ok(()), // Handled internally by engine
251        };
252
253        if let Err(e) = result {
254            // Engine-owned intent survives failure until its matching release
255            if retain_on_failure {
256                self.subscriptions_active
257                    .retain(key, retained.command_id(), retained);
258            }
259
260            log_command_error(&cmd_debug, &e);
261            return;
262        }
263
264        // A retry can establish a different physical identity from the failed attempt
265        if let Some(subscription) = self.subscriptions_active.get_mut(&key) {
266            subscription.command = retained.clone();
267        }
268        self.subscriptions_active
269            .retain(key, retained.command_id(), retained);
270    }
271
272    #[must_use]
273    #[rustfmt::skip]
274    pub(crate) fn has_active_subscription(&self, cmd: &SubscribeCommand) -> bool {
275        match cmd {
276            SubscribeCommand::Data(cmd) => self.subscriptions_custom.contains(&cmd.data_type),
277            SubscribeCommand::Instrument(cmd) => self.subscriptions_instrument.contains(&cmd.instrument_id),
278            SubscribeCommand::Instruments(cmd) => self.subscriptions_instrument_venue.contains(&cmd.venue),
279            SubscribeCommand::BookDeltas(cmd) => self.subscriptions_book_deltas.contains(&cmd.instrument_id),
280            SubscribeCommand::BookDepth10(cmd) => self.subscriptions_book_depth10.contains(&cmd.instrument_id),
281            SubscribeCommand::Quotes(cmd) => self.subscriptions_quotes.contains(&cmd.instrument_id),
282            SubscribeCommand::Trades(cmd) => self.subscriptions_trades.contains(&cmd.instrument_id),
283            SubscribeCommand::Bars(cmd) => self.subscriptions_bars.contains(&cmd.bar_type),
284            SubscribeCommand::MarkPrices(cmd) => self.subscriptions_mark_prices.contains(&cmd.instrument_id),
285            SubscribeCommand::IndexPrices(cmd) => self.subscriptions_index_prices.contains(&cmd.instrument_id),
286            SubscribeCommand::FundingRates(cmd) => self.subscriptions_funding_rates.contains(&cmd.instrument_id),
287            SubscribeCommand::InstrumentStatus(cmd) => self.subscriptions_instrument_status.contains(&cmd.instrument_id),
288            SubscribeCommand::InstrumentClose(cmd) => self.subscriptions_instrument_close.contains(&cmd.instrument_id),
289            SubscribeCommand::OptionGreeks(cmd) => self.subscriptions_option_greeks.contains(&cmd.instrument_id),
290            SubscribeCommand::BookSnapshots(_) | SubscribeCommand::OptionChain(_) => self.subscriptions_active.contains(&SubscriptionKey::from_subscribe(cmd)),
291        }
292    }
293
294    pub(crate) fn clear_subscription_state(&mut self) {
295        self.subscriptions_custom.clear();
296        self.subscriptions_book_deltas.clear();
297        self.subscriptions_book_depth10.clear();
298        self.subscriptions_quotes.clear();
299        self.subscriptions_trades.clear();
300        self.subscriptions_bars.clear();
301        self.subscriptions_instrument_status.clear();
302        self.subscriptions_instrument_close.clear();
303        self.subscriptions_instrument.clear();
304        self.subscriptions_instrument_venue.clear();
305        self.subscriptions_mark_prices.clear();
306        self.subscriptions_index_prices.clear();
307        self.subscriptions_funding_rates.clear();
308        self.subscriptions_option_greeks.clear();
309        self.subscriptions_active.clear();
310
311        #[cfg(feature = "defi")]
312        {
313            self.subscriptions_active_defi.clear();
314            self.subscriptions_blocks.clear();
315            self.subscriptions_pools.clear();
316            self.subscriptions_pool_swaps.clear();
317            self.subscriptions_pool_liquidity_updates.clear();
318            self.subscriptions_pool_fee_collects.clear();
319            self.subscriptions_pool_flash.clear();
320        }
321    }
322
323    #[inline]
324    pub fn execute_unsubscribe(&mut self, cmd: &UnsubscribeCommand) {
325        let key = SubscriptionKey::from_unsubscribe(cmd);
326        let command = match self.subscriptions_active.release(&key) {
327            SubscriptionRelease::Retained => return,
328            SubscriptionRelease::Final(subscribe) => {
329                subscribe.into_unsubscribe(cmd.command_id(), cmd.ts_init(), cmd.correlation_id())
330            }
331            SubscriptionRelease::Untracked => cmd.clone(),
332        };
333
334        if let Err(e) = match &command {
335            UnsubscribeCommand::Data(cmd) => self.unsubscribe(cmd),
336            UnsubscribeCommand::Instrument(cmd) => self.unsubscribe_instrument(cmd),
337            UnsubscribeCommand::Instruments(cmd) => self.unsubscribe_instruments(cmd),
338            UnsubscribeCommand::BookDeltas(cmd) => self.unsubscribe_book_deltas(cmd),
339            UnsubscribeCommand::BookDepth10(cmd) => self.unsubscribe_book_depth10(cmd),
340            UnsubscribeCommand::BookSnapshots(_) => Ok(()), // Handled internally by engine
341            UnsubscribeCommand::Quotes(cmd) => self.unsubscribe_quotes(cmd),
342            UnsubscribeCommand::Trades(cmd) => self.unsubscribe_trades(cmd),
343            UnsubscribeCommand::Bars(cmd) => self.unsubscribe_bars(cmd),
344            UnsubscribeCommand::MarkPrices(cmd) => self.unsubscribe_mark_prices(cmd),
345            UnsubscribeCommand::IndexPrices(cmd) => self.unsubscribe_index_prices(cmd),
346            UnsubscribeCommand::FundingRates(cmd) => self.unsubscribe_funding_rates(cmd),
347            UnsubscribeCommand::InstrumentStatus(cmd) => self.unsubscribe_instrument_status(cmd),
348            UnsubscribeCommand::InstrumentClose(cmd) => self.unsubscribe_instrument_close(cmd),
349            UnsubscribeCommand::OptionGreeks(cmd) => self.unsubscribe_option_greeks(cmd),
350            UnsubscribeCommand::OptionChain(_) => Ok(()), // Handled internally by engine
351        } {
352            log_command_error(&command, &e);
353        } else {
354            self.subscriptions_active.remove(&key);
355        }
356    }
357
358    /// Subscribes to a custom data type, updating internal state and forwarding to the client.
359    ///
360    /// # Errors
361    ///
362    /// Returns an error if the underlying client subscribe operation fails.
363    pub fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
364        let data_type = cmd.data_type.clone();
365        Self::execute_tracked_subscribe(
366            self.client.as_mut(),
367            &mut self.subscriptions_custom,
368            data_type,
369            "",
370            |client| client.subscribe(cmd),
371        )
372    }
373
374    /// Unsubscribes from a custom data type, updating internal state and forwarding to the client.
375    ///
376    /// # Errors
377    ///
378    /// Returns an error if the underlying client unsubscribe operation fails.
379    pub fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
380        Self::execute_tracked_unsubscribe(
381            self.client.as_mut(),
382            &mut self.subscriptions_custom,
383            &cmd.data_type,
384            "",
385            |client| client.unsubscribe(cmd),
386        )
387    }
388
389    /// Subscribes to instrument definitions for a venue, updating internal state and forwarding to the client.
390    ///
391    /// # Errors
392    ///
393    /// Returns an error if the underlying client subscribe operation fails.
394    fn subscribe_instruments(&mut self, cmd: SubscribeInstruments) -> anyhow::Result<()> {
395        Self::execute_tracked_subscribe(
396            self.client.as_mut(),
397            &mut self.subscriptions_instrument_venue,
398            cmd.venue,
399            "instruments",
400            |client| client.subscribe_instruments(cmd),
401        )
402    }
403
404    /// Unsubscribes from instrument definition updates for a venue, updating internal state and forwarding to the client.
405    ///
406    /// # Errors
407    ///
408    /// Returns an error if the underlying client unsubscribe operation fails.
409    fn unsubscribe_instruments(&mut self, cmd: &UnsubscribeInstruments) -> anyhow::Result<()> {
410        Self::execute_tracked_unsubscribe(
411            self.client.as_mut(),
412            &mut self.subscriptions_instrument_venue,
413            &cmd.venue,
414            "instruments",
415            |client| client.unsubscribe_instruments(cmd),
416        )
417    }
418
419    /// Subscribes to instrument definitions for a single instrument, updating internal state and forwarding to the client.
420    ///
421    /// # Errors
422    ///
423    /// Returns an error if the underlying client subscribe operation fails.
424    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
425        Self::execute_tracked_subscribe(
426            self.client.as_mut(),
427            &mut self.subscriptions_instrument,
428            cmd.instrument_id,
429            "instrument",
430            |client| client.subscribe_instrument(cmd),
431        )
432    }
433
434    /// Unsubscribes from instrument definition updates for a single instrument, updating internal state and forwarding to the client.
435    ///
436    /// # Errors
437    ///
438    /// Returns an error if the underlying client unsubscribe operation fails.
439    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
440        Self::execute_tracked_unsubscribe(
441            self.client.as_mut(),
442            &mut self.subscriptions_instrument,
443            &cmd.instrument_id,
444            "instrument",
445            |client| client.unsubscribe_instrument(cmd),
446        )
447    }
448
449    /// Subscribes to book deltas updates for an instrument, updating internal state and forwarding to the client.
450    ///
451    /// # Errors
452    ///
453    /// Returns an error if the underlying client subscribe operation fails.
454    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
455        Self::execute_tracked_subscribe(
456            self.client.as_mut(),
457            &mut self.subscriptions_book_deltas,
458            cmd.instrument_id,
459            "order book deltas",
460            |client| client.subscribe_book_deltas(cmd),
461        )
462    }
463
464    /// Unsubscribes from book deltas for an instrument, updating internal state and forwarding to the client.
465    ///
466    /// # Errors
467    ///
468    /// Returns an error if the underlying client unsubscribe operation fails.
469    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
470        Self::execute_tracked_unsubscribe(
471            self.client.as_mut(),
472            &mut self.subscriptions_book_deltas,
473            &cmd.instrument_id,
474            "order book deltas",
475            |client| client.unsubscribe_book_deltas(cmd),
476        )
477    }
478
479    /// Subscribes to book depth updates for an instrument, updating internal state and forwarding to the client.
480    ///
481    /// # Errors
482    ///
483    /// Returns an error if the underlying client subscribe operation fails.
484    fn subscribe_book_depth10(&mut self, cmd: SubscribeBookDepth10) -> anyhow::Result<()> {
485        Self::execute_tracked_subscribe(
486            self.client.as_mut(),
487            &mut self.subscriptions_book_depth10,
488            cmd.instrument_id,
489            "order book depth",
490            |client| client.subscribe_book_depth10(cmd),
491        )
492    }
493
494    /// Unsubscribes from book depth updates for an instrument, updating internal state and forwarding to the client.
495    ///
496    /// # Errors
497    ///
498    /// Returns an error if the underlying client unsubscribe operation fails.
499    fn unsubscribe_book_depth10(&mut self, cmd: &UnsubscribeBookDepth10) -> anyhow::Result<()> {
500        Self::execute_tracked_unsubscribe(
501            self.client.as_mut(),
502            &mut self.subscriptions_book_depth10,
503            &cmd.instrument_id,
504            "order book depth",
505            |client| client.unsubscribe_book_depth10(cmd),
506        )
507    }
508
509    /// Subscribes to quotes for an instrument, updating internal state and forwarding to the client.
510    ///
511    /// # Errors
512    ///
513    /// Returns an error if the underlying client subscribe operation fails.
514    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
515        Self::execute_tracked_subscribe(
516            self.client.as_mut(),
517            &mut self.subscriptions_quotes,
518            cmd.instrument_id,
519            "quotes",
520            |client| client.subscribe_quotes(cmd),
521        )
522    }
523
524    /// Unsubscribes from quotes for an instrument, updating internal state and forwarding to the client.
525    ///
526    /// # Errors
527    ///
528    /// Returns an error if the underlying client unsubscribe operation fails.
529    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
530        Self::execute_tracked_unsubscribe(
531            self.client.as_mut(),
532            &mut self.subscriptions_quotes,
533            &cmd.instrument_id,
534            "quotes",
535            |client| client.unsubscribe_quotes(cmd),
536        )
537    }
538
539    /// Subscribes to trades for an instrument, updating internal state and forwarding to the client.
540    ///
541    /// # Errors
542    ///
543    /// Returns an error if the underlying client subscribe operation fails.
544    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
545        Self::execute_tracked_subscribe(
546            self.client.as_mut(),
547            &mut self.subscriptions_trades,
548            cmd.instrument_id,
549            "trades",
550            |client| client.subscribe_trades(cmd),
551        )
552    }
553
554    /// Unsubscribes from trades for an instrument, updating internal state and forwarding to the client.
555    ///
556    /// # Errors
557    ///
558    /// Returns an error if the underlying client unsubscribe operation fails.
559    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
560        Self::execute_tracked_unsubscribe(
561            self.client.as_mut(),
562            &mut self.subscriptions_trades,
563            &cmd.instrument_id,
564            "trades",
565            |client| client.unsubscribe_trades(cmd),
566        )
567    }
568
569    /// Subscribes to bars for a bar type, updating internal state and forwarding to the client.
570    ///
571    /// # Errors
572    ///
573    /// Returns an error if the underlying client subscribe operation fails.
574    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
575        Self::execute_tracked_subscribe(
576            self.client.as_mut(),
577            &mut self.subscriptions_bars,
578            cmd.bar_type,
579            "bars",
580            |client| client.subscribe_bars(cmd),
581        )
582    }
583
584    /// Unsubscribes from bars for a bar type, updating internal state and forwarding to the client.
585    ///
586    /// # Errors
587    ///
588    /// Returns an error if the underlying client unsubscribe operation fails.
589    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
590        Self::execute_tracked_unsubscribe(
591            self.client.as_mut(),
592            &mut self.subscriptions_bars,
593            &cmd.bar_type,
594            "bars",
595            |client| client.unsubscribe_bars(cmd),
596        )
597    }
598
599    /// Subscribes to mark price updates for an instrument, updating internal state and forwarding to the client.
600    ///
601    /// # Errors
602    ///
603    /// Returns an error if the underlying client subscribe operation fails.
604    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
605        Self::execute_tracked_subscribe(
606            self.client.as_mut(),
607            &mut self.subscriptions_mark_prices,
608            cmd.instrument_id,
609            "mark prices",
610            |client| client.subscribe_mark_prices(cmd),
611        )
612    }
613
614    /// Unsubscribes from mark price updates for an instrument, updating internal state and forwarding to the client.
615    ///
616    /// # Errors
617    ///
618    /// Returns an error if the underlying client unsubscribe operation fails.
619    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
620        Self::execute_tracked_unsubscribe(
621            self.client.as_mut(),
622            &mut self.subscriptions_mark_prices,
623            &cmd.instrument_id,
624            "mark prices",
625            |client| client.unsubscribe_mark_prices(cmd),
626        )
627    }
628
629    /// Subscribes to index price updates for an instrument, updating internal state and forwarding to the client.
630    ///
631    /// # Errors
632    ///
633    /// Returns an error if the underlying client subscribe operation fails.
634    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
635        Self::execute_tracked_subscribe(
636            self.client.as_mut(),
637            &mut self.subscriptions_index_prices,
638            cmd.instrument_id,
639            "index prices",
640            |client| client.subscribe_index_prices(cmd),
641        )
642    }
643
644    /// Unsubscribes from index price updates for an instrument, updating internal state and forwarding to the client.
645    ///
646    /// # Errors
647    ///
648    /// Returns an error if the underlying client unsubscribe operation fails.
649    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
650        Self::execute_tracked_unsubscribe(
651            self.client.as_mut(),
652            &mut self.subscriptions_index_prices,
653            &cmd.instrument_id,
654            "index prices",
655            |client| client.unsubscribe_index_prices(cmd),
656        )
657    }
658
659    /// Subscribes to funding rate updates for an instrument, updating internal state and forwarding to the client.
660    ///
661    /// # Errors
662    ///
663    /// Returns an error if the underlying client subscribe operation fails.
664    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
665        Self::execute_tracked_subscribe(
666            self.client.as_mut(),
667            &mut self.subscriptions_funding_rates,
668            cmd.instrument_id,
669            "funding rates",
670            |client| client.subscribe_funding_rates(cmd),
671        )
672    }
673
674    /// Unsubscribes from funding rate updates 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_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
680        Self::execute_tracked_unsubscribe(
681            self.client.as_mut(),
682            &mut self.subscriptions_funding_rates,
683            &cmd.instrument_id,
684            "funding rates",
685            |client| client.unsubscribe_funding_rates(cmd),
686        )
687    }
688
689    /// Subscribes to instrument status updates for the specified instrument.
690    ///
691    /// # Errors
692    ///
693    /// Returns an error if the underlying client subscribe operation fails.
694    fn subscribe_instrument_status(
695        &mut self,
696        cmd: SubscribeInstrumentStatus,
697    ) -> anyhow::Result<()> {
698        Self::execute_tracked_subscribe(
699            self.client.as_mut(),
700            &mut self.subscriptions_instrument_status,
701            cmd.instrument_id,
702            "instrument status",
703            |client| client.subscribe_instrument_status(cmd),
704        )
705    }
706
707    /// Unsubscribes from instrument status updates for the specified instrument.
708    ///
709    /// # Errors
710    ///
711    /// Returns an error if the underlying client unsubscribe operation fails.
712    fn unsubscribe_instrument_status(
713        &mut self,
714        cmd: &UnsubscribeInstrumentStatus,
715    ) -> anyhow::Result<()> {
716        Self::execute_tracked_unsubscribe(
717            self.client.as_mut(),
718            &mut self.subscriptions_instrument_status,
719            &cmd.instrument_id,
720            "instrument status",
721            |client| client.unsubscribe_instrument_status(cmd),
722        )
723    }
724
725    /// Subscribes to instrument close events for the specified instrument.
726    ///
727    /// # Errors
728    ///
729    /// Returns an error if the underlying client subscribe operation fails.
730    fn subscribe_instrument_close(&mut self, cmd: SubscribeInstrumentClose) -> anyhow::Result<()> {
731        Self::execute_tracked_subscribe(
732            self.client.as_mut(),
733            &mut self.subscriptions_instrument_close,
734            cmd.instrument_id,
735            "instrument close",
736            |client| client.subscribe_instrument_close(cmd),
737        )
738    }
739
740    /// Unsubscribes from instrument close events for the specified instrument.
741    ///
742    /// # Errors
743    ///
744    /// Returns an error if the underlying client unsubscribe operation fails.
745    fn unsubscribe_instrument_close(
746        &mut self,
747        cmd: &UnsubscribeInstrumentClose,
748    ) -> anyhow::Result<()> {
749        Self::execute_tracked_unsubscribe(
750            self.client.as_mut(),
751            &mut self.subscriptions_instrument_close,
752            &cmd.instrument_id,
753            "instrument close",
754            |client| client.unsubscribe_instrument_close(cmd),
755        )
756    }
757
758    /// Subscribes to option greeks for an instrument, updating internal state and forwarding to the client.
759    ///
760    /// # Errors
761    ///
762    /// Returns an error if the underlying client subscribe operation fails.
763    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
764        Self::execute_tracked_subscribe(
765            self.client.as_mut(),
766            &mut self.subscriptions_option_greeks,
767            cmd.instrument_id,
768            "option greeks",
769            |client| client.subscribe_option_greeks(cmd),
770        )
771    }
772
773    /// Unsubscribes from option greeks for an instrument, updating internal state and forwarding to the client.
774    ///
775    /// # Errors
776    ///
777    /// Returns an error if the underlying client unsubscribe operation fails.
778    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
779        Self::execute_tracked_unsubscribe(
780            self.client.as_mut(),
781            &mut self.subscriptions_option_greeks,
782            &cmd.instrument_id,
783            "option greeks",
784            |client| client.unsubscribe_option_greeks(cmd),
785        )
786    }
787
788    pub(crate) fn execute_tracked_subscribe<T>(
789        client: &mut dyn DataClient,
790        set: &mut AHashSet<T>,
791        key: T,
792        data_type: &str,
793        subscribe: impl FnOnce(&mut dyn DataClient) -> anyhow::Result<()>,
794    ) -> anyhow::Result<()>
795    where
796        T: Eq + Hash + Display,
797    {
798        if set.contains(&key) {
799            return Ok(());
800        }
801
802        subscribe(client)?;
803
804        if data_type.is_empty() {
805            log_info!("Subscribed {key}", color = LogColor::Blue);
806        } else {
807            log_info!("Subscribed {key} {data_type}", color = LogColor::Blue);
808        }
809        set.insert(key);
810        Ok(())
811    }
812
813    pub(crate) fn execute_tracked_unsubscribe<T>(
814        client: &mut dyn DataClient,
815        set: &mut AHashSet<T>,
816        key: &T,
817        data_type: &str,
818        unsubscribe: impl FnOnce(&mut dyn DataClient) -> anyhow::Result<()>,
819    ) -> anyhow::Result<()>
820    where
821        T: Eq + Hash + Display,
822    {
823        if !set.contains(key) {
824            return Ok(());
825        }
826
827        unsubscribe(client)?;
828        set.remove(key);
829        if data_type.is_empty() {
830            log_info!("Unsubscribed {key}", color = LogColor::Blue);
831        } else {
832            log_info!("Unsubscribed {key} {data_type}", color = LogColor::Blue);
833        }
834        Ok(())
835    }
836
837    /// Sends a data request to the underlying client.
838    ///
839    /// # Errors
840    ///
841    /// Returns an error if the client request fails.
842    pub fn request_data(&self, req: RequestCustomData) -> anyhow::Result<()> {
843        self.client.request_data(req)
844    }
845
846    /// Sends a single instrument request to the client.
847    ///
848    /// # Errors
849    ///
850    /// Returns an error if the client fails to process the request.
851    pub fn request_instrument(&self, req: RequestInstrument) -> anyhow::Result<()> {
852        self.client.request_instrument(req)
853    }
854
855    /// Sends a batch instruments request to the client.
856    ///
857    /// # Errors
858    ///
859    /// Returns an error if the client fails to process the request.
860    pub fn request_instruments(&self, req: RequestInstruments) -> anyhow::Result<()> {
861        self.client.request_instruments(req)
862    }
863
864    /// Sends a book snapshot request for a given instrument.
865    ///
866    /// # Errors
867    ///
868    /// Returns an error if the client fails to process the book snapshot request.
869    pub fn request_book_snapshot(&self, req: RequestBookSnapshot) -> anyhow::Result<()> {
870        self.client.request_book_snapshot(req)
871    }
872
873    /// Sends a quotes request for a given instrument.
874    ///
875    /// # Errors
876    ///
877    /// Returns an error if the client fails to process the quotes request.
878    pub fn request_quotes(&self, req: RequestQuotes) -> anyhow::Result<()> {
879        self.client.request_quotes(req)
880    }
881
882    /// Sends a trades request for a given instrument.
883    ///
884    /// # Errors
885    ///
886    /// Returns an error if the client fails to process the trades request.
887    pub fn request_trades(&self, req: RequestTrades) -> anyhow::Result<()> {
888        self.client.request_trades(req)
889    }
890
891    /// Sends a funding rates request for a given instrument.
892    ///
893    /// # Errors
894    ///
895    /// Returns an error if the client fails to process the trades request.
896    pub fn request_funding_rates(&self, req: RequestFundingRates) -> anyhow::Result<()> {
897        self.client.request_funding_rates(req)
898    }
899
900    /// Sends an option-chain reference price request.
901    ///
902    /// # Errors
903    ///
904    /// Returns an error if the client fails to process the option-chain reference price request.
905    pub fn request_option_chain_reference_price(
906        &self,
907        req: RequestOptionChainReferencePrice,
908    ) -> anyhow::Result<()> {
909        self.client.request_option_chain_reference_price(req)
910    }
911
912    /// Sends a bars request for a given instrument and bar type.
913    ///
914    /// # Errors
915    ///
916    /// Returns an error if the client fails to process the bars request.
917    pub fn request_bars(&self, req: RequestBars) -> anyhow::Result<()> {
918        self.client.request_bars(req)
919    }
920
921    /// Sends an order book depths request for a given instrument.
922    ///
923    /// # Errors
924    ///
925    /// Returns an error if the client fails to process the order book depths request.
926    pub fn request_book_depth(&self, req: RequestBookDepth) -> anyhow::Result<()> {
927        self.client.request_book_depth(req)
928    }
929}
nautilus_data/client.rs

nautilus_data/
client.rs
