nautilus_binance/futures/websocket/streams/

parse_data.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Parsing for Binance Futures WebSocket JSON messages.

use std::str::FromStr;

use nautilus_core::nanos::UnixNanos;
use nautilus_model::{
    data::{
        BarSpecification, BarType, BookOrder, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate,
        OrderBookDelta, OrderBookDeltas, QuoteTick, TradeTick,
    },
    enums::{
        AggregationSource, AggressorSide, BarAggregation, BookAction, OrderSide, PriceType,
        RecordFlag,
    },
    identifiers::TradeId,
    instruments::{Instrument, InstrumentAny},
    types::Price,
};
use rust_decimal::Decimal;
use ustr::Ustr;

use super::{
    error::{BinanceWsError, BinanceWsResult},
    messages::{
        BinanceFuturesAggTradeMsg, BinanceFuturesBookTickerMsg, BinanceFuturesDepthUpdateMsg,
        BinanceFuturesKlineMsg, BinanceFuturesMarkPriceMsg, BinanceFuturesTickerMsg,
        BinanceFuturesTradeMsg,
    },
};
use crate::{
    common::{
        bar::BinanceBar,
        enums::{BinanceKlineInterval, BinanceWsEventType},
        parse::{
            parse_millis, parse_millis_or_init, parse_required_price_at_precision,
            parse_required_quantity_at_precision,
        },
    },
    data_types::{BinanceFuturesMarkPriceUpdate, BinanceFuturesTicker},
};

/// Parses an aggregate trade message into a `TradeTick`.
///
/// # Errors
///
/// Returns an error if parsing fails.
pub fn parse_agg_trade(
    msg: &BinanceFuturesAggTradeMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> BinanceWsResult<TradeTick> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let price = parse_required_price_at_precision(&msg.price, price_precision, "price")
        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
    let size = parse_required_quantity_at_precision(&msg.quantity, size_precision, "quantity")
        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;

    let aggressor_side = if msg.is_buyer_maker {
        AggressorSide::Sell
    } else {
        AggressorSide::Buy
    };

    let ts_event = parse_millis_or_init(msg.trade_time, "Futures aggregate trade time", ts_init);
    let trade_id = TradeId::new(msg.agg_trade_id.to_string());

    Ok(TradeTick::new(
        instrument_id,
        price,
        size,
        aggressor_side,
        trade_id,
        ts_event,
        ts_init,
    ))
}

/// Parses a trade message into a `TradeTick`.
///
/// # Errors
///
/// Returns an error if parsing fails.
pub fn parse_trade(
    msg: &BinanceFuturesTradeMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> BinanceWsResult<TradeTick> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let price = parse_required_price_at_precision(&msg.price, price_precision, "price")
        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
    let size = parse_required_quantity_at_precision(&msg.quantity, size_precision, "quantity")
        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;

    let aggressor_side = if msg.is_buyer_maker {
        AggressorSide::Sell
    } else {
        AggressorSide::Buy
    };

    let ts_event = parse_millis_or_init(msg.trade_time, "Futures trade time", ts_init);
    let trade_id = TradeId::new(msg.trade_id.to_string());

    Ok(TradeTick::new(
        instrument_id,
        price,
        size,
        aggressor_side,
        trade_id,
        ts_event,
        ts_init,
    ))
}

/// Parses a book ticker message into a `QuoteTick`.
///
/// # Errors
///
/// Returns an error if parsing fails.
pub fn parse_book_ticker(
    msg: &BinanceFuturesBookTickerMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> BinanceWsResult<QuoteTick> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let bid_price =
        parse_required_price_at_precision(&msg.best_bid_price, price_precision, "best_bid_price")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
    let bid_size =
        parse_required_quantity_at_precision(&msg.best_bid_qty, size_precision, "best_bid_qty")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
    let ask_price =
        parse_required_price_at_precision(&msg.best_ask_price, price_precision, "best_ask_price")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
    let ask_size =
        parse_required_quantity_at_precision(&msg.best_ask_qty, size_precision, "best_ask_qty")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;

    let ts_event = parse_millis_or_init(
        msg.transaction_time,
        "Futures book ticker transaction time",
        ts_init,
    );

    Ok(QuoteTick::new(
        instrument_id,
        bid_price,
        ask_price,
        bid_size,
        ask_size,
        ts_event,
        ts_init,
    ))
}

