nautilus_serialization/python/
arrow.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
16use std::{io::Cursor, sync::Arc};
17
18use arrow::{
19    datatypes::Schema,
20    ffi_stream::FFI_ArrowArrayStream,
21    ipc::{reader::StreamReader, writer::StreamWriter},
22    record_batch::{RecordBatch, RecordBatchIterator},
23};
24use nautilus_core::python::{to_pyruntime_err, to_pytype_err, to_pyvalue_err};
25use nautilus_model::{
26    data::{
27        Bar, FundingRateUpdate, IndexPriceUpdate, InstrumentStatus, MarkPriceUpdate, OptionGreeks,
28        OrderBookDelta, OrderBookDepth, QuoteTick, TradeTick,
29        close::InstrumentClose,
30        greeks::{GreeksData, PortfolioGreeks},
31    },
32    python::data::{
33        pyobjects_to_bars, pyobjects_to_book_deltas, pyobjects_to_index_prices,
34        pyobjects_to_instrument_closes, pyobjects_to_instrument_statuses, pyobjects_to_mark_prices,
35        pyobjects_to_option_greeks, pyobjects_to_quotes, pyobjects_to_trades,
36    },
37};
38use pyo3::{
39    conversion::IntoPyObjectExt,
40    prelude::*,
41    types::{PyBytes, PyCapsule, PyType},
42};
43
44use crate::arrow::{
45    ArrowSchemaProvider, DecodeFromRecordBatch, DecodeTypedFromRecordBatch, EncodeToRecordBatch,
46    bars_to_arrow_record_batch_bytes, book_deltas_to_arrow_record_batch_bytes,
47    book_depths_to_arrow_record_batch_bytes, encode_batch_with_metadata,
48    index_prices_to_arrow_record_batch_bytes, instrument_closes_to_arrow_record_batch_bytes,
49    instrument_status_to_arrow_record_batch_bytes, mark_prices_to_arrow_record_batch_bytes,
50    option_greeks_to_arrow_record_batch_bytes, quotes_to_arrow_record_batch_bytes,
51    trades_to_arrow_record_batch_bytes,
52};
53
54/// Transforms the given record `batch` into Python `bytes`.
55///
56/// # Errors
57///
58/// Returns an error if writing the Arrow IPC stream fails.
59pub fn arrow_record_batch_to_pybytes(py: Python, batch: &RecordBatch) -> PyResult<Py<PyBytes>> {
60    arrow_record_batches_to_pybytes(py, &batch.schema(), std::slice::from_ref(batch))
61}
62
63/// Transforms the given record `batches` into Python `bytes` as a single Arrow IPC stream.
64///
65/// # Errors
66///
67/// Returns an error if writing the Arrow IPC stream fails.
68pub fn arrow_record_batches_to_pybytes(
69    py: Python,
70    schema: &Schema,
71    batches: &[RecordBatch],
72) -> PyResult<Py<PyBytes>> {
73    let mut cursor = Cursor::new(Vec::new());
74    {
75        let mut writer = StreamWriter::try_new(&mut cursor, schema).map_err(to_pyruntime_err)?;
76
77        for batch in batches {
78            writer.write(batch).map_err(to_pyruntime_err)?;
79        }
80
81        writer.finish().map_err(to_pyruntime_err)?;
82    }
83
84    let buffer = cursor.into_inner();
85    let pybytes = PyBytes::new(py, &buffer);
86
87    Ok(pybytes.into())
88}
89
90/// Exports the given record `batches` as an Arrow C stream PyCapsule.
91///
92/// # Errors
93///
94/// Returns an error if creating the PyCapsule fails.
95pub fn arrow_record_batches_to_pyarrow_stream(
96    py: Python<'_>,
97    schema: &Schema,
98    batches: Vec<RecordBatch>,
99) -> PyResult<Py<PyAny>> {
100    let reader = RecordBatchIterator::new(batches.into_iter().map(Ok), Arc::new(schema.clone()));
101    let stream = FFI_ArrowArrayStream::new(Box::new(reader));
102
103    // The Arrow PyCapsule protocol requires this exact name for ArrowArrayStream values.
104    let capsule = PyCapsule::new_with_value(py, stream, c"arrow_array_stream")?;
105    Ok(capsule.into_any().unbind())
106}
107
108/// Returns a mapping from field names to Arrow data types for the given Rust data class.
109///
110/// # Errors
111///
112/// Returns an error if the class name is not recognized or schema extraction fails.
113#[pyfunction]
114#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
115pub fn get_arrow_schema_map(py: Python<'_>, cls: &Bound<'_, PyType>) -> PyResult<Py<PyAny>> {
116    let cls_str: String = cls.getattr("__name__")?.extract()?;
117    let result_map = match cls_str.as_str() {
118        stringify!(OrderBookDelta) => OrderBookDelta::get_schema_map(),
119        stringify!(OrderBookDepth) => OrderBookDepth::get_schema_map(),
120        stringify!(QuoteTick) => QuoteTick::get_schema_map(),
121        stringify!(TradeTick) => TradeTick::get_schema_map(),
122        stringify!(Bar) => Bar::get_schema_map(),
123        stringify!(MarkPriceUpdate) => MarkPriceUpdate::get_schema_map(),
124        stringify!(IndexPriceUpdate) => IndexPriceUpdate::get_schema_map(),
125        stringify!(FundingRateUpdate) => FundingRateUpdate::get_schema_map(),
126        stringify!(InstrumentStatus) => InstrumentStatus::get_schema_map(),
127        stringify!(OptionGreeks) => OptionGreeks::get_schema_map(),
128        stringify!(InstrumentClose) => InstrumentClose::get_schema_map(),
129        stringify!(GreeksData) => GreeksData::get_schema_map(),
130        stringify!(PortfolioGreeks) => PortfolioGreeks::get_schema_map(),
131        _ => {
132            return Err(to_pytype_err(format!(
133                "Arrow schema for `{cls_str}` is not currently implemented in Rust."
134            )));
135        }
136    };
137
138    result_map.into_py_any(py)
139}
140
141/// Returns an Arrow IPC stream containing the Rust schema for the given data class.
142///
143/// # Errors
144///
145/// Returns an error if the class name is not recognized or schema serialization fails.
146#[pyfunction]
147#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
148pub fn get_arrow_schema_bytes(py: Python<'_>, cls: &Bound<'_, PyType>) -> PyResult<Py<PyBytes>> {
149    let cls_str: String = cls.getattr("__name__")?.extract()?;
150    let schema = match cls_str.as_str() {
151        stringify!(OrderBookDelta) => OrderBookDelta::get_schema(None),
152        stringify!(OrderBookDepth) => OrderBookDepth::get_schema(None),
153        stringify!(QuoteTick) => QuoteTick::get_schema(None),
154        stringify!(TradeTick) => TradeTick::get_schema(None),
155        stringify!(Bar) => Bar::get_schema(None),
156        stringify!(MarkPriceUpdate) => MarkPriceUpdate::get_schema(None),
157        stringify!(IndexPriceUpdate) => IndexPriceUpdate::get_schema(None),
158        stringify!(FundingRateUpdate) => FundingRateUpdate::get_schema(None),
159        stringify!(InstrumentStatus) => InstrumentStatus::get_schema(None),
160        stringify!(OptionGreeks) => OptionGreeks::get_schema(None),
161        stringify!(InstrumentClose) => InstrumentClose::get_schema(None),
162        stringify!(GreeksData) => GreeksData::get_schema(None),
163        stringify!(PortfolioGreeks) => PortfolioGreeks::get_schema(None),
164        _ => {
165            return Err(to_pytype_err(format!(
166                "Arrow schema for `{cls_str}` is not currently implemented in Rust."
167            )));
168        }
169    };
170
171    arrow_record_batches_to_pybytes(py, &schema, &[])
172}
173
174/// Encodes supported Nautilus data or record objects as an Arrow IPC stream.
175///
176/// # Errors
177///
178/// Returns an error if:
179/// - `data` is empty.
180/// - `data` contains unsupported or incompatible objects.
181/// - `GreeksData` records have different instrument IDs.
182/// - Arrow encoding or IPC serialization fails.
183#[pyfunction]
184#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
185#[expect(clippy::missing_panics_doc)] // Guarded by empty check
186pub fn pyobjects_to_arrow_record_batch_bytes(
187    py: Python,
188    data: Vec<Bound<'_, PyAny>>,
189) -> PyResult<Py<PyBytes>> {
190    if data.is_empty() {
191        return Err(to_pyvalue_err("Empty data"));
192    }
193
194    let data_type: String = data
195        .first()
196        .unwrap() // SAFETY: Unwrap safe as already checked that `data` not empty
197        .getattr("__class__")?
198        .getattr("__name__")?
199        .extract()?;
200
201    match data_type.as_str() {
202        stringify!(OrderBookDelta) => {
203            let deltas = pyobjects_to_book_deltas(data)?;
204            py_book_deltas_to_arrow_record_batch_bytes(py, deltas)
205        }
206        stringify!(OrderBookDepth) => {
207            let depth_snapshots: Vec<OrderBookDepth> = data
208                .into_iter()
209                .map(|obj| obj.extract::<OrderBookDepth>().map_err(Into::into))
210                .collect::<PyResult<Vec<OrderBookDepth>>>()?;
211            py_book_depths_to_arrow_record_batch_bytes(py, depth_snapshots)
212        }
213        stringify!(QuoteTick) => {
214            let quotes = pyobjects_to_quotes(data)?;
215            py_quotes_to_arrow_record_batch_bytes(py, quotes)
216        }
217        stringify!(TradeTick) => {
218            let trades = pyobjects_to_trades(data)?;
219            py_trades_to_arrow_record_batch_bytes(py, trades)
220        }
221        stringify!(Bar) => {
222            let bars = pyobjects_to_bars(data)?;
223            py_bars_to_arrow_record_batch_bytes(py, bars)
224        }
225        stringify!(MarkPriceUpdate) => {
226            let updates = pyobjects_to_mark_prices(data)?;
227            py_mark_prices_to_arrow_record_batch_bytes(py, updates)
228        }
229        stringify!(IndexPriceUpdate) => {
230            let index_prices = pyobjects_to_index_prices(data)?;
231            py_index_prices_to_arrow_record_batch_bytes(py, index_prices)
232        }
233        stringify!(InstrumentStatus) => {
234            let statuses = pyobjects_to_instrument_statuses(data)?;
235            py_instrument_status_to_arrow_record_batch_bytes(py, statuses)
236        }
237        stringify!(OptionGreeks) => {
238            let greeks = pyobjects_to_option_greeks(data)?;
239            py_option_greeks_to_arrow_record_batch_bytes(py, greeks)
240        }
241        stringify!(InstrumentClose) => {
242            let closes = pyobjects_to_instrument_closes(data)?;
243            py_instrument_closes_to_arrow_record_batch_bytes(py, closes)
244        }
245        stringify!(GreeksData) => encode_py_records::<GreeksData>(py, data),
246        stringify!(PortfolioGreeks) => encode_py_records::<PortfolioGreeks>(py, data),
247        _ => Err(to_pyvalue_err(format!(
248            "unsupported data type: {data_type}"
249        ))),
250    }
251}
252
253fn encode_py_records<T>(py: Python<'_>, data: Vec<Bound<'_, PyAny>>) -> PyResult<Py<PyBytes>>
254where
255    T: EncodeToRecordBatch + pyo3::PyClass + Clone,
256{
257    let values = data
258        .into_iter()
259        .map(|obj| {
260            obj.extract::<PyRef<'_, T>>()
261                .map(|value| (*value).clone())
262                .map_err(PyErr::from)
263        })
264        .collect::<PyResult<Vec<_>>>()?;
265
266    let batch = encode_batch_with_metadata(&values).map_err(to_pyvalue_err)?;
267    arrow_record_batch_to_pybytes(py, &batch)
268}
269
270/// Converts a vector of `OrderBookDelta` into an Arrow `RecordBatch`.
271///
272/// # Errors
273///
274/// Returns an error if:
275/// - `data` is empty.
276/// - Instrument IDs differ, or non-clear precision metadata differs.
277/// - Encoding fails.
278#[pyfunction(name = "book_deltas_to_arrow_record_batch_bytes")]
279#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
280#[expect(clippy::needless_pass_by_value)]
281pub fn py_book_deltas_to_arrow_record_batch_bytes(
282    py: Python,
283    data: Vec<OrderBookDelta>,
284) -> PyResult<Py<PyBytes>> {
285    match book_deltas_to_arrow_record_batch_bytes(&data) {
286        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
287        Err(e) => Err(to_pyvalue_err(e)),
288    }
289}
290
291/// Converts a vector of `OrderBookDepth` into an Arrow `RecordBatch`.
292///
293/// # Errors
294///
295/// Returns an error if:
296/// - `data` is empty.
297/// - Metadata differs between rows.
298/// - Encoding fails.
299#[pyfunction(name = "book_depths_to_arrow_record_batch_bytes")]
300#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
301#[expect(clippy::needless_pass_by_value)]
302pub fn py_book_depths_to_arrow_record_batch_bytes(
303    py: Python,
304    data: Vec<OrderBookDepth>,
305) -> PyResult<Py<PyBytes>> {
306    match book_depths_to_arrow_record_batch_bytes(&data) {
307        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
308        Err(e) => Err(to_pyvalue_err(e)),
309    }
310}
311
312/// Converts a vector of `QuoteTick` into an Arrow `RecordBatch`.
313///
314/// # Errors
315///
316/// Returns an error if:
317/// - `data` is empty.
318/// - Metadata differs between rows.
319/// - Encoding fails.
320#[pyfunction(name = "quotes_to_arrow_record_batch_bytes")]
321#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
322#[expect(clippy::needless_pass_by_value)]
323pub fn py_quotes_to_arrow_record_batch_bytes(
324    py: Python,
325    data: Vec<QuoteTick>,
326) -> PyResult<Py<PyBytes>> {
327    match quotes_to_arrow_record_batch_bytes(&data) {
328        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
329        Err(e) => Err(to_pyvalue_err(e)),
330    }
331}
332
333/// Converts a vector of `TradeTick` into an Arrow `RecordBatch`.
334///
335/// # Errors
336///
337/// Returns an error if:
338/// - `data` is empty.
339/// - Metadata differs between rows.
340/// - Encoding fails.
341#[pyfunction(name = "trades_to_arrow_record_batch_bytes")]
342#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
343#[expect(clippy::needless_pass_by_value)]
344pub fn py_trades_to_arrow_record_batch_bytes(
345    py: Python,
346    data: Vec<TradeTick>,
347) -> PyResult<Py<PyBytes>> {
348    match trades_to_arrow_record_batch_bytes(&data) {
349        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
350        Err(e) => Err(to_pyvalue_err(e)),
351    }
352}
353
354/// Converts a vector of `Bar` into an Arrow `RecordBatch`.
355///
356/// # Errors
357///
358/// Returns an error if:
359/// - `data` is empty.
360/// - Metadata differs between rows.
361/// - Encoding fails.
362#[pyfunction(name = "bars_to_arrow_record_batch_bytes")]
363#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
364#[expect(clippy::needless_pass_by_value)]
365pub fn py_bars_to_arrow_record_batch_bytes(py: Python, data: Vec<Bar>) -> PyResult<Py<PyBytes>> {
366    match bars_to_arrow_record_batch_bytes(&data) {
367        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
368        Err(e) => Err(to_pyvalue_err(e)),
369    }
370}
371
372/// Converts a vector of `MarkPriceUpdate` into an Arrow `RecordBatch`.
373///
374/// # Errors
375///
376/// Returns an error if:
377/// - `data` is empty.
378/// - Metadata differs between rows.
379/// - Encoding fails.
380#[pyfunction(name = "mark_prices_to_arrow_record_batch_bytes")]
381#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
382#[expect(clippy::needless_pass_by_value)]
383pub fn py_mark_prices_to_arrow_record_batch_bytes(
384    py: Python,
385    data: Vec<MarkPriceUpdate>,
386) -> PyResult<Py<PyBytes>> {
387    match mark_prices_to_arrow_record_batch_bytes(&data) {
388        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
389        Err(e) => Err(to_pyvalue_err(e)),
390    }
391}
392
393/// Converts a vector of `IndexPriceUpdate` into an Arrow `RecordBatch`.
394///
395/// # Errors
396///
397/// Returns an error if:
398/// - `data` is empty.
399/// - Metadata differs between rows.
400/// - Encoding fails.
401#[pyfunction(name = "index_prices_to_arrow_record_batch_bytes")]
402#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
403#[expect(clippy::needless_pass_by_value)]
404pub fn py_index_prices_to_arrow_record_batch_bytes(
405    py: Python,
406    data: Vec<IndexPriceUpdate>,
407) -> PyResult<Py<PyBytes>> {
408    match index_prices_to_arrow_record_batch_bytes(&data) {
409        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
410        Err(e) => Err(to_pyvalue_err(e)),
411    }
412}
413
414/// Converts a vector of `InstrumentStatus` into an Arrow `RecordBatch`.
415///
416/// # Errors
417///
418/// Returns an error if:
419/// - `data` is empty.
420/// - Metadata differs between rows.
421/// - Encoding fails.
422#[pyfunction(name = "instrument_status_to_arrow_record_batch_bytes")]
423#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
424#[expect(clippy::needless_pass_by_value)]
425pub fn py_instrument_status_to_arrow_record_batch_bytes(
426    py: Python,
427    data: Vec<InstrumentStatus>,
428) -> PyResult<Py<PyBytes>> {
429    match instrument_status_to_arrow_record_batch_bytes(&data) {
430        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
431        Err(e) => Err(to_pyvalue_err(e)),
432    }
433}
434
435/// Converts a vector of `OptionGreeks` into an Arrow `RecordBatch`.
436///
437/// # Errors
438///
439/// Returns an error if:
440/// - `data` is empty.
441/// - Metadata differs between rows.
442/// - Encoding fails.
443#[pyfunction(name = "option_greeks_to_arrow_record_batch_bytes")]
444#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
445#[expect(clippy::needless_pass_by_value)]
446pub fn py_option_greeks_to_arrow_record_batch_bytes(
447    py: Python,
448    data: Vec<OptionGreeks>,
449) -> PyResult<Py<PyBytes>> {
450    match option_greeks_to_arrow_record_batch_bytes(&data) {
451        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
452        Err(e) => Err(to_pyvalue_err(e)),
453    }
454}
455
456/// Decodes Arrow IPC bytes into a list of `OptionGreeks`.
457///
458/// # Errors
459///
460/// Returns an error if decoding fails.
461#[pyfunction(name = "option_greeks_from_arrow_record_batch_bytes")]
462#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
463pub fn py_option_greeks_from_arrow_record_batch_bytes(
464    _py: Python,
465    data: Vec<u8>,
466) -> PyResult<Vec<OptionGreeks>> {
467    let cursor = Cursor::new(data);
468    let reader = StreamReader::try_new(cursor, None).map_err(to_pyruntime_err)?;
469
470    let mut results = Vec::new();
471
472    for batch_result in reader {
473        let batch = batch_result.map_err(to_pyruntime_err)?;
474        let metadata = batch.schema().metadata().clone();
475        let decoded = OptionGreeks::decode_batch(&metadata, batch).map_err(to_pyvalue_err)?;
476        results.extend(decoded);
477    }
478
479    Ok(results)
480}
481
482/// Decodes Arrow IPC bytes into a list of `InstrumentStatus`.
483///
484/// # Errors
485///
486/// Returns an error if decoding fails.
487#[pyfunction(name = "instrument_status_from_arrow_record_batch_bytes")]
488#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
489pub fn py_instrument_status_from_arrow_record_batch_bytes(
490    _py: Python,
491    data: Vec<u8>,
492) -> PyResult<Vec<InstrumentStatus>> {
493    let cursor = Cursor::new(data);
494    let reader = StreamReader::try_new(cursor, None).map_err(to_pyruntime_err)?;
495
496    let mut results = Vec::new();
497
498    for batch_result in reader {
499        let batch = batch_result.map_err(to_pyruntime_err)?;
500        let metadata = batch.schema().metadata().clone();
501        let decoded =
502            InstrumentStatus::decode_typed_batch(&metadata, batch).map_err(to_pyvalue_err)?;
503        results.extend(decoded);
504    }
505
506    Ok(results)
507}
508
509/// Converts a vector of `InstrumentClose` into an Arrow `RecordBatch`.
510///
511/// # Errors
512///
513/// Returns an error if:
514/// - `data` is empty.
515/// - Metadata differs between rows.
516/// - Encoding fails.
517#[pyfunction(name = "instrument_closes_to_arrow_record_batch_bytes")]
518#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.serialization")]
519#[expect(clippy::needless_pass_by_value)]
520pub fn py_instrument_closes_to_arrow_record_batch_bytes(
521    py: Python,
522    data: Vec<InstrumentClose>,
523) -> PyResult<Py<PyBytes>> {
524    match instrument_closes_to_arrow_record_batch_bytes(&data) {
525        Ok(batch) => arrow_record_batch_to_pybytes(py, &batch),
526        Err(e) => Err(to_pyvalue_err(e)),
527    }
528}
529
530#[cfg(test)]
531mod tests {
532    use std::collections::HashMap;
533
534    use nautilus_model::data::stubs::stub_instrument_status;
535    use pyo3::{
536        exceptions::{PyRuntimeError, PyTypeError, PyValueError},
537        types::PyString,
538    };
539    use rstest::rstest;
540
541    use super::*;
542    use crate::arrow::EncodeToRecordBatch;
543
544    #[rstest]
545    fn test_schema_bytes_and_map_match_rust_schema() {
546        Python::initialize();
547        Python::attach(|py| {
548            let cls = py.get_type::<InstrumentStatus>();
549            let bytes = get_arrow_schema_bytes(py, &cls).unwrap();
550            let mut reader = StreamReader::try_new(Cursor::new(bytes.as_bytes(py)), None).unwrap();
551            let map: HashMap<String, String> =
552                get_arrow_schema_map(py, &cls).unwrap().extract(py).unwrap();
553
554            assert_eq!(
555                reader.schema().as_ref(),
556                &InstrumentStatus::get_schema(None)
557            );
558            assert!(reader.next().is_none());
559            assert_eq!(map, InstrumentStatus::get_schema_map());
560        });
561    }
562
563    #[rstest]
564    fn test_schema_rejects_unsupported_class() {
565        Python::initialize();
566        Python::attach(|py| {
567            let cls = py.get_type::<PyString>();
568            let errors = [
569                get_arrow_schema_bytes(py, &cls).unwrap_err(),
570                get_arrow_schema_map(py, &cls).unwrap_err(),
571            ];
572
573            for error in errors {
574                assert!(error.is_instance_of::<PyTypeError>(py));
575                assert_eq!(
576                    error.value(py).to_string(),
577                    "Arrow schema for `str` is not currently implemented in Rust."
578                );
579            }
580        });
581    }
582
583    #[rstest]
584    fn test_status_ipc_preserves_multiple_batches() {
585        let first = stub_instrument_status();
586        let mut second = first;
587        second.ts_event = 31.into();
588        second.ts_init = 47.into();
589        second.reason = Some("venue halt".into());
590        second.is_trading = Some(false);
591        second.is_quoting = Some(true);
592        let rows = [first, second];
593        let batches = rows
594            .iter()
595            .map(|row| InstrumentStatus::encode_batch(&row.metadata(), &[*row]).unwrap())
596            .collect::<Vec<_>>();
597        Python::initialize();
598        Python::attach(|py| {
599            let bytes =
600                arrow_record_batches_to_pybytes(py, &batches[0].schema(), &batches).unwrap();
601            let decoded =
602                py_instrument_status_from_arrow_record_batch_bytes(py, bytes.as_bytes(py).to_vec())
603                    .unwrap();
604            let single =
605                py_instrument_status_to_arrow_record_batch_bytes(py, rows.to_vec()).unwrap();
606            let single_decoded = py_instrument_status_from_arrow_record_batch_bytes(
607                py,
608                single.as_bytes(py).to_vec(),
609            )
610            .unwrap();
611
612            assert_eq!(decoded, rows);
613            assert_eq!(single_decoded, rows);
614        });
615    }
616
617    #[rstest]
618    fn test_ipc_decode_rejects_invalid_stream() {
619        Python::initialize();
620        Python::attach(|py| {
621            let errors = [
622                py_instrument_status_from_arrow_record_batch_bytes(py, vec![1, 2, 3]).unwrap_err(),
623                py_option_greeks_from_arrow_record_batch_bytes(py, vec![1, 2, 3]).unwrap_err(),
624            ];
625
626            for error in errors {
627                assert!(error.is_instance_of::<PyRuntimeError>(py));
628                assert_eq!(
629                    error.value(py).to_string(),
630                    "Ipc error: Expected schema message, found empty stream."
631                );
632            }
633        });
634    }
635
636    #[rstest]
637    fn test_python_encoders_reject_empty_data() {
638        Python::initialize();
639        Python::attach(|py| {
640            let errors = [
641                py_book_deltas_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
642                py_book_depths_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
643                py_quotes_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
644                py_trades_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
645                py_bars_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
646                py_mark_prices_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
647                py_index_prices_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
648                py_instrument_status_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
649                py_option_greeks_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
650                py_instrument_closes_to_arrow_record_batch_bytes(py, vec![]).unwrap_err(),
651            ];
652
653            for error in errors {
654                assert!(error.is_instance_of::<PyValueError>(py));
655                assert_eq!(
656                    error.value(py).to_string(),
657                    crate::arrow::EncodingError::EmptyData.to_string()
658                );
659            }
660        });
661    }
662}
nautilus_serialization/python/arrow.rs

nautilus_serialization/python/
arrow.rs
