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

nautilus_model/data/
mod.rs
