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,
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        let price_values =
179            extract_column::<Decimal128Array>(cols, "price", 2, fixed_decimal_data_type())?;
180        let size_values =
181            extract_column::<Decimal128Array>(cols, "size", 3, fixed_decimal_data_type())?;
182        let order_id_values = extract_column::<UInt64Array>(cols, "order_id", 4, DataType::UInt64)?;
183        let flags_values = extract_column::<UInt8Array>(cols, "flags", 5, DataType::UInt8)?;
184        let sequence_values = extract_column::<UInt64Array>(cols, "sequence", 6, DataType::UInt64)?;
185        let ts_event_values = extract_column::<UInt64Array>(cols, "ts_event", 7, DataType::UInt64)?;
186        let ts_init_values = extract_column::<UInt64Array>(cols, "ts_init", 8, DataType::UInt64)?;
187
188        let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
189            .map(|i| {
190                let action_value = action_values.value(i);
191                let action = BookAction::from_str(action_value).map_err(|e| {
192                    EncodingError::ParseError(stringify!(BookAction), e.to_string())
193                })?;
194                let side_value = side_values.value(i);
195                let side = if side_value.eq_ignore_ascii_case("NO_ORDER_SIDE") {
196                    None
197                } else {
198                    Some(OrderSide::from_str(side_value).map_err(|e| {
199                        EncodingError::ParseError(stringify!(OrderSide), e.to_string())
200                    })?)
201                };
202                let price = decode_decimal_price(price_values, price_precision, "price", i)?;
203                let size = decode_decimal_quantity(size_values, size_precision, "size", i)?;
204                let order_id = decode_required_u64(order_id_values, "order_id", i)?;
205                let flags = decode_required_u8(flags_values, "flags", i)?;
206                let sequence = decode_required_u64(sequence_values, "sequence", i)?;
207                let ts_event = decode_required_timestamp(ts_event_values, "ts_event", i)?;
208                let ts_init = decode_required_timestamp(ts_init_values, "ts_init", i)?;
209
210                Ok(Self {
211                    instrument_id,
212                    action,
213                    order: BookOrder {
214                        side,
215                        price,
216                        size,
217                        order_id,
218                    },
219                    flags,
220                    sequence,
221                    ts_event,
222                    ts_init,
223                })
224            })
225            .collect();
226
227        result
228    }
229}
230
231impl DecodeDataFromRecordBatch for OrderBookDelta {
232    fn decode_data_batch(
233        metadata: &HashMap<String, String>,
234        record_batch: RecordBatch,
235    ) -> Result<Vec<Data>, EncodingError> {
236        let deltas: Vec<Self> = Self::decode_batch(metadata, record_batch)?;
237        Ok(deltas.into_iter().map(Data::from).collect())
238    }
239}
240
241#[cfg(test)]
242mod tests {
243    use std::sync::Arc;
244
245    use arrow::array::{Array, ArrayRef, TimestampNanosecondArray};
246    use nautilus_model::types::{
247        Price, Quantity,
248        fixed::FIXED_SCALAR,
249        price::{PRICE_UNDEF, PriceRaw},
250        quantity::{QUANTITY_UNDEF, QuantityRaw},
251    };
252    use pretty_assertions::assert_eq;
253    use rstest::rstest;
254
255    use super::*;
256    use crate::arrow::{KEY_TYPE_NAME, get_raw_price};
257
258    #[rstest]
259    fn test_get_schema() {
260        let instrument_id = InstrumentId::from("AAPL.XNAS");
261        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
262        let schema = OrderBookDelta::get_schema(Some(metadata.clone()));
263
264        let expected_fields = vec![
265            Field::new("action", enum_dictionary_data_type(), false),
266            Field::new("side", enum_dictionary_data_type(), false),
267            Field::new("price", fixed_decimal_data_type(), true),
268            Field::new("size", fixed_decimal_data_type(), true),
269            Field::new("order_id", DataType::UInt64, false),
270            Field::new("flags", DataType::UInt8, false),
271            Field::new("sequence", DataType::UInt64, false),
272            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
273            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
274            Field::new(KEY_IDENTIFIER, DataType::Utf8, true),
275        ];
276
277        let mut expected_metadata = metadata;
278        expected_metadata.insert(KEY_TYPE_NAME.to_string(), "OrderBookDelta".to_string());
279        let expected_schema = Schema::new_with_metadata(expected_fields, expected_metadata);
280        assert_eq!(schema, expected_schema);
281    }
282
283    #[rstest]
284    fn test_get_schema_map() {
285        let schema_map = OrderBookDelta::get_schema_map();
286        let fixed_size_binary = "Decimal128(38, 16)".to_string();
287
288        assert_eq!(schema_map.get("action").unwrap(), "Dictionary(Int8, Utf8)");
289        assert_eq!(schema_map.get("side").unwrap(), "Dictionary(Int8, Utf8)");
290        assert_eq!(*schema_map.get("price").unwrap(), fixed_size_binary);
291        assert_eq!(*schema_map.get("size").unwrap(), fixed_size_binary);
292        assert_eq!(schema_map.get("order_id").unwrap(), "UInt64");
293        assert_eq!(schema_map.get("flags").unwrap(), "UInt8");
294        assert_eq!(schema_map.get("sequence").unwrap(), "UInt64");
295        assert_eq!(
296            schema_map.get("ts_event").unwrap(),
297            "Timestamp(Nanosecond, Some(\"UTC\"))"
298        );
299        assert_eq!(
300            schema_map.get("ts_init").unwrap(),
301            "Timestamp(Nanosecond, Some(\"UTC\"))"
302        );
303        assert_eq!(schema_map.get(KEY_IDENTIFIER).unwrap(), "Utf8");
304    }
305
306    #[rstest]
307    fn clear_delta_rejects_other_instrument_chunk_metadata() {
308        let delta = OrderBookDelta::clear(InstrumentId::from("AAPL.XNAS"), 0, 1.into(), 1.into());
309        let metadata = OrderBookDelta::get_metadata(&InstrumentId::from("MSFT.XNAS"), 2, 0);
310
311        assert!(!delta.matches_chunk_metadata(&metadata));
312    }
313
314    #[rstest]
315    fn clear_delta_rejects_chunk_metadata_without_price_precision() {
316        let instrument_id = InstrumentId::from("AAPL.XNAS");
317        let delta = OrderBookDelta::clear(instrument_id, 0, 1.into(), 1.into());
318        let mut metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
319        metadata.remove(KEY_PRICE_PRECISION);
320
321        assert!(!delta.matches_chunk_metadata(&metadata));
322    }
323
324    #[rstest]
325    fn test_encode_batch() {
326        let instrument_id = InstrumentId::from("AAPL.XNAS");
327        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
328
329        let delta1 = OrderBookDelta {
330            instrument_id,
331            action: BookAction::Add,
332            order: BookOrder {
333                side: OrderSide::Buy.into(),
334                price: Price::from("100.10"),
335                size: Quantity::from(100),
336                order_id: 1,
337            },
338            flags: 0,
339            sequence: 1,
340            ts_event: 1.into(),
341            ts_init: 3.into(),
342        };
343
344        let delta2 = OrderBookDelta {
345            instrument_id,
346            action: BookAction::Update,
347            order: BookOrder {
348                side: OrderSide::Sell.into(),
349                price: Price::from("101.20"),
350                size: Quantity::from(200),
351                order_id: 2,
352            },
353            flags: 1,
354            sequence: 2,
355            ts_event: 2.into(),
356            ts_init: 4.into(),
357        };
358
359        let data = vec![delta1, delta2];
360        let record_batch = OrderBookDelta::encode_batch(&metadata, &data).unwrap();
361
362        let columns = record_batch.columns();
363        let action_values = extract_column_string(columns, "action", 0).unwrap();
364        let side_values = extract_column_string(columns, "side", 1).unwrap();
365        let price_values = columns[2]
366            .as_any()
367            .downcast_ref::<Decimal128Array>()
368            .unwrap();
369        let size_values = columns[3]
370            .as_any()
371            .downcast_ref::<Decimal128Array>()
372            .unwrap();
373        let order_id_values = columns[4].as_any().downcast_ref::<UInt64Array>().unwrap();
374        let flags_values = columns[5].as_any().downcast_ref::<UInt8Array>().unwrap();
375        let sequence_values = columns[6].as_any().downcast_ref::<UInt64Array>().unwrap();
376        let ts_event_values = columns[7]
377            .as_any()
378            .downcast_ref::<TimestampNanosecondArray>()
379            .unwrap();
380        let ts_init_values = columns[8]
381            .as_any()
382            .downcast_ref::<TimestampNanosecondArray>()
383            .unwrap();
384
385        assert_eq!(columns.len(), 10);
386        assert_eq!(action_values.len(), 2);
387        assert_eq!(action_values.value(0), "ADD");
388        assert_eq!(action_values.value(1), "UPDATE");
389        assert_eq!(side_values.len(), 2);
390        assert_eq!(side_values.value(0), "BUY");
391        assert_eq!(side_values.value(1), "SELL");
392
393        assert_eq!(price_values.len(), 2);
394        assert_eq!(
395            get_raw_price(price_values.value(0)),
396            (100.10 * FIXED_SCALAR) as PriceRaw
397        );
398        assert_eq!(
399            get_raw_price(price_values.value(1)),
400            (101.20 * FIXED_SCALAR) as PriceRaw
401        );
402
403        assert_eq!(size_values.len(), 2);
404        assert_eq!(
405            get_raw_price(size_values.value(0)),
406            (100.0 * FIXED_SCALAR) as PriceRaw
407        );
408        assert_eq!(
409            get_raw_price(size_values.value(1)),
410            (200.0 * FIXED_SCALAR) as PriceRaw
411        );
412        assert_eq!(order_id_values.len(), 2);
413        assert_eq!(order_id_values.value(0), 1);
414        assert_eq!(order_id_values.value(1), 2);
415        assert_eq!(flags_values.len(), 2);
416        assert_eq!(flags_values.value(0), 0);
417        assert_eq!(flags_values.value(1), 1);
418        assert_eq!(sequence_values.len(), 2);
419        assert_eq!(sequence_values.value(0), 1);
420        assert_eq!(sequence_values.value(1), 2);
421        assert_eq!(ts_event_values.len(), 2);
422        assert_eq!(ts_event_values.value(0), 1);
423        assert_eq!(ts_event_values.value(1), 2);
424        assert_eq!(ts_init_values.len(), 2);
425        assert_eq!(ts_init_values.value(0), 3);
426        assert_eq!(ts_init_values.value(1), 4);
427    }
428
429    #[rstest]
430    fn test_decode_batch() {
431        let instrument_id = InstrumentId::from("AAPL.XNAS");
432        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
433
434        let action = enum_dictionary_array([BookAction::Add, BookAction::Update]).unwrap();
435        let side = enum_dictionary_array([OrderSide::Buy, OrderSide::Buy]).unwrap();
436        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
437            &((101.10 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
438            &((101.20 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
439        ]);
440        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
441            &((10000.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
442            &((9000.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
443        ]);
444        let order_id = UInt64Array::from(vec![1, 2]);
445        let flags = UInt8Array::from(vec![0, 0]);
446        let sequence = UInt64Array::from(vec![1, 2]);
447        let ts_event = UInt64Array::from(vec![1, 2]);
448        let ts_init = UInt64Array::from(vec![3, 4]);
449
450        let record_batch = crate::arrow::record_batch_with_timestamps(
451            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
452                metadata.clone(),
453            )))
454            .into(),
455            vec![
456                Arc::new(action),
457                Arc::new(side),
458                Arc::new(price),
459                Arc::new(size),
460                Arc::new(order_id),
461                Arc::new(flags),
462                Arc::new(sequence),
463                Arc::new(ts_event),
464                Arc::new(ts_init),
465            ],
466        )
467        .unwrap();
468
469        let decoded_data = OrderBookDelta::decode_batch(&metadata, record_batch).unwrap();
470        assert_eq!(decoded_data.len(), 2);
471    }
472
473    #[rstest]
474    fn test_decode_batch_rejects_null_timestamp_with_field_and_row() {
475        let instrument_id = InstrumentId::from("AAPL.XNAS");
476        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
477        let delta = OrderBookDelta {
478            instrument_id,
479            action: BookAction::Add,
480            order: BookOrder {
481                side: OrderSide::Buy.into(),
482                price: Price::from("100.10"),
483                size: Quantity::from(100),
484                order_id: 1,
485            },
486            flags: 0,
487            sequence: 1,
488            ts_event: 1.into(),
489            ts_init: 2.into(),
490        };
491        let encoded = OrderBookDelta::encode_batch(&metadata, &[delta]).unwrap();
492        let mut columns = encoded.columns().to_vec();
493        columns[8] = Arc::new(TimestampNanosecondArray::from(vec![None]).with_timezone("UTC"));
494        let fields = encoded
495            .schema()
496            .fields()
497            .iter()
498            .map(|field| {
499                if field.name() == "ts_init" {
500                    Arc::new(field.as_ref().clone().with_nullable(true))
501                } else {
502                    Arc::clone(field)
503                }
504            })
505            .collect::<Vec<_>>();
506        let schema = Arc::new(Schema::new_with_metadata(fields, metadata.clone()));
507        let batch = RecordBatch::try_new(schema, columns).unwrap();
508
509        let error = OrderBookDelta::decode_batch(&metadata, batch).unwrap_err();
510
511        assert!(error.to_string().contains("ts_init"));
512        assert!(error.to_string().contains("row 0"));
513    }
514
515    #[rstest]
516    fn test_decode_batch_rejects_null_required_integers_with_field_and_row() {
517        let instrument_id = InstrumentId::from("AAPL.XNAS");
518        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
519        let delta = OrderBookDelta {
520            instrument_id,
521            action: BookAction::Add,
522            order: BookOrder::new(
523                OrderSide::Buy,
524                Price::from("100.10"),
525                Quantity::from(100),
526                1,
527            ),
528            flags: 0,
529            sequence: 1,
530            ts_event: 1.into(),
531            ts_init: 2.into(),
532        };
533        let encoded = OrderBookDelta::encode_batch(&metadata, &[delta]).unwrap();
534        let corruptions: [(usize, &str, ArrayRef); 3] = [
535            (4, "order_id", Arc::new(UInt64Array::from(vec![None]))),
536            (5, "flags", Arc::new(UInt8Array::from(vec![None]))),
537            (6, "sequence", Arc::new(UInt64Array::from(vec![None]))),
538        ];
539
540        for (index, field, column) in corruptions {
541            let mut columns = encoded.columns().to_vec();
542            columns[index] = column;
543            let fields = encoded
544                .schema()
545                .fields()
546                .iter()
547                .map(|schema_field| {
548                    if schema_field.name() == field {
549                        Arc::new(schema_field.as_ref().clone().with_nullable(true))
550                    } else {
551                        Arc::clone(schema_field)
552                    }
553                })
554                .collect::<Vec<_>>();
555            let schema = Arc::new(Schema::new_with_metadata(fields, metadata.clone()));
556            let batch = RecordBatch::try_new(schema, columns).unwrap();
557
558            let error = OrderBookDelta::decode_batch(&metadata, batch).unwrap_err();
559            assert!(error.to_string().contains(field));
560            assert!(error.to_string().contains("row 0"));
561        }
562    }
563
564    #[rstest]
565    fn test_decode_batch_with_undef_values() {
566        let instrument_id = InstrumentId::from("PLTR.XNAS");
567        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
568
569        // Create test data with 'R' (clear) action which has PRICE_UNDEF and QUANTITY_UNDEF
570        let action = enum_dictionary_array([BookAction::Clear, BookAction::Add]).unwrap();
571        let side = enum_dictionary_array(["NO_ORDER_SIDE", "BUY"]).unwrap();
572        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
573            &PRICE_UNDEF.to_le_bytes(),
574            &((100.50 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
575        ]);
576        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
577            &QUANTITY_UNDEF.to_le_bytes(),
578            &((1000.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
579        ]);
580        let order_id = UInt64Array::from(vec![0, 1]);
581        let flags = UInt8Array::from(vec![0, 0]);
582        let sequence = UInt64Array::from(vec![1, 2]);
583        let ts_event = UInt64Array::from(vec![1, 2]);
584        let ts_init = UInt64Array::from(vec![3, 4]);
585
586        let record_batch = crate::arrow::record_batch_with_timestamps(
587            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
588                metadata.clone(),
589            )))
590            .into(),
591            vec![
592                Arc::new(action),
593                Arc::new(side),
594                Arc::new(price),
595                Arc::new(size),
596                Arc::new(order_id),
597                Arc::new(flags),
598                Arc::new(sequence),
599                Arc::new(ts_event),
600                Arc::new(ts_init),
601            ],
602        )
603        .unwrap();
604
605        let decoded_data = OrderBookDelta::decode_batch(&metadata, record_batch).unwrap();
606        assert_eq!(decoded_data.len(), 2);
607        assert_eq!(decoded_data[0].order.price.raw(), PRICE_UNDEF);
608        assert_eq!(decoded_data[0].order.price.precision, 0);
609        assert_eq!(decoded_data[0].order.size.raw(), QUANTITY_UNDEF);
610        assert_eq!(decoded_data[0].order.size.precision, 0);
611        assert_eq!(decoded_data[1].order.price.precision, 2);
612        assert_eq!(decoded_data[1].order.size.precision, 0);
613    }
614
615    #[rstest]
616    fn test_decode_batch_invalid_price_returns_error() {
617        let instrument_id = InstrumentId::from("AAPL.XNAS");
618        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
619
620        let action = enum_dictionary_array([BookAction::Add]).unwrap();
621        let side = enum_dictionary_array([OrderSide::Buy]).unwrap();
622
623        let invalid_price: PriceRaw = PriceRaw::MAX - 1000;
624        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
625            &invalid_price.to_le_bytes(),
626        ]);
627        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
628            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
629        ]);
630        let order_id = UInt64Array::from(vec![1]);
631        let flags = UInt8Array::from(vec![0]);
632        let sequence = UInt64Array::from(vec![1]);
633        let ts_event = UInt64Array::from(vec![1]);
634        let ts_init = UInt64Array::from(vec![2]);
635
636        let record_batch = crate::arrow::record_batch_with_timestamps(
637            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
638                metadata.clone(),
639            )))
640            .into(),
641            vec![
642                Arc::new(action),
643                Arc::new(side),
644                Arc::new(price),
645                Arc::new(size),
646                Arc::new(order_id),
647                Arc::new(flags),
648                Arc::new(sequence),
649                Arc::new(ts_event),
650                Arc::new(ts_init),
651            ],
652        )
653        .unwrap();
654
655        let result = OrderBookDelta::decode_batch(&metadata, record_batch);
656        assert!(result.is_err());
657        let err = result.unwrap_err();
658        assert!(
659            err.to_string().contains("price") && err.to_string().contains("row 0"),
660            "Expected price error at row 0, was: {err}"
661        );
662    }
663
664    #[rstest]
665    fn test_decode_batch_invalid_action_returns_error() {
666        let instrument_id = InstrumentId::from("AAPL.XNAS");
667        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
668
669        let action = enum_dictionary_array(["INVALID"]).unwrap();
670        let side = enum_dictionary_array([OrderSide::Buy]).unwrap();
671        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
672            &((100.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
673        ]);
674        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
675            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
676        ]);
677        let order_id = UInt64Array::from(vec![1]);
678        let flags = UInt8Array::from(vec![0]);
679        let sequence = UInt64Array::from(vec![1]);
680        let ts_event = UInt64Array::from(vec![1]);
681        let ts_init = UInt64Array::from(vec![2]);
682
683        let record_batch = crate::arrow::record_batch_with_timestamps(
684            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
685                metadata.clone(),
686            )))
687            .into(),
688            vec![
689                Arc::new(action),
690                Arc::new(side),
691                Arc::new(price),
692                Arc::new(size),
693                Arc::new(order_id),
694                Arc::new(flags),
695                Arc::new(sequence),
696                Arc::new(ts_event),
697                Arc::new(ts_init),
698            ],
699        )
700        .unwrap();
701
702        let result = OrderBookDelta::decode_batch(&metadata, record_batch);
703        assert!(result.is_err());
704        let err = result.unwrap_err();
705        assert!(
706            err.to_string().contains("BookAction"),
707            "Expected BookAction error, was: {err}"
708        );
709    }
710
711    #[rstest]
712    fn test_decode_batch_missing_instrument_id_returns_error() {
713        let instrument_id = InstrumentId::from("AAPL.XNAS");
714        let mut metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
715        metadata.remove(KEY_INSTRUMENT_ID);
716
717        let action = enum_dictionary_array([BookAction::Add]).unwrap();
718        let side = enum_dictionary_array([OrderSide::Buy]).unwrap();
719        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
720            &((100.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
721        ]);
722        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
723            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
724        ]);
725        let order_id = UInt64Array::from(vec![1]);
726        let flags = UInt8Array::from(vec![0]);
727        let sequence = UInt64Array::from(vec![1]);
728        let ts_event = UInt64Array::from(vec![1]);
729        let ts_init = UInt64Array::from(vec![2]);
730
731        let record_batch = crate::arrow::record_batch_with_timestamps(
732            crate::arrow::schema_without_identifier_column(&OrderBookDelta::get_schema(Some(
733                metadata.clone(),
734            )))
735            .into(),
736            vec![
737                Arc::new(action),
738                Arc::new(side),
739                Arc::new(price),
740                Arc::new(size),
741                Arc::new(order_id),
742                Arc::new(flags),
743                Arc::new(sequence),
744                Arc::new(ts_event),
745                Arc::new(ts_init),
746            ],
747        )
748        .unwrap();
749
750        let result = OrderBookDelta::decode_batch(&metadata, record_batch);
751        assert!(result.is_err());
752        let err = result.unwrap_err();
753        assert!(
754            err.to_string().contains("instrument_id"),
755            "Expected missing instrument_id error, was: {err}"
756        );
757    }
758
759    #[rstest]
760    fn test_encode_decode_round_trip() {
761        let instrument_id = InstrumentId::from("AAPL.XNAS");
762        let metadata = OrderBookDelta::get_metadata(&instrument_id, 2, 0);
763
764        let delta1 = OrderBookDelta {
765            instrument_id,
766            action: BookAction::Add,
767            order: BookOrder {
768                side: OrderSide::Buy.into(),
769                price: Price::from("100.10"),
770                size: Quantity::from(100),
771                order_id: 1,
772            },
773            flags: 0,
774            sequence: 1,
775            ts_event: 1_000_000_000.into(),
776            ts_init: 1_000_000_001.into(),
777        };
778
779        let delta2 = OrderBookDelta {
780            instrument_id,
781            action: BookAction::Update,
782            order: BookOrder {
783                side: OrderSide::Sell.into(),
784                price: Price::from("101.20"),
785                size: Quantity::from(200),
786                order_id: 2,
787            },
788            flags: 1,
789            sequence: 2,
790            ts_event: 2_000_000_000.into(),
791            ts_init: 2_000_000_001.into(),
792        };
793
794        let original = vec![delta1, delta2];
795        let record_batch = OrderBookDelta::encode_batch(&metadata, &original).unwrap();
796        let decoded = OrderBookDelta::decode_batch(&metadata, record_batch).unwrap();
797
798        assert_eq!(decoded.len(), original.len());
799        for (orig, dec) in original.iter().zip(decoded.iter()) {
800            assert_eq!(dec.instrument_id, orig.instrument_id);
801            assert_eq!(dec.action, orig.action);
802            assert_eq!(dec.order.side, orig.order.side);
803            assert_eq!(dec.order.price, orig.order.price);
804            assert_eq!(dec.order.size, orig.order.size);
805            assert_eq!(dec.order.order_id, orig.order.order_id);
806            assert_eq!(dec.flags, orig.flags);
807            assert_eq!(dec.sequence, orig.sequence);
808            assert_eq!(dec.ts_event, orig.ts_event);
809            assert_eq!(dec.ts_init, orig.ts_init);
810        }
811    }
812}
nautilus_serialization/arrow/delta.rs

nautilus_serialization/arrow/
delta.rs
