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

nautilus_binance/spot/sbe/stream/
trades.rs
