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

nautilus_serialization/arrow/
delta.rs
