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

nautilus_binance/futures/websocket/streams/
parse_data.rs
