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