nautilus_serialization/arrow/
delta.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::{collections::HashMap, str::FromStr, sync::Arc};
17
18use arrow::{
19    array::{Decimal128Array, UInt8Array, UInt64Array},
20    datatypes::{DataType, Field, Schema},
21    error::ArrowError,
22    record_batch::RecordBatch,
23};
24#[cfg(test)]
25use nautilus_model::identifiers::InstrumentId;
26use nautilus_model::{
27    data::{BookOrder, OrderBookDelta},
28    enums::{BookAction, OrderSide},
29};
30
31use super::{
32    DecodeDataFromRecordBatch, EncodingError, KEY_IDENTIFIER, decode_decimal_price,
33    decode_decimal_quantity, decode_required_timestamp, decode_required_u8, decode_required_u64,
34    enum_dictionary_array, enum_dictionary_data_type, extract_column, extract_column_string,
35    fixed_decimal_data_type, identifier_array_from_display, metadata_with_type_name,
36    parse_metadata, price_decimal_array, quantity_decimal_array, validate_required_column,
37};
38#[cfg(test)]
39use super::{KEY_INSTRUMENT_ID, KEY_PRICE_PRECISION};
40use crate::arrow::{ArrowSchemaProvider, Data, DecodeFromRecordBatch, EncodeToRecordBatch};
41
42impl ArrowSchemaProvider for OrderBookDelta {
43    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
44        let fields = vec![
45            Field::new("action", enum_dictionary_data_type(), false),
46            Field::new("side", enum_dictionary_data_type(), false),
47            Field::new("price", fixed_decimal_data_type(), true),
48            Field::new("size", fixed_decimal_data_type(), true),
49            Field::new("order_id", DataType::UInt64, false),
50            Field::new("flags", DataType::UInt8, false),
51            Field::new("sequence", DataType::UInt64, false),
52            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
53            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
54            Field::new(KEY_IDENTIFIER, DataType::Utf8, true),
55        ];
56
57        Schema::new_with_metadata(fields, metadata_with_type_name("OrderBookDelta", metadata))
58    }
59}
60
61impl EncodeToRecordBatch for OrderBookDelta {
62    fn encode_batch<T>(
63        metadata: &HashMap<String, String>,
64        data: &[T],
65    ) -> Result<RecordBatch, ArrowError>
66    where
67        T: std::borrow::Borrow<Self>,
68    {
69        let mut order_id_builder = UInt64Array::builder(data.len());
70        let mut flags_builder = UInt8Array::builder(data.len());
71        let mut sequence_builder = UInt64Array::builder(data.len());
72        let mut ts_event_builder = UInt64Array::builder(data.len());
73        let mut ts_init_builder = UInt64Array::builder(data.len());
74
75        for delta in data.iter().map(std::borrow::Borrow::borrow) {
76            order_id_builder.append_value(delta.order.order_id);
77            flags_builder.append_value(delta.flags);
78            sequence_builder.append_value(delta.sequence);
79            ts_event_builder.append_value(delta.ts_event.as_u64());
80            ts_init_builder.append_value(delta.ts_init.as_u64());
81        }
82
83        let action_array = enum_dictionary_array(
84            data.iter()
85                .map(std::borrow::Borrow::borrow)
86                .map(|delta| delta.action),
87        )?;
88        let side_array =
89            enum_dictionary_array(data.iter().map(std::borrow::Borrow::borrow).map(|delta| {
90                delta
91                    .order
92                    .side
93                    .map_or_else(|| "NO_ORDER_SIDE".to_string(), |side| side.to_string())
94            }))?;
95        let price_array = price_decimal_array(
96            data.iter()
97                .map(std::borrow::Borrow::borrow)
98                .map(|delta| delta.order.price.raw()),
99            "price",
100        )?;
101        let size_array = quantity_decimal_array(
102            data.iter()
103                .map(std::borrow::Borrow::borrow)
104                .map(|delta| delta.order.size.raw()),
105            "size",
106        )?;
107        let order_id_array = order_id_builder.finish();
108        let flags_array = flags_builder.finish();
109        let sequence_array = sequence_builder.finish();
110        let ts_event_array = ts_event_builder.finish();
111        let ts_init_array = ts_init_builder.finish();
112
113        crate::arrow::record_batch_with_timestamps(
114            Self::get_schema(Some(metadata.clone())).into(),
115            vec![
116                Arc::new(action_array),
117                Arc::new(side_array),
118                Arc::new(price_array),
119                Arc::new(size_array),
120                Arc::new(order_id_array),
121                Arc::new(flags_array),
122                Arc::new(sequence_array),
123                Arc::new(ts_event_array),
124                Arc::new(ts_init_array),
125                Arc::new(identifier_array_from_display(
126                    data.iter()
127                        .map(std::borrow::Borrow::borrow)
128                        .map(|delta| delta.instrument_id),
129                )),
130            ],
131        )
132    }
133
134    fn metadata(&self) -> HashMap<String, String> {
135        Self::get_metadata(
136            &self.instrument_id,
137            self.order.price.precision,
138            self.order.size.precision,
139        )
140    }
141
142    /// Extracts metadata from the first non-clear delta, falling back to the first clear.
143    fn chunk_metadata<T>(chunk: &[T]) -> HashMap<String, String>
144    where
145        T: std::borrow::Borrow<Self>,
146    {
147        chunk
148            .iter()
149            .map(std::borrow::Borrow::borrow)
150            .find(|delta| delta.action != BookAction::Clear)
151            .or_else(|| chunk.first().map(std::borrow::Borrow::borrow))
152            .map(EncodeToRecordBatch::metadata)
153            .expect("Chunk must contain at least one element to encode")
154    }
155
156    fn matches_chunk_metadata(&self, metadata: &HashMap<String, String>) -> bool {
157        if self.action != BookAction::Clear {
158            return self.metadata() == *metadata;
159        }
160
161        parse_metadata(metadata)
162            .is_ok_and(|(instrument_id, _, _)| self.instrument_id == instrument_id)
163    }
164}
165
166impl DecodeFromRecordBatch for OrderBookDelta {
167    fn decode_batch(
168        metadata: &HashMap<String, String>,
169        record_batch: RecordBatch,
170    ) -> Result<Vec<Self>, EncodingError> {
171        let (instrument_id, price_precision, size_precision) = parse_metadata(metadata)?;
172        let record_batch = crate::arrow::record_batch_with_u64_timestamps(&record_batch)?;
173        let record_batch = &record_batch;
174        let cols = record_batch.columns();
175
176        let action_values = extract_column_string(cols, "action", 0)?;
177        let side_values = extract_column_string(cols, "side", 1)?;
178        validate_required_column(cols[0].as_ref(), "action")?;
179        validate_required_column(cols[1].as_ref(), "side")?;
180        let price_values =
181            extract_column::<Decimal128Array>(cols, "price", 2, fixed_decimal_data_type())?;
182        let size_values =
183            extract_column::<Decimal128Array>(cols, "size", 3, fixed_decimal_data_type())?;
184        let order_id_values = extract_column::<UInt64Array>(cols, "order_id", 4, DataType::UInt64)?;
185        let flags_values = extract_column::<UInt8Array>(cols, "flags", 5, DataType::UInt8)?;
186        let sequence_values = extract_column::<UInt64Array>(cols, "sequence", 6, DataType::UInt64)?;
187        let ts_event_values = extract_column::<UInt64Array>(cols, "ts_event", 7, DataType::UInt64)?;
188        let ts_init_values = extract_column::<UInt64Array>(cols, "ts_init", 8, DataType::UInt64)?;
189
190        let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
191            .map(|i| {
192                let action_value = action_values.value(i);
193                let action = BookAction::from_str(action_value).map_err(|e| {
194                    EncodingError::ParseError(stringify!(BookAction), e.to_string())
195                })?;
196                let side_value = side_values.value(i);
197                let side = if side_value.eq_ignore_ascii_case("NO_ORDER_SIDE") {
198                    None
199                } else {
200                    Some(OrderSide::from_str(side_value).map_err(|e| {
201                        EncodingError::ParseError(stringify!(OrderSide), e.to_string())
202                    })?)
203                };
204                let price = decode_decimal_price(price_values, price_precision, "price", i)?;
205                let size = decode_decimal_quantity(size_values, size_precision, "size", i)?;
206                let order_id = decode_required_u64(order_id_values, "order_id", i)?;
207                let flags = decode_required_u8(flags_values, "flags", i)?;
208                let sequence = decode_required_u64(sequence_values, "sequence", i)?;
209                let ts_event = decode_required_timestamp(ts_event_values, "ts_event", i)?;
210                let ts_init = decode_required_timestamp(ts_init_values, "ts_init", i)?;
211
212                Ok(Self {
213                    instrument_id,
214                    action,
215                    order: BookOrder {
216                        side,
217                        price,
218                        size,
219                        order_id,
220                    },
221                    flags,
222                    sequence,
223                    ts_event,
224                    ts_init,
225                })
226            })
227            .collect();
228
229        result
230    }
231}
232
233impl DecodeDataFromRecordBatch for OrderBookDelta {
234    fn decode_data_batch(
235        metadata: &HashMap<String, String>,
236        record_batch: RecordBatch,
237    ) -> Result<Vec<Data>, EncodingError> {
238        let deltas: Vec<Self> = Self::decode_batch(metadata, record_batch)?;
239        Ok(deltas.into_iter().map(Data::from).collect())
240    }
241}
242
243#[cfg(test)]
244mod tests {
245    use std::sync::Arc;
246
247    use arrow::array::{Array, ArrayRef, TimestampNanosecondArray};
248    use nautilus_model::types::{
249        Price, Quantity,
250        fixed::FIXED_SCALAR,
251        price::{PRICE_UNDEF, PriceRaw},
252        quantity::{QUANTITY_UNDEF, QuantityRaw},
253    };
254    use pretty_assertions::assert_eq;
255    use rstest::rstest;
256
257    use super::*;
258    use crate::arrow::{KEY_TYPE_NAME, get_raw_price};
259
260    #[rstest]
261    fn test_get_schema() {
262        let instrument_id = InstrumentId::from("AAPL.XNAS");
263        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
264        let schema = OrderBookDelta::get_schema(Some(metadata.clone()));
265
266        let expected_fields = vec![
267            Field::new("action", enum_dictionary_data_type(), false),
268            Field::new("side", enum_dictionary_data_type(), false),
269            Field::new("price", fixed_decimal_data_type(), true),
270            Field::new("size", fixed_decimal_data_type(), true),
271            Field::new("order_id", DataType::UInt64, false),
272            Field::new("flags", DataType::UInt8, false),
273            Field::new("sequence", DataType::UInt64, false),
274            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
275            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
276            Field::new(KEY_IDENTIFIER, DataType::Utf8, true),
277        ];
278
279        let mut expected_metadata = metadata;
280        expected_metadata.insert(KEY_TYPE_NAME.to_string(), "OrderBookDelta".to_string());
281        let expected_schema = Schema::new_with_metadata(expected_fields, expected_metadata);
282        assert_eq!(schema, expected_schema);
283    }
284
285    #[rstest]
286    fn test_get_schema_map() {
287        let schema_map = OrderBookDelta::get_schema_map();
288        let fixed_size_binary = "Decimal128(38, 16)".to_string();
289
290        assert_eq!(schema_map.get("action").unwrap(), "Dictionary(Int8, Utf8)");
291        assert_eq!(schema_map.get("side").unwrap(), "Dictionary(Int8, Utf8)");
292        assert_eq!(*schema_map.get("price").unwrap(), fixed_size_binary);
293        assert_eq!(*schema_map.get("size").unwrap(), fixed_size_binary);
294        assert_eq!(schema_map.get("order_id").unwrap(), "UInt64");
295        assert_eq!(schema_map.get("flags").unwrap(), "UInt8");
296        assert_eq!(schema_map.get("sequence").unwrap(), "UInt64");
297        assert_eq!(
298            schema_map.get("ts_event").unwrap(),
299            "Timestamp(Nanosecond, Some(\"UTC\"))"
300        );
301        assert_eq!(
302            schema_map.get("ts_init").unwrap(),
303            "Timestamp(Nanosecond, Some(\"UTC\"))"
304        );
305        assert_eq!(schema_map.get(KEY_IDENTIFIER).unwrap(), "Utf8");
306    }
307
308    #[rstest]
309    fn clear_delta_rejects_other_instrument_chunk_metadata() {
310        let delta = OrderBookDelta::clear(InstrumentId::from("AAPL.XNAS"), 0, 1.into(), 1.into());
311        let metadata = OrderBookDelta::get_metadata(&InstrumentId::from("MSFT.XNAS"), 2, 0);
312
313        assert!(!delta.matches_chunk_metadata(&metadata));
314    }
315
316    #[rstest]
317    fn clear_delta_rejects_chunk_metadata_without_price_precision() {
318        let instrument_id = InstrumentId::from("AAPL.XNAS");
319        let delta = OrderBookDelta::clear(instrument_id, 0, 1.into(), 1.into());
320        let mut metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
321        metadata.remove(KEY_PRICE_PRECISION);
322
323        assert!(!delta.matches_chunk_metadata(&metadata));
324    }
325
326    #[rstest]
327    fn test_encode_batch() {
328        let instrument_id = InstrumentId::from("AAPL.XNAS");
329        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
330
331        let delta1 = OrderBookDelta {
332            instrument_id,
333            action: BookAction::Add,
334            order: BookOrder {
335                side: OrderSide::Buy.into(),
336                price: Price::from("100.10"),
337                size: Quantity::from(100),
338                order_id: 1,
339            },
340            flags: 0,
341            sequence: 1,
342            ts_event: 1.into(),
343            ts_init: 3.into(),
344        };
345
346        let delta2 = OrderBookDelta {
347            instrument_id,
348            action: BookAction::Update,
349            order: BookOrder {
350                side: OrderSide::Sell.into(),
351                price: Price::from("101.20"),
352                size: Quantity::from(200),
353                order_id: 2,
354            },
355            flags: 1,
356            sequence: 2,
357            ts_event: 2.into(),
358            ts_init: 4.into(),
359        };
360
361        let data = vec![delta1, delta2];
362        let record_batch = OrderBookDelta::encode_batch(&metadata, &data).unwrap();
363
364        let columns = record_batch.columns();
365        let action_values = extract_column_string(columns, "action", 0).unwrap();
366        let side_values = extract_column_string(columns, "side", 1).unwrap();
367        let price_values = columns[2]
368            .as_any()
369            .downcast_ref::<Decimal128Array>()
370            .unwrap();
371        let size_values = columns[3]
372            .as_any()
373            .downcast_ref::<Decimal128Array>()
374            .unwrap();
375        let order_id_values = columns[4].as_any().downcast_ref::<UInt64Array>().unwrap();
376        let flags_values = columns[5].as_any().downcast_ref::<UInt8Array>().unwrap();
377        let sequence_values = columns[6].as_any().downcast_ref::<UInt64Array>().unwrap();
378        let ts_event_values = columns[7]
379            .as_any()
380            .downcast_ref::<TimestampNanosecondArray>()
381            .unwrap();
382        let ts_init_values = columns[8]
383            .as_any()
384            .downcast_ref::<TimestampNanosecondArray>()
385            .unwrap();
386
387        assert_eq!(columns.len(), 10);
388        assert_eq!(action_values.len(), 2);
389        assert_eq!(action_values.value(0), "ADD");
390        assert_eq!(action_values.value(1), "UPDATE");
391        assert_eq!(side_values.len(), 2);
392        assert_eq!(side_values.value(0), "BUY");
393        assert_eq!(side_values.value(1), "SELL");
394
395        assert_eq!(price_values.len(), 2);
396        assert_eq!(
397            get_raw_price(price_values.value(0)),
398            (100.10 * FIXED_SCALAR) as PriceRaw
399        );
400        assert_eq!(
401            get_raw_price(price_values.value(1)),
402            (101.20 * FIXED_SCALAR) as PriceRaw
403        );
404
405        assert_eq!(size_values.len(), 2);
406        assert_eq!(
407            get_raw_price(size_values.value(0)),
408            (100.0 * FIXED_SCALAR) as PriceRaw
409        );
410        assert_eq!(
411            get_raw_price(size_values.value(1)),
412            (200.0 * FIXED_SCALAR) as PriceRaw
413        );
414        assert_eq!(order_id_values.len(), 2);
415        assert_eq!(order_id_values.value(0), 1);
416        assert_eq!(order_id_values.value(1), 2);
417        assert_eq!(flags_values.len(), 2);
418        assert_eq!(flags_values.value(0), 0);
419        assert_eq!(flags_values.value(1), 1);
420        assert_eq!(sequence_values.len(), 2);
421        assert_eq!(sequence_values.value(0), 1);
422        assert_eq!(sequence_values.value(1), 2);
423        assert_eq!(ts_event_values.len(), 2);
424        assert_eq!(ts_event_values.value(0), 1);
425        assert_eq!(ts_event_values.value(1), 2);
426        assert_eq!(ts_init_values.len(), 2);
427        assert_eq!(ts_init_values.value(0), 3);
428        assert_eq!(ts_init_values.value(1), 4);
429    }
430
431    #[rstest]
432    fn test_decode_batch() {
433        let instrument_id = InstrumentId::from("AAPL.XNAS");
434        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
435
436        let action = enum_dictionary_array([BookAction::Add, BookAction::Update]).unwrap();
437        let side = enum_dictionary_array([OrderSide::Buy, OrderSide::Buy]).unwrap();
438        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
439            &((101.10 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
440            &((101.20 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
441        ]);
442        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
443            &((10000.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
444            &((9000.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
445        ]);
446        let order_id = UInt64Array::from(vec![1, 2]);
447        let flags = UInt8Array::from(vec![0, 0]);
448        let sequence = UInt64Array::from(vec![1, 2]);
449        let ts_event = UInt64Array::from(vec![1, 2]);
450        let ts_init = UInt64Array::from(vec![3, 4]);
451
452        let record_batch = crate::arrow::record_batch_with_timestamps(
453            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
454                metadata.clone(),
455            )))
456            .into(),
457            vec![
458                Arc::new(action),
459                Arc::new(side),
460                Arc::new(price),
461                Arc::new(size),
462                Arc::new(order_id),
463                Arc::new(flags),
464                Arc::new(sequence),
465                Arc::new(ts_event),
466                Arc::new(ts_init),
467            ],
468        )
469        .unwrap();
470
471        let decoded_data = OrderBookDelta::decode_batch(&metadata, record_batch).unwrap();
472        assert_eq!(decoded_data.len(), 2);
473    }
474
475    #[rstest]
476    fn test_decode_batch_rejects_null_action_with_field_and_row() {
477        use arrow::{array::StringDictionaryBuilder, datatypes::Int8Type};
478
479        let instrument_id = InstrumentId::from("AAPL.XNAS");
480        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
481        let delta = OrderBookDelta::clear(instrument_id, 0, 1.into(), 2.into());
482        let encoded = OrderBookDelta::encode_batch(&metadata, &[delta]).unwrap();
483        let mut action = StringDictionaryBuilder::<Int8Type>::new();
484        action.append_null();
485        let mut columns = encoded.columns().to_vec();
486        columns[0] = Arc::new(action.finish());
487
488        let fields = encoded
489            .schema()
490            .fields()
491            .iter()
492            .map(|field| {
493                if field.name() == "action" {
494                    Arc::new(field.as_ref().clone().with_nullable(true))
495                } else {
496                    Arc::clone(field)
497                }
498            })
499            .collect::<Vec<_>>();
500
501        let schema = Arc::new(Schema::new_with_metadata(fields, metadata.clone()));
502        let record_batch = RecordBatch::try_new(schema, columns).unwrap();
503
504        let result = OrderBookDelta::decode_batch(&metadata, record_batch);
505
506        assert!(matches!(
507            result,
508            Err(EncodingError::ParseError("action", ref message))
509                if message == "row 0: required value is null"
510        ));
511    }
512
513    #[rstest]
514    fn test_decode_batch_rejects_null_timestamp_with_field_and_row() {
515        let instrument_id = InstrumentId::from("AAPL.XNAS");
516        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
517        let delta = OrderBookDelta {
518            instrument_id,
519            action: BookAction::Add,
520            order: BookOrder {
521                side: OrderSide::Buy.into(),
522                price: Price::from("100.10"),
523                size: Quantity::from(100),
524                order_id: 1,
525            },
526            flags: 0,
527            sequence: 1,
528            ts_event: 1.into(),
529            ts_init: 2.into(),
530        };
531        let encoded = OrderBookDelta::encode_batch(&metadata, &[delta]).unwrap();
532        let mut columns = encoded.columns().to_vec();
533        columns[8] = Arc::new(TimestampNanosecondArray::from(vec![None]).with_timezone("UTC"));
534        let fields = encoded
535            .schema()
536            .fields()
537            .iter()
538            .map(|field| {
539                if field.name() == "ts_init" {
540                    Arc::new(field.as_ref().clone().with_nullable(true))
541                } else {
542                    Arc::clone(field)
543                }
544            })
545            .collect::<Vec<_>>();
546        let schema = Arc::new(Schema::new_with_metadata(fields, metadata.clone()));
547        let batch = RecordBatch::try_new(schema, columns).unwrap();
548
549        let error = OrderBookDelta::decode_batch(&metadata, batch).unwrap_err();
550
551        assert!(error.to_string().contains("ts_init"));
552        assert!(error.to_string().contains("row 0"));
553    }
554
555    #[rstest]
556    fn test_decode_batch_rejects_null_required_integers_with_field_and_row() {
557        let instrument_id = InstrumentId::from("AAPL.XNAS");
558        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
559        let delta = OrderBookDelta {
560            instrument_id,
561            action: BookAction::Add,
562            order: BookOrder::new(
563                OrderSide::Buy,
564                Price::from("100.10"),
565                Quantity::from(100),
566                1,
567            ),
568            flags: 0,
569            sequence: 1,
570            ts_event: 1.into(),
571            ts_init: 2.into(),
572        };
573        let encoded = OrderBookDelta::encode_batch(&metadata, &[delta]).unwrap();
574        let corruptions: [(usize, &str, ArrayRef); 3] = [
575            (4, "order_id", Arc::new(UInt64Array::from(vec![None]))),
576            (5, "flags", Arc::new(UInt8Array::from(vec![None]))),
577            (6, "sequence", Arc::new(UInt64Array::from(vec![None]))),
578        ];
579
580        for (index, field, column) in corruptions {
581            let mut columns = encoded.columns().to_vec();
582            columns[index] = column;
583            let fields = encoded
584                .schema()
585                .fields()
586                .iter()
587                .map(|schema_field| {
588                    if schema_field.name() == field {
589                        Arc::new(schema_field.as_ref().clone().with_nullable(true))
590                    } else {
591                        Arc::clone(schema_field)
592                    }
593                })
594                .collect::<Vec<_>>();
595            let schema = Arc::new(Schema::new_with_metadata(fields, metadata.clone()));
596            let batch = RecordBatch::try_new(schema, columns).unwrap();
597
598            let error = OrderBookDelta::decode_batch(&metadata, batch).unwrap_err();
599            assert!(error.to_string().contains(field));
600            assert!(error.to_string().contains("row 0"));
601        }
602    }
603
604    #[rstest]
605    fn test_decode_batch_with_undef_values() {
606        let instrument_id = InstrumentId::from("PLTR.XNAS");
607        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
608
609        // Create test data with 'R' (clear) action which has PRICE_UNDEF and QUANTITY_UNDEF
610        let action = enum_dictionary_array([BookAction::Clear, BookAction::Add]).unwrap();
611        let side = enum_dictionary_array(["NO_ORDER_SIDE", "BUY"]).unwrap();
612        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
613            &PRICE_UNDEF.to_le_bytes(),
614            &((100.50 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
615        ]);
616        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
617            &QUANTITY_UNDEF.to_le_bytes(),
618            &((1000.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
619        ]);
620        let order_id = UInt64Array::from(vec![0, 1]);
621        let flags = UInt8Array::from(vec![0, 0]);
622        let sequence = UInt64Array::from(vec![1, 2]);
623        let ts_event = UInt64Array::from(vec![1, 2]);
624        let ts_init = UInt64Array::from(vec![3, 4]);
625
626        let record_batch = crate::arrow::record_batch_with_timestamps(
627            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
628                metadata.clone(),
629            )))
630            .into(),
631            vec![
632                Arc::new(action),
633                Arc::new(side),
634                Arc::new(price),
635                Arc::new(size),
636                Arc::new(order_id),
637                Arc::new(flags),
638                Arc::new(sequence),
639                Arc::new(ts_event),
640                Arc::new(ts_init),
641            ],
642        )
643        .unwrap();
644
645        let decoded_data = OrderBookDelta::decode_batch(&metadata, record_batch).unwrap();
646        assert_eq!(decoded_data.len(), 2);
647        assert_eq!(decoded_data[0].order.price.raw(), PRICE_UNDEF);
648        assert_eq!(decoded_data[0].order.price.precision, 0);
649        assert_eq!(decoded_data[0].order.size.raw(), QUANTITY_UNDEF);
650        assert_eq!(decoded_data[0].order.size.precision, 0);
651        assert_eq!(decoded_data[1].order.price.precision, 2);
652        assert_eq!(decoded_data[1].order.size.precision, 0);
653    }
654
655    #[rstest]
656    fn test_decode_batch_invalid_price_returns_error() {
657        let instrument_id = InstrumentId::from("AAPL.XNAS");
658        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
659
660        let action = enum_dictionary_array([BookAction::Add]).unwrap();
661        let side = enum_dictionary_array([OrderSide::Buy]).unwrap();
662
663        let invalid_price: PriceRaw = PriceRaw::MAX - 1000;
664        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
665            &invalid_price.to_le_bytes(),
666        ]);
667        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
668            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
669        ]);
670        let order_id = UInt64Array::from(vec![1]);
671        let flags = UInt8Array::from(vec![0]);
672        let sequence = UInt64Array::from(vec![1]);
673        let ts_event = UInt64Array::from(vec![1]);
674        let ts_init = UInt64Array::from(vec![2]);
675
676        let record_batch = crate::arrow::record_batch_with_timestamps(
677            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
678                metadata.clone(),
679            )))
680            .into(),
681            vec![
682                Arc::new(action),
683                Arc::new(side),
684                Arc::new(price),
685                Arc::new(size),
686                Arc::new(order_id),
687                Arc::new(flags),
688                Arc::new(sequence),
689                Arc::new(ts_event),
690                Arc::new(ts_init),
691            ],
692        )
693        .unwrap();
694
695        let result = OrderBookDelta::decode_batch(&metadata, record_batch);
696        assert!(result.is_err());
697        let err = result.unwrap_err();
698        assert!(
699            err.to_string().contains("price") && err.to_string().contains("row 0"),
700            "Expected price error at row 0, was: {err}"
701        );
702    }
703
704    #[rstest]
705    fn test_decode_batch_invalid_action_returns_error() {
706        let instrument_id = InstrumentId::from("AAPL.XNAS");
707        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
708
709        let action = enum_dictionary_array(["INVALID"]).unwrap();
710        let side = enum_dictionary_array([OrderSide::Buy]).unwrap();
711        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
712            &((100.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
713        ]);
714        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
715            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
716        ]);
717        let order_id = UInt64Array::from(vec![1]);
718        let flags = UInt8Array::from(vec![0]);
719        let sequence = UInt64Array::from(vec![1]);
720        let ts_event = UInt64Array::from(vec![1]);
721        let ts_init = UInt64Array::from(vec![2]);
722
723        let record_batch = crate::arrow::record_batch_with_timestamps(
724            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
725                metadata.clone(),
726            )))
727            .into(),
728            vec![
729                Arc::new(action),
730                Arc::new(side),
731                Arc::new(price),
732                Arc::new(size),
733                Arc::new(order_id),
734                Arc::new(flags),
735                Arc::new(sequence),
736                Arc::new(ts_event),
737                Arc::new(ts_init),
738            ],
739        )
740        .unwrap();
741
742        let result = OrderBookDelta::decode_batch(&metadata, record_batch);
743        assert!(result.is_err());
744        let err = result.unwrap_err();
745        assert!(
746            err.to_string().contains("BookAction"),
747            "Expected BookAction error, was: {err}"
748        );
749    }
750
751    #[rstest]
752    fn test_decode_batch_missing_instrument_id_returns_error() {
753        let instrument_id = InstrumentId::from("AAPL.XNAS");
754        let mut metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
755        metadata.remove(KEY_INSTRUMENT_ID);
756
757        let action = enum_dictionary_array([BookAction::Add]).unwrap();
758        let side = enum_dictionary_array([OrderSide::Buy]).unwrap();
759        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
760            &((100.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
761        ]);
762        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
763            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
764        ]);
765        let order_id = UInt64Array::from(vec![1]);
766        let flags = UInt8Array::from(vec![0]);
767        let sequence = UInt64Array::from(vec![1]);
768        let ts_event = UInt64Array::from(vec![1]);
769        let ts_init = UInt64Array::from(vec![2]);
770
771        let record_batch = crate::arrow::record_batch_with_timestamps(
772            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
773                metadata.clone(),
774            )))
775            .into(),
776            vec![
777                Arc::new(action),
778                Arc::new(side),
779                Arc::new(price),
780                Arc::new(size),
781                Arc::new(order_id),
782                Arc::new(flags),
783                Arc::new(sequence),
784                Arc::new(ts_event),
785                Arc::new(ts_init),
786            ],
787        )
788        .unwrap();
789
790        let result = OrderBookDelta::decode_batch(&metadata, record_batch);
791        assert!(result.is_err());
792        let err = result.unwrap_err();
793        assert!(
794            err.to_string().contains("instrument_id"),
795            "Expected missing instrument_id error, was: {err}"
796        );
797    }
798
799    #[rstest]
800    fn test_encode_decode_round_trip() {
801        let instrument_id = InstrumentId::from("AAPL.XNAS");
802        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
803
804        let delta1 = OrderBookDelta {
805            instrument_id,
806            action: BookAction::Add,
807            order: BookOrder {
808                side: OrderSide::Buy.into(),
809                price: Price::from("100.10"),
810                size: Quantity::from(100),
811                order_id: 1,
812            },
813            flags: 0,
814            sequence: 1,
815            ts_event: 1_000_000_000.into(),
816            ts_init: 1_000_000_001.into(),
817        };
818
819        let delta2 = OrderBookDelta {
820            instrument_id,
821            action: BookAction::Update,
822            order: BookOrder {
823                side: OrderSide::Sell.into(),
824                price: Price::from("101.20"),
825                size: Quantity::from(200),
826                order_id: 2,
827            },
828            flags: 1,
829            sequence: 2,
830            ts_event: 2_000_000_000.into(),
831            ts_init: 2_000_000_001.into(),
832        };
833
834        let original = vec![delta1, delta2];
835        let record_batch = OrderBookDelta::encode_batch(&metadata, &original).unwrap();
836        let decoded = OrderBookDelta::decode_batch(&metadata, record_batch).unwrap();
837
838        assert_eq!(decoded.len(), original.len());
839        for (orig, dec) in original.iter().zip(decoded.iter()) {
840            assert_eq!(dec.instrument_id, orig.instrument_id);
841            assert_eq!(dec.action, orig.action);
842            assert_eq!(dec.order.side, orig.order.side);
843            assert_eq!(dec.order.price, orig.order.price);
844            assert_eq!(dec.order.size, orig.order.size);
845            assert_eq!(dec.order.order_id, orig.order.order_id);
846            assert_eq!(dec.flags, orig.flags);
847            assert_eq!(dec.sequence, orig.sequence);
848            assert_eq!(dec.ts_event, orig.ts_event);
849            assert_eq!(dec.ts_init, orig.ts_init);
850        }
851    }
852}
nautilus_serialization/arrow/delta.rs

nautilus_serialization/arrow/
delta.rs
