nautilus_serialization/arrow/
quote.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::{Decimal128Array, UInt64Array},
20    datatypes::{DataType, Field, Schema},
21    error::ArrowError,
22    record_batch::RecordBatch,
23};
24use nautilus_model::data::QuoteTick;
25#[cfg(test)]
26use nautilus_model::identifiers::InstrumentId;
27
28use super::{
29    DecodeDataFromRecordBatch, EncodingError, KEY_IDENTIFIER, decode_required_decimal_price,
30    decode_required_decimal_quantity, decode_required_timestamp, extract_column,
31    fixed_decimal_data_type, identifier_array_from_display, metadata_with_type_name,
32    parse_metadata, required_price_decimal_array, required_quantity_decimal_array,
33};
34#[cfg(test)]
35use super::{KEY_INSTRUMENT_ID, KEY_PRICE_PRECISION};
36use crate::arrow::{ArrowSchemaProvider, Data, DecodeFromRecordBatch, EncodeToRecordBatch};
37
38impl ArrowSchemaProvider for QuoteTick {
39    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
40        let fields = vec![
41            Field::new("bid_price", fixed_decimal_data_type(), false),
42            Field::new("ask_price", fixed_decimal_data_type(), false),
43            Field::new("bid_size", fixed_decimal_data_type(), false),
44            Field::new("ask_size", fixed_decimal_data_type(), false),
45            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
46            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
47            Field::new(KEY_IDENTIFIER, DataType::Utf8, true),
48        ];
49
50        Schema::new_with_metadata(fields, metadata_with_type_name("QuoteTick", metadata))
51    }
52}
53
54impl EncodeToRecordBatch for QuoteTick {
55    fn encode_batch<T>(
56        metadata: &HashMap<String, String>,
57        data: &[T],
58    ) -> Result<RecordBatch, ArrowError>
59    where
60        T: std::borrow::Borrow<Self>,
61    {
62        let mut ts_event_builder = UInt64Array::builder(data.len());
63        let mut ts_init_builder = UInt64Array::builder(data.len());
64
65        for quote in data.iter().map(std::borrow::Borrow::borrow) {
66            ts_event_builder.append_value(quote.ts_event.as_u64());
67            ts_init_builder.append_value(quote.ts_init.as_u64());
68        }
69
70        crate::arrow::record_batch_with_timestamps(
71            Self::get_schema(Some(metadata.clone())).into(),
72            vec![
73                Arc::new(required_price_decimal_array(
74                    data.iter()
75                        .map(std::borrow::Borrow::borrow)
76                        .map(|quote| quote.bid_price.raw()),
77                    "bid_price",
78                )?),
79                Arc::new(required_price_decimal_array(
80                    data.iter()
81                        .map(std::borrow::Borrow::borrow)
82                        .map(|quote| quote.ask_price.raw()),
83                    "ask_price",
84                )?),
85                Arc::new(required_quantity_decimal_array(
86                    data.iter()
87                        .map(std::borrow::Borrow::borrow)
88                        .map(|quote| quote.bid_size.raw()),
89                    "bid_size",
90                )?),
91                Arc::new(required_quantity_decimal_array(
92                    data.iter()
93                        .map(std::borrow::Borrow::borrow)
94                        .map(|quote| quote.ask_size.raw()),
95                    "ask_size",
96                )?),
97                Arc::new(ts_event_builder.finish()),
98                Arc::new(ts_init_builder.finish()),
99                Arc::new(identifier_array_from_display(
100                    data.iter()
101                        .map(std::borrow::Borrow::borrow)
102                        .map(|quote| quote.instrument_id),
103                )),
104            ],
105        )
106    }
107
108    fn metadata(&self) -> HashMap<String, String> {
109        Self::get_metadata(
110            &self.instrument_id,
111            self.bid_price.precision,
112            self.bid_size.precision,
113        )
114    }
115}
116
117impl DecodeFromRecordBatch for QuoteTick {
118    fn decode_batch(
119        metadata: &HashMap<String, String>,
120        record_batch: RecordBatch,
121    ) -> Result<Vec<Self>, EncodingError> {
122        let (instrument_id, price_precision, size_precision) = parse_metadata(metadata)?;
123        let record_batch = crate::arrow::record_batch_with_u64_timestamps(&record_batch)?;
124        let record_batch = &record_batch;
125        let cols = record_batch.columns();
126
127        let bid_price_values =
128            extract_column::<Decimal128Array>(cols, "bid_price", 0, fixed_decimal_data_type())?;
129        let ask_price_values =
130            extract_column::<Decimal128Array>(cols, "ask_price", 1, fixed_decimal_data_type())?;
131        let bid_size_values =
132            extract_column::<Decimal128Array>(cols, "bid_size", 2, fixed_decimal_data_type())?;
133        let ask_size_values =
134            extract_column::<Decimal128Array>(cols, "ask_size", 3, fixed_decimal_data_type())?;
135        let ts_event_values = extract_column::<UInt64Array>(cols, "ts_event", 4, DataType::UInt64)?;
136        let ts_init_values = extract_column::<UInt64Array>(cols, "ts_init", 5, DataType::UInt64)?;
137
138        let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
139            .map(|row| {
140                let bid_price = decode_required_decimal_price(
141                    bid_price_values,
142                    price_precision,
143                    "bid_price",
144                    row,
145                )?;
146                let ask_price = decode_required_decimal_price(
147                    ask_price_values,
148                    price_precision,
149                    "ask_price",
150                    row,
151                )?;
152                let bid_size = decode_required_decimal_quantity(
153                    bid_size_values,
154                    size_precision,
155                    "bid_size",
156                    row,
157                )?;
158                let ask_size = decode_required_decimal_quantity(
159                    ask_size_values,
160                    size_precision,
161                    "ask_size",
162                    row,
163                )?;
164                Ok(Self {
165                    instrument_id,
166                    bid_price,
167                    ask_price,
168                    bid_size,
169                    ask_size,
170                    ts_event: decode_required_timestamp(ts_event_values, "ts_event", row)?,
171                    ts_init: decode_required_timestamp(ts_init_values, "ts_init", row)?,
172                })
173            })
174            .collect();
175
176        result
177    }
178}
179
180impl DecodeDataFromRecordBatch for QuoteTick {
181    fn decode_data_batch(
182        metadata: &HashMap<String, String>,
183        record_batch: RecordBatch,
184    ) -> Result<Vec<Data>, EncodingError> {
185        let ticks: Vec<Self> = Self::decode_batch(metadata, record_batch)?;
186        Ok(ticks.into_iter().map(Data::from).collect())
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use std::{collections::HashMap, sync::Arc};
193
194    use arrow::array::{Array, StringArray, TimestampNanosecondArray};
195    use nautilus_model::types::{
196        Price, Quantity, fixed::FIXED_SCALAR, price::PriceRaw, quantity::QuantityRaw,
197    };
198    use rstest::rstest;
199
200    use super::*;
201    use crate::arrow::{KEY_IDENTIFIER, KEY_TYPE_NAME, get_raw_price, get_raw_quantity};
202
203    #[rstest]
204    fn test_quote_nanoseconds_round_trip_within_one_microsecond() {
205        let first = QuoteTick {
206            instrument_id: InstrumentId::from("AAPL.XNAS"),
207            bid_price: Price::from("123.45"),
208            ask_price: Price::from("123.67"),
209            bid_size: Quantity::from(17),
210            ask_size: Quantity::from(29),
211            ts_event: 1_788_652_800_123_456_789_u64.into(),
212            ts_init: 1_788_652_800_123_456_799_u64.into(),
213        };
214        let second = QuoteTick {
215            ts_event: 1_788_652_800_123_456_801_u64.into(),
216            ts_init: 1_788_652_800_123_456_899_u64.into(),
217            ..first
218        };
219        let values = vec![first, second];
220        let metadata = QuoteTick::get_metadata(&first.instrument_id, 2, 0);
221        let batch = QuoteTick::encode_batch(&metadata, &values).unwrap();
222
223        for field in ["ts_event", "ts_init"] {
224            assert_eq!(
225                batch.schema().field_with_name(field).unwrap().data_type(),
226                &DataType::Timestamp(arrow::datatypes::TimeUnit::Nanosecond, Some("UTC".into()))
227            );
228        }
229        let decoded = QuoteTick::decode_batch(&metadata, batch).unwrap();
230        assert_eq!(decoded, values);
231    }
232
233    #[rstest]
234    #[case(38, 2)]
235    #[case(38, 9)]
236    #[case(38, 18)]
237    #[case(37, 16)]
238    fn test_decode_rejects_decimal_type(
239        #[case] precision: u8,
240        #[case] scale: i8,
241        #[values(0, 1, 2, 3)] column: usize,
242    ) {
243        let quote = QuoteTick {
244            instrument_id: InstrumentId::from("AAPL.XNAS"),
245            bid_price: Price::from("123.45"),
246            ask_price: Price::from("123.67"),
247            bid_size: Quantity::from(17),
248            ask_size: Quantity::from(29),
249            ts_event: 1.into(),
250            ts_init: 2.into(),
251        };
252        let metadata = quote.metadata();
253        let batch = QuoteTick::encode_batch(&metadata, &[quote]).unwrap();
254        let mut columns = batch.columns().to_vec();
255        let values = columns[column]
256            .as_any()
257            .downcast_ref::<Decimal128Array>()
258            .unwrap()
259            .clone()
260            .with_precision_and_scale(precision, scale)
261            .unwrap();
262        columns[column] = Arc::new(values);
263        let mut fields = batch.schema().fields().to_vec();
264        let name = fields[column].name().clone();
265        fields[column] = Arc::new(
266            fields[column]
267                .as_ref()
268                .clone()
269                .with_data_type(DataType::Decimal128(precision, scale)),
270        );
271        let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap();
272
273        let error = QuoteTick::decode_batch(&metadata, batch).unwrap_err();
274
275        match error {
276            EncodingError::InvalidColumnType(field, index, expected, actual) => {
277                assert_eq!(field, name);
278                assert_eq!(index, column);
279                assert_eq!(expected, fixed_decimal_data_type());
280                assert_eq!(actual, DataType::Decimal128(precision, scale));
281            }
282            error => panic!("Unexpected error: {error}"),
283        }
284    }
285
286    #[rstest]
287    fn test_get_schema() {
288        let instrument_id = InstrumentId::from("AAPL.XNAS");
289        let metadata = QuoteTick::get_metadata(&instrument_id, 2, 0);
290        let schema = QuoteTick::get_schema(Some(metadata.clone()));
291
292        let mut expected_fields = Vec::with_capacity(7);
293
294        expected_fields.push(Field::new("bid_price", fixed_decimal_data_type(), false));
295        expected_fields.push(Field::new("ask_price", fixed_decimal_data_type(), false));
296
297        expected_fields.extend(vec![
298            Field::new("bid_size", fixed_decimal_data_type(), false),
299            Field::new("ask_size", fixed_decimal_data_type(), false),
300            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
301            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
302            Field::new(KEY_IDENTIFIER, DataType::Utf8, true),
303        ]);
304
305        let mut expected_metadata = metadata;
306        expected_metadata.insert(KEY_TYPE_NAME.to_string(), "QuoteTick".to_string());
307        let expected_schema = Schema::new_with_metadata(expected_fields, expected_metadata);
308        assert_eq!(schema, expected_schema);
309    }
310
311    #[rstest]
312    fn test_get_schema_map() {
313        let arrow_schema = QuoteTick::get_schema_map();
314        let mut expected_map = HashMap::new();
315
316        let fixed_size_binary = "Decimal128(38, 16)".to_string();
317        expected_map.insert("bid_price".to_string(), fixed_size_binary.clone());
318        expected_map.insert("ask_price".to_string(), fixed_size_binary.clone());
319        expected_map.insert("bid_size".to_string(), fixed_size_binary.clone());
320        expected_map.insert("ask_size".to_string(), fixed_size_binary);
321        expected_map.insert(
322            "ts_event".to_string(),
323            "Timestamp(Nanosecond, Some(\"UTC\"))".to_string(),
324        );
325        expected_map.insert(
326            "ts_init".to_string(),
327            "Timestamp(Nanosecond, Some(\"UTC\"))".to_string(),
328        );
329        expected_map.insert(KEY_IDENTIFIER.to_string(), "Utf8".to_string());
330        assert_eq!(arrow_schema, expected_map);
331    }
332
333    #[rstest]
334    fn test_encode_quote_tick() {
335        // Create test data
336        let instrument_id = InstrumentId::from("AAPL.XNAS");
337        let tick1 = QuoteTick {
338            instrument_id,
339            bid_price: Price::from("100.10"),
340            ask_price: Price::from("101.50"),
341            bid_size: Quantity::from(1000),
342            ask_size: Quantity::from(500),
343            ts_event: 1.into(),
344            ts_init: 3.into(),
345        };
346
347        let tick2 = QuoteTick {
348            instrument_id,
349            bid_price: Price::from("100.75"),
350            ask_price: Price::from("100.20"),
351            bid_size: Quantity::from(750),
352            ask_size: Quantity::from(300),
353            ts_event: 2.into(),
354            ts_init: 4.into(),
355        };
356
357        let data = vec![tick1, tick2];
358        let metadata = QuoteTick::get_metadata(&instrument_id, 2, 0);
359        let record_batch = QuoteTick::encode_batch(&metadata, &data).unwrap();
360
361        // Verify the encoded data
362        let columns = record_batch.columns();
363
364        let bid_price_values = columns[0]
365            .as_any()
366            .downcast_ref::<Decimal128Array>()
367            .unwrap();
368        let ask_price_values = columns[1]
369            .as_any()
370            .downcast_ref::<Decimal128Array>()
371            .unwrap();
372        assert_eq!(
373            get_raw_price(bid_price_values.value(0)),
374            (100.10 * FIXED_SCALAR) as PriceRaw
375        );
376        assert_eq!(
377            get_raw_price(bid_price_values.value(1)),
378            (100.75 * FIXED_SCALAR) as PriceRaw
379        );
380        assert_eq!(
381            get_raw_price(ask_price_values.value(0)),
382            (101.50 * FIXED_SCALAR) as PriceRaw
383        );
384        assert_eq!(
385            get_raw_price(ask_price_values.value(1)),
386            (100.20 * FIXED_SCALAR) as PriceRaw
387        );
388
389        let bid_size_values = columns[2]
390            .as_any()
391            .downcast_ref::<Decimal128Array>()
392            .unwrap();
393        let ask_size_values = columns[3]
394            .as_any()
395            .downcast_ref::<Decimal128Array>()
396            .unwrap();
397        let ts_event_values = columns[4]
398            .as_any()
399            .downcast_ref::<TimestampNanosecondArray>()
400            .unwrap();
401        let ts_init_values = columns[5]
402            .as_any()
403            .downcast_ref::<TimestampNanosecondArray>()
404            .unwrap();
405
406        let identifier_values = columns[6].as_any().downcast_ref::<StringArray>().unwrap();
407
408        assert_eq!(columns.len(), 7);
409        assert_eq!(bid_size_values.len(), 2);
410        assert_eq!(
411            get_raw_quantity(bid_size_values.value(0)),
412            (1000.0 * FIXED_SCALAR) as QuantityRaw
413        );
414        assert_eq!(
415            get_raw_quantity(bid_size_values.value(1)),
416            (750.0 * FIXED_SCALAR) as QuantityRaw
417        );
418        assert_eq!(ask_size_values.len(), 2);
419        assert_eq!(
420            get_raw_quantity(ask_size_values.value(0)),
421            (500.0 * FIXED_SCALAR) as QuantityRaw
422        );
423        assert_eq!(
424            get_raw_quantity(ask_size_values.value(1)),
425            (300.0 * FIXED_SCALAR) as QuantityRaw
426        );
427        assert_eq!(ts_event_values.len(), 2);
428        assert_eq!(ts_event_values.value(0), 1);
429        assert_eq!(ts_event_values.value(1), 2);
430        assert_eq!(ts_init_values.len(), 2);
431        assert_eq!(ts_init_values.value(0), 3);
432        assert_eq!(ts_init_values.value(1), 4);
433        assert_eq!(record_batch.schema().field(6).name(), KEY_IDENTIFIER);
434        assert_eq!(identifier_values.len(), 2);
435        assert_eq!(identifier_values.value(0), "AAPL.XNAS");
436        assert_eq!(identifier_values.value(1), "AAPL.XNAS");
437    }
438
439    #[rstest]
440    fn test_decode_batch() {
441        let instrument_id = InstrumentId::from("AAPL.XNAS");
442        let metadata = QuoteTick::get_metadata(&instrument_id, 2, 0);
443
444        let raw_bid1 = (100.00 * FIXED_SCALAR) as PriceRaw;
445        let raw_bid2 = (99.00 * FIXED_SCALAR) as PriceRaw;
446        let raw_ask1 = (101.00 * FIXED_SCALAR) as PriceRaw;
447        let raw_ask2 = (100.00 * FIXED_SCALAR) as PriceRaw;
448
449        let (bid_price, ask_price) = (
450            crate::arrow::test_support::decimal_array_from_bytes(vec![
451                &raw_bid1.to_le_bytes(),
452                &raw_bid2.to_le_bytes(),
453            ]),
454            crate::arrow::test_support::decimal_array_from_bytes(vec![
455                &raw_ask1.to_le_bytes(),
456                &raw_ask2.to_le_bytes(),
457            ]),
458        );
459
460        let bid_size = crate::arrow::test_support::decimal_array_from_bytes(vec![
461            &((100.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
462            &((90.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
463        ]);
464        let ask_size = crate::arrow::test_support::decimal_array_from_bytes(vec![
465            &((110.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
466            &((100.0 * FIXED_SCALAR) as PriceRaw).to_le_bytes(),
467        ]);
468        let ts_event = UInt64Array::from(vec![1, 2]);
469        let ts_init = UInt64Array::from(vec![3, 4]);
470
471        let record_batch = crate::arrow::record_batch_with_timestamps(
472            crate::arrow::schema_without_identifier_column(&QuoteTick::get_schema(Some(
473                metadata.clone(),
474            )))
475            .into(),
476            vec![
477                Arc::new(bid_price),
478                Arc::new(ask_price),
479                Arc::new(bid_size),
480                Arc::new(ask_size),
481                Arc::new(ts_event),
482                Arc::new(ts_init),
483            ],
484        )
485        .unwrap();
486
487        let decoded_data = QuoteTick::decode_batch(&metadata, record_batch).unwrap();
488        assert_eq!(decoded_data.len(), 2);
489
490        // Verify decoded values
491        assert_eq!(decoded_data[0].bid_price, Price::from_raw(raw_bid1, 2));
492        assert_eq!(decoded_data[0].ask_price, Price::from_raw(raw_ask1, 2));
493        assert_eq!(decoded_data[1].bid_price, Price::from_raw(raw_bid2, 2));
494        assert_eq!(decoded_data[1].ask_price, Price::from_raw(raw_ask2, 2));
495    }
496
497    #[rstest]
498    fn test_decode_batch_rejects_null_timestamp_with_field_and_row() {
499        let instrument_id = InstrumentId::from("AAPL.XNAS");
500        let metadata = QuoteTick::get_metadata(&instrument_id, 2, 0);
501        let quote = QuoteTick::new(
502            instrument_id,
503            Price::from("100.00"),
504            Price::from("101.00"),
505            Quantity::from(10),
506            Quantity::from(20),
507            1.into(),
508            2.into(),
509        );
510        let encoded = QuoteTick::encode_batch(&metadata, &[quote]).unwrap();
511        let mut columns = encoded.columns().to_vec();
512        columns[5] = Arc::new(TimestampNanosecondArray::from(vec![None]).with_timezone("UTC"));
513        let fields = encoded
514            .schema()
515            .fields()
516            .iter()
517            .map(|field| {
518                if field.name() == "ts_init" {
519                    Arc::new(field.as_ref().clone().with_nullable(true))
520                } else {
521                    Arc::clone(field)
522                }
523            })
524            .collect::<Vec<_>>();
525        let schema = Arc::new(Schema::new_with_metadata(fields, metadata.clone()));
526        let batch = RecordBatch::try_new(schema, columns).unwrap();
527
528        let error = QuoteTick::decode_batch(&metadata, batch).unwrap_err();
529
530        assert!(error.to_string().contains("ts_init"));
531        assert!(error.to_string().contains("row 0"));
532    }
533
534    #[rstest]
535    fn test_decode_batch_invalid_bid_price_returns_error() {
536        let instrument_id = InstrumentId::from("AAPL.XNAS");
537        let metadata = QuoteTick::get_metadata(&instrument_id, 2, 0);
538
539        let invalid_price: PriceRaw = PriceRaw::MAX - 1000;
540        let valid_price = (100.00 * FIXED_SCALAR) as PriceRaw;
541
542        let bid_price = crate::arrow::test_support::decimal_array_from_bytes(vec![
543            &invalid_price.to_le_bytes(),
544        ]);
545        let ask_price =
546            crate::arrow::test_support::decimal_array_from_bytes(vec![&valid_price.to_le_bytes()]);
547        let bid_size = crate::arrow::test_support::decimal_array_from_bytes(vec![
548            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
549        ]);
550        let ask_size = crate::arrow::test_support::decimal_array_from_bytes(vec![
551            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
552        ]);
553        let ts_event = UInt64Array::from(vec![1]);
554        let ts_init = UInt64Array::from(vec![2]);
555
556        let record_batch = crate::arrow::record_batch_with_timestamps(
557            crate::arrow::schema_without_identifier_column(&QuoteTick::get_schema(Some(
558                metadata.clone(),
559            )))
560            .into(),
561            vec![
562                Arc::new(bid_price),
563                Arc::new(ask_price),
564                Arc::new(bid_size),
565                Arc::new(ask_size),
566                Arc::new(ts_event),
567                Arc::new(ts_init),
568            ],
569        )
570        .unwrap();
571
572        let result = QuoteTick::decode_batch(&metadata, record_batch);
573        assert!(result.is_err());
574        let err = result.unwrap_err();
575        assert!(
576            err.to_string().contains("bid_price") && err.to_string().contains("row 0"),
577            "Expected bid_price error at row 0, was: {err}"
578        );
579    }
580
581    #[rstest]
582    fn test_decode_batch_invalid_ask_size_returns_error() {
583        use nautilus_model::types::{fixed::FIXED_PRECISION, quantity::QUANTITY_RAW_MAX};
584
585        let instrument_id = InstrumentId::from("AAPL.XNAS");
586        // Decode the size at full precision so the out-of-range raw value bypasses the
587        // precision-0 correction, which would otherwise round it back within the bound.
588        let metadata = QuoteTick::get_metadata(&instrument_id, 2, FIXED_PRECISION);
589
590        let valid_price = (100.00 * FIXED_SCALAR) as PriceRaw;
591        let bid_price =
592            crate::arrow::test_support::decimal_array_from_bytes(vec![&valid_price.to_le_bytes()]);
593        let ask_price =
594            crate::arrow::test_support::decimal_array_from_bytes(vec![&valid_price.to_le_bytes()]);
595        let bid_size = crate::arrow::test_support::decimal_array_from_bytes(vec![
596            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
597        ]);
598
599        let invalid_size = QUANTITY_RAW_MAX + 1;
600        let ask_size =
601            crate::arrow::test_support::decimal_array_from_bytes(vec![&invalid_size.to_le_bytes()]);
602        let ts_event = UInt64Array::from(vec![1]);
603        let ts_init = UInt64Array::from(vec![2]);
604
605        let record_batch = crate::arrow::record_batch_with_timestamps(
606            crate::arrow::schema_without_identifier_column(&QuoteTick::get_schema(Some(
607                metadata.clone(),
608            )))
609            .into(),
610            vec![
611                Arc::new(bid_price),
612                Arc::new(ask_price),
613                Arc::new(bid_size),
614                Arc::new(ask_size),
615                Arc::new(ts_event),
616                Arc::new(ts_init),
617            ],
618        )
619        .unwrap();
620
621        let result = QuoteTick::decode_batch(&metadata, record_batch);
622        assert!(result.is_err());
623        let err = result.unwrap_err();
624        assert!(
625            err.to_string().contains("ask_size") && err.to_string().contains("row 0"),
626            "Expected ask_size error at row 0, was: {err}"
627        );
628    }
629
630    #[rstest]
631    fn test_decode_batch_missing_instrument_id_returns_error() {
632        let instrument_id = InstrumentId::from("AAPL.XNAS");
633        let mut metadata = QuoteTick::get_metadata(&instrument_id, 2, 0);
634        metadata.remove(KEY_INSTRUMENT_ID);
635
636        let valid_price = (100.00 * FIXED_SCALAR) as PriceRaw;
637        let bid_price =
638            crate::arrow::test_support::decimal_array_from_bytes(vec![&valid_price.to_le_bytes()]);
639        let ask_price =
640            crate::arrow::test_support::decimal_array_from_bytes(vec![&valid_price.to_le_bytes()]);
641        let bid_size = crate::arrow::test_support::decimal_array_from_bytes(vec![
642            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
643        ]);
644        let ask_size = crate::arrow::test_support::decimal_array_from_bytes(vec![
645            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
646        ]);
647        let ts_event = UInt64Array::from(vec![1]);
648        let ts_init = UInt64Array::from(vec![2]);
649
650        let record_batch = crate::arrow::record_batch_with_timestamps(
651            crate::arrow::schema_without_identifier_column(&QuoteTick::get_schema(Some(
652                metadata.clone(),
653            )))
654            .into(),
655            vec![
656                Arc::new(bid_price),
657                Arc::new(ask_price),
658                Arc::new(bid_size),
659                Arc::new(ask_size),
660                Arc::new(ts_event),
661                Arc::new(ts_init),
662            ],
663        )
664        .unwrap();
665
666        let result = QuoteTick::decode_batch(&metadata, record_batch);
667        assert!(result.is_err());
668        let err = result.unwrap_err();
669        assert!(
670            err.to_string().contains("instrument_id"),
671            "Expected missing instrument_id error, was: {err}"
672        );
673    }
674
675    #[rstest]
676    fn test_decode_batch_missing_price_precision_returns_error() {
677        let instrument_id = InstrumentId::from("AAPL.XNAS");
678        let mut metadata = QuoteTick::get_metadata(&instrument_id, 2, 0);
679        metadata.remove(KEY_PRICE_PRECISION);
680
681        let valid_price = (100.00 * FIXED_SCALAR) as PriceRaw;
682        let bid_price =
683            crate::arrow::test_support::decimal_array_from_bytes(vec![&valid_price.to_le_bytes()]);
684        let ask_price =
685            crate::arrow::test_support::decimal_array_from_bytes(vec![&valid_price.to_le_bytes()]);
686        let bid_size = crate::arrow::test_support::decimal_array_from_bytes(vec![
687            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
688        ]);
689        let ask_size = crate::arrow::test_support::decimal_array_from_bytes(vec![
690            &((100.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
691        ]);
692        let ts_event = UInt64Array::from(vec![1]);
693        let ts_init = UInt64Array::from(vec![2]);
694
695        let record_batch = crate::arrow::record_batch_with_timestamps(
696            crate::arrow::schema_without_identifier_column(&QuoteTick::get_schema(Some(
697                metadata.clone(),
698            )))
699            .into(),
700            vec![
701                Arc::new(bid_price),
702                Arc::new(ask_price),
703                Arc::new(bid_size),
704                Arc::new(ask_size),
705                Arc::new(ts_event),
706                Arc::new(ts_init),
707            ],
708        )
709        .unwrap();
710
711        let result = QuoteTick::decode_batch(&metadata, record_batch);
712        assert!(result.is_err());
713        let err = result.unwrap_err();
714        assert!(
715            err.to_string().contains("price_precision"),
716            "Expected missing price_precision error, was: {err}"
717        );
718    }
719
720    #[rstest]
721    fn test_encode_decode_round_trip() {
722        let instrument_id = InstrumentId::from("AAPL.XNAS");
723        let metadata = QuoteTick::get_metadata(&instrument_id, 2, 0);
724
725        let tick1 = QuoteTick {
726            instrument_id,
727            bid_price: Price::from("100.10"),
728            ask_price: Price::from("100.20"),
729            bid_size: Quantity::from(1000),
730            ask_size: Quantity::from(500),
731            ts_event: 1_000_000_000.into(),
732            ts_init: 1_000_000_001.into(),
733        };
734
735        let tick2 = QuoteTick {
736            instrument_id,
737            bid_price: Price::from("100.15"),
738            ask_price: Price::from("100.25"),
739            bid_size: Quantity::from(750),
740            ask_size: Quantity::from(250),
741            ts_event: 2_000_000_000.into(),
742            ts_init: 2_000_000_001.into(),
743        };
744
745        let original = vec![tick1, tick2];
746        let record_batch = QuoteTick::encode_batch(&metadata, &original).unwrap();
747        let decoded = QuoteTick::decode_batch(&metadata, record_batch).unwrap();
748
749        assert_eq!(decoded.len(), original.len());
750        for (orig, dec) in original.iter().zip(decoded.iter()) {
751            assert_eq!(dec.instrument_id, orig.instrument_id);
752            assert_eq!(dec.bid_price, orig.bid_price);
753            assert_eq!(dec.ask_price, orig.ask_price);
754            assert_eq!(dec.bid_size, orig.bid_size);
755            assert_eq!(dec.ask_size, orig.ask_size);
756            assert_eq!(dec.ts_event, orig.ts_event);
757            assert_eq!(dec.ts_init, orig.ts_init);
758        }
759    }
760}
nautilus_serialization/arrow/quote.rs

nautilus_serialization/arrow/
quote.rs
