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, Deserializer, Serialize, Serializer};
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///
398/// A view retains its entire backing buffer. Serialization and debug output include only
399/// the visible rows.
400#[derive(Clone)]
401#[expect(
402    clippy::rc_buffer,
403    reason = "Batch views share Vec storage across ranges"
404)]
405pub struct BatchView<T> {
406    data: Arc<Vec<T>>,
407    range: Range<usize>,
408}
409
410impl<T> BatchView<T> {
411    /// Creates a new [`BatchView`] instance.
412    ///
413    /// # Panics
414    ///
415    /// Panics if `range` is invalid or exceeds `data.len()`.
416    #[must_use]
417    pub fn new(data: Arc<Vec<T>>, range: Range<usize>) -> Self {
418        assert!(range.start <= range.end, "invalid batch view range");
419        assert!(
420            range.end <= data.len(),
421            "batch view range exceeds data length"
422        );
423        Self { data, range }
424    }
425
426    #[must_use]
427    pub fn full(data: Arc<Vec<T>>) -> Self {
428        let len = data.len();
429        Self {
430            data,
431            range: 0..len,
432        }
433    }
434
435    /// Returns the backing buffer, including rows outside this view.
436    #[must_use]
437    pub fn arc(&self) -> &Arc<Vec<T>> {
438        &self.data
439    }
440
441    #[must_use]
442    pub fn range(&self) -> Range<usize> {
443        self.range.clone()
444    }
445
446    /// Returns a sub-view of this batch view.
447    ///
448    /// # Panics
449    ///
450    /// Panics if `[start, end)` is invalid or exceeds this view length.
451    #[must_use]
452    pub fn slice(&self, start: usize, end: usize) -> Self {
453        assert!(start <= end, "invalid batch slice range");
454        assert!(end <= self.len(), "batch slice range exceeds view length");
455        Self {
456            data: Arc::clone(&self.data),
457            range: (self.range.start + start)..(self.range.start + end),
458        }
459    }
460
461    /// Returns the visible rows for modification, copying only those rows if storage is shared.
462    pub fn make_mut(&mut self) -> &mut [T]
463    where
464        T: Clone,
465    {
466        if Arc::get_mut(&mut self.data).is_none() {
467            let data = self.to_vec();
468            self.range = 0..data.len();
469            self.data = Arc::new(data);
470        }
471
472        &mut Arc::make_mut(&mut self.data)[self.range.clone()]
473    }
474}
475
476impl<T: Debug> Debug for BatchView<T> {
477    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
478        f.debug_list().entries(self.iter()).finish()
479    }
480}
481
482impl<T: Serialize> Serialize for BatchView<T> {
483    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
484        self.as_ref().serialize(serializer)
485    }
486}
487
488impl<'de, T: Deserialize<'de>> Deserialize<'de> for BatchView<T> {
489    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
490        Vec::<T>::deserialize(deserializer).map(Self::from)
491    }
492}
493
494impl<T> From<Vec<T>> for BatchView<T> {
495    fn from(data: Vec<T>) -> Self {
496        Self::full(Arc::new(data))
497    }
498}
499
500impl<T> From<Arc<Vec<T>>> for BatchView<T> {
501    fn from(data: Arc<Vec<T>>) -> Self {
502        Self::full(data)
503    }
504}
505
506impl<T> AsRef<[T]> for BatchView<T> {
507    fn as_ref(&self) -> &[T] {
508        self
509    }
510}
511
512impl<T> Deref for BatchView<T> {
513    type Target = [T];
514
515    fn deref(&self) -> &Self::Target {
516        &self.data[self.range.clone()]
517    }
518}
519
520/// Shared typed batch used by replay and catalog fan-out paths.
521#[derive(Clone, Debug)]
522pub enum DataBatch {
523    Instrument(BatchView<InstrumentAny>),
524    BookDelta(BatchView<OrderBookDelta>),
525    BookDeltas(BatchView<OrderBookDeltas>),
526    BookDepth(BatchView<OrderBookDepth>),
527    Quote(BatchView<QuoteTick>),
528    Trade(BatchView<TradeTick>),
529    Bar(BatchView<Bar>),
530    MarkPrice(BatchView<MarkPriceUpdate>),
531    IndexPrice(BatchView<IndexPriceUpdate>),
532    FundingRate(BatchView<FundingRateUpdate>),
533    OptionGreeks(BatchView<OptionGreeks>),
534    InstrumentStatus(BatchView<InstrumentStatus>),
535    InstrumentClose(BatchView<InstrumentClose>),
536    Custom(BatchView<CustomData>),
537    #[cfg(feature = "defi")]
538    Defi(BatchView<DefiData>),
539}
540
541macro_rules! data_batch_from_data_vec {
542    (
543        ($data_type:ident, $input:ident);
544        $(($variant:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?
545    ) => {
546        match $data_type {
547            NautilusDataType::Custom { .. } => {
548                let expected_len = $input.len();
549                let custom = $input
550                    .into_iter()
551                    .filter_map(|item| match item {
552                        Data::Custom(custom) => Some(custom),
553                        _ => None,
554                    })
555                    .collect::<Vec<_>>();
556                anyhow::ensure!(
557                    custom.len() == expected_len,
558                    "catalog query for {} returned rows with another data type",
559                    $data_type,
560                );
561                Ok(Self::Custom(custom.into()))
562            }
563            $(
564                NautilusDataType::$variant => Ok(Self::$batch(
565                    to_variant_for_batch::<$type>($data_type, $input)?.into(),
566                )),
567            )+
568            #[cfg(feature = "defi")]
569            NautilusDataType::Defi => {
570                let expected_len = $input.len();
571                let defi = $input
572                    .into_iter()
573                    .filter_map(|item| match item {
574                        Data::Defi(defi) => Some(*defi),
575                        _ => None,
576                    })
577                    .collect::<Vec<_>>();
578                anyhow::ensure!(
579                    defi.len() == expected_len,
580                    "catalog query for {} returned rows with another data type",
581                    $data_type,
582                );
583                Ok(Self::Defi(defi.into()))
584            }
585        }
586    };
587}
588
589impl DataBatch {
590    /// Converts owned compatibility rows into a typed batch for `data_type`.
591    ///
592    /// This is a transition path for backends that still decode through [`Data`] but should not
593    /// expose legacy batches to typed replay.
594    ///
595    /// # Errors
596    ///
597    /// Returns an error if `data_type` is not replayable as a typed batch.
598    pub fn from_data_vec_for_type(
599        data_type: &NautilusDataType,
600        data: Vec<Data>,
601    ) -> anyhow::Result<Self> {
602        if matches!(data_type, NautilusDataType::OrderBookDelta)
603            && matches!(data.first(), Some(Data::BookDeltas(_)))
604        {
605            return Ok(Self::BookDeltas(
606                to_variant_for_batch::<OrderBookDeltas>(data_type, data)?.into(),
607            ));
608        }
609
610        crate::for_each_data_type!(data_batch_from_data_vec, data_type, data)
611    }
612
613    /// Groups mixed compatibility rows into typed batches.
614    ///
615    /// [`Data::BookDeltas`] values retain their event-batch boundary in a separate typed batch.
616    ///
617    /// # Errors
618    ///
619    /// Returns an error if a row cannot be represented by a typed batch.
620    pub fn from_data_vec_grouped(data: &[Data]) -> anyhow::Result<Vec<Self>> {
621        let mut groups = Vec::<(NautilusDataType, Vec<Data>)>::new();
622
623        for item in data.iter().cloned() {
624            let data_type = NautilusDataType::from_data(&item);
625            let is_deltas = matches!(item, Data::BookDeltas(_));
626
627            if let Some((_, group)) = groups.iter_mut().find(|(group_type, group)| {
628                group_type == &data_type
629                    && group
630                        .first()
631                        .is_some_and(|item| matches!(item, Data::BookDeltas(_)) == is_deltas)
632            }) {
633                group.push(item);
634            } else {
635                groups.push((data_type, vec![item]));
636            }
637        }
638
639        groups
640            .into_iter()
641            .map(|(data_type, data)| Self::from_data_vec_for_type(&data_type, data))
642            .collect()
643    }
644
645    #[must_use]
646    pub fn len(&self) -> usize {
647        match self {
648            Self::Custom(data) => data.len(),
649            Self::Instrument(data) => data.len(),
650            Self::BookDelta(data) => data.len(),
651            Self::BookDeltas(data) => data.len(),
652            Self::BookDepth(data) => data.len(),
653            Self::Quote(data) => data.len(),
654            Self::Trade(data) => data.len(),
655            Self::Bar(data) => data.len(),
656            Self::MarkPrice(data) => data.len(),
657            Self::IndexPrice(data) => data.len(),
658            Self::FundingRate(data) => data.len(),
659            Self::OptionGreeks(data) => data.len(),
660            Self::InstrumentStatus(data) => data.len(),
661            Self::InstrumentClose(data) => data.len(),
662            #[cfg(feature = "defi")]
663            Self::Defi(data) => data.len(),
664        }
665    }
666
667    #[must_use]
668    pub fn is_empty(&self) -> bool {
669        self.len() == 0
670    }
671
672    /// Returns the storage-family name of this batch's data type.
673    #[must_use]
674    pub fn data_type_name(&self) -> &'static str {
675        match self {
676            Self::Custom(_) => "custom",
677            Self::Instrument(_) => "instruments",
678            Self::BookDelta(_) => "order_book_deltas",
679            Self::BookDeltas(_) => "order_book_deltas_batches",
680            Self::BookDepth(_) => "order_book_depths",
681            Self::Quote(_) => "quotes",
682            Self::Trade(_) => "trades",
683            Self::Bar(_) => "bars",
684            Self::MarkPrice(_) => "mark_prices",
685            Self::IndexPrice(_) => "index_prices",
686            Self::FundingRate(_) => "funding_rates",
687            Self::OptionGreeks(_) => "option_greeks",
688            Self::InstrumentStatus(_) => "instrument_status",
689            Self::InstrumentClose(_) => "instrument_closes",
690            #[cfg(feature = "defi")]
691            Self::Defi(_) => "defi",
692        }
693    }
694
695    #[must_use]
696    pub fn is_monotonically_increasing_by_init(&self) -> bool {
697        (0..self.len())
698            .filter_map(|index| self.get(index))
699            .map(|data| data.ts_init())
700            .is_sorted()
701    }
702
703    #[must_use]
704    pub fn get(&self, index: usize) -> Option<DataRef<'_>> {
705        match self {
706            Self::Custom(data) => data.get(index).map(DataRef::Custom),
707            Self::Instrument(data) => data.get(index).map(DataRef::Instrument),
708            Self::BookDelta(data) => data.get(index).map(DataRef::BookDelta),
709            Self::BookDeltas(data) => data.get(index).map(DataRef::BookDeltas),
710            Self::BookDepth(data) => data.get(index).map(DataRef::BookDepth),
711            Self::Quote(data) => data.get(index).map(DataRef::Quote),
712            Self::Trade(data) => data.get(index).map(DataRef::Trade),
713            Self::Bar(data) => data.get(index).map(DataRef::Bar),
714            Self::MarkPrice(data) => data.get(index).map(DataRef::MarkPrice),
715            Self::IndexPrice(data) => data.get(index).map(DataRef::IndexPrice),
716            Self::FundingRate(data) => data.get(index).map(DataRef::FundingRate),
717            Self::OptionGreeks(data) => data.get(index).map(DataRef::OptionGreeks),
718            Self::InstrumentStatus(data) => data.get(index).map(DataRef::InstrumentStatus),
719            Self::InstrumentClose(data) => data.get(index).map(DataRef::InstrumentClose),
720            #[cfg(feature = "defi")]
721            Self::Defi(data) => data.get(index).map(DataRef::Defi),
722        }
723    }
724
725    #[must_use]
726    pub fn aligned_chunk(&self, start: usize, chunk_size: Option<usize>) -> Option<(Self, usize)> {
727        let len = self.len();
728        if start >= len {
729            return None;
730        }
731
732        let mut end = match chunk_size {
733            Some(size) => len.min(start + size.max(1)),
734            None => len,
735        };
736
737        if end < len
738            && let Some(boundary_ts) = self.get(end - 1).map(|data| data.ts_init())
739        {
740            while end < len
741                && self
742                    .get(end)
743                    .is_some_and(|data| data.ts_init() == boundary_ts)
744            {
745                end += 1;
746            }
747        }
748
749        Some((self.slice(start, end), end))
750    }
751
752    #[must_use]
753    pub fn slice(&self, start: usize, end: usize) -> Self {
754        if start == 0 && end == self.len() {
755            return self.clone();
756        }
757
758        match self {
759            Self::Custom(data) => Self::Custom(data.slice(start, end)),
760            Self::Instrument(data) => Self::Instrument(data.slice(start, end)),
761            Self::BookDelta(data) => Self::BookDelta(data.slice(start, end)),
762            Self::BookDeltas(data) => Self::BookDeltas(data.slice(start, end)),
763            Self::BookDepth(data) => Self::BookDepth(data.slice(start, end)),
764            Self::Quote(data) => Self::Quote(data.slice(start, end)),
765            Self::Trade(data) => Self::Trade(data.slice(start, end)),
766            Self::Bar(data) => Self::Bar(data.slice(start, end)),
767            Self::MarkPrice(data) => Self::MarkPrice(data.slice(start, end)),
768            Self::IndexPrice(data) => Self::IndexPrice(data.slice(start, end)),
769            Self::FundingRate(data) => Self::FundingRate(data.slice(start, end)),
770            Self::OptionGreeks(data) => Self::OptionGreeks(data.slice(start, end)),
771            Self::InstrumentStatus(data) => Self::InstrumentStatus(data.slice(start, end)),
772            Self::InstrumentClose(data) => Self::InstrumentClose(data.slice(start, end)),
773            #[cfg(feature = "defi")]
774            Self::Defi(data) => Self::Defi(data.slice(start, end)),
775        }
776    }
777
778    #[must_use]
779    pub fn to_data_vec_for_compat(&self) -> Vec<Data> {
780        (0..self.len())
781            .filter_map(|index| self.get(index).map(|item| item.to_owned_data()))
782            .collect()
783    }
784}
785
786/// Typed values that wrap into their [`DataBatch`] variant.
787pub trait IntoDataBatch: Sized {
788    /// Wraps owned typed values in their [`DataBatch`] variant.
789    #[must_use]
790    fn into_batch(data: Vec<Self>) -> DataBatch;
791}
792
793impl<T: IntoDataBatch> From<Vec<T>> for DataBatch {
794    fn from(data: Vec<T>) -> Self {
795        T::into_batch(data)
796    }
797}
798
799macro_rules! impl_into_data_batch {
800    ($(($variant:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?) => {
801        $(
802            impl IntoDataBatch for $type {
803                fn into_batch(data: Vec<Self>) -> DataBatch {
804                    DataBatch::$batch(data.into())
805                }
806            }
807        )+
808    };
809}
810
811for_each_data_type!(impl_into_data_batch);
812
813impl IntoDataBatch for OrderBookDeltas {
814    fn into_batch(data: Vec<Self>) -> DataBatch {
815        DataBatch::BookDeltas(data.into())
816    }
817}
818
819impl IntoDataBatch for CustomData {
820    fn into_batch(data: Vec<Self>) -> DataBatch {
821        DataBatch::Custom(data.into())
822    }
823}
824
825#[cfg(feature = "defi")]
826impl IntoDataBatch for DefiData {
827    fn into_batch(data: Vec<Self>) -> DataBatch {
828        DataBatch::Defi(data.into())
829    }
830}
831
832/// Typed values that unwrap from their [`DataBatch`] variant.
833pub trait FromDataBatch: Sized {
834    /// Unwraps a typed batch into owned values.
835    ///
836    /// # Errors
837    ///
838    /// Returns an error if `batch` holds another variant.
839    fn from_batch(batch: DataBatch) -> anyhow::Result<Vec<Self>>;
840}
841
842macro_rules! impl_from_data_batch {
843    ($(($variant:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?) => {
844        $(
845            impl FromDataBatch for $type {
846                fn from_batch(batch: DataBatch) -> anyhow::Result<Vec<Self>> {
847                    match batch {
848                        DataBatch::$batch(data) => Ok(data.as_ref().to_vec()),
849                        _ => anyhow::bail!(
850                            "expected a {} batch, found another data batch variant",
851                            stringify!($batch),
852                        ),
853                    }
854                }
855            }
856        )+
857    };
858}
859
860for_each_data_type!(impl_from_data_batch);
861
862impl FromDataBatch for CustomData {
863    fn from_batch(batch: DataBatch) -> anyhow::Result<Vec<Self>> {
864        match batch {
865            DataBatch::Custom(data) => Ok(data.as_ref().to_vec()),
866            _ => anyhow::bail!("expected a Custom batch, found another data batch variant"),
867        }
868    }
869}
870
871#[cfg(feature = "defi")]
872impl FromDataBatch for DefiData {
873    fn from_batch(batch: DataBatch) -> anyhow::Result<Vec<Self>> {
874        match batch {
875            DataBatch::Defi(data) => Ok(data.as_ref().to_vec()),
876            _ => anyhow::bail!("expected a Defi batch, found another data batch variant"),
877        }
878    }
879}
880
881/// Catalog record selector used to query Arrow-backed persisted records.
882#[derive(
883    Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize, strum::Display, strum::EnumIter,
884)]
885pub enum NautilusRecordType {
886    AccountState,
887    OrderInitialized,
888    OrderDenied,
889    OrderEmulated,
890    OrderSubmitted,
891    OrderAccepted,
892    OrderRejected,
893    OrderPendingCancel,
894    OrderCanceled,
895    OrderCancelRejected,
896    OrderExpired,
897    OrderTriggered,
898    OrderPendingUpdate,
899    OrderReleased,
900    OrderModifyRejected,
901    OrderUpdated,
902    OrderFilled,
903    OrderFillVoided,
904    PositionOpened,
905    PositionChanged,
906    PositionClosed,
907    PositionAdjusted,
908    OrderSnapshot,
909    PositionSnapshot,
910    OrderStatusReport,
911    FillReport,
912    PositionStatusReport,
913    ExecutionMassStatus,
914    #[cfg(feature = "defi")]
915    Defi,
916    GreeksData,
917    PortfolioGreeks,
918}
919
920impl FromStr for NautilusRecordType {
921    type Err = anyhow::Error;
922
923    fn from_str(s: &str) -> anyhow::Result<Self> {
924        match s {
925            "AccountState" | "account_state" => Ok(Self::AccountState),
926            "OrderInitialized" | "order_initialized" => Ok(Self::OrderInitialized),
927            "OrderDenied" | "order_denied" => Ok(Self::OrderDenied),
928            "OrderEmulated" | "order_emulated" => Ok(Self::OrderEmulated),
929            "OrderSubmitted" | "order_submitted" => Ok(Self::OrderSubmitted),
930            "OrderAccepted" | "order_accepted" => Ok(Self::OrderAccepted),
931            "OrderRejected" | "order_rejected" => Ok(Self::OrderRejected),
932            "OrderPendingCancel" | "order_pending_cancel" => Ok(Self::OrderPendingCancel),
933            "OrderCanceled" | "order_canceled" => Ok(Self::OrderCanceled),
934            "OrderCancelRejected" | "order_cancel_rejected" => Ok(Self::OrderCancelRejected),
935            "OrderExpired" | "order_expired" => Ok(Self::OrderExpired),
936            "OrderTriggered" | "order_triggered" => Ok(Self::OrderTriggered),
937            "OrderPendingUpdate" | "order_pending_update" => Ok(Self::OrderPendingUpdate),
938            "OrderReleased" | "order_released" => Ok(Self::OrderReleased),
939            "OrderModifyRejected" | "order_modify_rejected" => Ok(Self::OrderModifyRejected),
940            "OrderUpdated" | "order_updated" => Ok(Self::OrderUpdated),
941            "OrderFilled" | "order_filled" => Ok(Self::OrderFilled),
942            "OrderFillVoided" | "order_fill_voided" => Ok(Self::OrderFillVoided),
943            "PositionOpened" | "position_opened" => Ok(Self::PositionOpened),
944            "PositionChanged" | "position_changed" => Ok(Self::PositionChanged),
945            "PositionClosed" | "position_closed" => Ok(Self::PositionClosed),
946            "PositionAdjusted" | "position_adjusted" => Ok(Self::PositionAdjusted),
947            "OrderSnapshot" | "order_snapshot" => Ok(Self::OrderSnapshot),
948            "PositionSnapshot" | "position_snapshot" => Ok(Self::PositionSnapshot),
949            "OrderStatusReport" | "order_status_report" => Ok(Self::OrderStatusReport),
950            "FillReport" | "fill_report" => Ok(Self::FillReport),
951            "PositionStatusReport" | "position_status_report" => Ok(Self::PositionStatusReport),
952            "ExecutionMassStatus" | "execution_mass_status" => Ok(Self::ExecutionMassStatus),
953            "GreeksData" | "greeks_data" => Ok(Self::GreeksData),
954            "PortfolioGreeks" | "portfolio_greeks" => Ok(Self::PortfolioGreeks),
955            "custom" | "CustomData" => {
956                anyhow::bail!("custom data queries require NautilusDataType::Custom")
957            }
958            #[cfg(feature = "defi")]
959            "Defi" | "defi" => Ok(Self::Defi),
960            _ => anyhow::bail!("Invalid `NautilusRecordType`: '{s}'"),
961        }
962    }
963}
964
965impl<'de> Deserialize<'de> for Data {
966    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
967    where
968        D: serde::Deserializer<'de>,
969    {
970        use serde::de::Error;
971        let value = serde_json::Value::deserialize(deserializer)?;
972        let type_name = value
973            .get("type")
974            .and_then(|v| v.as_str())
975            .ok_or_else(|| D::Error::custom("Missing 'type' field in Data"))?;
976
977        match type_name {
978            "Instrument" => Ok(Self::Instrument(Box::new(
979                serde_json::from_value(
980                    value
981                        .get("data")
982                        .cloned()
983                        .ok_or_else(|| D::Error::custom("Missing 'data' field for Instrument"))?,
984                )
985                .map_err(D::Error::custom)?,
986            ))),
987            "OrderBookDelta" => Ok(Self::BookDelta(
988                serde_json::from_value(value).map_err(D::Error::custom)?,
989            )),
990            "OrderBookDeltas" => Ok(Self::BookDeltas(
991                serde_json::from_value(value).map_err(D::Error::custom)?,
992            )),
993            "OrderBookDepth" => Ok(Self::BookDepth(
994                serde_json::from_value(value).map_err(D::Error::custom)?,
995            )),
996            "QuoteTick" => Ok(Self::Quote(
997                serde_json::from_value(value).map_err(D::Error::custom)?,
998            )),
999            "TradeTick" => Ok(Self::Trade(
1000                serde_json::from_value(value).map_err(D::Error::custom)?,
1001            )),
1002            "Bar" => Ok(Self::Bar(
1003                serde_json::from_value(value).map_err(D::Error::custom)?,
1004            )),
1005            "MarkPriceUpdate" => Ok(Self::MarkPrice(
1006                serde_json::from_value(value).map_err(D::Error::custom)?,
1007            )),
1008            "IndexPriceUpdate" => Ok(Self::IndexPrice(
1009                serde_json::from_value(value).map_err(D::Error::custom)?,
1010            )),
1011            "FundingRateUpdate" => Ok(Self::FundingRate(
1012                serde_json::from_value(value).map_err(D::Error::custom)?,
1013            )),
1014            "OptionGreeks" => Ok(Self::OptionGreeks(
1015                serde_json::from_value(value).map_err(D::Error::custom)?,
1016            )),
1017            "InstrumentStatus" => Ok(Self::InstrumentStatus(
1018                serde_json::from_value(value).map_err(D::Error::custom)?,
1019            )),
1020            "InstrumentClose" => Ok(Self::InstrumentClose(
1021                serde_json::from_value(value).map_err(D::Error::custom)?,
1022            )),
1023            _ => {
1024                if let Some(data) =
1025                    deserialize_custom_from_json(type_name, &value).map_err(D::Error::custom)?
1026                {
1027                    Ok(data)
1028                } else {
1029                    Err(D::Error::custom(format!("Unknown Data type: {type_name}")))
1030                }
1031            }
1032        }
1033    }
1034}
1035
1036impl Clone for Data {
1037    fn clone(&self) -> Self {
1038        match self {
1039            Self::Custom(x) => Self::Custom(x.clone()),
1040            Self::Instrument(x) => Self::Instrument(x.clone()),
1041            Self::BookDelta(x) => Self::BookDelta(*x),
1042            Self::BookDeltas(x) => Self::BookDeltas(x.clone()),
1043            Self::BookDepth(x) => Self::BookDepth(x.clone()),
1044            Self::Quote(x) => Self::Quote(*x),
1045            Self::Trade(x) => Self::Trade(*x),
1046            Self::Bar(x) => Self::Bar(*x),
1047            Self::MarkPrice(x) => Self::MarkPrice(*x),
1048            Self::IndexPrice(x) => Self::IndexPrice(*x),
1049            Self::FundingRate(x) => Self::FundingRate(*x),
1050            Self::OptionGreeks(x) => Self::OptionGreeks(*x),
1051            Self::InstrumentStatus(x) => Self::InstrumentStatus(*x),
1052            Self::InstrumentClose(x) => Self::InstrumentClose(*x),
1053            #[cfg(feature = "defi")]
1054            Self::Defi(x) => Self::Defi(x.clone()),
1055        }
1056    }
1057}
1058
1059impl PartialEq for Data {
1060    fn eq(&self, other: &Self) -> bool {
1061        match (self, other) {
1062            (Self::Custom(a), Self::Custom(b)) => a == b,
1063            (Self::Instrument(a), Self::Instrument(b)) => a == b,
1064            (Self::BookDelta(a), Self::BookDelta(b)) => a == b,
1065            (Self::BookDeltas(a), Self::BookDeltas(b)) => a == b,
1066            (Self::BookDepth(a), Self::BookDepth(b)) => a == b,
1067            (Self::Quote(a), Self::Quote(b)) => a == b,
1068            (Self::Trade(a), Self::Trade(b)) => a == b,
1069            (Self::Bar(a), Self::Bar(b)) => a == b,
1070            (Self::MarkPrice(a), Self::MarkPrice(b)) => a == b,
1071            (Self::IndexPrice(a), Self::IndexPrice(b)) => a == b,
1072            (Self::FundingRate(a), Self::FundingRate(b)) => a == b,
1073            (Self::OptionGreeks(a), Self::OptionGreeks(b)) => a == b,
1074            (Self::InstrumentStatus(a), Self::InstrumentStatus(b)) => a == b,
1075            (Self::InstrumentClose(a), Self::InstrumentClose(b)) => a == b,
1076            #[cfg(feature = "defi")]
1077            (Self::Defi(a), Self::Defi(b)) => a == b,
1078            _ => false,
1079        }
1080    }
1081}
1082
1083impl Serialize for Data {
1084    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1085    where
1086        S: serde::Serializer,
1087    {
1088        match self {
1089            Self::Custom(x) => x.serialize(serializer),
1090            Self::Instrument(instrument) => serde_json::json!({
1091                "type": "Instrument",
1092                "data": instrument,
1093            })
1094            .serialize(serializer),
1095            Self::BookDelta(x) => x.serialize(serializer),
1096            Self::BookDeltas(x) => x.serialize(serializer),
1097            Self::BookDepth(x) => x.serialize(serializer),
1098            Self::Quote(x) => x.serialize(serializer),
1099            Self::Trade(x) => x.serialize(serializer),
1100            Self::Bar(x) => x.serialize(serializer),
1101            Self::MarkPrice(x) => x.serialize(serializer),
1102            Self::IndexPrice(x) => x.serialize(serializer),
1103            Self::FundingRate(x) => x.serialize(serializer),
1104            Self::OptionGreeks(x) => x.serialize(serializer),
1105            Self::InstrumentStatus(x) => x.serialize(serializer),
1106            Self::InstrumentClose(x) => x.serialize(serializer),
1107            #[cfg(feature = "defi")]
1108            Self::Defi(_) => Err(serde::ser::Error::custom(
1109                "Data::Defi serialization is not supported",
1110            )),
1111        }
1112    }
1113}
1114
1115macro_rules! impl_data_conversions {
1116    ($variant:ident, $type:ty) => {
1117        impl TryFrom<Data> for $type {
1118            type Error = ();
1119
1120            fn try_from(value: Data) -> Result<Self, Self::Error> {
1121                match value {
1122                    Data::$variant(x) => Ok(x),
1123                    _ => Err(()),
1124                }
1125            }
1126        }
1127
1128        impl From<$type> for Data {
1129            fn from(value: $type) -> Self {
1130                Self::$variant(value)
1131            }
1132        }
1133    };
1134}
1135
1136impl TryFrom<Data> for OrderBookDepth {
1137    type Error = ();
1138
1139    fn try_from(value: Data) -> Result<Self, Self::Error> {
1140        match value {
1141            Data::BookDepth(x) => Ok(*x),
1142            _ => Err(()),
1143        }
1144    }
1145}
1146
1147impl TryFrom<Data> for InstrumentAny {
1148    type Error = ();
1149
1150    fn try_from(value: Data) -> Result<Self, Self::Error> {
1151        match value {
1152            Data::Instrument(instrument) => Ok(*instrument),
1153            _ => Err(()),
1154        }
1155    }
1156}
1157
1158impl TryFrom<Data> for OrderBookDeltas {
1159    type Error = ();
1160
1161    fn try_from(value: Data) -> Result<Self, Self::Error> {
1162        match value {
1163            Data::BookDeltas(deltas) => Ok(*deltas),
1164            _ => Err(()),
1165        }
1166    }
1167}
1168
1169impl_data_conversions!(Quote, QuoteTick);
1170impl_data_conversions!(BookDelta, OrderBookDelta);
1171impl_data_conversions!(Trade, TradeTick);
1172impl_data_conversions!(Bar, Bar);
1173impl_data_conversions!(MarkPrice, MarkPriceUpdate);
1174impl_data_conversions!(IndexPrice, IndexPriceUpdate);
1175impl_data_conversions!(FundingRate, FundingRateUpdate);
1176impl_data_conversions!(OptionGreeks, OptionGreeks);
1177impl_data_conversions!(InstrumentStatus, InstrumentStatus);
1178impl_data_conversions!(InstrumentClose, InstrumentClose);
1179
1180/// Converts a vector of `Data` items to a specific variant type.
1181///
1182/// Filters and converts the data vector, keeping only items that can be
1183/// successfully converted to the target type `T`.
1184#[must_use]
1185pub fn to_variant<T: TryFrom<Data>>(data: Vec<Data>) -> Vec<T> {
1186    data.into_iter()
1187        .filter_map(|d| T::try_from(d).ok())
1188        .collect()
1189}
1190
1191fn to_variant_for_batch<T: TryFrom<Data>>(
1192    data_type: &NautilusDataType,
1193    data: Vec<Data>,
1194) -> anyhow::Result<Vec<T>> {
1195    let expected_len = data.len();
1196    let converted = to_variant(data);
1197    anyhow::ensure!(
1198        converted.len() == expected_len,
1199        "catalog query for {data_type} returned rows with another data type",
1200    );
1201    Ok(converted)
1202}
1203
1204impl Data {
1205    /// Returns the instrument ID for the data.
1206    #[must_use]
1207    pub fn instrument_id(&self) -> InstrumentId {
1208        DataRef::from(self).instrument_id()
1209    }
1210
1211    /// Returns whether the data is a type of order book data.
1212    #[must_use]
1213    pub fn is_order_book_data(&self) -> bool {
1214        DataRef::from(self).is_order_book_data()
1215    }
1216}
1217
1218impl From<InstrumentAny> for Data {
1219    fn from(value: InstrumentAny) -> Self {
1220        Self::Instrument(Box::new(value))
1221    }
1222}
1223
1224/// Marker trait for types that carry a creation timestamp.
1225///
1226/// `ts_init` is the moment (UNIX nanoseconds) when this value was first generated or
1227/// ingested by Nautilus. It can be used for sequencing, latency measurements,
1228/// or monitoring data-pipeline delays.
1229pub trait HasTsInit {
1230    /// Returns the UNIX timestamp (nanoseconds) when the instance was created.
1231    fn ts_init(&self) -> UnixNanos;
1232}
1233
1234impl HasTsInit for Data {
1235    fn ts_init(&self) -> UnixNanos {
1236        DataRef::from(self).ts_init()
1237    }
1238}
1239
1240/// Checks if the data slice is monotonically increasing by initialization timestamp.
1241///
1242/// Returns `true` if each element's `ts_init` is less than or equal to the next element's `ts_init`.
1243pub fn is_monotonically_increasing_by_init<T: HasTsInit>(data: &[T]) -> bool {
1244    data.array_windows()
1245        .all(|[a, b]| a.ts_init() <= b.ts_init())
1246}
1247
1248impl From<OrderBookDeltas> for Data {
1249    fn from(value: OrderBookDeltas) -> Self {
1250        Self::BookDeltas(Box::new(value))
1251    }
1252}
1253
1254impl From<OrderBookDepth> for Data {
1255    fn from(value: OrderBookDepth) -> Self {
1256        Self::BookDepth(Box::new(value))
1257    }
1258}
1259
1260#[cfg(feature = "defi")]
1261impl From<DefiData> for Data {
1262    fn from(value: DefiData) -> Self {
1263        Self::Defi(Box::new(value))
1264    }
1265}
1266
1267/// Invokes a macro with every built-in typed data family.
1268///
1269/// Each tuple contains the canonical family name, concrete type, `Data` variant,
1270/// `DataBatch` variant, and catalog path prefix. The first four fields use the same name whenever
1271/// the family has one concrete representation. Custom and DeFi data have no single built-in
1272/// concrete type; `Deltas` and `Data` are aggregate compatibility batches. Consumers handle
1273/// those variants explicitly.
1274#[macro_export]
1275macro_rules! for_each_data_type {
1276    ($macro:ident) => {
1277        $macro! {
1278            (Instrument, InstrumentAny, Instrument, Instrument, "instruments"),
1279            (QuoteTick, QuoteTick, Quote, Quote, "quotes"),
1280            (TradeTick, TradeTick, Trade, Trade, "trades"),
1281            (Bar, Bar, Bar, Bar, "bars"),
1282            (OrderBookDelta, OrderBookDelta, BookDelta, BookDelta, "order_book_deltas"),
1283            (OrderBookDepth, OrderBookDepth, BookDepth, BookDepth, "order_book_depths"),
1284            (MarkPriceUpdate, MarkPriceUpdate, MarkPrice, MarkPrice, "mark_prices"),
1285            (IndexPriceUpdate, IndexPriceUpdate, IndexPrice, IndexPrice, "index_prices"),
1286            (FundingRateUpdate, FundingRateUpdate, FundingRate, FundingRate, "funding_rates"),
1287            (InstrumentStatus, InstrumentStatus, InstrumentStatus, InstrumentStatus, "instrument_status"),
1288            (OptionGreeks, OptionGreeks, OptionGreeks, OptionGreeks, "option_greeks"),
1289            (InstrumentClose, InstrumentClose, InstrumentClose, InstrumentClose, "instrument_closes"),
1290        }
1291    };
1292    ($macro:ident, $($args:tt)*) => {
1293        $macro! {
1294            ($($args)*);
1295            (Instrument, InstrumentAny, Instrument, Instrument, "instruments"),
1296            (QuoteTick, QuoteTick, Quote, Quote, "quotes"),
1297            (TradeTick, TradeTick, Trade, Trade, "trades"),
1298            (Bar, Bar, Bar, Bar, "bars"),
1299            (OrderBookDelta, OrderBookDelta, BookDelta, BookDelta, "order_book_deltas"),
1300            (OrderBookDepth, OrderBookDepth, BookDepth, BookDepth, "order_book_depths"),
1301            (MarkPriceUpdate, MarkPriceUpdate, MarkPrice, MarkPrice, "mark_prices"),
1302            (IndexPriceUpdate, IndexPriceUpdate, IndexPrice, IndexPrice, "index_prices"),
1303            (FundingRateUpdate, FundingRateUpdate, FundingRate, FundingRate, "funding_rates"),
1304            (InstrumentStatus, InstrumentStatus, InstrumentStatus, InstrumentStatus, "instrument_status"),
1305            (OptionGreeks, OptionGreeks, OptionGreeks, OptionGreeks, "option_greeks"),
1306            (InstrumentClose, InstrumentClose, InstrumentClose, InstrumentClose, "instrument_closes"),
1307        }
1308    };
1309}
1310
1311#[cfg(test)]
1312mod tests {
1313    use std::sync::Arc;
1314
1315    use rstest::*;
1316
1317    use super::*;
1318    use crate::instruments::stubs::crypto_perpetual_ethusdt;
1319
1320    #[rstest]
1321    fn test_depth_family_matches_both_macro_forms() {
1322        macro_rules! family_rows {
1323            (($marker:ident); $($rows:tt)*) => { family_rows!($($rows)*) };
1324            ($(($family:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?) => {
1325                vec![$((stringify!($family), stringify!($type), stringify!($data), stringify!($batch), $prefix)),+]
1326            };
1327        }
1328        let rows = for_each_data_type!(family_rows);
1329        let rows_with_args = for_each_data_type!(family_rows, context);
1330        let depth = rows
1331            .iter()
1332            .find(|row| row.4 == "order_book_depths")
1333            .copied()
1334            .unwrap();
1335        assert_eq!(rows, rows_with_args);
1336        assert_eq!(
1337            depth,
1338            (
1339                "OrderBookDepth",
1340                "OrderBookDepth",
1341                "BookDepth",
1342                "BookDepth",
1343                "order_book_depths"
1344            )
1345        );
1346    }
1347
1348    #[rstest]
1349    fn data_instrument_json_roundtrips() {
1350        let data = Data::from(InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt()));
1351
1352        let encoded = serde_json::to_string(&data).unwrap();
1353        let decoded = serde_json::from_str::<Data>(&encoded).unwrap();
1354
1355        assert_eq!(decoded, data);
1356    }
1357
1358    #[rstest]
1359    #[case(NautilusDataType::Instrument, "Instrument")]
1360    #[case(NautilusDataType::QuoteTick, "QuoteTick")]
1361    #[case(NautilusDataType::OrderBookDelta, "OrderBookDelta")]
1362    #[case(
1363        NautilusDataType::Custom {
1364            type_name: "Example".to_string()
1365        },
1366        "Custom:Example"
1367    )]
1368    fn nautilus_data_type_display_from_str_roundtrips(
1369        #[case] nautilus_data_type: NautilusDataType,
1370        #[case] expected: &str,
1371    ) {
1372        assert_eq!(nautilus_data_type.to_string(), expected);
1373        assert_eq!(
1374            expected.parse::<NautilusDataType>().unwrap(),
1375            nautilus_data_type
1376        );
1377    }
1378
1379    #[rstest]
1380    #[case("instruments", NautilusDataType::Instrument)]
1381    #[case("quotes", NautilusDataType::QuoteTick)]
1382    #[case("order_book_deltas", NautilusDataType::OrderBookDelta)]
1383    #[case("instrument_closes", NautilusDataType::InstrumentClose)]
1384    // Plural class names, so callers can pass a Nautilus type's own name.
1385    #[case("OrderBookDeltas", NautilusDataType::OrderBookDelta)]
1386    #[case("OrderBookDepth", NautilusDataType::OrderBookDepth)]
1387    fn nautilus_data_type_storage_names_parse(
1388        #[case] value: &str,
1389        #[case] expected: NautilusDataType,
1390    ) {
1391        assert_eq!(value.parse::<NautilusDataType>().unwrap(), expected);
1392    }
1393
1394    #[rstest]
1395    #[case("OrderBookDepth10")]
1396    #[case("order_book_depth10")]
1397    fn nautilus_data_type_rejects_former_depth10_spellings(#[case] value: &str) {
1398        assert!(value.parse::<NautilusDataType>().is_err());
1399    }
1400
1401    #[rstest]
1402    #[case("OrderBook")]
1403    #[case("order_book")]
1404    fn nautilus_data_type_rejects_order_book_spellings(#[case] value: &str) {
1405        assert!(value.parse::<NautilusDataType>().is_err());
1406    }
1407
1408    #[rstest]
1409    #[case(NautilusRecordType::AccountState, "AccountState")]
1410    #[case(NautilusRecordType::OrderFilled, "OrderFilled")]
1411    #[case(NautilusRecordType::OrderFillVoided, "OrderFillVoided")]
1412    #[case(NautilusRecordType::ExecutionMassStatus, "ExecutionMassStatus")]
1413    #[case(NautilusRecordType::GreeksData, "GreeksData")]
1414    #[case(NautilusRecordType::PortfolioGreeks, "PortfolioGreeks")]
1415    fn nautilus_record_type_display_from_str_roundtrips(
1416        #[case] record_type: NautilusRecordType,
1417        #[case] expected: &str,
1418    ) {
1419        assert_eq!(record_type.to_string(), expected);
1420        assert_eq!(expected.parse::<NautilusRecordType>().unwrap(), record_type);
1421    }
1422
1423    #[rstest]
1424    #[case("account_state", NautilusRecordType::AccountState)]
1425    #[case("order_filled", NautilusRecordType::OrderFilled)]
1426    #[case("order_fill_voided", NautilusRecordType::OrderFillVoided)]
1427    #[case("position_snapshot", NautilusRecordType::PositionSnapshot)]
1428    #[case("greeks_data", NautilusRecordType::GreeksData)]
1429    #[case("portfolio_greeks", NautilusRecordType::PortfolioGreeks)]
1430    fn nautilus_record_type_storage_names_parse(
1431        #[case] value: &str,
1432        #[case] expected: NautilusRecordType,
1433    ) {
1434        assert_eq!(value.parse::<NautilusRecordType>().unwrap(), expected);
1435    }
1436
1437    #[rstest]
1438    fn data_batches_group_mixed_rows_in_first_seen_order() {
1439        let data = vec![
1440            Data::Quote(QuoteTick::default()),
1441            Data::Trade(TradeTick::default()),
1442            Data::Quote(QuoteTick::default()),
1443        ];
1444
1445        let batches = DataBatch::from_data_vec_grouped(&data).unwrap();
1446
1447        assert_eq!(batches.len(), 2);
1448        assert!(matches!(&batches[0], DataBatch::Quote(rows) if rows.len() == 2));
1449        assert!(matches!(&batches[1], DataBatch::Trade(rows) if rows.len() == 1));
1450    }
1451
1452    #[rstest]
1453    fn data_batches_preserve_order_book_delta_event_batches() {
1454        let instrument_id = InstrumentId::from("AUD/USD.SIM");
1455        let delta = OrderBookDelta::clear(instrument_id, 1, UnixNanos::from(2), UnixNanos::from(3));
1456        let data = vec![Data::BookDeltas(Box::new(OrderBookDeltas::new(
1457            instrument_id,
1458            vec![delta],
1459        )))];
1460
1461        let batches = DataBatch::from_data_vec_grouped(&data).unwrap();
1462
1463        assert_eq!(batches.len(), 1);
1464        assert!(
1465            matches!(&batches[0], DataBatch::BookDeltas(rows) if rows.len() == 1 && rows[0].deltas == vec![delta])
1466        );
1467    }
1468    #[rstest]
1469    fn test_batch_view_make_mut_reuses_unshared_backing() {
1470        let mut view = BatchView::new(Arc::new(vec![9, 3, 1, 2, 8]), 1..4);
1471        let backing = Arc::as_ptr(view.arc());
1472        let rows = view.as_ptr();
1473
1474        view.make_mut().sort_unstable();
1475
1476        assert!(std::ptr::eq(Arc::as_ptr(view.arc()), backing));
1477        assert_eq!(view.as_ptr(), rows);
1478        assert_eq!(view.as_ref(), &[1, 2, 3]);
1479        assert_eq!(view.range(), 1..4);
1480        assert_eq!(view.arc().as_slice(), &[9, 1, 2, 3, 8]);
1481    }
1482
1483    #[rstest]
1484    #[case::whole_view(0, 3)]
1485    #[case::nested(1, 3)]
1486    #[case::empty(1, 1)]
1487    fn test_batch_view_make_mut_copies_only_visible_rows(#[case] start: usize, #[case] end: usize) {
1488        let source = BatchView::new(Arc::new(vec![9, 3, 1, 2, 8]), 1..4);
1489        let mut view = source.slice(start, end);
1490        let mut expected = source[start..end].to_vec();
1491        expected.sort_unstable();
1492
1493        view.make_mut().sort_unstable();
1494
1495        assert!(!Arc::ptr_eq(source.arc(), view.arc()));
1496        assert_eq!(source.as_ref(), &[3, 1, 2]);
1497        assert_eq!(source.range(), 1..4);
1498        assert_eq!(source.arc().as_slice(), &[9, 3, 1, 2, 8]);
1499        assert_eq!(view.as_ref(), expected.as_slice());
1500        assert_eq!(view.range(), 0..expected.len());
1501        assert_eq!(view.arc().as_slice(), expected.as_slice());
1502    }
1503
1504    #[rstest]
1505    fn test_batch_view_make_mut_does_not_clone_hidden_rows() {
1506        #[derive(Debug)]
1507        struct Row(bool);
1508
1509        impl Clone for Row {
1510            fn clone(&self) -> Self {
1511                assert!(self.0, "Hidden rows must not be cloned");
1512                Self(self.0)
1513            }
1514        }
1515
1516        let backing = Arc::new(vec![Row(false), Row(true), Row(true), Row(false)]);
1517        let mut view = BatchView::new(Arc::clone(&backing), 1..3);
1518
1519        view.make_mut()[0].0 = false;
1520
1521        assert_eq!(view.range(), 0..2);
1522        assert_eq!(view.arc().len(), 2);
1523        assert!(!view[0].0);
1524        assert!(view[1].0);
1525        assert!(backing[1].0);
1526        assert!(backing[2].0);
1527    }
1528
1529    #[rstest]
1530    #[case::full(0, 5)]
1531    #[case::narrowed(1, 4)]
1532    #[case::empty(2, 2)]
1533    fn test_batch_view_serialization_and_debug_include_only_visible_rows(
1534        #[case] start: usize,
1535        #[case] end: usize,
1536    ) {
1537        let rows = Arc::new(vec![9, 3, 1, 2, 8]);
1538        let view = BatchView::new(Arc::clone(&rows), start..end);
1539        let visible = &rows[start..end];
1540        let json = serde_json::to_vec(&view).unwrap();
1541        let msgpack = rmp_serde::to_vec_named(&view).unwrap();
1542        let decoded_json: BatchView<i32> = serde_json::from_slice(&json).unwrap();
1543        let decoded_msgpack: BatchView<i32> = rmp_serde::from_slice(&msgpack).unwrap();
1544
1545        assert_eq!(json, serde_json::to_vec(visible).unwrap());
1546        assert_eq!(msgpack, rmp_serde::to_vec_named(visible).unwrap());
1547        assert_eq!(format!("{view:?}"), format!("{visible:?}"));
1548
1549        for decoded in [decoded_json, decoded_msgpack] {
1550            assert_eq!(decoded.as_ref(), visible);
1551            assert_eq!(decoded.range(), 0..visible.len());
1552            assert_eq!(decoded.arc().as_slice(), visible);
1553        }
1554    }
1555}
nautilus_model/data/mod.rs

nautilus_model/data/
mod.rs
