nautilus_model/python/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 close;
21pub mod delta;
22pub mod deltas;
23pub mod depth;
24pub mod forward;
25pub mod funding;
26pub mod greeks;
27pub mod option_chain;
28pub mod order;
29pub mod prices;
30pub mod quote;
31pub mod status;
32pub mod trade;
33
34#[cfg(feature = "python")]
35pub mod custom;
36
37#[cfg(feature = "python")]
38use nautilus_core::python::{
39    params::{params_to_pydict, pydict_to_params},
40    to_pyruntime_err, to_pytype_err, to_pyvalue_err,
41};
42#[cfg(feature = "defi")]
43use pyo3::IntoPyObjectExt;
44use pyo3::prelude::*;
45#[cfg(feature = "python")]
46use pyo3::types::PyDict;
47
48use crate::data::{
49    Bar, CustomData, Data, DataType, FundingRateUpdate, IndexPriceUpdate, InstrumentStatus,
50    MarkPriceUpdate, OptionGreeks, OrderBookDelta, QuoteTick, TradeTick, close::InstrumentClose,
51    is_monotonically_increasing_by_init, register_python_data_class,
52};
53
54const ERROR_MONOTONICITY: &str = "`data` was not monotonically increasing by the `ts_init` field";
55
56#[pymethods]
57#[cfg_attr(feature = "python", pyo3_stub_gen::derive::gen_stub_pymethods)]
58impl DataType {
59    /// Represents a data type including metadata.
60    #[new]
61    #[pyo3(signature = (type_name, metadata=None, identifier=None))]
62    fn py_new(
63        py: Python<'_>,
64        type_name: &str,
65        metadata: Option<Py<PyDict>>,
66        identifier: Option<String>,
67    ) -> PyResult<Self> {
68        let params = match metadata {
69            None => None,
70            Some(d) => pydict_to_params(py, &d)?,
71        };
72        Ok(Self::new(type_name, params, identifier))
73    }
74
75    fn __richcmp__(&self, other: &Self, op: pyo3::pyclass::CompareOp, py: Python<'_>) -> Py<PyAny> {
76        use nautilus_core::python::IntoPyObjectNautilusExt;
77
78        match op {
79            pyo3::pyclass::CompareOp::Eq => (self.topic() == other.topic()).into_py_any_unwrap(py),
80            pyo3::pyclass::CompareOp::Ne => (self.topic() != other.topic()).into_py_any_unwrap(py),
81            _ => py.NotImplemented(),
82        }
83    }
84
85    fn __hash__(&self) -> isize {
86        self.precomputed_hash() as isize
87    }
88
89    /// Returns the type name for the data type.
90    #[getter]
91    #[pyo3(name = "type_name")]
92    fn py_type_name(&self) -> &str {
93        self.type_name()
94    }
95
96    /// Returns the metadata for the data type.
97    #[getter]
98    #[pyo3(name = "metadata")]
99    fn py_metadata(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
100        match self.metadata() {
101            None => Ok(py.None()),
102            Some(p) => Ok(params_to_pydict(py, p)?
103                .bind(py)
104                .clone()
105                .into_any()
106                .unbind()),
107        }
108    }
109
110    /// Returns the messaging topic for the data type.
111    #[getter]
112    #[pyo3(name = "topic")]
113    fn py_topic(&self) -> &str {
114        self.topic()
115    }
116
117    /// Returns the optional catalog path identifier (can contain subdirs, e.g. `"venue//symbol"`).
118    #[getter]
119    #[pyo3(name = "identifier")]
120    fn py_identifier(&self) -> Option<&str> {
121        self.identifier()
122    }
123}
124
125/// Converts a [`Data`] variant into its Python model object.
126///
127/// # Errors
128///
129/// Returns an error if Python object allocation fails.
130pub fn data_to_pyobject(py: Python<'_>, data: Data) -> PyResult<Py<PyAny>> {
131    match data {
132        Data::Quote(quote) => Py::new(py, quote).map(Py::into_any),
133        Data::Trade(trade) => Py::new(py, trade).map(Py::into_any),
134        Data::Bar(bar) => Py::new(py, bar).map(Py::into_any),
135        Data::Delta(delta) => Py::new(py, delta).map(Py::into_any),
136        Data::Deltas(deltas) => Py::new(py, *deltas).map(Py::into_any),
137        Data::Depth10(depth) => Py::new(py, *depth).map(Py::into_any),
138        Data::IndexPrice(price) => Py::new(py, price).map(Py::into_any),
139        Data::MarkPrice(price) => Py::new(py, price).map(Py::into_any),
140        Data::FundingRate(funding) => Py::new(py, funding).map(Py::into_any),
141        Data::OptionGreeks(greeks) => Py::new(py, greeks).map(Py::into_any),
142        Data::InstrumentStatus(status) => Py::new(py, status).map(Py::into_any),
143        Data::InstrumentClose(close) => Py::new(py, close).map(Py::into_any),
144        Data::Custom(custom) => Py::new(py, custom).map(Py::into_any),
145        #[cfg(feature = "defi")]
146        Data::Defi(defi) => (*defi).into_py_any(py),
147    }
148}
149
150/// Transforms the given Python objects into a vector of [`OrderBookDelta`] objects.
151///
152/// # Errors
153///
154/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
155pub fn pyobjects_to_book_deltas(data: Vec<Bound<'_, PyAny>>) -> PyResult<Vec<OrderBookDelta>> {
156    let deltas: Vec<OrderBookDelta> = data
157        .into_iter()
158        .map(|obj| obj.extract::<OrderBookDelta>().map_err(PyErr::from))
159        .collect::<PyResult<Vec<OrderBookDelta>>>()?;
160
161    // Validate monotonically increasing
162    if !is_monotonically_increasing_by_init(&deltas) {
163        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
164    }
165
166    Ok(deltas)
167}
168
169/// Transforms the given Python objects into a vector of [`QuoteTick`] objects.
170///
171/// # Errors
172///
173/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
174pub fn pyobjects_to_quotes(data: Vec<Bound<'_, PyAny>>) -> PyResult<Vec<QuoteTick>> {
175    let quotes: Vec<QuoteTick> = data
176        .into_iter()
177        .map(|obj| obj.extract::<QuoteTick>().map_err(PyErr::from))
178        .collect::<PyResult<Vec<QuoteTick>>>()?;
179
180    // Validate monotonically increasing
181    if !is_monotonically_increasing_by_init(&quotes) {
182        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
183    }
184
185    Ok(quotes)
186}
187
188/// Transforms the given Python objects into a vector of [`TradeTick`] objects.
189///
190/// # Errors
191///
192/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
193pub fn pyobjects_to_trades(data: Vec<Bound<'_, PyAny>>) -> PyResult<Vec<TradeTick>> {
194    let trades: Vec<TradeTick> = data
195        .into_iter()
196        .map(|obj| obj.extract::<TradeTick>().map_err(PyErr::from))
197        .collect::<PyResult<Vec<TradeTick>>>()?;
198
199    // Validate monotonically increasing
200    if !is_monotonically_increasing_by_init(&trades) {
201        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
202    }
203
204    Ok(trades)
205}
206
207/// Transforms the given Python objects into a vector of [`Bar`] objects.
208///
209/// # Errors
210///
211/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
212pub fn pyobjects_to_bars(data: Vec<Bound<'_, PyAny>>) -> PyResult<Vec<Bar>> {
213    let bars: Vec<Bar> = data
214        .into_iter()
215        .map(|obj| obj.extract::<Bar>().map_err(PyErr::from))
216        .collect::<PyResult<Vec<Bar>>>()?;
217
218    // Validate monotonically increasing
219    if !is_monotonically_increasing_by_init(&bars) {
220        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
221    }
222
223    Ok(bars)
224}
225
226/// Transforms the given Python objects into a vector of [`MarkPriceUpdate`] objects.
227///
228/// # Errors
229///
230/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
231pub fn pyobjects_to_mark_prices(data: Vec<Bound<'_, PyAny>>) -> PyResult<Vec<MarkPriceUpdate>> {
232    let mark_prices: Vec<MarkPriceUpdate> = data
233        .into_iter()
234        .map(|obj| obj.extract::<MarkPriceUpdate>().map_err(PyErr::from))
235        .collect::<PyResult<Vec<MarkPriceUpdate>>>()?;
236
237    // Validate monotonically increasing
238    if !is_monotonically_increasing_by_init(&mark_prices) {
239        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
240    }
241
242    Ok(mark_prices)
243}
244
245/// Transforms the given Python objects into a vector of [`IndexPriceUpdate`] objects.
246///
247/// # Errors
248///
249/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
250pub fn pyobjects_to_index_prices(data: Vec<Bound<'_, PyAny>>) -> PyResult<Vec<IndexPriceUpdate>> {
251    let index_prices: Vec<IndexPriceUpdate> = data
252        .into_iter()
253        .map(|obj| obj.extract::<IndexPriceUpdate>().map_err(PyErr::from))
254        .collect::<PyResult<Vec<IndexPriceUpdate>>>()?;
255
256    // Validate monotonically increasing
257    if !is_monotonically_increasing_by_init(&index_prices) {
258        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
259    }
260
261    Ok(index_prices)
262}
263
264/// Transforms the given Python objects into a vector of [`InstrumentStatus`] objects.
265///
266/// # Errors
267///
268/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
269pub fn pyobjects_to_instrument_statuses(
270    data: Vec<Bound<'_, PyAny>>,
271) -> PyResult<Vec<InstrumentStatus>> {
272    let statuses: Vec<InstrumentStatus> = data
273        .into_iter()
274        .map(|obj| obj.extract::<InstrumentStatus>().map_err(PyErr::from))
275        .collect::<PyResult<Vec<InstrumentStatus>>>()?;
276
277    if !is_monotonically_increasing_by_init(&statuses) {
278        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
279    }
280
281    Ok(statuses)
282}
283
284/// Transforms the given Python objects into a vector of [`OptionGreeks`] objects.
285///
286/// # Errors
287///
288/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
289pub fn pyobjects_to_option_greeks(data: Vec<Bound<'_, PyAny>>) -> PyResult<Vec<OptionGreeks>> {
290    let greeks: Vec<OptionGreeks> = data
291        .into_iter()
292        .map(|obj| obj.extract::<OptionGreeks>().map_err(PyErr::from))
293        .collect::<PyResult<Vec<OptionGreeks>>>()?;
294
295    if !is_monotonically_increasing_by_init(&greeks) {
296        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
297    }
298
299    Ok(greeks)
300}
301
302/// Transforms the given Python objects into a vector of [`InstrumentClose`] objects.
303///
304/// # Errors
305///
306/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
307pub fn pyobjects_to_instrument_closes(
308    data: Vec<Bound<'_, PyAny>>,
309) -> PyResult<Vec<InstrumentClose>> {
310    let closes: Vec<InstrumentClose> = data
311        .into_iter()
312        .map(|obj| obj.extract::<InstrumentClose>().map_err(PyErr::from))
313        .collect::<PyResult<Vec<InstrumentClose>>>()?;
314
315    // Validate monotonically increasing
316    if !is_monotonically_increasing_by_init(&closes) {
317        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
318    }
319
320    Ok(closes)
321}
322
323/// Deserializes custom data from JSON bytes into a PyO3 `CustomData` wrapper.
324///
325/// # Errors
326///
327/// Returns a `PyErr` if the type is not registered or JSON deserialization fails.
328#[cfg(feature = "python")]
329#[pyfunction]
330pub fn deserialize_custom_from_json(type_name: &str, payload: &[u8]) -> PyResult<CustomData> {
331    use crate::data::registry;
332    let value: serde_json::Value = serde_json::from_slice(payload)
333        .map_err(|e| to_pyvalue_err(format!("Invalid JSON: {e}")))?;
334    let Some(Data::Custom(custom)) = registry::deserialize_custom_from_json(type_name, &value)
335        .map_err(|e| to_pyvalue_err(format!("Deserialization failed: {e}")))?
336    else {
337        return Err(to_pyvalue_err(format!(
338            "Custom data type \"{type_name}\" is not registered"
339        )));
340    };
341    Ok(custom)
342}
343
344/// Deserializes JSON value to `CustomData` via the data class's `from_json`.
345#[cfg(feature = "python")]
346fn py_json_deserialize_custom_data(
347    data_class: &pyo3::Py<pyo3::PyAny>,
348    value: &serde_json::Value,
349) -> Result<std::sync::Arc<dyn crate::data::CustomDataTrait>, anyhow::Error> {
350    use std::sync::Arc;
351
352    use crate::data::PythonCustomDataWrapper;
353
354    pyo3::Python::attach(|py| {
355        let json_str = serde_json::to_string(&value)?;
356        let json_module = py
357            .import("json")
358            .map_err(|e| anyhow::anyhow!("Failed to import json: {e}"))?;
359        let py_dict = json_module
360            .call_method1("loads", (json_str,))
361            .map_err(|e| anyhow::anyhow!("Failed to parse JSON: {e}"))?;
362
363        let instance = data_class
364            .bind(py)
365            .call_method1("from_json", (py_dict,))
366            .map_err(|e| anyhow::anyhow!("Failed to call from_json: {e}"))?;
367
368        let wrapper = PythonCustomDataWrapper::new(py, &instance)
369            .map_err(|e| anyhow::anyhow!("Failed to create wrapper: {e}"))?;
370
371        Ok(Arc::new(wrapper) as Arc<dyn crate::data::CustomDataTrait>)
372    })
373}
374
375/// Encodes `CustomData` items to `RecordBatch` via Python `encode_record_batch_py`.
376#[allow(unsafe_code)]
377#[cfg(all(feature = "python", feature = "arrow"))]
378fn py_encode_custom_data_to_record_batch(
379    items: &[std::sync::Arc<dyn crate::data::CustomDataTrait>],
380) -> Result<arrow::record_batch::RecordBatch, anyhow::Error> {
381    pyo3::Python::attach(|py| {
382        let py_items: Result<Vec<_>, _> = items.iter().map(|item| item.to_pyobject(py)).collect();
383        let py_items = py_items.map_err(|e| anyhow::anyhow!("Failed to convert to Python: {e}"))?;
384        let py_list = pyo3::types::PyList::new(py, &py_items)
385            .map_err(|e| anyhow::anyhow!("Failed to create list: {e}"))?;
386
387        let first = items
388            .first()
389            .ok_or_else(|| anyhow::anyhow!("No items to encode"))?;
390        let first_py = first.to_pyobject(py)?;
391
392        if first_py
393            .bind(py)
394            .hasattr("encode_record_batch_py")
395            .unwrap_or(false)
396        {
397            let py_batch = first_py
398                .bind(py)
399                .call_method1("encode_record_batch_py", (py_list,))
400                .map_err(|e| anyhow::anyhow!("Failed to call encode_record_batch_py: {e}"))?;
401
402            let mut ffi_array = arrow::ffi::FFI_ArrowArray::empty();
403            let mut ffi_schema = arrow::ffi::FFI_ArrowSchema::empty();
404
405            py_batch.call_method1(
406                "_export_to_c",
407                (
408                    (&raw mut ffi_array as usize),
409                    (&raw mut ffi_schema as usize),
410                ),
411            )?;
412
413            let schema = std::sync::Arc::new(arrow::datatypes::Schema::try_from(&ffi_schema)?);
414            let struct_array_data = unsafe {
415                arrow::ffi::from_ffi_and_data_type(
416                    ffi_array,
417                    arrow::datatypes::DataType::Struct(schema.fields().clone()),
418                )?
419            };
420            let struct_array = arrow::array::StructArray::from(struct_array_data);
421            Ok(arrow::record_batch::RecordBatch::from(&struct_array))
422        } else {
423            anyhow::bail!("Instances must have encode_record_batch_py method")
424        }
425    })
426}
427
428#[cfg(all(feature = "python", feature = "arrow"))]
429fn pyarrow_schema_to_arrow_schema(
430    py_schema: &pyo3::Bound<'_, pyo3::PyAny>,
431) -> PyResult<arrow::datatypes::Schema> {
432    let mut ffi_schema = arrow::ffi::FFI_ArrowSchema::empty();
433    py_schema.call_method1("_export_to_c", ((&raw mut ffi_schema as usize),))?;
434    arrow::datatypes::Schema::try_from(&ffi_schema)
435        .map_err(|e| to_pyvalue_err(format!("Failed to import PyArrow schema: {e}")))
436}
437
438/// Decodes `RecordBatch` to `CustomData` via Python `decode_record_batch_py`.
439#[allow(unsafe_code)]
440#[cfg(all(feature = "python", feature = "arrow"))]
441fn py_decode_record_batch_to_custom_data(
442    data_class: &pyo3::Py<pyo3::PyAny>,
443    metadata: &std::collections::HashMap<String, String>,
444    batch: arrow::record_batch::RecordBatch,
445) -> Result<Vec<crate::data::Data>, anyhow::Error> {
446    use std::sync::Arc;
447
448    use crate::data::PythonCustomDataWrapper;
449
450    pyo3::Python::attach(|py| {
451        let struct_array: arrow::array::StructArray = batch.into();
452        let array_data = arrow::array::Array::to_data(&struct_array);
453        let mut ffi_array = arrow::ffi::FFI_ArrowArray::new(&array_data);
454        let fields = match arrow::array::Array::data_type(&struct_array) {
455            arrow::datatypes::DataType::Struct(f) => f.clone(),
456            _ => unreachable!(),
457        };
458        let mut ffi_schema =
459            arrow::ffi::FFI_ArrowSchema::try_from(arrow::datatypes::DataType::Struct(fields))?;
460
461        let pyarrow = py.import("pyarrow")?;
462        let cls = pyarrow.getattr("RecordBatch")?;
463        let py_batch = cls.call_method1(
464            "_import_from_c",
465            (
466                (&raw mut ffi_array as usize),
467                (&raw mut ffi_schema as usize),
468            ),
469        )?;
470
471        let metadata_py = pyo3::types::PyDict::new(py);
472        for (k, v) in metadata {
473            metadata_py.set_item(k, v)?;
474        }
475
476        let py_list = data_class
477            .bind(py)
478            .call_method1("decode_record_batch_py", (metadata_py, py_batch))
479            .map_err(|e| anyhow::anyhow!("Failed to call decode_record_batch_py: {e}"))?;
480
481        let list = py_list
482            .cast::<pyo3::types::PyList>()
483            .map_err(|_| anyhow::anyhow!("Expected list from decode_record_batch_py"))?;
484
485        let mut result = Vec::new();
486        for item in list.iter() {
487            let wrapper = PythonCustomDataWrapper::new(py, &item)
488                .map_err(|e| anyhow::anyhow!("Failed to create wrapper: {e}"))?;
489            result.push(crate::data::Data::Custom(
490                crate::data::CustomData::from_arc(Arc::new(wrapper)),
491            ));
492        }
493        Ok(result)
494    })
495}
496
497/// Registers a custom data **type** (class) with the catalog registry.
498///
499/// Use this when you prefer to pass the class instead of a sample instance.
500/// The class must have:
501/// - `type_name_static()` class method or `__name__` (used as type name in storage)
502/// - `from_json(data)` class method
503/// - `decode_record_batch_py(metadata, batch)` class method
504/// - Instances must have `ts_event`, `ts_init`, and `encode_record_batch_py(items)`.
505///
506/// # Arguments
507///
508/// - `data_class` - The custom data class (e.g. `MarketTickPython` or `module.MarketTickData`)
509///
510/// # Errors
511///
512/// Returns a `PyErr` if the class lacks required methods or the type is already registered.
513///
514/// # Example
515///
516/// ```python
517/// import json
518///
519/// from nautilus_trader.model import register_custom_data_class
520///
521/// class MarketTickPython:
522///     ts_event = 0
523///     ts_init = 0
524///
525///     def to_json(self):
526///         return json.dumps(self.__dict__)
527///
528///     @classmethod
529///     def from_json(cls, data):
530///         instance = cls()
531///         instance.__dict__.update(data)
532///         return instance
533///
534///     def encode_record_batch_py(self, items):
535///         raise NotImplementedError("Arrow encoding is not configured")
536///
537///     @classmethod
538///     def decode_record_batch_py(cls, metadata, batch):
539///         raise NotImplementedError("Arrow decoding is not configured")
540///
541/// register_custom_data_class(MarketTickPython)
542/// ```
543///
544/// The Arrow methods may raise for a message-bus-only class, but must be implemented before catalog
545/// persistence is used.
546#[cfg(feature = "python")]
547#[pyfunction]
548#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.model")]
549pub fn register_custom_data_class(data_class: &Bound<'_, PyAny>) -> PyResult<()> {
550    use std::sync::Arc;
551
552    use crate::data::registry;
553
554    let _py = data_class.py();
555
556    let type_name: String = if data_class.hasattr("type_name_static")? {
557        data_class.call_method0("type_name_static")?.extract()?
558    } else {
559        data_class.getattr("__name__")?.extract()?
560    };
561
562    #[cfg(feature = "arrow")]
563    if !data_class.hasattr("decode_record_batch_py")? {
564        return Err(to_pytype_err(
565            "Custom data class must have decode_record_batch_py(metadata, batch) class method",
566        ));
567    }
568
569    if !data_class.hasattr("from_json")? {
570        return Err(to_pytype_err(
571            "Custom data class must have from_json(data) class method (Rust macro provides it)",
572        ));
573    }
574
575    register_python_data_class(&type_name, data_class);
576
577    if let Some(extractor) = registry::get_rust_extractor(&type_name) {
578        let _ = registry::ensure_py_extractor_registered(&type_name, extractor);
579    }
580
581    let data_class_for_json = data_class.clone().unbind();
582
583    let json_deserializer = Box::new(
584        move |value: serde_json::Value| -> Result<Arc<dyn crate::data::CustomDataTrait>, anyhow::Error> {
585            pyo3::Python::attach(|py| {
586                py_json_deserialize_custom_data(&data_class_for_json.clone_ref(py), &value)
587            })
588        },
589    );
590
591    registry::ensure_json_deserializer_registered(&type_name, json_deserializer).map_err(|e| {
592        to_pyruntime_err(format!(
593            "Failed to register JSON deserializer for {type_name}: {e}"
594        ))
595    })?;
596
597    #[cfg(feature = "arrow")]
598    {
599        let data_class_for_decode = data_class.clone().unbind();
600        let pyarrow_schema = data_class
601            .getattr("_schema")
602            .ok()
603            .filter(|s| s.hasattr("_export_to_c").unwrap_or(false));
604        let schema = if let Some(py_schema) = pyarrow_schema {
605            Arc::new(pyarrow_schema_to_arrow_schema(&py_schema)?)
606        } else if let Some(schema) = registry::get_arrow_schema(&type_name) {
607            schema
608        } else {
609            Arc::new(arrow::datatypes::Schema::empty())
610        };
611
612        let encoder = Box::new(
613            move |items: &[Arc<dyn crate::data::CustomDataTrait>]| -> Result<
614                arrow::record_batch::RecordBatch,
615                anyhow::Error,
616            > { py_encode_custom_data_to_record_batch(items) },
617        );
618
619        let decoder = Box::new(
620            move |metadata: &std::collections::HashMap<String, String>,
621                  batch: arrow::record_batch::RecordBatch|
622                  -> Result<Vec<crate::data::Data>, anyhow::Error> {
623                pyo3::Python::attach(|py| {
624                    py_decode_record_batch_to_custom_data(
625                        &data_class_for_decode.clone_ref(py),
626                        metadata,
627                        batch,
628                    )
629                })
630            },
631        );
632
633        registry::ensure_arrow_registered(&type_name, schema, encoder, decoder).map_err(|e| {
634            to_pyruntime_err(format!(
635                "Failed to register Arrow encoder/decoder for {type_name}: {e}"
636            ))
637        })?;
638    }
639
640    Ok(())
641}
642
643/// Transforms the given Python objects into a vector of [`FundingRateUpdate`] objects.
644///
645/// # Errors
646///
647/// Returns a `PyErr` if element conversion fails or the data is not monotonically increasing.
648pub fn pyobjects_to_funding_rates(data: Vec<Bound<'_, PyAny>>) -> PyResult<Vec<FundingRateUpdate>> {
649    let funding_rates: Vec<FundingRateUpdate> = data
650        .into_iter()
651        .map(|obj| obj.extract::<FundingRateUpdate>().map_err(PyErr::from))
652        .collect::<PyResult<Vec<FundingRateUpdate>>>()?;
653
654    // Validate monotonically increasing
655    if !is_monotonically_increasing_by_init(&funding_rates) {
656        return Err(to_pyvalue_err(ERROR_MONOTONICITY));
657    }
658
659    Ok(funding_rates)
660}
661
662#[cfg(test)]
663mod tests {
664    use std::sync::Once;
665
666    use rstest::rstest;
667
668    use super::*;
669    use crate::data::{
670        OrderBookDeltas, OrderBookDepth10,
671        stubs::{
672            quote_audusd, stub_bar, stub_delta, stub_deltas, stub_depth10, stub_trade_ethusdt_buy,
673        },
674    };
675
676    fn ensure_python_initialized() {
677        static INIT: Once = Once::new();
678        INIT.call_once(Python::initialize);
679    }
680
681    #[rstest]
682    fn data_to_pyobject_preserves_built_in_data() {
683        ensure_python_initialized();
684
685        let expected_delta = stub_delta();
686        let expected_deltas = stub_deltas();
687        let expected_depth = stub_depth10();
688        let expected_quote = quote_audusd();
689        let expected_trade = stub_trade_ethusdt_buy();
690        let expected_bar = stub_bar();
691
692        Python::attach(|py| {
693            let py_delta = data_to_pyobject(py, Data::Delta(expected_delta)).unwrap();
694            let py_deltas =
695                data_to_pyobject(py, Data::Deltas(Box::new(expected_deltas.clone()))).unwrap();
696            let py_depth = data_to_pyobject(py, Data::Depth10(Box::new(expected_depth))).unwrap();
697            let py_quote = data_to_pyobject(py, Data::Quote(expected_quote)).unwrap();
698            let py_trade = data_to_pyobject(py, Data::Trade(expected_trade)).unwrap();
699            let py_bar = data_to_pyobject(py, Data::Bar(expected_bar)).unwrap();
700
701            let actual_delta = *py_delta.bind(py).cast::<OrderBookDelta>().unwrap().borrow();
702            let actual_deltas = py_deltas
703                .bind(py)
704                .cast::<OrderBookDeltas>()
705                .unwrap()
706                .borrow()
707                .clone();
708            let actual_depth = *py_depth
709                .bind(py)
710                .cast::<OrderBookDepth10>()
711                .unwrap()
712                .borrow();
713            let actual_quote = *py_quote.bind(py).cast::<QuoteTick>().unwrap().borrow();
714            let actual_trade = *py_trade.bind(py).cast::<TradeTick>().unwrap().borrow();
715            let actual_bar = *py_bar.bind(py).cast::<Bar>().unwrap().borrow();
716
717            assert_eq!(actual_delta, expected_delta);
718            assert_eq!(actual_deltas.instrument_id, expected_deltas.instrument_id);
719            assert_eq!(actual_deltas.deltas, expected_deltas.deltas);
720            assert_eq!(actual_deltas.flags, expected_deltas.flags);
721            assert_eq!(actual_deltas.sequence, expected_deltas.sequence);
722            assert_eq!(actual_deltas.ts_event, expected_deltas.ts_event);
723            assert_eq!(actual_deltas.ts_init, expected_deltas.ts_init);
724            assert_eq!(actual_depth, expected_depth);
725            assert_eq!(actual_quote, expected_quote);
726            assert_eq!(actual_trade, expected_trade);
727            assert_eq!(actual_bar, expected_bar);
728        });
729    }
730
731    #[cfg(feature = "defi")]
732    #[rstest]
733    fn data_to_pyobject_preserves_defi_variant() {
734        use nautilus_core::UnixNanos;
735        use ustr::Ustr;
736
737        use crate::defi::{Blockchain, data::Block};
738
739        ensure_python_initialized();
740
741        let block = Block::new(
742            "0x1234".to_string(),
743            "0xabcd".to_string(),
744            42,
745            Ustr::from("0x0000000000000000000000000000000000000000"),
746            100_000,
747            50_000,
748            UnixNanos::from(1_700_000_000u64),
749            Some(Blockchain::Ethereum),
750        );
751
752        Python::attach(|py| {
753            let py_defi = data_to_pyobject(
754                py,
755                Data::Defi(Box::new(crate::defi::data::DefiData::Block(block))),
756            )
757            .unwrap();
758            let defi_type = py.get_type::<crate::defi::data::DefiData>();
759            let block_type = defi_type.getattr("Block").unwrap();
760
761            assert!(py_defi.bind(py).is_instance(&block_type).unwrap());
762            let actual_block = py_defi
763                .bind(py)
764                .getattr("_0")
765                .unwrap()
766                .extract::<Block>()
767                .unwrap();
768            assert_eq!(actual_block.hash, "0x1234");
769            assert_eq!(actual_block.parent_hash, "0xabcd");
770            assert_eq!(actual_block.number, 42);
771            assert_eq!(actual_block.gas_limit, 100_000);
772            assert_eq!(actual_block.gas_used, 50_000);
773            assert_eq!(actual_block.timestamp, UnixNanos::from(1_700_000_000u64));
774            assert_eq!(actual_block.chain, Some(Blockchain::Ethereum));
775        });
776    }
777}
nautilus_model/python/data/mod.rs

nautilus_model/python/data/
mod.rs
