nautilus_deribit/websocket/

parse.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 functions for converting Deribit WebSocket messages to Nautilus domain types.

use ahash::AHashMap;
use anyhow::Context;
use jiff::{Span, Timestamp, tz::Offset};
use nautilus_core::{
    UUID4, UnixNanos, datetime::NANOSECONDS_IN_MILLISECOND,
    serialization::deserialize_decimal_token_borrowed,
};
use nautilus_model::{
    data::{
        Bar, BarType, BookOrder, Data, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate,
        OrderBookDelta, OrderBookDeltas, QuoteTick, TradeTick, bar::BarSpecification,
        option_chain::OptionGreeks,
    },
    enums::{
        AggregationSource, AggressorSide, BarAggregation, BookAction, GreeksConvention,
        LiquiditySide, OrderSide, OrderStatus, OrderType, PositionSide, PriceType, RecordFlag,
        TimeInForce,
    },
    events::{OrderAccepted, OrderCanceled, OrderExpired, OrderUpdated},
    identifiers::{
        AccountId, ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, VenueOrderId,
    },
    instruments::{Instrument, InstrumentAny},
    reports::{FillReport, OrderStatusReport, PositionStatusReport},
    types::{Currency, Money, Price, Quantity},
};
use rust_decimal::{Decimal, prelude::ToPrimitive};
use serde_json::value::RawValue;
use ustr::Ustr;

use super::{
    enums::{DeribitBookAction, DeribitBookMsgType},
    messages::{
        DeribitBookMsg, DeribitChartMsg, DeribitOrderMsg, DeribitPerpetualMsg, DeribitQuoteMsg,
        DeribitTickerMsg, DeribitTradeMsg, DeribitUserTradeMsg,
    },
};
use crate::{common::parse::build_public_trade_id, http::models::DeribitPosition};

fn next_8_utc(from_ns: UnixNanos) -> anyhow::Result<UnixNanos> {
    let from_secs = from_ns.as_u64() / 1_000_000_000;
    let timestamp = Timestamp::from_second(i64::try_from(from_secs)?)?;
    let dt = Offset::UTC.to_datetime(timestamp);
    let date = if dt.hour() < 8 {
        dt.date()
    } else {
        dt.date().checked_add(Span::new().days(1))?
    };
    let next_8 = Offset::UTC.to_timestamp(date.at(8, 0, 0, 0))?;
    let nanos = u64::try_from(next_8.as_nanosecond())
        .context("GTD expiry timestamp out of UnixNanos range")?;
    Ok(UnixNanos::from(nanos))
}

