nautilus_binance/spot/websocket/public_json/
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 public JSON WebSocket messages.
17
18use std::str::FromStr;
19
20use anyhow::Context;
21use nautilus_core::nanos::UnixNanos;
22use nautilus_model::{
23    data::{
24        Bar, BarSpecification, BarType, BookOrder, OrderBookDelta, OrderBookDeltas, QuoteTick,
25        TradeTick,
26    },
27    enums::{
28        AggregationSource, AggressorSide, BarAggregation, BookAction, OrderSide, PriceType,
29        RecordFlag,
30    },
31    identifiers::TradeId,
32    instruments::{Instrument, InstrumentAny},
33    types::{Price, Quantity},
34};
35use rust_decimal::Decimal;
36
37use super::messages::{
38    BinanceSpotBookTickerMsg, BinanceSpotDepthDiffMsg, BinanceSpotKlineMsg,
39    BinanceSpotPartialDepthMsg, BinanceSpotTradeMsg,
40};
41use crate::common::{
42    enums::BinanceKlineInterval,
43    parse::{parse_price_at_precision, parse_quantity_at_precision},
44};
45
46fn parse_positive_price(raw: &str, precision: u8, field: &str) -> anyhow::Result<Price> {
47    parse_price_at_precision(raw, precision)
48        .ok_or_else(|| anyhow::anyhow!("invalid {field} `{raw}`"))
49}
50
51fn parse_positive_quantity(raw: &str, precision: u8, field: &str) -> anyhow::Result<Quantity> {
52    parse_quantity_at_precision(raw, precision)
53        .ok_or_else(|| anyhow::anyhow!("invalid {field} `{raw}`"))
54}
55
56fn parse_non_negative_quantity(raw: &str, precision: u8, field: &str) -> anyhow::Result<Quantity> {
57    let decimal = Decimal::from_str(raw).with_context(|| format!("invalid {field} `{raw}`"))?;
58    if decimal.is_sign_negative() {
59        anyhow::bail!("invalid {field} `{raw}`");
60    }
61
62    Quantity::from_decimal_dp(decimal, precision)
63        .map_err(|e| anyhow::anyhow!("invalid {field} `{raw}`: {e}"))
64}
65
66/// Parses a trade message into a `TradeTick`.
67///
68/// # Errors
69///
70/// Returns an error if price or quantity fields cannot be parsed.
71pub fn parse_trade(
72    msg: &BinanceSpotTradeMsg,
73    instrument: &InstrumentAny,
74    ts_init: UnixNanos,
75) -> anyhow::Result<TradeTick> {
76    let instrument_id = instrument.id();
77    let price_precision = instrument.price_precision();
78    let size_precision = instrument.size_precision();
79
80    let price = parse_positive_price(&msg.price, price_precision, "trade price")?;
81    let size = parse_positive_quantity(&msg.quantity, size_precision, "trade quantity")?;
82
83    let aggressor_side = if msg.is_buyer_maker {
84        AggressorSide::Seller
85    } else {
86        AggressorSide::Buyer
87    };
88
89    let ts_event = UnixNanos::from_millis(msg.trade_time as u64);
90
91    Ok(TradeTick::new(
92        instrument_id,
93        price,
94        size,
95        aggressor_side,
96        TradeId::new(msg.trade_id.to_string()),
97        ts_event,
98        ts_init,
99    ))
100}
101
102/// Parses a book ticker message into a `QuoteTick`.
103///
104/// # Errors
105///
106/// Returns an error if price or quantity fields cannot be parsed.
107pub fn parse_book_ticker(
108    msg: &BinanceSpotBookTickerMsg,
109    instrument: &InstrumentAny,
110    ts_init: UnixNanos,
111) -> anyhow::Result<QuoteTick> {
112    let instrument_id = instrument.id();
113    let price_precision = instrument.price_precision();
114    let size_precision = instrument.size_precision();
115
116    let bid_price = parse_positive_price(&msg.best_bid_price, price_precision, "bid price")?;
117    // A side that empties reports a zero size, which is a valid quote state.
118    let bid_size = parse_non_negative_quantity(&msg.best_bid_qty, size_precision, "bid quantity")?;
119    let ask_price = parse_positive_price(&msg.best_ask_price, price_precision, "ask price")?;
120    let ask_size = parse_non_negative_quantity(&msg.best_ask_qty, size_precision, "ask quantity")?;
121
122    // Spot bookTicker payloads on public streams do not consistently include
123    // event timestamps; fall back to receive time when absent.
124    let ts_event = msg
125        .transaction_time
126        .or(msg.event_time)
127        .and_then(|ts| u64::try_from(ts).ok())
128        .map_or(ts_init, UnixNanos::from_millis);
129
130    Ok(QuoteTick::new(
131        instrument_id,
132        bid_price,
133        ask_price,
134        bid_size,
135        ask_size,
136        ts_event,
137        ts_init,
138    ))
139}
140
141/// Parses a partial depth snapshot message into `OrderBookDeltas`.
142///
143/// Returns `None` when there are no usable levels.
144pub fn parse_depth_snapshot(
145    msg: &BinanceSpotPartialDepthMsg,
146    instrument: &InstrumentAny,
147    ts_init: UnixNanos,
148) -> Option<OrderBookDeltas> {
149    let instrument_id = instrument.id();
150    let price_precision = instrument.price_precision();
151    let size_precision = instrument.size_precision();
152
153    let mut deltas = Vec::with_capacity(msg.bids.len() + msg.asks.len() + 1);
154    deltas.push(OrderBookDelta::clear(instrument_id, 0, ts_init, ts_init));
155
156    for level in &msg.bids {
157        let Some(price) = parse_price_at_precision(&level[0], price_precision) else {
158            continue;
159        };
160        let Some(size) = parse_quantity_at_precision(&level[1], size_precision) else {
161            continue;
162        };
163
164        deltas.push(OrderBookDelta::new(
165            instrument_id,
166            BookAction::Add,
167            BookOrder::new(OrderSide::Buy, price, size, 0),
168            0,
169            0,
170            ts_init,
171            ts_init,
172        ));
173    }
174
175    for level in &msg.asks {
176        let Some(price) = parse_price_at_precision(&level[0], price_precision) else {
177            continue;
178        };
179        let Some(size) = parse_quantity_at_precision(&level[1], size_precision) else {
180            continue;
181        };
182
183        deltas.push(OrderBookDelta::new(
184            instrument_id,
185            BookAction::Add,
186            BookOrder::new(OrderSide::Sell, price, size, 0),
187            0,
188            0,
189            ts_init,
190            ts_init,
191        ));
192    }
193
194    if deltas.len() <= 1 {
195        return None;
196    }
197
198    // Mark the final emitted delta as the snapshot terminator. Assigning F_LAST by
199    // source index would drop the terminator whenever the last level fails to parse
200    // and is skipped above.
201    if let Some(last) = deltas.last_mut() {
202        last.flags |= RecordFlag::F_LAST as u8;
203    }
204
205    Some(OrderBookDeltas::new(instrument_id, deltas))
206}
207
208/// Parses a depth diff message into `OrderBookDeltas`.
209///
210/// # Errors
211///
212/// Returns an error if any price or quantity update cannot be parsed.
213pub fn parse_depth_diff(
214    msg: &BinanceSpotDepthDiffMsg,
215    instrument: &InstrumentAny,
216    ts_init: UnixNanos,
217) -> anyhow::Result<Option<OrderBookDeltas>> {
218    let instrument_id = instrument.id();
219    let price_precision = instrument.price_precision();
220    let size_precision = instrument.size_precision();
221    let ts_event = UnixNanos::from_millis(msg.event_time as u64);
222    let sequence = msg.final_update_id;
223
224    let mut deltas = Vec::with_capacity(msg.bids.len() + msg.asks.len());
225
226    for (i, level) in msg.bids.iter().enumerate() {
227        let price = parse_positive_price(&level[0], price_precision, "bid price")?;
228        let size = parse_non_negative_quantity(&level[1], size_precision, "bid quantity")?;
229        let action = if size.is_zero() {
230            BookAction::Delete
231        } else {
232            BookAction::Update
233        };
234        let flags = if i == msg.bids.len() - 1 && msg.asks.is_empty() {
235            RecordFlag::F_LAST as u8
236        } else {
237            0
238        };
239
240        deltas.push(OrderBookDelta::new(
241            instrument_id,
242            action,
243            BookOrder::new(OrderSide::Buy, price, size, 0),
244            flags,
245            sequence,
246            ts_event,
247            ts_init,
248        ));
249    }
250
251    for (i, level) in msg.asks.iter().enumerate() {
252        let price = parse_positive_price(&level[0], price_precision, "ask price")?;
253        let size = parse_non_negative_quantity(&level[1], size_precision, "ask quantity")?;
254        let action = if size.is_zero() {
255            BookAction::Delete
256        } else {
257            BookAction::Update
258        };
259        let flags = if i == msg.asks.len() - 1 {
260            RecordFlag::F_LAST as u8
261        } else {
262            0
263        };
264
265        deltas.push(OrderBookDelta::new(
266            instrument_id,
267            action,
268            BookOrder::new(OrderSide::Sell, price, size, 0),
269            flags,
270            sequence,
271            ts_event,
272            ts_init,
273        ));
274    }
275
276    if deltas.is_empty() {
277        return Ok(None);
278    }
279
280    Ok(Some(OrderBookDeltas::new(instrument_id, deltas)))
281}
282
283fn interval_to_bar_spec(interval: BinanceKlineInterval) -> BarSpecification {
284    match interval {
285        BinanceKlineInterval::Second1 => {
286            BarSpecification::new(1, BarAggregation::Second, PriceType::Last)
287        }
288        BinanceKlineInterval::Minute1 => {
289            BarSpecification::new(1, BarAggregation::Minute, PriceType::Last)
290        }
291        BinanceKlineInterval::Minute3 => {
292            BarSpecification::new(3, BarAggregation::Minute, PriceType::Last)
293        }
294        BinanceKlineInterval::Minute5 => {
295            BarSpecification::new(5, BarAggregation::Minute, PriceType::Last)
296        }
297        BinanceKlineInterval::Minute15 => {
298            BarSpecification::new(15, BarAggregation::Minute, PriceType::Last)
299        }
300        BinanceKlineInterval::Minute30 => {
301            BarSpecification::new(30, BarAggregation::Minute, PriceType::Last)
302        }
303        BinanceKlineInterval::Hour1 => {
304            BarSpecification::new(1, BarAggregation::Hour, PriceType::Last)
305        }
306        BinanceKlineInterval::Hour2 => {
307            BarSpecification::new(2, BarAggregation::Hour, PriceType::Last)
308        }
309        BinanceKlineInterval::Hour4 => {
310            BarSpecification::new(4, BarAggregation::Hour, PriceType::Last)
311        }
312        BinanceKlineInterval::Hour6 => {
313            BarSpecification::new(6, BarAggregation::Hour, PriceType::Last)
314        }
315        BinanceKlineInterval::Hour8 => {
316            BarSpecification::new(8, BarAggregation::Hour, PriceType::Last)
317        }
318        BinanceKlineInterval::Hour12 => {
319            BarSpecification::new(12, BarAggregation::Hour, PriceType::Last)
320        }
321        BinanceKlineInterval::Day1 => {
322            BarSpecification::new(1, BarAggregation::Day, PriceType::Last)
323        }
324        BinanceKlineInterval::Day3 => {
325            BarSpecification::new(3, BarAggregation::Day, PriceType::Last)
326        }
327        BinanceKlineInterval::Week1 => {
328            BarSpecification::new(1, BarAggregation::Week, PriceType::Last)
329        }
330        BinanceKlineInterval::Month1 => {
331            BarSpecification::new(1, BarAggregation::Month, PriceType::Last)
332        }
333    }
334}
335
336/// Parses a kline message into a closed `Bar`.
337///
338/// Returns `None` if the kline is not closed yet.
339///
340/// # Errors
341///
342/// Returns an error if any OHLCV field cannot be parsed.
343pub fn parse_kline(
344    msg: &BinanceSpotKlineMsg,
345    instrument: &InstrumentAny,
346    ts_init: UnixNanos,
347) -> anyhow::Result<Option<Bar>> {
348    if !msg.kline.is_closed {
349        return Ok(None);
350    }
351
352    let instrument_id = instrument.id();
353    let price_precision = instrument.price_precision();
354    let size_precision = instrument.size_precision();
355
356    let spec = interval_to_bar_spec(msg.kline.interval);
357    let bar_type = BarType::new(instrument_id, spec, AggregationSource::External);
358
359    let open = parse_positive_price(&msg.kline.open, price_precision, "open price")?;
360    let high = parse_positive_price(&msg.kline.high, price_precision, "high price")?;
361    let low = parse_positive_price(&msg.kline.low, price_precision, "low price")?;
362    let close = parse_positive_price(&msg.kline.close, price_precision, "close price")?;
363    let volume = parse_non_negative_quantity(&msg.kline.volume, size_precision, "volume")?;
364
365    let ts_event = UnixNanos::from_millis(msg.kline.close_time as u64);
366
367    Ok(Some(Bar::new(
368        bar_type, open, high, low, close, volume, ts_event, ts_init,
369    )))
370}
371
372#[cfg(test)]
373mod tests {
374    use rstest::rstest;
375    use ustr::Ustr;
376
377    use super::*;
378    use crate::{
379        common::parse::parse_spot_instrument_sbe,
380        spot::http::models::{
381            BinanceLotSizeFilterSbe, BinancePriceFilterSbe, BinanceSymbolFiltersSbe,
382            BinanceSymbolSbe,
383        },
384    };
385
386    fn sample_instrument() -> InstrumentAny {
387        let symbol = BinanceSymbolSbe {
388            symbol: "ETHUSDT".to_string(),
389            base_asset: "ETH".to_string(),
390            quote_asset: "USDT".to_string(),
391            base_asset_precision: 8,
392            quote_asset_precision: 8,
393            status: 0,
394            order_types: 0,
395            iceberg_allowed: true,
396            oco_allowed: true,
397            oto_allowed: false,
398            quote_order_qty_market_allowed: true,
399            allow_trailing_stop: true,
400            cancel_replace_allowed: true,
401            amend_allowed: true,
402            is_spot_trading_allowed: true,
403            is_margin_trading_allowed: false,
404            filters: BinanceSymbolFiltersSbe {
405                price_filter: Some(BinancePriceFilterSbe {
406                    price_exponent: -8,
407                    min_price: 1,
408                    max_price: 100_000_000_000_000,
409                    tick_size: 1,
410                }),
411                lot_size_filter: Some(BinanceLotSizeFilterSbe {
412                    qty_exponent: -8,
413                    min_qty: 1,
414                    max_qty: 900_000_000_000,
415                    step_size: 1,
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_parse_trade_preserves_decimal_precision() {
427        let instrument = sample_instrument();
428        let msg = BinanceSpotTradeMsg {
429            event_type: "trade".to_string(),
430            event_time: 1_700_000_000_000,
431            symbol: Ustr::from("ETHUSDT"),
432            trade_id: 42,
433            price: "123.45678901".to_string(),
434            quantity: "0.10000001".to_string(),
435            trade_time: 1_700_000_000_001,
436            is_buyer_maker: false,
437        };
438
439        let tick = parse_trade(&msg, &instrument, UnixNanos::from(1)).unwrap();
440        assert_eq!(
441            tick.price.as_decimal(),
442            Decimal::from_str("123.45678901").unwrap()
443        );
444        assert_eq!(
445            tick.size.as_decimal(),
446            Decimal::from_str("0.10000001").unwrap()
447        );
448    }
449
450    #[rstest]
451    fn test_parse_book_ticker_preserves_decimal_precision() {
452        let instrument = sample_instrument();
453        let msg = BinanceSpotBookTickerMsg {
454            event_type: None,
455            event_time: None,
456            symbol: Ustr::from("ETHUSDT"),
457            book_update_id: 100,
458            best_bid_price: "123.45678901".to_string(),
459            best_bid_qty: "1.23000000".to_string(),
460            best_ask_price: "123.45678909".to_string(),
461            best_ask_qty: "4.56000000".to_string(),
462            transaction_time: Some(1_700_000_000_002),
463        };
464
465        let quote = parse_book_ticker(&msg, &instrument, UnixNanos::from(1)).unwrap();
466        assert_eq!(
467            quote.bid_price.as_decimal(),
468            Decimal::from_str("123.45678901").unwrap()
469        );
470        assert_eq!(
471            quote.ask_price.as_decimal(),
472            Decimal::from_str("123.45678909").unwrap()
473        );
474        assert_eq!(
475            quote.bid_size.as_decimal(),
476            Decimal::from_str("1.23000000").unwrap()
477        );
478        assert_eq!(
479            quote.ask_size.as_decimal(),
480            Decimal::from_str("4.56000000").unwrap()
481        );
482    }
483
484    #[rstest]
485    fn test_parse_book_ticker_accepts_zero_bid_size() {
486        let instrument = sample_instrument();
487        // A side that empties reports a zero size; the quote must still be produced.
488        let msg = BinanceSpotBookTickerMsg {
489            event_type: None,
490            event_time: None,
491            symbol: Ustr::from("ETHUSDT"),
492            book_update_id: 1,
493            best_bid_price: "100.00000000".to_string(),
494            best_bid_qty: "0.00000000".to_string(),
495            best_ask_price: "101.00000000".to_string(),
496            best_ask_qty: "1.00000000".to_string(),
497            transaction_time: None,
498        };
499
500        let quote = parse_book_ticker(&msg, &instrument, UnixNanos::from(1))
501            .expect("zero bid size is a valid quote");
502        assert_eq!(quote.bid_size.as_decimal(), Decimal::from_str("0").unwrap());
503    }
504
505    #[rstest]
506    fn test_parse_depth_snapshot_sets_last_flag_when_final_level_skipped() {
507        let instrument = sample_instrument();
508        // The final ask level has a zero quantity and is skipped during parsing; the
509        // F_LAST terminator must still land on the last emitted delta.
510        let msg = BinanceSpotPartialDepthMsg {
511            symbol: Ustr::from("ETHUSDT"),
512            last_update_id: 1,
513            bids: vec![["100.00000000".to_string(), "1.00000000".to_string()]],
514            asks: vec![
515                ["101.00000000".to_string(), "2.00000000".to_string()],
516                ["102.00000000".to_string(), "0.00000000".to_string()],
517            ],
518        };
519
520        let deltas = parse_depth_snapshot(&msg, &instrument, UnixNanos::from(1))
521            .expect("snapshot should produce deltas");
522
523        let last = deltas.deltas.last().expect("at least one delta");
524        assert_ne!(last.flags & RecordFlag::F_LAST as u8, 0);
525    }
526
527    #[rstest]
528    fn test_parse_depth_diff_sets_delete_actions_and_last_flag_on_final_ask() {
529        let instrument = sample_instrument();
530        let msg = BinanceSpotDepthDiffMsg {
531            event_type: "depthUpdate".to_string(),
532            event_time: 1_700_000_000_000,
533            symbol: Ustr::from("ETHUSDT"),
534            first_update_id: 10,
535            final_update_id: 12,
536            bids: vec![
537                ["100.00000000".to_string(), "1.00000000".to_string()],
538                ["99.00000000".to_string(), "0.00000000".to_string()],
539            ],
540            asks: vec![
541                ["101.00000000".to_string(), "2.00000000".to_string()],
542                ["102.00000000".to_string(), "0.00000000".to_string()],
543            ],
544        };
545
546        let deltas = parse_depth_diff(&msg, &instrument, UnixNanos::from(1))
547            .unwrap()
548            .expect("depth diff should produce deltas");
549
550        assert_eq!(deltas.sequence, 12);
551        assert_eq!(deltas.deltas.len(), 4);
552        assert_eq!(deltas.deltas[0].action, BookAction::Update);
553        assert_eq!(deltas.deltas[0].order.side, OrderSide::Buy);
554        assert_eq!(deltas.deltas[0].flags, 0);
555        assert_eq!(deltas.deltas[1].action, BookAction::Delete);
556        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy);
557        assert_eq!(deltas.deltas[1].order.size.as_decimal(), Decimal::ZERO);
558        assert_eq!(deltas.deltas[1].flags, 0);
559        assert_eq!(deltas.deltas[2].action, BookAction::Update);
560        assert_eq!(deltas.deltas[2].order.side, OrderSide::Sell);
561        assert_eq!(deltas.deltas[2].flags, 0);
562        assert_eq!(deltas.deltas[3].action, BookAction::Delete);
563        assert_eq!(deltas.deltas[3].order.side, OrderSide::Sell);
564        assert_eq!(deltas.deltas[3].order.size.as_decimal(), Decimal::ZERO);
565        assert_eq!(deltas.deltas[3].flags, RecordFlag::F_LAST as u8);
566    }
567}
nautilus_binance/spot/websocket/public_json/parse.rs

nautilus_binance/spot/websocket/public_json/
parse.rs
