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.
80///
81/// # Errors
82///
83/// Returns an error if:
84/// - A price is out of range or a reserved sentinel value.
85/// - A quantity is negative, out of range, or zero at the instrument size precision.
86pub fn parse_trades_event(
87    event: &TradesStreamEvent,
88    instrument: &InstrumentAny,
89    ts_init: UnixNanos,
90) -> anyhow::Result<Vec<Data>> {
91    let instrument_id = instrument.id();
92    let price_precision = instrument.price_precision();
93    let size_precision = instrument.size_precision();
94
95    let mut trades = Vec::with_capacity(event.trades.len());
96    for t in &event.trades {
97        let price = parse_price(t.price_mantissa, event.price_exponent, price_precision)?;
98        let size = parse_quantity(t.qty_mantissa, event.qty_exponent, size_precision)?;
99        let ts_event = parse_micros_or_init(
100            event.transact_time_us,
101            "Spot SBE stream transaction time",
102            ts_init,
103        );
104
105        let trade = TradeTick::new_checked(
106            instrument_id,
107            price,
108            size,
109            if t.is_buyer_maker {
110                AggressorSide::Sell
111            } else {
112                AggressorSide::Buy
113            },
114            TradeId::new(t.id.to_string()),
115            ts_event,
116            ts_init,
117        )?;
118
119        trades.push(Data::from(trade));
120    }
121
122    Ok(trades)
123}
124
125/// Parses a best bid/ask event into a `QuoteTick` using the supplied adapter
126/// initialization timestamp.
127///
128/// # Errors
129///
130/// Returns an error if:
131/// - A price is out of range or a reserved sentinel value.
132/// - A quantity is negative or out of range.
133pub fn parse_bbo_event(
134    event: &BestBidAskStreamEvent,
135    instrument: &InstrumentAny,
136    ts_init: UnixNanos,
137) -> anyhow::Result<QuoteTick> {
138    let instrument_id = instrument.id();
139    let price_precision = instrument.price_precision();
140    let size_precision = instrument.size_precision();
141
142    let bid_price = parse_price(
143        event.bid_price_mantissa,
144        event.price_exponent,
145        price_precision,
146    )?;
147    let bid_size = parse_quantity(event.bid_qty_mantissa, event.qty_exponent, size_precision)?;
148    let ask_price = parse_price(
149        event.ask_price_mantissa,
150        event.price_exponent,
151        price_precision,
152    )?;
153    let ask_size = parse_quantity(event.ask_qty_mantissa, event.qty_exponent, size_precision)?;
154    let ts_event = parse_micros_or_init(event.event_time_us, "Spot SBE BBO event time", ts_init);
155
156    Ok(QuoteTick::new(
157        instrument_id,
158        bid_price,
159        ask_price,
160        bid_size,
161        ask_size,
162        ts_event,
163        ts_init,
164    ))
165}
166
167/// Parses a depth snapshot event into `OrderBookDeltas` using the supplied
168/// adapter initialization timestamp for the aggregate and every inner delta.
169///
170/// A snapshot without levels clears the book.
171///
172/// # Errors
173///
174/// Returns an error if:
175/// - A price is out of range or a reserved sentinel value.
176/// - A quantity is negative, out of range, or zero at the instrument size precision.
177pub fn parse_depth_snapshot(
178    event: &DepthSnapshotStreamEvent,
179    instrument: &InstrumentAny,
180    ts_init: UnixNanos,
181) -> anyhow::Result<OrderBookDeltas> {
182    let instrument_id = instrument.id();
183    let price_precision = instrument.price_precision();
184    let size_precision = instrument.size_precision();
185    let ts_event = parse_micros_or_init(
186        event.event_time_us,
187        "Spot SBE depth snapshot event time",
188        ts_init,
189    );
190    let sequence = event.book_update_id as u64;
191
192    let mut deltas = Vec::with_capacity(event.bids.len() + event.asks.len() + 1);
193
194    // Add clear delta first
195    deltas.push(OrderBookDelta::clear(
196        instrument_id,
197        sequence,
198        ts_event,
199        ts_init,
200    ));
201
202    for (side, levels) in [
203        (OrderSide::Buy, &event.bids),
204        (OrderSide::Sell, &event.asks),
205    ] {
206        for level in levels {
207            let price = parse_price(level.price_mantissa, event.price_exponent, price_precision)?;
208            let size = parse_quantity(level.qty_mantissa, event.qty_exponent, size_precision)?;
209            let order = BookOrder::new(side, price, size, 0);
210
211            deltas.push(OrderBookDelta::new_checked(
212                instrument_id,
213                BookAction::Add,
214                order,
215                RecordFlag::F_SNAPSHOT as u8,
216                sequence,
217                ts_event,
218                ts_init,
219            )?);
220        }
221    }
222
223    if let Some(last) = deltas.last_mut() {
224        last.flags |= RecordFlag::F_LAST as u8;
225    }
226
227    Ok(OrderBookDeltas::new(instrument_id, deltas))
228}
229
230/// Parses a depth diff event into `OrderBookDeltas` using the supplied adapter
231/// initialization timestamp for the aggregate and every inner delta.
232///
233/// Returns `Ok(None)` if the diff contains no updates.
234///
235/// # Errors
236///
237/// Returns an error if:
238/// - A price is out of range or a reserved sentinel value.
239/// - A quantity is negative or out of range.
240/// - A non-zero quantity rounds to zero at the instrument size precision.
241pub fn parse_depth_diff(
242    event: &DepthDiffStreamEvent,
243    instrument: &InstrumentAny,
244    ts_init: UnixNanos,
245) -> anyhow::Result<Option<OrderBookDeltas>> {
246    let instrument_id = instrument.id();
247    let price_precision = instrument.price_precision();
248    let size_precision = instrument.size_precision();
249    let ts_event = parse_micros_or_init(
250        event.event_time_us,
251        "Spot SBE depth diff event time",
252        ts_init,
253    );
254    let sequence = event.last_book_update_id as u64;
255
256    let mut deltas = Vec::with_capacity(event.bids.len() + event.asks.len());
257
258    // Add bid updates
259    for (i, level) in event.bids.iter().enumerate() {
260        let price = parse_price(level.price_mantissa, event.price_exponent, price_precision)?;
261        let size = parse_quantity(level.qty_mantissa, event.qty_exponent, size_precision)?;
262
263        // Zero size means delete, otherwise update
264        let action = if level.qty_mantissa == 0 {
265            BookAction::Delete
266        } else {
267            BookAction::Update
268        };
269
270        let flags = if i == event.bids.len() - 1 && event.asks.is_empty() {
271            RecordFlag::F_LAST as u8
272        } else {
273            0
274        };
275
276        let order = BookOrder::new(OrderSide::Buy, price, size, 0);
277
278        deltas.push(OrderBookDelta::new_checked(
279            instrument_id,
280            action,
281            order,
282            flags,
283            sequence,
284            ts_event,
285            ts_init,
286        )?);
287    }
288
289    // Add ask updates
290    for (i, level) in event.asks.iter().enumerate() {
291        let price = parse_price(level.price_mantissa, event.price_exponent, price_precision)?;
292        let size = parse_quantity(level.qty_mantissa, event.qty_exponent, size_precision)?;
293
294        let action = if level.qty_mantissa == 0 {
295            BookAction::Delete
296        } else {
297            BookAction::Update
298        };
299
300        let flags = if i == event.asks.len() - 1 {
301            RecordFlag::F_LAST as u8
302        } else {
303            0
304        };
305
306        let order = BookOrder::new(OrderSide::Sell, price, size, 0);
307
308        deltas.push(OrderBookDelta::new_checked(
309            instrument_id,
310            action,
311            order,
312            flags,
313            sequence,
314            ts_event,
315            ts_init,
316        )?);
317    }
318
319    if deltas.is_empty() {
320        return Ok(None);
321    }
322
323    Ok(Some(OrderBookDeltas::new(instrument_id, deltas)))
324}
325
326fn parse_price(mantissa: i64, exponent: i8, precision: u8) -> anyhow::Result<Price> {
327    let price = Price::from_mantissa_exponent_checked(mantissa, exponent, precision)?;
328    anyhow::ensure!(
329        !price.is_error() && !price.is_undefined(),
330        "invalid SBE price sentinel"
331    );
332    Ok(price)
333}
334
335fn parse_quantity(mantissa: i64, exponent: i8, precision: u8) -> anyhow::Result<Quantity> {
336    let mantissa =
337        u64::try_from(mantissa).map_err(|_| anyhow::anyhow!("invalid negative SBE quantity"))?;
338    Ok(Quantity::from_mantissa_exponent_checked(
339        mantissa, exponent, precision,
340    )?)
341}
342
343#[cfg(test)]
344mod tests {
345    use rstest::rstest;
346    use ustr::Ustr;
347
348    use super::*;
349    use crate::{
350        common::parse::parse_spot_instrument_sbe,
351        spot::{
352            http::models::{
353                BinanceLotSizeFilterSbe, BinancePriceFilterSbe, BinanceSymbolFiltersSbe,
354                BinanceSymbolSbe,
355            },
356            sbe::stream::{PriceLevel, STREAM_SCHEMA_ID, Trade},
357        },
358    };
359
360    fn make_bbo_buffer() -> Vec<u8> {
361        let mut buf = vec![0u8; 70];
362
363        // Header
364        buf[0..2].copy_from_slice(&50u16.to_le_bytes()); // block_length
365        buf[2..4].copy_from_slice(&template_id::BEST_BID_ASK_STREAM_EVENT.to_le_bytes());
366        buf[4..6].copy_from_slice(&STREAM_SCHEMA_ID.to_le_bytes());
367        buf[6..8].copy_from_slice(&0u16.to_le_bytes()); // version
368
369        // Body
370        let body = &mut buf[8..];
371        body[0..8].copy_from_slice(&1000000i64.to_le_bytes()); // event_time_us
372        body[8..16].copy_from_slice(&12345i64.to_le_bytes()); // book_update_id
373        body[16] = (-2i8) as u8; // price_exponent
374        body[17] = (-8i8) as u8; // qty_exponent
375        body[18..26].copy_from_slice(&4200000i64.to_le_bytes()); // bid_price
376        body[26..34].copy_from_slice(&100000000i64.to_le_bytes()); // bid_qty
377        body[34..42].copy_from_slice(&4200100i64.to_le_bytes()); // ask_price
378        body[42..50].copy_from_slice(&200000000i64.to_le_bytes()); // ask_qty
379
380        // Symbol: "BTCUSDT" (7 bytes)
381        body[50] = 7;
382        body[51..58].copy_from_slice(b"BTCUSDT");
383
384        buf
385    }
386
387    fn sample_instrument() -> InstrumentAny {
388        let symbol = BinanceSymbolSbe {
389            symbol: "ETHUSDT".to_string(),
390            base_asset: "ETH".to_string(),
391            quote_asset: "USDT".to_string(),
392            base_asset_precision: 8,
393            quote_asset_precision: 8,
394            status: 0,
395            order_types: 0,
396            iceberg_allowed: true,
397            oco_allowed: true,
398            oto_allowed: false,
399            quote_order_qty_market_allowed: true,
400            allow_trailing_stop: true,
401            cancel_replace_allowed: true,
402            amend_allowed: true,
403            is_spot_trading_allowed: true,
404            is_margin_trading_allowed: false,
405            filters: BinanceSymbolFiltersSbe {
406                notional_filters: Vec::new(),
407                price_filter: Some(BinancePriceFilterSbe {
408                    price_exponent: -8,
409                    min_price: 1_000_000,
410                    max_price: 100_000_000_000_000,
411                    tick_size: 1_000_000,
412                }),
413                lot_size_filter: Some(BinanceLotSizeFilterSbe {
414                    qty_exponent: -8,
415                    min_qty: 10_000,
416                    max_qty: 900_000_000_000,
417                    step_size: 10_000,
418                }),
419            },
420            permissions: vec![vec!["SPOT".to_string()]],
421        };
422
423        let ts = UnixNanos::from(1_700_000_000_000_000_000u64);
424        parse_spot_instrument_sbe(&symbol, ts, ts).unwrap()
425    }
426
427    #[rstest]
428    fn test_decode_empty_buffer() {
429        let err = decode_market_data(&[]).unwrap_err();
430        assert!(matches!(err, StreamDecodeError::BufferTooShort { .. }));
431    }
432
433    #[rstest]
434    fn test_decode_short_buffer() {
435        let buf = [0u8; 5];
436        let err = decode_market_data(&buf).unwrap_err();
437        assert!(matches!(err, StreamDecodeError::BufferTooShort { .. }));
438    }
439
440    #[rstest]
441    fn test_decode_wrong_schema() {
442        let mut buf = [0u8; 100];
443        buf[0..2].copy_from_slice(&50u16.to_le_bytes()); // block_length
444        buf[2..4].copy_from_slice(&template_id::BEST_BID_ASK_STREAM_EVENT.to_le_bytes());
445        buf[4..6].copy_from_slice(&99u16.to_le_bytes()); // Wrong schema
446        buf[6..8].copy_from_slice(&0u16.to_le_bytes()); // version
447
448        let err = decode_market_data(&buf).unwrap_err();
449        assert!(matches!(err, StreamDecodeError::SchemaMismatch { .. }));
450    }
451
452    #[rstest]
453    fn test_decode_unknown_template() {
454        let mut buf = [0u8; 100];
455        buf[0..2].copy_from_slice(&50u16.to_le_bytes()); // block_length
456        buf[2..4].copy_from_slice(&9999u16.to_le_bytes()); // Unknown template
457        buf[4..6].copy_from_slice(&STREAM_SCHEMA_ID.to_le_bytes());
458        buf[6..8].copy_from_slice(&0u16.to_le_bytes()); // version
459
460        let err = decode_market_data(&buf).unwrap_err();
461        assert!(matches!(err, StreamDecodeError::UnknownTemplateId(9999)));
462    }
463
464    #[rstest]
465    fn test_decode_valid_best_bid_ask() {
466        let buf = make_bbo_buffer();
467        let msg = decode_market_data(&buf).unwrap();
468
469        match msg {
470            MarketDataMessage::BestBidAsk(event) => {
471                assert_eq!(event.event_time_us, 1_000_000);
472                assert_eq!(event.symbol, Ustr::from("BTCUSDT"));
473            }
474            _ => panic!("Expected BestBidAsk"),
475        }
476    }
477
478    #[rstest]
479    fn test_parse_trades_event() {
480        let instrument = sample_instrument();
481        let ts_init = UnixNanos::from(1_800_000_000_000_000_000u64);
482        let event = TradesStreamEvent {
483            event_time_us: 1_700_000_000_000_000,
484            transact_time_us: 1_700_000_000_100_000,
485            price_exponent: -2,
486            qty_exponent: -4,
487            trades: vec![
488                Trade {
489                    id: 1,
490                    price_mantissa: 12_345,
491                    qty_mantissa: 25_000,
492                    is_buyer_maker: false,
493                },
494                Trade {
495                    id: 2,
496                    price_mantissa: 12_340,
497                    qty_mantissa: 10_000,
498                    is_buyer_maker: true,
499                },
500            ],
501            symbol: Ustr::from("ETHUSDT"),
502        };
503
504        let data = parse_trades_event(&event, &instrument, ts_init).unwrap();
505
506        assert_eq!(data.len(), 2);
507        match &data[0] {
508            Data::Trade(trade) => {
509                assert_eq!(trade.instrument_id, instrument.id());
510                assert_eq!(trade.price, Price::from("123.45"));
511                assert_eq!(trade.size, Quantity::from("2.5000"));
512                assert_eq!(trade.aggressor_side, AggressorSide::Buy);
513                assert_eq!(trade.trade_id, TradeId::new("1"));
514                assert_eq!(
515                    trade.ts_event,
516                    UnixNanos::from(1_700_000_000_100_000_000u64)
517                );
518                assert_eq!(trade.ts_init, ts_init);
519            }
520            other => panic!("Expected trade data, was {other:?}"),
521        }
522
523        match &data[1] {
524            Data::Trade(trade) => {
525                assert_eq!(trade.instrument_id, instrument.id());
526                assert_eq!(trade.price, Price::from("123.40"));
527                assert_eq!(trade.size, Quantity::from("1.0000"));
528                assert_eq!(trade.aggressor_side, AggressorSide::Sell);
529                assert_eq!(trade.trade_id, TradeId::new("2"));
530                assert_eq!(
531                    trade.ts_event,
532                    UnixNanos::from(1_700_000_000_100_000_000u64)
533                );
534                assert_eq!(trade.ts_init, ts_init);
535            }
536            other => panic!("Expected trade data, was {other:?}"),
537        }
538    }
539
540    #[rstest]
541    fn test_parse_bbo_event() {
542        let instrument = sample_instrument();
543        let ts_init = UnixNanos::from(1_800_000_000_000_000_000u64);
544        let event = BestBidAskStreamEvent {
545            event_time_us: 1_700_000_000_000_000,
546            book_update_id: 123,
547            price_exponent: -2,
548            qty_exponent: -4,
549            bid_price_mantissa: 12_345,
550            bid_qty_mantissa: 25_000,
551            ask_price_mantissa: 12_350,
552            ask_qty_mantissa: 30_000,
553            symbol: Ustr::from("ETHUSDT"),
554        };
555
556        let quote = parse_bbo_event(&event, &instrument, ts_init).unwrap();
557
558        assert_eq!(quote.instrument_id, instrument.id());
559        assert_eq!(quote.bid_price, Price::from("123.45"));
560        assert_eq!(quote.ask_price, Price::from("123.50"));
561        assert_eq!(quote.bid_size, Quantity::from("2.5000"));
562        assert_eq!(quote.ask_size, Quantity::from("3.0000"));
563        assert_eq!(
564            quote.ts_event,
565            UnixNanos::from(1_700_000_000_000_000_000u64)
566        );
567        assert_eq!(quote.ts_init, ts_init);
568    }
569
570    #[rstest]
571    #[case::negative(-1)]
572    #[case::overflow(i64::MAX)]
573    fn test_parse_bbo_event_falls_back_for_invalid_timestamp(#[case] event_time_us: i64) {
574        let instrument = sample_instrument();
575        let event = BestBidAskStreamEvent {
576            event_time_us,
577            book_update_id: 123,
578            price_exponent: -2,
579            qty_exponent: -4,
580            bid_price_mantissa: 12_345,
581            bid_qty_mantissa: 25_000,
582            ask_price_mantissa: 12_350,
583            ask_qty_mantissa: 30_000,
584            symbol: Ustr::from("ETHUSDT"),
585        };
586
587        let ts_init = UnixNanos::from(1);
588        let quote = parse_bbo_event(&event, &instrument, ts_init).unwrap();
589
590        assert_eq!(quote.ts_event, ts_init);
591        assert_eq!(quote.ts_init, ts_init);
592    }
593
594    #[rstest]
595    fn test_parse_depth_snapshot() {
596        let instrument = sample_instrument();
597        let ts_init = UnixNanos::from(1_800_000_000_000_000_000u64);
598        let event = DepthSnapshotStreamEvent {
599            event_time_us: 1_700_000_000_000_000,
600            book_update_id: 123,
601            price_exponent: -2,
602            qty_exponent: -4,
603            bids: vec![PriceLevel {
604                price_mantissa: 12_345,
605                qty_mantissa: 25_000,
606            }],
607            asks: vec![PriceLevel {
608                price_mantissa: 12_350,
609                qty_mantissa: 30_000,
610            }],
611            symbol: Ustr::from("ETHUSDT"),
612        };
613
614        let deltas = parse_depth_snapshot(&event, &instrument, ts_init).unwrap();
615
616        assert_eq!(deltas.instrument_id, instrument.id());
617        assert_eq!(deltas.deltas.len(), 3);
618        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
619        assert_eq!(deltas.deltas[1].action, BookAction::Add);
620        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy.into());
621        assert_eq!(deltas.deltas[1].order.price, Price::from("123.45"));
622        assert_eq!(deltas.deltas[1].order.size, Quantity::from("2.5000"));
623        assert_eq!(deltas.deltas[2].action, BookAction::Add);
624        assert_eq!(deltas.deltas[2].order.side, OrderSide::Sell.into());
625        assert_eq!(deltas.deltas[2].order.price, Price::from("123.50"));
626        assert_eq!(deltas.deltas[2].order.size, Quantity::from("3.0000"));
627        assert_eq!(deltas.deltas[1].flags, RecordFlag::F_SNAPSHOT as u8);
628        assert_eq!(
629            deltas.deltas[2].flags,
630            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
631        );
632        assert_eq!(deltas.deltas[0].sequence, 123);
633        assert_eq!(deltas.deltas[2].sequence, 123);
634        assert_eq!(
635            deltas.ts_event,
636            UnixNanos::from(1_700_000_000_000_000_000u64)
637        );
638        assert_eq!(deltas.ts_init, ts_init);
639        assert!(deltas.deltas.iter().all(|delta| delta.ts_init == ts_init));
640    }
641
642    #[rstest]
643    fn test_parse_depth_snapshot_empty_clears_book() {
644        let instrument = sample_instrument();
645        let event = DepthSnapshotStreamEvent {
646            event_time_us: 1_700_000_000_000_000,
647            book_update_id: 123,
648            price_exponent: -2,
649            qty_exponent: -4,
650            bids: vec![],
651            asks: vec![],
652            symbol: Ustr::from("ETHUSDT"),
653        };
654
655        let deltas = parse_depth_snapshot(&event, &instrument, UnixNanos::from(2)).unwrap();
656
657        assert_eq!(deltas.deltas.len(), 1);
658        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
659        assert_eq!(deltas.deltas[0].sequence, 123);
660        assert_eq!(
661            deltas.deltas[0].flags,
662            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
663        );
664    }
665
666    #[rstest]
667    fn test_parse_depth_diff() {
668        let instrument = sample_instrument();
669        let ts_init = UnixNanos::from(1_800_000_000_000_000_000u64);
670        let event = DepthDiffStreamEvent {
671            event_time_us: 1_700_000_000_000_000,
672            first_book_update_id: 100,
673            last_book_update_id: 101,
674            price_exponent: -2,
675            qty_exponent: -4,
676            bids: vec![
677                PriceLevel {
678                    price_mantissa: 12_345,
679                    qty_mantissa: 25_000,
680                },
681                PriceLevel {
682                    price_mantissa: 12_340,
683                    qty_mantissa: 0,
684                },
685            ],
686            asks: vec![PriceLevel {
687                price_mantissa: 12_350,
688                qty_mantissa: 30_000,
689            }],
690            symbol: Ustr::from("ETHUSDT"),
691        };
692
693        let deltas = parse_depth_diff(&event, &instrument, ts_init)
694            .unwrap()
695            .unwrap();
696
697        assert_eq!(deltas.instrument_id, instrument.id());
698        assert_eq!(deltas.deltas.len(), 3);
699        assert_eq!(deltas.deltas[0].order.price, Price::from("123.45"));
700        assert_eq!(deltas.deltas[0].order.size, Quantity::from("2.5000"));
701        assert_eq!(deltas.deltas[1].order.price, Price::from("123.40"));
702        assert_eq!(deltas.deltas[1].order.size, Quantity::from("0.0000"));
703        assert_eq!(deltas.deltas[2].order.price, Price::from("123.50"));
704        assert_eq!(deltas.deltas[2].order.size, Quantity::from("3.0000"));
705        assert_eq!(deltas.deltas[0].action, BookAction::Update);
706        assert_eq!(deltas.deltas[0].order.side, OrderSide::Buy.into());
707        assert_eq!(deltas.deltas[1].action, BookAction::Delete);
708        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy.into());
709        assert_eq!(deltas.deltas[2].action, BookAction::Update);
710        assert_eq!(deltas.deltas[2].order.side, OrderSide::Sell.into());
711        assert_eq!(deltas.deltas[2].flags, RecordFlag::F_LAST as u8);
712        assert_eq!(deltas.deltas[0].flags, 0);
713        assert_eq!(deltas.deltas[1].flags, 0);
714
715        for delta in &deltas.deltas {
716            assert_eq!(delta.instrument_id, instrument.id());
717            assert_eq!(delta.order.order_id, 0);
718            assert_eq!(delta.sequence, 101);
719            assert_eq!(
720                delta.ts_event,
721                UnixNanos::from(1_700_000_000_000_000_000u64)
722            );
723            assert_eq!(delta.ts_init, ts_init);
724        }
725
726        assert_eq!(
727            deltas.ts_event,
728            UnixNanos::from(1_700_000_000_000_000_000u64)
729        );
730        assert_eq!(deltas.ts_init, ts_init);
731        assert!(deltas.deltas.iter().all(|delta| delta.ts_init == ts_init));
732    }
733
734    #[rstest]
735    fn test_parse_depth_diff_empty_returns_none() {
736        let instrument = sample_instrument();
737        let event = DepthDiffStreamEvent {
738            event_time_us: 1_700_000_000_000_000,
739            first_book_update_id: 100,
740            last_book_update_id: 101,
741            price_exponent: -2,
742            qty_exponent: -4,
743            bids: vec![],
744            asks: vec![],
745            symbol: Ustr::from("ETHUSDT"),
746        };
747
748        let deltas = parse_depth_diff(&event, &instrument, UnixNanos::from(2)).unwrap();
749
750        assert!(deltas.is_none());
751    }
752
753    #[rstest]
754    #[case::price_exponent(1, 1, i8::MAX, 0)]
755    #[case::qty_exponent(1, 1, 0, i8::MAX)]
756    #[case::price_above_range(i64::MAX, 1, 0, 0)]
757    #[case::price_below_range(i64::MIN, 1, 0, 0)]
758    #[case::qty_above_range(1, i64::MAX, 0, 0)]
759    #[case::negative_qty(1, -1, 0, 0)]
760    #[case::minimum_qty(1, i64::MIN, 0, 0)]
761    #[case::negative_qty_rounds_to_zero(1, -1, 0, i8::MIN)]
762    fn test_parse_events_reject_invalid_numbers(
763        #[case] price_mantissa: i64,
764        #[case] qty_mantissa: i64,
765        #[case] price_exponent: i8,
766        #[case] qty_exponent: i8,
767        #[values(false, true)] ask: bool,
768    ) {
769        let instrument = sample_instrument();
770        let ts_init = UnixNanos::from(2);
771
772        let level = PriceLevel {
773            price_mantissa,
774            qty_mantissa,
775        };
776
777        let valid = PriceLevel {
778            price_mantissa: 12_345,
779            qty_mantissa: 25_000,
780        };
781
782        let (bid, ask) = if ask { (valid, level) } else { (level, valid) };
783
784        let trades = TradesStreamEvent {
785            event_time_us: 1,
786            transact_time_us: 1,
787            price_exponent,
788            qty_exponent,
789            trades: vec![
790                Trade {
791                    id: 1,
792                    price_mantissa: 0,
793                    qty_mantissa: 1,
794                    is_buyer_maker: false,
795                },
796                Trade {
797                    id: 2,
798                    price_mantissa,
799                    qty_mantissa,
800                    is_buyer_maker: true,
801                },
802            ],
803            symbol: Ustr::from("ETHUSDT"),
804        };
805
806        let bbo = BestBidAskStreamEvent {
807            event_time_us: 1,
808            book_update_id: 123,
809            price_exponent,
810            qty_exponent,
811            bid_price_mantissa: bid.price_mantissa,
812            bid_qty_mantissa: bid.qty_mantissa,
813            ask_price_mantissa: ask.price_mantissa,
814            ask_qty_mantissa: ask.qty_mantissa,
815            symbol: trades.symbol,
816        };
817
818        let snapshot = DepthSnapshotStreamEvent {
819            event_time_us: 1,
820            book_update_id: 123,
821            price_exponent,
822            qty_exponent,
823            bids: vec![bid],
824            asks: vec![ask],
825            symbol: trades.symbol,
826        };
827
828        let diff = DepthDiffStreamEvent {
829            event_time_us: 1,
830            first_book_update_id: 123,
831            last_book_update_id: 124,
832            price_exponent,
833            qty_exponent,
834            bids: snapshot.bids.clone(),
835            asks: snapshot.asks.clone(),
836            symbol: trades.symbol,
837        };
838
839        assert!(parse_trades_event(&trades, &instrument, ts_init).is_err());
840        assert!(parse_bbo_event(&bbo, &instrument, ts_init).is_err());
841        assert!(parse_depth_snapshot(&snapshot, &instrument, ts_init).is_err());
842        assert!(parse_depth_diff(&diff, &instrument, ts_init).is_err());
843    }
844
845    #[rstest]
846    #[case::zero(0, -4)]
847    #[case::rounded_to_zero(1, i8::MIN)]
848    fn test_parse_events_handle_zero_size(#[case] qty_mantissa: i64, #[case] qty_exponent: i8) {
849        let instrument = sample_instrument();
850        let ts_init = UnixNanos::from(2);
851
852        let trades = TradesStreamEvent {
853            event_time_us: 1,
854            transact_time_us: 1,
855            price_exponent: -2,
856            qty_exponent,
857            trades: vec![Trade {
858                id: 1,
859                price_mantissa: 12_345,
860                qty_mantissa,
861                is_buyer_maker: false,
862            }],
863            symbol: Ustr::from("ETHUSDT"),
864        };
865
866        let snapshot = DepthSnapshotStreamEvent {
867            event_time_us: 1,
868            book_update_id: 123,
869            price_exponent: -2,
870            qty_exponent,
871            bids: vec![PriceLevel {
872                price_mantissa: 12_345,
873                qty_mantissa,
874            }],
875            asks: vec![],
876            symbol: trades.symbol,
877        };
878
879        let diff = DepthDiffStreamEvent {
880            event_time_us: 1,
881            first_book_update_id: 123,
882            last_book_update_id: 124,
883            price_exponent: -2,
884            qty_exponent,
885            bids: snapshot.bids.clone(),
886            asks: vec![],
887            symbol: trades.symbol,
888        };
889
890        let bbo = BestBidAskStreamEvent {
891            event_time_us: 1,
892            book_update_id: 123,
893            price_exponent: -2,
894            qty_exponent,
895            bid_price_mantissa: 12_345,
896            bid_qty_mantissa: qty_mantissa,
897            ask_price_mantissa: 12_350,
898            ask_qty_mantissa: qty_mantissa,
899            symbol: trades.symbol,
900        };
901
902        assert!(parse_trades_event(&trades, &instrument, ts_init).is_err());
903        assert!(parse_depth_snapshot(&snapshot, &instrument, ts_init).is_err());
904        let result = parse_depth_diff(&diff, &instrument, ts_init);
905        if qty_mantissa == 0 {
906            let deltas = result.unwrap().unwrap();
907            assert_eq!(deltas.deltas[0].action, BookAction::Delete);
908            assert_eq!(
909                deltas.deltas[0].order.size,
910                Quantity::zero(instrument.size_precision())
911            );
912        } else {
913            assert!(result.is_err());
914        }
915
916        let quote = parse_bbo_event(&bbo, &instrument, ts_init).unwrap();
917        assert_eq!(quote.bid_size, Quantity::zero(instrument.size_precision()));
918        assert_eq!(quote.ask_size, Quantity::zero(instrument.size_precision()));
919    }
920
921    #[cfg(not(feature = "high-precision"))]
922    #[rstest]
923    fn test_parse_price_rejects_sentinel() {
924        assert!(parse_price(i64::MIN, -9, 9).is_err());
925    }
926}
nautilus_binance/spot/websocket/streams/parse.rs

nautilus_binance/spot/websocket/streams/
parse.rs