/// Parses a depth update message into `OrderBookDeltas`.
///
/// # Errors
///
/// Returns an error if parsing fails.
pub fn parse_depth_update(
    msg: &BinanceFuturesDepthUpdateMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> BinanceWsResult<OrderBookDeltas> {
    parse_book_depth(msg, instrument, ts_init, false)
}

pub(crate) fn parse_depth_snapshot(
    msg: &BinanceFuturesDepthUpdateMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> BinanceWsResult<OrderBookDeltas> {
    parse_book_depth(msg, instrument, ts_init, true)
}

fn parse_book_depth(
    msg: &BinanceFuturesDepthUpdateMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
    snapshot: bool,
) -> BinanceWsResult<OrderBookDeltas> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let ts_event = parse_millis_or_init(
        msg.transaction_time,
        "Futures depth update transaction time",
        ts_init,
    );

    let mut deltas = Vec::with_capacity(msg.bids.len() + msg.asks.len() + usize::from(snapshot));
    if snapshot {
        deltas.push(OrderBookDelta::clear(
            instrument_id,
            msg.final_update_id,
            ts_event,
            ts_init,
        ));
    }

    // Process bids
    for (i, bid) in msg.bids.iter().enumerate() {
        let price = parse_required_price_at_precision(&bid[0], price_precision, "bid_price")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
        let size = parse_required_quantity_at_precision(&bid[1], size_precision, "bid_quantity")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;

        let action = if size.is_zero() {
            BookAction::Delete
        } else if snapshot {
            BookAction::Add
        } else {
            BookAction::Update
        };

        let is_last = i == msg.bids.len() - 1 && msg.asks.is_empty();
        let flags = if is_last { RecordFlag::F_LAST as u8 } else { 0 };

        let order = BookOrder::new(OrderSide::Buy, price, size, 0);

        deltas.push(OrderBookDelta::new(
            instrument_id,
            action,
            order,
            flags,
            msg.final_update_id,
            ts_event,
            ts_init,
        ));
    }

    // Process asks
    for (i, ask) in msg.asks.iter().enumerate() {
        let price = parse_required_price_at_precision(&ask[0], price_precision, "ask_price")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
        let size = parse_required_quantity_at_precision(&ask[1], size_precision, "ask_quantity")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;

        let action = if size.is_zero() {
            BookAction::Delete
        } else if snapshot {
            BookAction::Add
        } else {
            BookAction::Update
        };

        let is_last = i == msg.asks.len() - 1;
        let flags = if is_last { RecordFlag::F_LAST as u8 } else { 0 };

        let order = BookOrder::new(OrderSide::Sell, price, size, 0);

        deltas.push(OrderBookDelta::new(
            instrument_id,
            action,
            order,
            flags,
            msg.final_update_id,
            ts_event,
            ts_init,
        ));
    }

    if snapshot && let Some(last) = deltas.last_mut() {
        last.flags |= RecordFlag::F_LAST as u8;
    }

    Ok(OrderBookDeltas::new(instrument_id, deltas))
}

/// Parses a mark price message into `MarkPriceUpdate`, `IndexPriceUpdate`, and `FundingRateUpdate`.
///
/// # Errors
///
/// Returns an error if parsing fails.
pub fn parse_mark_price(
    msg: &BinanceFuturesMarkPriceMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> BinanceWsResult<(
    MarkPriceUpdate,
    IndexPriceUpdate,
    FundingRateUpdate,
    BinanceFuturesMarkPriceUpdate,
)> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();

    let mark_price =
        parse_required_price_at_precision(&msg.mark_price, price_precision, "mark_price")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
    let index_price =
        parse_required_price_at_precision(&msg.index_price, price_precision, "index_price")
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
    let estimated_settle_price = msg
        .estimated_settle_price
        .parse::<Decimal>()
        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
    let estimated_settle_price = Price::from_decimal_dp(estimated_settle_price, price_precision)
        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;
    let funding_rate = msg
        .funding_rate
        .parse::<Decimal>()
        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;

    let ts_event = parse_millis_or_init(msg.event_time, "Futures mark price event time", ts_init);
    let next_funding_ns = if msg.next_funding_time > 0 {
        match parse_millis(
            msg.next_funding_time,
            "Futures mark price next funding time",
        ) {
            Ok(timestamp) => Some(timestamp),
            Err(e) => {
                log::warn!("{e}; omitting next funding time");
                None
            }
        }
    } else {
        None
    };

    let mark_update = MarkPriceUpdate::new(instrument_id, mark_price, ts_event, ts_init);

    let index_update = IndexPriceUpdate::new(instrument_id, index_price, ts_event, ts_init);

    let funding_update = FundingRateUpdate::new(
        instrument_id,
        funding_rate,
        None, // Binance does not provide the funding interval through WebSocket API
        next_funding_ns,
        ts_event,
        ts_init,
    );

    let custom_update = BinanceFuturesMarkPriceUpdate {
        instrument_id,
        mark_price,
        index_price,
        estimated_settle_price,
        funding_rate,
        next_funding_time: next_funding_ns,
        ts_event,
        ts_init,
    };

    Ok((mark_update, index_update, funding_update, custom_update))
}

/// Parses a 24-hour ticker message into `BinanceFuturesTicker` custom data.
///
/// # Errors
///
/// Returns an error if parsing fails.
pub fn parse_ticker(
    msg: &BinanceFuturesTickerMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> BinanceWsResult<BinanceFuturesTicker> {
    Ok(BinanceFuturesTicker::new(
        instrument.id(),
        parse_ticker_decimal("price_change", &msg.price_change)?,
        parse_ticker_decimal("price_change_percent", &msg.price_change_percent)?,
        parse_ticker_decimal("weighted_avg_price", &msg.weighted_avg_price)?,
        parse_ticker_decimal("last_price", &msg.last_price)?,
        parse_ticker_decimal("last_qty", &msg.last_qty)?,
        parse_ticker_decimal("open_price", &msg.open_price)?,
        parse_ticker_decimal("high_price", &msg.high_price)?,
        parse_ticker_decimal("low_price", &msg.low_price)?,
        parse_ticker_decimal("volume", &msg.volume)?,
        parse_ticker_decimal("quote_volume", &msg.quote_volume)?,
        parse_millis_or_init(msg.open_time, "Futures ticker open time", ts_init),
        parse_millis_or_init(msg.close_time, "Futures ticker close time", ts_init),
        msg.first_trade_id,
        msg.last_trade_id,
        msg.num_trades,
        parse_millis_or_init(msg.event_time, "Futures ticker event time", ts_init),
        ts_init,
    ))
}

fn parse_ticker_decimal(field: &str, value: &str) -> BinanceWsResult<Decimal> {
    Decimal::from_str(value).map_err(|e| {
        BinanceWsError::ParseError(format!("invalid Binance ticker {field}='{value}': {e}"))
    })
}

/// Converts a Binance kline interval to a Nautilus `BarSpecification`.
fn interval_to_bar_spec(interval: BinanceKlineInterval) -> BarSpecification {
    match interval {
        BinanceKlineInterval::Second1 => {
            BarSpecification::new(1, BarAggregation::Second, PriceType::Last)
        }
        BinanceKlineInterval::Minute1 => {
            BarSpecification::new(1, BarAggregation::Minute, PriceType::Last)
        }
        BinanceKlineInterval::Minute3 => {
            BarSpecification::new(3, BarAggregation::Minute, PriceType::Last)
        }
        BinanceKlineInterval::Minute5 => {
            BarSpecification::new(5, BarAggregation::Minute, PriceType::Last)
        }
        BinanceKlineInterval::Minute15 => {
            BarSpecification::new(15, BarAggregation::Minute, PriceType::Last)
        }
        BinanceKlineInterval::Minute30 => {
            BarSpecification::new(30, BarAggregation::Minute, PriceType::Last)
        }
        BinanceKlineInterval::Hour1 => {
            BarSpecification::new(1, BarAggregation::Hour, PriceType::Last)
        }
        BinanceKlineInterval::Hour2 => {
            BarSpecification::new(2, BarAggregation::Hour, PriceType::Last)
        }
        BinanceKlineInterval::Hour4 => {
            BarSpecification::new(4, BarAggregation::Hour, PriceType::Last)
        }
        BinanceKlineInterval::Hour6 => {
            BarSpecification::new(6, BarAggregation::Hour, PriceType::Last)
        }
        BinanceKlineInterval::Hour8 => {
            BarSpecification::new(8, BarAggregation::Hour, PriceType::Last)
        }
        BinanceKlineInterval::Hour12 => {
            BarSpecification::new(12, BarAggregation::Hour, PriceType::Last)
        }
        BinanceKlineInterval::Day1 => {
            BarSpecification::new(1, BarAggregation::Day, PriceType::Last)
        }
        BinanceKlineInterval::Day3 => {
            BarSpecification::new(3, BarAggregation::Day, PriceType::Last)
        }
        BinanceKlineInterval::Week1 => {
            BarSpecification::new(1, BarAggregation::Week, PriceType::Last)
        }
        BinanceKlineInterval::Month1 => {
            BarSpecification::new(1, BarAggregation::Month, PriceType::Last)
        }
    }
}

/// Parses a kline message into a `Bar`.
///
/// Returns `None` if the kline is not closed yet.
///
/// # Errors
///
/// Returns an error if parsing fails.
pub fn parse_kline(
    msg: &BinanceFuturesKlineMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> BinanceWsResult<Option<BinanceBar>> {
    // Only emit bars when the kline is closed
    if !msg.kline.is_closed {
        return Ok(None);
    }

    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let spec = interval_to_bar_spec(msg.kline.interval);
    let bar_type = BarType::new(instrument_id, spec, AggregationSource::External);

    let price = |field: &str, value: &str| {
        parse_required_price_at_precision(value, price_precision, field)
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))
    };
    let quantity = |field: &str, value: &str| {
        parse_required_quantity_at_precision(value, size_precision, field)
            .map_err(|e| BinanceWsError::ParseError(e.to_string()))
    };
    let decimal = |field: &str, value: &str| {
        Decimal::from_str(value)
            .map_err(|e| BinanceWsError::ParseError(format!("invalid {field} `{value}`: {e}")))
    };
    let count = u64::try_from(msg.kline.num_trades)
        .map_err(|e| BinanceWsError::ParseError(e.to_string()))?;

    // Use the kline close time as the event timestamp
    let ts_event = parse_millis_or_init(msg.kline.close_time, "Futures kline close time", ts_init);

    let bar = BinanceBar::new(
        bar_type,
        price("open", &msg.kline.open)?,
        price("high", &msg.kline.high)?,
        price("low", &msg.kline.low)?,
        price("close", &msg.kline.close)?,
        quantity("volume", &msg.kline.volume)?,
        decimal("quote volume", &msg.kline.quote_volume)?,
        count,
        decimal("taker buy base volume", &msg.kline.taker_buy_volume)?,
        decimal("taker buy quote volume", &msg.kline.taker_buy_quote_volume)?,
        ts_event,
        ts_init,
    );

    Ok(Some(bar))
}

/// Extracts the symbol from a raw JSON message.
pub fn extract_symbol(json: &serde_json::Value) -> Option<Ustr> {
    json.get("s").and_then(|v| v.as_str()).map(Ustr::from)
}

/// Extracts the event type from a raw JSON message.
pub fn extract_event_type(json: &serde_json::Value) -> Option<BinanceWsEventType> {
    json.get("e")
        .and_then(|v| serde_json::from_value(v.clone()).ok())
}

#[cfg(test)]
mod tests {
    use nautilus_model::{enums::BookType, orderbook::OrderBook, types::Quantity};
    use rstest::rstest;
    use rust_decimal_macros::dec;
    use serde::de::DeserializeOwned;
    use serde_json::json;

    use super::*;
    use crate::{
        common::{
            enums::{BinanceOrderStatus, BinanceSide, BinanceTradingStatus},
            parse::parse_usdm_instrument,
            testing::{load_fixture_string, load_json_fixture},
        },
        futures::{
            http::models::BinanceFuturesUsdSymbol,
            websocket::streams::messages::{BinanceFuturesLiquidationMsg, BinanceFuturesTickerMsg},
        },
    };

    const PRICE_PRECISION: u8 = 8;
    const SIZE_PRECISION: u8 = 3;

    fn sample_futures_symbol() -> BinanceFuturesUsdSymbol {
        BinanceFuturesUsdSymbol {
            symbol: Ustr::from("BTCUSDT"),
            pair: Ustr::from("BTCUSDT"),
            contract_type: "PERPETUAL".to_string(),
            delivery_date: 4_133_404_800_000,
            onboard_date: 1_569_398_400_000,
            status: BinanceTradingStatus::Trading,
            maint_margin_percent: "2.5000".to_string(),
            required_margin_percent: "5.0000".to_string(),
            base_asset: Ustr::from("BTC"),
            quote_asset: Ustr::from("USDT"),
            margin_asset: Ustr::from("USDT"),
            price_precision: PRICE_PRECISION as i32,
            quantity_precision: SIZE_PRECISION as i32,
            base_asset_precision: 8,
            quote_precision: 8,
            underlying_type: Some("COIN".to_string()),
            underlying_sub_type: vec!["PoW".to_string()],
            settle_plan: None,
            trigger_protect: Some("0.0500".to_string()),
            liquidation_fee: Some("0.012500".to_string()),
            market_take_bound: Some("0.05".to_string()),
            order_types: vec!["LIMIT".to_string(), "MARKET".to_string()],
            time_in_force: vec!["GTC".to_string(), "IOC".to_string()],
            filters: vec![
                json!({
                    "filterType": "PRICE_FILTER",
                    "tickSize": "0.00000001",
                    "maxPrice": "1000000",
                    "minPrice": "0.00000001"
                }),
                json!({
                    "filterType": "LOT_SIZE",
                    "stepSize": "0.001",
                    "maxQty": "1000",
                    "minQty": "0.001"
                }),
            ],
        }
    }

    fn sample_instrument() -> InstrumentAny {
        let ts = UnixNanos::from(1_700_000_000_000_000_000u64);
        parse_usdm_instrument(&sample_futures_symbol(), ts, ts).unwrap()
    }

    fn load_market_fixture<T: DeserializeOwned>(filename: &str) -> T {
        let path = format!("futures/market_data_json/{filename}");
        serde_json::from_str(&load_fixture_string(&path))
            .unwrap_or_else(|e| panic!("Failed to parse fixture {path}: {e}"))
    }

    #[rstest]
    fn test_parse_agg_trade() {
        let instrument = sample_instrument();
        let msg: BinanceFuturesAggTradeMsg = load_market_fixture("agg_trade_stream.json");
        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);

        let trade = parse_agg_trade(&msg, &instrument, ts_init).unwrap();

        assert_eq!(trade.instrument_id, instrument.id());
        assert_eq!(trade.price, Price::new(0.001, PRICE_PRECISION));
        assert_eq!(trade.size, Quantity::new(100.0, SIZE_PRECISION));
        assert_eq!(trade.aggressor_side, AggressorSide::Sell);
        assert_eq!(trade.trade_id, TradeId::new("5933014"));
        assert_eq!(trade.ts_event, UnixNanos::from(123_456_785_000_000u64));
        assert_eq!(trade.ts_init, ts_init);
    }

    #[rstest]
    #[case::negative(-1)]
    #[case::overflow(i64::MAX)]
    fn test_parse_agg_trade_falls_back_for_invalid_timestamp(#[case] trade_time: i64) {
        let instrument = sample_instrument();
        let mut msg: BinanceFuturesAggTradeMsg = load_market_fixture("agg_trade_stream.json");
        msg.trade_time = trade_time;

        let ts_init = UnixNanos::from(1);
        let trade = parse_agg_trade(&msg, &instrument, ts_init).unwrap();

        assert_eq!(trade.ts_event, ts_init);
        assert_eq!(trade.ts_init, ts_init);
    }

    #[rstest]
    fn test_parse_trade() {
        let instrument = sample_instrument();
        let msg: BinanceFuturesTradeMsg = load_market_fixture("trade_stream.json");
        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);

        let trade = parse_trade(&msg, &instrument, ts_init).unwrap();

        assert_eq!(trade.instrument_id, instrument.id());
        assert_eq!(trade.price, Price::new(0.001, PRICE_PRECISION));
        assert_eq!(trade.size, Quantity::new(100.0, SIZE_PRECISION));
        assert_eq!(trade.aggressor_side, AggressorSide::Sell);
        assert_eq!(trade.trade_id, TradeId::new("5933014"));
        assert_eq!(trade.ts_event, UnixNanos::from(123_456_785_000_000u64));
        assert_eq!(trade.ts_init, ts_init);
    }

    #[rstest]
    fn test_parse_book_ticker() {
        let instrument = sample_instrument();
        let msg: BinanceFuturesBookTickerMsg = load_market_fixture("book_ticker_stream.json");
        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);

        let quote = parse_book_ticker(&msg, &instrument, ts_init).unwrap();

        assert_eq!(quote.instrument_id, instrument.id());
        assert_eq!(quote.bid_price, Price::new(25.3519, PRICE_PRECISION));
        assert_eq!(quote.ask_price, Price::new(25.3652, PRICE_PRECISION));
        assert_eq!(quote.bid_size, Quantity::new(31.21, SIZE_PRECISION));
        assert_eq!(quote.ask_size, Quantity::new(40.66, SIZE_PRECISION));
        assert_eq!(
            quote.ts_event,
            UnixNanos::from(1_568_014_460_891_000_000u64)
        );
        assert_eq!(quote.ts_init, ts_init);
    }

    #[rstest]
    fn test_parse_depth_update() {
        let instrument = sample_instrument();
        let msg: BinanceFuturesDepthUpdateMsg = load_market_fixture("depth_update_stream.json");
        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);

        let deltas = parse_depth_update(&msg, &instrument, ts_init).unwrap();

        assert_eq!(deltas.instrument_id, instrument.id());
        assert_eq!(deltas.deltas.len(), 2);
        assert_eq!(deltas.sequence, 160);
        assert_eq!(deltas.ts_event, UnixNanos::from(123_456_788_000_000u64));
        assert_eq!(deltas.ts_init, ts_init);
        assert_eq!(deltas.deltas[0].action, BookAction::Update);
        assert_eq!(deltas.deltas[0].order.side, OrderSide::Buy.into());
        assert_eq!(
            deltas.deltas[0].order.price,
            Price::new(0.0024, PRICE_PRECISION)
        );
        assert_eq!(
            deltas.deltas[0].order.size,
            Quantity::new(10.0, SIZE_PRECISION)
        );
        assert_eq!(deltas.deltas[1].action, BookAction::Update);
        assert_eq!(deltas.deltas[1].order.side, OrderSide::Sell.into());
        assert_eq!(
            deltas.deltas[1].order.price,
            Price::new(0.0026, PRICE_PRECISION)
        );
        assert_eq!(
            deltas.deltas[1].order.size,
            Quantity::new(100.0, SIZE_PRECISION)
        );
        assert_eq!(deltas.deltas[1].flags, RecordFlag::F_LAST as u8);
    }

    #[rstest]
    #[case::five(5)]
    #[case::ten(10)]
    #[case::twenty(20)]
    fn test_parse_depth_snapshot_replaces_levels(#[case] depth: usize) {
        let instrument = sample_instrument();
        let mut msg: BinanceFuturesDepthUpdateMsg = load_market_fixture("depth_update_stream.json");
        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);
        let mut book = OrderBook::new(instrument.id(), BookType::L2_MBP);

        for offset in [0, 100] {
            msg.bids = (0..depth)
                .map(|i| [(1000 - offset - i).to_string(), "2.000".into()])
                .collect();
            msg.asks = (0..depth)
                .map(|i| [(2000 + offset + i).to_string(), "3.000".into()])
                .collect();
            let snapshot = parse_depth_snapshot(&msg, &instrument, ts_init).unwrap();
            let mut expected = vec![OrderBookDelta::clear(
                instrument.id(),
                msg.final_update_id,
                snapshot.ts_event,
                ts_init,
            )];

            for (side, levels) in [(OrderSide::Buy, &msg.bids), (OrderSide::Sell, &msg.asks)] {
                for level in levels {
                    expected.push(OrderBookDelta::new(
                        instrument.id(),
                        BookAction::Add,
                        BookOrder::new(
                            side,
                            Price::from_str(&format!("{}.00000000", level[0])).unwrap(),
                            Quantity::from_str(&level[1]).unwrap(),
                            0,
                        ),
                        0,
                        msg.final_update_id,
                        UnixNanos::from(123_456_788_000_000u64),
                        ts_init,
                    ));
                }
            }
            expected.last_mut().unwrap().flags = RecordFlag::F_LAST as u8;
            assert_eq!(snapshot.instrument_id, instrument.id());
            assert_eq!(snapshot.sequence, msg.final_update_id);
            assert_eq!(snapshot.deltas, expected);
            book.apply_deltas(&snapshot).unwrap();
            assert_eq!(book.bids(None).count(), depth);
            assert_eq!(book.asks(None).count(), depth);
            for (actual, input) in book.bids(None).zip(&msg.bids) {
                assert_eq!(
                    actual.price.value.as_decimal(),
                    Decimal::from_str(&input[0]).unwrap()
                );
                assert_eq!(actual.size_decimal(), dec!(2));
            }

            for (actual, input) in book.asks(None).zip(&msg.asks) {
                assert_eq!(
                    actual.price.value.as_decimal(),
                    Decimal::from_str(&input[0]).unwrap()
                );
                assert_eq!(actual.size_decimal(), dec!(3));
            }
        }

        msg.bids.clear();
        msg.asks.clear();
        let empty = parse_depth_snapshot(&msg, &instrument, ts_init).unwrap();
        book.apply_deltas(&empty).unwrap();
        assert_eq!(empty.deltas.len(), 1);
        assert_eq!(
            empty.deltas[0].flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
        assert_eq!(book.bids(None).count(), 0);
        assert_eq!(book.asks(None).count(), 0);
    }

    #[rstest]
    fn test_market_data_parsers_preserve_decimal_prices() {
        let instrument = sample_instrument();
        let ts_init = UnixNanos::from(1);
        let price = "123456789.12345678";
        let mut agg: BinanceFuturesAggTradeMsg = load_market_fixture("agg_trade_stream.json");
        agg.price = price.to_string();
        let mut trade: BinanceFuturesTradeMsg = load_market_fixture("trade_stream.json");
        trade.price = price.to_string();
        let mut book: BinanceFuturesBookTickerMsg = load_market_fixture("book_ticker_stream.json");
        book.best_bid_price = price.to_string();
        book.best_ask_price = "123456789.87654321".to_string();
        let mut depth: BinanceFuturesDepthUpdateMsg =
            load_market_fixture("depth_update_stream.json");
        depth.bids[0][0] = price.to_string();
        depth.asks[0][0] = book.best_ask_price.clone();
        depth.asks[0][1] = "0.000".to_string();

        let agg = parse_agg_trade(&agg, &instrument, ts_init).unwrap();
        let trade = parse_trade(&trade, &instrument, ts_init).unwrap();
        let book = parse_book_ticker(&book, &instrument, ts_init).unwrap();
        let depth = parse_depth_update(&depth, &instrument, ts_init).unwrap();

        assert_eq!(agg.price, Price::from(price));
        assert_eq!(trade.price, Price::from(price));
        assert_eq!(book.bid_price, Price::from(price));
        assert_eq!(book.ask_price, Price::from("123456789.87654321"));
        assert_eq!(depth.deltas[0].order.price, Price::from(price));
        assert_eq!(
            depth.deltas[1].order.price,
            Price::from("123456789.87654321")
        );
        assert_eq!(depth.deltas[1].order.size, Quantity::from("0.000"));
        assert_eq!(depth.deltas[1].action, BookAction::Delete);
    }

    #[rstest]
    fn test_parse_mark_price() {
        let instrument = sample_instrument();
        let msg: BinanceFuturesMarkPriceMsg = load_market_fixture("mark_price_stream.json");
        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);

        let (mark, index, funding, custom) = parse_mark_price(&msg, &instrument, ts_init).unwrap();

        assert_eq!(mark.instrument_id, instrument.id());
        assert_eq!(mark.value, Price::new(11794.15, PRICE_PRECISION));
        assert_eq!(index.value, Price::new(11784.62659091, PRICE_PRECISION));
        assert_eq!(mark.ts_event, UnixNanos::from(1_562_305_380_000_000_000u64));
        assert_eq!(funding.instrument_id, instrument.id());
        assert_eq!(funding.rate.to_string(), "0.00038167");
        assert_eq!(
            funding.next_funding_ns,
            Some(UnixNanos::from(1_562_306_400_000_000_000u64))
        );
        assert_eq!(
            funding.ts_event,
            UnixNanos::from(1_562_305_380_000_000_000u64)
        );
        assert_eq!(funding.ts_init, ts_init);
        assert_eq!(custom.instrument_id, instrument.id());
        assert_eq!(custom.mark_price, Price::from("11794.15000000"));
        assert_eq!(custom.index_price, Price::from("11784.62659091"));
        assert_eq!(custom.estimated_settle_price, Price::from("11784.25641265"));
        assert_eq!(custom.funding_rate, dec!(0.00038167));
        assert_eq!(custom.next_funding_time, funding.next_funding_ns);
        assert_eq!(custom.ts_event, mark.ts_event);
        assert_eq!(custom.ts_init, ts_init);
    }

    #[rstest]
    #[case::zero(0)]
    #[case::negative(-1)]
    fn test_parse_mark_price_preserves_missing_funding_time(#[case] next_funding_time: i64) {
        let instrument = sample_instrument();
        let mut msg: BinanceFuturesMarkPriceMsg = load_market_fixture("mark_price_stream.json");
        msg.next_funding_time = next_funding_time;

        let (_, _, funding, custom) =
            parse_mark_price(&msg, &instrument, UnixNanos::from(1)).unwrap();

        assert_eq!(funding.next_funding_ns, None);
        assert_eq!(custom.next_funding_time, None);
    }

    #[rstest]
    fn test_parse_kline_closed() {
        let instrument = sample_instrument();
        let msg: BinanceFuturesKlineMsg = load_market_fixture("kline_stream_closed.json");
        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);

        let bar = parse_kline(&msg, &instrument, ts_init).unwrap().unwrap();

        assert_eq!(bar.bar_type.instrument_id(), instrument.id());
        assert_eq!(bar.open, Price::new(0.001, PRICE_PRECISION));
        assert_eq!(bar.high, Price::new(0.0025, PRICE_PRECISION));
        assert_eq!(bar.low, Price::new(0.001, PRICE_PRECISION));
        assert_eq!(bar.close, Price::new(0.002, PRICE_PRECISION));
        assert_eq!(bar.volume, Quantity::new(1000.0, SIZE_PRECISION));
        assert_eq!(bar.quote_volume, dec!(1.0000));
        assert_eq!(bar.count, 100);
        assert_eq!(bar.taker_buy_base_volume, dec!(500));
        assert_eq!(bar.taker_buy_quote_volume, dec!(0.500));
        assert_eq!(bar.ts_event, UnixNanos::from(1_638_747_719_999_000_000u64));
        assert_eq!(bar.ts_init, ts_init);
    }

    #[rstest]
    fn test_parse_kline_open_returns_none() {
        let instrument = sample_instrument();
        let msg: BinanceFuturesKlineMsg = load_market_fixture("kline_stream_open.json");

        let bar = parse_kline(&msg, &instrument, UnixNanos::default()).unwrap();

        assert!(bar.is_none());
    }

    #[rstest]
    fn test_mark_price_msg_deserializes_optional_ap() {
        let json = r#"{
            "e": "markPriceUpdate",
            "E": 1562305380000,
            "s": "BTCUSDT",
            "p": "11794.15000000",
            "ap": "11792.85000000",
            "i": "11784.62659091",
            "P": "11784.25641265",
            "r": "0.00038167",
            "T": 1562306400000
        }"#;

        let msg: BinanceFuturesMarkPriceMsg = serde_json::from_str(json).unwrap();
        assert_eq!(msg.mark_price_moving_avg.as_deref(), Some("11792.85000000"));

        let legacy: BinanceFuturesMarkPriceMsg = load_market_fixture("mark_price_stream.json");
        assert!(legacy.mark_price_moving_avg.is_none());
    }

    #[rstest]
    fn test_parse_mark_price_funding_rate_fields() {
        let instrument = sample_instrument();
        let msg: BinanceFuturesMarkPriceMsg = load_market_fixture("mark_price_stream.json");
        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);

        let (_mark, _index, funding, _custom) =
            parse_mark_price(&msg, &instrument, ts_init).unwrap();

        assert_eq!(funding.instrument_id, instrument.id());
        assert_eq!(funding.rate.to_string(), "0.00038167");
        assert!(funding.interval.is_none());
        assert_eq!(
            funding.next_funding_ns,
            Some(UnixNanos::from(1_562_306_400_000_000_000u64))
        );
        assert_eq!(
            funding.ts_event,
            UnixNanos::from(1_562_305_380_000_000_000u64)
        );
        assert_eq!(funding.ts_init, ts_init);
    }

    #[rstest]
    fn test_deserialize_liquidation_msg() {
        let msg: BinanceFuturesLiquidationMsg = load_market_fixture("liquidation_stream.json");

        assert_eq!(msg.event_type, "forceOrder");
        assert_eq!(msg.event_time, 1_568_014_460_893);
        assert_eq!(msg.order.symbol, Ustr::from("BTCUSDT"));
        assert_eq!(msg.order.side, BinanceSide::Sell);
        assert_eq!(msg.order.original_qty, "0.014");
        assert_eq!(msg.order.average_price, "9910.12345678");
        assert_eq!(msg.order.status, BinanceOrderStatus::Filled);
        assert_eq!(msg.order.accumulated_qty, "0.014");
        assert_eq!(msg.order.trade_time, 1_568_014_460_893);
    }

    #[rstest]
    fn test_deserialize_ticker_msg() {
        let msg: BinanceFuturesTickerMsg = load_market_fixture("ticker_stream.json");

        assert_eq!(msg.event_type, "24hrTicker");
        assert_eq!(msg.symbol, Ustr::from("BTCUSDT"));
        assert_eq!(msg.price_change, "-131.40000000");
        assert_eq!(msg.price_change_percent, "-0.786");
        assert_eq!(msg.weighted_avg_price, "16628.97377498");
        assert_eq!(msg.last_price, "16584.60000000");
        assert_eq!(msg.open_price, "16716.00000000");
        assert_eq!(msg.high_price, "16764.89000000");
        assert_eq!(msg.low_price, "16456.51000000");
        assert_eq!(msg.volume, "122474.816");
        assert_eq!(msg.quote_volume, "2036102085.69746400");
        assert_eq!(msg.num_trades, 142853);
    }

    #[rstest]
    fn test_parse_ticker() {
        let instrument = sample_instrument();
        let msg: BinanceFuturesTickerMsg = load_market_fixture("ticker_stream.json");
        let ts_init = UnixNanos::from(1_700_000_001_000_000_000u64);

        let ticker = parse_ticker(&msg, &instrument, ts_init).unwrap();

        assert_eq!(ticker.instrument_id, instrument.id());
        assert_eq!(ticker.price_change, dec!(-131.40000000));
        assert_eq!(ticker.price_change_percent, dec!(-0.786));
        assert_eq!(ticker.weighted_avg_price, dec!(16628.97377498));
        assert_eq!(ticker.last_price, dec!(16584.60000000));
        assert_eq!(ticker.last_qty, dec!(0.002));
        assert_eq!(ticker.open_price, dec!(16716.00000000));
        assert_eq!(ticker.high_price, dec!(16764.89000000));
        assert_eq!(ticker.low_price, dec!(16456.51000000));
        assert_eq!(ticker.volume, dec!(122474.816));
        assert_eq!(ticker.quote_volume, dec!(2036102085.69746400));
        assert_eq!(ticker.open_time, UnixNanos::from_millis(1_672_429_382_136));
        assert_eq!(ticker.close_time, UnixNanos::from_millis(1_672_515_782_136));
        assert_eq!(ticker.first_trade_id, 2_289_691);
        assert_eq!(ticker.last_trade_id, 2_432_543);
        assert_eq!(ticker.num_trades, 142_853);
        assert_eq!(ticker.ts_event, UnixNanos::from_millis(1_672_515_782_136));
        assert_eq!(ticker.ts_init, ts_init);
    }

    #[rstest]
    fn test_parse_ticker_rejects_invalid_numeric_field() {
        let instrument = sample_instrument();
        let mut msg: BinanceFuturesTickerMsg = load_market_fixture("ticker_stream.json");
        msg.last_price = "not-a-decimal".to_string();

        let result = parse_ticker(&msg, &instrument, UnixNanos::default());

        assert!(result.is_err());
    }

    #[rstest]
    fn test_extract_symbol() {
        let json = load_json_fixture("futures/market_data_json/book_ticker_stream.json");

        let symbol = extract_symbol(&json);

        assert_eq!(symbol, Some(Ustr::from("BNBUSDT")));
    }

    #[rstest]
    fn test_extract_event_type() {
        let json = load_json_fixture("futures/market_data_json/mark_price_stream.json");

        let event_type = extract_event_type(&json);

        assert_eq!(event_type, Some(BinanceWsEventType::MarkPriceUpdate));
    }

    #[rstest]
    fn test_extract_event_type_force_order() {
        let json = load_json_fixture("futures/market_data_json/liquidation_stream.json");

        let event_type = extract_event_type(&json);

        assert_eq!(event_type, Some(BinanceWsEventType::ForceOrder));
    }

    #[rstest]
    fn test_extract_event_type_ticker() {
        let json = load_json_fixture("futures/market_data_json/ticker_stream.json");

        let event_type = extract_event_type(&json);

        assert_eq!(event_type, Some(BinanceWsEventType::Ticker24Hr));
    }
}