nautilus_common/messages/data/
mod.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//! Data specific messages such as subscriptions and requests.
17
18use std::{any::Any, sync::Arc};
19
20use nautilus_core::{Params, UUID4, UnixNanos};
21use nautilus_model::{
22    data::BarType,
23    identifiers::{ClientId, Venue},
24};
25use serde::{Deserialize, Serialize};
26
27pub mod request;
28pub mod response;
29pub mod subscribe;
30pub mod unsubscribe;
31
32/// Params key used to flag a book subscription as targeting a parent symbol.
33///
34/// When the boolean value is `true`, the subscription fans out across all
35/// instruments that resolve from the parent components (see
36/// [`InstrumentId::parse_parent_components`]). When absent or `false`, the
37/// subscription is routed to the concrete instrument id only.
38///
39/// [`InstrumentId::parse_parent_components`]: nautilus_model::identifiers::InstrumentId::parse_parent_components
40pub const PARAMS_IS_PARENT: &str = "is_parent";
41
42// Re-exports
43pub use request::{
44    RequestBars, RequestBookDeltas, RequestBookDepth, RequestBookSnapshot, RequestCustomData,
45    RequestFundingRates, RequestInstrument, RequestInstruments, RequestJoin,
46    RequestOptionChainReferencePrice, RequestQuotes, RequestTrades,
47};
48pub use response::{
49    BarsResponse, BookDeltasResponse, BookDepthResponse, BookResponse, CustomDataResponse,
50    FundingRatesResponse, InstrumentResponse, InstrumentsResponse,
51    OptionChainReferencePriceResponse, QuotesResponse, TradesResponse,
52};
53pub use subscribe::{
54    SubscribeBars, SubscribeBookDeltas, SubscribeBookDepth, SubscribeBookSnapshots,
55    SubscribeCustomData, SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument,
56    SubscribeInstrumentClose, SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices,
57    SubscribeOptionChain, SubscribeOptionGreeks, SubscribeQuotes, SubscribeTrades,
58};
59pub use unsubscribe::{
60    UnsubscribeBars, UnsubscribeBookDeltas, UnsubscribeBookDepth, UnsubscribeBookSnapshots,
61    UnsubscribeCustomData, UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrument,
62    UnsubscribeInstrumentClose, UnsubscribeInstrumentStatus, UnsubscribeInstruments,
63    UnsubscribeMarkPrices, UnsubscribeOptionChain, UnsubscribeOptionGreeks, UnsubscribeQuotes,
64    UnsubscribeTrades,
65};
66
67#[cfg(feature = "defi")]
68use crate::messages::defi::{DefiRequestCommand, DefiSubscribeCommand, DefiUnsubscribeCommand};
69
70#[non_exhaustive]
71#[derive(Clone, Debug, PartialEq)]
72pub enum DataCommand {
73    Request(RequestCommand),
74    Subscribe(SubscribeCommand),
75    Unsubscribe(UnsubscribeCommand),
76    #[cfg(feature = "defi")]
77    DefiRequest(DefiRequestCommand),
78    #[cfg(feature = "defi")]
79    DefiSubscribe(DefiSubscribeCommand),
80    #[cfg(feature = "defi")]
81    DefiUnsubscribe(DefiUnsubscribeCommand),
82}
83
84impl DataCommand {
85    /// Converts the command to a dyn Any trait object for messaging.
86    pub fn as_any(&self) -> &dyn Any {
87        self
88    }
89
90    /// Converts a subscribe variant into its matching unsubscribe variant.
91    ///
92    /// Returns `None` for request and unsubscribe variants.
93    pub(crate) fn into_unsubscribe(self, command_id: UUID4, ts_init: UnixNanos) -> Option<Self> {
94        match self {
95            Self::Subscribe(command) => {
96                let correlation_id = matches!(
97                    command,
98                    SubscribeCommand::BookDeltas(_)
99                        | SubscribeCommand::BookDepth(_)
100                        | SubscribeCommand::BookSnapshots(_)
101                )
102                .then(|| command.command_id());
103                Some(Self::Unsubscribe(command.into_unsubscribe(
104                    command_id,
105                    ts_init,
106                    correlation_id,
107                )))
108            }
109            #[cfg(feature = "defi")]
110            Self::DefiSubscribe(command) => Some(Self::DefiUnsubscribe(
111                command.into_unsubscribe(command_id, ts_init),
112            )),
113            _ => None,
114        }
115    }
116}
117
118#[derive(Clone, Debug, Serialize, Deserialize)]
119pub enum SubscribeCommand {
120    Data(SubscribeCustomData),
121    Instrument(SubscribeInstrument),
122    Instruments(SubscribeInstruments),
123    BookDeltas(SubscribeBookDeltas),
124    BookDepth(SubscribeBookDepth),
125    BookSnapshots(SubscribeBookSnapshots),
126    Quotes(SubscribeQuotes),
127    Trades(SubscribeTrades),
128    Bars(SubscribeBars),
129    MarkPrices(SubscribeMarkPrices),
130    IndexPrices(SubscribeIndexPrices),
131    FundingRates(SubscribeFundingRates),
132    InstrumentStatus(SubscribeInstrumentStatus),
133    InstrumentClose(SubscribeInstrumentClose),
134    OptionGreeks(SubscribeOptionGreeks),
135    OptionChain(SubscribeOptionChain),
136}
137
138impl PartialEq for SubscribeCommand {
139    fn eq(&self, other: &Self) -> bool {
140        self.command_id() == other.command_id()
141    }
142}
143
144impl SubscribeCommand {
145    /// Converts the command to a dyn Any trait object for messaging.
146    pub fn as_any(&self) -> &dyn Any {
147        self
148    }
149
150    /// Converts this subscribe command into its matching unsubscribe command.
151    ///
152    /// Preserves the subscribed data identity and client route while replacing the command ID and
153    /// initialization timestamp. It also preserves parameters and sets the supplied correlation ID
154    /// when the matching unsubscribe command supports those fields.
155    #[must_use]
156    pub fn into_unsubscribe(
157        self,
158        command_id: UUID4,
159        ts_init: UnixNanos,
160        correlation_id: Option<UUID4>,
161    ) -> UnsubscribeCommand {
162        match self {
163            Self::Data(cmd) => UnsubscribeCommand::Data(UnsubscribeCustomData::new(
164                cmd.client_id,
165                cmd.venue,
166                cmd.data_type,
167                command_id,
168                ts_init,
169                correlation_id,
170                cmd.params,
171            )),
172            Self::Instrument(cmd) => UnsubscribeCommand::Instrument(UnsubscribeInstrument::new(
173                cmd.instrument_id,
174                cmd.client_id,
175                cmd.venue,
176                command_id,
177                ts_init,
178                correlation_id,
179                cmd.params,
180            )),
181            Self::Instruments(cmd) => UnsubscribeCommand::Instruments(UnsubscribeInstruments::new(
182                cmd.client_id,
183                cmd.venue,
184                command_id,
185                ts_init,
186                correlation_id,
187                cmd.params,
188            )),
189            Self::BookDeltas(cmd) => UnsubscribeCommand::BookDeltas(UnsubscribeBookDeltas::new(
190                cmd.instrument_id,
191                cmd.client_id,
192                cmd.venue,
193                command_id,
194                ts_init,
195                correlation_id,
196                cmd.params,
197            )),
198            Self::BookDepth(cmd) => UnsubscribeCommand::BookDepth(UnsubscribeBookDepth::new(
199                cmd.instrument_id,
200                cmd.client_id,
201                cmd.venue,
202                command_id,
203                ts_init,
204                correlation_id,
205                cmd.params,
206            )),
207            Self::BookSnapshots(cmd) => {
208                UnsubscribeCommand::BookSnapshots(UnsubscribeBookSnapshots::new(
209                    cmd.instrument_id,
210                    cmd.interval_ms,
211                    cmd.client_id,
212                    cmd.venue,
213                    command_id,
214                    ts_init,
215                    correlation_id,
216                    cmd.params,
217                ))
218            }
219            Self::Quotes(cmd) => UnsubscribeCommand::Quotes(UnsubscribeQuotes::new(
220                cmd.instrument_id,
221                cmd.client_id,
222                cmd.venue,
223                command_id,
224                ts_init,
225                correlation_id,
226                cmd.params,
227            )),
228            Self::Trades(cmd) => UnsubscribeCommand::Trades(UnsubscribeTrades::new(
229                cmd.instrument_id,
230                cmd.client_id,
231                cmd.venue,
232                command_id,
233                ts_init,
234                correlation_id,
235                cmd.params,
236            )),
237            Self::Bars(cmd) => UnsubscribeCommand::Bars(UnsubscribeBars::new(
238                cmd.bar_type,
239                cmd.client_id,
240                cmd.venue,
241                command_id,
242                ts_init,
243                correlation_id,
244                cmd.params,
245            )),
246            Self::MarkPrices(cmd) => UnsubscribeCommand::MarkPrices(UnsubscribeMarkPrices::new(
247                cmd.instrument_id,
248                cmd.client_id,
249                cmd.venue,
250                command_id,
251                ts_init,
252                correlation_id,
253                cmd.params,
254            )),
255            Self::IndexPrices(cmd) => UnsubscribeCommand::IndexPrices(UnsubscribeIndexPrices::new(
256                cmd.instrument_id,
257                cmd.client_id,
258                cmd.venue,
259                command_id,
260                ts_init,
261                correlation_id,
262                cmd.params,
263            )),
264            Self::FundingRates(cmd) => {
265                UnsubscribeCommand::FundingRates(UnsubscribeFundingRates::new(
266                    cmd.instrument_id,
267                    cmd.client_id,
268                    cmd.venue,
269                    command_id,
270                    ts_init,
271                    correlation_id,
272                    cmd.params,
273                ))
274            }
275            Self::InstrumentStatus(cmd) => {
276                UnsubscribeCommand::InstrumentStatus(UnsubscribeInstrumentStatus::new(
277                    cmd.instrument_id,
278                    cmd.client_id,
279                    cmd.venue,
280                    command_id,
281                    ts_init,
282                    correlation_id,
283                    cmd.params,
284                ))
285            }
286            Self::InstrumentClose(cmd) => {
287                UnsubscribeCommand::InstrumentClose(UnsubscribeInstrumentClose::new(
288                    cmd.instrument_id,
289                    cmd.client_id,
290                    cmd.venue,
291                    command_id,
292                    ts_init,
293                    correlation_id,
294                    cmd.params,
295                ))
296            }
297            Self::OptionGreeks(cmd) => {
298                UnsubscribeCommand::OptionGreeks(UnsubscribeOptionGreeks::new(
299                    cmd.instrument_id,
300                    cmd.client_id,
301                    cmd.venue,
302                    command_id,
303                    ts_init,
304                    correlation_id,
305                    cmd.params,
306                ))
307            }
308            Self::OptionChain(cmd) => {
309                let mut unsubscribe = UnsubscribeOptionChain::new(
310                    cmd.series_id,
311                    command_id,
312                    ts_init,
313                    cmd.client_id,
314                    cmd.venue,
315                );
316                unsubscribe.params = cmd.params;
317                UnsubscribeCommand::OptionChain(unsubscribe)
318            }
319        }
320    }
321
322    pub fn command_id(&self) -> UUID4 {
323        match self {
324            Self::Data(cmd) => cmd.command_id,
325            Self::Instrument(cmd) => cmd.command_id,
326            Self::Instruments(cmd) => cmd.command_id,
327            Self::BookDeltas(cmd) => cmd.command_id,
328            Self::BookDepth(cmd) => cmd.command_id,
329            Self::BookSnapshots(cmd) => cmd.command_id,
330            Self::Quotes(cmd) => cmd.command_id,
331            Self::Trades(cmd) => cmd.command_id,
332            Self::Bars(cmd) => cmd.command_id,
333            Self::MarkPrices(cmd) => cmd.command_id,
334            Self::IndexPrices(cmd) => cmd.command_id,
335            Self::FundingRates(cmd) => cmd.command_id,
336            Self::InstrumentStatus(cmd) => cmd.command_id,
337            Self::InstrumentClose(cmd) => cmd.command_id,
338            Self::OptionGreeks(cmd) => cmd.command_id,
339            Self::OptionChain(cmd) => cmd.command_id,
340        }
341    }
342
343    pub fn client_id(&self) -> Option<&ClientId> {
344        match self {
345            Self::Data(cmd) => cmd.client_id.as_ref(),
346            Self::Instrument(cmd) => cmd.client_id.as_ref(),
347            Self::Instruments(cmd) => cmd.client_id.as_ref(),
348            Self::BookDeltas(cmd) => cmd.client_id.as_ref(),
349            Self::BookDepth(cmd) => cmd.client_id.as_ref(),
350            Self::BookSnapshots(cmd) => cmd.client_id.as_ref(),
351            Self::Quotes(cmd) => cmd.client_id.as_ref(),
352            Self::Trades(cmd) => cmd.client_id.as_ref(),
353            Self::MarkPrices(cmd) => cmd.client_id.as_ref(),
354            Self::IndexPrices(cmd) => cmd.client_id.as_ref(),
355            Self::FundingRates(cmd) => cmd.client_id.as_ref(),
356            Self::Bars(cmd) => cmd.client_id.as_ref(),
357            Self::InstrumentStatus(cmd) => cmd.client_id.as_ref(),
358            Self::InstrumentClose(cmd) => cmd.client_id.as_ref(),
359            Self::OptionGreeks(cmd) => cmd.client_id.as_ref(),
360            Self::OptionChain(cmd) => cmd.client_id.as_ref(),
361        }
362    }
363
364    pub fn venue(&self) -> Option<&Venue> {
365        match self {
366            Self::Data(cmd) => cmd.venue.as_ref(),
367            Self::Instrument(cmd) => cmd.venue.as_ref(),
368            Self::Instruments(cmd) => Some(&cmd.venue),
369            Self::BookDeltas(cmd) => cmd.venue.as_ref(),
370            Self::BookDepth(cmd) => cmd.venue.as_ref(),
371            Self::BookSnapshots(cmd) => cmd.venue.as_ref(),
372            Self::Quotes(cmd) => cmd.venue.as_ref(),
373            Self::Trades(cmd) => cmd.venue.as_ref(),
374            Self::MarkPrices(cmd) => cmd.venue.as_ref(),
375            Self::IndexPrices(cmd) => cmd.venue.as_ref(),
376            Self::FundingRates(cmd) => cmd.venue.as_ref(),
377            Self::Bars(cmd) => cmd.venue.as_ref(),
378            Self::InstrumentStatus(cmd) => cmd.venue.as_ref(),
379            Self::InstrumentClose(cmd) => cmd.venue.as_ref(),
380            Self::OptionGreeks(cmd) => cmd.venue.as_ref(),
381            Self::OptionChain(cmd) => cmd.venue.as_ref(),
382        }
383    }
384
385    pub fn ts_init(&self) -> UnixNanos {
386        match self {
387            Self::Data(cmd) => cmd.ts_init,
388            Self::Instrument(cmd) => cmd.ts_init,
389            Self::Instruments(cmd) => cmd.ts_init,
390            Self::BookDeltas(cmd) => cmd.ts_init,
391            Self::BookDepth(cmd) => cmd.ts_init,
392            Self::BookSnapshots(cmd) => cmd.ts_init,
393            Self::Quotes(cmd) => cmd.ts_init,
394            Self::Trades(cmd) => cmd.ts_init,
395            Self::MarkPrices(cmd) => cmd.ts_init,
396            Self::IndexPrices(cmd) => cmd.ts_init,
397            Self::FundingRates(cmd) => cmd.ts_init,
398            Self::Bars(cmd) => cmd.ts_init,
399            Self::InstrumentStatus(cmd) => cmd.ts_init,
400            Self::InstrumentClose(cmd) => cmd.ts_init,
401            Self::OptionGreeks(cmd) => cmd.ts_init,
402            Self::OptionChain(cmd) => cmd.ts_init,
403        }
404    }
405
406    pub fn correlation_id(&self) -> Option<UUID4> {
407        match self {
408            Self::Data(cmd) => cmd.correlation_id,
409            Self::Instrument(cmd) => cmd.correlation_id,
410            Self::Instruments(cmd) => cmd.correlation_id,
411            Self::BookDeltas(cmd) => cmd.correlation_id,
412            Self::BookDepth(cmd) => cmd.correlation_id,
413            Self::BookSnapshots(cmd) => cmd.correlation_id,
414            Self::Quotes(cmd) => cmd.correlation_id,
415            Self::Trades(cmd) => cmd.correlation_id,
416            Self::MarkPrices(cmd) => cmd.correlation_id,
417            Self::IndexPrices(cmd) => cmd.correlation_id,
418            Self::FundingRates(cmd) => cmd.correlation_id,
419            Self::Bars(cmd) => cmd.correlation_id,
420            Self::InstrumentStatus(cmd) => cmd.correlation_id,
421            Self::InstrumentClose(cmd) => cmd.correlation_id,
422            Self::OptionGreeks(cmd) => cmd.correlation_id,
423            Self::OptionChain(cmd) => cmd.correlation_id,
424        }
425    }
426
427    pub fn params(&self) -> Option<&Params> {
428        match self {
429            Self::Data(cmd) => cmd.params.as_ref(),
430            Self::Instrument(cmd) => cmd.params.as_ref(),
431            Self::Instruments(cmd) => cmd.params.as_ref(),
432            Self::BookDeltas(cmd) => cmd.params.as_ref(),
433            Self::BookDepth(cmd) => cmd.params.as_ref(),
434            Self::BookSnapshots(cmd) => cmd.params.as_ref(),
435            Self::Quotes(cmd) => cmd.params.as_ref(),
436            Self::Trades(cmd) => cmd.params.as_ref(),
437            Self::Bars(cmd) => cmd.params.as_ref(),
438            Self::MarkPrices(cmd) => cmd.params.as_ref(),
439            Self::IndexPrices(cmd) => cmd.params.as_ref(),
440            Self::FundingRates(cmd) => cmd.params.as_ref(),
441            Self::InstrumentStatus(cmd) => cmd.params.as_ref(),
442            Self::InstrumentClose(cmd) => cmd.params.as_ref(),
443            Self::OptionGreeks(cmd) => cmd.params.as_ref(),
444            Self::OptionChain(cmd) => cmd.params.as_ref(),
445        }
446    }
447}
448
449#[derive(Clone, Debug, Serialize, Deserialize)]
450pub enum UnsubscribeCommand {
451    Data(UnsubscribeCustomData),
452    Instrument(UnsubscribeInstrument),
453    Instruments(UnsubscribeInstruments),
454    BookDeltas(UnsubscribeBookDeltas),
455    BookDepth(UnsubscribeBookDepth),
456    BookSnapshots(UnsubscribeBookSnapshots),
457    Quotes(UnsubscribeQuotes),
458    Trades(UnsubscribeTrades),
459    Bars(UnsubscribeBars),
460    MarkPrices(UnsubscribeMarkPrices),
461    IndexPrices(UnsubscribeIndexPrices),
462    FundingRates(UnsubscribeFundingRates),
463    InstrumentStatus(UnsubscribeInstrumentStatus),
464    InstrumentClose(UnsubscribeInstrumentClose),
465    OptionGreeks(UnsubscribeOptionGreeks),
466    OptionChain(UnsubscribeOptionChain),
467}
468
469impl PartialEq for UnsubscribeCommand {
470    fn eq(&self, other: &Self) -> bool {
471        self.command_id() == other.command_id()
472    }
473}
474
475impl UnsubscribeCommand {
476    /// Converts the command to a dyn Any trait object for messaging.
477    pub fn as_any(&self) -> &dyn Any {
478        self
479    }
480
481    pub fn command_id(&self) -> UUID4 {
482        match self {
483            Self::Data(cmd) => cmd.command_id,
484            Self::Instrument(cmd) => cmd.command_id,
485            Self::Instruments(cmd) => cmd.command_id,
486            Self::BookDeltas(cmd) => cmd.command_id,
487            Self::BookDepth(cmd) => cmd.command_id,
488            Self::BookSnapshots(cmd) => cmd.command_id,
489            Self::Quotes(cmd) => cmd.command_id,
490            Self::Trades(cmd) => cmd.command_id,
491            Self::Bars(cmd) => cmd.command_id,
492            Self::MarkPrices(cmd) => cmd.command_id,
493            Self::IndexPrices(cmd) => cmd.command_id,
494            Self::FundingRates(cmd) => cmd.command_id,
495            Self::InstrumentStatus(cmd) => cmd.command_id,
496            Self::InstrumentClose(cmd) => cmd.command_id,
497            Self::OptionGreeks(cmd) => cmd.command_id,
498            Self::OptionChain(cmd) => cmd.command_id,
499        }
500    }
501
502    pub fn client_id(&self) -> Option<&ClientId> {
503        match self {
504            Self::Data(cmd) => cmd.client_id.as_ref(),
505            Self::Instrument(cmd) => cmd.client_id.as_ref(),
506            Self::Instruments(cmd) => cmd.client_id.as_ref(),
507            Self::BookDeltas(cmd) => cmd.client_id.as_ref(),
508            Self::BookDepth(cmd) => cmd.client_id.as_ref(),
509            Self::BookSnapshots(cmd) => cmd.client_id.as_ref(),
510            Self::Quotes(cmd) => cmd.client_id.as_ref(),
511            Self::Trades(cmd) => cmd.client_id.as_ref(),
512            Self::Bars(cmd) => cmd.client_id.as_ref(),
513            Self::MarkPrices(cmd) => cmd.client_id.as_ref(),
514            Self::IndexPrices(cmd) => cmd.client_id.as_ref(),
515            Self::FundingRates(cmd) => cmd.client_id.as_ref(),
516            Self::InstrumentStatus(cmd) => cmd.client_id.as_ref(),
517            Self::InstrumentClose(cmd) => cmd.client_id.as_ref(),
518            Self::OptionGreeks(cmd) => cmd.client_id.as_ref(),
519            Self::OptionChain(cmd) => cmd.client_id.as_ref(),
520        }
521    }
522
523    pub fn venue(&self) -> Option<&Venue> {
524        match self {
525            Self::Data(cmd) => cmd.venue.as_ref(),
526            Self::Instrument(cmd) => cmd.venue.as_ref(),
527            Self::Instruments(cmd) => Some(&cmd.venue),
528            Self::BookDeltas(cmd) => cmd.venue.as_ref(),
529            Self::BookDepth(cmd) => cmd.venue.as_ref(),
530            Self::BookSnapshots(cmd) => cmd.venue.as_ref(),
531            Self::Quotes(cmd) => cmd.venue.as_ref(),
532            Self::Trades(cmd) => cmd.venue.as_ref(),
533            Self::Bars(cmd) => cmd.venue.as_ref(),
534            Self::MarkPrices(cmd) => cmd.venue.as_ref(),
535            Self::IndexPrices(cmd) => cmd.venue.as_ref(),
536            Self::FundingRates(cmd) => cmd.venue.as_ref(),
537            Self::InstrumentStatus(cmd) => cmd.venue.as_ref(),
538            Self::InstrumentClose(cmd) => cmd.venue.as_ref(),
539            Self::OptionGreeks(cmd) => cmd.venue.as_ref(),
540            Self::OptionChain(cmd) => cmd.venue.as_ref(),
541        }
542    }
543
544    pub fn ts_init(&self) -> UnixNanos {
545        match self {
546            Self::Data(cmd) => cmd.ts_init,
547            Self::Instrument(cmd) => cmd.ts_init,
548            Self::Instruments(cmd) => cmd.ts_init,
549            Self::BookDeltas(cmd) => cmd.ts_init,
550            Self::BookDepth(cmd) => cmd.ts_init,
551            Self::BookSnapshots(cmd) => cmd.ts_init,
552            Self::Quotes(cmd) => cmd.ts_init,
553            Self::Trades(cmd) => cmd.ts_init,
554            Self::MarkPrices(cmd) => cmd.ts_init,
555            Self::IndexPrices(cmd) => cmd.ts_init,
556            Self::FundingRates(cmd) => cmd.ts_init,
557            Self::Bars(cmd) => cmd.ts_init,
558            Self::InstrumentStatus(cmd) => cmd.ts_init,
559            Self::InstrumentClose(cmd) => cmd.ts_init,
560            Self::OptionGreeks(cmd) => cmd.ts_init,
561            Self::OptionChain(cmd) => cmd.ts_init,
562        }
563    }
564
565    pub fn correlation_id(&self) -> Option<UUID4> {
566        match self {
567            Self::Data(cmd) => cmd.correlation_id,
568            Self::Instrument(cmd) => cmd.correlation_id,
569            Self::Instruments(cmd) => cmd.correlation_id,
570            Self::BookDeltas(cmd) => cmd.correlation_id,
571            Self::BookDepth(cmd) => cmd.correlation_id,
572            Self::BookSnapshots(cmd) => cmd.correlation_id,
573            Self::Quotes(cmd) => cmd.correlation_id,
574            Self::Trades(cmd) => cmd.correlation_id,
575            Self::MarkPrices(cmd) => cmd.correlation_id,
576            Self::IndexPrices(cmd) => cmd.correlation_id,
577            Self::FundingRates(cmd) => cmd.correlation_id,
578            Self::Bars(cmd) => cmd.correlation_id,
579            Self::InstrumentStatus(cmd) => cmd.correlation_id,
580            Self::InstrumentClose(cmd) => cmd.correlation_id,
581            Self::OptionGreeks(cmd) => cmd.correlation_id,
582            Self::OptionChain(_) => None,
583        }
584    }
585}
586
587#[allow(
588    clippy::ref_option,
589    reason = "callers pass borrowed Option fields directly"
590)]
591fn check_client_id_or_venue(client_id: &Option<ClientId>, venue: &Option<Venue>) {
592    assert!(
593        client_id.is_some() || venue.is_some(),
594        "Both `client_id` and `venue` were None"
595    );
596}
597
598#[derive(Clone, Debug, Serialize, Deserialize)]
599pub enum RequestCommand {
600    Data(RequestCustomData),
601    Instrument(RequestInstrument),
602    Instruments(RequestInstruments),
603    BookSnapshot(RequestBookSnapshot),
604    BookDeltas(RequestBookDeltas),
605    BookDepth(RequestBookDepth),
606    Quotes(RequestQuotes),
607    Trades(RequestTrades),
608    FundingRates(RequestFundingRates),
609    OptionChainReferencePrice(RequestOptionChainReferencePrice),
610    Bars(RequestBars),
611    Join(RequestJoin),
612}
613
614impl PartialEq for RequestCommand {
615    fn eq(&self, other: &Self) -> bool {
616        self.request_id() == other.request_id()
617    }
618}
619
620impl RequestCommand {
621    /// Converts the command to a dyn Any trait object for messaging.
622    pub fn as_any(&self) -> &dyn Any {
623        self
624    }
625
626    pub fn request_id(&self) -> &UUID4 {
627        match self {
628            Self::Data(cmd) => &cmd.request_id,
629            Self::Instrument(cmd) => &cmd.request_id,
630            Self::Instruments(cmd) => &cmd.request_id,
631            Self::BookSnapshot(cmd) => &cmd.request_id,
632            Self::BookDeltas(cmd) => &cmd.request_id,
633            Self::BookDepth(cmd) => &cmd.request_id,
634            Self::Quotes(cmd) => &cmd.request_id,
635            Self::Trades(cmd) => &cmd.request_id,
636            Self::FundingRates(cmd) => &cmd.request_id,
637            Self::OptionChainReferencePrice(cmd) => &cmd.request_id,
638            Self::Bars(cmd) => &cmd.request_id,
639            Self::Join(cmd) => &cmd.request_id,
640        }
641    }
642
643    pub fn client_id(&self) -> Option<&ClientId> {
644        match self {
645            Self::Data(cmd) => Some(&cmd.client_id),
646            Self::Instrument(cmd) => cmd.client_id.as_ref(),
647            Self::Instruments(cmd) => cmd.client_id.as_ref(),
648            Self::BookSnapshot(cmd) => cmd.client_id.as_ref(),
649            Self::BookDeltas(cmd) => cmd.client_id.as_ref(),
650            Self::BookDepth(cmd) => cmd.client_id.as_ref(),
651            Self::Quotes(cmd) => cmd.client_id.as_ref(),
652            Self::Trades(cmd) => cmd.client_id.as_ref(),
653            Self::FundingRates(cmd) => cmd.client_id.as_ref(),
654            Self::OptionChainReferencePrice(cmd) => cmd.client_id.as_ref(),
655            Self::Bars(cmd) => cmd.client_id.as_ref(),
656            Self::Join(_) => None,
657        }
658    }
659
660    pub fn venue(&self) -> Option<&Venue> {
661        match self {
662            Self::Data(_) => None,
663            Self::Instrument(cmd) => Some(&cmd.instrument_id.venue),
664            Self::Instruments(cmd) => cmd.venue.as_ref(),
665            Self::BookSnapshot(cmd) => Some(&cmd.instrument_id.venue),
666            Self::BookDeltas(cmd) => Some(&cmd.instrument_id.venue),
667            Self::BookDepth(cmd) => Some(&cmd.instrument_id.venue),
668            Self::Quotes(cmd) => Some(&cmd.instrument_id.venue),
669            Self::Trades(cmd) => Some(&cmd.instrument_id.venue),
670            Self::FundingRates(cmd) => Some(&cmd.instrument_id.venue),
671            Self::OptionChainReferencePrice(cmd) => Some(&cmd.series_id.venue),
672            // TODO: Extract the below somewhere
673            Self::Bars(cmd) => match &cmd.bar_type {
674                BarType::Standard { instrument_id, .. } => Some(&instrument_id.venue),
675                BarType::Composite { instrument_id, .. } => Some(&instrument_id.venue),
676            },
677            Self::Join(_) => None,
678        }
679    }
680
681    pub fn ts_init(&self) -> UnixNanos {
682        match self {
683            Self::Data(cmd) => cmd.ts_init,
684            Self::Instrument(cmd) => cmd.ts_init,
685            Self::Instruments(cmd) => cmd.ts_init,
686            Self::BookSnapshot(cmd) => cmd.ts_init,
687            Self::BookDeltas(cmd) => cmd.ts_init,
688            Self::BookDepth(cmd) => cmd.ts_init,
689            Self::Quotes(cmd) => cmd.ts_init,
690            Self::Trades(cmd) => cmd.ts_init,
691            Self::FundingRates(cmd) => cmd.ts_init,
692            Self::OptionChainReferencePrice(cmd) => cmd.ts_init,
693            Self::Bars(cmd) => cmd.ts_init,
694            Self::Join(cmd) => cmd.ts_init,
695        }
696    }
697}
698
699#[derive(Clone, Debug)]
700pub enum DataResponse {
701    Data(CustomDataResponse),
702    Instrument(Box<InstrumentResponse>),
703    Instruments(InstrumentsResponse),
704    Book(BookResponse),
705    BookDeltas(BookDeltasResponse),
706    BookDepth(BookDepthResponse),
707    Quotes(QuotesResponse),
708    Trades(TradesResponse),
709    FundingRates(FundingRatesResponse),
710    OptionChainReferencePrice(OptionChainReferencePriceResponse),
711    Bars(BarsResponse),
712}
713
714impl DataResponse {
715    /// Converts the command to a dyn Any trait object for messaging.
716    pub fn as_any(&self) -> &dyn Any {
717        self
718    }
719
720    pub fn correlation_id(&self) -> &UUID4 {
721        match self {
722            Self::Data(resp) => &resp.correlation_id,
723            Self::Instrument(resp) => &resp.correlation_id,
724            Self::Instruments(resp) => &resp.correlation_id,
725            Self::Book(resp) => &resp.correlation_id,
726            Self::BookDeltas(resp) => &resp.correlation_id,
727            Self::BookDepth(resp) => &resp.correlation_id,
728            Self::Quotes(resp) => &resp.correlation_id,
729            Self::Trades(resp) => &resp.correlation_id,
730            Self::FundingRates(resp) => &resp.correlation_id,
731            Self::OptionChainReferencePrice(resp) => &resp.correlation_id,
732            Self::Bars(resp) => &resp.correlation_id,
733        }
734    }
735
736    /// Returns a short variant name for compact logging.
737    #[must_use]
738    pub fn kind(&self) -> &'static str {
739        match self {
740            Self::Data(_) => "Data",
741            Self::Instrument(_) => "Instrument",
742            Self::Instruments(_) => "Instruments",
743            Self::Book(_) => "Book",
744            Self::BookDeltas(_) => "BookDeltas",
745            Self::BookDepth(_) => "BookDepth",
746            Self::Quotes(_) => "Quotes",
747            Self::Trades(_) => "Trades",
748            Self::FundingRates(_) => "FundingRates",
749            Self::OptionChainReferencePrice(_) => "OptionChainReferencePrice",
750            Self::Bars(_) => "Bars",
751        }
752    }
753
754    /// Returns the number of records carried by the response, where defined.
755    ///
756    /// Returns `None` for singular or opaque variants (`Data`, `Instrument`, `Book`)
757    /// where a record count is not meaningful.
758    #[must_use]
759    pub fn record_count(&self) -> Option<usize> {
760        match self {
761            Self::Data(_) | Self::Instrument(_) | Self::Book(_) => None,
762            Self::Instruments(resp) => Some(resp.data.len()),
763            Self::BookDeltas(resp) => Some(resp.data.len()),
764            Self::BookDepth(resp) => Some(resp.data.len()),
765            Self::Quotes(resp) => Some(resp.data.len()),
766            Self::Trades(resp) => Some(resp.data.len()),
767            Self::FundingRates(resp) => Some(resp.data.len()),
768            Self::OptionChainReferencePrice(_) => None,
769            Self::Bars(resp) => Some(resp.data.len()),
770        }
771    }
772
773    /// Trims vector payloads to the inclusive `[start, end]` window on `ts_init`.
774    ///
775    /// Applies to variants whose payload elements implement `HasTsInit`
776    /// (`BookDeltas`, `BookDepth`, `Quotes`, `Trades`, `FundingRates`,
777    /// `Bars`, `Instruments`). Other variants are untouched: singular payloads
778    /// (`Instrument`, `Book`),
779    /// `OptionChainReferencePrice` (singular), and the opaque custom
780    /// `Data` variant.
781    pub fn trim_to_bounds(&mut self) {
782        match self {
783            Self::Quotes(r) => response::trim_data_to_bounds(&mut r.data, r.start, r.end),
784            Self::Trades(r) => response::trim_data_to_bounds(&mut r.data, r.start, r.end),
785            Self::FundingRates(r) => response::trim_data_to_bounds(&mut r.data, r.start, r.end),
786            Self::Bars(r) => response::trim_data_to_bounds(&mut r.data, r.start, r.end),
787            Self::Instruments(r) => response::trim_data_to_bounds(&mut r.data, r.start, r.end),
788            Self::BookDeltas(r) => response::trim_data_to_bounds(&mut r.data, r.start, r.end),
789            Self::BookDepth(r) => response::trim_data_to_bounds(&mut r.data, r.start, r.end),
790            Self::Data(_)
791            | Self::Instrument(_)
792            | Self::Book(_)
793            | Self::OptionChainReferencePrice(_) => {}
794        }
795    }
796}
797
798pub type Payload = Arc<dyn Any + Send + Sync>;
799
800/// Returns `true` when `params` carries the [`PARAMS_IS_PARENT`] flag set to `true`.
801///
802/// Absent or non-boolean values resolve to `false`, keeping the default subscription
803/// path concrete (exact topic).
804#[must_use]
805pub fn is_parent_subscription(params: Option<&Params>) -> bool {
806    params
807        .and_then(|p| p.get_bool(PARAMS_IS_PARENT))
808        .unwrap_or(false)
809}
nautilus_common/messages/data/mod.rs

nautilus_common/messages/data/
mod.rs
