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

nautilus_serialization/arrow/
trade.rs
