nautilus_serialization/sbe/market/
book.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 nautilus_model::{
17    data::{BookOrder, OrderBookDelta, OrderBookDeltas, OrderBookDepth},
18    enums::OrderSide,
19};
20
21use super::{
22    super::{SbeCursor, SbeDecodeError, SbeEncodeError, SbeWriter},
23    FromSbeReuse, MarketSbeMessage,
24    common::{
25        BOOK_ORDER_BLOCK_LENGTH, DEPTH10_COUNTS_BLOCK_LENGTH, DEPTH10_LEVEL_BLOCK_LENGTH,
26        DEPTH10_LEVEL_COUNT, GROUP_HEADER_16_LENGTH, MessageHeader,
27        ORDER_BOOK_DELTA_GROUP_BLOCK_LENGTH, decode_book_action, decode_header,
28        decode_instrument_id, decode_order_side, decode_price, decode_quantity, decode_unix_nanos,
29        encode_group_header_16, encode_instrument_id, encode_price, encode_quantity,
30        encode_unix_nanos, encoded_instrument_id_size, validate_header,
31    },
32    template_id,
33};
34
35impl MarketSbeMessage for BookOrder {
36    const TEMPLATE_ID: u16 = template_id::BOOK_ORDER;
37    const BLOCK_LENGTH: u16 = BOOK_ORDER_BLOCK_LENGTH;
38
39    fn encode_body(&self, writer: &mut SbeWriter<'_>) -> Result<(), SbeEncodeError> {
40        encode_book_order(writer, self);
41        Ok(())
42    }
43
44    fn decode_body(cursor: &mut SbeCursor<'_>) -> Result<Self, SbeDecodeError> {
45        decode_book_order(cursor)
46    }
47}
48
49impl MarketSbeMessage for OrderBookDelta {
50    const TEMPLATE_ID: u16 = template_id::ORDER_BOOK_DELTA;
51    const BLOCK_LENGTH: u16 = ORDER_BOOK_DELTA_GROUP_BLOCK_LENGTH;
52
53    fn encode_body(&self, writer: &mut SbeWriter<'_>) -> Result<(), SbeEncodeError> {
54        encode_order_book_delta_fields(writer, self);
55        encode_instrument_id(writer, &self.instrument_id)
56    }
57
58    fn decode_body(cursor: &mut SbeCursor<'_>) -> Result<Self, SbeDecodeError> {
59        let action = decode_book_action(cursor)?;
60        let order = decode_book_order(cursor)?;
61        let flags = cursor.read_u8()?;
62        let sequence = cursor.read_u64_le()?;
63        let ts_event = decode_unix_nanos(cursor)?;
64        let ts_init = decode_unix_nanos(cursor)?;
65        let instrument_id = decode_instrument_id(cursor)?;
66
67        Ok(Self {
68            instrument_id,
69            action,
70            order,
71            flags,
72            sequence,
73            ts_event,
74            ts_init,
75        })
76    }
77
78    fn encoded_body_size(&self) -> usize {
79        usize::from(Self::BLOCK_LENGTH) + encoded_instrument_id_size(&self.instrument_id)
80    }
81}
82
83impl MarketSbeMessage for OrderBookDeltas {
84    const TEMPLATE_ID: u16 = template_id::ORDER_BOOK_DELTAS;
85    const BLOCK_LENGTH: u16 = 25;
86
87    fn encode_body(&self, writer: &mut SbeWriter<'_>) -> Result<(), SbeEncodeError> {
88        writer.write_u8(self.flags);
89        writer.write_u64_le(self.sequence);
90        encode_unix_nanos(writer, self.ts_event);
91        encode_unix_nanos(writer, self.ts_init);
92        encode_instrument_id(writer, &self.instrument_id)?;
93        encode_group_header_16(
94            writer,
95            "OrderBookDeltas.deltas",
96            self.deltas.len(),
97            ORDER_BOOK_DELTA_GROUP_BLOCK_LENGTH,
98        )?;
99
100        for delta in &self.deltas {
101            <OrderBookDelta as MarketSbeMessage>::encode_body(delta, writer)?;
102        }
103        Ok(())
104    }
105
106    fn decode_body(cursor: &mut SbeCursor<'_>) -> Result<Self, SbeDecodeError> {
107        let mut scratch = Vec::new();
108        decode_order_book_deltas_body(cursor, &mut scratch)
109    }
110
111    fn encoded_body_size(&self) -> usize {
112        usize::from(Self::BLOCK_LENGTH)
113            + encoded_instrument_id_size(&self.instrument_id)
114            + GROUP_HEADER_16_LENGTH
115            + self
116                .deltas
117                .iter()
118                .map(MarketSbeMessage::encoded_body_size)
119                .sum::<usize>()
120    }
121}
122
123impl FromSbeReuse for OrderBookDeltas {
124    type Scratch = Vec<OrderBookDelta>;
125
126    fn from_sbe_reuse(
127        bytes: &[u8],
128        scratch: &mut Vec<OrderBookDelta>,
129    ) -> Result<Self, SbeDecodeError> {
130        let mut cursor = SbeCursor::new(bytes);
131        let header = decode_header(&mut cursor)?;
132        validate_header(
133            header,
134            <Self as MarketSbeMessage>::TEMPLATE_ID,
135            <Self as MarketSbeMessage>::BLOCK_LENGTH,
136        )?;
137        decode_order_book_deltas_body(&mut cursor, scratch)
138    }
139}
140
141fn decode_order_book_deltas_body(
142    cursor: &mut SbeCursor<'_>,
143    scratch: &mut Vec<OrderBookDelta>,
144) -> Result<OrderBookDeltas, SbeDecodeError> {
145    let flags = cursor.read_u8()?;
146    let sequence = cursor.read_u64_le()?;
147    let ts_event = decode_unix_nanos(cursor)?;
148    let ts_init = decode_unix_nanos(cursor)?;
149    let instrument_id = decode_instrument_id(cursor)?;
150    let (block_length, count) = cursor.read_group_header_16()?;
151
152    if block_length != ORDER_BOOK_DELTA_GROUP_BLOCK_LENGTH {
153        return Err(SbeDecodeError::InvalidBlockLength {
154            expected: ORDER_BOOK_DELTA_GROUP_BLOCK_LENGTH,
155            actual: block_length,
156        });
157    }
158
159    let count = usize::from(count);
160    scratch.clear();
161    scratch.reserve(count);
162
163    for _ in 0..count {
164        scratch.push(<OrderBookDelta as MarketSbeMessage>::decode_body(cursor)?);
165    }
166
167    Ok(OrderBookDeltas {
168        instrument_id,
169        deltas: std::mem::take(scratch),
170        flags,
171        sequence,
172        ts_event,
173        ts_init,
174    })
175}
176
177impl MarketSbeMessage for OrderBookDepth {
178    const TEMPLATE_ID: u16 = template_id::ORDER_BOOK_DEPTH;
179    const BLOCK_LENGTH: u16 =
180        (DEPTH10_LEVEL_BLOCK_LENGTH * 20) + (DEPTH10_COUNTS_BLOCK_LENGTH as u16 * 2) + 25;
181
182    fn template_id(&self) -> u16 {
183        if depth_uses_legacy_layout(self) {
184            Self::TEMPLATE_ID
185        } else {
186            template_id::ORDER_BOOK_DEPTH_FULL
187        }
188    }
189
190    fn block_length(&self) -> u16 {
191        if depth_uses_legacy_layout(self) {
192            Self::BLOCK_LENGTH
193        } else {
194            DEPTH_FULL_BLOCK_LENGTH
195        }
196    }
197
198    fn decode_with_header(
199        header: MessageHeader,
200        cursor: &mut SbeCursor<'_>,
201    ) -> Result<Self, SbeDecodeError> {
202        if header.template_id == template_id::ORDER_BOOK_DEPTH_FULL {
203            validate_header(
204                header,
205                template_id::ORDER_BOOK_DEPTH_FULL,
206                DEPTH_FULL_BLOCK_LENGTH,
207            )?;
208            decode_depth_full(cursor)
209        } else {
210            validate_header(header, Self::TEMPLATE_ID, Self::BLOCK_LENGTH)?;
211            Self::decode_body(cursor)
212        }
213    }
214
215    fn encode_body(&self, writer: &mut SbeWriter<'_>) -> Result<(), SbeEncodeError> {
216        if !depth_uses_legacy_layout(self) {
217            return encode_depth_full(self, writer);
218        }
219
220        for bid in &self.bids {
221            encode_price(writer, &bid.price);
222            encode_quantity(writer, &bid.size);
223        }
224
225        for ask in &self.asks {
226            encode_price(writer, &ask.price);
227            encode_quantity(writer, &ask.size);
228        }
229
230        for count in &self.bid_counts {
231            writer.write_u32_le(*count);
232        }
233
234        for count in &self.ask_counts {
235            writer.write_u32_le(*count);
236        }
237        writer.write_u8(self.flags);
238        writer.write_u64_le(self.sequence);
239        encode_unix_nanos(writer, self.ts_event);
240        encode_unix_nanos(writer, self.ts_init);
241        encode_instrument_id(writer, &self.instrument_id)
242    }
243
244    fn decode_body(cursor: &mut SbeCursor<'_>) -> Result<Self, SbeDecodeError> {
245        let mut bids = [BookOrder::default(); DEPTH10_LEVEL_COUNT];
246        let mut asks = [BookOrder::default(); DEPTH10_LEVEL_COUNT];
247
248        for bid in &mut bids {
249            *bid = BookOrder::new(
250                OrderSide::Buy,
251                decode_price(cursor)?,
252                decode_quantity(cursor)?,
253                0,
254            );
255        }
256
257        for ask in &mut asks {
258            *ask = BookOrder::new(
259                OrderSide::Sell,
260                decode_price(cursor)?,
261                decode_quantity(cursor)?,
262                0,
263            );
264        }
265
266        let mut bid_counts = [0u32; DEPTH10_LEVEL_COUNT];
267        let mut ask_counts = [0u32; DEPTH10_LEVEL_COUNT];
268
269        for count in &mut bid_counts {
270            *count = cursor.read_u32_le()?;
271        }
272
273        for count in &mut ask_counts {
274            *count = cursor.read_u32_le()?;
275        }
276
277        let flags = cursor.read_u8()?;
278        let sequence = cursor.read_u64_le()?;
279        let ts_event = decode_unix_nanos(cursor)?;
280        let ts_init = decode_unix_nanos(cursor)?;
281        let instrument_id = decode_instrument_id(cursor)?;
282
283        Ok(Self {
284            instrument_id,
285            bids: bids.into(),
286            asks: asks.into(),
287            bid_counts: bid_counts.into(),
288            ask_counts: ask_counts.into(),
289            flags,
290            sequence,
291            ts_event,
292            ts_init,
293        })
294    }
295
296    fn encoded_body_size(&self) -> usize {
297        let body = if depth_uses_legacy_layout(self) {
298            usize::from(Self::BLOCK_LENGTH)
299        } else {
300            usize::from(DEPTH_FULL_BLOCK_LENGTH)
301                + GROUP_HEADER_16_LENGTH * 2
302                + (self.bids.len() + self.asks.len()) * usize::from(DEPTH_FULL_LEVEL_BLOCK_LENGTH)
303        };
304
305        body + encoded_instrument_id_size(&self.instrument_id)
306    }
307}
308
309const DEPTH_FULL_BLOCK_LENGTH: u16 = 25;
310const DEPTH_FULL_LEVEL_BLOCK_LENGTH: u16 = BOOK_ORDER_BLOCK_LENGTH + 4;
311
312fn depth_uses_legacy_layout(value: &OrderBookDepth) -> bool {
313    value.bids.len() == DEPTH10_LEVEL_COUNT
314        && value.asks.len() == DEPTH10_LEVEL_COUNT
315        && value.bid_counts.len() == DEPTH10_LEVEL_COUNT
316        && value.ask_counts.len() == DEPTH10_LEVEL_COUNT
317        && value
318            .bids
319            .iter()
320            .all(|order| order.side == Some(OrderSide::Buy) && order.order_id == 0)
321        && value
322            .asks
323            .iter()
324            .all(|order| order.side == Some(OrderSide::Sell) && order.order_id == 0)
325}
326
327fn encode_depth_full(
328    value: &OrderBookDepth,
329    writer: &mut SbeWriter<'_>,
330) -> Result<(), SbeEncodeError> {
331    for (group, count, expected) in [
332        ("bid_counts", value.bid_counts.len(), value.bids.len()),
333        ("ask_counts", value.ask_counts.len(), value.asks.len()),
334    ] {
335        if count != expected {
336            return Err(SbeEncodeError::InvalidGroupSize {
337                group,
338                count,
339                expected,
340            });
341        }
342    }
343
344    writer.write_u8(value.flags);
345    writer.write_u64_le(value.sequence);
346    encode_unix_nanos(writer, value.ts_event);
347    encode_unix_nanos(writer, value.ts_init);
348    encode_instrument_id(writer, &value.instrument_id)?;
349    for (group, orders, counts) in [
350        (
351            "OrderBookDepth.bids",
352            value.bids.as_slice(),
353            value.bid_counts.as_slice(),
354        ),
355        (
356            "OrderBookDepth.asks",
357            value.asks.as_slice(),
358            value.ask_counts.as_slice(),
359        ),
360    ] {
361        encode_group_header_16(writer, group, orders.len(), DEPTH_FULL_LEVEL_BLOCK_LENGTH)?;
362        for (order, count) in orders.iter().zip(counts) {
363            encode_book_order(writer, order);
364            writer.write_u32_le(*count);
365        }
366    }
367
368    Ok(())
369}
370
371pub(super) fn decode_depth_full(
372    cursor: &mut SbeCursor<'_>,
373) -> Result<OrderBookDepth, SbeDecodeError> {
374    let flags = cursor.read_u8()?;
375    let sequence = cursor.read_u64_le()?;
376    let ts_event = decode_unix_nanos(cursor)?;
377    let ts_init = decode_unix_nanos(cursor)?;
378    let instrument_id = decode_instrument_id(cursor)?;
379    let (bids, bid_counts) = decode_depth_levels(cursor)?;
380    let (asks, ask_counts) = decode_depth_levels(cursor)?;
381    OrderBookDepth::new_checked(
382        instrument_id,
383        bids,
384        asks,
385        bid_counts,
386        ask_counts,
387        flags,
388        sequence,
389        ts_event,
390        ts_init,
391    )
392    .map_err(|_| SbeDecodeError::InvalidValue {
393        field: "OrderBookDepth",
394    })
395}
396
397fn decode_depth_levels(
398    cursor: &mut SbeCursor<'_>,
399) -> Result<(Vec<BookOrder>, Vec<u32>), SbeDecodeError> {
400    let (block_length, count) = cursor.read_group_header_16()?;
401    if block_length != DEPTH_FULL_LEVEL_BLOCK_LENGTH {
402        return Err(SbeDecodeError::InvalidBlockLength {
403            expected: DEPTH_FULL_LEVEL_BLOCK_LENGTH,
404            actual: block_length,
405        });
406    }
407
408    let count = usize::from(count);
409    cursor.require(count * usize::from(DEPTH_FULL_LEVEL_BLOCK_LENGTH))?;
410    let mut orders = Vec::with_capacity(count);
411    let mut counts = Vec::with_capacity(count);
412    for _ in 0..count {
413        orders.push(decode_book_order(cursor)?);
414        counts.push(cursor.read_u32_le()?);
415    }
416
417    Ok((orders, counts))
418}
419
420fn encode_book_order(writer: &mut SbeWriter<'_>, order: &BookOrder) {
421    encode_price(writer, &order.price);
422    encode_quantity(writer, &order.size);
423    writer.write_u8(order.side.map_or(0, |side| side as u8));
424    writer.write_u64_le(order.order_id);
425}
426
427fn decode_book_order(cursor: &mut SbeCursor<'_>) -> Result<BookOrder, SbeDecodeError> {
428    let price = decode_price(cursor)?;
429    let size = decode_quantity(cursor)?;
430    let side = decode_order_side(cursor)?;
431    let order_id = cursor.read_u64_le()?;
432    Ok(BookOrder {
433        side,
434        price,
435        size,
436        order_id,
437    })
438}
439
440fn encode_order_book_delta_fields(writer: &mut SbeWriter<'_>, delta: &OrderBookDelta) {
441    writer.write_u8(delta.action as u8);
442    encode_book_order(writer, &delta.order);
443    writer.write_u8(delta.flags);
444    writer.write_u64_le(delta.sequence);
445    encode_unix_nanos(writer, delta.ts_event);
446    encode_unix_nanos(writer, delta.ts_init);
447}
nautilus_serialization/sbe/market/book.rs

nautilus_serialization/sbe/market/
book.rs
