nautilus_model/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 types and shared representations for the trading domain model.
17//!
18//! [`Data`] provides an owned, heterogeneous representation of built-in data, while [`DataRef`]
19//! provides borrowed access to the same variants. [`DataBatch`] preserves concrete element types
20//! for homogeneous storage and exposes individual items through the borrowed representation.
21
22pub mod bar;
23pub mod batch;
24pub mod bet;
25pub mod black_scholes;
26pub mod close;
27pub mod custom;
28pub mod data_type;
29pub mod delta;
30pub mod deltas;
31pub mod depth;
32pub mod funding;
33pub mod greeks;
34pub mod option_chain;
35pub mod order;
36pub mod prices;
37pub mod quote;
38pub mod registry;
39pub mod status;
40pub mod trade;
41
42#[cfg(any(test, feature = "test-support"))]
43pub mod stubs;
44
45use std::{
46    fmt::{Debug, Display},
47    ops::{Deref, Range},
48    str::FromStr,
49    sync::Arc,
50};
51
52use nautilus_core::UnixNanos;
53use serde::{Deserialize, Serialize};
54
55#[cfg(feature = "defi")]
56use crate::defi::DefiData;
57use crate::{
58    for_each_data_type,
59    identifiers::InstrumentId,
60    instruments::{Instrument, InstrumentAny},
61};
62
63// Re-exports
64#[rustfmt::skip]  // Keep these grouped
65pub use bar::{Bar, BarSpecification, BarType};
66pub use black_scholes::Greeks;
67pub use close::InstrumentClose;
68#[cfg(feature = "python")]
69pub use custom::PythonCustomDataWrapper;
70pub use custom::{
71    CustomData, CustomDataTrait, ensure_custom_data_json_registered, register_custom_data_json,
72};
73#[cfg(feature = "python")]
74pub use custom::{
75    get_python_data_class, reconstruct_python_custom_data, register_python_data_class,
76};
77pub use data_type::DataType;
78pub use delta::OrderBookDelta;
79pub use deltas::OrderBookDeltas;
80pub use depth::{DEPTH_INLINE_LEN, DEPTH10_LEN, OrderBookDepth};
81pub use funding::FundingRateUpdate;
82pub use greeks::{
83    BlackScholesGreeksResult, GreeksData, HasGreeks, OptionGreekValues, PortfolioGreeks,
84    YieldCurveData, black_scholes_greeks, imply_vol_and_greeks, refine_vol_and_greeks,
85};
86pub use option_chain::{OptionChainSlice, OptionGreeks, OptionStrikeData, StrikeRange};
87pub use order::{BookOrder, NULL_ORDER};
88pub use prices::{IndexPriceUpdate, MarkPriceUpdate};
89pub use quote::QuoteTick;
90#[cfg(feature = "arrow")]
91pub use registry::{
92    ArrowDecoder, ArrowEncoder, decode_custom_from_arrow, encode_custom_to_arrow,
93    ensure_arrow_registered, get_arrow_schema, register_arrow, validate_custom_arrow_schema,
94};
95#[cfg(feature = "python")]
96pub use registry::{
97    PyExtractor, ensure_py_extractor_registered, ensure_rust_extractor_factory_registered,
98    ensure_rust_extractor_registered, get_rust_extractor, register_py_extractor,
99    register_rust_extractor, register_rust_extractor_factory, try_extract_from_py,
100};
101pub use registry::{
102    deserialize_custom_from_json, ensure_json_deserializer_registered, register_json_deserializer,
103};
104pub use status::InstrumentStatus;
105pub use trade::TradeTick;
106
107/// Arrow schema-map name for compact string enum columns.
108pub const ARROW_ENUM_DICTIONARY: &str = "Dictionary(Int8, Utf8)";
109/// Arrow schema-map name for UTC nanosecond instants.
110pub const ARROW_TIMESTAMP_NANOSECOND: &str = "Timestamp(Nanosecond, Some(\"UTC\"))";
111
112/// A built-in Nautilus data type.
113///
114/// Not recommended for storing large amounts of data, as the largest variant is significantly
115/// larger (~10x) than the smallest.
116#[derive(Debug)]
117pub enum Data {
118    Custom(CustomData),
119    Instrument(Box<InstrumentAny>),
120    BookDelta(OrderBookDelta),
121    BookDeltas(Box<OrderBookDeltas>),
122    BookDepth(Box<OrderBookDepth>), // This variant is significantly larger
123    Quote(QuoteTick),
124    Trade(TradeTick),
125    Bar(Bar),
126    MarkPrice(MarkPriceUpdate),
127    IndexPrice(IndexPriceUpdate),
128    FundingRate(FundingRateUpdate),
129    OptionGreeks(OptionGreeks),
130    InstrumentStatus(InstrumentStatus),
131    InstrumentClose(InstrumentClose),
132    #[cfg(feature = "defi")]
133    Defi(Box<DefiData>), // This variant is significantly larger
134}
135
136/// Data family selector used by request and catalog APIs.
137///
138/// This is a type-level descriptor, not a decoded data value. [`Data::BookDeltas`] maps to
139/// [`NautilusDataType::OrderBookDelta`] because both share the same storage and request family.
140#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
141pub enum NautilusDataType {
142    /// User-defined data type identified by `type_name`.
143    Custom { type_name: String },
144    /// Instrument definitions.
145    Instrument,
146    /// Order book deltas, single or batched.
147    OrderBookDelta,
148    /// Order book depth snapshots.
149    OrderBookDepth,
150    /// Quote ticks.
151    QuoteTick,
152    /// Trade ticks.
153    TradeTick,
154    /// Aggregated bars.
155    Bar,
156    /// Mark price updates.
157    MarkPriceUpdate,
158    /// Index price updates.
159    IndexPriceUpdate,
160    /// Funding rate updates.
161    FundingRateUpdate,
162    /// Option greeks.
163    OptionGreeks,
164    /// Instrument status updates.
165    InstrumentStatus,
166    /// Instrument closes.
167    InstrumentClose,
168    /// Decentralized finance data.
169    #[cfg(feature = "defi")]
170    Defi,
171}
172
173impl Display for NautilusDataType {
174    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
175        match self {
176            Self::Custom { type_name } => write!(f, "Custom:{type_name}"),
177            Self::Instrument => f.write_str("Instrument"),
178            Self::OrderBookDelta => f.write_str("OrderBookDelta"),
179            Self::OrderBookDepth => f.write_str("OrderBookDepth"),
180            Self::QuoteTick => f.write_str("QuoteTick"),
181            Self::TradeTick => f.write_str("TradeTick"),
182            Self::Bar => f.write_str("Bar"),
183            Self::MarkPriceUpdate => f.write_str("MarkPriceUpdate"),
184            Self::IndexPriceUpdate => f.write_str("IndexPriceUpdate"),
185            Self::FundingRateUpdate => f.write_str("FundingRateUpdate"),
186            Self::OptionGreeks => f.write_str("OptionGreeks"),
187            Self::InstrumentStatus => f.write_str("InstrumentStatus"),
188            Self::InstrumentClose => f.write_str("InstrumentClose"),
189            #[cfg(feature = "defi")]
190            Self::Defi => f.write_str("Defi"),
191        }
192    }
193}
194
195impl FromStr for NautilusDataType {
196    type Err = anyhow::Error;
197
198    fn from_str(s: &str) -> anyhow::Result<Self> {
199        match s {
200            custom if custom.starts_with("Custom:") => Ok(Self::Custom {
201                type_name: custom.trim_start_matches("Custom:").to_string(),
202            }),
203            "Instrument" | "instruments" | "instrument" => Ok(Self::Instrument),
204            "QuoteTick" | "quotes" | "quote" | "quote_tick" => Ok(Self::QuoteTick),
205            "TradeTick" | "trades" | "trade" | "trade_tick" => Ok(Self::TradeTick),
206            "Bar" | "bars" | "bar" => Ok(Self::Bar),
207            "OrderBookDelta" | "OrderBookDeltas" | "order_book_deltas" | "order_book_delta" => {
208                Ok(Self::OrderBookDelta)
209            }
210            "OrderBookDepth" | "order_book_depths" => Ok(Self::OrderBookDepth),
211            "MarkPriceUpdate" | "mark_price_updates" | "mark_prices" | "mark_price_update" => {
212                Ok(Self::MarkPriceUpdate)
213            }
214            "IndexPriceUpdate" | "index_price_updates" | "index_prices" | "index_price_update" => {
215                Ok(Self::IndexPriceUpdate)
216            }
217            "FundingRateUpdate" | "funding_rate_update" | "funding_rates" => {
218                Ok(Self::FundingRateUpdate)
219            }
220            "InstrumentStatus" | "instrument_status" => Ok(Self::InstrumentStatus),
221            "OptionGreeks" | "option_greeks" => Ok(Self::OptionGreeks),
222            "InstrumentClose" | "instrument_closes" | "instrument_close" => {
223                Ok(Self::InstrumentClose)
224            }
225            #[cfg(feature = "defi")]
226            "Defi" => Ok(Self::Defi),
227            _ => anyhow::bail!("Invalid `NautilusDataType`: '{s}'"),
228        }
229    }
230}
231
232impl NautilusDataType {
233    /// Returns the discriminant tag for a [`Data`] value.
234    #[must_use]
235    pub fn from_data(data: &Data) -> Self {
236        match data {
237            Data::Custom(c) => Self::Custom {
238                type_name: c.data_type.type_name().to_string(),
239            },
240            Data::Instrument(_) => Self::Instrument,
241            Data::Quote(_) => Self::QuoteTick,
242            Data::Trade(_) => Self::TradeTick,
243            Data::Bar(_) => Self::Bar,
244            Data::BookDelta(_) | Data::BookDeltas(_) => Self::OrderBookDelta,
245            Data::BookDepth(_) => Self::OrderBookDepth,
246            Data::MarkPrice(_) => Self::MarkPriceUpdate,
247            Data::IndexPrice(_) => Self::IndexPriceUpdate,
248            Data::FundingRate(_) => Self::FundingRateUpdate,
249            Data::InstrumentStatus(_) => Self::InstrumentStatus,
250            Data::OptionGreeks(_) => Self::OptionGreeks,
251            Data::InstrumentClose(_) => Self::InstrumentClose,
252            #[cfg(feature = "defi")]
253            Data::Defi(_) => Self::Defi,
254        }
255    }
256}
257
258/// Borrowed data item used by typed replay paths.
259#[derive(Clone, Copy, Debug)]
260pub enum DataRef<'a> {
261    Instrument(&'a InstrumentAny),
262    BookDelta(&'a OrderBookDelta),
263    BookDeltas(&'a OrderBookDeltas),
264    BookDepth(&'a OrderBookDepth),
265    Quote(&'a QuoteTick),
266    Trade(&'a TradeTick),
267    Bar(&'a Bar),
268    MarkPrice(&'a MarkPriceUpdate),
269    IndexPrice(&'a IndexPriceUpdate),
270    FundingRate(&'a FundingRateUpdate),
271    OptionGreeks(&'a OptionGreeks),
272    InstrumentStatus(&'a InstrumentStatus),
273    InstrumentClose(&'a InstrumentClose),
274    Custom(&'a CustomData),
275    #[cfg(feature = "defi")]
276    Defi(&'a DefiData),
277}
278
279impl<'a> From<&'a Data> for DataRef<'a> {
280    fn from(data: &'a Data) -> Self {
281        match data {
282            Data::Custom(custom) => Self::Custom(custom),
283            Data::Instrument(instrument) => Self::Instrument(instrument),
284            Data::BookDelta(delta) => Self::BookDelta(delta),
285            Data::BookDeltas(deltas) => Self::BookDeltas(deltas),
286            Data::BookDepth(depth) => Self::BookDepth(depth),
287            Data::Quote(quote) => Self::Quote(quote),
288            Data::Trade(trade) => Self::Trade(trade),
289            Data::Bar(bar) => Self::Bar(bar),
290            Data::MarkPrice(mark_price) => Self::MarkPrice(mark_price),
291            Data::IndexPrice(index_price) => Self::IndexPrice(index_price),
292            Data::FundingRate(funding_rate) => Self::FundingRate(funding_rate),
293            Data::OptionGreeks(greeks) => Self::OptionGreeks(greeks),
294            Data::InstrumentStatus(status) => Self::InstrumentStatus(status),
295            Data::InstrumentClose(close) => Self::InstrumentClose(close),
296            #[cfg(feature = "defi")]
297            Data::Defi(defi) => Self::Defi(defi),
298        }
299    }
300}
301
302impl DataRef<'_> {
303    /// Returns the instrument ID for the data.
304    #[must_use]
305    pub fn instrument_id(&self) -> InstrumentId {
306        match self {
307            Self::Custom(custom) => custom
308                .data_type
309                .identifier()
310                .and_then(|s| InstrumentId::from_str(s).ok())
311                .or_else(|| {
312                    custom
313                        .data_type
314                        .metadata()
315                        .and_then(|m| m.get_str("instrument_id"))
316                        .and_then(|s| InstrumentId::from_str(s).ok())
317                })
318                .unwrap_or_else(|| InstrumentId::from("NULL.NULL")),
319            Self::Instrument(instrument) => instrument.id(),
320            Self::BookDelta(delta) => delta.instrument_id,
321            Self::BookDeltas(deltas) => deltas.instrument_id,
322            Self::BookDepth(depth) => depth.instrument_id,
323            Self::Quote(quote) => quote.instrument_id,
324            Self::Trade(trade) => trade.instrument_id,
325            Self::Bar(bar) => bar.bar_type.instrument_id(),
326            Self::MarkPrice(mark_price) => mark_price.instrument_id,
327            Self::IndexPrice(index_price) => index_price.instrument_id,
328            Self::FundingRate(funding_rate) => funding_rate.instrument_id,
329            Self::OptionGreeks(greeks) => greeks.instrument_id,
330            Self::InstrumentStatus(status) => status.instrument_id,
331            Self::InstrumentClose(close) => close.instrument_id,
332            #[cfg(feature = "defi")]
333            Self::Defi(defi) => defi.instrument_id(),
334        }
335    }
336
337    /// Returns whether the data is a type of order book data.
338    #[must_use]
339    pub fn is_order_book_data(&self) -> bool {
340        matches!(
341            self,
342            Self::BookDelta(_) | Self::BookDeltas(_) | Self::BookDepth(_)
343        )
344    }
345
346    /// Materializes this borrowed item as an owned [`Data`] enum for compatibility.
347    #[must_use]
348    pub fn to_owned_data(&self) -> Data {
349        match self {
350            Self::Custom(custom) => Data::Custom((**custom).clone()),
351            Self::Instrument(instrument) => Data::Instrument(Box::new((*instrument).clone())),
352            Self::BookDelta(delta) => Data::BookDelta(**delta),
353            Self::BookDeltas(deltas) => Data::BookDeltas(Box::new((**deltas).clone())),
354            Self::BookDepth(depth) => Data::BookDepth(Box::new((*depth).clone())),
355            Self::Quote(quote) => Data::Quote(**quote),
356            Self::Trade(trade) => Data::Trade(**trade),
357            Self::Bar(bar) => Data::Bar(**bar),
358            Self::MarkPrice(mark_price) => Data::MarkPrice(**mark_price),
359            Self::IndexPrice(index_price) => Data::IndexPrice(**index_price),
360            Self::FundingRate(funding_rate) => Data::FundingRate(**funding_rate),
361            Self::OptionGreeks(greeks) => Data::OptionGreeks(**greeks),
362            Self::InstrumentStatus(status) => Data::InstrumentStatus(**status),
363            Self::InstrumentClose(close) => Data::InstrumentClose(**close),
364            #[cfg(feature = "defi")]
365            Self::Defi(defi) => Data::Defi(Box::new((**defi).clone())),
366        }
367    }
368}
369
370impl HasTsInit for DataRef<'_> {
371    fn ts_init(&self) -> UnixNanos {
372        match self {
373            Self::Custom(custom) => custom.data.ts_init(),
374            Self::Instrument(instrument) => Instrument::ts_init(*instrument),
375            Self::BookDelta(delta) => delta.ts_init,
376            Self::BookDeltas(deltas) => deltas.ts_init,
377            Self::BookDepth(depth) => depth.ts_init,
378            Self::Quote(quote) => quote.ts_init,
379            Self::Trade(trade) => trade.ts_init,
380            Self::Bar(bar) => bar.ts_init,
381            Self::MarkPrice(mark_price) => mark_price.ts_init,
382            Self::IndexPrice(index_price) => index_price.ts_init,
383            Self::FundingRate(funding_rate) => funding_rate.ts_init,
384            Self::OptionGreeks(greeks) => greeks.ts_init,
385            Self::InstrumentStatus(status) => status.ts_init,
386            Self::InstrumentClose(close) => close.ts_init,
387            #[cfg(feature = "defi")]
388            Self::Defi(defi) => defi.ts_init(),
389        }
390    }
391}
392
393/// Range view over a shared typed data batch.
394#[derive(Clone, Debug)]
395#[expect(
396    clippy::rc_buffer,
397    reason = "Backtest batch views share full Vec batches by design"
398)]
399pub struct BatchView<T> {
400    data: Arc<Vec<T>>,
401    range: Range<usize>,
402}
403
404impl<T> BatchView<T> {
405    /// Creates a new [`BatchView`] instance.
406    ///
407    /// # Panics
408    ///
409    /// Panics if `range` is invalid or exceeds `data.len()`.
410    #[must_use]
411    pub fn new(data: Arc<Vec<T>>, range: Range<usize>) -> Self {
412        assert!(range.start <= range.end, "invalid batch view range");
413        assert!(
414            range.end <= data.len(),
415            "batch view range exceeds data length"
416        );
417        Self { data, range }
418    }
419
420    #[must_use]
421    pub fn full(data: Arc<Vec<T>>) -> Self {
422        let len = data.len();
423        Self {
424            data,
425            range: 0..len,
426        }
427    }
428
429    #[must_use]
430    pub fn arc(&self) -> &Arc<Vec<T>> {
431        &self.data
432    }
433
434    #[must_use]
435    pub fn range(&self) -> Range<usize> {
436        self.range.clone()
437    }
438
439    /// Returns a sub-view of this batch view.
440    ///
441    /// # Panics
442    ///
443    /// Panics if `[start, end)` is invalid or exceeds this view length.
444    #[must_use]
445    pub fn slice(&self, start: usize, end: usize) -> Self {
446        assert!(start <= end, "invalid batch slice range");
447        assert!(end <= self.len(), "batch slice range exceeds view length");
448        Self {
449            data: self.data.clone(),
450            range: (self.range.start + start)..(self.range.start + end),
451        }
452    }
453    /// Returns this view for modification, cloning shared storage when needed.
454    pub fn make_mut(&mut self) -> &mut [T]
455    where
456        T: Clone,
457    {
458        &mut Arc::make_mut(&mut self.data)[self.range.clone()]
459    }
460}
461
462impl<T> From<Vec<T>> for BatchView<T> {
463    fn from(data: Vec<T>) -> Self {
464        Self::full(Arc::new(data))
465    }
466}
467
468impl<T> From<Arc<Vec<T>>> for BatchView<T> {
469    fn from(data: Arc<Vec<T>>) -> Self {
470        Self::full(data)
471    }
472}
473
474impl<T> AsRef<[T]> for BatchView<T> {
475    fn as_ref(&self) -> &[T] {
476        self
477    }
478}
479
480impl<T> Deref for BatchView<T> {
481    type Target = [T];
482
483    fn deref(&self) -> &Self::Target {
484        &self.data[self.range.clone()]
485    }
486}
487
488/// Shared typed batch used by replay and catalog fan-out paths.
489#[derive(Clone, Debug)]
490pub enum DataBatch {
491    Instrument(BatchView<InstrumentAny>),
492    BookDelta(BatchView<OrderBookDelta>),
493    BookDeltas(BatchView<OrderBookDeltas>),
494    BookDepth(BatchView<OrderBookDepth>),
495    Quote(BatchView<QuoteTick>),
496    Trade(BatchView<TradeTick>),
497    Bar(BatchView<Bar>),
498    MarkPrice(BatchView<MarkPriceUpdate>),
499    IndexPrice(BatchView<IndexPriceUpdate>),
500    FundingRate(BatchView<FundingRateUpdate>),
501    OptionGreeks(BatchView<OptionGreeks>),
502    InstrumentStatus(BatchView<InstrumentStatus>),
503    InstrumentClose(BatchView<InstrumentClose>),
504    Custom(BatchView<CustomData>),
505    #[cfg(feature = "defi")]
506    Defi(BatchView<DefiData>),
507}
508
509macro_rules! data_batch_from_data_vec {
510    (
511        ($data_type:ident, $input:ident);
512        $(($variant:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?
513    ) => {
514        match $data_type {
515            NautilusDataType::Custom { .. } => {
516                let expected_len = $input.len();
517                let custom = $input
518                    .into_iter()
519                    .filter_map(|item| match item {
520                        Data::Custom(custom) => Some(custom),
521                        _ => None,
522                    })
523                    .collect::<Vec<_>>();
524                anyhow::ensure!(
525                    custom.len() == expected_len,
526                    "catalog query for {} returned rows with another data type",
527                    $data_type,
528                );
529                Ok(Self::Custom(custom.into()))
530            }
531            $(
532                NautilusDataType::$variant => Ok(Self::$batch(
533                    to_variant_for_batch::<$type>($data_type, $input)?.into(),
534                )),
535            )+
536            #[cfg(feature = "defi")]
537            NautilusDataType::Defi => {
538                let expected_len = $input.len();
539                let defi = $input
540                    .into_iter()
541                    .filter_map(|item| match item {
542                        Data::Defi(defi) => Some(*defi),
543                        _ => None,
544                    })
545                    .collect::<Vec<_>>();
546                anyhow::ensure!(
547                    defi.len() == expected_len,
548                    "catalog query for {} returned rows with another data type",
549                    $data_type,
550                );
551                Ok(Self::Defi(defi.into()))
552            }
553        }
554    };
555}
556
557impl DataBatch {
558    /// Converts owned compatibility rows into a typed batch for `data_type`.
559    ///
560    /// This is a transition path for backends that still decode through [`Data`] but should not
561    /// expose legacy batches to typed replay.
562    ///
563    /// # Errors
564    ///
565    /// Returns an error if `data_type` is not replayable as a typed batch.
566    pub fn from_data_vec_for_type(
567        data_type: &NautilusDataType,
568        data: Vec<Data>,
569    ) -> anyhow::Result<Self> {
570        if matches!(data_type, NautilusDataType::OrderBookDelta)
571            && matches!(data.first(), Some(Data::BookDeltas(_)))
572        {
573            return Ok(Self::BookDeltas(
574                to_variant_for_batch::<OrderBookDeltas>(data_type, data)?.into(),
575            ));
576        }
577
578        crate::for_each_data_type!(data_batch_from_data_vec, data_type, data)
579    }
580
581    /// Groups mixed compatibility rows into typed batches.
582    ///
583    /// [`Data::BookDeltas`] values retain their event-batch boundary in a separate typed batch.
584    ///
585    /// # Errors
586    ///
587    /// Returns an error if a row cannot be represented by a typed batch.
588    pub fn from_data_vec_grouped(data: &[Data]) -> anyhow::Result<Vec<Self>> {
589        let mut groups = Vec::<(NautilusDataType, Vec<Data>)>::new();
590
591        for item in data.iter().cloned() {
592            let data_type = NautilusDataType::from_data(&item);
593            let is_deltas = matches!(item, Data::BookDeltas(_));
594
595            if let Some((_, group)) = groups.iter_mut().find(|(group_type, group)| {
596                group_type == &data_type
597                    && group
598                        .first()
599                        .is_some_and(|item| matches!(item, Data::BookDeltas(_)) == is_deltas)
600            }) {
601                group.push(item);
602            } else {
603                groups.push((data_type, vec![item]));
604            }
605        }
606
607        groups
608            .into_iter()
609            .map(|(data_type, data)| Self::from_data_vec_for_type(&data_type, data))
610            .collect()
611    }
612
613    #[must_use]
614    pub fn len(&self) -> usize {
615        match self {
616            Self::Custom(data) => data.len(),
617            Self::Instrument(data) => data.len(),
618            Self::BookDelta(data) => data.len(),
619            Self::BookDeltas(data) => data.len(),
620            Self::BookDepth(data) => data.len(),
621            Self::Quote(data) => data.len(),
622            Self::Trade(data) => data.len(),
623            Self::Bar(data) => data.len(),
624            Self::MarkPrice(data) => data.len(),
625            Self::IndexPrice(data) => data.len(),
626            Self::FundingRate(data) => data.len(),
627            Self::OptionGreeks(data) => data.len(),
628            Self::InstrumentStatus(data) => data.len(),
629            Self::InstrumentClose(data) => data.len(),
630            #[cfg(feature = "defi")]
631            Self::Defi(data) => data.len(),
632        }
633    }
634
635    #[must_use]
636    pub fn is_empty(&self) -> bool {
637        self.len() == 0
638    }
639
640    /// Returns the storage-family name of this batch's data type.
641    #[must_use]
642    pub fn data_type_name(&self) -> &'static str {
643        match self {
644            Self::Custom(_) => "custom",
645            Self::Instrument(_) => "instruments",
646            Self::BookDelta(_) => "order_book_deltas",
647            Self::BookDeltas(_) => "order_book_deltas_batches",
648            Self::BookDepth(_) => "order_book_depths",
649            Self::Quote(_) => "quotes",
650            Self::Trade(_) => "trades",
651            Self::Bar(_) => "bars",
652            Self::MarkPrice(_) => "mark_prices",
653            Self::IndexPrice(_) => "index_prices",
654            Self::FundingRate(_) => "funding_rates",
655            Self::OptionGreeks(_) => "option_greeks",
656            Self::InstrumentStatus(_) => "instrument_status",
657            Self::InstrumentClose(_) => "instrument_closes",
658            #[cfg(feature = "defi")]
659            Self::Defi(_) => "defi",
660        }
661    }
662
663    #[must_use]
664    pub fn is_monotonically_increasing_by_init(&self) -> bool {
665        (0..self.len())
666            .filter_map(|index| self.get(index))
667            .map(|data| data.ts_init())
668            .is_sorted()
669    }
670
671    #[must_use]
672    pub fn get(&self, index: usize) -> Option<DataRef<'_>> {
673        match self {
674            Self::Custom(data) => data.get(index).map(DataRef::Custom),
675            Self::Instrument(data) => data.get(index).map(DataRef::Instrument),
676            Self::BookDelta(data) => data.get(index).map(DataRef::BookDelta),
677            Self::BookDeltas(data) => data.get(index).map(DataRef::BookDeltas),
678            Self::BookDepth(data) => data.get(index).map(DataRef::BookDepth),
679            Self::Quote(data) => data.get(index).map(DataRef::Quote),
680            Self::Trade(data) => data.get(index).map(DataRef::Trade),
681            Self::Bar(data) => data.get(index).map(DataRef::Bar),
682            Self::MarkPrice(data) => data.get(index).map(DataRef::MarkPrice),
683            Self::IndexPrice(data) => data.get(index).map(DataRef::IndexPrice),
684            Self::FundingRate(data) => data.get(index).map(DataRef::FundingRate),
685            Self::OptionGreeks(data) => data.get(index).map(DataRef::OptionGreeks),
686            Self::InstrumentStatus(data) => data.get(index).map(DataRef::InstrumentStatus),
687            Self::InstrumentClose(data) => data.get(index).map(DataRef::InstrumentClose),
688            #[cfg(feature = "defi")]
689            Self::Defi(data) => data.get(index).map(DataRef::Defi),
690        }
691    }
692
693    #[must_use]
694    pub fn aligned_chunk(&self, start: usize, chunk_size: Option<usize>) -> Option<(Self, usize)> {
695        let len = self.len();
696        if start >= len {
697            return None;
698        }
699
700        let mut end = match chunk_size {
701            Some(size) => len.min(start + size.max(1)),
702            None => len,
703        };
704
705        if end < len
706            && let Some(boundary_ts) = self.get(end - 1).map(|data| data.ts_init())
707        {
708            while end < len
709                && self
710                    .get(end)
711                    .is_some_and(|data| data.ts_init() == boundary_ts)
712            {
713                end += 1;
714            }
715        }
716
717        Some((self.slice(start, end), end))
718    }
719
720    #[must_use]
721    pub fn slice(&self, start: usize, end: usize) -> Self {
722        if start == 0 && end == self.len() {
723            return self.clone();
724        }
725
726        match self {
727            Self::Custom(data) => Self::Custom(data.slice(start, end)),
728            Self::Instrument(data) => Self::Instrument(data.slice(start, end)),
729            Self::BookDelta(data) => Self::BookDelta(data.slice(start, end)),
730            Self::BookDeltas(data) => Self::BookDeltas(data.slice(start, end)),
731            Self::BookDepth(data) => Self::BookDepth(data.slice(start, end)),
732            Self::Quote(data) => Self::Quote(data.slice(start, end)),
733            Self::Trade(data) => Self::Trade(data.slice(start, end)),
734            Self::Bar(data) => Self::Bar(data.slice(start, end)),
735            Self::MarkPrice(data) => Self::MarkPrice(data.slice(start, end)),
736            Self::IndexPrice(data) => Self::IndexPrice(data.slice(start, end)),
737            Self::FundingRate(data) => Self::FundingRate(data.slice(start, end)),
738            Self::OptionGreeks(data) => Self::OptionGreeks(data.slice(start, end)),
739            Self::InstrumentStatus(data) => Self::InstrumentStatus(data.slice(start, end)),
740            Self::InstrumentClose(data) => Self::InstrumentClose(data.slice(start, end)),
741            #[cfg(feature = "defi")]
742            Self::Defi(data) => Self::Defi(data.slice(start, end)),
743        }
744    }
745
746    #[must_use]
747    pub fn to_data_vec_for_compat(&self) -> Vec<Data> {
748        (0..self.len())
749            .filter_map(|index| self.get(index).map(|item| item.to_owned_data()))
750            .collect()
751    }
752}
753
754/// Typed values that wrap into their [`DataBatch`] variant.
755pub trait IntoDataBatch: Sized {
756    /// Wraps owned typed values in their [`DataBatch`] variant.
757    #[must_use]
758    fn into_batch(data: Vec<Self>) -> DataBatch;
759}
760
761impl<T: IntoDataBatch> From<Vec<T>> for DataBatch {
762    fn from(data: Vec<T>) -> Self {
763        T::into_batch(data)
764    }
765}
766
767macro_rules! impl_into_data_batch {
768    ($(($variant:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?) => {
769        $(
770            impl IntoDataBatch for $type {
771                fn into_batch(data: Vec<Self>) -> DataBatch {
772                    DataBatch::$batch(data.into())
773                }
774            }
775        )+
776    };
777}
778
779for_each_data_type!(impl_into_data_batch);
780
781impl IntoDataBatch for OrderBookDeltas {
782    fn into_batch(data: Vec<Self>) -> DataBatch {
783        DataBatch::BookDeltas(data.into())
784    }
785}
786
787impl IntoDataBatch for CustomData {
788    fn into_batch(data: Vec<Self>) -> DataBatch {
789        DataBatch::Custom(data.into())
790    }
791}
792
793#[cfg(feature = "defi")]
794impl IntoDataBatch for DefiData {
795    fn into_batch(data: Vec<Self>) -> DataBatch {
796        DataBatch::Defi(data.into())
797    }
798}
799
800/// Typed values that unwrap from their [`DataBatch`] variant.
801pub trait FromDataBatch: Sized {
802    /// Unwraps a typed batch into owned values.
803    ///
804    /// # Errors
805    ///
806    /// Returns an error if `batch` holds another variant.
807    fn from_batch(batch: DataBatch) -> anyhow::Result<Vec<Self>>;
808}
809
810macro_rules! impl_from_data_batch {
811    ($(($variant:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?) => {
812        $(
813            impl FromDataBatch for $type {
814                fn from_batch(batch: DataBatch) -> anyhow::Result<Vec<Self>> {
815                    match batch {
816                        DataBatch::$batch(data) => Ok(data.as_ref().to_vec()),
817                        _ => anyhow::bail!(
818                            "expected a {} batch, found another data batch variant",
819                            stringify!($batch),
820                        ),
821                    }
822                }
823            }
824        )+
825    };
826}
827
828for_each_data_type!(impl_from_data_batch);
829
830impl FromDataBatch for CustomData {
831    fn from_batch(batch: DataBatch) -> anyhow::Result<Vec<Self>> {
832        match batch {
833            DataBatch::Custom(data) => Ok(data.as_ref().to_vec()),
834            _ => anyhow::bail!("expected a Custom batch, found another data batch variant"),
835        }
836    }
837}
838
839#[cfg(feature = "defi")]
840impl FromDataBatch for DefiData {
841    fn from_batch(batch: DataBatch) -> anyhow::Result<Vec<Self>> {
842        match batch {
843            DataBatch::Defi(data) => Ok(data.as_ref().to_vec()),
844            _ => anyhow::bail!("expected a Defi batch, found another data batch variant"),
845        }
846    }
847}
848
849/// Catalog record selector used to query Arrow-backed persisted records.
850#[derive(
851    Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize, strum::Display, strum::EnumIter,
852)]
853pub enum NautilusRecordType {
854    AccountState,
855    OrderInitialized,
856    OrderDenied,
857    OrderEmulated,
858    OrderSubmitted,
859    OrderAccepted,
860    OrderRejected,
861    OrderPendingCancel,
862    OrderCanceled,
863    OrderCancelRejected,
864    OrderExpired,
865    OrderTriggered,
866    OrderPendingUpdate,
867    OrderReleased,
868    OrderModifyRejected,
869    OrderUpdated,
870    OrderFilled,
871    OrderFillVoided,
872    PositionOpened,
873    PositionChanged,
874    PositionClosed,
875    PositionAdjusted,
876    OrderSnapshot,
877    PositionSnapshot,
878    OrderStatusReport,
879    FillReport,
880    PositionStatusReport,
881    ExecutionMassStatus,
882    #[cfg(feature = "defi")]
883    Defi,
884}
885
886impl FromStr for NautilusRecordType {
887    type Err = anyhow::Error;
888
889    fn from_str(s: &str) -> anyhow::Result<Self> {
890        match s {
891            "AccountState" | "account_state" => Ok(Self::AccountState),
892            "OrderInitialized" | "order_initialized" => Ok(Self::OrderInitialized),
893            "OrderDenied" | "order_denied" => Ok(Self::OrderDenied),
894            "OrderEmulated" | "order_emulated" => Ok(Self::OrderEmulated),
895            "OrderSubmitted" | "order_submitted" => Ok(Self::OrderSubmitted),
896            "OrderAccepted" | "order_accepted" => Ok(Self::OrderAccepted),
897            "OrderRejected" | "order_rejected" => Ok(Self::OrderRejected),
898            "OrderPendingCancel" | "order_pending_cancel" => Ok(Self::OrderPendingCancel),
899            "OrderCanceled" | "order_canceled" => Ok(Self::OrderCanceled),
900            "OrderCancelRejected" | "order_cancel_rejected" => Ok(Self::OrderCancelRejected),
901            "OrderExpired" | "order_expired" => Ok(Self::OrderExpired),
902            "OrderTriggered" | "order_triggered" => Ok(Self::OrderTriggered),
903            "OrderPendingUpdate" | "order_pending_update" => Ok(Self::OrderPendingUpdate),
904            "OrderReleased" | "order_released" => Ok(Self::OrderReleased),
905            "OrderModifyRejected" | "order_modify_rejected" => Ok(Self::OrderModifyRejected),
906            "OrderUpdated" | "order_updated" => Ok(Self::OrderUpdated),
907            "OrderFilled" | "order_filled" => Ok(Self::OrderFilled),
908            "OrderFillVoided" | "order_fill_voided" => Ok(Self::OrderFillVoided),
909            "PositionOpened" | "position_opened" => Ok(Self::PositionOpened),
910            "PositionChanged" | "position_changed" => Ok(Self::PositionChanged),
911            "PositionClosed" | "position_closed" => Ok(Self::PositionClosed),
912            "PositionAdjusted" | "position_adjusted" => Ok(Self::PositionAdjusted),
913            "OrderSnapshot" | "order_snapshot" => Ok(Self::OrderSnapshot),
914            "PositionSnapshot" | "position_snapshot" => Ok(Self::PositionSnapshot),
915            "OrderStatusReport" | "order_status_report" => Ok(Self::OrderStatusReport),
916            "FillReport" | "fill_report" => Ok(Self::FillReport),
917            "PositionStatusReport" | "position_status_report" => Ok(Self::PositionStatusReport),
918            "ExecutionMassStatus" | "execution_mass_status" => Ok(Self::ExecutionMassStatus),
919            "custom" | "CustomData" => {
920                anyhow::bail!("custom data queries require NautilusDataType::Custom")
921            }
922            #[cfg(feature = "defi")]
923            "Defi" | "defi" => Ok(Self::Defi),
924            _ => anyhow::bail!("Invalid `NautilusRecordType`: '{s}'"),
925        }
926    }
927}
928
929impl<'de> Deserialize<'de> for Data {
930    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
931    where
932        D: serde::Deserializer<'de>,
933    {
934        use serde::de::Error;
935        let value = serde_json::Value::deserialize(deserializer)?;
936        let type_name = value
937            .get("type")
938            .and_then(|v| v.as_str())
939            .ok_or_else(|| D::Error::custom("Missing 'type' field in Data"))?;
940
941        match type_name {
942            "Instrument" => Ok(Self::Instrument(Box::new(
943                serde_json::from_value(
944                    value
945                        .get("data")
946                        .cloned()
947                        .ok_or_else(|| D::Error::custom("Missing 'data' field for Instrument"))?,
948                )
949                .map_err(D::Error::custom)?,
950            ))),
951            "OrderBookDelta" => Ok(Self::BookDelta(
952                serde_json::from_value(value).map_err(D::Error::custom)?,
953            )),
954            "OrderBookDeltas" => Ok(Self::BookDeltas(
955                serde_json::from_value(value).map_err(D::Error::custom)?,
956            )),
957            "OrderBookDepth" => Ok(Self::BookDepth(
958                serde_json::from_value(value).map_err(D::Error::custom)?,
959            )),
960            "QuoteTick" => Ok(Self::Quote(
961                serde_json::from_value(value).map_err(D::Error::custom)?,
962            )),
963            "TradeTick" => Ok(Self::Trade(
964                serde_json::from_value(value).map_err(D::Error::custom)?,
965            )),
966            "Bar" => Ok(Self::Bar(
967                serde_json::from_value(value).map_err(D::Error::custom)?,
968            )),
969            "MarkPriceUpdate" => Ok(Self::MarkPrice(
970                serde_json::from_value(value).map_err(D::Error::custom)?,
971            )),
972            "IndexPriceUpdate" => Ok(Self::IndexPrice(
973                serde_json::from_value(value).map_err(D::Error::custom)?,
974            )),
975            "FundingRateUpdate" => Ok(Self::FundingRate(
976                serde_json::from_value(value).map_err(D::Error::custom)?,
977            )),
978            "OptionGreeks" => Ok(Self::OptionGreeks(
979                serde_json::from_value(value).map_err(D::Error::custom)?,
980            )),
981            "InstrumentStatus" => Ok(Self::InstrumentStatus(
982                serde_json::from_value(value).map_err(D::Error::custom)?,
983            )),
984            "InstrumentClose" => Ok(Self::InstrumentClose(
985                serde_json::from_value(value).map_err(D::Error::custom)?,
986            )),
987            _ => {
988                if let Some(data) =
989                    deserialize_custom_from_json(type_name, &value).map_err(D::Error::custom)?
990                {
991                    Ok(data)
992                } else {
993                    Err(D::Error::custom(format!("Unknown Data type: {type_name}")))
994                }
995            }
996        }
997    }
998}
999
1000impl Clone for Data {
1001    fn clone(&self) -> Self {
1002        match self {
1003            Self::Custom(x) => Self::Custom(x.clone()),
1004            Self::Instrument(x) => Self::Instrument(x.clone()),
1005            Self::BookDelta(x) => Self::BookDelta(*x),
1006            Self::BookDeltas(x) => Self::BookDeltas(x.clone()),
1007            Self::BookDepth(x) => Self::BookDepth(x.clone()),
1008            Self::Quote(x) => Self::Quote(*x),
1009            Self::Trade(x) => Self::Trade(*x),
1010            Self::Bar(x) => Self::Bar(*x),
1011            Self::MarkPrice(x) => Self::MarkPrice(*x),
1012            Self::IndexPrice(x) => Self::IndexPrice(*x),
1013            Self::FundingRate(x) => Self::FundingRate(*x),
1014            Self::OptionGreeks(x) => Self::OptionGreeks(*x),
1015            Self::InstrumentStatus(x) => Self::InstrumentStatus(*x),
1016            Self::InstrumentClose(x) => Self::InstrumentClose(*x),
1017            #[cfg(feature = "defi")]
1018            Self::Defi(x) => Self::Defi(x.clone()),
1019        }
1020    }
1021}
1022
1023impl PartialEq for Data {
1024    fn eq(&self, other: &Self) -> bool {
1025        match (self, other) {
1026            (Self::Custom(a), Self::Custom(b)) => a == b,
1027            (Self::Instrument(a), Self::Instrument(b)) => a == b,
1028            (Self::BookDelta(a), Self::BookDelta(b)) => a == b,
1029            (Self::BookDeltas(a), Self::BookDeltas(b)) => a == b,
1030            (Self::BookDepth(a), Self::BookDepth(b)) => a == b,
1031            (Self::Quote(a), Self::Quote(b)) => a == b,
1032            (Self::Trade(a), Self::Trade(b)) => a == b,
1033            (Self::Bar(a), Self::Bar(b)) => a == b,
1034            (Self::MarkPrice(a), Self::MarkPrice(b)) => a == b,
1035            (Self::IndexPrice(a), Self::IndexPrice(b)) => a == b,
1036            (Self::FundingRate(a), Self::FundingRate(b)) => a == b,
1037            (Self::OptionGreeks(a), Self::OptionGreeks(b)) => a == b,
1038            (Self::InstrumentStatus(a), Self::InstrumentStatus(b)) => a == b,
1039            (Self::InstrumentClose(a), Self::InstrumentClose(b)) => a == b,
1040            #[cfg(feature = "defi")]
1041            (Self::Defi(a), Self::Defi(b)) => a == b,
1042            _ => false,
1043        }
1044    }
1045}
1046
1047impl Serialize for Data {
1048    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1049    where
1050        S: serde::Serializer,
1051    {
1052        match self {
1053            Self::Custom(x) => x.serialize(serializer),
1054            Self::Instrument(instrument) => serde_json::json!({
1055                "type": "Instrument",
1056                "data": instrument,
1057            })
1058            .serialize(serializer),
1059            Self::BookDelta(x) => x.serialize(serializer),
1060            Self::BookDeltas(x) => x.serialize(serializer),
1061            Self::BookDepth(x) => x.serialize(serializer),
1062            Self::Quote(x) => x.serialize(serializer),
1063            Self::Trade(x) => x.serialize(serializer),
1064            Self::Bar(x) => x.serialize(serializer),
1065            Self::MarkPrice(x) => x.serialize(serializer),
1066            Self::IndexPrice(x) => x.serialize(serializer),
1067            Self::FundingRate(x) => x.serialize(serializer),
1068            Self::OptionGreeks(x) => x.serialize(serializer),
1069            Self::InstrumentStatus(x) => x.serialize(serializer),
1070            Self::InstrumentClose(x) => x.serialize(serializer),
1071            #[cfg(feature = "defi")]
1072            Self::Defi(_) => Err(serde::ser::Error::custom(
1073                "Data::Defi serialization is not supported",
1074            )),
1075        }
1076    }
1077}
1078
1079macro_rules! impl_data_conversions {
1080    ($variant:ident, $type:ty) => {
1081        impl TryFrom<Data> for $type {
1082            type Error = ();
1083
1084            fn try_from(value: Data) -> Result<Self, Self::Error> {
1085                match value {
1086                    Data::$variant(x) => Ok(x),
1087                    _ => Err(()),
1088                }
1089            }
1090        }
1091
1092        impl From<$type> for Data {
1093            fn from(value: $type) -> Self {
1094                Self::$variant(value)
1095            }
1096        }
1097    };
1098}
1099
1100impl TryFrom<Data> for OrderBookDepth {
1101    type Error = ();
1102
1103    fn try_from(value: Data) -> Result<Self, Self::Error> {
1104        match value {
1105            Data::BookDepth(x) => Ok(*x),
1106            _ => Err(()),
1107        }
1108    }
1109}
1110
1111impl TryFrom<Data> for InstrumentAny {
1112    type Error = ();
1113
1114    fn try_from(value: Data) -> Result<Self, Self::Error> {
1115        match value {
1116            Data::Instrument(instrument) => Ok(*instrument),
1117            _ => Err(()),
1118        }
1119    }
1120}
1121
1122impl TryFrom<Data> for OrderBookDeltas {
1123    type Error = ();
1124
1125    fn try_from(value: Data) -> Result<Self, Self::Error> {
1126        match value {
1127            Data::BookDeltas(deltas) => Ok(*deltas),
1128            _ => Err(()),
1129        }
1130    }
1131}
1132
1133impl_data_conversions!(Quote, QuoteTick);
1134impl_data_conversions!(BookDelta, OrderBookDelta);
1135impl_data_conversions!(Trade, TradeTick);
1136impl_data_conversions!(Bar, Bar);
1137impl_data_conversions!(MarkPrice, MarkPriceUpdate);
1138impl_data_conversions!(IndexPrice, IndexPriceUpdate);
1139impl_data_conversions!(FundingRate, FundingRateUpdate);
1140impl_data_conversions!(OptionGreeks, OptionGreeks);
1141impl_data_conversions!(InstrumentStatus, InstrumentStatus);
1142impl_data_conversions!(InstrumentClose, InstrumentClose);
1143
1144/// Converts a vector of `Data` items to a specific variant type.
1145///
1146/// Filters and converts the data vector, keeping only items that can be
1147/// successfully converted to the target type `T`.
1148#[must_use]
1149pub fn to_variant<T: TryFrom<Data>>(data: Vec<Data>) -> Vec<T> {
1150    data.into_iter()
1151        .filter_map(|d| T::try_from(d).ok())
1152        .collect()
1153}
1154
1155fn to_variant_for_batch<T: TryFrom<Data>>(
1156    data_type: &NautilusDataType,
1157    data: Vec<Data>,
1158) -> anyhow::Result<Vec<T>> {
1159    let expected_len = data.len();
1160    let converted = to_variant(data);
1161    anyhow::ensure!(
1162        converted.len() == expected_len,
1163        "catalog query for {data_type} returned rows with another data type",
1164    );
1165    Ok(converted)
1166}
1167
1168impl Data {
1169    /// Returns the instrument ID for the data.
1170    #[must_use]
1171    pub fn instrument_id(&self) -> InstrumentId {
1172        DataRef::from(self).instrument_id()
1173    }
1174
1175    /// Returns whether the data is a type of order book data.
1176    #[must_use]
1177    pub fn is_order_book_data(&self) -> bool {
1178        DataRef::from(self).is_order_book_data()
1179    }
1180}
1181
1182impl From<InstrumentAny> for Data {
1183    fn from(value: InstrumentAny) -> Self {
1184        Self::Instrument(Box::new(value))
1185    }
1186}
1187
1188/// Marker trait for types that carry a creation timestamp.
1189///
1190/// `ts_init` is the moment (UNIX nanoseconds) when this value was first generated or
1191/// ingested by Nautilus. It can be used for sequencing, latency measurements,
1192/// or monitoring data-pipeline delays.
1193pub trait HasTsInit {
1194    /// Returns the UNIX timestamp (nanoseconds) when the instance was created.
1195    fn ts_init(&self) -> UnixNanos;
1196}
1197
1198impl HasTsInit for Data {
1199    fn ts_init(&self) -> UnixNanos {
1200        DataRef::from(self).ts_init()
1201    }
1202}
1203
1204/// Checks if the data slice is monotonically increasing by initialization timestamp.
1205///
1206/// Returns `true` if each element's `ts_init` is less than or equal to the next element's `ts_init`.
1207pub fn is_monotonically_increasing_by_init<T: HasTsInit>(data: &[T]) -> bool {
1208    data.array_windows()
1209        .all(|[a, b]| a.ts_init() <= b.ts_init())
1210}
1211
1212impl From<OrderBookDeltas> for Data {
1213    fn from(value: OrderBookDeltas) -> Self {
1214        Self::BookDeltas(Box::new(value))
1215    }
1216}
1217
1218impl From<OrderBookDepth> for Data {
1219    fn from(value: OrderBookDepth) -> Self {
1220        Self::BookDepth(Box::new(value))
1221    }
1222}
1223
1224#[cfg(feature = "defi")]
1225impl From<DefiData> for Data {
1226    fn from(value: DefiData) -> Self {
1227        Self::Defi(Box::new(value))
1228    }
1229}
1230
1231/// Invokes a macro with every built-in typed data family.
1232///
1233/// Each tuple contains the canonical family name, concrete type, `Data` variant,
1234/// `DataBatch` variant, and catalog path prefix. The first four fields use the same name whenever
1235/// the family has one concrete representation. Custom and DeFi data have no single built-in
1236/// concrete type; `Deltas` and `Data` are aggregate compatibility batches. Consumers handle
1237/// those variants explicitly.
1238#[macro_export]
1239macro_rules! for_each_data_type {
1240    ($macro:ident) => {
1241        $macro! {
1242            (Instrument, InstrumentAny, Instrument, Instrument, "instruments"),
1243            (QuoteTick, QuoteTick, Quote, Quote, "quotes"),
1244            (TradeTick, TradeTick, Trade, Trade, "trades"),
1245            (Bar, Bar, Bar, Bar, "bars"),
1246            (OrderBookDelta, OrderBookDelta, BookDelta, BookDelta, "order_book_deltas"),
1247            (OrderBookDepth, OrderBookDepth, BookDepth, BookDepth, "order_book_depths"),
1248            (MarkPriceUpdate, MarkPriceUpdate, MarkPrice, MarkPrice, "mark_prices"),
1249            (IndexPriceUpdate, IndexPriceUpdate, IndexPrice, IndexPrice, "index_prices"),
1250            (FundingRateUpdate, FundingRateUpdate, FundingRate, FundingRate, "funding_rates"),
1251            (InstrumentStatus, InstrumentStatus, InstrumentStatus, InstrumentStatus, "instrument_status"),
1252            (OptionGreeks, OptionGreeks, OptionGreeks, OptionGreeks, "option_greeks"),
1253            (InstrumentClose, InstrumentClose, InstrumentClose, InstrumentClose, "instrument_closes"),
1254        }
1255    };
1256    ($macro:ident, $($args:tt)*) => {
1257        $macro! {
1258            ($($args)*);
1259            (Instrument, InstrumentAny, Instrument, Instrument, "instruments"),
1260            (QuoteTick, QuoteTick, Quote, Quote, "quotes"),
1261            (TradeTick, TradeTick, Trade, Trade, "trades"),
1262            (Bar, Bar, Bar, Bar, "bars"),
1263            (OrderBookDelta, OrderBookDelta, BookDelta, BookDelta, "order_book_deltas"),
1264            (OrderBookDepth, OrderBookDepth, BookDepth, BookDepth, "order_book_depths"),
1265            (MarkPriceUpdate, MarkPriceUpdate, MarkPrice, MarkPrice, "mark_prices"),
1266            (IndexPriceUpdate, IndexPriceUpdate, IndexPrice, IndexPrice, "index_prices"),
1267            (FundingRateUpdate, FundingRateUpdate, FundingRate, FundingRate, "funding_rates"),
1268            (InstrumentStatus, InstrumentStatus, InstrumentStatus, InstrumentStatus, "instrument_status"),
1269            (OptionGreeks, OptionGreeks, OptionGreeks, OptionGreeks, "option_greeks"),
1270            (InstrumentClose, InstrumentClose, InstrumentClose, InstrumentClose, "instrument_closes"),
1271        }
1272    };
1273}
1274
1275#[cfg(test)]
1276mod tests {
1277    use std::sync::Arc;
1278
1279    use rstest::*;
1280
1281    use super::*;
1282    use crate::instruments::stubs::crypto_perpetual_ethusdt;
1283
1284    #[rstest]
1285    fn test_depth_family_matches_both_macro_forms() {
1286        macro_rules! family_rows {
1287            (($marker:ident); $($rows:tt)*) => { family_rows!($($rows)*) };
1288            ($(($family:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?) => {
1289                vec![$((stringify!($family), stringify!($type), stringify!($data), stringify!($batch), $prefix)),+]
1290            };
1291        }
1292        let rows = for_each_data_type!(family_rows);
1293        let rows_with_args = for_each_data_type!(family_rows, context);
1294        let depth = rows
1295            .iter()
1296            .find(|row| row.4 == "order_book_depths")
1297            .copied()
1298            .unwrap();
1299        assert_eq!(rows, rows_with_args);
1300        assert_eq!(
1301            depth,
1302            (
1303                "OrderBookDepth",
1304                "OrderBookDepth",
1305                "BookDepth",
1306                "BookDepth",
1307                "order_book_depths"
1308            )
1309        );
1310    }
1311
1312    #[rstest]
1313    fn data_instrument_json_roundtrips() {
1314        let data = Data::from(InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt()));
1315
1316        let encoded = serde_json::to_string(&data).unwrap();
1317        let decoded = serde_json::from_str::<Data>(&encoded).unwrap();
1318
1319        assert_eq!(decoded, data);
1320    }
1321
1322    #[rstest]
1323    #[case(NautilusDataType::Instrument, "Instrument")]
1324    #[case(NautilusDataType::QuoteTick, "QuoteTick")]
1325    #[case(NautilusDataType::OrderBookDelta, "OrderBookDelta")]
1326    #[case(
1327        NautilusDataType::Custom {
1328            type_name: "Example".to_string()
1329        },
1330        "Custom:Example"
1331    )]
1332    fn nautilus_data_type_display_from_str_roundtrips(
1333        #[case] nautilus_data_type: NautilusDataType,
1334        #[case] expected: &str,
1335    ) {
1336        assert_eq!(nautilus_data_type.to_string(), expected);
1337        assert_eq!(
1338            expected.parse::<NautilusDataType>().unwrap(),
1339            nautilus_data_type
1340        );
1341    }
1342
1343    #[rstest]
1344    #[case("instruments", NautilusDataType::Instrument)]
1345    #[case("quotes", NautilusDataType::QuoteTick)]
1346    #[case("order_book_deltas", NautilusDataType::OrderBookDelta)]
1347    #[case("instrument_closes", NautilusDataType::InstrumentClose)]
1348    // Plural class names, so callers can pass a Nautilus type's own name.
1349    #[case("OrderBookDeltas", NautilusDataType::OrderBookDelta)]
1350    #[case("OrderBookDepth", NautilusDataType::OrderBookDepth)]
1351    fn nautilus_data_type_storage_names_parse(
1352        #[case] value: &str,
1353        #[case] expected: NautilusDataType,
1354    ) {
1355        assert_eq!(value.parse::<NautilusDataType>().unwrap(), expected);
1356    }
1357
1358    #[rstest]
1359    #[case("OrderBookDepth10")]
1360    #[case("order_book_depth10")]
1361    fn nautilus_data_type_rejects_former_depth10_spellings(#[case] value: &str) {
1362        assert!(value.parse::<NautilusDataType>().is_err());
1363    }
1364
1365    #[rstest]
1366    #[case("OrderBook")]
1367    #[case("order_book")]
1368    fn nautilus_data_type_rejects_order_book_spellings(#[case] value: &str) {
1369        assert!(value.parse::<NautilusDataType>().is_err());
1370    }
1371
1372    #[rstest]
1373    #[case(NautilusRecordType::AccountState, "AccountState")]
1374    #[case(NautilusRecordType::OrderFilled, "OrderFilled")]
1375    #[case(NautilusRecordType::OrderFillVoided, "OrderFillVoided")]
1376    #[case(NautilusRecordType::ExecutionMassStatus, "ExecutionMassStatus")]
1377    fn nautilus_record_type_display_from_str_roundtrips(
1378        #[case] record_type: NautilusRecordType,
1379        #[case] expected: &str,
1380    ) {
1381        assert_eq!(record_type.to_string(), expected);
1382        assert_eq!(expected.parse::<NautilusRecordType>().unwrap(), record_type);
1383    }
1384
1385    #[rstest]
1386    #[case("account_state", NautilusRecordType::AccountState)]
1387    #[case("order_filled", NautilusRecordType::OrderFilled)]
1388    #[case("order_fill_voided", NautilusRecordType::OrderFillVoided)]
1389    #[case("position_snapshot", NautilusRecordType::PositionSnapshot)]
1390    fn nautilus_record_type_storage_names_parse(
1391        #[case] value: &str,
1392        #[case] expected: NautilusRecordType,
1393    ) {
1394        assert_eq!(value.parse::<NautilusRecordType>().unwrap(), expected);
1395    }
1396
1397    #[rstest]
1398    fn data_batches_group_mixed_rows_in_first_seen_order() {
1399        let data = vec![
1400            Data::Quote(QuoteTick::default()),
1401            Data::Trade(TradeTick::default()),
1402            Data::Quote(QuoteTick::default()),
1403        ];
1404
1405        let batches = DataBatch::from_data_vec_grouped(&data).unwrap();
1406
1407        assert_eq!(batches.len(), 2);
1408        assert!(matches!(&batches[0], DataBatch::Quote(rows) if rows.len() == 2));
1409        assert!(matches!(&batches[1], DataBatch::Trade(rows) if rows.len() == 1));
1410    }
1411
1412    #[rstest]
1413    fn data_batches_preserve_order_book_delta_event_batches() {
1414        let instrument_id = InstrumentId::from("AUD/USD.SIM");
1415        let delta = OrderBookDelta::clear(instrument_id, 1, UnixNanos::from(2), UnixNanos::from(3));
1416        let data = vec![Data::BookDeltas(Box::new(OrderBookDeltas::new(
1417            instrument_id,
1418            vec![delta],
1419        )))];
1420
1421        let batches = DataBatch::from_data_vec_grouped(&data).unwrap();
1422
1423        assert_eq!(batches.len(), 1);
1424        assert!(
1425            matches!(&batches[0], DataBatch::BookDeltas(rows) if rows.len() == 1 && rows[0].deltas == vec![delta])
1426        );
1427    }
1428    #[rstest]
1429    fn test_batch_view_make_mut_reuses_unshared_backing() {
1430        let mut view = BatchView::from(vec![3, 1, 2]);
1431        let backing = Arc::as_ptr(view.arc());
1432
1433        view.make_mut().sort_unstable();
1434
1435        assert!(std::ptr::eq(Arc::as_ptr(view.arc()), backing));
1436        assert_eq!(view.as_ref(), &[1, 2, 3]);
1437    }
1438
1439    #[rstest]
1440    fn test_batch_view_make_mut_clones_shared_backing_within_range() {
1441        let source = BatchView::new(Arc::new(vec![9, 3, 1, 2]), 1..4);
1442        let mut view = source.clone();
1443
1444        view.make_mut().sort_unstable();
1445
1446        assert!(!Arc::ptr_eq(source.arc(), view.arc()));
1447        assert_eq!(source.as_ref(), &[3, 1, 2]);
1448        assert_eq!(view.as_ref(), &[1, 2, 3]);
1449        assert_eq!(view.arc().as_slice(), &[9, 1, 2, 3]);
1450    }
1451}
nautilus_model/data/mod.rs

nautilus_model/data/
mod.rs
