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

nautilus_binance/futures/websocket/streams/
parse_data.rs
