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, parse_metadata, required_price_decimal_array,
33    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(), true),
43            Field::new("size", fixed_decimal_data_type(), true),
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        match metadata {
52            Some(metadata) => Schema::new_with_metadata(fields, metadata),
53            None => Schema::new(fields),
54        }
55    }
56}
57
58impl EncodeToRecordBatch for TradeTick {
59    fn encode_batch<T>(
60        metadata: &HashMap<String, String>,
61        data: &[T],
62    ) -> Result<RecordBatch, ArrowError>
63    where
64        T: std::borrow::Borrow<Self>,
65    {
66        let mut trade_id_builder = StringBuilder::new();
67        let mut ts_event_builder = UInt64Array::builder(data.len());
68        let mut ts_init_builder = UInt64Array::builder(data.len());
69
70        for tick in data.iter().map(std::borrow::Borrow::borrow) {
71            trade_id_builder.append_value(tick.trade_id.to_string());
72            ts_event_builder.append_value(tick.ts_event.as_u64());
73            ts_init_builder.append_value(tick.ts_init.as_u64());
74        }
75
76        let price_array = Arc::new(required_price_decimal_array(
77            data.iter()
78                .map(std::borrow::Borrow::borrow)
79                .map(|tick| tick.price.raw()),
80            "price",
81        )?);
82        let size_array = Arc::new(required_quantity_decimal_array(
83            data.iter()
84                .map(std::borrow::Borrow::borrow)
85                .map(|tick| tick.size.raw()),
86            "size",
87        )?);
88        let aggressor_side_array = Arc::new(enum_dictionary_array(
89            data.iter()
90                .map(std::borrow::Borrow::borrow)
91                .map(|tick| tick.aggressor_side),
92        )?);
93        let trade_id_array = Arc::new(trade_id_builder.finish());
94        let ts_event_array = Arc::new(ts_event_builder.finish());
95        let ts_init_array = Arc::new(ts_init_builder.finish());
96
97        crate::arrow::record_batch_with_timestamps(
98            Self::get_schema(Some(metadata.clone())).into(),
99            vec![
100                price_array,
101                size_array,
102                aggressor_side_array,
103                trade_id_array,
104                ts_event_array,
105                ts_init_array,
106                Arc::new(identifier_array_from_display(
107                    data.iter()
108                        .map(std::borrow::Borrow::borrow)
109                        .map(|tick| tick.instrument_id),
110                )),
111            ],
112        )
113    }
114
115    fn metadata(&self) -> HashMap<String, String> {
116        Self::get_metadata(
117            &self.instrument_id,
118            self.price.precision,
119            self.size.precision,
120        )
121    }
122}
123
124impl DecodeFromRecordBatch for TradeTick {
125    fn decode_batch(
126        metadata: &HashMap<String, String>,
127        record_batch: RecordBatch,
128    ) -> Result<Vec<Self>, EncodingError> {
129        let (instrument_id, price_precision, size_precision) = parse_metadata(metadata)?;
130        let record_batch = crate::arrow::record_batch_with_u64_timestamps(&record_batch)?;
131        let record_batch = &record_batch;
132        let cols = record_batch.columns();
133
134        let price_values =
135            extract_column::<Decimal128Array>(cols, "price", 0, fixed_decimal_data_type())?;
136
137        let size_values =
138            extract_column::<Decimal128Array>(cols, "size", 1, fixed_decimal_data_type())?;
139
140        let aggressor_side_values = extract_column_string(cols, "aggressor_side", 2)?;
141        let ts_event_values = extract_column::<UInt64Array>(cols, "ts_event", 4, DataType::UInt64)?;
142        let ts_init_values = extract_column::<UInt64Array>(cols, "ts_init", 5, DataType::UInt64)?;
143
144        // Datafusion reads trade_ids as StringView
145        let trade_id_values: Vec<TradeId> = if record_batch
146            .schema()
147            .field_with_name("trade_id")?
148            .data_type()
149            == &DataType::Utf8View
150        {
151            extract_column::<StringViewArray>(cols, "trade_id", 3, DataType::Utf8View)?
152                .iter()
153                .enumerate()
154                .map(|(i, id)| {
155                    id.map(TradeId::from).ok_or_else(|| {
156                        EncodingError::ParseError("trade_id", format!("NULL value at row {i}"))
157                    })
158                })
159                .collect::<Result<Vec<_>, _>>()?
160        } else {
161            extract_column::<StringArray>(cols, "trade_id", 3, DataType::Utf8)?
162                .iter()
163                .enumerate()
164                .map(|(i, id)| {
165                    id.map(TradeId::from).ok_or_else(|| {
166                        EncodingError::ParseError("trade_id", format!("NULL value at row {i}"))
167                    })
168                })
169                .collect::<Result<Vec<_>, _>>()?
170        };
171
172        let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
173            .map(|i| {
174                let price =
175                    decode_required_decimal_price(price_values, price_precision, "price", i)?;
176                let size =
177                    decode_required_decimal_quantity(size_values, size_precision, "size", i)?;
178                let aggressor_side_value = aggressor_side_values.value(i);
179                let aggressor_side =
180                    AggressorSide::from_str(aggressor_side_value).map_err(|e| {
181                        EncodingError::ParseError(stringify!(AggressorSide), e.to_string())
182                    })?;
183                let trade_id = trade_id_values[i];
184                let ts_event = decode_required_timestamp(ts_event_values, "ts_event", i)?;
185                let ts_init = decode_required_timestamp(ts_init_values, "ts_init", i)?;
186
187                Ok(Self {
188                    instrument_id,
189                    price,
190                    size,
191                    aggressor_side,
192                    trade_id,
193                    ts_event,
194                    ts_init,
195                })
196            })
197            .collect();
198
199        result
200    }
201}
202
203impl DecodeDataFromRecordBatch for TradeTick {
204    fn decode_data_batch(
205        metadata: &HashMap<String, String>,
206        record_batch: RecordBatch,
207    ) -> Result<Vec<Data>, EncodingError> {
208        let ticks: Vec<Self> = Self::decode_batch(metadata, record_batch)?;
209        Ok(ticks.into_iter().map(Data::from).collect())
210    }
211}
212
213#[cfg(test)]
214mod tests {
215    use std::sync::Arc;
216
217    use arrow::array::{Array, Decimal128Array, TimestampNanosecondArray, UInt64Array};
218    use nautilus_model::types::{
219        Price, Quantity, fixed::FIXED_SCALAR, price::PriceRaw, quantity::QuantityRaw,
220    };
221    use rstest::rstest;
222
223    use super::*;
224    use crate::arrow::{get_raw_price, get_raw_quantity};
225
226    #[rstest]
227    fn test_get_schema() {
228        let instrument_id = InstrumentId::from("AAPL.XNAS");
229        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
230        let schema = TradeTick::get_schema(Some(metadata.clone()));
231
232        let mut expected_fields = Vec::with_capacity(7);
233
234        expected_fields.push(Field::new("price", fixed_decimal_data_type(), true));
235
236        expected_fields.extend(vec![
237            Field::new("size", fixed_decimal_data_type(), true),
238            Field::new("aggressor_side", enum_dictionary_data_type(), false),
239            Field::new("trade_id", DataType::Utf8, false),
240            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
241            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
242            Field::new(KEY_IDENTIFIER, DataType::Utf8, true),
243        ]);
244
245        let expected_schema = Schema::new_with_metadata(expected_fields, 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    fn test_decode_batch_invalid_price_returns_error() {
462        let instrument_id = InstrumentId::from("AAPL.XNAS");
463        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
464
465        let invalid_price: PriceRaw = PriceRaw::MAX - 1000;
466        let price = crate::arrow::test_support::decimal_array_from_bytes(vec![
467            &invalid_price.to_le_bytes(),
468        ]);
469        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
470            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
471        ]);
472        let aggressor_side = enum_dictionary_array([AggressorSide::NoAggressor]).unwrap();
473        let trade_id = StringArray::from(vec!["1"]);
474        let ts_event = UInt64Array::from(vec![1]);
475        let ts_init = UInt64Array::from(vec![2]);
476
477        let record_batch = crate::arrow::record_batch_with_timestamps(
478            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
479                metadata.clone(),
480            )))
481            .into(),
482            vec![
483                Arc::new(price),
484                Arc::new(size),
485                Arc::new(aggressor_side),
486                Arc::new(trade_id),
487                Arc::new(ts_event),
488                Arc::new(ts_init),
489            ],
490        )
491        .unwrap();
492
493        let result = TradeTick::decode_batch(&metadata, record_batch);
494        assert!(result.is_err());
495        let err = result.unwrap_err();
496        assert!(
497            err.to_string().contains("price") && err.to_string().contains("row 0"),
498            "Expected price error at row 0, was: {err}"
499        );
500    }
501
502    #[rstest]
503    fn test_decode_batch_invalid_size_returns_error() {
504        use nautilus_model::types::{fixed::FIXED_PRECISION, quantity::QUANTITY_RAW_MAX};
505
506        let instrument_id = InstrumentId::from("AAPL.XNAS");
507        // Decode the size at full precision so the out-of-range raw value bypasses the
508        // precision-0 correction, which would otherwise round it back within the bound.
509        let metadata = TradeTick::get_metadata(&instrument_id, 2, FIXED_PRECISION);
510
511        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
512        let price =
513            crate::arrow::test_support::decimal_array_from_bytes(vec![&raw_price.to_le_bytes()]);
514
515        let invalid_size = QUANTITY_RAW_MAX + 1;
516        let size =
517            crate::arrow::test_support::decimal_array_from_bytes(vec![&invalid_size.to_le_bytes()]);
518        let aggressor_side = enum_dictionary_array([AggressorSide::NoAggressor]).unwrap();
519        let trade_id = StringArray::from(vec!["1"]);
520        let ts_event = UInt64Array::from(vec![1]);
521        let ts_init = UInt64Array::from(vec![2]);
522
523        let record_batch = crate::arrow::record_batch_with_timestamps(
524            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
525                metadata.clone(),
526            )))
527            .into(),
528            vec![
529                Arc::new(price),
530                Arc::new(size),
531                Arc::new(aggressor_side),
532                Arc::new(trade_id),
533                Arc::new(ts_event),
534                Arc::new(ts_init),
535            ],
536        )
537        .unwrap();
538
539        let result = TradeTick::decode_batch(&metadata, record_batch);
540        assert!(result.is_err());
541        let err = result.unwrap_err();
542        assert!(
543            err.to_string().contains("size") && err.to_string().contains("row 0"),
544            "Expected size error at row 0, was: {err}"
545        );
546    }
547
548    #[rstest]
549    fn test_decode_batch_invalid_aggressor_side_returns_error() {
550        let instrument_id = InstrumentId::from("AAPL.XNAS");
551        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
552
553        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
554        let price =
555            crate::arrow::test_support::decimal_array_from_bytes(vec![&raw_price.to_le_bytes()]);
556        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
557            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
558        ]);
559
560        let aggressor_side = enum_dictionary_array(["INVALID"]).unwrap();
561        let trade_id = StringArray::from(vec!["1"]);
562        let ts_event = UInt64Array::from(vec![1]);
563        let ts_init = UInt64Array::from(vec![2]);
564
565        let record_batch = crate::arrow::record_batch_with_timestamps(
566            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
567                metadata.clone(),
568            )))
569            .into(),
570            vec![
571                Arc::new(price),
572                Arc::new(size),
573                Arc::new(aggressor_side),
574                Arc::new(trade_id),
575                Arc::new(ts_event),
576                Arc::new(ts_init),
577            ],
578        )
579        .unwrap();
580
581        let result = TradeTick::decode_batch(&metadata, record_batch);
582        assert!(result.is_err());
583        let err = result.unwrap_err();
584        assert!(
585            err.to_string().contains("AggressorSide"),
586            "Expected AggressorSide error, was: {err}"
587        );
588    }
589
590    #[rstest]
591    fn test_decode_batch_missing_instrument_id_returns_error() {
592        let instrument_id = InstrumentId::from("AAPL.XNAS");
593        let mut metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
594        metadata.remove(KEY_INSTRUMENT_ID);
595
596        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
597        let price =
598            crate::arrow::test_support::decimal_array_from_bytes(vec![&raw_price.to_le_bytes()]);
599        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
600            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
601        ]);
602        let aggressor_side = enum_dictionary_array([AggressorSide::NoAggressor]).unwrap();
603        let trade_id = StringArray::from(vec!["1"]);
604        let ts_event = UInt64Array::from(vec![1]);
605        let ts_init = UInt64Array::from(vec![2]);
606
607        let record_batch = crate::arrow::record_batch_with_timestamps(
608            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
609                metadata.clone(),
610            )))
611            .into(),
612            vec![
613                Arc::new(price),
614                Arc::new(size),
615                Arc::new(aggressor_side),
616                Arc::new(trade_id),
617                Arc::new(ts_event),
618                Arc::new(ts_init),
619            ],
620        )
621        .unwrap();
622
623        let result = TradeTick::decode_batch(&metadata, record_batch);
624        assert!(result.is_err());
625        let err = result.unwrap_err();
626        assert!(
627            err.to_string().contains("instrument_id"),
628            "Expected missing instrument_id error, was: {err}"
629        );
630    }
631
632    #[rstest]
633    fn test_decode_batch_missing_price_precision_returns_error() {
634        let instrument_id = InstrumentId::from("AAPL.XNAS");
635        let mut metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
636        metadata.remove(KEY_PRICE_PRECISION);
637
638        let raw_price = (100.00 * FIXED_SCALAR) as PriceRaw;
639        let price =
640            crate::arrow::test_support::decimal_array_from_bytes(vec![&raw_price.to_le_bytes()]);
641        let size = crate::arrow::test_support::decimal_array_from_bytes(vec![
642            &((1000.0 * FIXED_SCALAR) as QuantityRaw).to_le_bytes(),
643        ]);
644        let aggressor_side = enum_dictionary_array([AggressorSide::NoAggressor]).unwrap();
645        let trade_id = StringArray::from(vec!["1"]);
646        let ts_event = UInt64Array::from(vec![1]);
647        let ts_init = UInt64Array::from(vec![2]);
648
649        let record_batch = crate::arrow::record_batch_with_timestamps(
650            crate::arrow::schema_without_identifier_column(&TradeTick::get_schema(Some(
651                metadata.clone(),
652            )))
653            .into(),
654            vec![
655                Arc::new(price),
656                Arc::new(size),
657                Arc::new(aggressor_side),
658                Arc::new(trade_id),
659                Arc::new(ts_event),
660                Arc::new(ts_init),
661            ],
662        )
663        .unwrap();
664
665        let result = TradeTick::decode_batch(&metadata, record_batch);
666        assert!(result.is_err());
667        let err = result.unwrap_err();
668        assert!(
669            err.to_string().contains("price_precision"),
670            "Expected missing price_precision error, was: {err}"
671        );
672    }
673
674    #[rstest]
675    fn test_encode_decode_round_trip() {
676        let instrument_id = InstrumentId::from("AAPL.XNAS");
677        let metadata = TradeTick::get_metadata(&instrument_id, 2, 0);
678
679        let tick1 = TradeTick {
680            instrument_id,
681            price: Price::from("100.10"),
682            size: Quantity::from(1000),
683            aggressor_side: AggressorSide::Buy,
684            trade_id: TradeId::new("trade-123"),
685            ts_event: 1_000_000_000.into(),
686            ts_init: 1_000_000_001.into(),
687        };
688
689        let tick2 = TradeTick {
690            instrument_id,
691            price: Price::from("100.50"),
692            size: Quantity::from(500),
693            aggressor_side: AggressorSide::Sell,
694            trade_id: TradeId::new("trade-456"),
695            ts_event: 2_000_000_000.into(),
696            ts_init: 2_000_000_001.into(),
697        };
698
699        let original = vec![tick1, tick2];
700        let record_batch = TradeTick::encode_batch(&metadata, &original).unwrap();
701        let decoded = TradeTick::decode_batch(&metadata, record_batch).unwrap();
702
703        assert_eq!(decoded.len(), original.len());
704        for (orig, dec) in original.iter().zip(decoded.iter()) {
705            assert_eq!(dec.instrument_id, orig.instrument_id);
706            assert_eq!(dec.price, orig.price);
707            assert_eq!(dec.size, orig.size);
708            assert_eq!(dec.aggressor_side, orig.aggressor_side);
709            assert_eq!(dec.trade_id, orig.trade_id);
710            assert_eq!(dec.ts_event, orig.ts_event);
711            assert_eq!(dec.ts_init, orig.ts_init);
712        }
713    }
714}
nautilus_serialization/arrow/trade.rs

nautilus_serialization/arrow/
trade.rs
