nautilus_serialization/arrow/
depth.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, sync::Arc};
17
18use arrow::{
19    array::{
20        Array, FixedSizeBinaryArray, FixedSizeBinaryBuilder, UInt8Array, UInt32Array, UInt64Array,
21    },
22    datatypes::{DataType, Field, Schema},
23    error::ArrowError,
24    record_batch::RecordBatch,
25};
26use nautilus_model::{
27    data::{
28        depth::{DEPTH10_LEN, OrderBookDepth10},
29        order::BookOrder,
30    },
31    enums::OrderSide,
32    types::{PRICE_UNDEF, QUANTITY_UNDEF, fixed::PRECISION_BYTES},
33};
34
35use super::{
36    DecodeDataFromRecordBatch, EncodingError, decode_price, decode_quantity, extract_column,
37    get_raw_price, get_raw_quantity, parse_price_size_metadata, validate_precision_bytes,
38};
39use crate::arrow::{ArrowSchemaProvider, Data, DecodeFromRecordBatch, EncodeToRecordBatch};
40
41fn get_field_data() -> Vec<(&'static str, DataType)> {
42    vec![
43        ("bid_price", DataType::FixedSizeBinary(PRECISION_BYTES)),
44        ("ask_price", DataType::FixedSizeBinary(PRECISION_BYTES)),
45        ("bid_size", DataType::FixedSizeBinary(PRECISION_BYTES)),
46        ("ask_size", DataType::FixedSizeBinary(PRECISION_BYTES)),
47        ("bid_count", DataType::UInt32),
48        ("ask_count", DataType::UInt32),
49    ]
50}
51
52impl ArrowSchemaProvider for OrderBookDepth10 {
53    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
54        let mut fields = Vec::new();
55        let field_data = get_field_data();
56
57        // Schema is of the form:
58        // bid_price_0, bid_price_1, ..., bid_price_9, ask_price_0, ask_price_1
59        for (name, data_type) in field_data {
60            for i in 0..DEPTH10_LEN {
61                fields.push(Field::new(format!("{name}_{i}"), data_type.clone(), false));
62            }
63        }
64
65        fields.push(Field::new("flags", DataType::UInt8, false));
66        fields.push(Field::new("sequence", DataType::UInt64, false));
67        fields.push(Field::new("ts_event", DataType::UInt64, false));
68        fields.push(Field::new("ts_init", DataType::UInt64, false));
69
70        match metadata {
71            Some(metadata) => Schema::new_with_metadata(fields, metadata),
72            None => Schema::new(fields),
73        }
74    }
75}
76
77impl EncodeToRecordBatch for OrderBookDepth10 {
78    fn encode_batch(
79        metadata: &HashMap<String, String>,
80        data: &[Self],
81    ) -> Result<RecordBatch, ArrowError> {
82        let mut bid_price_builders = Vec::with_capacity(DEPTH10_LEN);
83        let mut ask_price_builders = Vec::with_capacity(DEPTH10_LEN);
84        let mut bid_size_builders = Vec::with_capacity(DEPTH10_LEN);
85        let mut ask_size_builders = Vec::with_capacity(DEPTH10_LEN);
86        let mut bid_count_builders = Vec::with_capacity(DEPTH10_LEN);
87        let mut ask_count_builders = Vec::with_capacity(DEPTH10_LEN);
88
89        for _ in 0..DEPTH10_LEN {
90            bid_price_builders.push(FixedSizeBinaryBuilder::with_capacity(
91                data.len(),
92                PRECISION_BYTES,
93            ));
94            ask_price_builders.push(FixedSizeBinaryBuilder::with_capacity(
95                data.len(),
96                PRECISION_BYTES,
97            ));
98            bid_size_builders.push(FixedSizeBinaryBuilder::with_capacity(
99                data.len(),
100                PRECISION_BYTES,
101            ));
102            ask_size_builders.push(FixedSizeBinaryBuilder::with_capacity(
103                data.len(),
104                PRECISION_BYTES,
105            ));
106            bid_count_builders.push(UInt32Array::builder(data.len()));
107            ask_count_builders.push(UInt32Array::builder(data.len()));
108        }
109
110        let mut flags_builder = UInt8Array::builder(data.len());
111        let mut sequence_builder = UInt64Array::builder(data.len());
112        let mut ts_event_builder = UInt64Array::builder(data.len());
113        let mut ts_init_builder = UInt64Array::builder(data.len());
114
115        for depth in data {
116            for i in 0..DEPTH10_LEN {
117                bid_price_builders[i]
118                    .append_value(depth.bids[i].price.raw().to_le_bytes())
119                    .unwrap();
120                ask_price_builders[i]
121                    .append_value(depth.asks[i].price.raw().to_le_bytes())
122                    .unwrap();
123                bid_size_builders[i]
124                    .append_value(depth.bids[i].size.raw().to_le_bytes())
125                    .unwrap();
126                ask_size_builders[i]
127                    .append_value(depth.asks[i].size.raw().to_le_bytes())
128                    .unwrap();
129                bid_count_builders[i].append_value(depth.bid_counts[i]);
130                ask_count_builders[i].append_value(depth.ask_counts[i]);
131            }
132
133            flags_builder.append_value(depth.flags);
134            sequence_builder.append_value(depth.sequence);
135            ts_event_builder.append_value(depth.ts_event.as_u64());
136            ts_init_builder.append_value(depth.ts_init.as_u64());
137        }
138
139        let bid_price_arrays = bid_price_builders
140            .into_iter()
141            .map(|mut b| Arc::new(b.finish()) as Arc<dyn Array>)
142            .collect::<Vec<_>>();
143        let ask_price_arrays = ask_price_builders
144            .into_iter()
145            .map(|mut b| Arc::new(b.finish()) as Arc<dyn Array>)
146            .collect::<Vec<_>>();
147        let bid_size_arrays = bid_size_builders
148            .into_iter()
149            .map(|mut b| Arc::new(b.finish()) as Arc<dyn Array>)
150            .collect::<Vec<_>>();
151        let ask_size_arrays = ask_size_builders
152            .into_iter()
153            .map(|mut b| Arc::new(b.finish()) as Arc<dyn Array>)
154            .collect::<Vec<_>>();
155        let bid_count_arrays = bid_count_builders
156            .into_iter()
157            .map(|mut b| Arc::new(b.finish()) as Arc<dyn Array>)
158            .collect::<Vec<_>>();
159        let ask_count_arrays = ask_count_builders
160            .into_iter()
161            .map(|mut b| Arc::new(b.finish()) as Arc<dyn Array>)
162            .collect::<Vec<_>>();
163
164        let flags_array = Arc::new(flags_builder.finish()) as Arc<dyn Array>;
165        let sequence_array = Arc::new(sequence_builder.finish()) as Arc<dyn Array>;
166        let ts_event_array = Arc::new(ts_event_builder.finish()) as Arc<dyn Array>;
167        let ts_init_array = Arc::new(ts_init_builder.finish()) as Arc<dyn Array>;
168
169        let mut columns = Vec::new();
170        columns.extend(bid_price_arrays);
171        columns.extend(ask_price_arrays);
172        columns.extend(bid_size_arrays);
173        columns.extend(ask_size_arrays);
174        columns.extend(bid_count_arrays);
175        columns.extend(ask_count_arrays);
176        columns.push(flags_array);
177        columns.push(sequence_array);
178        columns.push(ts_event_array);
179        columns.push(ts_init_array);
180
181        RecordBatch::try_new(Self::get_schema(Some(metadata.clone())).into(), columns)
182    }
183
184    fn metadata(&self) -> HashMap<String, String> {
185        Self::get_metadata(
186            &self.instrument_id,
187            self.bids[0].price.precision,
188            self.bids[0].size.precision,
189        )
190    }
191}
192
193impl DecodeFromRecordBatch for OrderBookDepth10 {
194    fn decode_batch(
195        metadata: &HashMap<String, String>,
196        record_batch: RecordBatch,
197    ) -> Result<Vec<Self>, EncodingError> {
198        let (instrument_id, price_precision, size_precision) = parse_price_size_metadata(metadata)?;
199        let cols = record_batch.columns();
200
201        let mut bid_prices = Vec::with_capacity(DEPTH10_LEN);
202        let mut ask_prices = Vec::with_capacity(DEPTH10_LEN);
203        let mut bid_sizes = Vec::with_capacity(DEPTH10_LEN);
204        let mut ask_sizes = Vec::with_capacity(DEPTH10_LEN);
205        let mut bid_counts = Vec::with_capacity(DEPTH10_LEN);
206        let mut ask_counts = Vec::with_capacity(DEPTH10_LEN);
207
208        macro_rules! extract_depth_column {
209            ($array:ty, $name:literal, $i:expr, $offset:expr, $type:expr) => {
210                extract_column::<$array>(cols, concat!($name, "_", stringify!($i)), $offset, $type)?
211            };
212        }
213
214        for i in 0..DEPTH10_LEN {
215            bid_prices.push(extract_depth_column!(
216                FixedSizeBinaryArray,
217                "bid_price",
218                i,
219                i,
220                DataType::FixedSizeBinary(PRECISION_BYTES)
221            ));
222            ask_prices.push(extract_depth_column!(
223                FixedSizeBinaryArray,
224                "ask_price",
225                i,
226                DEPTH10_LEN + i,
227                DataType::FixedSizeBinary(PRECISION_BYTES)
228            ));
229            bid_sizes.push(extract_depth_column!(
230                FixedSizeBinaryArray,
231                "bid_size",
232                i,
233                2 * DEPTH10_LEN + i,
234                DataType::FixedSizeBinary(PRECISION_BYTES)
235            ));
236            ask_sizes.push(extract_depth_column!(
237                FixedSizeBinaryArray,
238                "ask_size",
239                i,
240                3 * DEPTH10_LEN + i,
241                DataType::FixedSizeBinary(PRECISION_BYTES)
242            ));
243            bid_counts.push(extract_depth_column!(
244                UInt32Array,
245                "bid_count",
246                i,
247                4 * DEPTH10_LEN + i,
248                DataType::UInt32
249            ));
250            ask_counts.push(extract_depth_column!(
251                UInt32Array,
252                "ask_count",
253                i,
254                5 * DEPTH10_LEN + i,
255                DataType::UInt32
256            ));
257        }
258
259        for i in 0..DEPTH10_LEN {
260            validate_precision_bytes(bid_prices[i], "bid_price")?;
261            validate_precision_bytes(ask_prices[i], "ask_price")?;
262            validate_precision_bytes(bid_sizes[i], "bid_size")?;
263            validate_precision_bytes(ask_sizes[i], "ask_size")?;
264        }
265
266        let flags = extract_column::<UInt8Array>(cols, "flags", 6 * DEPTH10_LEN, DataType::UInt8)?;
267        let sequence =
268            extract_column::<UInt64Array>(cols, "sequence", 6 * DEPTH10_LEN + 1, DataType::UInt64)?;
269        let ts_event =
270            extract_column::<UInt64Array>(cols, "ts_event", 6 * DEPTH10_LEN + 2, DataType::UInt64)?;
271        let ts_init =
272            extract_column::<UInt64Array>(cols, "ts_init", 6 * DEPTH10_LEN + 3, DataType::UInt64)?;
273
274        // Map record batch rows to vector of OrderBookDepth10
275        let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
276            .map(|row| {
277                let mut bids = [BookOrder::default(); DEPTH10_LEN];
278                let mut asks = [BookOrder::default(); DEPTH10_LEN];
279                let mut bid_count_arr = [0u32; DEPTH10_LEN];
280                let mut ask_count_arr = [0u32; DEPTH10_LEN];
281
282                for i in 0..DEPTH10_LEN {
283                    // Undefined levels (PRICE_UNDEF / QUANTITY_UNDEF sentinels) are
284                    // decoded as NULL_ORDER so the matching engine skips them during
285                    // precision validation.
286                    let bid_price_bytes = bid_prices[i].value(row);
287                    let bid_size_bytes = bid_sizes[i].value(row);
288                    if get_raw_price(bid_price_bytes) == PRICE_UNDEF
289                        || get_raw_quantity(bid_size_bytes) == QUANTITY_UNDEF
290                    {
291                        bids[i] = BookOrder::default();
292                    } else {
293                        let bid_price =
294                            decode_price(bid_price_bytes, price_precision, "bid_price", row)?;
295                        let bid_size =
296                            decode_quantity(bid_size_bytes, size_precision, "bid_size", row)?;
297                        bids[i] = BookOrder::new(OrderSide::Buy, bid_price, bid_size, 0);
298                    }
299
300                    let ask_price_bytes = ask_prices[i].value(row);
301                    let ask_size_bytes = ask_sizes[i].value(row);
302                    if get_raw_price(ask_price_bytes) == PRICE_UNDEF
303                        || get_raw_quantity(ask_size_bytes) == QUANTITY_UNDEF
304                    {
305                        asks[i] = BookOrder::default();
306                    } else {
307                        let ask_price =
308                            decode_price(ask_price_bytes, price_precision, "ask_price", row)?;
309                        let ask_size =
310                            decode_quantity(ask_size_bytes, size_precision, "ask_size", row)?;
311                        asks[i] = BookOrder::new(OrderSide::Sell, ask_price, ask_size, 0);
312                    }
313
314                    bid_count_arr[i] = bid_counts[i].value(row);
315                    ask_count_arr[i] = ask_counts[i].value(row);
316                }
317
318                Ok(Self {
319                    instrument_id,
320                    bids,
321                    asks,
322                    bid_counts: bid_count_arr,
323                    ask_counts: ask_count_arr,
324                    flags: flags.value(row),
325                    sequence: sequence.value(row),
326                    ts_event: ts_event.value(row).into(),
327                    ts_init: ts_init.value(row).into(),
328                })
329            })
330            .collect();
331
332        result
333    }
334}
335
336impl DecodeDataFromRecordBatch for OrderBookDepth10 {
337    fn decode_data_batch(
338        metadata: &HashMap<String, String>,
339        record_batch: RecordBatch,
340    ) -> Result<Vec<Data>, EncodingError> {
341        let depths: Vec<Self> = Self::decode_batch(metadata, record_batch)?;
342        Ok(depths.into_iter().map(Data::from).collect())
343    }
344}
345
346#[cfg(test)]
347mod tests {
348    use arrow::datatypes::{DataType, Field};
349    use nautilus_model::{
350        data::stubs::stub_depth10,
351        enums::OrderSide,
352        identifiers::InstrumentId,
353        types::{Price, Quantity, fixed::FIXED_SCALAR, price::PriceRaw, quantity::QuantityRaw},
354    };
355    use pretty_assertions::assert_eq;
356    use rstest::rstest;
357
358    use super::*;
359    use crate::arrow::{KEY_INSTRUMENT_ID, KEY_PRICE_PRECISION, get_raw_price, get_raw_quantity};
360
361    #[rstest]
362    fn test_get_schema() {
363        let instrument_id = InstrumentId::from("AAPL.XNAS");
364        let metadata = OrderBookDepth10::get_metadata(&instrument_id, 2, 0);
365        let schema = OrderBookDepth10::get_schema(Some(metadata));
366
367        let mut group_count = 0;
368        let field_data = get_field_data();
369        for (name, data_type) in field_data {
370            for i in 0..DEPTH10_LEN {
371                let field = schema.field(i + group_count * DEPTH10_LEN).clone();
372                assert_eq!(
373                    field,
374                    Field::new(format!("{name}_{i}"), data_type.clone(), false)
375                );
376            }
377
378            group_count += 1;
379        }
380
381        let flags_field = schema.field(group_count * DEPTH10_LEN).clone();
382        assert_eq!(flags_field, Field::new("flags", DataType::UInt8, false));
383        let sequence_field = schema.field(group_count * DEPTH10_LEN + 1).clone();
384        assert_eq!(
385            sequence_field,
386            Field::new("sequence", DataType::UInt64, false)
387        );
388        let ts_event_field = schema.field(group_count * DEPTH10_LEN + 2).clone();
389        assert_eq!(
390            ts_event_field,
391            Field::new("ts_event", DataType::UInt64, false)
392        );
393        let ts_init_field = schema.field(group_count * DEPTH10_LEN + 3).clone();
394        assert_eq!(
395            ts_init_field,
396            Field::new("ts_init", DataType::UInt64, false)
397        );
398
399        assert_eq!(schema.metadata()["instrument_id"], "AAPL.XNAS");
400        assert_eq!(schema.metadata()["price_precision"], "2");
401        assert_eq!(schema.metadata()["size_precision"], "0");
402    }
403
404    #[rstest]
405    fn test_get_schema_map() {
406        let schema_map = OrderBookDepth10::get_schema_map();
407
408        let field_data = get_field_data();
409        for (name, data_type) in field_data {
410            for i in 0..DEPTH10_LEN {
411                let field = schema_map.get(&format!("{name}_{i}")).map(String::as_str);
412                assert_eq!(field, Some(format!("{data_type:?}").as_str()));
413            }
414        }
415
416        assert_eq!(schema_map.get("flags").map(String::as_str), Some("UInt8"));
417        assert_eq!(
418            schema_map.get("sequence").map(String::as_str),
419            Some("UInt64")
420        );
421        assert_eq!(
422            schema_map.get("ts_event").map(String::as_str),
423            Some("UInt64")
424        );
425        assert_eq!(
426            schema_map.get("ts_init").map(String::as_str),
427            Some("UInt64")
428        );
429    }
430
431    #[rstest]
432    fn test_encode_batch(stub_depth10: OrderBookDepth10) {
433        let instrument_id = InstrumentId::from("AAPL.XNAS");
434        let price_precision = 2;
435        let metadata = OrderBookDepth10::get_metadata(&instrument_id, price_precision, 0);
436
437        let data = vec![stub_depth10];
438        let record_batch = OrderBookDepth10::encode_batch(&metadata, &data).unwrap();
439        let columns = record_batch.columns();
440
441        assert_eq!(columns.len(), DEPTH10_LEN * 6 + 4);
442
443        // Extract and test bid prices
444        let bid_prices: Vec<_> = (0..DEPTH10_LEN)
445            .map(|i| {
446                columns[i]
447                    .as_any()
448                    .downcast_ref::<FixedSizeBinaryArray>()
449                    .unwrap()
450            })
451            .collect();
452
453        let expected_bid_prices: Vec<f64> =
454            vec![99.0, 98.0, 97.0, 96.0, 95.0, 94.0, 93.0, 92.0, 91.0, 90.0];
455
456        for (i, bid_price) in bid_prices.iter().enumerate() {
457            assert_eq!(bid_price.len(), 1);
458            assert_eq!(
459                get_raw_price(bid_price.value(0)),
460                (expected_bid_prices[i] * FIXED_SCALAR) as PriceRaw
461            );
462            assert_eq!(
463                Price::from_raw(get_raw_price(bid_price.value(0)), price_precision).as_f64(),
464                expected_bid_prices[i]
465            );
466        }
467
468        // Extract and test ask prices
469        let ask_prices: Vec<_> = (0..DEPTH10_LEN)
470            .map(|i| {
471                columns[DEPTH10_LEN + i]
472                    .as_any()
473                    .downcast_ref::<FixedSizeBinaryArray>()
474                    .unwrap()
475            })
476            .collect();
477
478        let expected_ask_prices: Vec<f64> = vec![
479            100.0, 101.0, 102.0, 103.0, 104.0, 105.0, 106.0, 107.0, 108.0, 109.0,
480        ];
481
482        for (i, ask_price) in ask_prices.iter().enumerate() {
483            assert_eq!(ask_price.len(), 1);
484            assert_eq!(
485                get_raw_price(ask_price.value(0)),
486                (expected_ask_prices[i] * FIXED_SCALAR) as PriceRaw
487            );
488            assert_eq!(
489                Price::from_raw(get_raw_price(ask_price.value(0)), price_precision).as_f64(),
490                expected_ask_prices[i]
491            );
492        }
493
494        // Extract and test bid sizes
495        let bid_sizes: Vec<_> = (0..DEPTH10_LEN)
496            .map(|i| {
497                columns[2 * DEPTH10_LEN + i]
498                    .as_any()
499                    .downcast_ref::<FixedSizeBinaryArray>()
500                    .unwrap()
501            })
502            .collect();
503
504        for (i, bid_size) in bid_sizes.iter().enumerate() {
505            assert_eq!(bid_size.len(), 1);
506            assert_eq!(
507                get_raw_quantity(bid_size.value(0)),
508                ((100.0 * FIXED_SCALAR * (i + 1) as f64) as QuantityRaw)
509            );
510        }
511
512        // Extract and test ask sizes
513        let ask_sizes: Vec<_> = (0..DEPTH10_LEN)
514            .map(|i| {
515                columns[3 * DEPTH10_LEN + i]
516                    .as_any()
517                    .downcast_ref::<FixedSizeBinaryArray>()
518                    .unwrap()
519            })
520            .collect();
521
522        for (i, ask_size) in ask_sizes.iter().enumerate() {
523            assert_eq!(ask_size.len(), 1);
524            assert_eq!(
525                get_raw_quantity(ask_size.value(0)),
526                ((100.0 * FIXED_SCALAR * ((i + 1) as f64)) as QuantityRaw)
527            );
528        }
529
530        // Extract and test bid counts
531        let bid_counts: Vec<_> = (0..DEPTH10_LEN)
532            .map(|i| {
533                columns[4 * DEPTH10_LEN + i]
534                    .as_any()
535                    .downcast_ref::<UInt32Array>()
536                    .unwrap()
537            })
538            .collect();
539
540        for count_values in bid_counts {
541            assert_eq!(count_values.len(), 1);
542            assert_eq!(count_values.value(0), 1);
543        }
544
545        // Extract and test ask counts
546        let ask_counts: Vec<_> = (0..DEPTH10_LEN)
547            .map(|i| {
548                columns[5 * DEPTH10_LEN + i]
549                    .as_any()
550                    .downcast_ref::<UInt32Array>()
551                    .unwrap()
552            })
553            .collect();
554
555        for count_values in ask_counts {
556            assert_eq!(count_values.len(), 1);
557            assert_eq!(count_values.value(0), 1);
558        }
559
560        // Test remaining fields
561        let flags_values = columns[6 * DEPTH10_LEN]
562            .as_any()
563            .downcast_ref::<UInt8Array>()
564            .unwrap();
565        let sequence_values = columns[6 * DEPTH10_LEN + 1]
566            .as_any()
567            .downcast_ref::<UInt64Array>()
568            .unwrap();
569        let ts_event_values = columns[6 * DEPTH10_LEN + 2]
570            .as_any()
571            .downcast_ref::<UInt64Array>()
572            .unwrap();
573        let ts_init_values = columns[6 * DEPTH10_LEN + 3]
574            .as_any()
575            .downcast_ref::<UInt64Array>()
576            .unwrap();
577
578        assert_eq!(flags_values.len(), 1);
579        assert_eq!(flags_values.value(0), 0);
580        assert_eq!(sequence_values.len(), 1);
581        assert_eq!(sequence_values.value(0), 0);
582        assert_eq!(ts_event_values.len(), 1);
583        assert_eq!(ts_event_values.value(0), 1);
584        assert_eq!(ts_init_values.len(), 1);
585        assert_eq!(ts_init_values.value(0), 2);
586    }
587
588    #[rstest]
589    fn test_decode_batch(stub_depth10: OrderBookDepth10) {
590        let instrument_id = InstrumentId::from("AAPL.XNAS");
591        let metadata = OrderBookDepth10::get_metadata(&instrument_id, 2, 0);
592
593        let data = vec![stub_depth10];
594        let record_batch = OrderBookDepth10::encode_batch(&metadata, &data).unwrap();
595        let decoded_data = OrderBookDepth10::decode_batch(&metadata, record_batch).unwrap();
596
597        assert_eq!(decoded_data.len(), 1);
598    }
599
600    #[rstest]
601    fn test_decode_batch_missing_instrument_id_returns_error(stub_depth10: OrderBookDepth10) {
602        let instrument_id = InstrumentId::from("AAPL.XNAS");
603        let mut metadata = OrderBookDepth10::get_metadata(&instrument_id, 2, 0);
604        let record_batch = OrderBookDepth10::encode_batch(&metadata, &[stub_depth10]).unwrap();
605
606        metadata.remove(KEY_INSTRUMENT_ID);
607
608        let result = OrderBookDepth10::decode_batch(&metadata, record_batch);
609        assert!(result.is_err());
610        let err = result.unwrap_err();
611        assert!(
612            err.to_string().contains("instrument_id"),
613            "Expected missing instrument_id error, was: {err}"
614        );
615    }
616
617    #[rstest]
618    fn test_decode_batch_missing_price_precision_returns_error(stub_depth10: OrderBookDepth10) {
619        let instrument_id = InstrumentId::from("AAPL.XNAS");
620        let mut metadata = OrderBookDepth10::get_metadata(&instrument_id, 2, 0);
621        let record_batch = OrderBookDepth10::encode_batch(&metadata, &[stub_depth10]).unwrap();
622
623        metadata.remove(KEY_PRICE_PRECISION);
624
625        let result = OrderBookDepth10::decode_batch(&metadata, record_batch);
626        assert!(result.is_err());
627        let err = result.unwrap_err();
628        assert!(
629            err.to_string().contains("price_precision"),
630            "Expected missing price_precision error, was: {err}"
631        );
632    }
633
634    #[rstest]
635    fn test_encode_decode_round_trip(stub_depth10: OrderBookDepth10) {
636        let instrument_id = InstrumentId::from("AAPL.XNAS");
637        let metadata = OrderBookDepth10::get_metadata(&instrument_id, 2, 0);
638
639        let original = vec![stub_depth10];
640        let record_batch = OrderBookDepth10::encode_batch(&metadata, &original).unwrap();
641        let decoded = OrderBookDepth10::decode_batch(&metadata, record_batch).unwrap();
642
643        assert_eq!(decoded.len(), original.len());
644        let orig = &original[0];
645        let dec = &decoded[0];
646
647        assert_eq!(dec.instrument_id, orig.instrument_id);
648        assert_eq!(dec.flags, orig.flags);
649        assert_eq!(dec.sequence, orig.sequence);
650        assert_eq!(dec.ts_event, orig.ts_event);
651        assert_eq!(dec.ts_init, orig.ts_init);
652
653        for i in 0..DEPTH10_LEN {
654            assert_eq!(
655                dec.bids[i].price, orig.bids[i].price,
656                "bid price mismatch at level {i}"
657            );
658            assert_eq!(
659                dec.bids[i].size, orig.bids[i].size,
660                "bid size mismatch at level {i}"
661            );
662            assert_eq!(
663                dec.asks[i].price, orig.asks[i].price,
664                "ask price mismatch at level {i}"
665            );
666            assert_eq!(
667                dec.asks[i].size, orig.asks[i].size,
668                "ask size mismatch at level {i}"
669            );
670        }
671    }
672
673    // Each case toggles the price and/or size sentinel on one bid level and one
674    // ask level. Any sentinel in either field must decode to NULL_ORDER; the
675    // "neither" case is the control proving defined levels still round-trip.
676    #[rstest]
677    #[case::price_only(true, false)]
678    #[case::size_only(false, true)]
679    #[case::both(true, true)]
680    #[case::neither(false, false)]
681    fn test_decode_batch_with_undefined_levels(
682        stub_depth10: OrderBookDepth10,
683        #[case] price_undef: bool,
684        #[case] size_undef: bool,
685    ) {
686        let instrument_id = InstrumentId::from("AAPL.XNAS");
687        let price_precision = 2;
688        let size_precision = 0;
689        let metadata =
690            OrderBookDepth10::get_metadata(&instrument_id, price_precision, size_precision);
691
692        let mut depth = stub_depth10;
693        let original_bid = depth.bids[5];
694        let original_ask = depth.asks[7];
695        let sentinel_bid_price = if price_undef {
696            Price::from_raw(PRICE_UNDEF, 0)
697        } else {
698            original_bid.price
699        };
700        let sentinel_bid_size = if size_undef {
701            Quantity::from_raw(QUANTITY_UNDEF, 0)
702        } else {
703            original_bid.size
704        };
705        depth.bids[5] = BookOrder {
706            side: OrderSide::Buy.into(),
707            price: sentinel_bid_price,
708            size: sentinel_bid_size,
709            order_id: 0,
710        };
711        let sentinel_ask_price = if price_undef {
712            Price::from_raw(PRICE_UNDEF, 0)
713        } else {
714            original_ask.price
715        };
716        let sentinel_ask_size = if size_undef {
717            Quantity::from_raw(QUANTITY_UNDEF, 0)
718        } else {
719            original_ask.size
720        };
721        depth.asks[7] = BookOrder {
722            side: OrderSide::Sell.into(),
723            price: sentinel_ask_price,
724            size: sentinel_ask_size,
725            order_id: 0,
726        };
727
728        let record_batch = OrderBookDepth10::encode_batch(&metadata, &[depth]).unwrap();
729        let decoded = OrderBookDepth10::decode_batch(&metadata, record_batch).unwrap();
730
731        assert_eq!(decoded.len(), 1);
732        let decoded = &decoded[0];
733
734        let expect_null = price_undef || size_undef;
735        if expect_null {
736            assert_eq!(decoded.bids[5].side, None);
737            assert_eq!(decoded.bids[5].price.raw(), 0);
738            assert_eq!(decoded.bids[5].price.precision, 0);
739            assert_eq!(decoded.bids[5].size.raw(), 0);
740            assert_eq!(decoded.bids[5].size.precision, 0);
741
742            assert_eq!(decoded.asks[7].side, None);
743            assert_eq!(decoded.asks[7].price.raw(), 0);
744            assert_eq!(decoded.asks[7].price.precision, 0);
745            assert_eq!(decoded.asks[7].size.raw(), 0);
746            assert_eq!(decoded.asks[7].size.precision, 0);
747        } else {
748            assert_eq!(decoded.bids[5].side, Some(OrderSide::Buy));
749            assert_eq!(decoded.bids[5].price, original_bid.price);
750            assert_eq!(decoded.bids[5].size, original_bid.size);
751            assert_eq!(decoded.asks[7].side, Some(OrderSide::Sell));
752            assert_eq!(decoded.asks[7].price, original_ask.price);
753            assert_eq!(decoded.asks[7].size, original_ask.size);
754        }
755
756        // Surrounding defined levels always round-trip with the instrument precision
757        assert_eq!(decoded.bids[0].side, Some(OrderSide::Buy));
758        assert_eq!(decoded.bids[0].price.precision, price_precision);
759        assert_eq!(decoded.bids[0].size.precision, size_precision);
760        assert_eq!(decoded.asks[0].side, Some(OrderSide::Sell));
761        assert_eq!(decoded.asks[0].price.precision, price_precision);
762        assert_eq!(decoded.asks[0].size.precision, size_precision);
763    }
764}
nautilus_serialization/arrow/depth.rs

nautilus_serialization/arrow/
depth.rs
