1use std::{collections::HashMap, sync::Arc};
17
18use arrow::{
19 array::{Array, Decimal128Array, ListArray, StructArray, UInt8Array, UInt32Array, UInt64Array},
20 buffer::{OffsetBuffer, ScalarBuffer},
21 datatypes::{DataType, Field, Fields, Schema},
22 error::ArrowError,
23 record_batch::RecordBatch,
24};
25#[cfg(test)]
26use nautilus_model::identifiers::InstrumentId;
27use nautilus_model::{
28 data::{
29 depth::{OrderBookDepth, is_depth_level_present},
30 order::BookOrder,
31 },
32 enums::OrderSide,
33 types::{price::PriceRaw, quantity::QuantityRaw},
34};
35
36use super::{
37 DecodeDataFromRecordBatch, EMPTY_DEPTH_PRECISION, EncodingError, KEY_IDENTIFIER,
38 decode_decimal_price, decode_decimal_quantity, decode_required_timestamp, decode_required_u8,
39 decode_required_u64, extract_column, fixed_decimal_data_type, identifier_array_from_display,
40 metadata_with_type_name, parse_metadata, price_decimal_array, price_raw_to_decimal,
41 quantity_decimal_array,
42};
43#[cfg(test)]
44use super::{KEY_INSTRUMENT_ID, KEY_PRICE_PRECISION, KEY_SIZE_PRECISION};
45use crate::arrow::{ArrowSchemaProvider, Data, DecodeFromRecordBatch, EncodeToRecordBatch};
46
47fn depth_level_fields() -> Fields {
48 vec![
49 Field::new("price", fixed_decimal_data_type(), false),
50 Field::new("size", fixed_decimal_data_type(), false),
51 Field::new("count", DataType::UInt32, false),
52 Field::new("order_id", DataType::UInt64, false),
53 ]
54 .into()
55}
56
57fn depth_side_data_type() -> DataType {
58 let fields = depth_level_fields();
59 DataType::List(Arc::new(Field::new(
60 "item",
61 DataType::Struct(fields),
62 false,
63 )))
64}
65
66impl ArrowSchemaProvider for OrderBookDepth {
67 fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
68 let mut fields = vec![
69 Field::new("bids", depth_side_data_type(), false),
70 Field::new("asks", depth_side_data_type(), false),
71 ];
72 fields.push(Field::new("flags", DataType::UInt8, false));
73 fields.push(Field::new("sequence", DataType::UInt64, false));
74 fields.push(Field::new(
75 "ts_event",
76 crate::arrow::timestamp_data_type(),
77 false,
78 ));
79 fields.push(Field::new(
80 "ts_init",
81 crate::arrow::timestamp_data_type(),
82 false,
83 ));
84 fields.push(Field::new(KEY_IDENTIFIER, DataType::Utf8, true));
85
86 Schema::new_with_metadata(fields, metadata_with_type_name("OrderBookDepth", metadata))
87 }
88}
89
90impl EncodeToRecordBatch for OrderBookDepth {
91 fn encode_batch<T>(
92 metadata: &HashMap<String, String>,
93 data: &[T],
94 ) -> Result<RecordBatch, ArrowError>
95 where
96 T: std::borrow::Borrow<Self>,
97 {
98 let bid_capacity = data
99 .iter()
100 .map(std::borrow::Borrow::borrow)
101 .map(|depth| depth.bids.len())
102 .sum();
103 let ask_capacity = data
104 .iter()
105 .map(std::borrow::Borrow::borrow)
106 .map(|depth| depth.asks.len())
107 .sum();
108 let mut bid_prices = Vec::with_capacity(bid_capacity);
109 let mut ask_prices = Vec::with_capacity(ask_capacity);
110 let mut bid_sizes = Vec::with_capacity(bid_capacity);
111 let mut ask_sizes = Vec::with_capacity(ask_capacity);
112 let mut bid_order_ids = Vec::with_capacity(bid_capacity);
113 let mut ask_order_ids = Vec::with_capacity(ask_capacity);
114 let mut bid_counts = Vec::with_capacity(bid_capacity);
115 let mut ask_counts = Vec::with_capacity(ask_capacity);
116 let mut bid_offsets = Vec::with_capacity(data.len() + 1);
117 let mut ask_offsets = Vec::with_capacity(data.len() + 1);
118 bid_offsets.push(0);
119 ask_offsets.push(0);
120
121 let mut flags_builder = UInt8Array::builder(data.len());
122 let mut sequence_builder = UInt64Array::builder(data.len());
123 let mut ts_event_builder = UInt64Array::builder(data.len());
124 let mut ts_init_builder = UInt64Array::builder(data.len());
125
126 for depth in data.iter().map(std::borrow::Borrow::borrow) {
127 if depth.bids.len() != depth.bid_counts.len()
128 || depth.asks.len() != depth.ask_counts.len()
129 {
130 return Err(ArrowError::InvalidArgumentError(format!(
131 "OrderBookDepth for '{}' has mismatched level and count lengths: \
132 bids {} vs bid_counts {}, asks {} vs ask_counts {}",
133 depth.instrument_id,
134 depth.bids.len(),
135 depth.bid_counts.len(),
136 depth.asks.len(),
137 depth.ask_counts.len(),
138 )));
139 }
140
141 for (bid, count) in depth.bids.iter().zip(&depth.bid_counts) {
142 price_raw_to_decimal(bid.price.raw(), "bids.price")?;
143 if is_depth_level_present(bid) {
144 bid_prices.push(bid.price.raw());
145 bid_sizes.push(bid.size.raw());
146 bid_order_ids.push(bid.order_id);
147 bid_counts.push(*count);
148 }
149 }
150
151 for (ask, count) in depth.asks.iter().zip(&depth.ask_counts) {
152 price_raw_to_decimal(ask.price.raw(), "asks.price")?;
153 if is_depth_level_present(ask) {
154 ask_prices.push(ask.price.raw());
155 ask_sizes.push(ask.size.raw());
156 ask_order_ids.push(ask.order_id);
157 ask_counts.push(*count);
158 }
159 }
160 bid_offsets.push(i32::try_from(bid_prices.len()).map_err(|_| {
161 ArrowError::InvalidArgumentError(
162 "Depth bid values exceed the Arrow List offset range".to_string(),
163 )
164 })?);
165 ask_offsets.push(i32::try_from(ask_prices.len()).map_err(|_| {
166 ArrowError::InvalidArgumentError(
167 "Depth ask values exceed the Arrow List offset range".to_string(),
168 )
169 })?);
170
171 flags_builder.append_value(depth.flags);
172 sequence_builder.append_value(depth.sequence);
173 ts_event_builder.append_value(depth.ts_event.as_u64());
174 ts_init_builder.append_value(depth.ts_init.as_u64());
175 }
176
177 let bids = depth_side_array(
178 bid_prices,
179 bid_sizes,
180 bid_counts,
181 bid_order_ids,
182 bid_offsets,
183 "bids.price",
184 )?;
185 let asks = depth_side_array(
186 ask_prices,
187 ask_sizes,
188 ask_counts,
189 ask_order_ids,
190 ask_offsets,
191 "asks.price",
192 )?;
193
194 let flags_array = Arc::new(flags_builder.finish()) as Arc<dyn Array>;
195 let sequence_array = Arc::new(sequence_builder.finish()) as Arc<dyn Array>;
196 let ts_event_array = Arc::new(ts_event_builder.finish()) as Arc<dyn Array>;
197 let ts_init_array = Arc::new(ts_init_builder.finish()) as Arc<dyn Array>;
198
199 let mut columns = vec![
200 Arc::new(bids) as Arc<dyn Array>,
201 Arc::new(asks) as Arc<dyn Array>,
202 ];
203 columns.push(flags_array);
204 columns.push(sequence_array);
205 columns.push(ts_event_array);
206 columns.push(ts_init_array);
207 columns.push(Arc::new(identifier_array_from_display(
208 data.iter()
209 .map(std::borrow::Borrow::borrow)
210 .map(|depth| depth.instrument_id),
211 )));
212
213 crate::arrow::record_batch_with_timestamps(
214 Self::get_schema(Some(metadata.clone())).into(),
215 columns,
216 )
217 }
218
219 fn metadata(&self) -> HashMap<String, String> {
220 let precision = self
221 .bids
222 .first()
223 .or_else(|| self.asks.first())
224 .map_or(EMPTY_DEPTH_PRECISION, |level| {
225 (level.price.precision, level.size.precision)
226 });
227 Self::get_metadata(&self.instrument_id, precision.0, precision.1)
228 }
229
230 fn chunk_metadata<T>(chunk: &[T]) -> HashMap<String, String>
231 where
232 T: std::borrow::Borrow<Self>,
233 {
234 let first = chunk
235 .first()
236 .map(std::borrow::Borrow::borrow)
237 .expect("Chunk must contain at least one element to encode");
238 let precision = chunk
239 .iter()
240 .map(std::borrow::Borrow::borrow)
241 .flat_map(|depth| depth.bids.iter().chain(&depth.asks))
242 .find(|order| is_depth_level_present(order))
243 .map_or(EMPTY_DEPTH_PRECISION, |order| {
244 (order.price.precision, order.size.precision)
245 });
246
247 Self::get_metadata(&first.instrument_id, precision.0, precision.1)
248 }
249
250 fn matches_chunk_metadata(&self, metadata: &HashMap<String, String>) -> bool {
251 let Ok((instrument_id, price_precision, size_precision)) = parse_metadata(metadata) else {
252 return false;
253 };
254
255 if self.instrument_id != instrument_id {
256 return false;
257 }
258
259 self.bids
260 .iter()
261 .chain(&self.asks)
262 .filter(|order| is_depth_level_present(order))
263 .all(|order| {
264 order.price.precision == price_precision && order.size.precision == size_precision
265 })
266 }
267}
268
269fn depth_side_array(
270 prices: Vec<PriceRaw>,
271 sizes: Vec<QuantityRaw>,
272 counts: Vec<u32>,
273 order_ids: Vec<u64>,
274 offsets: Vec<i32>,
275 price_field: &'static str,
276) -> Result<ListArray, ArrowError> {
277 let fields = depth_level_fields();
278 let values = StructArray::try_new(
279 fields.clone(),
280 vec![
281 Arc::new(price_decimal_array(prices, price_field)?),
282 Arc::new(quantity_decimal_array(sizes, "size")?),
283 Arc::new(UInt32Array::from(counts)),
284 Arc::new(UInt64Array::from(order_ids)),
285 ],
286 None,
287 )?;
288 ListArray::try_new(
289 Arc::new(Field::new("item", DataType::Struct(fields), false)),
290 OffsetBuffer::new(ScalarBuffer::from(offsets)),
291 Arc::new(values),
292 None,
293 )
294}
295
296impl DecodeFromRecordBatch for OrderBookDepth {
297 fn decode_batch(
298 metadata: &HashMap<String, String>,
299 record_batch: RecordBatch,
300 ) -> Result<Vec<Self>, EncodingError> {
301 let (instrument_id, price_precision, size_precision) = parse_metadata(metadata)?;
302 let record_batch = crate::arrow::record_batch_with_u64_timestamps(&record_batch)?;
303
304 let (bid_list, bid_values, bid_prices, bid_sizes, bid_counts, bid_order_ids) =
305 depth_side_values(&record_batch, "bids")?;
306 let (ask_list, ask_values, ask_prices, ask_sizes, ask_counts, ask_order_ids) =
307 depth_side_values(&record_batch, "asks")?;
308
309 let flags = named_column::<UInt8Array>(&record_batch, "flags", DataType::UInt8)?;
310 let sequence = named_column::<UInt64Array>(&record_batch, "sequence", DataType::UInt64)?;
311 let ts_event = named_column::<UInt64Array>(&record_batch, "ts_event", DataType::UInt64)?;
312 let ts_init = named_column::<UInt64Array>(&record_batch, "ts_init", DataType::UInt64)?;
313
314 let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
316 .map(|row| {
317 let (bid_start, bid_end) = depth_row_range(bid_list, "bids", row)?;
318 let bid_end = depth_present_end(bid_values, "bids", row, bid_start, bid_end)?;
319 let (ask_start, ask_end) = depth_row_range(ask_list, "asks", row)?;
320 let ask_end = depth_present_end(ask_values, "asks", row, ask_start, ask_end)?;
321 let mut bids = Vec::with_capacity(bid_end - bid_start);
322 let mut asks = Vec::with_capacity(ask_end - ask_start);
323 let mut bid_count_arr = Vec::with_capacity(bid_end - bid_start);
324 let mut ask_count_arr = Vec::with_capacity(ask_end - ask_start);
325
326 for value_index in bid_start..bid_end {
327 ensure_depth_value(bid_prices, "price", "bids.price", row, value_index)?;
328 ensure_depth_value(bid_sizes, "size", "bids.size", row, value_index)?;
329 ensure_depth_value(bid_counts, "count", "bids.count", row, value_index)?;
330 ensure_depth_value(
331 bid_order_ids,
332 "order_id",
333 "bids.order_id",
334 row,
335 value_index,
336 )?;
337 let bid_price = decode_decimal_price(
338 bid_prices,
339 price_precision,
340 "bids.price",
341 value_index,
342 )?;
343 let bid_size = decode_decimal_quantity(
344 bid_sizes,
345 size_precision,
346 "bids.size",
347 value_index,
348 )?;
349 bids.push(BookOrder::new(
350 OrderSide::Buy,
351 bid_price,
352 bid_size,
353 bid_order_ids.value(value_index),
354 ));
355 bid_count_arr.push(bid_counts.value(value_index));
356 }
357
358 for value_index in ask_start..ask_end {
359 ensure_depth_value(ask_prices, "price", "asks.price", row, value_index)?;
360 ensure_depth_value(ask_sizes, "size", "asks.size", row, value_index)?;
361 ensure_depth_value(ask_counts, "count", "asks.count", row, value_index)?;
362 ensure_depth_value(
363 ask_order_ids,
364 "order_id",
365 "asks.order_id",
366 row,
367 value_index,
368 )?;
369 let ask_price = decode_decimal_price(
370 ask_prices,
371 price_precision,
372 "asks.price",
373 value_index,
374 )?;
375 let ask_size = decode_decimal_quantity(
376 ask_sizes,
377 size_precision,
378 "asks.size",
379 value_index,
380 )?;
381 asks.push(BookOrder::new(
382 OrderSide::Sell,
383 ask_price,
384 ask_size,
385 ask_order_ids.value(value_index),
386 ));
387 ask_count_arr.push(ask_counts.value(value_index));
388 }
389
390 Self::new_checked(
391 instrument_id,
392 bids,
393 asks,
394 bid_count_arr,
395 ask_count_arr,
396 decode_required_u8(flags, "flags", row)?,
397 decode_required_u64(sequence, "sequence", row)?,
398 decode_required_timestamp(ts_event, "ts_event", row)?,
399 decode_required_timestamp(ts_init, "ts_init", row)?,
400 )
401 .map_err(|e| EncodingError::ParseError("depth", format!("row {row}: {e}")))
402 })
403 .collect();
404
405 result
406 }
407}
408
409fn named_column<'a, T: Array + 'static>(
410 record_batch: &'a RecordBatch,
411 name: &'static str,
412 data_type: DataType,
413) -> Result<&'a T, EncodingError> {
414 let index = record_batch.schema().index_of(name)?;
415 extract_column::<T>(record_batch.columns(), name, index, data_type)
416}
417
418#[allow(clippy::type_complexity)]
419fn depth_side_values<'a>(
420 record_batch: &'a RecordBatch,
421 name: &'static str,
422) -> Result<
423 (
424 &'a ListArray,
425 &'a StructArray,
426 &'a Decimal128Array,
427 &'a Decimal128Array,
428 &'a UInt32Array,
429 &'a UInt64Array,
430 ),
431 EncodingError,
432> {
433 let index = record_batch.schema().index_of(name)?;
434 let list = dynamic_column::<ListArray>(record_batch, name, index, &depth_side_data_type())?;
435 let values = list
436 .values()
437 .as_any()
438 .downcast_ref::<StructArray>()
439 .ok_or_else(|| {
440 ArrowError::CastError(format!(
441 "Invalid list value type `{name}`: expected {}, found {}",
442 DataType::Struct(depth_level_fields()),
443 list.value_type(),
444 ))
445 })?;
446 Ok((
447 list,
448 values,
449 depth_struct_column(values, name, "price", &fixed_decimal_data_type())?,
450 depth_struct_column(values, name, "size", &fixed_decimal_data_type())?,
451 depth_struct_column(values, name, "count", &DataType::UInt32)?,
452 depth_struct_column(values, name, "order_id", &DataType::UInt64)?,
453 ))
454}
455
456fn depth_present_end(
457 values: &StructArray,
458 side: &'static str,
459 row: usize,
460 start: usize,
461 end: usize,
462) -> Result<usize, EncodingError> {
463 let mut present_end = end;
464 while present_end > start && values.is_null(present_end - 1) {
465 present_end -= 1;
466 }
467
468 if values.null_count() > 0 && (start..present_end).any(|index| values.is_null(index)) {
469 return Err(EncodingError::ParseError(
470 side,
471 format!("row {row}: levels must be contiguous"),
472 ));
473 }
474
475 Ok(present_end)
476}
477
478fn ensure_depth_value(
479 values: &dyn Array,
480 kind: &'static str,
481 column: &'static str,
482 row: usize,
483 value_index: usize,
484) -> Result<(), EncodingError> {
485 if values.is_null(value_index) {
486 return Err(EncodingError::ParseError(
487 column,
488 format!("{kind} column '{column}' row {row} is null"),
489 ));
490 }
491
492 Ok(())
493}
494
495fn depth_row_range(
496 list: &ListArray,
497 name: &'static str,
498 row: usize,
499) -> Result<(usize, usize), EncodingError> {
500 if list.is_null(row) {
501 return Err(ArrowError::InvalidArgumentError(format!(
502 "Depth side `{name}` is null at row {row}"
503 ))
504 .into());
505 }
506 let offsets = list.value_offsets();
507 let start = usize::try_from(offsets[row]).map_err(|_| {
508 ArrowError::InvalidArgumentError(format!(
509 "Depth side `{name}` has a negative offset at row {row}"
510 ))
511 })?;
512 let end = usize::try_from(offsets[row + 1]).map_err(|_| {
513 ArrowError::InvalidArgumentError(format!(
514 "Depth side `{name}` has a negative offset at row {row}"
515 ))
516 })?;
517 Ok((start, end))
518}
519
520fn depth_struct_column<'a, T: Array + 'static>(
521 values: &'a StructArray,
522 side: &str,
523 name: &str,
524 expected_type: &DataType,
525) -> Result<&'a T, EncodingError> {
526 let column = values
527 .column_by_name(name)
528 .ok_or_else(|| ArrowError::SchemaError(format!("Missing depth field `{side}.{name}`")))?;
529 column
530 .as_any()
531 .downcast_ref::<T>()
532 .filter(|_| column.data_type() == expected_type)
533 .ok_or_else(|| {
534 ArrowError::CastError(format!(
535 "Invalid depth field `{side}.{name}`: expected {expected_type}, found {}",
536 column.data_type(),
537 ))
538 .into()
539 })
540}
541
542fn dynamic_column<'a, T: Array + 'static>(
543 record_batch: &'a RecordBatch,
544 name: &str,
545 index: usize,
546 expected_type: &DataType,
547) -> Result<&'a T, EncodingError> {
548 let column = record_batch.columns().get(index).ok_or_else(|| {
549 ArrowError::SchemaError(format!("Missing data column: `{name}` at index {index}"))
550 })?;
551 column.as_any().downcast_ref::<T>().ok_or_else(|| {
552 ArrowError::CastError(format!(
553 "Invalid column type `{name}` at index {index}: expected {expected_type}, found {}",
554 column.data_type(),
555 ))
556 .into()
557 })
558}
559
560impl DecodeDataFromRecordBatch for OrderBookDepth {
561 fn decode_data_batch(
562 metadata: &HashMap<String, String>,
563 record_batch: RecordBatch,
564 ) -> Result<Vec<Data>, EncodingError> {
565 let depths: Vec<Self> = Self::decode_batch(metadata, record_batch)?;
566 Ok(depths.into_iter().map(Data::from).collect())
567 }
568}
569
570#[cfg(test)]
571mod tests {
572 use arrow::{
573 array::TimestampNanosecondArray,
574 datatypes::{DataType, Field},
575 };
576 use nautilus_model::{
577 data::{BookOrder, DEPTH10_LEN, stubs::stub_depth10},
578 enums::{BookType, OrderSide, RecordFlag},
579 orderbook::OrderBook,
580 types::{
581 PRICE_ERROR, PRICE_UNDEF, Price, QUANTITY_UNDEF, Quantity, fixed::FIXED_SCALAR,
582 price::PriceRaw, quantity::QuantityRaw,
583 },
584 };
585 use pretty_assertions::assert_eq;
586 use rstest::rstest;
587
588 use super::*;
589 use crate::arrow::{get_raw_price, get_raw_quantity};
590
591 #[rstest]
592 fn test_get_schema() {
593 let instrument_id = InstrumentId::from("AAPL.XNAS");
594 let metadata = OrderBookDepth::get_metadata(&instrument_id, 2, 0);
595 let schema = OrderBookDepth::get_schema(Some(metadata));
596
597 assert_eq!(
598 schema.field(0),
599 &Field::new("bids", depth_side_data_type(), false),
600 );
601 assert_eq!(
602 schema.field(1),
603 &Field::new("asks", depth_side_data_type(), false),
604 );
605 let flags_field = schema.field(2).clone();
606 assert_eq!(flags_field, Field::new("flags", DataType::UInt8, false));
607 let sequence_field = schema.field(3).clone();
608 assert_eq!(
609 sequence_field,
610 Field::new("sequence", DataType::UInt64, false)
611 );
612 let ts_event_field = schema.field(4).clone();
613 assert_eq!(
614 ts_event_field,
615 Field::new("ts_event", crate::arrow::timestamp_data_type(), false)
616 );
617 let ts_init_field = schema.field(5).clone();
618 assert_eq!(
619 ts_init_field,
620 Field::new("ts_init", crate::arrow::timestamp_data_type(), false)
621 );
622
623 assert_eq!(schema.metadata()["instrument_id"], "AAPL.XNAS");
624 assert_eq!(schema.metadata()["price_precision"], "2");
625 assert_eq!(schema.metadata()["size_precision"], "0");
626 }
627
628 #[rstest]
629 fn test_get_schema_map() {
630 let schema_map = OrderBookDepth::get_schema_map();
631
632 let depth_type = format!("{:?}", depth_side_data_type());
633 assert_eq!(
634 schema_map.get("bids").map(String::as_str),
635 Some(depth_type.as_str()),
636 );
637 assert_eq!(
638 schema_map.get("asks").map(String::as_str),
639 Some(depth_type.as_str()),
640 );
641
642 assert_eq!(schema_map.get("flags").map(String::as_str), Some("UInt8"));
643 assert_eq!(
644 schema_map.get("sequence").map(String::as_str),
645 Some("UInt64")
646 );
647 assert_eq!(
648 schema_map.get("ts_event").map(String::as_str),
649 Some("Timestamp(Nanosecond, Some(\"UTC\"))")
650 );
651 assert_eq!(
652 schema_map.get("ts_init").map(String::as_str),
653 Some("Timestamp(Nanosecond, Some(\"UTC\"))")
654 );
655 assert_eq!(
656 schema_map.get(KEY_IDENTIFIER).map(String::as_str),
657 Some("Utf8")
658 );
659 }
660
661 #[rstest]
662 fn test_chunk_metadata_skips_leading_empty_snapshot(stub_depth10: OrderBookDepth) {
663 let mut empty = stub_depth10.clone();
664 empty.bids.clear();
665 empty.asks.clear();
666 empty.bid_counts.clear();
667 empty.ask_counts.clear();
668
669 let metadata = OrderBookDepth::chunk_metadata(&[empty.clone(), stub_depth10.clone()]);
670
671 assert_eq!(
672 metadata[KEY_INSTRUMENT_ID],
673 stub_depth10.instrument_id.to_string()
674 );
675 assert_eq!(metadata[KEY_PRICE_PRECISION], "2");
676 assert_eq!(metadata[KEY_SIZE_PRECISION], "0");
677 assert!(empty.matches_chunk_metadata(&metadata));
678 assert!(stub_depth10.matches_chunk_metadata(&metadata));
679 }
680
681 #[rstest]
682 fn test_all_empty_depth_chunk_uses_zero_precision(stub_depth10: OrderBookDepth) {
683 let mut first = stub_depth10;
684 first.bids.clear();
685 first.asks.clear();
686 first.bid_counts.clear();
687 first.ask_counts.clear();
688 let mut second = first.clone();
689 second.sequence = 2;
690
691 let depths = [first, second];
692 let metadata = OrderBookDepth::chunk_metadata(&depths);
693 let batch = OrderBookDepth::encode_batch(&metadata, &depths).unwrap();
694 let decoded = OrderBookDepth::decode_batch(&metadata, batch).unwrap();
695
696 assert_eq!(
697 metadata[KEY_INSTRUMENT_ID],
698 depths[0].instrument_id.to_string()
699 );
700 assert_eq!(metadata[KEY_PRICE_PRECISION], "0");
701 assert_eq!(metadata[KEY_SIZE_PRECISION], "0");
702 assert_eq!(decoded, depths);
703 }
704
705 #[rstest]
706 fn test_encode_batch(stub_depth10: OrderBookDepth) {
707 let instrument_id = InstrumentId::from("AAPL.XNAS");
708 let price_precision = 2;
709 let metadata = OrderBookDepth::get_metadata(&instrument_id, price_precision, 0);
710
711 let data = vec![stub_depth10];
712 let record_batch = OrderBookDepth::encode_batch(&metadata, &data).unwrap();
713 let columns = record_batch.columns();
714
715 assert_eq!(columns.len(), 7);
716
717 let bids = columns[0]
718 .as_any()
719 .downcast_ref::<ListArray>()
720 .unwrap()
721 .values()
722 .as_any()
723 .downcast_ref::<StructArray>()
724 .unwrap();
725 let bid_prices = bids
726 .column_by_name("price")
727 .unwrap()
728 .as_any()
729 .downcast_ref::<Decimal128Array>()
730 .unwrap();
731
732 let expected_bid_prices: Vec<f64> =
733 vec![99.0, 98.0, 97.0, 96.0, 95.0, 94.0, 93.0, 92.0, 91.0, 90.0];
734
735 for (i, expected) in expected_bid_prices.iter().enumerate() {
736 assert_eq!(
737 get_raw_price(bid_prices.value(i)),
738 (expected * FIXED_SCALAR) as PriceRaw
739 );
740 assert_eq!(
741 Price::from_raw(get_raw_price(bid_prices.value(i)), price_precision).as_f64(),
742 *expected
743 );
744 }
745
746 let asks = columns[1]
747 .as_any()
748 .downcast_ref::<ListArray>()
749 .unwrap()
750 .values()
751 .as_any()
752 .downcast_ref::<StructArray>()
753 .unwrap();
754 let ask_prices = asks
755 .column_by_name("price")
756 .unwrap()
757 .as_any()
758 .downcast_ref::<Decimal128Array>()
759 .unwrap();
760
761 let expected_ask_prices: Vec<f64> = vec![
762 100.0, 101.0, 102.0, 103.0, 104.0, 105.0, 106.0, 107.0, 108.0, 109.0,
763 ];
764
765 for (i, expected) in expected_ask_prices.iter().enumerate() {
766 assert_eq!(
767 get_raw_price(ask_prices.value(i)),
768 (expected * FIXED_SCALAR) as PriceRaw
769 );
770 assert_eq!(
771 Price::from_raw(get_raw_price(ask_prices.value(i)), price_precision).as_f64(),
772 *expected
773 );
774 }
775
776 let bid_sizes = bids
777 .column_by_name("size")
778 .unwrap()
779 .as_any()
780 .downcast_ref::<Decimal128Array>()
781 .unwrap();
782
783 for i in 0..DEPTH10_LEN {
784 assert_eq!(
785 get_raw_quantity(bid_sizes.value(i)),
786 ((100.0 * FIXED_SCALAR * (i + 1) as f64) as QuantityRaw)
787 );
788 }
789
790 let ask_sizes = asks
791 .column_by_name("size")
792 .unwrap()
793 .as_any()
794 .downcast_ref::<Decimal128Array>()
795 .unwrap();
796
797 for i in 0..DEPTH10_LEN {
798 assert_eq!(
799 get_raw_quantity(ask_sizes.value(i)),
800 ((100.0 * FIXED_SCALAR * ((i + 1) as f64)) as QuantityRaw)
801 );
802 }
803
804 let bid_order_ids = bids
805 .column_by_name("order_id")
806 .unwrap()
807 .as_any()
808 .downcast_ref::<UInt64Array>()
809 .unwrap();
810
811 for i in 0..DEPTH10_LEN {
812 assert_eq!(bid_order_ids.value(i), (i + 1) as u64);
813 }
814
815 let ask_order_ids = asks
816 .column_by_name("order_id")
817 .unwrap()
818 .as_any()
819 .downcast_ref::<UInt64Array>()
820 .unwrap();
821
822 for i in 0..DEPTH10_LEN {
823 assert_eq!(ask_order_ids.value(i), (DEPTH10_LEN + i + 1) as u64);
824 }
825
826 let flags_values = columns[2].as_any().downcast_ref::<UInt8Array>().unwrap();
827 let sequence_values = columns[3].as_any().downcast_ref::<UInt64Array>().unwrap();
828 let ts_event_values = columns[4]
829 .as_any()
830 .downcast_ref::<TimestampNanosecondArray>()
831 .unwrap();
832 let ts_init_values = columns[5]
833 .as_any()
834 .downcast_ref::<TimestampNanosecondArray>()
835 .unwrap();
836
837 assert_eq!(flags_values.len(), 1);
838 assert_eq!(flags_values.value(0), 0);
839 assert_eq!(sequence_values.len(), 1);
840 assert_eq!(sequence_values.value(0), 0);
841 assert_eq!(ts_event_values.len(), 1);
842 assert_eq!(ts_event_values.value(0), 1);
843 assert_eq!(ts_init_values.len(), 1);
844 assert_eq!(ts_init_values.value(0), 2);
845 }
846
847 #[rstest]
848 fn test_encode_batch_rejects_price_error(mut stub_depth10: OrderBookDepth) {
849 stub_depth10.bids[0].price = Price::from_raw(PRICE_ERROR, 0);
850 let metadata = stub_depth10.metadata();
851
852 let error = OrderBookDepth::encode_batch(&metadata, &[stub_depth10]).unwrap_err();
853
854 assert_eq!(
855 error.to_string(),
856 format!(
857 "Invalid argument error: Price field 'bids.price' contains PRICE_ERROR raw value {PRICE_ERROR}"
858 ),
859 );
860 }
861
862 #[rstest]
863 fn test_decode_batch(stub_depth10: OrderBookDepth) {
864 let instrument_id = InstrumentId::from("AAPL.XNAS");
865 let metadata = OrderBookDepth::get_metadata(&instrument_id, 2, 0);
866
867 let data = vec![stub_depth10];
868 let record_batch = OrderBookDepth::encode_batch(&metadata, &data).unwrap();
869 let decoded_data = OrderBookDepth::decode_batch(&metadata, record_batch).unwrap();
870
871 assert_eq!(decoded_data.len(), 1);
872 }
873
874 #[rstest]
875 #[case(38, 2)]
876 #[case(38, 9)]
877 #[case(38, 18)]
878 #[case(37, 16)]
879 fn test_decode_rejects_nested_decimal_type(
880 stub_depth10: OrderBookDepth,
881 #[case] precision: u8,
882 #[case] scale: i8,
883 #[values("bids", "asks")] side: &str,
884 #[values("price", "size")] name: &str,
885 ) {
886 let metadata = stub_depth10.metadata();
887 let batch = OrderBookDepth::encode_batch(&metadata, &[stub_depth10]).unwrap();
888 let side_index = batch.schema().index_of(side).unwrap();
889 let list = batch
890 .column(side_index)
891 .as_any()
892 .downcast_ref::<ListArray>()
893 .unwrap();
894 let values = list
895 .values()
896 .as_any()
897 .downcast_ref::<StructArray>()
898 .unwrap();
899 let column = values
900 .fields()
901 .iter()
902 .position(|field| field.name() == name)
903 .unwrap();
904 let decimals = values
905 .column(column)
906 .as_any()
907 .downcast_ref::<Decimal128Array>()
908 .unwrap()
909 .clone()
910 .with_precision_and_scale(precision, scale)
911 .unwrap();
912 let mut value_fields = values.fields().to_vec();
913 value_fields[column] = Arc::new(Field::new(
914 name,
915 DataType::Decimal128(precision, scale),
916 false,
917 ));
918 let value_fields: Fields = value_fields.into();
919 let mut value_columns = values.columns().to_vec();
920 value_columns[column] = Arc::new(decimals);
921 let values = StructArray::try_new(value_fields.clone(), value_columns, None).unwrap();
922 let item = Arc::new(Field::new("item", DataType::Struct(value_fields), false));
923 let list = ListArray::try_new(
924 Arc::clone(&item),
925 list.offsets().clone(),
926 Arc::new(values),
927 None,
928 )
929 .unwrap();
930 let mut fields = batch.schema().fields().to_vec();
931 fields[side_index] = Arc::new(Field::new(side, DataType::List(item), false));
932 let mut columns = batch.columns().to_vec();
933 columns[side_index] = Arc::new(list);
934 let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap();
935
936 let error = OrderBookDepth::decode_batch(&metadata, batch).unwrap_err();
937
938 assert_eq!(
939 error.to_string(),
940 format!(
941 "Arrow error: Cast error: Invalid depth field `{side}.{name}`: expected {}, found {}",
942 fixed_decimal_data_type(),
943 DataType::Decimal128(precision, scale),
944 )
945 );
946 }
947
948 #[rstest]
949 fn test_decode_batch_rejects_null_nested_price(stub_depth10: OrderBookDepth) {
950 let metadata = stub_depth10.metadata();
951 let batch =
952 OrderBookDepth::encode_batch(&metadata, std::slice::from_ref(&stub_depth10)).unwrap();
953 let bids_index = batch.schema().index_of("bids").unwrap();
954 let bids = batch
955 .column(bids_index)
956 .as_any()
957 .downcast_ref::<ListArray>()
958 .unwrap();
959 let values = bids.values();
960 let values = values.as_any().downcast_ref::<StructArray>().unwrap();
961 let prices = values
962 .column_by_name("price")
963 .unwrap()
964 .as_any()
965 .downcast_ref::<Decimal128Array>()
966 .unwrap();
967 let prices = Decimal128Array::from(
968 (0..prices.len())
969 .map(|index| (index != 0).then(|| prices.value(index)))
970 .collect::<Vec<_>>(),
971 )
972 .with_precision_and_scale(38, 16)
973 .unwrap();
974 let mut value_fields = values.fields().to_vec();
975 value_fields[0] = Arc::new(Field::new("price", fixed_decimal_data_type(), true));
976 let value_fields: Fields = value_fields.into();
977 let mut value_columns = values.columns().to_vec();
978 value_columns[0] = Arc::new(prices);
979 let values =
980 StructArray::try_new(value_fields.clone(), value_columns, values.nulls().cloned())
981 .unwrap();
982 let item = Arc::new(Field::new("item", DataType::Struct(value_fields), false));
983 let bids = ListArray::try_new(
984 Arc::clone(&item),
985 bids.offsets().clone(),
986 Arc::new(values),
987 bids.nulls().cloned(),
988 )
989 .unwrap();
990 let mut columns = batch.columns().to_vec();
991 columns[bids_index] = Arc::new(bids);
992 let mut fields = batch.schema().fields().to_vec();
993 fields[bids_index] = Arc::new(Field::new("bids", DataType::List(item), false));
994 let schema = Arc::new(Schema::new_with_metadata(
995 fields,
996 batch.schema().metadata().clone(),
997 ));
998 let malformed = RecordBatch::try_new(schema, columns).unwrap();
999
1000 let error = OrderBookDepth::decode_batch(&metadata, malformed).unwrap_err();
1001
1002 assert_eq!(
1003 error.to_string(),
1004 "Error parsing `bids.price`: price column 'bids.price' row 0 is null"
1005 );
1006 }
1007
1008 #[rstest]
1009 fn test_decode_batch_uses_column_names(stub_depth10: OrderBookDepth) {
1010 let metadata = OrderBookDepth::get_metadata(&stub_depth10.instrument_id, 2, 0);
1011 let batch =
1012 OrderBookDepth::encode_batch(&metadata, std::slice::from_ref(&stub_depth10)).unwrap();
1013 let schema = batch.schema();
1014 let mut fields = schema.fields().to_vec();
1015 fields.reverse();
1016 let mut columns = batch.columns().to_vec();
1017 columns.reverse();
1018 let reordered = crate::arrow::record_batch_with_timestamps(
1019 Arc::new(Schema::new_with_metadata(fields, schema.metadata().clone())),
1020 columns,
1021 )
1022 .unwrap();
1023
1024 let decoded = OrderBookDepth::decode_batch(&metadata, reordered).unwrap();
1025
1026 assert_eq!(decoded, vec![stub_depth10]);
1027 }
1028
1029 #[rstest]
1030 fn test_depth_column_type_error_names_column() {
1031 let batch = crate::arrow::record_batch_with_timestamps(
1032 Arc::new(Schema::new(vec![Field::new(
1033 "bids",
1034 DataType::UInt64,
1035 false,
1036 )])),
1037 vec![Arc::new(UInt64Array::from(vec![1]))],
1038 )
1039 .unwrap();
1040
1041 let error = depth_side_values(&batch, "bids").unwrap_err();
1042
1043 match error {
1044 EncodingError::ArrowError(ArrowError::CastError(message)) => assert_eq!(
1045 message,
1046 "Invalid column type `bids` at index 0: expected List(non-null \
1047 Struct(\"price\": non-null Decimal128(38, 16), \"size\": non-null \
1048 Decimal128(38, 16), \"count\": non-null UInt32, \"order_id\": non-null UInt64)), \
1049 found UInt64",
1050 ),
1051 other => panic!("expected Arrow cast error, was {other:?}"),
1052 }
1053 }
1054
1055 #[rstest]
1056 fn test_decode_batch_missing_instrument_id_returns_error(stub_depth10: OrderBookDepth) {
1057 let instrument_id = InstrumentId::from("AAPL.XNAS");
1058 let mut metadata = OrderBookDepth::get_metadata(&instrument_id, 2, 0);
1059 let record_batch = OrderBookDepth::encode_batch(&metadata, &[stub_depth10]).unwrap();
1060
1061 metadata.remove(KEY_INSTRUMENT_ID);
1062
1063 let result = OrderBookDepth::decode_batch(&metadata, record_batch);
1064 assert!(result.is_err());
1065 let err = result.unwrap_err();
1066 assert!(
1067 err.to_string().contains("instrument_id"),
1068 "Expected missing instrument_id error, was: {err}"
1069 );
1070 }
1071
1072 #[rstest]
1073 fn test_decode_batch_missing_price_precision_returns_error(stub_depth10: OrderBookDepth) {
1074 let instrument_id = InstrumentId::from("AAPL.XNAS");
1075 let mut metadata = OrderBookDepth::get_metadata(&instrument_id, 2, 0);
1076 let record_batch = OrderBookDepth::encode_batch(&metadata, &[stub_depth10]).unwrap();
1077
1078 metadata.remove(KEY_PRICE_PRECISION);
1079
1080 let result = OrderBookDepth::decode_batch(&metadata, record_batch);
1081 assert!(result.is_err());
1082 let err = result.unwrap_err();
1083 assert!(
1084 err.to_string().contains("price_precision"),
1085 "Expected missing price_precision error, was: {err}"
1086 );
1087 }
1088
1089 #[rstest]
1090 fn test_encode_decode_round_trip(stub_depth10: OrderBookDepth) {
1091 let instrument_id = InstrumentId::from("AAPL.XNAS");
1092 let metadata = OrderBookDepth::get_metadata(&instrument_id, 2, 0);
1093
1094 let original = vec![stub_depth10];
1095 let record_batch = OrderBookDepth::encode_batch(&metadata, &original).unwrap();
1096 let decoded = OrderBookDepth::decode_batch(&metadata, record_batch).unwrap();
1097
1098 assert_eq!(decoded.len(), original.len());
1099 let orig = &original[0];
1100 let dec = &decoded[0];
1101
1102 assert_eq!(dec.instrument_id, orig.instrument_id);
1103 assert_eq!(dec.flags, orig.flags);
1104 assert_eq!(dec.sequence, orig.sequence);
1105 assert_eq!(dec.ts_event, orig.ts_event);
1106 assert_eq!(dec.ts_init, orig.ts_init);
1107
1108 for i in 0..DEPTH10_LEN {
1109 assert_eq!(
1110 dec.bids[i].price, orig.bids[i].price,
1111 "bid price mismatch at level {i}"
1112 );
1113 assert_eq!(
1114 dec.bids[i].size, orig.bids[i].size,
1115 "bid size mismatch at level {i}"
1116 );
1117 assert_eq!(
1118 dec.bids[i].order_id, orig.bids[i].order_id,
1119 "bid order ID mismatch at level {i}"
1120 );
1121 assert_eq!(
1122 dec.asks[i].price, orig.asks[i].price,
1123 "ask price mismatch at level {i}"
1124 );
1125 assert_eq!(
1126 dec.asks[i].size, orig.asks[i].size,
1127 "ask size mismatch at level {i}"
1128 );
1129 assert_eq!(
1130 dec.asks[i].order_id, orig.asks[i].order_id,
1131 "ask order ID mismatch at level {i}"
1132 );
1133 }
1134 }
1135
1136 #[rstest]
1137 fn zero_size_level_round_trip_preserves_applied_book() {
1138 let instrument_id = InstrumentId::from("AAPL.XNAS");
1139 let zero_bid = BookOrder::new(
1140 OrderSide::Buy,
1141 Price::from("100.00"),
1142 Quantity::from("0"),
1143 1,
1144 );
1145 let bid = BookOrder::new(
1146 OrderSide::Buy,
1147 Price::from("99.00"),
1148 Quantity::from("10"),
1149 2,
1150 );
1151 let ask = BookOrder::new(
1152 OrderSide::Sell,
1153 Price::from("101.00"),
1154 Quantity::from("20"),
1155 3,
1156 );
1157 let original = OrderBookDepth {
1158 instrument_id,
1159 bids: [zero_bid, bid].into_iter().collect(),
1160 asks: [ask].into_iter().collect(),
1161 bid_counts: [11, 12].into_iter().collect(),
1162 ask_counts: [21].into_iter().collect(),
1163 flags: RecordFlag::F_SNAPSHOT as u8,
1164 sequence: 1,
1165 ts_event: 2.into(),
1166 ts_init: 3.into(),
1167 };
1168 let metadata = original.metadata();
1169 let batch =
1170 OrderBookDepth::encode_batch(&metadata, std::slice::from_ref(&original)).unwrap();
1171 let [decoded] = OrderBookDepth::decode_batch(&metadata, batch)
1172 .unwrap()
1173 .try_into()
1174 .unwrap();
1175 let mut expected_book = OrderBook::new(instrument_id, BookType::L2_MBP);
1176 let mut decoded_book = OrderBook::new(instrument_id, BookType::L2_MBP);
1177
1178 expected_book.apply_depth(&original).unwrap();
1179 decoded_book.apply_depth(&decoded).unwrap();
1180
1181 assert_eq!(decoded.bids.as_slice(), &[bid]);
1182 assert_eq!(decoded.bid_counts.as_slice(), &[12]);
1183 assert_eq!(decoded.asks.as_slice(), &[ask]);
1184 assert_eq!(decoded.ask_counts.as_slice(), &[21]);
1185 assert_eq!(decoded_book, expected_book);
1186 }
1187
1188 #[rstest]
1189 #[case(5)]
1190 #[case(10)]
1191 #[case(25)]
1192 fn test_encode_decode_runtime_depth_round_trip(
1193 stub_depth10: OrderBookDepth,
1194 #[case] depth: usize,
1195 ) {
1196 let value = OrderBookDepth::new(
1197 stub_depth10.instrument_id,
1198 stub_depth10.bids.iter().copied().cycle().take(depth),
1199 stub_depth10.asks.iter().copied().cycle().take(depth),
1200 0..u32::try_from(depth).unwrap(),
1201 100..100 + u32::try_from(depth).unwrap(),
1202 stub_depth10.flags,
1203 stub_depth10.sequence,
1204 stub_depth10.ts_event,
1205 stub_depth10.ts_init,
1206 );
1207 let metadata = value.metadata();
1208 let record_batch =
1209 OrderBookDepth::encode_batch(&metadata, std::slice::from_ref(&value)).unwrap();
1210 let bids = record_batch
1211 .column_by_name("bids")
1212 .unwrap()
1213 .as_any()
1214 .downcast_ref::<ListArray>()
1215 .unwrap();
1216 let asks = record_batch
1217 .column_by_name("asks")
1218 .unwrap()
1219 .as_any()
1220 .downcast_ref::<ListArray>()
1221 .unwrap();
1222
1223 assert_eq!(bids.value_length(0), i32::try_from(depth).unwrap());
1224 assert_eq!(asks.value_length(0), i32::try_from(depth).unwrap());
1225 assert_eq!(
1226 OrderBookDepth::decode_batch(&metadata, record_batch).unwrap(),
1227 vec![value]
1228 );
1229 }
1230
1231 #[rstest]
1232 fn test_encode_batch_rejects_mismatched_level_and_count_lengths(stub_depth10: OrderBookDepth) {
1233 let mut value = stub_depth10;
1234 value.bid_counts.pop();
1235 let metadata = value.metadata();
1236 let error =
1237 OrderBookDepth::encode_batch(&metadata, std::slice::from_ref(&value)).unwrap_err();
1238 assert!(
1239 error
1240 .to_string()
1241 .contains("mismatched level and count lengths")
1242 );
1243 assert!(error.to_string().contains("bids 10 vs bid_counts 9"));
1244 }
1245
1246 #[rstest]
1247 #[case::price_only(true, false)]
1248 #[case::size_only(false, true)]
1249 #[case::both(true, true)]
1250 #[case::neither(false, false)]
1251 fn test_decode_batch_with_undefined_levels(
1252 stub_depth10: OrderBookDepth,
1253 #[case] price_undef: bool,
1254 #[case] size_undef: bool,
1255 ) {
1256 let instrument_id = InstrumentId::from("AAPL.XNAS");
1257 let price_precision = 2;
1258 let size_precision = 0;
1259 let metadata =
1260 OrderBookDepth::get_metadata(&instrument_id, price_precision, size_precision);
1261
1262 let original_depth = stub_depth10;
1263 let mut depth = original_depth.clone();
1264 let original_bid = depth.bids[5];
1265 let original_ask = depth.asks[7];
1266 let sentinel_bid_price = if price_undef {
1267 Price::from_raw(PRICE_UNDEF, 0)
1268 } else {
1269 original_bid.price
1270 };
1271 let sentinel_bid_size = if size_undef {
1272 Quantity::from_raw(QUANTITY_UNDEF, 0)
1273 } else {
1274 original_bid.size
1275 };
1276 depth.bids[5] = BookOrder {
1277 side: OrderSide::Buy.into(),
1278 price: sentinel_bid_price,
1279 size: sentinel_bid_size,
1280 order_id: 0,
1281 };
1282 let sentinel_ask_price = if price_undef {
1283 Price::from_raw(PRICE_UNDEF, 0)
1284 } else {
1285 original_ask.price
1286 };
1287 let sentinel_ask_size = if size_undef {
1288 Quantity::from_raw(QUANTITY_UNDEF, 0)
1289 } else {
1290 original_ask.size
1291 };
1292 depth.asks[7] = BookOrder {
1293 side: OrderSide::Sell.into(),
1294 price: sentinel_ask_price,
1295 size: sentinel_ask_size,
1296 order_id: 0,
1297 };
1298
1299 let record_batch = OrderBookDepth::encode_batch(&metadata, &[depth]);
1300
1301 let record_batch = record_batch.unwrap();
1302 let bids = record_batch
1303 .column_by_name("bids")
1304 .unwrap()
1305 .as_any()
1306 .downcast_ref::<ListArray>()
1307 .unwrap();
1308 let asks = record_batch
1309 .column_by_name("asks")
1310 .unwrap()
1311 .as_any()
1312 .downcast_ref::<ListArray>()
1313 .unwrap();
1314 let expected_len = if price_undef || size_undef { 9 } else { 10 };
1315 assert_eq!(bids.value_length(0), expected_len);
1316 assert_eq!(asks.value_length(0), expected_len);
1317 let decoded = OrderBookDepth::decode_batch(&metadata, record_batch).unwrap();
1318
1319 assert_eq!(decoded.len(), 1);
1320 let decoded = &decoded[0];
1321
1322 let expect_null = price_undef || size_undef;
1323 if expect_null {
1324 assert_eq!(decoded.bids.len(), 9);
1325 assert_eq!(decoded.asks.len(), 9);
1326 assert_eq!(decoded.bids[5], original_depth.bids[6]);
1327 assert_eq!(decoded.bids[8], original_depth.bids[9]);
1328 assert_eq!(decoded.asks[7], original_depth.asks[8]);
1329 assert_eq!(decoded.asks[8], original_depth.asks[9]);
1330 } else {
1331 assert_eq!(decoded.bids[5].side, Some(OrderSide::Buy));
1332 assert_eq!(decoded.bids[5].price, original_bid.price);
1333 assert_eq!(decoded.bids[5].size, original_bid.size);
1334 assert_eq!(decoded.asks[7].side, Some(OrderSide::Sell));
1335 assert_eq!(decoded.asks[7].price, original_ask.price);
1336 assert_eq!(decoded.asks[7].size, original_ask.size);
1337 }
1338
1339 assert_eq!(decoded.bids[0].side, Some(OrderSide::Buy));
1341 assert_eq!(decoded.bids[0].price.precision, price_precision);
1342 assert_eq!(decoded.bids[0].size.precision, size_precision);
1343 assert_eq!(decoded.asks[0].side, Some(OrderSide::Sell));
1344 assert_eq!(decoded.asks[0].price.precision, price_precision);
1345 assert_eq!(decoded.asks[0].size.precision, size_precision);
1346 }
1347}