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 for the trading domain model.
17
18pub mod bar;
19pub mod bet;
20pub mod black_scholes;
21pub mod close;
22pub mod custom;
23pub mod delta;
24pub mod deltas;
25pub mod depth;
26pub mod forward;
27pub mod funding;
28pub mod greeks;
29pub mod option_chain;
30pub mod order;
31pub mod prices;
32pub mod quote;
33pub mod registry;
34pub mod status;
35pub mod trade;
36
37#[cfg(any(test, feature = "stubs"))]
38pub mod stubs;
39
40use std::{
41    fmt::{Debug, Display},
42    hash::{Hash, Hasher},
43    str::FromStr,
44};
45
46use nautilus_core::{Params, UnixNanos};
47use serde::{Deserialize, Serialize};
48use serde_json::{Value as JsonValue, to_string};
49
50#[cfg(feature = "defi")]
51use crate::defi::DefiData;
52// Re-exports
53#[rustfmt::skip]  // Keep these grouped
54pub use bar::{Bar, BarSpecification, BarType};
55pub use black_scholes::Greeks;
56pub use close::InstrumentClose;
57#[cfg(feature = "python")]
58pub use custom::PythonCustomDataWrapper;
59pub use custom::{
60    CustomData, CustomDataTrait, ensure_custom_data_json_registered, register_custom_data_json,
61};
62#[cfg(feature = "python")]
63pub use custom::{
64    get_python_data_class, reconstruct_python_custom_data, register_python_data_class,
65};
66pub use delta::OrderBookDelta;
67pub use deltas::{OrderBookDeltas, OrderBookDeltas_API};
68pub use depth::{DEPTH10_LEN, OrderBookDepth10};
69pub use forward::ForwardPrice;
70pub use funding::FundingRateUpdate;
71pub use greeks::{
72    BlackScholesGreeksResult, GreeksData, HasGreeks, OptionGreekValues, PortfolioGreeks,
73    YieldCurveData, black_scholes_greeks, imply_vol_and_greeks, refine_vol_and_greeks,
74};
75pub use option_chain::{OptionChainSlice, OptionGreeks, OptionStrikeData, StrikeRange};
76pub use order::{BookOrder, NULL_ORDER};
77pub use prices::{IndexPriceUpdate, MarkPriceUpdate};
78pub use quote::QuoteTick;
79#[cfg(feature = "arrow")]
80pub use registry::{
81    ArrowDecoder, ArrowEncoder, decode_custom_from_arrow, encode_custom_to_arrow,
82    ensure_arrow_registered, get_arrow_schema, register_arrow,
83};
84#[cfg(feature = "python")]
85pub use registry::{
86    PyExtractor, ensure_py_extractor_registered, ensure_rust_extractor_factory_registered,
87    ensure_rust_extractor_registered, get_rust_extractor, register_py_extractor,
88    register_rust_extractor, register_rust_extractor_factory, try_extract_from_py,
89};
90pub use registry::{
91    deserialize_custom_from_json, ensure_json_deserializer_registered, register_json_deserializer,
92};
93pub use status::InstrumentStatus;
94pub use trade::TradeTick;
95
96use crate::identifiers::{InstrumentId, Venue};
97/// A built-in Nautilus data type.
98///
99/// Not recommended for storing large amounts of data, as the largest variant is significantly
100/// larger (10x) than the smallest.
101#[derive(Debug)]
102pub enum Data {
103    Delta(OrderBookDelta),
104    Deltas(OrderBookDeltas_API),
105    Depth10(Box<OrderBookDepth10>), // This variant is significantly larger
106    Quote(QuoteTick),
107    Trade(TradeTick),
108    Bar(Bar),
109    MarkPriceUpdate(MarkPriceUpdate), // TODO: Rename to MarkPrice once Cython gone
110    IndexPriceUpdate(IndexPriceUpdate), // TODO: Rename to IndexPrice once Cython gone
111    FundingRateUpdate(FundingRateUpdate),
112    InstrumentStatus(InstrumentStatus),
113    OptionGreeks(OptionGreeks),
114    InstrumentClose(InstrumentClose),
115    Custom(CustomData),
116    #[cfg(feature = "defi")]
117    Defi(Box<DefiData>), // This variant is significantly larger
118}
119
120/// A C-compatible representation of [`Data`] for FFI.
121///
122/// This enum matches the standard variants of [`Data`] but excludes the `Custom`
123/// variant which is not FFI-safe.
124#[cfg(feature = "ffi")]
125#[repr(C)]
126#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
127#[allow(non_camel_case_types)]
128pub enum DataFFI {
129    Delta(OrderBookDelta),
130    Deltas(OrderBookDeltas_API),
131    Depth10(Box<OrderBookDepth10>),
132    Quote(QuoteTick),
133    Trade(TradeTick),
134    Bar(Bar),
135    MarkPriceUpdate(MarkPriceUpdate),
136    IndexPriceUpdate(IndexPriceUpdate),
137    InstrumentClose(InstrumentClose),
138}
139
140#[cfg(feature = "ffi")]
141impl TryFrom<Data> for DataFFI {
142    type Error = anyhow::Error;
143
144    fn try_from(value: Data) -> Result<Self, Self::Error> {
145        match value {
146            Data::Delta(x) => Ok(Self::Delta(x)),
147            Data::Deltas(x) => Ok(Self::Deltas(x)),
148            Data::Depth10(x) => Ok(Self::Depth10(x)),
149            Data::Quote(x) => Ok(Self::Quote(x)),
150            Data::Trade(x) => Ok(Self::Trade(x)),
151            Data::Bar(x) => Ok(Self::Bar(x)),
152            Data::MarkPriceUpdate(x) => Ok(Self::MarkPriceUpdate(x)),
153            Data::IndexPriceUpdate(x) => Ok(Self::IndexPriceUpdate(x)),
154            Data::FundingRateUpdate(_) => {
155                anyhow::bail!("Cannot convert Data::FundingRateUpdate to DataFFI")
156            }
157            Data::InstrumentStatus(_) => {
158                anyhow::bail!("Cannot convert Data::InstrumentStatus to DataFFI")
159            }
160            Data::OptionGreeks(_) => {
161                anyhow::bail!("Cannot convert Data::OptionGreeks to DataFFI")
162            }
163            Data::InstrumentClose(x) => Ok(Self::InstrumentClose(x)),
164            Data::Custom(_) => anyhow::bail!("Cannot convert Data::Custom to DataFFI"),
165            #[cfg(feature = "defi")]
166            Data::Defi(_) => anyhow::bail!("Cannot convert Data::Defi to DataFFI"),
167        }
168    }
169}
170
171#[cfg(feature = "ffi")]
172impl From<DataFFI> for Data {
173    fn from(value: DataFFI) -> Self {
174        match value {
175            DataFFI::Delta(x) => Self::Delta(x),
176            DataFFI::Deltas(x) => Self::Deltas(x),
177            DataFFI::Depth10(x) => Self::Depth10(x),
178            DataFFI::Quote(x) => Self::Quote(x),
179            DataFFI::Trade(x) => Self::Trade(x),
180            DataFFI::Bar(x) => Self::Bar(x),
181            DataFFI::MarkPriceUpdate(x) => Self::MarkPriceUpdate(x),
182            DataFFI::IndexPriceUpdate(x) => Self::IndexPriceUpdate(x),
183            DataFFI::InstrumentClose(x) => Self::InstrumentClose(x),
184        }
185    }
186}
187
188impl<'de> Deserialize<'de> for Data {
189    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
190    where
191        D: serde::Deserializer<'de>,
192    {
193        use serde::de::Error;
194        let value = serde_json::Value::deserialize(deserializer)?;
195        let type_name = value
196            .get("type")
197            .and_then(|v| v.as_str())
198            .ok_or_else(|| D::Error::custom("Missing 'type' field in Data"))?
199            .to_string();
200
201        match type_name.as_str() {
202            "OrderBookDelta" => Ok(Self::Delta(
203                serde_json::from_value(value).map_err(D::Error::custom)?,
204            )),
205            "OrderBookDeltas" => Ok(Self::Deltas(
206                serde_json::from_value(value).map_err(D::Error::custom)?,
207            )),
208            "OrderBookDepth10" => Ok(Self::Depth10(
209                serde_json::from_value(value).map_err(D::Error::custom)?,
210            )),
211            "QuoteTick" => Ok(Self::Quote(
212                serde_json::from_value(value).map_err(D::Error::custom)?,
213            )),
214            "TradeTick" => Ok(Self::Trade(
215                serde_json::from_value(value).map_err(D::Error::custom)?,
216            )),
217            "Bar" => Ok(Self::Bar(
218                serde_json::from_value(value).map_err(D::Error::custom)?,
219            )),
220            "MarkPriceUpdate" => Ok(Self::MarkPriceUpdate(
221                serde_json::from_value(value).map_err(D::Error::custom)?,
222            )),
223            "IndexPriceUpdate" => Ok(Self::IndexPriceUpdate(
224                serde_json::from_value(value).map_err(D::Error::custom)?,
225            )),
226            "FundingRateUpdate" => Ok(Self::FundingRateUpdate(
227                serde_json::from_value(value).map_err(D::Error::custom)?,
228            )),
229            "InstrumentStatus" => Ok(Self::InstrumentStatus(
230                serde_json::from_value(value).map_err(D::Error::custom)?,
231            )),
232            "OptionGreeks" => Ok(Self::OptionGreeks(
233                serde_json::from_value(value).map_err(D::Error::custom)?,
234            )),
235            "InstrumentClose" => Ok(Self::InstrumentClose(
236                serde_json::from_value(value).map_err(D::Error::custom)?,
237            )),
238            _ => {
239                if let Some(data) =
240                    deserialize_custom_from_json(&type_name, &value).map_err(D::Error::custom)?
241                {
242                    Ok(data)
243                } else {
244                    Err(D::Error::custom(format!("Unknown Data type: {type_name}")))
245                }
246            }
247        }
248    }
249}
250
251impl Clone for Data {
252    fn clone(&self) -> Self {
253        match self {
254            Self::Delta(x) => Self::Delta(*x),
255            Self::Deltas(x) => Self::Deltas(x.clone()),
256            Self::Depth10(x) => Self::Depth10(x.clone()),
257            Self::Quote(x) => Self::Quote(*x),
258            Self::Trade(x) => Self::Trade(*x),
259            Self::Bar(x) => Self::Bar(*x),
260            Self::MarkPriceUpdate(x) => Self::MarkPriceUpdate(*x),
261            Self::IndexPriceUpdate(x) => Self::IndexPriceUpdate(*x),
262            Self::FundingRateUpdate(x) => Self::FundingRateUpdate(*x),
263            Self::InstrumentStatus(x) => Self::InstrumentStatus(*x),
264            Self::OptionGreeks(x) => Self::OptionGreeks(*x),
265            Self::InstrumentClose(x) => Self::InstrumentClose(*x),
266            Self::Custom(x) => Self::Custom(x.clone()),
267            #[cfg(feature = "defi")]
268            Self::Defi(x) => Self::Defi(x.clone()),
269        }
270    }
271}
272
273impl PartialEq for Data {
274    fn eq(&self, other: &Self) -> bool {
275        match (self, other) {
276            (Self::Delta(a), Self::Delta(b)) => a == b,
277            (Self::Deltas(a), Self::Deltas(b)) => a == b,
278            (Self::Depth10(a), Self::Depth10(b)) => a == b,
279            (Self::Quote(a), Self::Quote(b)) => a == b,
280            (Self::Trade(a), Self::Trade(b)) => a == b,
281            (Self::Bar(a), Self::Bar(b)) => a == b,
282            (Self::MarkPriceUpdate(a), Self::MarkPriceUpdate(b)) => a == b,
283            (Self::IndexPriceUpdate(a), Self::IndexPriceUpdate(b)) => a == b,
284            (Self::FundingRateUpdate(a), Self::FundingRateUpdate(b)) => a == b,
285            (Self::InstrumentStatus(a), Self::InstrumentStatus(b)) => a == b,
286            (Self::OptionGreeks(a), Self::OptionGreeks(b)) => a == b,
287            (Self::InstrumentClose(a), Self::InstrumentClose(b)) => a == b,
288            (Self::Custom(a), Self::Custom(b)) => a == b,
289            #[cfg(feature = "defi")]
290            (Self::Defi(a), Self::Defi(b)) => a == b,
291            _ => false,
292        }
293    }
294}
295
296impl Serialize for Data {
297    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
298    where
299        S: serde::Serializer,
300    {
301        match self {
302            Self::Delta(x) => x.serialize(serializer),
303            Self::Deltas(x) => x.serialize(serializer),
304            Self::Depth10(x) => x.serialize(serializer),
305            Self::Quote(x) => x.serialize(serializer),
306            Self::Trade(x) => x.serialize(serializer),
307            Self::Bar(x) => x.serialize(serializer),
308            Self::MarkPriceUpdate(x) => x.serialize(serializer),
309            Self::IndexPriceUpdate(x) => x.serialize(serializer),
310            Self::FundingRateUpdate(x) => x.serialize(serializer),
311            Self::InstrumentStatus(x) => x.serialize(serializer),
312            Self::OptionGreeks(x) => x.serialize(serializer),
313            Self::InstrumentClose(x) => x.serialize(serializer),
314            Self::Custom(x) => x.serialize(serializer),
315            #[cfg(feature = "defi")]
316            Self::Defi(_) => Err(serde::ser::Error::custom(
317                "Data::Defi serialization is not supported",
318            )),
319        }
320    }
321}
322
323macro_rules! impl_try_from_data {
324    ($variant:ident, $type:ty) => {
325        impl TryFrom<Data> for $type {
326            type Error = ();
327
328            fn try_from(value: Data) -> Result<Self, Self::Error> {
329                match value {
330                    Data::$variant(x) => Ok(x),
331                    _ => Err(()),
332                }
333            }
334        }
335    };
336}
337
338impl TryFrom<Data> for OrderBookDepth10 {
339    type Error = ();
340
341    fn try_from(value: Data) -> Result<Self, Self::Error> {
342        match value {
343            Data::Depth10(x) => Ok(*x),
344            _ => Err(()),
345        }
346    }
347}
348
349impl_try_from_data!(Quote, QuoteTick);
350impl_try_from_data!(Delta, OrderBookDelta);
351impl_try_from_data!(Deltas, OrderBookDeltas_API);
352impl_try_from_data!(Trade, TradeTick);
353impl_try_from_data!(Bar, Bar);
354impl_try_from_data!(MarkPriceUpdate, MarkPriceUpdate);
355impl_try_from_data!(IndexPriceUpdate, IndexPriceUpdate);
356impl_try_from_data!(FundingRateUpdate, FundingRateUpdate);
357impl_try_from_data!(InstrumentStatus, InstrumentStatus);
358impl_try_from_data!(OptionGreeks, OptionGreeks);
359impl_try_from_data!(InstrumentClose, InstrumentClose);
360
361/// Converts a vector of `Data` items to a specific variant type.
362///
363/// Filters and converts the data vector, keeping only items that can be
364/// successfully converted to the target type `T`.
365#[must_use]
366pub fn to_variant<T: TryFrom<Data>>(data: Vec<Data>) -> Vec<T> {
367    data.into_iter()
368        .filter_map(|d| T::try_from(d).ok())
369        .collect()
370}
371
372impl Data {
373    /// Returns the instrument ID for the data.
374    #[must_use]
375    pub fn instrument_id(&self) -> InstrumentId {
376        match self {
377            Self::Delta(delta) => delta.instrument_id,
378            Self::Deltas(deltas) => deltas.instrument_id,
379            Self::Depth10(depth) => depth.instrument_id,
380            Self::Quote(quote) => quote.instrument_id,
381            Self::Trade(trade) => trade.instrument_id,
382            Self::Bar(bar) => bar.bar_type.instrument_id(),
383            Self::MarkPriceUpdate(mark_price) => mark_price.instrument_id,
384            Self::IndexPriceUpdate(index_price) => index_price.instrument_id,
385            Self::FundingRateUpdate(funding_rate) => funding_rate.instrument_id,
386            Self::InstrumentStatus(status) => status.instrument_id,
387            Self::OptionGreeks(greeks) => greeks.instrument_id,
388            Self::InstrumentClose(close) => close.instrument_id,
389            Self::Custom(custom) => custom
390                .data_type
391                .identifier()
392                .and_then(|s| InstrumentId::from_str(s).ok())
393                .or_else(|| {
394                    custom
395                        .data_type
396                        .metadata()
397                        .and_then(|m| m.get_str("instrument_id"))
398                        .and_then(|s| InstrumentId::from_str(s).ok())
399                })
400                .unwrap_or_else(|| InstrumentId::from("NULL.NULL")),
401            #[cfg(feature = "defi")]
402            Self::Defi(defi) => defi.instrument_id(),
403        }
404    }
405
406    /// Returns whether the data is a type of order book data.
407    #[must_use]
408    pub fn is_order_book_data(&self) -> bool {
409        matches!(self, Self::Delta(_) | Self::Deltas(_) | Self::Depth10(_))
410    }
411}
412
413/// Marker trait for types that carry a creation timestamp.
414///
415/// `ts_init` is the moment (UNIX nanoseconds) when this value was first generated or
416/// ingested by Nautilus. It can be used for sequencing, latency measurements,
417/// or monitoring data-pipeline delays.
418pub trait HasTsInit {
419    /// Returns the UNIX timestamp (nanoseconds) when the instance was created.
420    fn ts_init(&self) -> UnixNanos;
421}
422
423/// Trait for data types that have a catalog path prefix.
424pub trait CatalogPathPrefix {
425    /// Returns the path prefix (directory name) for this data type.
426    fn path_prefix() -> &'static str;
427}
428
429/// Macro for implementing [`CatalogPathPrefix`] for data types.
430///
431/// This macro provides a convenient way to implement the trait for multiple types
432/// with their corresponding path prefixes.
433///
434/// # Parameters
435///
436/// - `$type`: The data type to implement the trait for.
437/// - `$path`: The path prefix string for that type.
438#[macro_export]
439macro_rules! impl_catalog_path_prefix {
440    ($type:ty, $path:expr) => {
441        impl $crate::data::CatalogPathPrefix for $type {
442            fn path_prefix() -> &'static str {
443                $path
444            }
445        }
446    };
447}
448
449// Standard implementations for financial data types
450impl_catalog_path_prefix!(QuoteTick, "quotes");
451impl_catalog_path_prefix!(TradeTick, "trades");
452impl_catalog_path_prefix!(OrderBookDelta, "order_book_deltas");
453impl_catalog_path_prefix!(OrderBookDepth10, "order_book_depths");
454impl_catalog_path_prefix!(Bar, "bars");
455impl_catalog_path_prefix!(IndexPriceUpdate, "index_prices");
456impl_catalog_path_prefix!(MarkPriceUpdate, "mark_prices");
457impl_catalog_path_prefix!(FundingRateUpdate, "funding_rate_update");
458impl_catalog_path_prefix!(InstrumentStatus, "instrument_status");
459impl_catalog_path_prefix!(OptionGreeks, "option_greeks");
460impl_catalog_path_prefix!(InstrumentClose, "instrument_closes");
461
462use crate::instruments::InstrumentAny;
463impl_catalog_path_prefix!(InstrumentAny, "instruments");
464
465impl HasTsInit for Data {
466    fn ts_init(&self) -> UnixNanos {
467        match self {
468            Self::Delta(d) => d.ts_init,
469            Self::Deltas(d) => d.ts_init,
470            Self::Depth10(d) => d.ts_init,
471            Self::Quote(q) => q.ts_init,
472            Self::Trade(t) => t.ts_init,
473            Self::Bar(b) => b.ts_init,
474            Self::MarkPriceUpdate(p) => p.ts_init,
475            Self::IndexPriceUpdate(p) => p.ts_init,
476            Self::FundingRateUpdate(f) => f.ts_init,
477            Self::InstrumentStatus(s) => s.ts_init,
478            Self::OptionGreeks(g) => g.ts_init,
479            Self::InstrumentClose(c) => c.ts_init,
480            Self::Custom(c) => c.data.ts_init(),
481            #[cfg(feature = "defi")]
482            Self::Defi(d) => d.ts_init(),
483        }
484    }
485}
486
487/// Checks if the data slice is monotonically increasing by initialization timestamp.
488///
489/// Returns `true` if each element's `ts_init` is less than or equal to the next element's `ts_init`.
490pub fn is_monotonically_increasing_by_init<T: HasTsInit>(data: &[T]) -> bool {
491    data.array_windows()
492        .all(|[a, b]| a.ts_init() <= b.ts_init())
493}
494
495impl From<OrderBookDelta> for Data {
496    fn from(value: OrderBookDelta) -> Self {
497        Self::Delta(value)
498    }
499}
500
501impl From<OrderBookDeltas_API> for Data {
502    fn from(value: OrderBookDeltas_API) -> Self {
503        Self::Deltas(value)
504    }
505}
506
507impl From<OrderBookDepth10> for Data {
508    fn from(value: OrderBookDepth10) -> Self {
509        Self::Depth10(Box::new(value))
510    }
511}
512
513impl From<QuoteTick> for Data {
514    fn from(value: QuoteTick) -> Self {
515        Self::Quote(value)
516    }
517}
518
519impl From<TradeTick> for Data {
520    fn from(value: TradeTick) -> Self {
521        Self::Trade(value)
522    }
523}
524
525impl From<Bar> for Data {
526    fn from(value: Bar) -> Self {
527        Self::Bar(value)
528    }
529}
530
531impl From<MarkPriceUpdate> for Data {
532    fn from(value: MarkPriceUpdate) -> Self {
533        Self::MarkPriceUpdate(value)
534    }
535}
536
537impl From<IndexPriceUpdate> for Data {
538    fn from(value: IndexPriceUpdate) -> Self {
539        Self::IndexPriceUpdate(value)
540    }
541}
542
543impl From<FundingRateUpdate> for Data {
544    fn from(value: FundingRateUpdate) -> Self {
545        Self::FundingRateUpdate(value)
546    }
547}
548
549impl From<InstrumentStatus> for Data {
550    fn from(value: InstrumentStatus) -> Self {
551        Self::InstrumentStatus(value)
552    }
553}
554
555impl From<OptionGreeks> for Data {
556    fn from(value: OptionGreeks) -> Self {
557        Self::OptionGreeks(value)
558    }
559}
560
561impl From<InstrumentClose> for Data {
562    fn from(value: InstrumentClose) -> Self {
563        Self::InstrumentClose(value)
564    }
565}
566
567#[cfg(feature = "defi")]
568impl From<DefiData> for Data {
569    fn from(value: DefiData) -> Self {
570        Self::Defi(Box::new(value))
571    }
572}
573
574/// Builds a string-only view of a JSON value for use in topic (key=value).
575fn value_to_topic_string(v: &JsonValue) -> String {
576    if let Some(s) = v.as_str() {
577        return s.to_string();
578    }
579
580    if let Some(n) = v.as_u64() {
581        return n.to_string();
582    }
583
584    if let Some(n) = v.as_i64() {
585        return n.to_string();
586    }
587
588    if let Some(b) = v.as_bool() {
589        return b.to_string();
590    }
591
592    if let Some(f) = v.as_f64() {
593        return f.to_string();
594    }
595
596    if v.is_null() {
597        return "null".to_string();
598    }
599    serde_json::to_string(v).unwrap_or_default()
600}
601
602/// Builds the topic suffix from Params (string-only view: key=value joined by ".").
603fn params_to_topic_suffix(params: &Params) -> String {
604    params
605        .iter()
606        .map(|(k, v)| format!("{k}={}", value_to_topic_string(v)))
607        .collect::<Vec<_>>()
608        .join(".")
609}
610
611/// Represents a data type including metadata.
612#[derive(Clone, Serialize, Deserialize)]
613#[cfg_attr(
614    feature = "python",
615    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.model", from_py_object)
616)]
617#[cfg_attr(
618    feature = "python",
619    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.model")
620)]
621pub struct DataType {
622    type_name: String,
623    metadata: Option<Params>,
624    topic: String,
625    hash: u64,
626    identifier: Option<String>,
627}
628
629impl DataType {
630    /// Creates a new [`DataType`] instance.
631    #[must_use]
632    pub fn new(type_name: &str, metadata: Option<Params>, identifier: Option<String>) -> Self {
633        // Precompute topic from type_name + metadata (string-only view for backward compatibility)
634        let topic = if let Some(ref meta) = metadata {
635            if meta.is_empty() {
636                type_name.to_string()
637            } else {
638                format!("{type_name}.{}", params_to_topic_suffix(meta))
639            }
640        } else {
641            type_name.to_string()
642        };
643
644        // Precompute hash
645        let mut hasher = std::collections::hash_map::DefaultHasher::new();
646        topic.hash(&mut hasher);
647
648        Self {
649            type_name: type_name.to_owned(),
650            metadata,
651            topic,
652            hash: hasher.finish(),
653            identifier,
654        }
655    }
656
657    /// Creates a [`DataType`] from persisted parts (`type_name`, topic, metadata).
658    /// Hash is recomputed from topic. Use when restoring from legacy `data_type` column.
659    /// Identifier is set to None.
660    #[must_use]
661    pub fn from_parts(type_name: &str, topic: &str, metadata: Option<Params>) -> Self {
662        let mut hasher = std::collections::hash_map::DefaultHasher::new();
663        topic.hash(&mut hasher);
664        Self {
665            type_name: type_name.to_owned(),
666            metadata,
667            topic: topic.to_owned(),
668            hash: hasher.finish(),
669            identifier: None,
670        }
671    }
672
673    /// Serializes to JSON for persistence (`type_name`, metadata, identifier; no topic, no hash).
674    ///
675    /// # Errors
676    ///
677    /// Returns a JSON serialization error if the data cannot be serialized.
678    pub fn to_persistence_json(&self) -> Result<String, serde_json::Error> {
679        let mut map = serde_json::Map::new();
680        map.insert(
681            "type_name".to_string(),
682            serde_json::Value::String(self.type_name.clone()),
683        );
684        map.insert(
685            "metadata".to_string(),
686            self.metadata.as_ref().map_or(serde_json::Value::Null, |m| {
687                serde_json::to_value(m).unwrap_or(serde_json::Value::Null)
688            }),
689        );
690
691        if let Some(ref id) = self.identifier {
692            map.insert(
693                "identifier".to_string(),
694                serde_json::Value::String(id.clone()),
695            );
696        }
697        serde_json::to_string(&serde_json::Value::Object(map))
698    }
699
700    /// Deserializes from JSON produced by `to_persistence_json`.
701    /// Accepts legacy JSON with `topic` (ignored); topic is rebuilt from `type_name` + metadata.
702    ///
703    /// # Errors
704    ///
705    /// Returns an error if the string is not valid JSON or missing required fields.
706    pub fn from_persistence_json(s: &str) -> Result<Self, anyhow::Error> {
707        let value: serde_json::Value =
708            serde_json::from_str(s).map_err(|e| anyhow::anyhow!("Invalid data_type JSON: {e}"))?;
709        let obj = value
710            .as_object()
711            .ok_or_else(|| anyhow::anyhow!("data_type must be a JSON object"))?;
712        let type_name = obj
713            .get("type_name")
714            .and_then(|v| v.as_str())
715            .ok_or_else(|| anyhow::anyhow!("data_type must have type_name"))?
716            .to_string();
717        let metadata = obj.get("metadata").and_then(|m| {
718            if m.is_null() {
719                None
720            } else {
721                let p: Params = serde_json::from_value(m.clone()).ok()?;
722                if p.is_empty() { None } else { Some(p) }
723            }
724        });
725        let identifier = obj
726            .get("identifier")
727            .and_then(|v| v.as_str())
728            .map(String::from);
729        Ok(Self::new(&type_name, metadata, identifier))
730    }
731
732    /// Returns the type name for the data type.
733    #[must_use]
734    pub fn type_name(&self) -> &str {
735        self.type_name.as_str()
736    }
737
738    /// Returns the metadata for the data type.
739    #[must_use]
740    pub fn metadata(&self) -> Option<&Params> {
741        self.metadata.as_ref()
742    }
743
744    /// Returns a string representation of the metadata.
745    #[must_use]
746    pub fn metadata_str(&self) -> String {
747        self.metadata.as_ref().map_or_else(
748            || "null".to_string(),
749            |metadata| to_string(metadata).unwrap_or_default(),
750        )
751    }
752
753    /// Returns metadata as a string-only map (e.g. for Arrow schema metadata).
754    #[must_use]
755    pub fn metadata_string_map(&self) -> Option<std::collections::HashMap<String, String>> {
756        self.metadata.as_ref().map(|p| {
757            p.iter()
758                .map(|(k, v)| (k.clone(), value_to_topic_string(v)))
759                .collect()
760        })
761    }
762
763    /// Returns the precomputed hash for this data type.
764    #[must_use]
765    pub fn precomputed_hash(&self) -> u64 {
766        self.hash
767    }
768
769    /// Returns the messaging topic for the data type.
770    #[must_use]
771    pub fn topic(&self) -> &str {
772        self.topic.as_str()
773    }
774
775    /// Returns the optional catalog path identifier (can contain subdirs, e.g. `"venue//symbol"`).
776    #[must_use]
777    pub fn identifier(&self) -> Option<&str> {
778        self.identifier.as_deref()
779    }
780
781    /// Returns an [`Option<InstrumentId>`] parsed from the metadata.
782    ///
783    /// # Panics
784    ///
785    /// This function panics if:
786    /// - There is no metadata.
787    /// - The `instrument_id` value contained in the metadata is invalid.
788    #[must_use]
789    pub fn instrument_id(&self) -> Option<InstrumentId> {
790        let metadata = self.metadata.as_ref().expect("metadata was `None`");
791        let instrument_id = metadata.get_str("instrument_id")?;
792        Some(
793            InstrumentId::from_str(instrument_id)
794                .expect("Invalid `InstrumentId` for 'instrument_id'"),
795        )
796    }
797
798    /// Returns an [`Option<Venue>`] parsed from the metadata.
799    ///
800    /// # Panics
801    ///
802    /// This function panics if:
803    /// - There is no metadata.
804    /// - The `venue` value contained in the metadata is invalid.
805    #[must_use]
806    pub fn venue(&self) -> Option<Venue> {
807        let metadata = self.metadata.as_ref().expect("metadata was `None`");
808        let venue_str = metadata.get_str("venue")?;
809        Some(Venue::from(venue_str))
810    }
811
812    /// Returns an [`Option<UnixNanos>`] parsed from the metadata `start` field.
813    ///
814    /// # Panics
815    ///
816    /// This function panics if:
817    /// - There is no metadata.
818    /// - The `start` value contained in the metadata is invalid.
819    #[must_use]
820    pub fn start(&self) -> Option<UnixNanos> {
821        let metadata = self.metadata.as_ref()?;
822        let start_str = metadata.get_str("start")?;
823        Some(UnixNanos::from_str(start_str).expect("Invalid `UnixNanos` for 'start'"))
824    }
825
826    /// Returns an [`Option<UnixNanos>`] parsed from the metadata `end` field.
827    ///
828    /// # Panics
829    ///
830    /// This function panics if:
831    /// - There is no metadata.
832    /// - The `end` value contained in the metadata is invalid.
833    #[must_use]
834    pub fn end(&self) -> Option<UnixNanos> {
835        let metadata = self.metadata.as_ref()?;
836        let end_str = metadata.get_str("end")?;
837        Some(UnixNanos::from_str(end_str).expect("Invalid `UnixNanos` for 'end'"))
838    }
839
840    /// Returns an [`Option<usize>`] parsed from the metadata `limit` field.
841    ///
842    /// # Panics
843    ///
844    /// This function panics if:
845    /// - There is no metadata.
846    /// - The `limit` value contained in the metadata is invalid.
847    #[must_use]
848    pub fn limit(&self) -> Option<usize> {
849        let metadata = self.metadata.as_ref()?;
850        metadata.get_usize("limit").or_else(|| {
851            metadata
852                .get_str("limit")
853                .map(|s| s.parse::<usize>().expect("Invalid `usize` for 'limit'"))
854        })
855    }
856}
857
858impl PartialEq for DataType {
859    fn eq(&self, other: &Self) -> bool {
860        self.topic == other.topic
861    }
862}
863
864impl Eq for DataType {}
865
866impl PartialOrd for DataType {
867    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
868        Some(self.cmp(other))
869    }
870}
871
872impl Ord for DataType {
873    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
874        self.topic.cmp(&other.topic)
875    }
876}
877
878impl Hash for DataType {
879    fn hash<H: Hasher>(&self, state: &mut H) {
880        self.hash.hash(state);
881    }
882}
883
884impl Display for DataType {
885    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
886        write!(f, "{}", self.topic)
887    }
888}
889
890impl Debug for DataType {
891    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
892        write!(
893            f,
894            "DataType(type_name={}, metadata={:?}, identifier={:?})",
895            self.type_name, self.metadata, self.identifier
896        )
897    }
898}
899
900#[cfg(test)]
901mod tests {
902    use std::hash::DefaultHasher;
903
904    use rstest::*;
905    use serde_json::json;
906
907    use super::*;
908
909    fn params_from_json(value: serde_json::Value) -> Params {
910        serde_json::from_value(value).expect("valid Params JSON")
911    }
912
913    #[cfg(feature = "ffi")]
914    #[rstest]
915    fn test_funding_rate_update_does_not_convert_to_data_ffi() {
916        let funding_rate = FundingRateUpdate::new(
917            InstrumentId::from("BTCUSDT-PERP.BINANCE"),
918            "0.0001".parse().unwrap(),
919            Some(480),
920            Some(UnixNanos::from(1_000_000_000)),
921            UnixNanos::from(1),
922            UnixNanos::from(2),
923        );
924
925        let err = DataFFI::try_from(Data::FundingRateUpdate(funding_rate)).unwrap_err();
926
927        assert_eq!(
928            err.to_string(),
929            "Cannot convert Data::FundingRateUpdate to DataFFI"
930        );
931    }
932
933    #[rstest]
934    fn test_data_type_creation_with_metadata() {
935        let metadata = Some(params_from_json(
936            json!({"key1": "value1", "key2": "value2"}),
937        ));
938        let data_type = DataType::new("ExampleType", metadata.clone(), None);
939
940        assert_eq!(data_type.type_name(), "ExampleType");
941        assert_eq!(data_type.topic(), "ExampleType.key1=value1.key2=value2");
942        assert_eq!(data_type.metadata(), metadata.as_ref());
943    }
944
945    #[rstest]
946    fn test_data_type_creation_without_metadata() {
947        let data_type = DataType::new("ExampleType", None, None);
948
949        assert_eq!(data_type.type_name(), "ExampleType");
950        assert_eq!(data_type.topic(), "ExampleType");
951        assert_eq!(data_type.metadata(), None);
952    }
953
954    #[rstest]
955    fn test_data_type_equality() {
956        let metadata1 = Some(params_from_json(json!({"key1": "value1"})));
957        let metadata2 = Some(params_from_json(json!({"key1": "value1"})));
958
959        let data_type1 = DataType::new("ExampleType", metadata1, None);
960        let data_type2 = DataType::new("ExampleType", metadata2, None);
961
962        assert_eq!(data_type1, data_type2);
963    }
964
965    #[rstest]
966    fn test_data_type_inequality() {
967        let metadata1 = Some(params_from_json(json!({"key1": "value1"})));
968        let metadata2 = Some(params_from_json(json!({"key2": "value2"})));
969
970        let data_type1 = DataType::new("ExampleType", metadata1, None);
971        let data_type2 = DataType::new("ExampleType", metadata2, None);
972
973        assert_ne!(data_type1, data_type2);
974    }
975
976    #[rstest]
977    fn test_data_type_ordering() {
978        let metadata1 = Some(params_from_json(json!({"key1": "value1"})));
979        let metadata2 = Some(params_from_json(json!({"key2": "value2"})));
980
981        let data_type1 = DataType::new("ExampleTypeA", metadata1, None);
982        let data_type2 = DataType::new("ExampleTypeB", metadata2, None);
983
984        assert!(data_type1 < data_type2);
985    }
986
987    #[rstest]
988    fn test_data_type_hash() {
989        let metadata = Some(params_from_json(json!({"key1": "value1"})));
990
991        let data_type1 = DataType::new("ExampleType", metadata.clone(), None);
992        let data_type2 = DataType::new("ExampleType", metadata, None);
993
994        let mut hasher1 = DefaultHasher::new();
995        data_type1.hash(&mut hasher1);
996        let hash1 = hasher1.finish();
997
998        let mut hasher2 = DefaultHasher::new();
999        data_type2.hash(&mut hasher2);
1000        let hash2 = hasher2.finish();
1001
1002        assert_eq!(hash1, hash2);
1003    }
1004
1005    #[rstest]
1006    fn test_data_type_display() {
1007        let metadata = Some(params_from_json(json!({"key1": "value1"})));
1008        let data_type = DataType::new("ExampleType", metadata, None);
1009
1010        assert_eq!(format!("{data_type}"), "ExampleType.key1=value1");
1011    }
1012
1013    #[rstest]
1014    fn test_data_type_debug() {
1015        let metadata = Some(params_from_json(json!({"key1": "value1"})));
1016        let data_type = DataType::new("ExampleType", metadata.clone(), None);
1017
1018        assert_eq!(
1019            format!("{data_type:?}"),
1020            format!("DataType(type_name=ExampleType, metadata={metadata:?}, identifier=None)")
1021        );
1022    }
1023
1024    #[rstest]
1025    fn test_parse_instrument_id_from_metadata() {
1026        let instrument_id_str = "MSFT.XNAS";
1027        let metadata = Some(params_from_json(
1028            json!({"instrument_id": instrument_id_str}),
1029        ));
1030        let data_type = DataType::new("InstrumentAny", metadata, None);
1031
1032        assert_eq!(
1033            data_type.instrument_id().unwrap(),
1034            InstrumentId::from_str(instrument_id_str).unwrap()
1035        );
1036    }
1037
1038    #[rstest]
1039    fn test_parse_venue_from_metadata() {
1040        let venue_str = "BINANCE";
1041        let metadata = Some(params_from_json(json!({"venue": venue_str})));
1042        let data_type = DataType::new(stringify!(InstrumentAny), metadata, None);
1043
1044        assert_eq!(data_type.venue().unwrap(), Venue::new(venue_str));
1045    }
1046
1047    #[rstest]
1048    fn test_parse_start_from_metadata() {
1049        let start_ns = 1_600_054_595_844_758_000;
1050        let metadata = Some(params_from_json(json!({"start": start_ns.to_string()})));
1051        let data_type = DataType::new(stringify!(TradeTick), metadata, None);
1052
1053        assert_eq!(data_type.start().unwrap(), UnixNanos::from(start_ns),);
1054    }
1055
1056    #[rstest]
1057    fn test_parse_end_from_metadata() {
1058        let end_ns = 1_720_954_595_844_758_000;
1059        let metadata = Some(params_from_json(json!({"end": end_ns.to_string()})));
1060        let data_type = DataType::new(stringify!(TradeTick), metadata, None);
1061
1062        assert_eq!(data_type.end().unwrap(), UnixNanos::from(end_ns),);
1063    }
1064
1065    #[rstest]
1066    fn test_parse_limit_from_metadata() {
1067        let limit = 1000;
1068        let metadata = Some(params_from_json(json!({"limit": limit})));
1069        let data_type = DataType::new(stringify!(TradeTick), metadata, None);
1070
1071        assert_eq!(data_type.limit().unwrap(), limit);
1072    }
1073
1074    #[rstest]
1075    fn test_data_type_persistence_json_with_identifier() {
1076        let data_type = DataType::new("MyCustomType", None, Some("venue//symbol".to_string()));
1077        let json = data_type.to_persistence_json().unwrap();
1078        assert!(!json.contains("topic"));
1079        assert!(json.contains("\"identifier\":\"venue//symbol\""));
1080        let restored = DataType::from_persistence_json(&json).unwrap();
1081        assert_eq!(restored.type_name(), "MyCustomType");
1082        assert_eq!(restored.identifier(), Some("venue//symbol"));
1083        assert_eq!(restored.topic(), "MyCustomType");
1084    }
1085
1086    #[rstest]
1087    fn test_data_type_identifier_getter() {
1088        let data_type = DataType::new("T", None, Some("id".to_string()));
1089        assert_eq!(data_type.identifier(), Some("id"));
1090        let data_type_no_id = DataType::new("T", None, None);
1091        assert_eq!(data_type_no_id.identifier(), None);
1092    }
1093}
nautilus_model/data/mod.rs

nautilus_model/data/
mod.rs
