nautilus_serialization/sbe/
market.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
16//! Hand-written SBE codecs for Nautilus market data types.
17
18mod bars;
19mod book;
20mod common;
21mod data_any;
22mod ticks;
23
24use nautilus_model::data::{
25    Bar, FundingRateUpdate, IndexPriceUpdate, InstrumentClose, InstrumentStatus, MarkPriceUpdate,
26    OptionGreeks, OrderBookDelta, OrderBookDeltas, OrderBookDepth, QuoteTick, TradeTick,
27};
28
29use self::common::{HEADER_LENGTH, MessageHeader, decode_header, encode_header, validate_header};
30use super::{SbeCursor, SbeDecodeError, SbeEncodeError, SbeWriter};
31
32pub const MARKET_SCHEMA_ID: u16 = 1;
33pub const MARKET_SCHEMA_VERSION: u16 = 0;
34
35pub(super) mod data_any_variant {
36    pub(crate) const ORDER_BOOK_DELTA: u16 = 0;
37    pub(crate) const ORDER_BOOK_DELTAS: u16 = 1;
38    pub(crate) const ORDER_BOOK_DEPTH: u16 = 2;
39    pub(crate) const QUOTE: u16 = 3;
40    pub(crate) const TRADE: u16 = 4;
41    pub(crate) const BAR: u16 = 5;
42    pub(crate) const MARK_PRICE: u16 = 6;
43    pub(crate) const INDEX_PRICE: u16 = 7;
44    pub(crate) const FUNDING_RATE: u16 = 8;
45    pub(crate) const OPTION_GREEKS: u16 = 9;
46    pub(crate) const INSTRUMENT_STATUS: u16 = 10;
47    pub(crate) const INSTRUMENT_CLOSE: u16 = 11;
48    pub(crate) const ORDER_BOOK_DEPTH_FULL: u16 = 12;
49}
50
51pub(super) mod template_id {
52    pub(crate) const BOOK_ORDER: u16 = 30_001;
53    pub(crate) const ORDER_BOOK_DELTA: u16 = 30_002;
54    pub(crate) const ORDER_BOOK_DELTAS: u16 = 30_003;
55    pub(crate) const ORDER_BOOK_DEPTH: u16 = 30_004;
56    pub(crate) const QUOTE_TICK: u16 = 30_005;
57    pub(crate) const TRADE_TICK: u16 = 30_006;
58    pub(crate) const BAR_TYPE: u16 = 30_007;
59    pub(crate) const BAR: u16 = 30_008;
60    pub(crate) const MARK_PRICE_UPDATE: u16 = 30_009;
61    pub(crate) const INDEX_PRICE_UPDATE: u16 = 30_010;
62    pub(crate) const FUNDING_RATE_UPDATE: u16 = 30_011;
63    pub(crate) const OPTION_GREEKS: u16 = 30_012;
64    pub(crate) const INSTRUMENT_STATUS: u16 = 30_013;
65    pub(crate) const INSTRUMENT_CLOSE: u16 = 30_014;
66    pub(crate) const DATA_ANY: u16 = 30_015;
67    pub(crate) const ORDER_BOOK_DEPTH_FULL: u16 = 30_016;
68}
69
70#[expect(clippy::large_enum_variant)]
71#[derive(Debug, Clone, PartialEq)]
72pub enum DataAny {
73    BookDelta(OrderBookDelta),
74    BookDeltas(OrderBookDeltas),
75    BookDepth(OrderBookDepth),
76    Quote(QuoteTick),
77    Trade(TradeTick),
78    Bar(Bar),
79    MarkPrice(MarkPriceUpdate),
80    IndexPrice(IndexPriceUpdate),
81    FundingRate(FundingRateUpdate),
82    OptionGreeks(OptionGreeks),
83    InstrumentStatus(InstrumentStatus),
84    InstrumentClose(InstrumentClose),
85}
86
87pub trait ToSbe {
88    /// Encodes the value into an SBE message buffer.
89    ///
90    /// # Errors
91    ///
92    /// Returns an error if any field cannot be encoded into the target SBE wire format.
93    fn to_sbe(&self) -> Result<Vec<u8>, SbeEncodeError>;
94
95    /// Encodes the value into the provided SBE message buffer.
96    ///
97    /// This method clears any existing bytes in `buf` before encoding.
98    ///
99    /// # Errors
100    ///
101    /// Returns an error if any field cannot be encoded into the target SBE wire format.
102    fn to_sbe_into(&self, buf: &mut Vec<u8>) -> Result<(), SbeEncodeError> {
103        let bytes = self.to_sbe()?;
104        buf.clear();
105        buf.extend_from_slice(&bytes);
106        Ok(())
107    }
108}
109
110pub trait FromSbe: Sized {
111    /// Decodes the value from an SBE message buffer.
112    ///
113    /// # Errors
114    ///
115    /// Returns an error if the header is invalid or the payload is malformed.
116    fn from_sbe(bytes: &[u8]) -> Result<Self, SbeDecodeError>;
117}
118
119/// Extension of [`FromSbe`] that reuses allocations between decodes.
120///
121/// Scalar messages decode without heap allocation, so they do not need this trait. Types with
122/// growable internal buffers (for example [`OrderBookDeltas`] with its `Vec<OrderBookDelta>`)
123/// implement it to let callers supply a pre-allocated scratch buffer and avoid per-message
124/// allocation in hot paths.
125pub trait FromSbeReuse: FromSbe {
126    /// Scratch buffer whose allocation is reused across decodes.
127    type Scratch;
128
129    /// Decodes a value from an SBE message buffer, reusing `scratch`'s allocation.
130    ///
131    /// On success, ownership of the allocation moves from `scratch` into the returned value and
132    /// `scratch` is left in its empty state. To continue reusing the allocation, move the buffer
133    /// back from the returned value (for example `scratch = std::mem::take(&mut result.deltas)`).
134    ///
135    /// # Errors
136    ///
137    /// Returns an error if the header is invalid or the payload is malformed.
138    fn from_sbe_reuse(bytes: &[u8], scratch: &mut Self::Scratch) -> Result<Self, SbeDecodeError>;
139}
140
141pub(super) trait MarketSbeMessage: Sized {
142    const TEMPLATE_ID: u16;
143    const BLOCK_LENGTH: u16;
144
145    fn template_id(&self) -> u16 {
146        Self::TEMPLATE_ID
147    }
148
149    fn block_length(&self) -> u16 {
150        Self::BLOCK_LENGTH
151    }
152
153    fn decode_with_header(
154        header: MessageHeader,
155        cursor: &mut SbeCursor<'_>,
156    ) -> Result<Self, SbeDecodeError> {
157        validate_header(header, Self::TEMPLATE_ID, Self::BLOCK_LENGTH)?;
158        Self::decode_body(cursor)
159    }
160
161    fn encode_body(&self, writer: &mut SbeWriter<'_>) -> Result<(), SbeEncodeError>;
162
163    fn decode_body(cursor: &mut SbeCursor<'_>) -> Result<Self, SbeDecodeError>;
164
165    fn encoded_body_size(&self) -> usize {
166        usize::from(Self::BLOCK_LENGTH)
167    }
168}
169
170impl<T> ToSbe for T
171where
172    T: MarketSbeMessage,
173{
174    #[inline]
175    fn to_sbe(&self) -> Result<Vec<u8>, SbeEncodeError> {
176        let encoded_size = HEADER_LENGTH + self.encoded_body_size();
177        let mut buf = Vec::with_capacity(encoded_size);
178        encode_into_uninit(self, &mut buf, encoded_size)?;
179        Ok(buf)
180    }
181
182    #[inline]
183    fn to_sbe_into(&self, buf: &mut Vec<u8>) -> Result<(), SbeEncodeError> {
184        let encoded_size = HEADER_LENGTH + self.encoded_body_size();
185        buf.clear();
186        buf.reserve(encoded_size);
187        encode_into_uninit(self, buf, encoded_size)
188    }
189}
190
191impl<T> FromSbe for T
192where
193    T: MarketSbeMessage,
194{
195    #[inline]
196    fn from_sbe(bytes: &[u8]) -> Result<Self, SbeDecodeError> {
197        let mut cursor = SbeCursor::new(bytes);
198        let header = decode_header(&mut cursor)?;
199        T::decode_with_header(header, &mut cursor)
200    }
201}
202
203// Writes an SBE message into the spare capacity of `buf` without zero
204// initialization, then commits the length on success. Caller must ensure
205// `buf.len() == 0` and `buf.capacity() >= encoded_size`.
206#[inline]
207#[allow(
208    unsafe_code,
209    reason = "set_len commits writes the SbeWriter has already made into spare capacity"
210)]
211#[allow(
212    clippy::panic_in_result_fn,
213    reason = "load-bearing safety check for the unsafe set_len; panic is the right outcome"
214)]
215fn encode_into_uninit<T>(
216    value: &T,
217    buf: &mut Vec<u8>,
218    encoded_size: usize,
219) -> Result<(), SbeEncodeError>
220where
221    T: MarketSbeMessage,
222{
223    debug_assert_eq!(buf.len(), 0);
224    debug_assert!(buf.capacity() >= encoded_size);
225
226    let spare = &mut buf.spare_capacity_mut()[..encoded_size];
227    let mut writer = SbeWriter::new_uninit(spare);
228    encode_header(
229        &mut writer,
230        value.block_length(),
231        value.template_id(),
232        MARKET_SCHEMA_ID,
233        MARKET_SCHEMA_VERSION,
234    );
235    value.encode_body(&mut writer)?;
236
237    // Load-bearing for the unsafe `set_len` below: this is the invariant that
238    // converts the writer's per-byte initialization into Vec-level safety. Run
239    // in release builds too so a future size mismatch panics rather than
240    // commits uninit bytes.
241    assert_eq!(
242        writer.pos(),
243        encoded_size,
244        "SBE encode_body wrote {} bytes but encoded_body_size reported {}",
245        writer.pos(),
246        encoded_size,
247    );
248
249    // SAFETY: the writer panics if it attempts to write past `encoded_size`,
250    // the assert above confirms it wrote exactly `encoded_size` bytes, and
251    // errors propagate before `set_len` runs. Reaching this line means the
252    // first `encoded_size` bytes of `buf` hold initialized u8 values.
253    unsafe {
254        buf.set_len(encoded_size);
255    }
256    Ok(())
257}
nautilus_serialization/sbe/market.rs

nautilus_serialization/sbe/
market.rs
