nautilus_binance/spot/sbe/stream/
depth_snapshot.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//! Depth snapshot stream event decoder.
17//!
18//! Message layout (after 8-byte header):
19//! - eventTime: i64 (microseconds)
20//! - bookUpdateId: i64
21//! - priceExponent: i8
22//! - qtyExponent: i8
23//! - bids group (groupSize16Encoding: u16 blockLength + u16 numInGroup):
24//!   - price: i64 (mantissa)
25//!   - qty: i64 (mantissa)
26//! - asks group (groupSize16Encoding: u16 blockLength + u16 numInGroup):
27//!   - price: i64 (mantissa)
28//!   - qty: i64 (mantissa)
29//! - symbol: varString8
30
31use ustr::Ustr;
32
33use super::{MessageHeader, PriceLevel, StreamDecodeError};
34use crate::spot::sbe::cursor::SbeCursor;
35
36/// Depth snapshot stream event (top N levels of order book).
37#[derive(Debug, Clone)]
38pub struct DepthSnapshotStreamEvent {
39    /// Event timestamp in microseconds.
40    pub event_time_us: i64,
41    /// Book update ID for sequencing.
42    pub book_update_id: i64,
43    /// Price exponent (prices = mantissa * 10^exponent).
44    pub price_exponent: i8,
45    /// Quantity exponent (quantities = mantissa * 10^exponent).
46    pub qty_exponent: i8,
47    /// Bid levels (best bid first).
48    pub bids: Vec<PriceLevel>,
49    /// Ask levels (best ask first).
50    pub asks: Vec<PriceLevel>,
51    /// Trading symbol.
52    pub symbol: Ustr,
53}
54
55impl DepthSnapshotStreamEvent {
56    /// Fixed block length (excluding header, groups, and variable-length data).
57    pub const BLOCK_LENGTH: usize = 18;
58
59    /// Decode from SBE buffer (including 8-byte header).
60    ///
61    /// # Errors
62    ///
63    /// Returns an error if buffer is too short, group size exceeds limits, or data is otherwise
64    /// invalid.
65    pub fn decode(buf: &[u8]) -> Result<Self, StreamDecodeError> {
66        let header = MessageHeader::decode(buf)?;
67        header.validate_schema()?;
68        Self::decode_validated(buf)
69    }
70
71    /// Decode from an SBE buffer whose header has already been validated.
72    pub(crate) fn decode_validated(buf: &[u8]) -> Result<Self, StreamDecodeError> {
73        let mut cursor = SbeCursor::new_at(buf, MessageHeader::ENCODED_LENGTH);
74        Self::decode_body(&mut cursor)
75    }
76
77    #[inline]
78    fn decode_body(cursor: &mut SbeCursor<'_>) -> Result<Self, StreamDecodeError> {
79        let event_time_us = cursor.read_i64_le()?;
80        let book_update_id = cursor.read_i64_le()?;
81        let price_exponent = cursor.read_i8()?;
82        let qty_exponent = cursor.read_i8()?;
83
84        let bids = PriceLevel::decode_group(cursor)?;
85        let asks = PriceLevel::decode_group(cursor)?;
86
87        let symbol = Ustr::from(cursor.read_var_string8_ref()?);
88
89        Ok(Self {
90            event_time_us,
91            book_update_id,
92            price_exponent,
93            qty_exponent,
94            bids,
95            asks,
96            symbol,
97        })
98    }
99
100    /// Get price as f64 for a level.
101    #[inline]
102    #[must_use]
103    pub fn level_price(&self, level: &PriceLevel) -> f64 {
104        super::mantissa_to_f64(level.price_mantissa, self.price_exponent)
105    }
106
107    /// Get quantity as f64 for a level.
108    #[inline]
109    #[must_use]
110    pub fn level_qty(&self, level: &PriceLevel) -> f64 {
111        super::mantissa_to_f64(level.qty_mantissa, self.qty_exponent)
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use rstest::rstest;
118
119    use super::*;
120    use crate::spot::sbe::stream::{STREAM_SCHEMA_ID, template_id};
121
122    fn make_valid_buffer(num_bids: usize, num_asks: usize) -> Vec<u8> {
123        let level_block_len = 16u16;
124        let body_size = 18
125            + 4
126            + (num_bids * level_block_len as usize)
127            + 4
128            + (num_asks * level_block_len as usize)
129            + 8;
130        let mut buf = vec![0u8; 8 + body_size];
131
132        // Header
133        buf[0..2].copy_from_slice(&18u16.to_le_bytes()); // block_length
134        buf[2..4].copy_from_slice(&template_id::DEPTH_SNAPSHOT_STREAM_EVENT.to_le_bytes());
135        buf[4..6].copy_from_slice(&STREAM_SCHEMA_ID.to_le_bytes());
136        buf[6..8].copy_from_slice(&0u16.to_le_bytes()); // version
137
138        // Body
139        let body = &mut buf[8..];
140        body[0..8].copy_from_slice(&1000000i64.to_le_bytes()); // event_time_us
141        body[8..16].copy_from_slice(&12345i64.to_le_bytes()); // book_update_id
142        body[16] = (-2i8) as u8; // price_exponent
143        body[17] = (-8i8) as u8; // qty_exponent
144
145        let mut offset = 18;
146
147        // Bids group header
148        body[offset..offset + 2].copy_from_slice(&level_block_len.to_le_bytes());
149        body[offset + 2..offset + 4].copy_from_slice(&(num_bids as u16).to_le_bytes());
150        offset += 4;
151
152        // Bids
153        for i in 0..num_bids {
154            body[offset..offset + 8].copy_from_slice(&(4200000i64 - i as i64 * 100).to_le_bytes());
155            body[offset + 8..offset + 16].copy_from_slice(&100000000i64.to_le_bytes());
156            offset += level_block_len as usize;
157        }
158
159        // Asks group header
160        body[offset..offset + 2].copy_from_slice(&level_block_len.to_le_bytes());
161        body[offset + 2..offset + 4].copy_from_slice(&(num_asks as u16).to_le_bytes());
162        offset += 4;
163
164        // Asks
165        for i in 0..num_asks {
166            body[offset..offset + 8].copy_from_slice(&(4200100i64 + i as i64 * 100).to_le_bytes());
167            body[offset + 8..offset + 16].copy_from_slice(&200000000i64.to_le_bytes());
168            offset += level_block_len as usize;
169        }
170
171        // Symbol: "BTCUSDT"
172        body[offset] = 7;
173        body[offset + 1..offset + 8].copy_from_slice(b"BTCUSDT");
174
175        buf
176    }
177
178    #[rstest]
179    fn test_decode_valid() {
180        let buf = make_valid_buffer(5, 5);
181        let event = DepthSnapshotStreamEvent::decode(&buf).unwrap();
182
183        assert_eq!(event.event_time_us, 1000000);
184        assert_eq!(event.book_update_id, 12345);
185        assert_eq!(event.bids.len(), 5);
186        assert_eq!(event.asks.len(), 5);
187        assert_eq!(event.symbol, "BTCUSDT");
188    }
189
190    #[rstest]
191    fn test_decode_empty_books() {
192        let buf = make_valid_buffer(0, 0);
193        let event = DepthSnapshotStreamEvent::decode(&buf).unwrap();
194
195        assert!(event.bids.is_empty());
196        assert!(event.asks.is_empty());
197    }
198
199    #[rstest]
200    #[case::bids(false)]
201    #[case::asks(true)]
202    fn test_decode_rejects_undersized_depth_blocks(
203        #[case] asks: bool,
204        #[values(0, 1, 10_000)] count: u16,
205    ) {
206        for block_length in 0..PriceLevel::ENCODED_LENGTH as u16 {
207            let mut buf = make_valid_buffer(1, 1);
208
209            let offset = MessageHeader::ENCODED_LENGTH
210                + DepthSnapshotStreamEvent::BLOCK_LENGTH
211                + if asks {
212                    4 + PriceLevel::ENCODED_LENGTH
213                } else {
214                    0
215                };
216
217            buf[offset..offset + 2].copy_from_slice(&block_length.to_le_bytes());
218            buf[offset + 2..offset + 4].copy_from_slice(&u16::to_le_bytes(count));
219
220            let expected = StreamDecodeError::InvalidBlockLength {
221                expected: PriceLevel::ENCODED_LENGTH as u16,
222                actual: block_length,
223            };
224
225            assert_eq!(
226                DepthSnapshotStreamEvent::decode(&buf).unwrap_err(),
227                expected
228            );
229            assert_eq!(
230                DepthSnapshotStreamEvent::decode_validated(&buf).unwrap_err(),
231                expected
232            );
233        }
234    }
235
236    #[rstest]
237    fn test_decode_depth_extensions_and_symbol() {
238        let mut buf = make_valid_buffer(2, 2);
239        let width = PriceLevel::ENCODED_LENGTH;
240        let bids_header = MessageHeader::ENCODED_LENGTH + DepthSnapshotStreamEvent::BLOCK_LENGTH;
241        let asks_header = bids_header + 4 + 2 * width;
242        for header in [asks_header, bids_header] {
243            for index in (0..2).rev() {
244                let end = header + 4 + (index + 1) * width;
245                buf.splice(end..end, [0xFF; 3]);
246            }
247
248            buf[header..header + 2].copy_from_slice(&((width + 3) as u16).to_le_bytes());
249        }
250
251        let event = DepthSnapshotStreamEvent::decode(&buf).unwrap();
252
253        assert_eq!(event.event_time_us, 1_000_000);
254        assert_eq!(event.book_update_id, 12_345);
255        assert_eq!(event.price_exponent, -2);
256        assert_eq!(event.qty_exponent, -8);
257        assert_eq!(
258            event
259                .bids
260                .iter()
261                .map(|level| (level.price_mantissa, level.qty_mantissa))
262                .collect::<Vec<_>>(),
263            vec![(4_200_000, 100_000_000), (4_199_900, 100_000_000)]
264        );
265        assert_eq!(
266            event
267                .asks
268                .iter()
269                .map(|level| (level.price_mantissa, level.qty_mantissa))
270                .collect::<Vec<_>>(),
271            vec![(4_200_100, 200_000_000), (4_200_200, 200_000_000)]
272        );
273        assert_eq!(event.symbol, "BTCUSDT");
274    }
275
276    #[rstest]
277    fn test_decode_truncated() {
278        let mut buf = make_valid_buffer(10, 10);
279        buf.truncate(100); // Truncate in the middle
280        let err = DepthSnapshotStreamEvent::decode(&buf).unwrap_err();
281        assert!(matches!(err, StreamDecodeError::BufferTooShort { .. }));
282    }
283
284    #[rstest]
285    fn test_decode_wrong_schema() {
286        let mut buf = make_valid_buffer(5, 5);
287        buf[4..6].copy_from_slice(&99u16.to_le_bytes());
288        let err = DepthSnapshotStreamEvent::decode(&buf).unwrap_err();
289        assert!(matches!(err, StreamDecodeError::SchemaMismatch { .. }));
290    }
291
292    #[rstest]
293    fn test_decode_validated_matches_decode() {
294        let buf = make_valid_buffer(2, 3);
295
296        let decode_event = DepthSnapshotStreamEvent::decode(&buf).unwrap();
297        let validated_event = DepthSnapshotStreamEvent::decode_validated(&buf).unwrap();
298
299        assert_eq!(validated_event.event_time_us, decode_event.event_time_us);
300        assert_eq!(validated_event.book_update_id, decode_event.book_update_id);
301        assert_eq!(validated_event.price_exponent, decode_event.price_exponent);
302        assert_eq!(validated_event.qty_exponent, decode_event.qty_exponent);
303        assert_eq!(validated_event.bids.len(), decode_event.bids.len());
304        assert_eq!(validated_event.asks.len(), decode_event.asks.len());
305        assert_eq!(validated_event.symbol, decode_event.symbol);
306    }
307}
nautilus_binance/spot/sbe/stream/depth_snapshot.rs

nautilus_binance/spot/sbe/stream/
depth_snapshot.rs
