nautilus_serialization/arrow/
trade.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        FixedSizeBinaryArray, FixedSizeBinaryBuilder, StringArray, StringBuilder, StringViewArray,
21        UInt8Array, UInt64Array,
22    },
23    datatypes::{DataType, Field, Schema},
24    error::ArrowError,
25    record_batch::RecordBatch,
26};
27use nautilus_model::{
28    data::TradeTick, enums::AggressorSide, identifiers::TradeId, types::fixed::PRECISION_BYTES,
29};
30
31use super::{
32    DecodeDataFromRecordBatch, EncodingError, decode_price, decode_quantity, extract_column,
33    parse_price_size_metadata, validate_precision_bytes,
34};
35use crate::arrow::{ArrowSchemaProvider, Data, DecodeFromRecordBatch, EncodeToRecordBatch};
36
37impl ArrowSchemaProvider for TradeTick {
38    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
39        let fields = vec![
40            Field::new("price", DataType::FixedSizeBinary(PRECISION_BYTES), false),
41            Field::new("size", DataType::FixedSizeBinary(PRECISION_BYTES), false),
42            Field::new("aggressor_side", DataType::UInt8, false),
43            Field::new("trade_id", DataType::Utf8, false),
44            Field::new("ts_event", DataType::UInt64, false),
45            Field::new("ts_init", DataType::UInt64, false),
46        ];
47
48        match metadata {
49            Some(metadata) => Schema::new_with_metadata(fields, metadata),
50            None => Schema::new(fields),
51        }
52    }
53}
54
55impl EncodeToRecordBatch for TradeTick {
56    fn encode_batch(
57        metadata: &HashMap<String, String>,
58        data: &[Self],
59    ) -> Result<RecordBatch, ArrowError> {
60        let mut price_builder = FixedSizeBinaryBuilder::with_capacity(data.len(), PRECISION_BYTES);
61        let mut size_builder = FixedSizeBinaryBuilder::with_capacity(data.len(), PRECISION_BYTES);
62
63        let mut aggressor_side_builder = UInt8Array::builder(data.len());
64        let mut trade_id_builder = StringBuilder::new();
65        let mut ts_event_builder = UInt64Array::builder(data.len());
66        let mut ts_init_builder = UInt64Array::builder(data.len());
67
68        for tick in data {
69            price_builder
70                .append_value(tick.price.raw().to_le_bytes())
71                .unwrap();
72            size_builder
73                .append_value(tick.size.raw().to_le_bytes())
74                .unwrap();
75            aggressor_side_builder.append_value(tick.aggressor_side as u8);
76            trade_id_builder.append_value(tick.trade_id.to_string());
77            ts_event_builder.append_value(tick.ts_event.as_u64());
78            ts_init_builder.append_value(tick.ts_init.as_u64());
79        }
80
81        let price_array = Arc::new(price_builder.finish());
82        let size_array = Arc::new(size_builder.finish());
83        let aggressor_side_array = Arc::new(aggressor_side_builder.finish());
84        let trade_id_array = Arc::new(trade_id_builder.finish());
85        let ts_event_array = Arc::new(ts_event_builder.finish());
86        let ts_init_array = Arc::new(ts_init_builder.finish());
87
88        RecordBatch::try_new(
89            Self::get_schema(Some(metadata.clone())).into(),
90            vec![
91                price_array,
92                size_array,
93                aggressor_side_array,
94                trade_id_array,
95                ts_event_array,
96                ts_init_array,
97            ],
98        )
99    }
100
101    fn metadata(&self) -> HashMap<String, String> {
102        Self::get_metadata(
103            &self.instrument_id,
104            self.price.precision,
105            self.size.precision,
106        )
107    }
108}
109
110impl DecodeFromRecordBatch for TradeTick {
111    fn decode_batch(
112        metadata: &HashMap<String, String>,
113        record_batch: RecordBatch,
114    ) -> Result<Vec<Self>, EncodingError> {
115        let (instrument_id, price_precision, size_precision) = parse_price_size_metadata(metadata)?;
116        let cols = record_batch.columns();
117
118        let price_values = extract_column::<FixedSizeBinaryArray>(
119            cols,
120            "price",
121            0,
122            DataType::FixedSizeBinary(PRECISION_BYTES),
123        )?;
124
125        let size_values = extract_column::<FixedSizeBinaryArray>(
126            cols,
127            "size",
128            1,
129            DataType::FixedSizeBinary(PRECISION_BYTES),
130        )?;
131
132        validate_precision_bytes(price_values, "price")?;
133        validate_precision_bytes(size_values, "size")?;
134
135        let aggressor_side_values =
136            extract_column::<UInt8Array>(cols, "aggressor_side", 2, DataType::UInt8)?;
137        let ts_event_values = extract_column::<UInt64Array>(cols, "ts_event", 4, DataType::UInt64)?;
138        let ts_init_values = extract_column::<UInt64Array>(cols, "ts_init", 5, DataType::UInt64)?;
139
140        // Datafusion reads trade_ids as StringView
141        let trade_id_values: Vec<TradeId> = if record_batch
142            .schema()
143            .field_with_name("trade_id")?
144            .data_type()
145            == &DataType::Utf8View
146        {
147            extract_column::<StringViewArray>(cols, "trade_id", 3, DataType::Utf8View)?
148                .iter()
149                .enumerate()
150                .map(|(i, id)| {
151                    id.map(TradeId::from).ok_or_else(|| {
152                        EncodingError::ParseError("trade_id", format!("NULL value at row {i}"))
153                    })
154                })
155                .collect::<Result<Vec<_>, _>>()?
156        } else {
157            extract_column::<StringArray>(cols, "trade_id", 3, DataType::Utf8)?
158                .iter()
159                .enumerate()
160                .map(|(i, id)| {
161                    id.map(TradeId::from).ok_or_else(|| {
162                        EncodingError::ParseError("trade_id", format!("NULL value at row {i}"))
163                    })
164                })
165                .collect::<Result<Vec<_>, _>>()?
166        };
167
168        let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
169            .map(|i| {
170                let price = decode_price(price_values.value(i), price_precision, "price", i)?;
171                let size = decode_quantity(size_values.value(i), size_precision, "size", i)?;
172                let aggressor_side_value = aggressor_side_values.value(i);
173                let aggressor_side = AggressorSide::from_repr(aggressor_side_value as usize)
174                    .ok_or_else(|| {
175                        EncodingError::ParseError(
176                            stringify!(AggressorSide),
177                            format!("Invalid enum value, was {aggressor_side_value}"),
178                        )
179                    })?;
180                let trade_id = trade_id_values[i];
181                let ts_event = ts_event_values.value(i).into();
182                let ts_init = ts_init_values.value(i).into();
183
184                Ok(Self {
185                    instrument_id,
186                    price,
187                    size,
188                    aggressor_side,
189                    trade_id,
190                    ts_event,
191                    ts_init,
192                })
193            })
194            .collect();
195
196        result
197    }
198}
199
200impl DecodeDataFromRecordBatch for TradeTick {
201    fn decode_data_batch(
202        metadata: &HashMap<String, String>,
203        record_batch: RecordBatch,
204    ) -> Result<Vec<Data>, EncodingError> {
205        let ticks: Vec<Self> = Self::decode_batch(metadata, record_batch)?;
206        Ok(ticks.into_iter().map(Data::from).collect())
207    }
208}
209
210#[cfg(test)]
211mod tests {
212    use std::sync::Arc;
213
214    use arrow::{
215        array::{Array, UInt8Array, UInt64Array},
216        record_batch::RecordBatch,
217    };
218    use nautilus_model::{
219        identifiers::InstrumentId,
220        types::{Price, Quantity, fixed::FIXED_SCALAR, price::PriceRaw, quantity::QuantityRaw},
221    };
222    use rstest::rstest;
223
224    use super::*;
225    use crate::arrow::{
226        KEY_INSTRUMENT_ID, KEY_PRICE_PRECISION, fixed_size_binary, get_raw_price, get_raw_quantity,
227    };
228
229    #[rstest]
230    fn test_get_schema() {
231        let instrument_id = InstrumentId::from("AAPL.XNAS");
232        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
233        let schema = TradeTick::get_schema(Some(metadata.clone()));
234
235        let mut expected_fields = Vec::with_capacity(6);
236
237        expected_fields.push(Field::new(
238            "price",
239            DataType::FixedSizeBinary(PRECISION_BYTES),
240            false,
241        ));
242
243        expected_fields.extend(vec![
244            Field::new("size", DataType::FixedSizeBinary(PRECISION_BYTES), false),
245            Field::new("aggressor_side", DataType::UInt8, false),
246            Field::new("trade_id", DataType::Utf8, false),
247            Field::new("ts_event", DataType::UInt64, false),
248            Field::new("ts_init", DataType::UInt64, false),
249        ]);
250
251        let expected_schema = Schema::new_with_metadata(expected_fields, metadata);
252        assert_eq!(schema, expected_schema);
253    }
254
255    #[rstest]
256    fn test_get_schema_map() {
257        let schema_map = TradeTick::get_schema_map();
258        let mut expected_map = HashMap::new();
259
260        let precision_bytes = format!("FixedSizeBinary({PRECISION_BYTES})");
261        expected_map.insert("price".to_string(), precision_bytes.clone());
262        expected_map.insert("size".to_string(), precision_bytes);
263        expected_map.insert("aggressor_side".to_string(), "UInt8".to_string());
264        expected_map.insert("trade_id".to_string(), "Utf8".to_string());
265        expected_map.insert("ts_event".to_string(), "UInt64".to_string());
266        expected_map.insert("ts_init".to_string(), "UInt64".to_string());
267        assert_eq!(schema_map, expected_map);
268    }
269
270    #[rstest]
271    fn test_encode_trade_tick() {
272        let instrument_id = InstrumentId::from("AAPL.XNAS");
273        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
274
275        let tick1 = TradeTick {
276            instrument_id,
277            price: Price::from("100.10"),
278            size: Quantity::from(1000),
279            aggressor_side: AggressorSide::Buy,
280            trade_id: TradeId::new("1"),
281            ts_event: 1.into(),
282            ts_init: 3.into(),
283        };
284
285        let tick2 = TradeTick {
286            instrument_id,
287            price: Price::from("100.50"),
288            size: Quantity::from(500),
289            aggressor_side: AggressorSide::Sell,
290            trade_id: TradeId::new("2"),
291            ts_event: 2.into(),
292            ts_init: 4.into(),
293        };
294
295        let data = vec![tick1, tick2];
296        let record_batch = TradeTick::encode_batch(&metadata, &data).unwrap();
297        let columns = record_batch.columns();
298
299        let price_values = columns[0]
300            .as_any()
301            .downcast_ref::<FixedSizeBinaryArray>()
302            .unwrap();
303        assert_eq!(
304            get_raw_price(price_values.value(0)),
305            (100.10 * FIXED_SCALAR) as PriceRaw
306        );
307        assert_eq!(
308            get_raw_price(price_values.value(1)),
309            (100.50 * FIXED_SCALAR) as PriceRaw
310        );
311
312        let size_values = columns[1]
313            .as_any()
314            .downcast_ref::<FixedSizeBinaryArray>()
315            .unwrap();
316        assert_eq!(
317            get_raw_quantity(size_values.value(0)),
318            (1000.0 * FIXED_SCALAR) as QuantityRaw
319        );
320        assert_eq!(
321            get_raw_quantity(size_values.value(1)),
322            (500.0 * FIXED_SCALAR) as QuantityRaw
323        );
324
325        let aggressor_side_values = columns[2].as_any().downcast_ref::<UInt8Array>().unwrap();
326        let trade_id_values = columns[3].as_any().downcast_ref::<StringArray>().unwrap();
327        let ts_event_values = columns[4].as_any().downcast_ref::<UInt64Array>().unwrap();
328        let ts_init_values = columns[5].as_any().downcast_ref::<UInt64Array>().unwrap();
329
330        assert_eq!(columns.len(), 6);
331        assert_eq!(size_values.len(), 2);
332        assert_eq!(
333            get_raw_quantity(size_values.value(0)),
334            (1000.0 * FIXED_SCALAR) as QuantityRaw
335        );
336        assert_eq!(
337            get_raw_quantity(size_values.value(1)),
338            (500.0 * FIXED_SCALAR) as QuantityRaw
339        );
340        assert_eq!(aggressor_side_values.len(), 2);
341        assert_eq!(aggressor_side_values.value(0), 1);
342        assert_eq!(aggressor_side_values.value(1), 2);
343        assert_eq!(trade_id_values.len(), 2);
344        assert_eq!(trade_id_values.value(0), "1");
345        assert_eq!(trade_id_values.value(1), "2");
346        assert_eq!(ts_event_values.len(), 2);
347        assert_eq!(ts_event_values.value(0), 1);
348        assert_eq!(ts_event_values.value(1), 2);
349        assert_eq!(ts_init_values.len(), 2);
350        assert_eq!(ts_init_values.value(0), 3);
351        assert_eq!(ts_init_values.value(1), 4);
352    }
353
354    #[rstest]
355    fn test_decode_batch() {
356        let instrument_id = InstrumentId::from("AAPL.XNAS");
357        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
358
359        let raw_price1 = (100.00 * FIXED_SCALAR) as PriceRaw;
360        let raw_price2 = (101.00 * FIXED_SCALAR) as PriceRaw;
361        let price = fixed_size_binary(vec![&raw_price1.to_le_bytes(), &raw_price2.to_le_bytes()]);
362
363        let size = fixed_size_binary(vec![
364            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
365            &((900.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
366        ]);
367        let aggressor_side = UInt8Array::from(vec![0, 1]); // 0 for BUY, 1 for SELL
368        let trade_id = StringArray::from(vec!["1", "2"]);
369        let ts_event = UInt64Array::from(vec![1, 2]);
370        let ts_init = UInt64Array::from(vec![3, 4]);
371
372        let record_batch = RecordBatch::try_new(
373            TradeTick::get_schema(Some(metadata.clone())).into(),
374            vec![
375                Arc::new(price),
376                Arc::new(size),
377                Arc::new(aggressor_side),
378                Arc::new(trade_id),
379                Arc::new(ts_event),
380                Arc::new(ts_init),
381            ],
382        )
383        .unwrap();
384
385        let decoded_data = TradeTick::decode_batch(&metadata, record_batch).unwrap();
386        assert_eq!(decoded_data.len(), 2);
387        assert_eq!(decoded_data[0].price, Price::from_raw(raw_price1, 2));
388        assert_eq!(decoded_data[1].price, Price::from_raw(raw_price2, 2));
389    }
390
391    #[rstest]
392    fn test_decode_batch_null_trade_id_returns_error() {
393        use arrow::datatypes::Field;
394
395        let instrument_id = InstrumentId::from("AAPL.XNAS");
396        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
397
398        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
399        let price = fixed_size_binary(vec![&raw_price.to_le_bytes()]);
400        let size = fixed_size_binary(vec![
401            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
402        ]);
403        let aggressor_side = UInt8Array::from(vec![0]);
404
405        let trade_id: StringArray = vec![None::<&str>].into();
406        let ts_event = UInt64Array::from(vec![1]);
407        let ts_init = UInt64Array::from(vec![2]);
408
409        // Create schema with nullable trade_id to simulate external data source
410        let fields = vec![
411            Field::new("price", DataType::FixedSizeBinary(PRECISION_BYTES), false),
412            Field::new("size", DataType::FixedSizeBinary(PRECISION_BYTES), false),
413            Field::new("aggressor_side", DataType::UInt8, false),
414            Field::new("trade_id", DataType::Utf8, true), // nullable
415            Field::new("ts_event", DataType::UInt64, false),
416            Field::new("ts_init", DataType::UInt64, false),
417        ];
418        let schema = Schema::new_with_metadata(fields, metadata.clone());
419
420        let record_batch = RecordBatch::try_new(
421            schema.into(),
422            vec![
423                Arc::new(price),
424                Arc::new(size),
425                Arc::new(aggressor_side),
426                Arc::new(trade_id),
427                Arc::new(ts_event),
428                Arc::new(ts_init),
429            ],
430        )
431        .unwrap();
432
433        let result = TradeTick::decode_batch(&metadata, record_batch);
434        assert!(result.is_err());
435        let err = result.unwrap_err();
436        assert!(
437            err.to_string().contains("NULL value at row 0"),
438            "Expected NULL error, was: {err}"
439        );
440    }
441
442    #[rstest]
443    fn test_decode_batch_invalid_price_returns_error() {
444        let instrument_id = InstrumentId::from("AAPL.XNAS");
445        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
446
447        let invalid_price: PriceRaw = PriceRaw::MAX - 1000;
448        let price = fixed_size_binary(vec![&invalid_price.to_le_bytes()]);
449        let size = fixed_size_binary(vec![
450            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
451        ]);
452        let aggressor_side = UInt8Array::from(vec![0]);
453        let trade_id = StringArray::from(vec!["1"]);
454        let ts_event = UInt64Array::from(vec![1]);
455        let ts_init = UInt64Array::from(vec![2]);
456
457        let record_batch = RecordBatch::try_new(
458            TradeTick::get_schema(Some(metadata.clone())).into(),
459            vec![
460                Arc::new(price),
461                Arc::new(size),
462                Arc::new(aggressor_side),
463                Arc::new(trade_id),
464                Arc::new(ts_event),
465                Arc::new(ts_init),
466            ],
467        )
468        .unwrap();
469
470        let result = TradeTick::decode_batch(&metadata, record_batch);
471        assert!(result.is_err());
472        let err = result.unwrap_err();
473        assert!(
474            err.to_string().contains("price") && err.to_string().contains("row 0"),
475            "Expected price error at row 0, was: {err}"
476        );
477    }
478
479    #[rstest]
480    fn test_decode_batch_invalid_size_returns_error() {
481        use nautilus_model::types::{fixed::FIXED_PRECISION, quantity::QUANTITY_RAW_MAX};
482
483        let instrument_id = InstrumentId::from("AAPL.XNAS");
484        // Decode the size at full precision so the out-of-range raw value bypasses the
485        // precision-0 correction, which would otherwise round it back within the bound.
486        let metadata = TradeTick::get_metadata(&instrument_id, 2, FIXED_PRECISION);
487
488        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
489        let price = fixed_size_binary(vec![&raw_price.to_le_bytes()]);
490
491        let invalid_size = QUANTITY_RAW_MAX + 1;
492        let size = fixed_size_binary(vec![&invalid_size.to_le_bytes()]);
493        let aggressor_side = UInt8Array::from(vec![0]);
494        let trade_id = StringArray::from(vec!["1"]);
495        let ts_event = UInt64Array::from(vec![1]);
496        let ts_init = UInt64Array::from(vec![2]);
497
498        let record_batch = RecordBatch::try_new(
499            TradeTick::get_schema(Some(metadata.clone())).into(),
500            vec![
501                Arc::new(price),
502                Arc::new(size),
503                Arc::new(aggressor_side),
504                Arc::new(trade_id),
505                Arc::new(ts_event),
506                Arc::new(ts_init),
507            ],
508        )
509        .unwrap();
510
511        let result = TradeTick::decode_batch(&metadata, record_batch);
512        assert!(result.is_err());
513        let err = result.unwrap_err();
514        assert!(
515            err.to_string().contains("size") && err.to_string().contains("row 0"),
516            "Expected size error at row 0, was: {err}"
517        );
518    }
519
520    #[rstest]
521    fn test_decode_batch_invalid_aggressor_side_returns_error() {
522        let instrument_id = InstrumentId::from("AAPL.XNAS");
523        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
524
525        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
526        let price = fixed_size_binary(vec![&raw_price.to_le_bytes()]);
527        let size = fixed_size_binary(vec![
528            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
529        ]);
530
531        let aggressor_side = UInt8Array::from(vec![99]);
532        let trade_id = StringArray::from(vec!["1"]);
533        let ts_event = UInt64Array::from(vec![1]);
534        let ts_init = UInt64Array::from(vec![2]);
535
536        let record_batch = RecordBatch::try_new(
537            TradeTick::get_schema(Some(metadata.clone())).into(),
538            vec![
539                Arc::new(price),
540                Arc::new(size),
541                Arc::new(aggressor_side),
542                Arc::new(trade_id),
543                Arc::new(ts_event),
544                Arc::new(ts_init),
545            ],
546        )
547        .unwrap();
548
549        let result = TradeTick::decode_batch(&metadata, record_batch);
550        assert!(result.is_err());
551        let err = result.unwrap_err();
552        assert!(
553            err.to_string().contains("AggressorSide"),
554            "Expected AggressorSide error, was: {err}"
555        );
556    }
557
558    #[rstest]
559    fn test_decode_batch_missing_instrument_id_returns_error() {
560        let instrument_id = InstrumentId::from("AAPL.XNAS");
561        let mut metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
562        metadata.remove(KEY_INSTRUMENT_ID);
563
564        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
565        let price = fixed_size_binary(vec![&raw_price.to_le_bytes()]);
566        let size = fixed_size_binary(vec![
567            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
568        ]);
569        let aggressor_side = UInt8Array::from(vec![0]);
570        let trade_id = StringArray::from(vec!["1"]);
571        let ts_event = UInt64Array::from(vec![1]);
572        let ts_init = UInt64Array::from(vec![2]);
573
574        let record_batch = RecordBatch::try_new(
575            TradeTick::get_schema(Some(metadata.clone())).into(),
576            vec![
577                Arc::new(price),
578                Arc::new(size),
579                Arc::new(aggressor_side),
580                Arc::new(trade_id),
581                Arc::new(ts_event),
582                Arc::new(ts_init),
583            ],
584        )
585        .unwrap();
586
587        let result = TradeTick::decode_batch(&metadata, record_batch);
588        assert!(result.is_err());
589        let err = result.unwrap_err();
590        assert!(
591            err.to_string().contains("instrument_id"),
592            "Expected missing instrument_id error, was: {err}"
593        );
594    }
595
596    #[rstest]
597    fn test_decode_batch_missing_price_precision_returns_error() {
598        let instrument_id = InstrumentId::from("AAPL.XNAS");
599        let mut metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
600        metadata.remove(KEY_PRICE_PRECISION);
601
602        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
603        let price = fixed_size_binary(vec![&raw_price.to_le_bytes()]);
604        let size = fixed_size_binary(vec![
605            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
606        ]);
607        let aggressor_side = UInt8Array::from(vec![0]);
608        let trade_id = StringArray::from(vec!["1"]);
609        let ts_event = UInt64Array::from(vec![1]);
610        let ts_init = UInt64Array::from(vec![2]);
611
612        let record_batch = RecordBatch::try_new(
613            TradeTick::get_schema(Some(metadata.clone())).into(),
614            vec![
615                Arc::new(price),
616                Arc::new(size),
617                Arc::new(aggressor_side),
618                Arc::new(trade_id),
619                Arc::new(ts_event),
620                Arc::new(ts_init),
621            ],
622        )
623        .unwrap();
624
625        let result = TradeTick::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() {
636        let instrument_id = InstrumentId::from("AAPL.XNAS");
637        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
638
639        let tick1 = TradeTick {
640            instrument_id,
641            price: Price::from("100.10"),
642            size: Quantity::from(1000),
643            aggressor_side: AggressorSide::Buy,
644            trade_id: TradeId::new("trade-123"),
645            ts_event: 1_000_000_000.into(),
646            ts_init: 1_000_000_001.into(),
647        };
648
649        let tick2 = TradeTick {
650            instrument_id,
651            price: Price::from("100.50"),
652            size: Quantity::from(500),
653            aggressor_side: AggressorSide::Sell,
654            trade_id: TradeId::new("trade-456"),
655            ts_event: 2_000_000_000.into(),
656            ts_init: 2_000_000_001.into(),
657        };
658
659        let original = vec![tick1, tick2];
660        let record_batch = TradeTick::encode_batch(&metadata, &original).unwrap();
661        let decoded = TradeTick::decode_batch(&metadata, record_batch).unwrap();
662
663        assert_eq!(decoded.len(), original.len());
664        for (orig, dec) in original.iter().zip(decoded.iter()) {
665            assert_eq!(dec.instrument_id, orig.instrument_id);
666            assert_eq!(dec.price, orig.price);
667            assert_eq!(dec.size, orig.size);
668            assert_eq!(dec.aggressor_side, orig.aggressor_side);
669            assert_eq!(dec.trade_id, orig.trade_id);
670            assert_eq!(dec.ts_event, orig.ts_event);
671            assert_eq!(dec.ts_init, orig.ts_init);
672        }
673    }
674}
nautilus_serialization/arrow/trade.rs

nautilus_serialization/arrow/
trade.rs
