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,
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        let ts_event_values = extract_column::<UInt64Array>(cols, "ts_event", 4, DataType::UInt64)?;
139        let ts_init_values = extract_column::<UInt64Array>(cols, "ts_init", 5, DataType::UInt64)?;
140
141        // Datafusion reads trade_ids as StringView
142        let trade_id_values: Vec<TradeId> = if record_batch
143            .schema()
144            .field_with_name("trade_id")?
145            .data_type()
146            == &DataType::Utf8View
147        {
148            extract_column::<StringViewArray>(cols, "trade_id", 3, DataType::Utf8View)?
149                .iter()
150                .enumerate()
151                .map(|(i, id)| {
152                    id.map(TradeId::from).ok_or_else(|| {
153                        EncodingError::ParseError("trade_id", format!("NULL value at row {i}"))
154                    })
155                })
156                .collect::<Result<Vec<_>, _>>()?
157        } else {
158            extract_column::<StringArray>(cols, "trade_id", 3, DataType::Utf8)?
159                .iter()
160                .enumerate()
161                .map(|(i, id)| {
162                    id.map(TradeId::from).ok_or_else(|| {
163                        EncodingError::ParseError("trade_id", format!("NULL value at row {i}"))
164                    })
165                })
166                .collect::<Result<Vec<_>, _>>()?
167        };
168
169        let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
170            .map(|i| {
171                let price =
172                    decode_required_decimal_price(price_values, price_precision, "price", i)?;
173                let size =
174                    decode_required_decimal_quantity(size_values, size_precision, "size", i)?;
175                let aggressor_side_value = aggressor_side_values.value(i);
176                let aggressor_side =
177                    AggressorSide::from_str(aggressor_side_value).map_err(|e| {
178                        EncodingError::ParseError(stringify!(AggressorSide), e.to_string())
179                    })?;
180                let trade_id = trade_id_values[i];
181                let ts_event = decode_required_timestamp(ts_event_values, "ts_event", i)?;
182                let ts_init = decode_required_timestamp(ts_init_values, "ts_init", i)?;
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::array::{Array, Decimal128Array, TimestampNanosecondArray, UInt64Array};
215    use nautilus_model::types::{
216        Price, Quantity, fixed::FIXED_SCALAR, price::PriceRaw, quantity::QuantityRaw,
217    };
218    use rstest::rstest;
219
220    use super::*;
221    use crate::arrow::{KEY_TYPE_NAME, get_raw_price, get_raw_quantity};
222
223    #[rstest]
224    fn test_get_schema() {
225        let instrument_id = InstrumentId::from("AAPL.XNAS");
226        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
227        let schema = TradeTick::get_schema(Some(metadata.clone()));
228
229        let mut expected_fields = Vec::with_capacity(7);
230
231        expected_fields.push(Field::new("price", fixed_decimal_data_type(), false));
232
233        expected_fields.extend(vec![
234            Field::new("size", fixed_decimal_data_type(), false),
235            Field::new("aggressor_side", enum_dictionary_data_type(), false),
236            Field::new("trade_id", DataType::Utf8, false),
237            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
238            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
239            Field::new(KEY_IDENTIFIER, DataType::Utf8, true),
240        ]);
241
242        let mut expected_metadata = metadata;
243        expected_metadata.insert(KEY_TYPE_NAME.to_string(), "TradeTick".to_string());
244        let expected_schema = Schema::new_with_metadata(expected_fields, expected_metadata);
245        assert_eq!(schema, expected_schema);
246    }
247
248    #[rstest]
249    fn test_get_schema_map() {
250        let schema_map = TradeTick::get_schema_map();
251        let mut expected_map = HashMap::new();
252
253        let precision_bytes = "Decimal128(38, 16)".to_string();
254        expected_map.insert("price".to_string(), precision_bytes.clone());
255        expected_map.insert("size".to_string(), precision_bytes);
256        expected_map.insert(
257            "aggressor_side".to_string(),
258            "Dictionary(Int8, Utf8)".to_string(),
259        );
260        expected_map.insert("trade_id".to_string(), "Utf8".to_string());
261        expected_map.insert(
262            "ts_event".to_string(),
263            "Timestamp(Nanosecond, Some(\"UTC\"))".to_string(),
264        );
265        expected_map.insert(
266            "ts_init".to_string(),
267            "Timestamp(Nanosecond, Some(\"UTC\"))".to_string(),
268        );
269        expected_map.insert(KEY_IDENTIFIER.to_string(), "Utf8".to_string());
270        assert_eq!(schema_map, expected_map);
271    }
272
273    #[rstest]
274    fn test_encode_trade_tick() {
275        let instrument_id = InstrumentId::from("AAPL.XNAS");
276        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
277
278        let tick1 = TradeTick {
279            instrument_id,
280            price: Price::from("100.10"),
281            size: Quantity::from(1000),
282            aggressor_side: AggressorSide::Buy,
283            trade_id: TradeId::new("1"),
284            ts_event: 1.into(),
285            ts_init: 3.into(),
286        };
287
288        let tick2 = TradeTick {
289            instrument_id,
290            price: Price::from("100.50"),
291            size: Quantity::from(500),
292            aggressor_side: AggressorSide::Sell,
293            trade_id: TradeId::new("2"),
294            ts_event: 2.into(),
295            ts_init: 4.into(),
296        };
297
298        let data = vec![tick1, tick2];
299        let record_batch = TradeTick::encode_batch(&metadata, &data).unwrap();
300        let columns = record_batch.columns();
301
302        let price_values = columns[0]
303            .as_any()
304            .downcast_ref::<Decimal128Array>()
305            .unwrap();
306        assert_eq!(
307            get_raw_price(price_values.value(0)),
308            (100.10 * FIXED_SCALAR) as PriceRaw
309        );
310        assert_eq!(
311            get_raw_price(price_values.value(1)),
312            (100.50 * FIXED_SCALAR) as PriceRaw
313        );
314
315        let size_values = columns[1]
316            .as_any()
317            .downcast_ref::<Decimal128Array>()
318            .unwrap();
319        assert_eq!(
320            get_raw_quantity(size_values.value(0)),
321            (1000.0 * FIXED_SCALAR) as QuantityRaw
322        );
323        assert_eq!(
324            get_raw_quantity(size_values.value(1)),
325            (500.0 * FIXED_SCALAR) as QuantityRaw
326        );
327
328        let aggressor_side_values = extract_column_string(columns, "aggressor_side", 2).unwrap();
329        let trade_id_values = columns[3].as_any().downcast_ref::<StringArray>().unwrap();
330        let ts_event_values = columns[4]
331            .as_any()
332            .downcast_ref::<TimestampNanosecondArray>()
333            .unwrap();
334        let ts_init_values = columns[5]
335            .as_any()
336            .downcast_ref::<TimestampNanosecondArray>()
337            .unwrap();
338
339        assert_eq!(columns.len(), 7);
340        assert_eq!(size_values.len(), 2);
341        assert_eq!(
342            get_raw_quantity(size_values.value(0)),
343            (1000.0 * FIXED_SCALAR) as QuantityRaw
344        );
345        assert_eq!(
346            get_raw_quantity(size_values.value(1)),
347            (500.0 * FIXED_SCALAR) as QuantityRaw
348        );
349        assert_eq!(aggressor_side_values.len(), 2);
350        assert_eq!(aggressor_side_values.value(0), "BUY");
351        assert_eq!(aggressor_side_values.value(1), "SELL");
352        assert_eq!(trade_id_values.len(), 2);
353        assert_eq!(trade_id_values.value(0), "1");
354        assert_eq!(trade_id_values.value(1), "2");
355        assert_eq!(ts_event_values.len(), 2);
356        assert_eq!(ts_event_values.value(0), 1);
357        assert_eq!(ts_event_values.value(1), 2);
358        assert_eq!(ts_init_values.len(), 2);
359        assert_eq!(ts_init_values.value(0), 3);
360        assert_eq!(ts_init_values.value(1), 4);
361    }
362
363    #[rstest]
364    fn test_decode_batch() {
365        let instrument_id = InstrumentId::from("AAPL.XNAS");
366        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
367
368        let raw_price1 = (100.00 * FIXED_SCALAR) as PriceRaw;
369        let raw_price2 = (101.00 * FIXED_SCALAR) as PriceRaw;
370        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
371            &raw_price1.to_le_bytes(),
372            &raw_price2.to_le_bytes(),
373        ]);
374
375        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
376            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
377            &((900.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
378        ]);
379        let aggressor_side =
380            enum_dictionary_array([AggressorSide::NoAggressor, AggressorSide::Buy]).unwrap();
381        let trade_id = StringArray::from(vec!["1", "2"]);
382        let ts_event = UInt64Array::from(vec![1, 2]);
383        let ts_init = UInt64Array::from(vec![3, 4]);
384
385        let record_batch = crate::arrow::record_batch_with_timestamps(
386            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
387                metadata.clone(),
388            )))
389            .into(),
390            vec![
391                Arc::new(price),
392                Arc::new(size),
393                Arc::new(aggressor_side),
394                Arc::new(trade_id),
395                Arc::new(ts_event),
396                Arc::new(ts_init),
397            ],
398        )
399        .unwrap();
400
401        let decoded_data = TradeTick::decode_batch(&metadata, record_batch).unwrap();
402        assert_eq!(decoded_data.len(), 2);
403        assert_eq!(decoded_data[0].price, Price::from_raw(raw_price1, 2));
404        assert_eq!(decoded_data[1].price, Price::from_raw(raw_price2, 2));
405    }
406
407    #[rstest]
408    fn test_decode_batch_null_trade_id_returns_error() {
409        use arrow::datatypes::Field;
410
411        let instrument_id = InstrumentId::from("AAPL.XNAS");
412        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
413
414        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
415        let price =
416            crate::arrow::test_support::decimal_array_from_bytes(vec![&raw_price.to_le_bytes()]);
417        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
418            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
419        ]);
420        let aggressor_side = enum_dictionary_array([AggressorSide::NoAggressor]).unwrap();
421
422        let trade_id: StringArray = vec![None::<&str>].into();
423        let ts_event = UInt64Array::from(vec![1]);
424        let ts_init = UInt64Array::from(vec![2]);
425
426        // Create schema with nullable trade_id to simulate external data source
427        let fields = vec![
428            Field::new("price", fixed_decimal_data_type(), false),
429            Field::new("size", fixed_decimal_data_type(), false),
430            Field::new("aggressor_side", enum_dictionary_data_type(), false),
431            Field::new("trade_id", DataType::Utf8, true), // nullable
432            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
433            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
434        ];
435        let schema = Schema::new_with_metadata(fields, metadata.clone());
436
437        let record_batch = crate::arrow::record_batch_with_timestamps(
438            schema.into(),
439            vec![
440                Arc::new(price),
441                Arc::new(size),
442                Arc::new(aggressor_side),
443                Arc::new(trade_id),
444                Arc::new(ts_event),
445                Arc::new(ts_init),
446            ],
447        )
448        .unwrap();
449
450        let result = TradeTick::decode_batch(&metadata, record_batch);
451        assert!(result.is_err());
452        let err = result.unwrap_err();
453        assert!(
454            err.to_string().contains("NULL value at row 0"),
455            "Expected NULL error, was: {err}"
456        );
457    }
458
459    #[rstest]
460    fn test_decode_batch_invalid_price_returns_error() {
461        let instrument_id = InstrumentId::from("AAPL.XNAS");
462        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
463
464        let invalid_price: PriceRaw = PriceRaw::MAX - 1000;
465        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
466            &invalid_price.to_le_bytes(),
467        ]);
468        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
469            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
470        ]);
471        let aggressor_side = enum_dictionary_array([AggressorSide::NoAggressor]).unwrap();
472        let trade_id = StringArray::from(vec!["1"]);
473        let ts_event = UInt64Array::from(vec![1]);
474        let ts_init = UInt64Array::from(vec![2]);
475
476        let record_batch = crate::arrow::record_batch_with_timestamps(
477            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
478                metadata.clone(),
479            )))
480            .into(),
481            vec![
482                Arc::new(price),
483                Arc::new(size),
484                Arc::new(aggressor_side),
485                Arc::new(trade_id),
486                Arc::new(ts_event),
487                Arc::new(ts_init),
488            ],
489        )
490        .unwrap();
491
492        let result = TradeTick::decode_batch(&metadata, record_batch);
493        assert!(result.is_err());
494        let err = result.unwrap_err();
495        assert!(
496            err.to_string().contains("price") && err.to_string().contains("row 0"),
497            "Expected price error at row 0, was: {err}"
498        );
499    }
500
501    #[rstest]
502    fn test_decode_batch_invalid_size_returns_error() {
503        use nautilus_model::types::{fixed::FIXED_PRECISION, quantity::QUANTITY_RAW_MAX};
504
505        let instrument_id = InstrumentId::from("AAPL.XNAS");
506        // Decode the size at full precision so the out-of-range raw value bypasses the
507        // precision-0 correction, which would otherwise round it back within the bound.
508        let metadata = TradeTick::get_metadata(&instrument_id, 2, FIXED_PRECISION);
509
510        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
511        let price =
512            crate::arrow::test_support::decimal_array_from_bytes(vec![&raw_price.to_le_bytes()]);
513
514        let invalid_size = QUANTITY_RAW_MAX + 1;
515        let size =
516            crate::arrow::test_support::decimal_array_from_bytes(vec![&invalid_size.to_le_bytes()]);
517        let aggressor_side = enum_dictionary_array([AggressorSide::NoAggressor]).unwrap();
518        let trade_id = StringArray::from(vec!["1"]);
519        let ts_event = UInt64Array::from(vec![1]);
520        let ts_init = UInt64Array::from(vec![2]);
521
522        let record_batch = crate::arrow::record_batch_with_timestamps(
523            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
524                metadata.clone(),
525            )))
526            .into(),
527            vec![
528                Arc::new(price),
529                Arc::new(size),
530                Arc::new(aggressor_side),
531                Arc::new(trade_id),
532                Arc::new(ts_event),
533                Arc::new(ts_init),
534            ],
535        )
536        .unwrap();
537
538        let result = TradeTick::decode_batch(&metadata, record_batch);
539        assert!(result.is_err());
540        let err = result.unwrap_err();
541        assert!(
542            err.to_string().contains("size") && err.to_string().contains("row 0"),
543            "Expected size error at row 0, was: {err}"
544        );
545    }
546
547    #[rstest]
548    fn test_decode_batch_invalid_aggressor_side_returns_error() {
549        let instrument_id = InstrumentId::from("AAPL.XNAS");
550        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
551
552        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
553        let price =
554            crate::arrow::test_support::decimal_array_from_bytes(vec![&raw_price.to_le_bytes()]);
555        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
556            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
557        ]);
558
559        let aggressor_side = enum_dictionary_array(["INVALID"]).unwrap();
560        let trade_id = StringArray::from(vec!["1"]);
561        let ts_event = UInt64Array::from(vec![1]);
562        let ts_init = UInt64Array::from(vec![2]);
563
564        let record_batch = crate::arrow::record_batch_with_timestamps(
565            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
566                metadata.clone(),
567            )))
568            .into(),
569            vec![
570                Arc::new(price),
571                Arc::new(size),
572                Arc::new(aggressor_side),
573                Arc::new(trade_id),
574                Arc::new(ts_event),
575                Arc::new(ts_init),
576            ],
577        )
578        .unwrap();
579
580        let result = TradeTick::decode_batch(&metadata, record_batch);
581        assert!(result.is_err());
582        let err = result.unwrap_err();
583        assert!(
584            err.to_string().contains("AggressorSide"),
585            "Expected AggressorSide error, was: {err}"
586        );
587    }
588
589    #[rstest]
590    fn test_decode_batch_missing_instrument_id_returns_error() {
591        let instrument_id = InstrumentId::from("AAPL.XNAS");
592        let mut metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
593        metadata.remove(KEY_INSTRUMENT_ID);
594
595        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
596        let price =
597            crate::arrow::test_support::decimal_array_from_bytes(vec![&raw_price.to_le_bytes()]);
598        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
599            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
600        ]);
601        let aggressor_side = enum_dictionary_array([AggressorSide::NoAggressor]).unwrap();
602        let trade_id = StringArray::from(vec!["1"]);
603        let ts_event = UInt64Array::from(vec![1]);
604        let ts_init = UInt64Array::from(vec![2]);
605
606        let record_batch = crate::arrow::record_batch_with_timestamps(
607            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
608                metadata.clone(),
609            )))
610            .into(),
611            vec![
612                Arc::new(price),
613                Arc::new(size),
614                Arc::new(aggressor_side),
615                Arc::new(trade_id),
616                Arc::new(ts_event),
617                Arc::new(ts_init),
618            ],
619        )
620        .unwrap();
621
622        let result = TradeTick::decode_batch(&metadata, record_batch);
623        assert!(result.is_err());
624        let err = result.unwrap_err();
625        assert!(
626            err.to_string().contains("instrument_id"),
627            "Expected missing instrument_id error, was: {err}"
628        );
629    }
630
631    #[rstest]
632    fn test_decode_batch_missing_price_precision_returns_error() {
633        let instrument_id = InstrumentId::from("AAPL.XNAS");
634        let mut metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
635        metadata.remove(KEY_PRICE_PRECISION);
636
637        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
638        let price =
639            crate::arrow::test_support::decimal_array_from_bytes(vec![&raw_price.to_le_bytes()]);
640        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
641            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
642        ]);
643        let aggressor_side = enum_dictionary_array([AggressorSide::NoAggressor]).unwrap();
644        let trade_id = StringArray::from(vec!["1"]);
645        let ts_event = UInt64Array::from(vec![1]);
646        let ts_init = UInt64Array::from(vec![2]);
647
648        let record_batch = crate::arrow::record_batch_with_timestamps(
649            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
650                metadata.clone(),
651            )))
652            .into(),
653            vec![
654                Arc::new(price),
655                Arc::new(size),
656                Arc::new(aggressor_side),
657                Arc::new(trade_id),
658                Arc::new(ts_event),
659                Arc::new(ts_init),
660            ],
661        )
662        .unwrap();
663
664        let result = TradeTick::decode_batch(&metadata, record_batch);
665        assert!(result.is_err());
666        let err = result.unwrap_err();
667        assert!(
668            err.to_string().contains("price_precision"),
669            "Expected missing price_precision error, was: {err}"
670        );
671    }
672
673    #[rstest]
674    fn test_encode_decode_round_trip() {
675        let instrument_id = InstrumentId::from("AAPL.XNAS");
676        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
677
678        let tick1 = TradeTick {
679            instrument_id,
680            price: Price::from("100.10"),
681            size: Quantity::from(1000),
682            aggressor_side: AggressorSide::Buy,
683            trade_id: TradeId::new("trade-123"),
684            ts_event: 1_000_000_000.into(),
685            ts_init: 1_000_000_001.into(),
686        };
687
688        let tick2 = TradeTick {
689            instrument_id,
690            price: Price::from("100.50"),
691            size: Quantity::from(500),
692            aggressor_side: AggressorSide::Sell,
693            trade_id: TradeId::new("trade-456"),
694            ts_event: 2_000_000_000.into(),
695            ts_init: 2_000_000_001.into(),
696        };
697
698        let original = vec![tick1, tick2];
699        let record_batch = TradeTick::encode_batch(&metadata, &original).unwrap();
700        let decoded = TradeTick::decode_batch(&metadata, record_batch).unwrap();
701
702        assert_eq!(decoded.len(), original.len());
703        for (orig, dec) in original.iter().zip(decoded.iter()) {
704            assert_eq!(dec.instrument_id, orig.instrument_id);
705            assert_eq!(dec.price, orig.price);
706            assert_eq!(dec.size, orig.size);
707            assert_eq!(dec.aggressor_side, orig.aggressor_side);
708            assert_eq!(dec.trade_id, orig.trade_id);
709            assert_eq!(dec.ts_event, orig.ts_event);
710            assert_eq!(dec.ts_init, orig.ts_init);
711        }
712    }
713}
nautilus_serialization/arrow/trade.rs

nautilus_serialization/arrow/
trade.rs
