nautilus_binance/futures/websocket/streams/
parse_data.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 for Binance Futures WebSocket JSON messages.
17
18use std::str::FromStr;
19
20use nautilus_core::nanos::UnixNanos;
21use nautilus_model::{
22    data::{
23        BarSpecification, BarType, BookOrder, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate,
24        OrderBookDelta, OrderBookDeltas, QuoteTick, TradeTick,
25    },
26    enums::{
27        AggregationSource, AggressorSide, BarAggregation, BookAction, OrderSide, PriceType,
28        RecordFlag,
29    },
30    identifiers::TradeId,
31    instruments::{Instrument, InstrumentAny},
32    types::Price,
33};
34use rust_decimal::Decimal;
35use ustr::Ustr;
36
37use super::{
38    error::{BinanceWsError, BinanceWsResult},
39    messages::{
40        BinanceFuturesAggTradeMsg, BinanceFuturesBookTickerMsg, BinanceFuturesDepthUpdateMsg,
41        BinanceFuturesKlineMsg, BinanceFuturesMarkPriceMsg, BinanceFuturesTickerMsg,
42        BinanceFuturesTradeMsg,
43    },
44};
45use crate::{
46    common::{
47        bar::BinanceBar,
48        enums::{BinanceKlineInterval, BinanceWsEventType},
49        parse::{
50            parse_millis, parse_millis_or_init, parse_required_price_at_precision,
51            parse_required_quantity_at_precision,
52        },
53    },
54    data_types::{BinanceFuturesMarkPriceUpdate, BinanceFuturesTicker},
55};
56
57/// Parses an aggregate trade message into a `TradeTick`.
58///
59/// # Errors
60///
61/// Returns an error if parsing fails.
62pub fn parse_agg_trade(
63    msg: &BinanceFuturesAggTradeMsg,
64    instrument: &InstrumentAny,
65    ts_init: UnixNanos,
66) -> BinanceWsResult<TradeTick> {
67    let instrument_id = instrument.id();
68    let price_precision = instrument.price_precision();
69    let size_precision = instrument.size_precision();
70
71    let price = parse_required_price_at_precision(&msg.price, price_precision, "price")
72        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
73    let size = parse_required_quantity_at_precision(&msg.quantity, size_precision, "quantity")
74        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
75
76    let aggressor_side = if msg.is_buyer_maker {
77        AggressorSide::Sell
78    } else {
79        AggressorSide::Buy
80    };
81
82    let ts_event = parse_millis_or_init(msg.trade_time, "Futures aggregate trade time", ts_init);
83    let trade_id = TradeId::new(msg.agg_trade_id.to_string());
84
85    Ok(TradeTick::new(
86        instrument_id,
87        price,
88        size,
89        aggressor_side,
90        trade_id,
91        ts_event,
92        ts_init,
93    ))
94}
95
96/// Parses a trade message into a `TradeTick`.
97///
98/// # Errors
99///
100/// Returns an error if parsing fails.
101pub fn parse_trade(
102    msg: &BinanceFuturesTradeMsg,
103    instrument: &InstrumentAny,
104    ts_init: UnixNanos,
105) -> BinanceWsResult<TradeTick> {
106    let instrument_id = instrument.id();
107    let price_precision = instrument.price_precision();
108    let size_precision = instrument.size_precision();
109
110    let price = parse_required_price_at_precision(&msg.price, price_precision, "price")
111        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
112    let size = parse_required_quantity_at_precision(&msg.quantity, size_precision, "quantity")
113        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
114
115    let aggressor_side = if msg.is_buyer_maker {
116        AggressorSide::Sell
117    } else {
118        AggressorSide::Buy
119    };
120
121    let ts_event = parse_millis_or_init(msg.trade_time, "Futures trade time", ts_init);
122    let trade_id = TradeId::new(msg.trade_id.to_string());
123
124    Ok(TradeTick::new(
125        instrument_id,
126        price,
127        size,
128        aggressor_side,
129        trade_id,
130        ts_event,
131        ts_init,
132    ))
133}
134
135/// Parses a book ticker message into a `QuoteTick`.
136///
137/// # Errors
138///
139/// Returns an error if parsing fails.
140pub fn parse_book_ticker(
141    msg: &BinanceFuturesBookTickerMsg,
142    instrument: &InstrumentAny,
143    ts_init: UnixNanos,
144) -> BinanceWsResult<QuoteTick> {
145    let instrument_id = instrument.id();
146    let price_precision = instrument.price_precision();
147    let size_precision = instrument.size_precision();
148
149    let bid_price =
150        parse_required_price_at_precision(&msg.best_bid_price, price_precision, "best_bid_price")
151            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
152    let bid_size =
153        parse_required_quantity_at_precision(&msg.best_bid_qty, size_precision, "best_bid_qty")
154            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
155    let ask_price =
156        parse_required_price_at_precision(&msg.best_ask_price, price_precision, "best_ask_price")
157            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
158    let ask_size =
159        parse_required_quantity_at_precision(&msg.best_ask_qty, size_precision, "best_ask_qty")
160            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
161
162    let ts_event = parse_millis_or_init(
163        msg.transaction_time,
164        "Futures book ticker transaction time",
165        ts_init,
166    );
167
168    Ok(QuoteTick::new(
169        instrument_id,
170        bid_price,
171        ask_price,
172        bid_size,
173        ask_size,
174        ts_event,
175        ts_init,
176    ))
177}
178
179/// Parses a depth update message into `OrderBookDeltas`.
180///
181/// # Errors
182///
183/// Returns an error if parsing fails or the update contains no levels.
184pub fn parse_depth_update(
185    msg: &BinanceFuturesDepthUpdateMsg,
186    instrument: &InstrumentAny,
187    ts_init: UnixNanos,
188) -> BinanceWsResult<OrderBookDeltas> {
189    parse_book_depth(msg, instrument, ts_init, false)
190}
191
192pub(crate) fn parse_depth_snapshot(
193    msg: &BinanceFuturesDepthUpdateMsg,
194    instrument: &InstrumentAny,
195    ts_init: UnixNanos,
196) -> BinanceWsResult<OrderBookDeltas> {
197    parse_book_depth(msg, instrument, ts_init, true)
198}
199
200fn parse_book_depth(
201    msg: &BinanceFuturesDepthUpdateMsg,
202    instrument: &InstrumentAny,
203    ts_init: UnixNanos,
204    snapshot: bool,
205) -> BinanceWsResult<OrderBookDeltas> {
206    let instrument_id = instrument.id();
207    let price_precision = instrument.price_precision();
208    let size_precision = instrument.size_precision();
209
210    let ts_event = parse_millis_or_init(
211        msg.transaction_time,
212        "Futures depth update transaction time",
213        ts_init,
214    );
215
216    let mut deltas = Vec::with_capacity(msg.bids.len() + msg.asks.len() + usize::from(snapshot));
217    if snapshot {
218        deltas.push(OrderBookDelta::clear(
219            instrument_id,
220            msg.final_update_id,
221            ts_event,
222            ts_init,
223        ));
224    }
225
226    let flags = if snapshot {
227        RecordFlag::F_SNAPSHOT as u8
228    } else {
229        0
230    };
231
232    // Process bids
233    for bid in &msg.bids {
234        let price = parse_required_price_at_precision(&bid[0], price_precision, "bid_price")
235            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
236        let size = parse_required_quantity_at_precision(&bid[1], size_precision, "bid_quantity")
237            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
238
239        let action = if size.is_zero() {
240            BookAction::Delete
241        } else if snapshot {
242            BookAction::Add
243        } else {
244            BookAction::Update
245        };
246
247        let order = BookOrder::new(OrderSide::Buy, price, size, 0);
248
249        deltas.push(OrderBookDelta::new(
250            instrument_id,
251            action,
252            order,
253            flags,
254            msg.final_update_id,
255            ts_event,
256            ts_init,
257        ));
258    }
259
260    // Process asks
261    for ask in &msg.asks {
262        let price = parse_required_price_at_precision(&ask[0], price_precision, "ask_price")
263            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
264        let size = parse_required_quantity_at_precision(&ask[1], size_precision, "ask_quantity")
265            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
266
267        let action = if size.is_zero() {
268            BookAction::Delete
269        } else if snapshot {
270            BookAction::Add
271        } else {
272            BookAction::Update
273        };
274
275        let order = BookOrder::new(OrderSide::Sell, price, size, 0);
276
277        deltas.push(OrderBookDelta::new(
278            instrument_id,
279            action,
280            order,
281            flags,
282            msg.final_update_id,
283            ts_event,
284            ts_init,
285        ));
286    }
287
288    if let Some(last) = deltas.last_mut() {
289        last.flags |= RecordFlag::F_LAST as u8;
290    }
291
292    OrderBookDeltas::new_checked(instrument_id, deltas)
293        .map_err(|e| BinanceWsError::ParseError(e.to_string()))
294}
295
296/// Parses a mark price message into `MarkPriceUpdate`, `IndexPriceUpdate`, and `FundingRateUpdate`.
297///
298/// # Errors
299///
300/// Returns an error if parsing fails.
301pub fn parse_mark_price(
302    msg: &BinanceFuturesMarkPriceMsg,
303    instrument: &InstrumentAny,
304    ts_init: UnixNanos,
305) -> BinanceWsResult<(
306    MarkPriceUpdate,
307    IndexPriceUpdate,
308    FundingRateUpdate,
309    BinanceFuturesMarkPriceUpdate,
310)> {
311    let instrument_id = instrument.id();
312    let price_precision = instrument.price_precision();
313
314    let mark_price =
315        parse_required_price_at_precision(&msg.mark_price, price_precision, "mark_price")
316            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
317    let index_price =
318        parse_required_price_at_precision(&msg.index_price, price_precision, "index_price")
319            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
320    let estimated_settle_price = msg
321        .estimated_settle_price
322        .parse::<Decimal>()
323        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
324    let estimated_settle_price = Price::from_decimal_dp(estimated_settle_price, price_precision)
325        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
326    let funding_rate = msg
327        .funding_rate
328        .parse::<Decimal>()
329        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
330
331    let ts_event = parse_millis_or_init(msg.event_time, "Futures mark price event time", ts_init);
332    let next_funding_ns = if msg.next_funding_time > 0 {
333        match parse_millis(
334            msg.next_funding_time,
335            "Futures mark price next funding time",
336        ) {
337            Ok(timestamp) => Some(timestamp),
338            Err(e) => {
339                log::warn!("{e}; omitting next funding time");
340                None
341            }
342        }
343    } else {
344        None
345    };
346
347    let mark_update = MarkPriceUpdate::new(instrument_id, mark_price, ts_event, ts_init);
348
349    let index_update = IndexPriceUpdate::new(instrument_id, index_price, ts_event, ts_init);
350
351    let funding_update = FundingRateUpdate::new(
352        instrument_id,
353        funding_rate,
354        None, // Binance does not provide the funding interval through WebSocket API
355        next_funding_ns,
356        ts_event,
357        ts_init,
358    );
359
360    let custom_update = BinanceFuturesMarkPriceUpdate {
361        instrument_id,
362        mark_price,
363        index_price,
364        estimated_settle_price,
365        funding_rate,
366        next_funding_time: next_funding_ns,
367        ts_event,
368        ts_init,
369    };
370
371    Ok((mark_update, index_update, funding_update, custom_update))
372}
373
374/// Parses a 24-hour ticker message into `BinanceFuturesTicker` custom data.
375///
376/// # Errors
377///
378/// Returns an error if parsing fails.
379pub fn parse_ticker(
380    msg: &BinanceFuturesTickerMsg,
381    instrument: &InstrumentAny,
382    ts_init: UnixNanos,
383) -> BinanceWsResult<BinanceFuturesTicker> {
384    Ok(BinanceFuturesTicker::new(
385        instrument.id(),
386        parse_ticker_decimal("price_change", &msg.price_change)?,
387        parse_ticker_decimal("price_change_percent", &msg.price_change_percent)?,
388        parse_ticker_decimal("weighted_avg_price", &msg.weighted_avg_price)?,
389        parse_ticker_decimal("last_price", &msg.last_price)?,
390        parse_ticker_decimal("last_qty", &msg.last_qty)?,
391        parse_ticker_decimal("open_price", &msg.open_price)?,
392        parse_ticker_decimal("high_price", &msg.high_price)?,
393        parse_ticker_decimal("low_price", &msg.low_price)?,
394        parse_ticker_decimal("volume", &msg.volume)?,
395        parse_ticker_decimal("quote_volume", &msg.quote_volume)?,
396        parse_millis_or_init(msg.open_time, "Futures ticker open time", ts_init),
397        parse_millis_or_init(msg.close_time, "Futures ticker close time", ts_init),
398        msg.first_trade_id,
399        msg.last_trade_id,
400        msg.num_trades,
401        parse_millis_or_init(msg.event_time, "Futures ticker event time", ts_init),
402        ts_init,
403    ))
404}
405
406fn parse_ticker_decimal(field: &str, value: &str) -> BinanceWsResult<Decimal> {
407    Decimal::from_str(value).map_err(|e| {
408        BinanceWsError::ParseError(format!("invalid Binance ticker {field}='{value}': {e}"))
409    })
410}
411
412/// Converts a Binance kline interval to a Nautilus `BarSpecification`.
413fn interval_to_bar_spec(interval: BinanceKlineInterval) -> BarSpecification {
414    match interval {
415        BinanceKlineInterval::Second1 => {
416            BarSpecification::new(1, BarAggregation::Second, PriceType::Last)
417        }
418        BinanceKlineInterval::Minute1 => {
419            BarSpecification::new(1, BarAggregation::Minute, PriceType::Last)
420        }
421        BinanceKlineInterval::Minute3 => {
422            BarSpecification::new(3, BarAggregation::Minute, PriceType::Last)
423        }
424        BinanceKlineInterval::Minute5 => {
425            BarSpecification::new(5, BarAggregation::Minute, PriceType::Last)
426        }
427        BinanceKlineInterval::Minute15 => {
428            BarSpecification::new(15, BarAggregation::Minute, PriceType::Last)
429        }
430        BinanceKlineInterval::Minute30 => {
431            BarSpecification::new(30, BarAggregation::Minute, PriceType::Last)
432        }
433        BinanceKlineInterval::Hour1 => {
434            BarSpecification::new(1, BarAggregation::Hour, PriceType::Last)
435        }
436        BinanceKlineInterval::Hour2 => {
437            BarSpecification::new(2, BarAggregation::Hour, PriceType::Last)
438        }
439        BinanceKlineInterval::Hour4 => {
440            BarSpecification::new(4, BarAggregation::Hour, PriceType::Last)
441        }
442        BinanceKlineInterval::Hour6 => {
443            BarSpecification::new(6, BarAggregation::Hour, PriceType::Last)
444        }
445        BinanceKlineInterval::Hour8 => {
446            BarSpecification::new(8, BarAggregation::Hour, PriceType::Last)
447        }
448        BinanceKlineInterval::Hour12 => {
449            BarSpecification::new(12, BarAggregation::Hour, PriceType::Last)
450        }
451        BinanceKlineInterval::Day1 => {
452            BarSpecification::new(1, BarAggregation::Day, PriceType::Last)
453        }
454        BinanceKlineInterval::Day3 => {
455            BarSpecification::new(3, BarAggregation::Day, PriceType::Last)
456        }
457        BinanceKlineInterval::Week1 => {
458            BarSpecification::new(1, BarAggregation::Week, PriceType::Last)
459        }
460        BinanceKlineInterval::Month1 => {
461            BarSpecification::new(1, BarAggregation::Month, PriceType::Last)
462        }
463    }
464}
465
466/// Parses a kline message into a `Bar`.
467///
468/// Returns `None` if the kline is not closed yet.
469///
470/// # Errors
471///
472/// Returns an error if parsing fails.
473pub fn parse_kline(
474    msg: &BinanceFuturesKlineMsg,
475    instrument: &InstrumentAny,
476    ts_init: UnixNanos,
477) -> BinanceWsResult<Option<BinanceBar>> {
478    // Only emit bars when the kline is closed
479    if !msg.kline.is_closed {
480        return Ok(None);
481    }
482
483    let instrument_id = instrument.id();
484    let price_precision = instrument.price_precision();
485    let size_precision = instrument.size_precision();
486
487    let spec = interval_to_bar_spec(msg.kline.interval);
488    let bar_type = BarType::new(instrument_id, spec, AggregationSource::External);
489
490    let price = |field: &str, value: &str| {
491        parse_required_price_at_precision(value, price_precision, field)
492            .map_err(|e| BinanceWsError::ParseError(e.to_string()))
493    };
494    let quantity = |field: &str, value: &str| {
495        parse_required_quantity_at_precision(value, size_precision, field)
496            .map_err(|e| BinanceWsError::ParseError(e.to_string()))
497    };
498    let decimal = |field: &str, value: &str| {
499        Decimal::from_str(value)
500            .map_err(|e| BinanceWsError::ParseError(format!("invalid {field} `{value}`: {e}")))
501    };
502    let count = u64::try_from(msg.kline.num_trades)
503        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
504
505    // Use the kline close time as the event timestamp
506    let ts_event = parse_millis_or_init(msg.kline.close_time, "Futures kline close time", ts_init);
507
508    let bar = BinanceBar::new(
509        bar_type,
510        price("open", &msg.kline.open)?,
511        price("high", &msg.kline.high)?,
512        price("low", &msg.kline.low)?,
513        price("close", &msg.kline.close)?,
514        quantity("volume", &msg.kline.volume)?,
515        decimal("quote volume", &msg.kline.quote_volume)?,
516        count,
517        decimal("taker buy base volume", &msg.kline.taker_buy_volume)?,
518        decimal("taker buy quote volume", &msg.kline.taker_buy_quote_volume)?,
519        ts_event,
520        ts_init,
521    );
522
523    Ok(Some(bar))
524}
525
526/// Extracts the symbol from a raw JSON message.
527pub fn extract_symbol(json: &serde_json::Value) -> Option<Ustr> {
528    json.get("s").and_then(|v| v.as_str()).map(Ustr::from)
529}
530
531/// Extracts the event type from a raw JSON message.
532pub fn extract_event_type(json: &serde_json::Value) -> Option<BinanceWsEventType> {
533    json.get("e")
534        .and_then(|v| serde_json::from_value(v.clone()).ok())
535}
536
537#[cfg(test)]
538mod tests {
539    use nautilus_model::{enums::BookType, orderbook::OrderBook, types::Quantity};
540    use rstest::rstest;
541    use rust_decimal_macros::dec;
542    use serde::de::DeserializeOwned;
543    use serde_json::json;
544
545    use super::*;
546    use crate::{
547        common::{
548            enums::{BinanceOrderStatus, BinanceSide, BinanceTradingStatus},
549            parse::parse_usdm_instrument,
550            testing::{load_fixture_string, load_json_fixture},
551        },
552        futures::{
553            http::models::BinanceFuturesUsdSymbol,
554            websocket::streams::messages::{BinanceFuturesLiquidationMsg, BinanceFuturesTickerMsg},
555        },
556    };
557
558    const PRICE_PRECISION: u8 = 8;
559    const SIZE_PRECISION: u8 = 3;
560
561    fn sample_futures_symbol() -> BinanceFuturesUsdSymbol {
562        BinanceFuturesUsdSymbol {
563            symbol: Ustr::from("BTCUSDT"),
564            pair: Ustr::from("BTCUSDT"),
565            contract_type: "PERPETUAL".to_string(),
566            delivery_date: 4_133_404_800_000,
567            onboard_date: 1_569_398_400_000,
568            status: BinanceTradingStatus::Trading,
569            maint_margin_percent: "2.5000".to_string(),
570            required_margin_percent: "5.0000".to_string(),
571            base_asset: Ustr::from("BTC"),
572            quote_asset: Ustr::from("USDT"),
573            margin_asset: Ustr::from("USDT"),
574            price_precision: PRICE_PRECISION as i32,
575            quantity_precision: SIZE_PRECISION as i32,
576            base_asset_precision: 8,
577            quote_precision: 8,
578            underlying_type: Some("COIN".to_string()),
579            underlying_sub_type: vec!["PoW".to_string()],
580            settle_plan: None,
581            trigger_protect: Some("0.0500".to_string()),
582            liquidation_fee: Some("0.012500".to_string()),
583            market_take_bound: Some("0.05".to_string()),
584            order_types: vec!["LIMIT".to_string(), "MARKET".to_string()],
585            time_in_force: vec!["GTC".to_string(), "IOC".to_string()],
586            filters: vec![
587                json!({
588                    "filterType": "PRICE_FILTER",
589                    "tickSize": "0.00000001",
590                    "maxPrice": "1000000",
591                    "minPrice": "0.00000001"
592                }),
593                json!({
594                    "filterType": "LOT_SIZE",
595                    "stepSize": "0.001",
596                    "maxQty": "1000",
597                    "minQty": "0.001"
598                }),
599            ],
600        }
601    }
602
603    fn sample_instrument() -> InstrumentAny {
604        let ts = UnixNanos::from(1_700_000_000_000_000_000u64);
605        parse_usdm_instrument(&sample_futures_symbol(), ts, ts).unwrap()
606    }
607
608    fn load_market_fixture<T: DeserializeOwned>(filename: &str) -> T {
609        let path = format!("futures/market_data_json/{filename}");
610        serde_json::from_str(&load_fixture_string(&path))
611            .unwrap_or_else(|e| panic!("Failed to parse fixture {path}: {e}"))
612    }
613
614    #[rstest]
615    fn test_parse_agg_trade() {
616        let instrument = sample_instrument();
617        let msg: BinanceFuturesAggTradeMsg = load_market_fixture("agg_trade_stream.json");
618        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
619
620        let trade = parse_agg_trade(&msg, &instrument, ts_init).unwrap();
621
622        assert_eq!(trade.instrument_id, instrument.id());
623        assert_eq!(trade.price, Price::new(0.001, PRICE_PRECISION));
624        assert_eq!(trade.size, Quantity::new(100.0, SIZE_PRECISION));
625        assert_eq!(trade.aggressor_side, AggressorSide::Sell);
626        assert_eq!(trade.trade_id, TradeId::new("5933014"));
627        assert_eq!(trade.ts_event, UnixNanos::from(123_456_785_000_000u64));
628        assert_eq!(trade.ts_init, ts_init);
629    }
630
631    #[rstest]
632    #[case::negative(-1)]
633    #[case::overflow(i64::MAX)]
634    fn test_parse_agg_trade_falls_back_for_invalid_timestamp(#[case] trade_time: i64) {
635        let instrument = sample_instrument();
636        let mut msg: BinanceFuturesAggTradeMsg = load_market_fixture("agg_trade_stream.json");
637        msg.trade_time = trade_time;
638
639        let ts_init = UnixNanos::from(1);
640        let trade = parse_agg_trade(&msg, &instrument, ts_init).unwrap();
641
642        assert_eq!(trade.ts_event, ts_init);
643        assert_eq!(trade.ts_init, ts_init);
644    }
645
646    #[rstest]
647    fn test_parse_trade() {
648        let instrument = sample_instrument();
649        let msg: BinanceFuturesTradeMsg = load_market_fixture("trade_stream.json");
650        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
651
652        let trade = parse_trade(&msg, &instrument, ts_init).unwrap();
653
654        assert_eq!(trade.instrument_id, instrument.id());
655        assert_eq!(trade.price, Price::new(0.001, PRICE_PRECISION));
656        assert_eq!(trade.size, Quantity::new(100.0, SIZE_PRECISION));
657        assert_eq!(trade.aggressor_side, AggressorSide::Sell);
658        assert_eq!(trade.trade_id, TradeId::new("5933014"));
659        assert_eq!(trade.ts_event, UnixNanos::from(123_456_785_000_000u64));
660        assert_eq!(trade.ts_init, ts_init);
661    }
662
663    #[rstest]
664    fn test_parse_book_ticker() {
665        let instrument = sample_instrument();
666        let msg: BinanceFuturesBookTickerMsg = load_market_fixture("book_ticker_stream.json");
667        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
668
669        let quote = parse_book_ticker(&msg, &instrument, ts_init).unwrap();
670
671        assert_eq!(quote.instrument_id, instrument.id());
672        assert_eq!(quote.bid_price, Price::new(25.3519, PRICE_PRECISION));
673        assert_eq!(quote.ask_price, Price::new(25.3652, PRICE_PRECISION));
674        assert_eq!(quote.bid_size, Quantity::new(31.21, SIZE_PRECISION));
675        assert_eq!(quote.ask_size, Quantity::new(40.66, SIZE_PRECISION));
676        assert_eq!(
677            quote.ts_event,
678            UnixNanos::from(1_568_014_460_891_000_000u64)
679        );
680        assert_eq!(quote.ts_init, ts_init);
681    }
682
683    #[rstest]
684    fn test_parse_depth_update() {
685        let instrument = sample_instrument();
686        let msg: BinanceFuturesDepthUpdateMsg = load_market_fixture("depth_update_stream.json");
687        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
688
689        let deltas = parse_depth_update(&msg, &instrument, ts_init).unwrap();
690
691        assert_eq!(deltas.instrument_id, instrument.id());
692        assert_eq!(deltas.deltas.len(), 2);
693        assert_eq!(deltas.sequence, 160);
694        assert_eq!(deltas.ts_event, UnixNanos::from(123_456_788_000_000u64));
695        assert_eq!(deltas.ts_init, ts_init);
696        assert_eq!(deltas.deltas[0].action, BookAction::Update);
697        assert_eq!(deltas.deltas[0].order.side, OrderSide::Buy.into());
698        assert_eq!(
699            deltas.deltas[0].order.price,
700            Price::new(0.0024, PRICE_PRECISION)
701        );
702        assert_eq!(
703            deltas.deltas[0].order.size,
704            Quantity::new(10.0, SIZE_PRECISION)
705        );
706        assert_eq!(deltas.deltas[1].action, BookAction::Update);
707        assert_eq!(deltas.deltas[1].order.side, OrderSide::Sell.into());
708        assert_eq!(
709            deltas.deltas[1].order.price,
710            Price::new(0.0026, PRICE_PRECISION)
711        );
712        assert_eq!(
713            deltas.deltas[1].order.size,
714            Quantity::new(100.0, SIZE_PRECISION)
715        );
716        assert_eq!(deltas.deltas[1].flags, RecordFlag::F_LAST as u8);
717    }
718
719    #[rstest]
720    fn test_parse_depth_update_without_levels_returns_error() {
721        let instrument = sample_instrument();
722        let mut msg: BinanceFuturesDepthUpdateMsg = load_market_fixture("depth_update_stream.json");
723        msg.bids.clear();
724        msg.asks.clear();
725
726        let result = parse_depth_update(&msg, &instrument, UnixNanos::from(1));
727
728        assert!(matches!(result, Err(BinanceWsError::ParseError(_))));
729    }
730
731    #[rstest]
732    #[case::five(5)]
733    #[case::ten(10)]
734    #[case::twenty(20)]
735    fn test_parse_depth_snapshot_replaces_levels(#[case] depth: usize) {
736        let instrument = sample_instrument();
737        let mut msg: BinanceFuturesDepthUpdateMsg = load_market_fixture("depth_update_stream.json");
738        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
739        let mut book = OrderBook::new(instrument.id(), BookType::L2_MBP);
740
741        for offset in [0, 100] {
742            msg.bids = (0..depth)
743                .map(|i| [(1000 - offset - i).to_string(), "2.000".into()])
744                .collect();
745            msg.asks = (0..depth)
746                .map(|i| [(2000 + offset + i).to_string(), "3.000".into()])
747                .collect();
748            let snapshot = parse_depth_snapshot(&msg, &instrument, ts_init).unwrap();
749            let mut expected = vec![OrderBookDelta::clear(
750                instrument.id(),
751                msg.final_update_id,
752                snapshot.ts_event,
753                ts_init,
754            )];
755
756            for (side, levels) in [(OrderSide::Buy, &msg.bids), (OrderSide::Sell, &msg.asks)] {
757                for level in levels {
758                    expected.push(OrderBookDelta::new(
759                        instrument.id(),
760                        BookAction::Add,
761                        BookOrder::new(
762                            side,
763                            Price::from_str(&format!("{}.00000000", level[0])).unwrap(),
764                            Quantity::from_str(&level[1]).unwrap(),
765                            0,
766                        ),
767                        RecordFlag::F_SNAPSHOT as u8,
768                        msg.final_update_id,
769                        UnixNanos::from(123_456_788_000_000u64),
770                        ts_init,
771                    ));
772                }
773            }
774
775            expected.last_mut().unwrap().flags =
776                RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8;
777            assert_eq!(snapshot.instrument_id, instrument.id());
778            assert_eq!(snapshot.sequence, msg.final_update_id);
779            assert_eq!(snapshot.deltas, expected);
780            book.apply_deltas(&snapshot).unwrap();
781            assert_eq!(book.bids(None).count(), depth);
782            assert_eq!(book.asks(None).count(), depth);
783            for (actual, input) in book.bids(None).zip(&msg.bids) {
784                assert_eq!(
785                    actual.price.value.as_decimal(),
786                    Decimal::from_str(&input[0]).unwrap()
787                );
788                assert_eq!(actual.size_decimal(), dec!(2));
789            }
790
791            for (actual, input) in book.asks(None).zip(&msg.asks) {
792                assert_eq!(
793                    actual.price.value.as_decimal(),
794                    Decimal::from_str(&input[0]).unwrap()
795                );
796                assert_eq!(actual.size_decimal(), dec!(3));
797            }
798        }
799
800        msg.bids.clear();
801        msg.asks.clear();
802        let empty = parse_depth_snapshot(&msg, &instrument, ts_init).unwrap();
803        book.apply_deltas(&empty).unwrap();
804        assert_eq!(empty.deltas.len(), 1);
805        assert_eq!(
806            empty.deltas[0].flags,
807            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
808        );
809        assert_eq!(book.bids(None).count(), 0);
810        assert_eq!(book.asks(None).count(), 0);
811    }
812
813    #[rstest]
814    fn test_market_data_parsers_preserve_decimal_prices() {
815        let instrument = sample_instrument();
816        let ts_init = UnixNanos::from(1);
817        let price = "123456789.12345678";
818        let mut agg: BinanceFuturesAggTradeMsg = load_market_fixture("agg_trade_stream.json");
819        agg.price = price.to_string();
820        let mut trade: BinanceFuturesTradeMsg = load_market_fixture("trade_stream.json");
821        trade.price = price.to_string();
822        let mut book: BinanceFuturesBookTickerMsg = load_market_fixture("book_ticker_stream.json");
823        book.best_bid_price = price.to_string();
824        book.best_ask_price = "123456789.87654321".to_string();
825        let mut depth: BinanceFuturesDepthUpdateMsg =
826            load_market_fixture("depth_update_stream.json");
827        depth.bids[0][0] = price.to_string();
828        depth.asks[0][0] = book.best_ask_price.clone();
829        depth.asks[0][1] = "0.000".to_string();
830
831        let agg = parse_agg_trade(&agg, &instrument, ts_init).unwrap();
832        let trade = parse_trade(&trade, &instrument, ts_init).unwrap();
833        let book = parse_book_ticker(&book, &instrument, ts_init).unwrap();
834        let depth = parse_depth_update(&depth, &instrument, ts_init).unwrap();
835
836        assert_eq!(agg.price, Price::from(price));
837        assert_eq!(trade.price, Price::from(price));
838        assert_eq!(book.bid_price, Price::from(price));
839        assert_eq!(book.ask_price, Price::from("123456789.87654321"));
840        assert_eq!(depth.deltas[0].order.price, Price::from(price));
841        assert_eq!(
842            depth.deltas[1].order.price,
843            Price::from("123456789.87654321")
844        );
845        assert_eq!(depth.deltas[1].order.size, Quantity::from("0.000"));
846        assert_eq!(depth.deltas[1].action, BookAction::Delete);
847    }
848
849    #[rstest]
850    fn test_parse_mark_price() {
851        let instrument = sample_instrument();
852        let msg: BinanceFuturesMarkPriceMsg = load_market_fixture("mark_price_stream.json");
853        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
854
855        let (mark, index, funding, custom) = parse_mark_price(&msg, &instrument, ts_init).unwrap();
856
857        assert_eq!(mark.instrument_id, instrument.id());
858        assert_eq!(mark.value, Price::new(11794.15, PRICE_PRECISION));
859        assert_eq!(index.value, Price::new(11784.62659091, PRICE_PRECISION));
860        assert_eq!(mark.ts_event, UnixNanos::from(1_562_305_380_000_000_000u64));
861        assert_eq!(funding.instrument_id, instrument.id());
862        assert_eq!(funding.rate.to_string(), "0.00038167");
863        assert_eq!(
864            funding.next_funding_ns,
865            Some(UnixNanos::from(1_562_306_400_000_000_000u64))
866        );
867        assert_eq!(
868            funding.ts_event,
869            UnixNanos::from(1_562_305_380_000_000_000u64)
870        );
871        assert_eq!(funding.ts_init, ts_init);
872        assert_eq!(custom.instrument_id, instrument.id());
873        assert_eq!(custom.mark_price, Price::from("11794.15000000"));
874        assert_eq!(custom.index_price, Price::from("11784.62659091"));
875        assert_eq!(custom.estimated_settle_price, Price::from("11784.25641265"));
876        assert_eq!(custom.funding_rate, dec!(0.00038167));
877        assert_eq!(custom.next_funding_time, funding.next_funding_ns);
878        assert_eq!(custom.ts_event, mark.ts_event);
879        assert_eq!(custom.ts_init, ts_init);
880    }
881
882    #[rstest]
883    #[case::zero(0)]
884    #[case::negative(-1)]
885    fn test_parse_mark_price_preserves_missing_funding_time(#[case] next_funding_time: i64) {
886        let instrument = sample_instrument();
887        let mut msg: BinanceFuturesMarkPriceMsg = load_market_fixture("mark_price_stream.json");
888        msg.next_funding_time = next_funding_time;
889
890        let (_, _, funding, custom) =
891            parse_mark_price(&msg, &instrument, UnixNanos::from(1)).unwrap();
892
893        assert_eq!(funding.next_funding_ns, None);
894        assert_eq!(custom.next_funding_time, None);
895    }
896
897    #[rstest]
898    fn test_parse_kline_closed() {
899        let instrument = sample_instrument();
900        let msg: BinanceFuturesKlineMsg = load_market_fixture("kline_stream_closed.json");
901        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
902
903        let bar = parse_kline(&msg, &instrument, ts_init).unwrap().unwrap();
904
905        assert_eq!(bar.bar_type.instrument_id(), instrument.id());
906        assert_eq!(bar.open, Price::new(0.001, PRICE_PRECISION));
907        assert_eq!(bar.high, Price::new(0.0025, PRICE_PRECISION));
908        assert_eq!(bar.low, Price::new(0.001, PRICE_PRECISION));
909        assert_eq!(bar.close, Price::new(0.002, PRICE_PRECISION));
910        assert_eq!(bar.volume, Quantity::new(1000.0, SIZE_PRECISION));
911        assert_eq!(bar.quote_volume, dec!(1.0000));
912        assert_eq!(bar.count, 100);
913        assert_eq!(bar.taker_buy_base_volume, dec!(500));
914        assert_eq!(bar.taker_buy_quote_volume, dec!(0.500));
915        assert_eq!(bar.ts_event, UnixNanos::from(1_638_747_719_999_000_000u64));
916        assert_eq!(bar.ts_init, ts_init);
917    }
918
919    #[rstest]
920    fn test_parse_kline_open_returns_none() {
921        let instrument = sample_instrument();
922        let msg: BinanceFuturesKlineMsg = load_market_fixture("kline_stream_open.json");
923
924        let bar = parse_kline(&msg, &instrument, UnixNanos::default()).unwrap();
925
926        assert!(bar.is_none());
927    }
928
929    #[rstest]
930    fn test_mark_price_msg_deserializes_optional_ap() {
931        let json = r#"{
932            "e": "markPriceUpdate",
933            "E": 1562305380000,
934            "s": "BTCUSDT",
935            "p": "11794.15000000",
936            "ap": "11792.85000000",
937            "i": "11784.62659091",
938            "P": "11784.25641265",
939            "r": "0.00038167",
940            "T": 1562306400000
941        }"#;
942
943        let msg: BinanceFuturesMarkPriceMsg = serde_json::from_str(json).unwrap();
944        assert_eq!(msg.mark_price_moving_avg.as_deref(), Some("11792.85000000"));
945
946        let legacy: BinanceFuturesMarkPriceMsg = load_market_fixture("mark_price_stream.json");
947        assert!(legacy.mark_price_moving_avg.is_none());
948    }
949
950    #[rstest]
951    fn test_parse_mark_price_funding_rate_fields() {
952        let instrument = sample_instrument();
953        let msg: BinanceFuturesMarkPriceMsg = load_market_fixture("mark_price_stream.json");
954        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
955
956        let (_mark, _index, funding, _custom) =
957            parse_mark_price(&msg, &instrument, ts_init).unwrap();
958
959        assert_eq!(funding.instrument_id, instrument.id());
960        assert_eq!(funding.rate.to_string(), "0.00038167");
961        assert!(funding.interval.is_none());
962        assert_eq!(
963            funding.next_funding_ns,
964            Some(UnixNanos::from(1_562_306_400_000_000_000u64))
965        );
966        assert_eq!(
967            funding.ts_event,
968            UnixNanos::from(1_562_305_380_000_000_000u64)
969        );
970        assert_eq!(funding.ts_init, ts_init);
971    }
972
973    #[rstest]
974    fn test_deserialize_liquidation_msg() {
975        let msg: BinanceFuturesLiquidationMsg = load_market_fixture("liquidation_stream.json");
976
977        assert_eq!(msg.event_type, "forceOrder");
978        assert_eq!(msg.event_time, 1_568_014_460_893);
979        assert_eq!(msg.order.symbol, Ustr::from("BTCUSDT"));
980        assert_eq!(msg.order.side, BinanceSide::Sell);
981        assert_eq!(msg.order.original_qty, "0.014");
982        assert_eq!(msg.order.average_price, "9910.12345678");
983        assert_eq!(msg.order.status, BinanceOrderStatus::Filled);
984        assert_eq!(msg.order.accumulated_qty, "0.014");
985        assert_eq!(msg.order.trade_time, 1_568_014_460_893);
986    }
987
988    #[rstest]
989    fn test_deserialize_ticker_msg() {
990        let msg: BinanceFuturesTickerMsg = load_market_fixture("ticker_stream.json");
991
992        assert_eq!(msg.event_type, "24hrTicker");
993        assert_eq!(msg.symbol, Ustr::from("BTCUSDT"));
994        assert_eq!(msg.price_change, "-131.40000000");
995        assert_eq!(msg.price_change_percent, "-0.786");
996        assert_eq!(msg.weighted_avg_price, "16628.97377498");
997        assert_eq!(msg.last_price, "16584.60000000");
998        assert_eq!(msg.open_price, "16716.00000000");
999        assert_eq!(msg.high_price, "16764.89000000");
1000        assert_eq!(msg.low_price, "16456.51000000");
1001        assert_eq!(msg.volume, "122474.816");
1002        assert_eq!(msg.quote_volume, "2036102085.69746400");
1003        assert_eq!(msg.num_trades, 142853);
1004    }
1005
1006    #[rstest]
1007    fn test_parse_ticker() {
1008        let instrument = sample_instrument();
1009        let msg: BinanceFuturesTickerMsg = load_market_fixture("ticker_stream.json");
1010        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
1011
1012        let ticker = parse_ticker(&msg, &instrument, ts_init).unwrap();
1013
1014        assert_eq!(ticker.instrument_id, instrument.id());
1015        assert_eq!(ticker.price_change, dec!(-131.40000000));
1016        assert_eq!(ticker.price_change_percent, dec!(-0.786));
1017        assert_eq!(ticker.weighted_avg_price, dec!(16628.97377498));
1018        assert_eq!(ticker.last_price, dec!(16584.60000000));
1019        assert_eq!(ticker.last_qty, dec!(0.002));
1020        assert_eq!(ticker.open_price, dec!(16716.00000000));
1021        assert_eq!(ticker.high_price, dec!(16764.89000000));
1022        assert_eq!(ticker.low_price, dec!(16456.51000000));
1023        assert_eq!(ticker.volume, dec!(122474.816));
1024        assert_eq!(ticker.quote_volume, dec!(2036102085.69746400));
1025        assert_eq!(ticker.open_time, UnixNanos::from_millis(1_672_429_382_136));
1026        assert_eq!(ticker.close_time, UnixNanos::from_millis(1_672_515_782_136));
1027        assert_eq!(ticker.first_trade_id, 2_289_691);
1028        assert_eq!(ticker.last_trade_id, 2_432_543);
1029        assert_eq!(ticker.num_trades, 142_853);
1030        assert_eq!(ticker.ts_event, UnixNanos::from_millis(1_672_515_782_136));
1031        assert_eq!(ticker.ts_init, ts_init);
1032    }
1033
1034    #[rstest]
1035    fn test_parse_ticker_rejects_invalid_numeric_field() {
1036        let instrument = sample_instrument();
1037        let mut msg: BinanceFuturesTickerMsg = load_market_fixture("ticker_stream.json");
1038        msg.last_price = "not-a-decimal".to_string();
1039
1040        let result = parse_ticker(&msg, &instrument, UnixNanos::default());
1041
1042        assert!(result.is_err());
1043    }
1044
1045    #[rstest]
1046    fn test_extract_symbol() {
1047        let json = load_json_fixture("futures/market_data_json/book_ticker_stream.json");
1048
1049        let symbol = extract_symbol(&json);
1050
1051        assert_eq!(symbol, Some(Ustr::from("BNBUSDT")));
1052    }
1053
1054    #[rstest]
1055    fn test_extract_event_type() {
1056        let json = load_json_fixture("futures/market_data_json/mark_price_stream.json");
1057
1058        let event_type = extract_event_type(&json);
1059
1060        assert_eq!(event_type, Some(BinanceWsEventType::MarkPriceUpdate));
1061    }
1062
1063    #[rstest]
1064    fn test_extract_event_type_force_order() {
1065        let json = load_json_fixture("futures/market_data_json/liquidation_stream.json");
1066
1067        let event_type = extract_event_type(&json);
1068
1069        assert_eq!(event_type, Some(BinanceWsEventType::ForceOrder));
1070    }
1071
1072    #[rstest]
1073    fn test_extract_event_type_ticker() {
1074        let json = load_json_fixture("futures/market_data_json/ticker_stream.json");
1075
1076        let event_type = extract_event_type(&json);
1077
1078        assert_eq!(event_type, Some(BinanceWsEventType::Ticker24Hr));
1079    }
1080}
nautilus_binance/futures/websocket/streams/parse_data.rs

nautilus_binance/futures/websocket/streams/
parse_data.rs
