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

nautilus_binance/spot/sbe/stream/
depth_diff.rs
