nautilus_binance/spot/websocket/streams/
parse.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//! Parsing utilities for Binance Spot WebSocket SBE messages.
17
18use nautilus_core::nanos::UnixNanos;
19use nautilus_model::{
20    data::{BookOrder, Data, OrderBookDelta, OrderBookDeltas, QuoteTick, TradeTick},
21    enums::{AggressorSide, BookAction, OrderSide, RecordFlag},
22    identifiers::TradeId,
23    instruments::{Instrument, InstrumentAny},
24    types::{Price, Quantity},
25};
26
27use crate::{
28    common::parse::parse_micros_or_init,
29    spot::sbe::stream::{
30        BestBidAskStreamEvent, DepthDiffStreamEvent, DepthSnapshotStreamEvent, MessageHeader,
31        StreamDecodeError, TradesStreamEvent, template_id,
32    },
33};
34
35/// Decoded market data message.
36#[derive(Debug)]
37pub enum MarketDataMessage {
38    /// Trade event.
39    Trades(TradesStreamEvent),
40    /// Best bid/ask update.
41    BestBidAsk(BestBidAskStreamEvent),
42    /// Order book snapshot.
43    DepthSnapshot(DepthSnapshotStreamEvent),
44    /// Order book diff update.
45    DepthDiff(DepthDiffStreamEvent),
46}
47
48/// Decode an SBE binary frame into a market data message.
49///
50/// Validates the message header (including schema ID) and routes to the
51/// appropriate decoder based on template ID.
52///
53/// # Errors
54///
55/// Returns an error if the buffer is too short, schema validation fails,
56/// or the template ID is unknown.
57pub fn decode_market_data(buf: &[u8]) -> Result<MarketDataMessage, StreamDecodeError> {
58    let header = MessageHeader::decode(buf)?;
59    header.validate_schema()?;
60
61    match header.template_id {
62        template_id::TRADES_STREAM_EVENT => Ok(MarketDataMessage::Trades(
63            TradesStreamEvent::decode_validated(buf)?,
64        )),
65        template_id::BEST_BID_ASK_STREAM_EVENT => Ok(MarketDataMessage::BestBidAsk(
66            BestBidAskStreamEvent::decode_validated(buf)?,
67        )),
68        template_id::DEPTH_SNAPSHOT_STREAM_EVENT => Ok(MarketDataMessage::DepthSnapshot(
69            DepthSnapshotStreamEvent::decode_validated(buf)?,
70        )),
71        template_id::DEPTH_DIFF_STREAM_EVENT => Ok(MarketDataMessage::DepthDiff(
72            DepthDiffStreamEvent::decode_validated(buf)?,
73        )),
74        _ => Err(StreamDecodeError::UnknownTemplateId(header.template_id)),
75    }
76}
77
78/// Parses a trades stream event into a vector of `TradeTick` using the supplied
79/// adapter initialization timestamp.
80pub fn parse_trades_event(
81    event: &TradesStreamEvent,
82    instrument: &InstrumentAny,
83    ts_init: UnixNanos,
84) -> Vec<Data> {
85    let instrument_id = instrument.id();
86    let price_precision = instrument.price_precision();
87    let size_precision = instrument.size_precision();
88
89    event
90        .trades
91        .iter()
92        .map(|t| {
93            let price = Price::from_mantissa_exponent(
94                t.price_mantissa,
95                event.price_exponent,
96                price_precision,
97            );
98            let size = Quantity::from_mantissa_exponent(
99                t.qty_mantissa as u64,
100                event.qty_exponent,
101                size_precision,
102            );
103            let ts_event = parse_micros_or_init(
104                event.transact_time_us,
105                "Spot SBE stream transaction time",
106                ts_init,
107            );
108
109            let trade = TradeTick::new(
110                instrument_id,
111                price,
112                size,
113                if t.is_buyer_maker {
114                    AggressorSide::Sell
115                } else {
116                    AggressorSide::Buy
117                },
118                TradeId::new(t.id.to_string()),
119                ts_event,
120                ts_init,
121            );
122            Data::from(trade)
123        })
124        .collect()
125}
126
127/// Parses a best bid/ask event into a `QuoteTick` using the supplied adapter
128/// initialization timestamp.
129pub fn parse_bbo_event(
130    event: &BestBidAskStreamEvent,
131    instrument: &InstrumentAny,
132    ts_init: UnixNanos,
133) -> QuoteTick {
134    let instrument_id = instrument.id();
135    let price_precision = instrument.price_precision();
136    let size_precision = instrument.size_precision();
137
138    let bid_price = Price::from_mantissa_exponent(
139        event.bid_price_mantissa,
140        event.price_exponent,
141        price_precision,
142    );
143    let bid_size = Quantity::from_mantissa_exponent(
144        event.bid_qty_mantissa as u64,
145        event.qty_exponent,
146        size_precision,
147    );
148    let ask_price = Price::from_mantissa_exponent(
149        event.ask_price_mantissa,
150        event.price_exponent,
151        price_precision,
152    );
153    let ask_size = Quantity::from_mantissa_exponent(
154        event.ask_qty_mantissa as u64,
155        event.qty_exponent,
156        size_precision,
157    );
158    let ts_event = parse_micros_or_init(event.event_time_us, "Spot SBE BBO event time", ts_init);
159
160    QuoteTick::new(
161        instrument_id,
162        bid_price,
163        ask_price,
164        bid_size,
165        ask_size,
166        ts_event,
167        ts_init,
168    )
169}
170
171/// Parses a depth snapshot event into `OrderBookDeltas` using the supplied
172/// adapter initialization timestamp for the aggregate and every inner delta.
173///
174/// A snapshot without levels clears the book.
175pub fn parse_depth_snapshot(
176    event: &DepthSnapshotStreamEvent,
177    instrument: &InstrumentAny,
178    ts_init: UnixNanos,
179) -> OrderBookDeltas {
180    let instrument_id = instrument.id();
181    let price_precision = instrument.price_precision();
182    let size_precision = instrument.size_precision();
183    let ts_event = parse_micros_or_init(
184        event.event_time_us,
185        "Spot SBE depth snapshot event time",
186        ts_init,
187    );
188    let sequence = event.book_update_id as u64;
189
190    let mut deltas = Vec::with_capacity(event.bids.len() + event.asks.len() + 1);
191
192    // Add clear delta first
193    deltas.push(OrderBookDelta::clear(
194        instrument_id,
195        sequence,
196        ts_event,
197        ts_init,
198    ));
199
200    for (side, levels) in [
201        (OrderSide::Buy, &event.bids),
202        (OrderSide::Sell, &event.asks),
203    ] {
204        for level in levels {
205            let price = Price::from_mantissa_exponent(
206                level.price_mantissa,
207                event.price_exponent,
208                price_precision,
209            );
210            let size = Quantity::from_mantissa_exponent(
211                level.qty_mantissa as u64,
212                event.qty_exponent,
213                size_precision,
214            );
215            let order = BookOrder::new(side, price, size, 0);
216
217            deltas.push(OrderBookDelta::new(
218                instrument_id,
219                BookAction::Add,
220                order,
221                RecordFlag::F_SNAPSHOT as u8,
222                sequence,
223                ts_event,
224                ts_init,
225            ));
226        }
227    }
228
229    if let Some(last) = deltas.last_mut() {
230        last.flags |= RecordFlag::F_LAST as u8;
231    }
232
233    OrderBookDeltas::new(instrument_id, deltas)
234}
235
236/// Parses a depth diff event into `OrderBookDeltas` using the supplied adapter
237/// initialization timestamp for the aggregate and every inner delta.
238///
239/// Returns `None` if the diff contains no updates.
240pub fn parse_depth_diff(
241    event: &DepthDiffStreamEvent,
242    instrument: &InstrumentAny,
243    ts_init: UnixNanos,
244) -> Option<OrderBookDeltas> {
245    let instrument_id = instrument.id();
246    let price_precision = instrument.price_precision();
247    let size_precision = instrument.size_precision();
248    let ts_event = parse_micros_or_init(
249        event.event_time_us,
250        "Spot SBE depth diff event time",
251        ts_init,
252    );
253    let sequence = event.last_book_update_id as u64;
254
255    let mut deltas = Vec::with_capacity(event.bids.len() + event.asks.len());
256
257    // Add bid updates
258    for (i, level) in event.bids.iter().enumerate() {
259        let price = Price::from_mantissa_exponent(
260            level.price_mantissa,
261            event.price_exponent,
262            price_precision,
263        );
264        let size = Quantity::from_mantissa_exponent(
265            level.qty_mantissa as u64,
266            event.qty_exponent,
267            size_precision,
268        );
269
270        // Zero size means delete, otherwise update
271        let action = if level.qty_mantissa == 0 {
272            BookAction::Delete
273        } else {
274            BookAction::Update
275        };
276
277        let flags = if i == event.bids.len() - 1 && event.asks.is_empty() {
278            RecordFlag::F_LAST as u8
279        } else {
280            0
281        };
282
283        let order = BookOrder::new(OrderSide::Buy, price, size, 0);
284
285        deltas.push(OrderBookDelta::new(
286            instrument_id,
287            action,
288            order,
289            flags,
290            sequence,
291            ts_event,
292            ts_init,
293        ));
294    }
295
296    // Add ask updates
297    for (i, level) in event.asks.iter().enumerate() {
298        let price = Price::from_mantissa_exponent(
299            level.price_mantissa,
300            event.price_exponent,
301            price_precision,
302        );
303        let size = Quantity::from_mantissa_exponent(
304            level.qty_mantissa as u64,
305            event.qty_exponent,
306            size_precision,
307        );
308
309        let action = if level.qty_mantissa == 0 {
310            BookAction::Delete
311        } else {
312            BookAction::Update
313        };
314
315        let flags = if i == event.asks.len() - 1 {
316            RecordFlag::F_LAST as u8
317        } else {
318            0
319        };
320
321        let order = BookOrder::new(OrderSide::Sell, price, size, 0);
322
323        deltas.push(OrderBookDelta::new(
324            instrument_id,
325            action,
326            order,
327            flags,
328            sequence,
329            ts_event,
330            ts_init,
331        ));
332    }
333
334    if deltas.is_empty() {
335        return None;
336    }
337
338    Some(OrderBookDeltas::new(instrument_id, deltas))
339}
340
341#[cfg(test)]
342mod tests {
343    use rstest::rstest;
344    use ustr::Ustr;
345
346    use super::*;
347    use crate::{
348        common::parse::parse_spot_instrument_sbe,
349        spot::{
350            http::models::{
351                BinanceLotSizeFilterSbe, BinancePriceFilterSbe, BinanceSymbolFiltersSbe,
352                BinanceSymbolSbe,
353            },
354            sbe::stream::{PriceLevel, STREAM_SCHEMA_ID, Trade},
355        },
356    };
357
358    fn make_bbo_buffer() -> Vec<u8> {
359        let mut buf = vec![0u8; 70];
360
361        // Header
362        buf[0..2].copy_from_slice(&50u16.to_le_bytes()); // block_length
363        buf[2..4].copy_from_slice(&template_id::BEST_BID_ASK_STREAM_EVENT.to_le_bytes());
364        buf[4..6].copy_from_slice(&STREAM_SCHEMA_ID.to_le_bytes());
365        buf[6..8].copy_from_slice(&0u16.to_le_bytes()); // version
366
367        // Body
368        let body = &mut buf[8..];
369        body[0..8].copy_from_slice(&1000000i64.to_le_bytes()); // event_time_us
370        body[8..16].copy_from_slice(&12345i64.to_le_bytes()); // book_update_id
371        body[16] = (-2i8) as u8; // price_exponent
372        body[17] = (-8i8) as u8; // qty_exponent
373        body[18..26].copy_from_slice(&4200000i64.to_le_bytes()); // bid_price
374        body[26..34].copy_from_slice(&100000000i64.to_le_bytes()); // bid_qty
375        body[34..42].copy_from_slice(&4200100i64.to_le_bytes()); // ask_price
376        body[42..50].copy_from_slice(&200000000i64.to_le_bytes()); // ask_qty
377
378        // Symbol: "BTCUSDT" (7 bytes)
379        body[50] = 7;
380        body[51..58].copy_from_slice(b"BTCUSDT");
381
382        buf
383    }
384
385    fn sample_instrument() -> InstrumentAny {
386        let symbol = BinanceSymbolSbe {
387            symbol: "ETHUSDT".to_string(),
388            base_asset: "ETH".to_string(),
389            quote_asset: "USDT".to_string(),
390            base_asset_precision: 8,
391            quote_asset_precision: 8,
392            status: 0,
393            order_types: 0,
394            iceberg_allowed: true,
395            oco_allowed: true,
396            oto_allowed: false,
397            quote_order_qty_market_allowed: true,
398            allow_trailing_stop: true,
399            cancel_replace_allowed: true,
400            amend_allowed: true,
401            is_spot_trading_allowed: true,
402            is_margin_trading_allowed: false,
403            filters: BinanceSymbolFiltersSbe {
404                notional_filters: Vec::new(),
405                price_filter: Some(BinancePriceFilterSbe {
406                    price_exponent: -8,
407                    min_price: 1_000_000,
408                    max_price: 100_000_000_000_000,
409                    tick_size: 1_000_000,
410                }),
411                lot_size_filter: Some(BinanceLotSizeFilterSbe {
412                    qty_exponent: -8,
413                    min_qty: 10_000,
414                    max_qty: 900_000_000_000,
415                    step_size: 10_000,
416                }),
417            },
418            permissions: vec![vec!["SPOT".to_string()]],
419        };
420
421        let ts = UnixNanos::from(1_700_000_000_000_000_000u64);
422        parse_spot_instrument_sbe(&symbol, ts, ts).unwrap()
423    }
424
425    #[rstest]
426    fn test_decode_empty_buffer() {
427        let err = decode_market_data(&[]).unwrap_err();
428        assert!(matches!(err, StreamDecodeError::BufferTooShort { .. }));
429    }
430
431    #[rstest]
432    fn test_decode_short_buffer() {
433        let buf = [0u8; 5];
434        let err = decode_market_data(&buf).unwrap_err();
435        assert!(matches!(err, StreamDecodeError::BufferTooShort { .. }));
436    }
437
438    #[rstest]
439    fn test_decode_wrong_schema() {
440        let mut buf = [0u8; 100];
441        buf[0..2].copy_from_slice(&50u16.to_le_bytes()); // block_length
442        buf[2..4].copy_from_slice(&template_id::BEST_BID_ASK_STREAM_EVENT.to_le_bytes());
443        buf[4..6].copy_from_slice(&99u16.to_le_bytes()); // Wrong schema
444        buf[6..8].copy_from_slice(&0u16.to_le_bytes()); // version
445
446        let err = decode_market_data(&buf).unwrap_err();
447        assert!(matches!(err, StreamDecodeError::SchemaMismatch { .. }));
448    }
449
450    #[rstest]
451    fn test_decode_unknown_template() {
452        let mut buf = [0u8; 100];
453        buf[0..2].copy_from_slice(&50u16.to_le_bytes()); // block_length
454        buf[2..4].copy_from_slice(&9999u16.to_le_bytes()); // Unknown template
455        buf[4..6].copy_from_slice(&STREAM_SCHEMA_ID.to_le_bytes());
456        buf[6..8].copy_from_slice(&0u16.to_le_bytes()); // version
457
458        let err = decode_market_data(&buf).unwrap_err();
459        assert!(matches!(err, StreamDecodeError::UnknownTemplateId(9999)));
460    }
461
462    #[rstest]
463    fn test_decode_valid_best_bid_ask() {
464        let buf = make_bbo_buffer();
465        let msg = decode_market_data(&buf).unwrap();
466
467        match msg {
468            MarketDataMessage::BestBidAsk(event) => {
469                assert_eq!(event.event_time_us, 1_000_000);
470                assert_eq!(event.symbol, Ustr::from("BTCUSDT"));
471            }
472            _ => panic!("Expected BestBidAsk"),
473        }
474    }
475
476    #[rstest]
477    fn test_parse_trades_event() {
478        let instrument = sample_instrument();
479        let ts_init = UnixNanos::from(1_800_000_000_000_000_000u64);
480        let event = TradesStreamEvent {
481            event_time_us: 1_700_000_000_000_000,
482            transact_time_us: 1_700_000_000_100_000,
483            price_exponent: -2,
484            qty_exponent: -4,
485            trades: vec![
486                Trade {
487                    id: 1,
488                    price_mantissa: 12_345,
489                    qty_mantissa: 25_000,
490                    is_buyer_maker: false,
491                },
492                Trade {
493                    id: 2,
494                    price_mantissa: 12_340,
495                    qty_mantissa: 10_000,
496                    is_buyer_maker: true,
497                },
498            ],
499            symbol: Ustr::from("ETHUSDT"),
500        };
501
502        let data = parse_trades_event(&event, &instrument, ts_init);
503
504        assert_eq!(data.len(), 2);
505        match &data[0] {
506            Data::Trade(trade) => {
507                assert_eq!(trade.instrument_id, instrument.id());
508                assert_eq!(trade.price, Price::new(123.45, 2));
509                assert_eq!(trade.size, Quantity::new(2.5, 4));
510                assert_eq!(trade.aggressor_side, AggressorSide::Buy);
511                assert_eq!(trade.trade_id, TradeId::new("1"));
512                assert_eq!(
513                    trade.ts_event,
514                    UnixNanos::from(1_700_000_000_100_000_000u64)
515                );
516                assert_eq!(trade.ts_init, ts_init);
517            }
518            other => panic!("Expected trade data, was {other:?}"),
519        }
520
521        match &data[1] {
522            Data::Trade(trade) => {
523                assert_eq!(trade.aggressor_side, AggressorSide::Sell);
524                assert_eq!(trade.ts_init, ts_init);
525            }
526            other => panic!("Expected trade data, was {other:?}"),
527        }
528    }
529
530    #[rstest]
531    fn test_parse_bbo_event() {
532        let instrument = sample_instrument();
533        let ts_init = UnixNanos::from(1_800_000_000_000_000_000u64);
534        let event = BestBidAskStreamEvent {
535            event_time_us: 1_700_000_000_000_000,
536            book_update_id: 123,
537            price_exponent: -2,
538            qty_exponent: -4,
539            bid_price_mantissa: 12_345,
540            bid_qty_mantissa: 25_000,
541            ask_price_mantissa: 12_350,
542            ask_qty_mantissa: 30_000,
543            symbol: Ustr::from("ETHUSDT"),
544        };
545
546        let quote = parse_bbo_event(&event, &instrument, ts_init);
547
548        assert_eq!(quote.instrument_id, instrument.id());
549        assert_eq!(quote.bid_price, Price::new(123.45, 2));
550        assert_eq!(quote.ask_price, Price::new(123.50, 2));
551        assert_eq!(quote.bid_size, Quantity::new(2.5, 4));
552        assert_eq!(quote.ask_size, Quantity::new(3.0, 4));
553        assert_eq!(
554            quote.ts_event,
555            UnixNanos::from(1_700_000_000_000_000_000u64)
556        );
557        assert_eq!(quote.ts_init, ts_init);
558    }
559
560    #[rstest]
561    #[case::negative(-1)]
562    #[case::overflow(i64::MAX)]
563    fn test_parse_bbo_event_falls_back_for_invalid_timestamp(#[case] event_time_us: i64) {
564        let instrument = sample_instrument();
565        let event = BestBidAskStreamEvent {
566            event_time_us,
567            book_update_id: 123,
568            price_exponent: -2,
569            qty_exponent: -4,
570            bid_price_mantissa: 12_345,
571            bid_qty_mantissa: 25_000,
572            ask_price_mantissa: 12_350,
573            ask_qty_mantissa: 30_000,
574            symbol: Ustr::from("ETHUSDT"),
575        };
576
577        let ts_init = UnixNanos::from(1);
578        let quote = parse_bbo_event(&event, &instrument, ts_init);
579
580        assert_eq!(quote.ts_event, ts_init);
581        assert_eq!(quote.ts_init, ts_init);
582    }
583
584    #[rstest]
585    fn test_parse_depth_snapshot() {
586        let instrument = sample_instrument();
587        let ts_init = UnixNanos::from(1_800_000_000_000_000_000u64);
588        let event = DepthSnapshotStreamEvent {
589            event_time_us: 1_700_000_000_000_000,
590            book_update_id: 123,
591            price_exponent: -2,
592            qty_exponent: -4,
593            bids: vec![PriceLevel {
594                price_mantissa: 12_345,
595                qty_mantissa: 25_000,
596            }],
597            asks: vec![PriceLevel {
598                price_mantissa: 12_350,
599                qty_mantissa: 30_000,
600            }],
601            symbol: Ustr::from("ETHUSDT"),
602        };
603
604        let deltas = parse_depth_snapshot(&event, &instrument, ts_init);
605
606        assert_eq!(deltas.instrument_id, instrument.id());
607        assert_eq!(deltas.deltas.len(), 3);
608        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
609        assert_eq!(deltas.deltas[1].action, BookAction::Add);
610        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy.into());
611        assert_eq!(deltas.deltas[1].order.price, Price::new(123.45, 2));
612        assert_eq!(deltas.deltas[1].order.size, Quantity::new(2.5, 4));
613        assert_eq!(deltas.deltas[2].action, BookAction::Add);
614        assert_eq!(deltas.deltas[2].order.side, OrderSide::Sell.into());
615        assert_eq!(deltas.deltas[2].order.price, Price::new(123.50, 2));
616        assert_eq!(deltas.deltas[2].order.size, Quantity::new(3.0, 4));
617        assert_eq!(deltas.deltas[1].flags, RecordFlag::F_SNAPSHOT as u8);
618        assert_eq!(
619            deltas.deltas[2].flags,
620            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
621        );
622        assert_eq!(deltas.deltas[0].sequence, 123);
623        assert_eq!(deltas.deltas[2].sequence, 123);
624        assert_eq!(
625            deltas.ts_event,
626            UnixNanos::from(1_700_000_000_000_000_000u64)
627        );
628        assert_eq!(deltas.ts_init, ts_init);
629        assert!(deltas.deltas.iter().all(|delta| delta.ts_init == ts_init));
630    }
631
632    #[rstest]
633    fn test_parse_depth_snapshot_empty_clears_book() {
634        let instrument = sample_instrument();
635        let event = DepthSnapshotStreamEvent {
636            event_time_us: 1_700_000_000_000_000,
637            book_update_id: 123,
638            price_exponent: -2,
639            qty_exponent: -4,
640            bids: vec![],
641            asks: vec![],
642            symbol: Ustr::from("ETHUSDT"),
643        };
644
645        let deltas = parse_depth_snapshot(&event, &instrument, UnixNanos::from(2));
646
647        assert_eq!(deltas.deltas.len(), 1);
648        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
649        assert_eq!(deltas.deltas[0].sequence, 123);
650        assert_eq!(
651            deltas.deltas[0].flags,
652            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
653        );
654    }
655
656    #[rstest]
657    fn test_parse_depth_diff() {
658        let instrument = sample_instrument();
659        let ts_init = UnixNanos::from(1_800_000_000_000_000_000u64);
660        let event = DepthDiffStreamEvent {
661            event_time_us: 1_700_000_000_000_000,
662            first_book_update_id: 100,
663            last_book_update_id: 101,
664            price_exponent: -2,
665            qty_exponent: -4,
666            bids: vec![
667                PriceLevel {
668                    price_mantissa: 12_345,
669                    qty_mantissa: 25_000,
670                },
671                PriceLevel {
672                    price_mantissa: 12_340,
673                    qty_mantissa: 0,
674                },
675            ],
676            asks: vec![PriceLevel {
677                price_mantissa: 12_350,
678                qty_mantissa: 30_000,
679            }],
680            symbol: Ustr::from("ETHUSDT"),
681        };
682
683        let deltas = parse_depth_diff(&event, &instrument, ts_init).unwrap();
684
685        assert_eq!(deltas.instrument_id, instrument.id());
686        assert_eq!(deltas.deltas.len(), 3);
687        assert_eq!(deltas.deltas[0].action, BookAction::Update);
688        assert_eq!(deltas.deltas[0].order.side, OrderSide::Buy.into());
689        assert_eq!(deltas.deltas[1].action, BookAction::Delete);
690        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy.into());
691        assert_eq!(deltas.deltas[2].action, BookAction::Update);
692        assert_eq!(deltas.deltas[2].order.side, OrderSide::Sell.into());
693        assert_eq!(deltas.deltas[2].flags, RecordFlag::F_LAST as u8);
694        assert_eq!(deltas.deltas[0].sequence, 101);
695        assert_eq!(deltas.deltas[2].sequence, 101);
696        assert_eq!(
697            deltas.ts_event,
698            UnixNanos::from(1_700_000_000_000_000_000u64)
699        );
700        assert_eq!(deltas.ts_init, ts_init);
701        assert!(deltas.deltas.iter().all(|delta| delta.ts_init == ts_init));
702    }
703
704    #[rstest]
705    fn test_parse_depth_diff_empty_returns_none() {
706        let instrument = sample_instrument();
707        let event = DepthDiffStreamEvent {
708            event_time_us: 1_700_000_000_000_000,
709            first_book_update_id: 100,
710            last_book_update_id: 101,
711            price_exponent: -2,
712            qty_exponent: -4,
713            bids: vec![],
714            asks: vec![],
715            symbol: Ustr::from("ETHUSDT"),
716        };
717
718        let deltas = parse_depth_diff(&event, &instrument, UnixNanos::from(2));
719
720        assert!(deltas.is_none());
721    }
722}
nautilus_binance/spot/websocket/streams/parse.rs

nautilus_binance/spot/websocket/streams/
parse.rs