/// Parses a Deribit trade message into a Nautilus `TradeTick`.
///
/// # Errors
///
/// Returns an error if the trade cannot be parsed.
pub fn parse_trade_msg(
    msg: &DeribitTradeMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<TradeTick> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let price = Price::from_decimal_dp(msg.price, price_precision)?;
    let size = Quantity::from_decimal_dp(msg.amount.abs(), size_precision)?;

    let aggressor_side = match msg.direction.as_str() {
        "buy" => AggressorSide::Buy,
        "sell" => AggressorSide::Sell,
        _ => AggressorSide::NoAggressor,
    };

    let trade_id = build_public_trade_id(
        &msg.trade_id,
        msg.block_rfq_id,
        msg.block_trade_id.as_deref(),
        msg.combo_id.as_deref(),
    );
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

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

/// Parses a vector of Deribit trade messages into Nautilus `Data` items.
pub fn parse_trades_data(
    trades: &[DeribitTradeMsg],
    instruments_cache: &AHashMap<Ustr, InstrumentAny>,
    ts_init: UnixNanos,
) -> Vec<Data> {
    trades
        .iter()
        .filter_map(|msg| {
            instruments_cache
                .get(&msg.instrument_name)
                .and_then(|inst| parse_trade_msg(msg, inst, ts_init).ok())
                .map(Data::Trade)
        })
        .collect()
}

fn parse_snapshot_level(
    level: &[&RawValue],
    index: usize,
    side: &str,
    instrument_name: &str,
) -> Option<(Decimal, Decimal)> {
    let (price_val, amount_val) = if level.len() >= 3 {
        let price = deserialize_decimal_token_borrowed(level[1])
            .ok()
            .or_else(|| {
                log::warn!(
                    "Failed to parse {side} price at index {index} for {instrument_name}: {level:?}"
                );
                None
            })?;

        let amount = deserialize_decimal_token_borrowed(level[2]).ok().or_else(|| {
            log::warn!(
                "Failed to parse {side} amount at index {index} for {instrument_name}: {level:?}"
            );
            None
        })?;
        (price, amount)
    } else if level.len() >= 2 {
        let price = deserialize_decimal_token_borrowed(level[0])
            .ok()
            .or_else(|| {
                log::warn!(
                    "Failed to parse {side} price at index {index} for {instrument_name}: {level:?}"
                );
                None
            })?;

        let amount = deserialize_decimal_token_borrowed(level[1]).ok().or_else(|| {
            log::warn!(
                "Failed to parse {side} amount at index {index} for {instrument_name}: {level:?}"
            );
            None
        })?;
        (price, amount)
    } else {
        log::warn!(
            "Invalid {side} format at index {index} for {instrument_name}: expected 2-3 elements, was {}",
            level.len()
        );
        return None;
    };

    if price_val <= Decimal::ZERO {
        log::warn!(
            "Invalid {side} price {price_val} at index {index} for {instrument_name}: {level:?}"
        );
        return None;
    }

    Some((price_val, amount_val))
}

fn parse_delta_level(
    level: &[&RawValue],
    index: usize,
    side: &str,
    instrument_name: &str,
) -> Option<(BookAction, Decimal, Decimal)> {
    if level.len() < 3 {
        log::warn!(
            "Invalid {side} delta format at index {index} for {instrument_name}: expected 3 elements, was {}",
            level.len()
        );
        return None;
    }

    let action_str = serde_json::from_str::<&str>(level[0].get())
        .ok()
        .or_else(|| {
            log::warn!(
                "Failed to parse {side} action at index {index} for {instrument_name}: {level:?}"
            );
            None
        })?;

    let deribit_action: DeribitBookAction = action_str.parse().ok().or_else(|| {
        log::warn!(
            "Unknown {side} action '{action_str}' at index {index} for {instrument_name}: {level:?}"
        );
        None
    })?;

    let price_val = deserialize_decimal_token_borrowed(level[1])
        .ok()
        .or_else(|| {
            log::warn!(
                "Failed to parse {side} price at index {index} for {instrument_name}: {level:?}"
            );
            None
        })?;

    let amount_val = deserialize_decimal_token_borrowed(level[2])
        .ok()
        .or_else(|| {
            log::warn!(
                "Failed to parse {side} amount at index {index} for {instrument_name}: {level:?}"
            );
            None
        })?;

    if price_val <= Decimal::ZERO {
        log::warn!(
            "Invalid {side} price {price_val} at index {index} for {instrument_name}: {level:?}"
        );
        return None;
    }

    Some((deribit_action.into(), price_val, amount_val))
}

/// Parses a Deribit order book snapshot into Nautilus `OrderBookDeltas`.
///
/// Skips a malformed level, and a level whose amount is not positive or rounds to zero at the
/// instrument's size precision.
///
/// # Errors
///
/// Returns an error if a level's price or amount cannot convert to `Price` or `Quantity`, such as a
/// value outside their range.
pub fn parse_book_snapshot(
    msg: &DeribitBookMsg<'_>,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderBookDeltas> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

    let mut deltas = Vec::new();

    let has_levels = !msg.bids.is_empty() || !msg.asks.is_empty();

    // All snapshot deltas get F_SNAPSHOT; CLEAR also gets F_LAST if no levels follow
    let clear_flags = if has_levels {
        RecordFlag::F_SNAPSHOT as u8
    } else {
        RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
    };

    deltas.push(OrderBookDelta::new(
        instrument_id,
        BookAction::Clear,
        BookOrder::default(),
        clear_flags,
        msg.change_id,
        ts_event,
        ts_init,
    ));

    let sides = [
        (OrderSide::Buy, "bid", &msg.bids, 0),
        (OrderSide::Sell, "ask", &msg.asks, msg.bids.len()),
    ];

    for (order_side, side, levels, order_id_offset) in sides {
        for (i, level) in levels.iter().enumerate() {
            let Some((price_val, amount_val)) =
                parse_snapshot_level(level, i, side, msg.instrument_name.as_str())
            else {
                continue;
            };

            if amount_val <= Decimal::ZERO {
                continue;
            }

            let price = Price::from_decimal_dp(price_val, price_precision)?;
            let size = Quantity::from_decimal_dp(amount_val, size_precision)?;

            if is_zero_at_size_precision(size, amount_val, side, i, msg.instrument_name.as_str()) {
                continue;
            }

            deltas.push(OrderBookDelta::new_checked(
                instrument_id,
                BookAction::Add,
                BookOrder::new(order_side, price, size, (order_id_offset + i) as u64),
                RecordFlag::F_SNAPSHOT as u8,
                msg.change_id,
                ts_event,
                ts_init,
            )?);
        }
    }

    if let Some(last) = deltas.last_mut() {
        *last = OrderBookDelta::new(
            last.instrument_id,
            last.action,
            last.order,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8,
            last.sequence,
            last.ts_event,
            last.ts_init,
        );
    }

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

/// Parses a Deribit order book change (delta) into Nautilus `OrderBookDeltas`.
///
/// Skips a malformed level. A `new` or `change` level whose amount rounds to zero at the
/// instrument's size precision becomes a delete.
///
/// # Errors
///
/// Returns an error if a level's price or amount cannot convert to `Price` or `Quantity`, such as a
/// value outside their range.
pub fn parse_book_delta(
    msg: &DeribitBookMsg<'_>,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderBookDeltas> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

    let mut deltas = Vec::new();

    let sides = [
        (OrderSide::Buy, "bid", &msg.bids, 0),
        (OrderSide::Sell, "ask", &msg.asks, msg.bids.len()),
    ];

    for (order_side, side, levels, order_id_offset) in sides {
        for (i, level) in levels.iter().enumerate() {
            let Some((action, price_val, amount_val)) =
                parse_delta_level(level, i, side, msg.instrument_name.as_str())
            else {
                continue;
            };

            let price = Price::from_decimal_dp(price_val, price_precision)?;
            let size = Quantity::from_decimal_dp(amount_val.abs(), size_precision)?;

            let action = if action != BookAction::Delete
                && is_zero_at_size_precision(
                    size,
                    amount_val,
                    side,
                    i,
                    msg.instrument_name.as_str(),
                ) {
                BookAction::Delete
            } else {
                action
            };

            deltas.push(OrderBookDelta::new_checked(
                instrument_id,
                action,
                BookOrder::new(order_side, price, size, (order_id_offset + i) as u64),
                0,
                msg.change_id,
                ts_event,
                ts_init,
            )?);
        }
    }

    // Set F_LAST flag on the last delta
    if let Some(last) = deltas.last_mut() {
        *last = OrderBookDelta::new(
            last.instrument_id,
            last.action,
            last.order,
            RecordFlag::F_LAST as u8,
            last.sequence,
            last.ts_event,
            last.ts_init,
        );
    }

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

/// Parses a Deribit order book message (snapshot or delta) into Nautilus `OrderBookDeltas`.
///
/// # Errors
///
/// Returns an error if a level's price or amount cannot convert to `Price` or `Quantity`, such as a
/// value outside their range.
pub fn parse_book_msg(
    msg: &DeribitBookMsg<'_>,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderBookDeltas> {
    match msg.msg_type {
        DeribitBookMsgType::Snapshot => parse_book_snapshot(msg, instrument, ts_init),
        DeribitBookMsgType::Change => parse_book_delta(msg, instrument, ts_init),
    }
}

// An amount below half the size increment rounds to zero, which a book level cannot hold, so
// callers treat the level as absent.
fn is_zero_at_size_precision(
    size: Quantity,
    amount: Decimal,
    side: &str,
    index: usize,
    instrument_name: &str,
) -> bool {
    if size.is_positive() {
        return false;
    }

    log::warn!(
        "Book {side} amount {amount} at index {index} for {instrument_name} is zero at size precision {}, treating level as absent",
        size.precision
    );
    true
}

/// Parses a Deribit ticker message into a Nautilus `QuoteTick`.
///
/// # Errors
///
/// Returns an error if the quote cannot be parsed or prices are missing.
pub fn parse_ticker_to_quote(
    msg: &DeribitTickerMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<QuoteTick> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let bid_price_val = msg
        .best_bid_price
        .context("Missing best_bid_price in ticker")?;
    let ask_price_val = msg
        .best_ask_price
        .context("Missing best_ask_price in ticker")?;

    let bid_price = Price::from_decimal_dp(bid_price_val, price_precision)?;
    let ask_price = Price::from_decimal_dp(ask_price_val, price_precision)?;
    let bid_size =
        Quantity::from_decimal_dp(msg.best_bid_amount.unwrap_or_default(), size_precision)?;
    let ask_size =
        Quantity::from_decimal_dp(msg.best_ask_amount.unwrap_or_default(), size_precision)?;
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

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

/// Parses a Deribit quote message into a Nautilus `QuoteTick`.
///
/// # Errors
///
/// Returns an error if the quote cannot be parsed.
pub fn parse_quote_msg(
    msg: &DeribitQuoteMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<QuoteTick> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let bid_price = Price::from_decimal_dp(msg.best_bid_price, price_precision)?;
    let ask_price = Price::from_decimal_dp(msg.best_ask_price, price_precision)?;
    let bid_size = Quantity::from_decimal_dp(msg.best_bid_amount, size_precision)?;
    let ask_size = Quantity::from_decimal_dp(msg.best_ask_amount, size_precision)?;
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

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

/// Parses a Deribit ticker message into a Nautilus `MarkPriceUpdate`.
///
/// # Errors
///
/// Returns an error if the price cannot be converted to the required precision.
pub fn parse_ticker_to_mark_price(
    msg: &DeribitTickerMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<MarkPriceUpdate> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let value = Price::from_decimal_dp(msg.mark_price, price_precision)?;
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

    Ok(MarkPriceUpdate::new(
        instrument_id,
        value,
        ts_event,
        ts_init,
    ))
}

/// Parses a Deribit ticker message into a Nautilus `IndexPriceUpdate`.
///
/// # Errors
///
/// Returns an error if the price cannot be converted to the required precision.
pub fn parse_ticker_to_index_price(
    msg: &DeribitTickerMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<IndexPriceUpdate> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let value = Price::from_decimal_dp(msg.index_price, price_precision)?;
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

    Ok(IndexPriceUpdate::new(
        instrument_id,
        value,
        ts_event,
        ts_init,
    ))
}

/// Parses a Deribit ticker message into a Nautilus `FundingRateUpdate`.
///
/// Returns `None` if the instrument is not a perpetual or the funding rate is not available.
#[must_use]
pub fn parse_ticker_to_funding_rate(
    msg: &DeribitTickerMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> Option<FundingRateUpdate> {
    // current_funding is only available for perpetual instruments
    let rate = msg.current_funding?;
    let instrument_id = instrument.id();
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

    // Deribit ticker doesn't include next_funding_time, set to None
    Some(FundingRateUpdate::new(
        instrument_id,
        rate,
        None, // Deribit exchanges funding every few seconds, instead of in set intervals like other exchanges
        None, // next_funding_ns not available in ticker
        ts_event,
        ts_init,
    ))
}

/// Parses a Deribit ticker message into a Nautilus `OptionGreeks`.
///
/// Returns `None` if the ticker message does not contain Greeks (non-option instrument).
#[must_use]
pub fn parse_ticker_to_option_greeks(
    msg: &DeribitTickerMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> Option<OptionGreeks> {
    let deribit_greeks = msg.greeks.as_ref()?;
    let instrument_id = instrument.id();
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

    Some(OptionGreeks {
        instrument_id,
        convention: GreeksConvention::BlackScholes,
        greeks: deribit_greeks.to_greek_values(),
        mark_iv: msg.mark_iv.and_then(|v| v.to_f64()),
        bid_iv: msg.bid_iv.and_then(|v| v.to_f64()),
        ask_iv: msg.ask_iv.and_then(|v| v.to_f64()),
        underlying_price: msg.underlying_price.and_then(|v| v.to_f64()),
        open_interest: Some(msg.open_interest.to_f64().unwrap_or(0.0)),
        ts_event,
        ts_init,
    })
}

/// Parses a Deribit perpetual channel message into a Nautilus `FundingRateUpdate`.
///
/// The perpetual channel (`perpetual.{instrument}.{interval}`) provides dedicated
/// funding rate updates with the `interest` field representing the current funding rate.
#[must_use]
pub fn parse_perpetual_to_funding_rate(
    msg: &DeribitPerpetualMsg,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> FundingRateUpdate {
    let instrument_id = instrument.id();
    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

    FundingRateUpdate::new(
        instrument_id,
        msg.interest,
        None, // Deribit exchanges funding every few seconds, instead of in set intervals like other exchanges
        None, // next_funding_ns not available in perpetual channel
        ts_event,
        ts_init,
    )
}

/// Converts a Deribit chart resolution and instrument to a Nautilus BarType.
///
/// Deribit resolutions: "1", "3", "5", "10", "15", "30", "60", "120", "180", "360", "720", "1D"
///
/// # Errors
///
/// Returns an error if the resolution string is invalid or BarType construction fails.
pub fn resolution_to_bar_type(
    instrument_id: InstrumentId,
    resolution: &str,
) -> anyhow::Result<BarType> {
    let (step, aggregation) = match resolution {
        "1" => (1, BarAggregation::Minute),
        "3" => (3, BarAggregation::Minute),
        "5" => (5, BarAggregation::Minute),
        "10" => (10, BarAggregation::Minute),
        "15" => (15, BarAggregation::Minute),
        "30" => (30, BarAggregation::Minute),
        "60" => (1, BarAggregation::Hour),
        "120" => (2, BarAggregation::Hour),
        "180" => (3, BarAggregation::Hour),
        "360" => (6, BarAggregation::Hour),
        "720" => (12, BarAggregation::Hour),
        "1D" => (1, BarAggregation::Day),
        _ => anyhow::bail!("Unsupported Deribit resolution: {resolution}"),
    };

    let spec = BarSpecification::new_checked(step, aggregation, PriceType::Last)
        .context("invalid Deribit bar resolution")?;
    Ok(BarType::new(
        instrument_id,
        spec,
        AggregationSource::External,
    ))
}

/// Parses a Deribit chart message from a WebSocket subscription into a [`Bar`].
///
/// Converts a single OHLCV data point from the `chart.trades.{instrument}.{resolution}` channel
/// into a Nautilus Bar object.
///
/// When `use_cost_for_volume` is true, `Bar.volume` is populated from `chart_msg.cost` (USD) to
/// match instruments whose trade `amount` is in USD (inverse perpetuals / inverse futures).
/// Otherwise `chart_msg.volume` (base currency) is used. Callers derive this from the instrument
/// via [`crate::common::parse::use_cost_for_bar_volume`].
///
/// # Errors
///
/// Returns an error if:
/// - Price or volume values are invalid.
/// - Bar construction fails validation.
pub fn parse_chart_msg(
    chart_msg: &DeribitChartMsg,
    bar_type: BarType,
    price_precision: u8,
    size_precision: u8,
    use_cost_for_volume: bool,
    timestamp_on_close: bool,
    ts_init: UnixNanos,
) -> anyhow::Result<Bar> {
    let open =
        Price::from_decimal_dp(chart_msg.open, price_precision).context("Invalid open price")?;
    let high =
        Price::from_decimal_dp(chart_msg.high, price_precision).context("Invalid high price")?;
    let low =
        Price::from_decimal_dp(chart_msg.low, price_precision).context("Invalid low price")?;
    let close =
        Price::from_decimal_dp(chart_msg.close, price_precision).context("Invalid close price")?;

    let raw_volume = if use_cost_for_volume {
        chart_msg.cost
    } else {
        chart_msg.volume
    };

    let volume = Quantity::from_decimal_dp(raw_volume, size_precision).context("Invalid volume")?;

    // Convert timestamp from milliseconds to nanoseconds
    let mut ts_event = UnixNanos::from(chart_msg.tick * NANOSECONDS_IN_MILLISECOND);

    // Adjust timestamp to close time if configured
    if timestamp_on_close {
        let interval_ns = bar_type.spec().timedelta().as_nanos();
        let interval_ns = u64::try_from(interval_ns)
            .context("bar interval overflowed the u64 range for nanoseconds")?;
        let updated = ts_event
            .as_u64()
            .checked_add(interval_ns)
            .context("bar timestamp overflowed when adjusting to close time")?;
        ts_event = UnixNanos::from(updated);
    }

    Bar::new_checked(bar_type, open, high, low, close, volume, ts_event, ts_init)
        .context("Invalid OHLC bar")
}

/// Parses a Deribit user order message into a Nautilus `OrderStatusReport`.
///
/// # Errors
///
/// Returns an error if the order data cannot be parsed.
pub fn parse_user_order_msg(
    msg: &DeribitOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderStatusReport> {
    let instrument_id = instrument.id();
    let venue_order_id = VenueOrderId::new(&msg.order_id);

    let order_side = match msg.direction.as_str() {
        "buy" => OrderSide::Buy,
        "sell" => OrderSide::Sell,
        _ => anyhow::bail!("Unknown order direction: {}", msg.direction),
    };

    // Map Deribit order type to Nautilus
    let order_type = parse_deribit_order_type(&msg.order_type);

    // Deribit supports: good_til_cancelled, good_til_day, fill_or_kill, immediate_or_cancel
    let time_in_force = match msg.time_in_force.as_str() {
        "good_til_cancelled" => TimeInForce::Gtc,
        "good_til_day" => TimeInForce::Gtd,
        "fill_or_kill" => TimeInForce::Fok,
        "immediate_or_cancel" => TimeInForce::Ioc,
        other => {
            log::warn!("Unknown time_in_force '{other}', defaulting to GTC");
            TimeInForce::Gtc
        }
    };

    // Map Deribit order state to Nautilus status
    let order_status = match msg.order_state.as_str() {
        "open" => {
            if msg.filled_amount.is_zero() {
                OrderStatus::Accepted
            } else {
                OrderStatus::PartiallyFilled
            }
        }
        "filled" => OrderStatus::Filled,
        "rejected" => OrderStatus::Rejected,
        "cancelled" => OrderStatus::Canceled,
        "untriggered" => OrderStatus::Accepted, // Pending trigger
        other => {
            log::warn!("Unknown Deribit order_state '{other}', defaulting to Accepted");
            OrderStatus::Accepted
        }
    };

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

    let quantity = Quantity::from_decimal_dp(msg.amount, size_precision)?;
    let filled_qty = Quantity::from_decimal_dp(msg.filled_amount, size_precision)?;

    let ts_accepted = UnixNanos::new(msg.creation_timestamp * NANOSECONDS_IN_MILLISECOND);
    let ts_last = UnixNanos::new(msg.last_update_timestamp * NANOSECONDS_IN_MILLISECOND);

    let mut report = OrderStatusReport::new(
        account_id,
        instrument_id,
        None, // order_list_id
        venue_order_id,
        order_side.into(),
        order_type,
        time_in_force,
        order_status,
        quantity,
        filled_qty,
        ts_accepted,
        ts_last,
        ts_init,
        Some(UUID4::new()),
    );

    // Add client order ID if present
    if let Some(ref label) = msg.label
        && !label.is_empty()
    {
        report = report.with_client_order_id(ClientOrderId::new(label));
    }

    // Add price for limit orders
    if let Some(price_val) = msg.price
        && !price_val.is_zero()
    {
        let price = Price::from_decimal_dp(price_val, price_precision)?;
        report = report.with_price(price);
    }

    if time_in_force == TimeInForce::Gtd {
        let expire_time = next_8_utc(ts_accepted)?;
        report = report.with_expire_time(expire_time);
    }

    // Add average price if filled
    if let Some(avg_price) = msg.average_price
        && !avg_price.is_zero()
    {
        report = report.with_avg_px(avg_price);
    }

    // Add trigger price for stop/take orders
    if let Some(trigger_price) = msg.trigger_price
        && !trigger_price.is_zero()
    {
        let trigger = Price::from_decimal_dp(trigger_price, price_precision)?;
        report = report.with_trigger_price(trigger);
    }

    if msg.post_only {
        report = report.with_post_only(true);
    }

    if msg.reduce_only {
        report = report.with_reduce_only(true);
    }

    // Add cancel/reject reason
    if let Some(ref reason) = msg.reject_reason {
        report = report.with_cancel_reason(reason.clone());
    } else if let Some(ref reason) = msg.cancel_reason {
        report = report.with_cancel_reason(reason.clone());
    }

    Ok(report)
}

#[must_use]
pub(crate) fn parse_deribit_order_type(order_type: &str) -> OrderType {
    match order_type {
        "limit" => OrderType::Limit,
        "market" => OrderType::Market,
        "stop_limit" => OrderType::StopLimit,
        "stop_market" => OrderType::StopMarket,
        "take_limit" => OrderType::LimitIfTouched,
        "take_market" => OrderType::MarketIfTouched,
        other => {
            log::warn!("Unknown Deribit order_type '{other}', defaulting to Limit");
            OrderType::Limit
        }
    }
}

/// Parses a Deribit user trade message into a Nautilus `FillReport`.
///
/// # Errors
///
/// Returns an error if the trade data cannot be parsed.
pub fn parse_user_trade_msg(
    msg: &DeribitUserTradeMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    ts_init: UnixNanos,
) -> anyhow::Result<FillReport> {
    let instrument_id = instrument.id();
    let venue_order_id = VenueOrderId::new(&msg.order_id);
    let trade_id = TradeId::new(&msg.trade_id);

    // Deribit marks liquidation-triggered trades with "M" (maker liquidated),
    // "T" (taker liquidated), or "MT" (both). Absent means a normal trade.
    if let Some(liq) = msg.liquidation.as_deref().filter(|s| !s.is_empty()) {
        let who = match liq {
            "M" => "maker",
            "T" => "taker",
            "MT" => "both",
            _ => liq,
        };
        log::warn!(
            "Liquidation trade: {} trade_id={} order_id={} liquidation_side={} direction={} amount={} price={}",
            instrument_id,
            msg.trade_id,
            msg.order_id,
            who,
            msg.direction,
            msg.amount,
            msg.price,
        );
    }

    let order_side = match msg.direction.as_str() {
        "buy" => OrderSide::Buy,
        "sell" => OrderSide::Sell,
        _ => anyhow::bail!("Unknown trade direction: {}", msg.direction),
    };

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

    let last_qty = Quantity::from_decimal_dp(msg.amount, size_precision)?;
    let last_px = Price::from_decimal_dp(msg.price, price_precision)?;

    let liquidity_side = match msg.liquidity.as_str() {
        "M" => LiquiditySide::Maker,
        "T" => LiquiditySide::Taker,
        _ => LiquiditySide::NoLiquiditySide,
    };

    // Get fee currency from the fee_currency field
    let fee_currency = Currency::from(&msg.fee_currency);
    let commission = Money::from_decimal(msg.fee, fee_currency)?;

    let ts_event = UnixNanos::new(msg.timestamp * NANOSECONDS_IN_MILLISECOND);

    let client_order_id = msg
        .label
        .as_ref()
        .filter(|l| !l.is_empty())
        .map(ClientOrderId::new);

    Ok(FillReport::new(
        account_id,
        instrument_id,
        venue_order_id,
        trade_id,
        order_side,
        last_qty,
        last_px,
        commission,
        liquidity_side,
        client_order_id,
        None, // venue_position_id
        ts_event,
        ts_init,
        None, // report_id
    ))
}

/// Parses a Deribit position into a Nautilus `PositionStatusReport`.
///
/// # Arguments
/// - `position` - The Deribit position data from `/private/get_positions`
/// - `instrument` - The corresponding Nautilus instrument
/// - `account_id` - The account ID for the report
/// - `ts_init` - Initialization timestamp
///
/// # Returns
/// A `PositionStatusReport` representing the current position state.
#[must_use]
pub fn parse_position_status_report(
    position: &DeribitPosition,
    instrument: &InstrumentAny,
    account_id: AccountId,
    ts_init: UnixNanos,
) -> PositionStatusReport {
    let instrument_id = instrument.id();
    let size_precision = instrument.size_precision();

    let signed_qty = Quantity::from_decimal_dp(position.size.abs(), size_precision)
        .unwrap_or_else(|_| Quantity::zero(size_precision));

    let position_side = match position.direction.as_str() {
        "buy" => PositionSide::Long,
        "sell" => PositionSide::Short,
        _ => PositionSide::Flat,
    };

    // Use average_price directly as it's already a Decimal
    let avg_px_open = Some(position.average_price);

    PositionStatusReport::new(
        account_id,
        instrument_id,
        position_side,
        signed_qty,
        ts_init,
        ts_init,
        Some(UUID4::new()),
        None, // venue_position_id
        avg_px_open,
    )
}

/// Parsed order event result from a Deribit order message.
///
/// This enum represents the discrete order events that can be derived from
/// Deribit order state transitions.
#[derive(Debug, Clone)]
pub enum ParsedOrderEvent {
    /// Order was accepted by the venue.
    Accepted(OrderAccepted),
    /// Order was canceled.
    Canceled(OrderCanceled),
    /// Order expired.
    Expired(OrderExpired),
    /// Order was updated (amended).
    Updated(OrderUpdated),
    /// No event to emit (e.g., already processed or intermediate state).
    None,
}

/// Extracts the client order ID from a Deribit order message label.
fn extract_client_order_id(msg: &DeribitOrderMsg) -> Option<ClientOrderId> {
    msg.label
        .as_ref()
        .filter(|l| !l.is_empty())
        .map(ClientOrderId::new)
}

/// Parses a Deribit order message into an `OrderAccepted` event.
///
/// This should be called when an order transitions to "open" state for the first time
/// or when a buy/sell response is received successfully.
#[must_use]
pub fn parse_order_accepted(
    msg: &DeribitOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    ts_init: UnixNanos,
) -> OrderAccepted {
    let client_order_id =
        extract_client_order_id(msg).unwrap_or_else(|| ClientOrderId::new(&msg.order_id));
    parse_order_accepted_with_client_order_id(
        msg,
        instrument,
        account_id,
        trader_id,
        strategy_id,
        client_order_id,
        ts_init,
    )
}

#[must_use]
pub(crate) fn parse_order_accepted_with_client_order_id(
    msg: &DeribitOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    client_order_id: ClientOrderId,
    ts_init: UnixNanos,
) -> OrderAccepted {
    let instrument_id = instrument.id();
    let venue_order_id = VenueOrderId::new(&msg.order_id);
    let ts_event = UnixNanos::new(msg.last_update_timestamp * NANOSECONDS_IN_MILLISECOND);

    OrderAccepted::new(
        trader_id,
        strategy_id,
        instrument_id,
        client_order_id,
        venue_order_id,
        account_id,
        nautilus_core::UUID4::new(),
        ts_event,
        ts_init,
        false, // reconciliation
    )
}

/// Parses a Deribit order message into an `OrderCanceled` event.
///
/// This should be called when an order transitions to "cancelled" state.
#[must_use]
pub fn parse_order_canceled(
    msg: &DeribitOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    ts_init: UnixNanos,
) -> OrderCanceled {
    let client_order_id =
        extract_client_order_id(msg).unwrap_or_else(|| ClientOrderId::new(&msg.order_id));
    parse_order_canceled_with_client_order_id(
        msg,
        instrument,
        account_id,
        trader_id,
        strategy_id,
        client_order_id,
        ts_init,
    )
}

#[must_use]
pub(crate) fn parse_order_canceled_with_client_order_id(
    msg: &DeribitOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    client_order_id: ClientOrderId,
    ts_init: UnixNanos,
) -> OrderCanceled {
    let instrument_id = instrument.id();
    let venue_order_id = VenueOrderId::new(&msg.order_id);
    let ts_event = UnixNanos::new(msg.last_update_timestamp * NANOSECONDS_IN_MILLISECOND);

    OrderCanceled::new(
        trader_id,
        strategy_id,
        instrument_id,
        client_order_id,
        nautilus_core::UUID4::new(),
        ts_event,
        ts_init,
        false, // reconciliation
        Some(venue_order_id),
        Some(account_id),
        msg.cancel_reason.as_deref().map(Ustr::from),
    )
}

/// Parses a Deribit order message into an `OrderExpired` event.
///
/// This should be called when an order transitions to "expired" state
/// (e.g., GTD orders that reached their expiry time).
#[must_use]
pub fn parse_order_expired(
    msg: &DeribitOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    ts_init: UnixNanos,
) -> OrderExpired {
    let client_order_id =
        extract_client_order_id(msg).unwrap_or_else(|| ClientOrderId::new(&msg.order_id));
    parse_order_expired_with_client_order_id(
        msg,
        instrument,
        account_id,
        trader_id,
        strategy_id,
        client_order_id,
        ts_init,
    )
}

#[must_use]
pub(crate) fn parse_order_expired_with_client_order_id(
    msg: &DeribitOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    client_order_id: ClientOrderId,
    ts_init: UnixNanos,
) -> OrderExpired {
    let instrument_id = instrument.id();
    let venue_order_id = VenueOrderId::new(&msg.order_id);
    let ts_event = UnixNanos::new(msg.last_update_timestamp * NANOSECONDS_IN_MILLISECOND);

    OrderExpired::new(
        trader_id,
        strategy_id,
        instrument_id,
        client_order_id,
        nautilus_core::UUID4::new(),
        ts_event,
        ts_init,
        false, // reconciliation
        Some(venue_order_id),
        Some(account_id),
    )
}

/// Parses a Deribit order message into an `OrderUpdated` event.
///
/// This should be called when an order is amended (price or quantity changed).
#[must_use]
pub fn parse_order_updated(
    msg: &DeribitOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    ts_init: UnixNanos,
) -> OrderUpdated {
    let client_order_id =
        extract_client_order_id(msg).unwrap_or_else(|| ClientOrderId::new(&msg.order_id));
    parse_order_updated_with_client_order_id(
        msg,
        instrument,
        account_id,
        trader_id,
        strategy_id,
        client_order_id,
        ts_init,
    )
}

#[must_use]
pub(crate) fn parse_order_updated_with_client_order_id(
    msg: &DeribitOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    client_order_id: ClientOrderId,
    ts_init: UnixNanos,
) -> OrderUpdated {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let venue_order_id = VenueOrderId::new(&msg.order_id);
    let quantity = Quantity::from_decimal_dp(msg.amount, size_precision)
        .unwrap_or_else(|_| Quantity::zero(size_precision));
    let price = msg
        .price
        .and_then(|p| Price::from_decimal_dp(p, price_precision).ok());
    let trigger_price = msg
        .trigger_price
        .and_then(|p| Price::from_decimal_dp(p, price_precision).ok());
    let ts_event = UnixNanos::new(msg.last_update_timestamp * NANOSECONDS_IN_MILLISECOND);

    OrderUpdated::new(
        trader_id,
        strategy_id,
        instrument_id,
        client_order_id,
        quantity,
        nautilus_core::UUID4::new(),
        ts_event,
        ts_init,
        false, // reconciliation
        Some(venue_order_id),
        Some(account_id),
        price,
        trigger_price,
        None,  // protection_price
        false, // is_quote_quantity
    )
}

/// Determines the appropriate order event based on the Deribit order state.
///
/// This function analyzes the order state and returns the corresponding event type.
/// It's used by the handler to determine which event to emit for a given order update.
///
/// # Arguments
/// - `order_state` - The Deribit order state string ("open", "filled", "cancelled", etc.)
/// - `is_new_order` - Whether this is the first time we're seeing this order
/// - `was_amended` - Whether this update is due to an amendment (edit) operation
///
/// # Returns
/// The type of event that should be emitted, or `None` if no event should be emitted.
#[must_use]
pub fn determine_order_event_type(
    order_state: &str,
    is_new_order: bool,
    was_amended: bool,
) -> OrderEventType {
    match order_state {
        "open" | "untriggered" => {
            if was_amended {
                OrderEventType::Updated
            } else if is_new_order {
                OrderEventType::Accepted
            } else {
                // Order is still open, no event needed (partial fill handled separately)
                OrderEventType::None
            }
        }
        "cancelled" => OrderEventType::Canceled,
        "expired" => OrderEventType::Expired,
        "filled" => {
            // Fills are handled through the user.trades channel
            OrderEventType::None
        }
        "rejected" => {
            // Rejections are handled separately via OrderRejected
            OrderEventType::None
        }
        other => {
            log::warn!("Unknown Deribit order_state '{other}' in event routing, dropping");
            OrderEventType::None
        }
    }
}

/// Order event type to be emitted.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OrderEventType {
    /// Emit OrderAccepted event.
    Accepted,
    /// Emit OrderCanceled event.
    Canceled,
    /// Emit OrderExpired event.
    Expired,
    /// Emit OrderUpdated event.
    Updated,
    /// No event to emit.
    None,
}

#[cfg(test)]
mod tests {
    use rstest::rstest;
    use rust_decimal_macros::dec;
    use serde::de::DeserializeOwned;

    use super::*;
    use crate::{
        common::{
            parse::{parse_deribit_instrument_any, parse_portfolio_to_account_state},
            testing::load_test_json,
        },
        http::models::{DeribitInstrument, DeribitJsonRpcResponse},
        websocket::messages::{DeribitPortfolioMsg, DeribitWsMessage, parse_raw_message},
    };

    /// Creates a BTC-PERPETUAL test instrument.
    fn test_perpetual_instrument() -> InstrumentAny {
        let json = load_test_json("http_get_instruments.json");
        let response: DeribitJsonRpcResponse<Vec<DeribitInstrument>> =
            serde_json::from_str(&json).unwrap();
        let instrument = &response.result.unwrap()[0];
        parse_deribit_instrument_any(instrument, UnixNanos::default(), UnixNanos::default())
            .unwrap()
            .unwrap()
    }

    #[rstest]
    fn test_parse_trade_msg_sell() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_trades.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let trades: Vec<DeribitTradeMsg> =
            serde_json::from_str(&response["params"]["data"].to_string()).unwrap();
        let msg = &trades[0];

        let tick = parse_trade_msg(msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(tick.instrument_id, instrument.id());
        assert_eq!(tick.price, instrument.make_price(92294.5));
        assert_eq!(tick.size, instrument.make_qty(10.0, None));
        assert_eq!(tick.aggressor_side, AggressorSide::Sell);
        assert_eq!(tick.trade_id.to_string(), "403691824");
        assert_eq!(tick.ts_event, UnixNanos::new(1_765_531_356_452_000_000));
    }

    #[rstest]
    fn test_parse_trade_msg_buy() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_trades.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let trades: Vec<DeribitTradeMsg> =
            serde_json::from_str(&response["params"]["data"].to_string()).unwrap();
        let msg = &trades[1];

        let tick = parse_trade_msg(msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(tick.instrument_id, instrument.id());
        assert_eq!(tick.price, instrument.make_price(92288.5));
        assert_eq!(tick.size, instrument.make_qty(750.0, None));
        assert_eq!(tick.aggressor_side, AggressorSide::Sell);
        assert_eq!(tick.trade_id.to_string(), "403691825");
    }

    fn make_trade_msg(
        instrument_name: &str,
        trade_id: &str,
        block_trade_id: Option<&str>,
        block_rfq_id: Option<i64>,
        combo_id: Option<&str>,
    ) -> DeribitTradeMsg {
        let raw = serde_json::json!({
            "trade_id": trade_id,
            "instrument_name": instrument_name,
            "price": 92294.5,
            "amount": 10.0,
            "direction": "buy",
            "timestamp": 1_765_531_356_452_u64,
            "trade_seq": 1,
            "tick_direction": 0,
            "index_price": 92276.75,
            "mark_price": 92287.11,
            "block_trade_id": block_trade_id,
            "block_rfq_id": block_rfq_id,
            "combo_id": combo_id,
        });
        serde_json::from_str(&raw.to_string()).unwrap()
    }

    #[rstest]
    fn test_parse_trade_msg_tags_block_trade() {
        let instrument = test_perpetual_instrument();
        let msg = make_trade_msg("BTC-PERPETUAL", "244343055", Some("12345"), None, None);
        let tick = parse_trade_msg(&msg, &instrument, UnixNanos::default()).unwrap();
        assert_eq!(tick.trade_id.to_string(), "BLK-244343055");
    }

    #[rstest]
    fn test_parse_trade_msg_tags_block_rfq() {
        let instrument = test_perpetual_instrument();
        let msg = make_trade_msg("BTC-PERPETUAL", "244343055", None, Some(99), None);
        let tick = parse_trade_msg(&msg, &instrument, UnixNanos::default()).unwrap();
        assert_eq!(tick.trade_id.to_string(), "RFQ-244343055");
    }

    #[rstest]
    fn test_parse_trade_msg_tags_combo_leg() {
        // Per-leg trade originating from a combo: combo_id is set by Deribit
        // even though the instrument is a plain perp / option, so downstream
        // sees `COMBO-` and can detect combo-origin fills.
        let instrument = test_perpetual_instrument();
        let msg = make_trade_msg(
            "BTC-PERPETUAL",
            "244343055",
            None,
            None,
            Some("BTC-FS-25DEC26_PERP"),
        );
        let tick = parse_trade_msg(&msg, &instrument, UnixNanos::default()).unwrap();
        assert_eq!(tick.trade_id.to_string(), "COMBO-244343055");
    }

    fn load_combo_option_instrument() -> InstrumentAny {
        let combo_json = load_test_json("http_get_instruments_option_combo.json");
        let combo_response: DeribitJsonRpcResponse<Vec<DeribitInstrument>> =
            serde_json::from_str(&combo_json).unwrap();
        let combo_raw = combo_response
            .result
            .unwrap()
            .into_iter()
            .find(|i| i.instrument_name == "BTC-CS-19MAY26-70000_75000")
            .expect("fixture must contain BTC-CS-19MAY26-70000_75000");
        parse_deribit_instrument_any(&combo_raw, UnixNanos::default(), UnixNanos::default())
            .unwrap()
            .unwrap()
    }

    fn load_combo_trade_msgs() -> Vec<DeribitTradeMsg> {
        let json = load_test_json("ws_trades_option_combo.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        serde_json::from_str(&response["params"]["data"].to_string()).unwrap()
    }

    #[rstest]
    fn test_parse_trades_data_combo_emits_single_tick() {
        // Step 1 finding: combo legs already publish on their own per-leg
        // streams. Parsing a combo trade message should produce exactly one
        // TradeTick (for the combo), not N+1.
        let combo_inst = load_combo_option_instrument();
        let mut cache: AHashMap<Ustr, InstrumentAny> = AHashMap::new();
        cache.insert(Ustr::from("BTC-CS-19MAY26-70000_75000"), combo_inst.clone());

        let trades = load_combo_trade_msgs();
        // Sanity: the combo message itself carries the legs array.
        let legs = trades[0].legs.as_ref().expect("combo trade must have legs");
        assert_eq!(legs.len(), 2);

        let data = parse_trades_data(&trades, &cache, UnixNanos::default());
        assert_eq!(data.len(), 1, "should emit one tick for the combo only");

        let Data::Trade(tick) = &data[0] else {
            panic!("expected Data::Trade");
        };
        assert_eq!(tick.instrument_id, combo_inst.id());
        // Exact values from fixture; independent of the instrument under test
        // so a regression in price/size precision is caught here too.
        assert_eq!(tick.price, Price::from("0.0639"));
        assert_eq!(tick.size, Quantity::from("0.1"));
        assert_eq!(tick.aggressor_side, AggressorSide::Sell);
        assert_eq!(tick.trade_id.to_string(), "244365193");
    }

    #[rstest]
    fn test_parse_trades_data_combo_not_cached_emits_no_tick() {
        // When the combo InstrumentAny is not in the WS handler cache,
        // parse_trades_data must drop the message rather than panic or
        // synthesize a tick against an unknown instrument.
        let cache: AHashMap<Ustr, InstrumentAny> = AHashMap::new();
        let trades = load_combo_trade_msgs();
        let data = parse_trades_data(&trades, &cache, UnixNanos::default());
        assert!(data.is_empty(), "uncached combo must not emit ticks");
    }

    #[rstest]
    fn test_parse_trades_data_mixed_combo_and_per_leg() {
        // Combo trade and a plain per-leg trade on the underlying perpetual,
        // both cached, must each produce a tick against their own instrument.
        let combo_inst = load_combo_option_instrument();
        let perp_inst = test_perpetual_instrument();

        let mut cache: AHashMap<Ustr, InstrumentAny> = AHashMap::new();
        cache.insert(Ustr::from("BTC-CS-19MAY26-70000_75000"), combo_inst.clone());
        cache.insert(Ustr::from("BTC-PERPETUAL"), perp_inst.clone());

        let mut trades = load_combo_trade_msgs();
        let perp_msgs: Vec<DeribitTradeMsg> = {
            let json = load_test_json("ws_trades.json");
            let response: serde_json::Value = serde_json::from_str(&json).unwrap();
            serde_json::from_str(&response["params"]["data"].to_string()).unwrap()
        };
        trades.extend(perp_msgs);

        let data = parse_trades_data(&trades, &cache, UnixNanos::default());
        // 1 combo + 2 plain BTC-PERPETUAL trades from the existing fixture.
        assert_eq!(data.len(), 3);

        let mut combo_ticks = 0;
        let mut perp_ticks = 0;

        for item in &data {
            let Data::Trade(tick) = item else {
                panic!("expected Data::Trade, was {item:?}");
            };

            if tick.instrument_id == combo_inst.id() {
                combo_ticks += 1;
                assert_eq!(tick.trade_id.to_string(), "244365193");
            } else if tick.instrument_id == perp_inst.id() {
                perp_ticks += 1;
            } else {
                panic!("unexpected instrument_id: {}", tick.instrument_id);
            }
        }
        assert_eq!(combo_ticks, 1);
        assert_eq!(perp_ticks, 2);
    }

    #[rstest]
    fn test_parse_book_snapshot() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_book_snapshot.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let data = response["params"]["data"].to_string();
        let msg: DeribitBookMsg = serde_json::from_str(&data).unwrap();

        let deltas = parse_book_snapshot(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(deltas.instrument_id, instrument.id());
        // Should have CLEAR + 5 bids + 5 asks = 11 deltas
        assert_eq!(deltas.deltas.len(), 11);

        // First delta should be CLEAR
        assert_eq!(deltas.deltas[0].action, BookAction::Clear);

        // Check first bid
        let first_bid = &deltas.deltas[1];
        assert_eq!(first_bid.action, BookAction::Add);
        assert_eq!(first_bid.order.side, OrderSide::Buy.into());
        assert_eq!(first_bid.order.price, instrument.make_price(42500.0));
        assert_eq!(first_bid.order.size, instrument.make_qty(1000.0, None));

        // Check first ask
        let first_ask = &deltas.deltas[6];
        assert_eq!(first_ask.action, BookAction::Add);
        assert_eq!(first_ask.order.side, OrderSide::Sell.into());
        assert_eq!(first_ask.order.price, instrument.make_price(42501.0));
        assert_eq!(first_ask.order.size, instrument.make_qty(800.0, None));

        // Check F_LAST flag on last delta
        let last = deltas.deltas.last().unwrap();
        assert_eq!(
            last.flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8
        );
    }

    #[rstest]
    fn test_parse_book_delta() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_book_delta.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let data = response["params"]["data"].to_string();
        let msg: DeribitBookMsg = serde_json::from_str(&data).unwrap();

        let deltas = parse_book_delta(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(deltas.instrument_id, instrument.id());
        // Should have 2 bid deltas + 2 ask deltas = 4 deltas
        assert_eq!(deltas.deltas.len(), 4);

        // Check first bid - "change" action
        let bid_change = &deltas.deltas[0];
        assert_eq!(bid_change.action, BookAction::Update);
        assert_eq!(bid_change.order.side, OrderSide::Buy.into());
        assert_eq!(bid_change.order.price, instrument.make_price(42500.0));
        assert_eq!(bid_change.order.size, instrument.make_qty(950.0, None));

        // Check second bid - "new" action
        let bid_new = &deltas.deltas[1];
        assert_eq!(bid_new.action, BookAction::Add);
        assert_eq!(bid_new.order.side, OrderSide::Buy.into());
        assert_eq!(bid_new.order.price, instrument.make_price(42498.5));
        assert_eq!(bid_new.order.size, instrument.make_qty(300.0, None));

        // Check first ask - "delete" action
        let ask_delete = &deltas.deltas[2];
        assert_eq!(ask_delete.action, BookAction::Delete);
        assert_eq!(ask_delete.order.side, OrderSide::Sell.into());
        assert_eq!(ask_delete.order.price, instrument.make_price(42501.0));
        assert_eq!(ask_delete.order.size, instrument.make_qty(0.0, None));

        // Check second ask - "change" action
        let ask_change = &deltas.deltas[3];
        assert_eq!(ask_change.action, BookAction::Update);
        assert_eq!(ask_change.order.side, OrderSide::Sell.into());
        assert_eq!(ask_change.order.price, instrument.make_price(42501.5));
        assert_eq!(ask_change.order.size, instrument.make_qty(700.0, None));

        // Check F_LAST flag on last delta
        let last = deltas.deltas.last().unwrap();
        assert_eq!(
            last.flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8
        );
    }

    #[rstest]
    fn test_parse_ticker_to_quote() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_ticker.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let msg: DeribitTickerMsg =
            serde_json::from_str(&response["params"]["data"].to_string()).unwrap();

        // Verify the message was deserialized correctly
        assert_eq!(msg.instrument_name, "BTC-PERPETUAL");
        assert_eq!(msg.timestamp, 1_765_541_474_086);
        assert_eq!(msg.best_bid_price, Some(dec!(92283.5)));
        assert_eq!(msg.best_ask_price, Some(dec!(92284.0)));
        assert_eq!(msg.best_bid_amount, Some(dec!(117660.0)));
        assert_eq!(msg.best_ask_amount, Some(dec!(186520.0)));
        assert_eq!(msg.mark_price, dec!(92281.78));
        assert_eq!(msg.index_price, dec!(92263.55));
        assert_eq!(msg.open_interest, dec!(1132329370.0));

        let quote = parse_ticker_to_quote(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(quote.instrument_id, instrument.id());
        assert_eq!(quote.bid_price, instrument.make_price(92283.5));
        assert_eq!(quote.ask_price, instrument.make_price(92284.0));
        assert_eq!(quote.bid_size, instrument.make_qty(117660.0, None));
        assert_eq!(quote.ask_size, instrument.make_qty(186520.0, None));
        assert_eq!(quote.ts_event, UnixNanos::new(1_765_541_474_086_000_000));
    }

    #[rstest]
    fn test_parse_quote_msg() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_quote.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let msg: DeribitQuoteMsg =
            serde_json::from_str(&response["params"]["data"].to_string()).unwrap();

        // Verify the message was deserialized correctly
        assert_eq!(msg.instrument_name, "BTC-PERPETUAL");
        assert_eq!(msg.timestamp, 1_765_541_767_174);
        assert_eq!(msg.best_bid_price, dec!(92288.0));
        assert_eq!(msg.best_ask_price, dec!(92288.5));
        assert_eq!(msg.best_bid_amount, dec!(133440.0));
        assert_eq!(msg.best_ask_amount, dec!(99470.0));

        let quote = parse_quote_msg(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(quote.instrument_id, instrument.id());
        assert_eq!(quote.bid_price, instrument.make_price(92288.0));
        assert_eq!(quote.ask_price, instrument.make_price(92288.5));
        assert_eq!(quote.bid_size, instrument.make_qty(133440.0, None));
        assert_eq!(quote.ask_size, instrument.make_qty(99470.0, None));
        assert_eq!(quote.ts_event, UnixNanos::new(1_765_541_767_174_000_000));
    }

    #[rstest]
    fn test_parse_book_msg_snapshot() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_book_snapshot.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let data = response["params"]["data"].to_string();
        let msg: DeribitBookMsg = serde_json::from_str(&data).unwrap();

        // Validate raw message format - snapshots use 3-element arrays: ["new", price, amount]
        assert_eq!(
            msg.bids[0].len(),
            3,
            "Snapshot bids should have 3 elements: [action, price, amount]"
        );
        assert_eq!(
            msg.bids[0][0].get(),
            r#""new""#,
            "First element should be 'new' action for snapshot"
        );
        assert_eq!(
            msg.asks[0].len(),
            3,
            "Snapshot asks should have 3 elements: [action, price, amount]"
        );
        assert_eq!(
            msg.asks[0][0].get(),
            r#""new""#,
            "First element should be 'new' action for snapshot"
        );

        let deltas = parse_book_msg(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(deltas.instrument_id, instrument.id());
        // Should have CLEAR + 5 bids + 5 asks = 11 deltas
        assert_eq!(deltas.deltas.len(), 11);

        // First delta should be CLEAR
        assert_eq!(deltas.deltas[0].action, BookAction::Clear);

        // Verify first bid was parsed correctly from ["new", 42500.0, 1000.0]
        let first_bid = &deltas.deltas[1];
        assert_eq!(first_bid.action, BookAction::Add);
        assert_eq!(first_bid.order.side, OrderSide::Buy.into());
        assert_eq!(first_bid.order.price, instrument.make_price(42500.0));
        assert_eq!(first_bid.order.size, instrument.make_qty(1000.0, None));

        // Verify first ask was parsed correctly from ["new", 42501.0, 800.0]
        let first_ask = &deltas.deltas[6];
        assert_eq!(first_ask.action, BookAction::Add);
        assert_eq!(first_ask.order.side, OrderSide::Sell.into());
        assert_eq!(first_ask.order.price, instrument.make_price(42501.0));
        assert_eq!(first_ask.order.size, instrument.make_qty(800.0, None));
    }

    #[rstest]
    fn test_parse_book_msg_delta() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_book_delta.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let data = response["params"]["data"].to_string();
        let msg: DeribitBookMsg = serde_json::from_str(&data).unwrap();

        // Validate raw message format - deltas use 3-element arrays: [action, price, amount]
        assert_eq!(
            msg.bids[0].len(),
            3,
            "Delta bids should have 3 elements: [action, price, amount]"
        );
        assert_eq!(
            msg.bids[0][0].get(),
            r#""change""#,
            "First bid should be 'change' action"
        );
        assert_eq!(
            msg.bids[1][0].get(),
            r#""new""#,
            "Second bid should be 'new' action"
        );
        assert_eq!(
            msg.asks[0].len(),
            3,
            "Delta asks should have 3 elements: [action, price, amount]"
        );
        assert_eq!(
            msg.asks[0][0].get(),
            r#""delete""#,
            "First ask should be 'delete' action"
        );

        let deltas = parse_book_msg(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(deltas.instrument_id, instrument.id());
        // Should have 2 bid deltas + 2 ask deltas = 4 deltas
        assert_eq!(deltas.deltas.len(), 4);

        // Delta should not have CLEAR action
        assert_ne!(deltas.deltas[0].action, BookAction::Clear);

        // Verify first bid "change" action was parsed correctly from ["change", 42500.0, 950.0]
        let bid_change = &deltas.deltas[0];
        assert_eq!(bid_change.action, BookAction::Update);
        assert_eq!(bid_change.order.side, OrderSide::Buy.into());
        assert_eq!(bid_change.order.price, instrument.make_price(42500.0));
        assert_eq!(bid_change.order.size, instrument.make_qty(950.0, None));

        // Verify second bid "new" action was parsed correctly from ["new", 42498.5, 300.0]
        let bid_new = &deltas.deltas[1];
        assert_eq!(bid_new.action, BookAction::Add);
        assert_eq!(bid_new.order.side, OrderSide::Buy.into());
        assert_eq!(bid_new.order.price, instrument.make_price(42498.5));
        assert_eq!(bid_new.order.size, instrument.make_qty(300.0, None));

        // Verify first ask "delete" action was parsed correctly from ["delete", 42501.0, 0.0]
        let ask_delete = &deltas.deltas[2];
        assert_eq!(ask_delete.action, BookAction::Delete);
        assert_eq!(ask_delete.order.side, OrderSide::Sell.into());
        assert_eq!(ask_delete.order.price, instrument.make_price(42501.0));

        // Verify second ask "change" action was parsed correctly from ["change", 42501.5, 700.0]
        let ask_change = &deltas.deltas[3];
        assert_eq!(ask_change.action, BookAction::Update);
        assert_eq!(ask_change.order.side, OrderSide::Sell.into());
        assert_eq!(ask_change.order.price, instrument.make_price(42501.5));
        assert_eq!(ask_change.order.size, instrument.make_qty(700.0, None));
    }

    #[rstest]
    fn test_parse_book_grouped_snapshot() {
        // Test parsing grouped book channel format: book.{instrument}.{group}.{depth}.{interval}
        // This format has NO type field and uses 2-element arrays [price, amount]
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_book_grouped_snapshot.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let data = response["params"]["data"].to_string();
        let msg: DeribitBookMsg = serde_json::from_str(&data).unwrap();

        // Validate raw message format - grouped channel uses 2-element arrays: [price, amount]
        assert_eq!(
            msg.bids[0].len(),
            2,
            "Grouped bids should have 2 elements: [price, amount]"
        );
        assert_eq!(
            msg.asks[0].len(),
            2,
            "Grouped asks should have 2 elements: [price, amount]"
        );

        // Verify msg_type defaults to Snapshot (grouped channel has no type field)
        assert_eq!(
            msg.msg_type,
            DeribitBookMsgType::Snapshot,
            "Grouped channel should default to Snapshot type"
        );

        let deltas = parse_book_snapshot(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(deltas.instrument_id, instrument.id());
        // Should have CLEAR + 10 bids + 10 asks = 21 deltas
        assert_eq!(deltas.deltas.len(), 21);

        // First delta should be CLEAR
        assert_eq!(deltas.deltas[0].action, BookAction::Clear);

        // Verify first bid was parsed correctly from [89532.5, 254900.0]
        let first_bid = &deltas.deltas[1];
        assert_eq!(first_bid.action, BookAction::Add);
        assert_eq!(first_bid.order.side, OrderSide::Buy.into());
        assert_eq!(first_bid.order.price, instrument.make_price(89532.5));
        assert_eq!(first_bid.order.size, instrument.make_qty(254900.0, None));

        // Verify first ask was parsed correctly from [89533.0, 91570.0]
        let first_ask = &deltas.deltas[11];
        assert_eq!(first_ask.action, BookAction::Add);
        assert_eq!(first_ask.order.side, OrderSide::Sell.into());
        assert_eq!(first_ask.order.price, instrument.make_price(89533.0));
        assert_eq!(first_ask.order.size, instrument.make_qty(91570.0, None));

        // Check F_LAST flag on last delta
        let last = deltas.deltas.last().unwrap();
        assert_eq!(
            last.flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8
        );
    }

    #[rstest]
    fn test_parse_ticker_to_mark_price() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_ticker.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let msg: DeribitTickerMsg =
            serde_json::from_str(&response["params"]["data"].to_string()).unwrap();

        let mark_price =
            parse_ticker_to_mark_price(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(mark_price.instrument_id, instrument.id());
        assert_eq!(mark_price.value, instrument.make_price(92281.78));
        assert_eq!(
            mark_price.ts_event,
            UnixNanos::new(1_765_541_474_086_000_000)
        );
    }

    #[rstest]
    fn test_parse_ticker_to_index_price() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_ticker.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let msg: DeribitTickerMsg =
            serde_json::from_str(&response["params"]["data"].to_string()).unwrap();

        let index_price =
            parse_ticker_to_index_price(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(index_price.instrument_id, instrument.id());
        assert_eq!(index_price.value, instrument.make_price(92263.55));
        assert_eq!(
            index_price.ts_event,
            UnixNanos::new(1_765_541_474_086_000_000)
        );
    }

    #[rstest]
    fn test_parse_ticker_to_funding_rate() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_ticker.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let msg: DeribitTickerMsg =
            serde_json::from_str(&response["params"]["data"].to_string()).unwrap();

        // Verify current_funding exists in the message
        assert!(msg.current_funding.is_some());

        let funding_rate =
            parse_ticker_to_funding_rate(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(funding_rate.instrument_id, instrument.id());
        // The test fixture has current_funding value
        assert_eq!(
            funding_rate.ts_event,
            UnixNanos::new(1_765_541_474_086_000_000)
        );
        assert!(funding_rate.interval.is_none());
        assert!(funding_rate.next_funding_ns.is_none()); // Not available in ticker
    }

    #[rstest]
    fn test_resolution_to_bar_type_1_minute() {
        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
        let bar_type = resolution_to_bar_type(instrument_id, "1").unwrap();

        assert_eq!(bar_type.instrument_id(), instrument_id);
        assert_eq!(bar_type.spec().step.get(), 1);
        assert_eq!(bar_type.spec().aggregation, BarAggregation::Minute);
        assert_eq!(bar_type.spec().price_type, PriceType::Last);
        assert_eq!(bar_type.aggregation_source(), AggregationSource::External);
    }

    #[rstest]
    fn test_resolution_to_bar_type_60_minute() {
        let instrument_id = InstrumentId::from("ETH-PERPETUAL.DERIBIT");
        let bar_type = resolution_to_bar_type(instrument_id, "60").unwrap();

        assert_eq!(bar_type.instrument_id(), instrument_id);
        assert_eq!(bar_type.spec().step.get(), 1);
        assert_eq!(bar_type.spec().aggregation, BarAggregation::Hour);
    }

    #[rstest]
    fn test_resolution_to_bar_type_daily() {
        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
        let bar_type = resolution_to_bar_type(instrument_id, "1D").unwrap();

        assert_eq!(bar_type.instrument_id(), instrument_id);
        assert_eq!(bar_type.spec().step.get(), 1);
        assert_eq!(bar_type.spec().aggregation, BarAggregation::Day);
    }

    #[rstest]
    fn test_resolution_to_bar_type_invalid() {
        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
        let result = resolution_to_bar_type(instrument_id, "invalid");

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Unsupported Deribit resolution")
        );
    }

    #[rstest]
    fn test_parse_chart_msg_uses_cost() {
        let instrument = test_perpetual_instrument();
        assert!(
            instrument.is_inverse(),
            "test fixture is expected to be an inverse perp"
        );

        let json = load_test_json("ws_chart.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();
        let chart_msg: DeribitChartMsg =
            serde_json::from_str(&response["params"]["data"].to_string()).unwrap();

        // Verify chart message was deserialized correctly
        assert_eq!(chart_msg.tick, 1_767_200_040_000);
        assert_eq!(chart_msg.open, dec!(87490.0));
        assert_eq!(chart_msg.high, dec!(87500.0));
        assert_eq!(chart_msg.low, dec!(87465.0));
        assert_eq!(chart_msg.close, dec!(87474.0));
        assert_eq!(chart_msg.volume, dec!(0.95978896));
        assert_eq!(chart_msg.cost, dec!(83970.0));

        let bar_type = resolution_to_bar_type(instrument.id(), "1").unwrap();

        // Test with timestamp_on_close=true (default)
        let bar = parse_chart_msg(
            &chart_msg,
            bar_type,
            instrument.price_precision(),
            instrument.size_precision(),
            true, // use_cost_for_volume
            true,
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(bar.bar_type, bar_type);
        assert_eq!(bar.open, instrument.make_price(87490.0));
        assert_eq!(bar.high, instrument.make_price(87500.0));
        assert_eq!(bar.low, instrument.make_price(87465.0));
        assert_eq!(bar.close, instrument.make_price(87474.0));
        assert_eq!(bar.volume, instrument.make_qty(83970.0, None));

        // ts_event should be close time (open + 1 minute)
        assert_eq!(bar.ts_event, UnixNanos::new(1_767_200_100_000_000_000));
    }

    #[rstest]
    fn test_parse_order_buy_response() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_order_buy_response.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();

        // Parse the order from the response (buy/sell responses wrap order in {"order": ...})
        let order_msg: DeribitOrderMsg =
            serde_json::from_str(&response["result"]["order"].to_string()).unwrap();

        // Verify deserialization
        assert_eq!(order_msg.order_id, "USDC-104819327443");
        assert_eq!(
            order_msg.label,
            Some("O-19700101-000000-001-001-1".to_string())
        );
        assert_eq!(order_msg.direction, "buy");
        assert_eq!(order_msg.order_state, "open");
        assert_eq!(order_msg.order_type, "limit");
        assert_eq!(order_msg.price, Some(dec!(2973.55)));
        assert_eq!(order_msg.amount, dec!(0.001));
        assert_eq!(order_msg.filled_amount, rust_decimal::Decimal::ZERO);
        assert!(order_msg.post_only);
        assert!(!order_msg.reduce_only);

        // Test parse_order_accepted
        let account_id = AccountId::new("DERIBIT-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("PMM-001");

        let accepted = parse_order_accepted(
            &order_msg,
            &instrument,
            account_id,
            trader_id,
            strategy_id,
            UnixNanos::default(),
        );

        assert_eq!(
            accepted.client_order_id.to_string(),
            "O-19700101-000000-001-001-1"
        );
        assert_eq!(accepted.venue_order_id.to_string(), "USDC-104819327443");
        assert_eq!(accepted.trader_id, trader_id);
        assert_eq!(accepted.strategy_id, strategy_id);
        assert_eq!(accepted.account_id, account_id);
    }

    #[rstest]
    fn test_parse_order_sell_response() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_order_sell_response.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();

        let order_msg: DeribitOrderMsg =
            serde_json::from_str(&response["result"]["order"].to_string()).unwrap();

        // Verify deserialization
        assert_eq!(order_msg.order_id, "USDC-104819327458");
        assert_eq!(
            order_msg.label,
            Some("O-19700101-000000-001-001-2".to_string())
        );
        assert_eq!(order_msg.direction, "sell");
        assert_eq!(order_msg.order_state, "open");
        assert_eq!(order_msg.price, Some(dec!(3286.7)));
        assert_eq!(order_msg.amount, dec!(0.001));

        // Test parse_order_accepted for sell order
        let account_id = AccountId::new("DERIBIT-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("PMM-001");

        let accepted = parse_order_accepted(
            &order_msg,
            &instrument,
            account_id,
            trader_id,
            strategy_id,
            UnixNanos::default(),
        );

        assert_eq!(
            accepted.client_order_id.to_string(),
            "O-19700101-000000-001-001-2"
        );
        assert_eq!(accepted.venue_order_id.to_string(), "USDC-104819327458");
        assert_eq!(accepted.trader_id, trader_id);
        assert_eq!(accepted.strategy_id, strategy_id);
        assert_eq!(accepted.account_id, account_id);
    }

    #[rstest]
    fn test_parse_order_edit_response() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_order_edit_response.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();

        let order_msg: DeribitOrderMsg =
            serde_json::from_str(&response["result"]["order"].to_string()).unwrap();

        // Verify deserialization - edit response has replaced=true in raw JSON
        assert_eq!(order_msg.order_id, "USDC-104819327443");
        assert_eq!(
            order_msg.label,
            Some("O-19700101-000000-001-001-1".to_string())
        );
        assert_eq!(order_msg.direction, "buy");
        assert_eq!(order_msg.order_state, "open");
        assert!(order_msg.replaced);
        assert_eq!(order_msg.price, Some(dec!(3067.2))); // New price after edit

        // Test parse_order_updated
        let account_id = AccountId::new("DERIBIT-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("PMM-001");

        let updated = parse_order_updated(
            &order_msg,
            &instrument,
            account_id,
            trader_id,
            strategy_id,
            UnixNanos::default(),
        );

        assert_eq!(
            updated.client_order_id.to_string(),
            "O-19700101-000000-001-001-1"
        );
        assert_eq!(
            updated.venue_order_id.unwrap().to_string(),
            "USDC-104819327443"
        );
        // Note: 0.001 truncates to 0.0 due to BTC-PERPETUAL size_precision=0
        assert_eq!(updated.quantity.as_f64(), 0.0);
    }

    #[rstest]
    fn test_parse_order_cancel_response() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_order_cancel_response.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();

        // Cancel response has order fields directly in result (not wrapped)
        let order_msg: DeribitOrderMsg =
            serde_json::from_str(&response["result"].to_string()).unwrap();

        // Verify deserialization
        assert_eq!(order_msg.order_id, "USDC-104819327443");
        assert_eq!(
            order_msg.label,
            Some("O-19700101-000000-001-001-1".to_string())
        );
        assert_eq!(order_msg.order_state, "cancelled");
        assert_eq!(order_msg.cancel_reason, Some("user_request".to_string()));

        // Test parse_order_canceled
        let account_id = AccountId::new("DERIBIT-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("PMM-001");

        let canceled = parse_order_canceled(
            &order_msg,
            &instrument,
            account_id,
            trader_id,
            strategy_id,
            UnixNanos::default(),
        );

        assert_eq!(
            canceled.client_order_id.to_string(),
            "O-19700101-000000-001-001-1"
        );
        assert_eq!(
            canceled.venue_order_id.unwrap().to_string(),
            "USDC-104819327443"
        );
        assert_eq!(canceled.trader_id, trader_id);
        assert_eq!(canceled.strategy_id, strategy_id);
    }

    #[rstest]
    fn test_parse_order_stop_market_response() {
        // Regression for https://github.com/nautechsystems/nautilus_trader/issues/3925
        // Deribit returns the literal string "market_price" for the price of
        // trigger market orders; the deserializer must map this to None rather
        // than failing with "Invalid decimal: unknown character".
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_order_stop_market_response.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();

        let order_msg: DeribitOrderMsg =
            serde_json::from_str(&response["result"]["order"].to_string()).unwrap();

        assert_eq!(order_msg.order_id, "USDC-104819327499");
        assert_eq!(order_msg.order_type, "stop_market");
        assert_eq!(order_msg.order_state, "untriggered");
        assert_eq!(order_msg.price, None);
        assert_eq!(order_msg.trigger_price, Some(dec!(2228.0)));
        assert_eq!(order_msg.trigger.as_deref(), Some("mark_price"));
        assert!(order_msg.reduce_only);

        let account_id = AccountId::new("DERIBIT-001");
        let report =
            parse_user_order_msg(&order_msg, &instrument, account_id, UnixNanos::default())
                .unwrap();

        assert_eq!(report.order_type, OrderType::StopMarket);
        assert_eq!(report.order_status, OrderStatus::Accepted);
        assert!(report.price.is_none());
        assert!(report.trigger_price.is_some());
        assert!(report.reduce_only);
    }

    #[rstest]
    fn test_parse_order_stop_market_response_missing_filled_amount() {
        // Regression for https://github.com/nautechsystems/nautilus_trader/issues/3995
        // Deribit omits `filled_amount` for untriggered trigger market orders;
        // the deserializer must treat the missing field as zero rather than
        // failing with "missing field `filled_amount`".
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_order_stop_market_no_filled_amount.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();

        let order_msg: DeribitOrderMsg =
            serde_json::from_str(&response["result"]["order"].to_string()).unwrap();

        assert_eq!(order_msg.order_id, "USDC-SLMB-19641");
        assert_eq!(order_msg.order_type, "stop_market");
        assert_eq!(order_msg.order_state, "untriggered");
        assert_eq!(order_msg.filled_amount, rust_decimal::Decimal::ZERO);
        assert_eq!(order_msg.average_price, None);

        let account_id = AccountId::new("DERIBIT-001");
        let report =
            parse_user_order_msg(&order_msg, &instrument, account_id, UnixNanos::default())
                .unwrap();

        assert_eq!(report.order_type, OrderType::StopMarket);
        assert_eq!(report.order_status, OrderStatus::Accepted);
        assert_eq!(report.filled_qty.as_f64(), 0.0);
    }

    #[rstest]
    fn test_parse_user_order_msg_to_status_report() {
        let instrument = test_perpetual_instrument();
        let json = load_test_json("ws_order_buy_response.json");
        let response: serde_json::Value = serde_json::from_str(&json).unwrap();

        let order_msg: DeribitOrderMsg =
            serde_json::from_str(&response["result"]["order"].to_string()).unwrap();

        let account_id = AccountId::new("DERIBIT-001");
        let report =
            parse_user_order_msg(&order_msg, &instrument, account_id, UnixNanos::default())
                .unwrap();

        assert_eq!(report.venue_order_id.to_string(), "USDC-104819327443");
        assert_eq!(
            report.client_order_id.unwrap().to_string(),
            "O-19700101-000000-001-001-1"
        );
        assert_eq!(report.order_side, OrderSide::Buy.into());
        assert_eq!(report.order_type, OrderType::Limit);
        assert_eq!(report.time_in_force, TimeInForce::Gtc);
        assert_eq!(report.order_status, OrderStatus::Accepted);
        // Note: 0.001 truncates to 0.0 due to BTC-PERPETUAL size_precision=0
        assert_eq!(report.quantity.as_f64(), 0.0);
        assert_eq!(report.filled_qty.as_f64(), 0.0);
        assert!(report.post_only);
        assert!(!report.reduce_only);
    }

    #[rstest]
    fn test_determine_order_event_type() {
        // New order -> Accepted
        assert_eq!(
            determine_order_event_type("open", true, false),
            OrderEventType::Accepted
        );

        // Amended order -> Updated
        assert_eq!(
            determine_order_event_type("open", false, true),
            OrderEventType::Updated
        );

        // Cancelled order -> Canceled
        assert_eq!(
            determine_order_event_type("cancelled", false, false),
            OrderEventType::Canceled
        );

        // Expired order -> Expired
        assert_eq!(
            determine_order_event_type("expired", false, false),
            OrderEventType::Expired
        );

        // Filled order -> None (handled via trades)
        assert_eq!(
            determine_order_event_type("filled", false, false),
            OrderEventType::None
        );
    }

    fn perpetual_instrument_with_increments(
        tick_size: &str,
        min_trade_amount: &str,
    ) -> InstrumentAny {
        let json = load_test_json("http_get_instruments.json");
        let mut response: serde_json::Value = serde_json::from_str(&json).unwrap();
        response["result"][0]["tick_size"] = serde_json::json!(tick_size);
        response["result"][0]["min_trade_amount"] = serde_json::json!(min_trade_amount);
        let response: DeribitJsonRpcResponse<Vec<DeribitInstrument>> =
            serde_json::from_str(&response.to_string()).unwrap();
        let instrument = &response.result.unwrap()[0];
        parse_deribit_instrument_any(instrument, UnixNanos::default(), UnixNanos::default())
            .unwrap()
            .unwrap()
    }

    fn parse_notification_data<T: DeserializeOwned>(fixture: &str) -> T {
        serde_json::from_str(&notification_data(fixture)).unwrap()
    }

    fn notification_data(fixture: &str) -> String {
        let json = load_test_json(fixture);

        let DeribitWsMessage::Notification(notification) = parse_raw_message(&json).unwrap() else {
            panic!("Expected notification from {fixture}");
        };

        notification.params.data.get().to_string()
    }

    #[rstest]
    fn test_parse_user_trade_msg_keeps_decimals_that_collapse_in_f64() {
        let instrument = perpetual_instrument_with_increments("0.000000001", "0.000000001");
        let trades: Vec<DeribitUserTradeMsg> = parse_notification_data("ws_user_trades_exact.json");
        let account_id = AccountId::new("DERIBIT-001");
        let btc = Currency::BTC();

        let first = parse_user_trade_msg(&trades[0], &instrument, account_id, UnixNanos::default())
            .unwrap();
        let second =
            parse_user_trade_msg(&trades[1], &instrument, account_id, UnixNanos::default())
                .unwrap();

        let fee_f64 = "1000000000.12345678".parse::<f64>().unwrap();
        let next_fee_f64 = "1000000000.12345679".parse::<f64>().unwrap();
        assert_eq!(fee_f64.to_bits(), next_fee_f64.to_bits());
        assert_eq!(instrument.price_precision(), 9);
        assert_eq!(instrument.size_precision(), 9);
        assert_eq!(first.last_px, Price::from("100000000.123456789"));
        assert_eq!(first.last_qty, Quantity::from("100000000.123456789"));
        assert_eq!(
            first.commission,
            Money::from_decimal(dec!(1000000000.12345678), btc).unwrap()
        );
        assert_eq!(second.last_px, Price::from("100000000.123456788"));
        assert_eq!(second.last_qty, Quantity::from("100000000.123456788"));
        assert_eq!(
            second.commission,
            Money::from_decimal(dec!(1000000000.12345679), btc).unwrap()
        );
        assert_eq!(trades[0].mark_price, dec!(100000000.123456788));
        assert_eq!(trades[0].profit_loss, Some(dec!(0.30000000000000004)));
    }

    #[rstest]
    fn test_parse_user_trade_msg_rounds_exact_token_half_even() {
        let instrument = perpetual_instrument_with_increments("0.01", "0.01");
        let trades: Vec<DeribitUserTradeMsg> = parse_notification_data("ws_user_trades_exact.json");
        let account_id = AccountId::new("DERIBIT-001");

        let fill = parse_user_trade_msg(&trades[2], &instrument, account_id, UnixNanos::default())
            .unwrap();

        let token_f64 = "100000000.005000001".parse::<f64>().unwrap();
        let via_f64 = Price::from_decimal_dp(Decimal::try_from(token_f64).unwrap(), 2).unwrap();
        assert_eq!(via_f64, Price::from("100000000.00"));
        assert_eq!(fill.last_px, Price::from("100000000.01"));
        assert_eq!(fill.last_qty, Quantity::from("100000000.01"));
        assert_eq!(
            fill.commission,
            Money::from_decimal(dec!(0.00011403), Currency::BTC()).unwrap()
        );
    }

    #[rstest]
    fn test_parse_portfolio_keeps_decimals_that_collapse_in_f64() {
        let portfolio: DeribitPortfolioMsg = parse_notification_data("ws_portfolio_exact.json");
        let account_id = AccountId::new("DERIBIT-001");
        let btc = Currency::BTC();

        let state =
            parse_portfolio_to_account_state(&portfolio, account_id, UnixNanos::default()).unwrap();

        assert_eq!(state.balances.len(), 1);
        assert_eq!(
            state.balances[0].total,
            Money::from_decimal(dec!(1000000000.12345679), btc).unwrap()
        );
        assert_eq!(
            state.balances[0].free,
            Money::from_decimal(dec!(1000000000.12345678), btc).unwrap()
        );
        assert_eq!(
            state.balances[0].locked,
            Money::from_decimal(dec!(0.00000001), btc).unwrap()
        );
        assert_eq!(state.margins.len(), 1);
        assert_eq!(
            state.margins[0].initial,
            Money::from_decimal(dec!(1000000000.12345678), btc).unwrap()
        );
        assert_eq!(state.margins[0].maintenance, Money::zero(btc));
        assert_eq!(
            portfolio.maintenance_margin,
            dec!(0.0000000000000000555111512313)
        );
    }

    #[rstest]
    fn test_parse_book_snapshot_keeps_decimals_that_collapse_in_f64() {
        let instrument = perpetual_instrument_with_increments("0.000000001", "0.000000001");
        let data = notification_data("ws_book_snapshot_exact.json");
        let msg: DeribitBookMsg = serde_json::from_str(&data).unwrap();

        let deltas = parse_book_msg(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(deltas.deltas.len(), 3);
        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
        assert_eq!(
            deltas.deltas[1].order.price,
            Price::from("100000000.123456789")
        );
        assert_eq!(
            deltas.deltas[1].order.size,
            Quantity::from("100000000.123456789")
        );
        assert_eq!(
            deltas.deltas[2].order.price,
            Price::from("100000000.123456788")
        );
        assert_eq!(
            deltas.deltas[2].order.size,
            Quantity::from("100000000.123456788")
        );
    }

    #[rstest]
    fn test_parse_book_delta_rounds_levels_beyond_decimal_scale() {
        let instrument = perpetual_instrument_with_increments("0.000000001", "0.000000001");
        let data = notification_data("ws_book_delta_exact.json");
        let msg: DeribitBookMsg = serde_json::from_str(&data).unwrap();

        let deltas = parse_book_msg(&msg, &instrument, UnixNanos::default()).unwrap();

        assert_eq!(deltas.deltas.len(), 2);
        assert_eq!(deltas.deltas[0].action, BookAction::Update);
        assert_eq!(
            deltas.deltas[0].order.price,
            Price::from("100000000.123456789")
        );
        assert_eq!(deltas.deltas[0].order.size, Quantity::from("0.123456789"));
        assert_eq!(deltas.deltas[1].action, BookAction::Add);
        assert_eq!(
            deltas.deltas[1].order.price,
            Price::from("100000000.123456788")
        );
        assert_eq!(deltas.deltas[1].order.size, Quantity::from("0.000012346"));
    }

    #[rstest]
    fn test_parse_book_delta_maps_zero_rounding_amounts_to_delete() {
        let instrument = perpetual_instrument_with_increments("0.5", "1");
        let data = r#"{"type":"change","instrument_name":"BTC-PERPETUAL","timestamp":1700000000000,
            "change_id":101,"prev_change_id":100,
            "bids":[["change",42500.0,1.1102230246251565e-16],["new",42499.5,0.4]],
            "asks":[["change",42500.5,20.0],["delete",42501.0,0.0]]}"#;
        let msg: DeribitBookMsg = serde_json::from_str(data).unwrap();

        let deltas = parse_book_msg(&msg, &instrument, UnixNanos::default()).unwrap();

        let actual: Vec<_> = deltas
            .deltas
            .iter()
            .map(|delta| {
                (
                    delta.action,
                    delta.order.side,
                    delta.order.price,
                    delta.order.size,
                    delta.order.order_id,
                    delta.flags,
                    delta.sequence,
                )
            })
            .collect();

        assert_eq!(
            actual,
            vec![
                (
                    BookAction::Delete,
                    Some(OrderSide::Buy),
                    Price::from("42500.0"),
                    Quantity::from("0"),
                    0,
                    0,
                    101,
                ),
                (
                    BookAction::Delete,
                    Some(OrderSide::Buy),
                    Price::from("42499.5"),
                    Quantity::from("0"),
                    1,
                    0,
                    101,
                ),
                (
                    BookAction::Update,
                    Some(OrderSide::Sell),
                    Price::from("42500.5"),
                    Quantity::from("20"),
                    2,
                    0,
                    101,
                ),
                (
                    BookAction::Delete,
                    Some(OrderSide::Sell),
                    Price::from("42501.0"),
                    Quantity::from("0"),
                    3,
                    RecordFlag::F_LAST as u8,
                    101,
                ),
            ]
        );
    }

    #[rstest]
    fn test_parse_book_snapshot_skips_zero_rounding_amounts() {
        let instrument = perpetual_instrument_with_increments("0.5", "1");
        let data = r#"{"type":"snapshot","instrument_name":"BTC-PERPETUAL","timestamp":1700000000000,
            "change_id":100,
            "bids":[["new",42500.0,1.1102230246251565e-16],["new",42499.5,10.0]],
            "asks":[["new",42500.5,0.4]]}"#;
        let msg: DeribitBookMsg = serde_json::from_str(data).unwrap();

        let deltas = parse_book_msg(&msg, &instrument, UnixNanos::default()).unwrap();

        let actual: Vec<_> = deltas
            .deltas
            .iter()
            .map(|delta| {
                (
                    delta.action,
                    delta.order.side,
                    delta.order.price,
                    delta.order.size,
                    delta.order.order_id,
                    delta.flags,
                )
            })
            .collect();

        assert_eq!(
            actual,
            vec![
                (
                    BookAction::Clear,
                    None,
                    Price::from("0"),
                    Quantity::from("0"),
                    0,
                    RecordFlag::F_SNAPSHOT as u8,
                ),
                (
                    BookAction::Add,
                    Some(OrderSide::Buy),
                    Price::from("42499.5"),
                    Quantity::from("10"),
                    1,
                    RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8,
                ),
            ]
        );
    }

    #[rstest]
    #[case::snapshot(
        r#"{"type":"snapshot","instrument_name":"BTC-PERPETUAL","timestamp":1700000000000,
            "change_id":100,"bids":[["new",100000000000000000000.0,10.0]],"asks":[]}"#
    )]
    #[case::delta(
        r#"{"type":"change","instrument_name":"BTC-PERPETUAL","timestamp":1700000000000,
            "change_id":101,"prev_change_id":100,
            "bids":[["new",100000000000000000000.0,10.0]],"asks":[]}"#
    )]
    fn test_parse_book_msg_rejects_unconvertible_price(#[case] data: &str) {
        let instrument = perpetual_instrument_with_increments("0.5", "1");
        let msg: DeribitBookMsg = serde_json::from_str(data).unwrap();

        let error = parse_book_msg(&msg, &instrument, UnixNanos::default()).unwrap_err();

        let expected = Price::from_decimal_dp(dec!(100000000000000000000.0), 1).unwrap_err();
        assert_eq!(error.to_string(), expected.to_string());
    }
}