nautilus_hyperliquid/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.
// -------------------------------------------------------------------------------------------------

//! Parsers for Hyperliquid WebSocket payloads.

use anyhow::Context;
use nautilus_core::{nanos::UnixNanos, uuid::UUID4};
use nautilus_model::{
    data::{
        Bar, BarType, BookOrder, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate,
        OrderBookDelta, OrderBookDeltas, OrderBookDepth, QuoteTick, TradeTick, depth::DEPTH10_LEN,
    },
    enums::{
        AggressorSide, BookAction, LiquiditySide, OrderSide, OrderStatus, OrderType, RecordFlag,
        TimeInForce,
    },
    identifiers::{AccountId, ClientOrderId, TradeId, VenueOrderId},
    instruments::{Instrument, InstrumentAny},
    reports::{FillReport, OrderStatusReport},
    types::{Money, Price, Quantity},
};
use rust_decimal::Decimal;

use super::messages::{
    CandleData, TwapStateData, WsActiveAssetCtxData, WsBboData, WsBookData, WsFillData,
    WsOrderData, WsTradeData, WsTwapHistoryData, WsTwapSliceFillData,
};
use crate::{
    common::{
        converters::hyperliquid_time_in_force_to_nautilus,
        enums::{HyperliquidFillDirection, HyperliquidTimeInForce},
        parse::{
            is_conditional_order_data, make_fill_trade_id, millis_to_nanos,
            parse_trigger_order_type,
        },
    },
    data_types::{
        HyperliquidOpenInterest, HyperliquidPublicTrade, HyperliquidTwapHistory,
        HyperliquidTwapSliceFill,
    },
};

fn parse_price(
    value: Decimal,
    instrument: &InstrumentAny,
    field_name: &str,
) -> anyhow::Result<Price> {
    Price::from_decimal_dp(value, instrument.price_precision())
        .with_context(|| format!("Failed to create price from '{value}' for {field_name}"))
}

fn parse_quantity(
    value: Decimal,
    instrument: &InstrumentAny,
    field_name: &str,
) -> anyhow::Result<Quantity> {
    Quantity::from_decimal_dp(value.abs(), instrument.size_precision())
        .with_context(|| format!("Failed to create quantity from '{value}' for {field_name}"))
}

/// Parses a WebSocket trade frame into a [`TradeTick`].
pub fn parse_ws_trade_tick(
    trade: &WsTradeData,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<TradeTick> {
    let price = parse_price(trade.px, instrument, "trade.px")?;
    let size = parse_quantity(trade.sz, instrument, "trade.sz")?;
    let aggressor = AggressorSide::from(trade.side);
    let trade_id = TradeId::new_checked(trade.tid.to_string())
        .context("invalid trade identifier in Hyperliquid trade message")?;
    let ts_event = millis_to_nanos(trade.time)?;

    TradeTick::new_checked(
        instrument.id(),
        price,
        size,
        aggressor,
        trade_id,
        ts_event,
        ts_init,
    )
    .context("failed to construct TradeTick from Hyperliquid trade message")
}

/// Parses a WebSocket trade frame into a complete public Hyperliquid trade.
pub fn parse_ws_public_trade(
    trade: &WsTradeData,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<HyperliquidPublicTrade> {
    let price = parse_price(trade.px, instrument, "trade.px")?;
    let size = parse_quantity(trade.sz, instrument, "trade.sz")?;
    let ts_event = millis_to_nanos(trade.time)?;

    Ok(HyperliquidPublicTrade::new(
        instrument.id(),
        price,
        size,
        AggressorSide::from(trade.side),
        trade.tid.to_string(),
        trade.users[0].clone(),
        trade.users[1].clone(),
        trade.hash.clone(),
        ts_event,
        ts_init,
    ))
}

/// Parses a WebSocket L2 order book message into [`OrderBookDeltas`].
pub fn parse_ws_order_book_deltas(
    book: &WsBookData,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderBookDeltas> {
    let ts_event = millis_to_nanos(book.time)?;
    let bids = &book.levels[0];
    let asks = &book.levels[1];
    let mut deltas = Vec::with_capacity(1 + bids.len() + asks.len());

    // Treat every book payload as a snapshot: clear existing depth and rebuild it
    deltas.push(OrderBookDelta::clear(instrument.id(), 0, ts_event, ts_init));

    for level in bids {
        let price = parse_price(level.px, instrument, "book.bid.px")?;
        let size = parse_quantity(level.sz, instrument, "book.bid.sz")?;

        if !size.is_positive() {
            continue;
        }

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

        let delta = OrderBookDelta::new(
            instrument.id(),
            BookAction::Add,
            order,
            RecordFlag::F_LAST as u8,
            0, // sequence
            ts_event,
            ts_init,
        );

        deltas.push(delta);
    }

    for level in asks {
        let price = parse_price(level.px, instrument, "book.ask.px")?;
        let size = parse_quantity(level.sz, instrument, "book.ask.sz")?;

        if !size.is_positive() {
            continue;
        }

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

        let delta = OrderBookDelta::new(
            instrument.id(),
            BookAction::Add,
            order,
            RecordFlag::F_LAST as u8,
            0, // sequence
            ts_event,
            ts_init,
        );

        deltas.push(delta);
    }

    Ok(OrderBookDeltas::new(instrument.id(), deltas))
}

/// Parses a WebSocket L2 order book snapshot into [`OrderBookDepth`].
///
/// Hyperliquid's `l2Book` subscription emits snapshots of bid/ask levels.
/// Fills any missing levels past the venue-provided depth with zero-size
/// placeholder orders so the fixed-size `[BookOrder; 10]` arrays are
/// always fully populated.
pub fn parse_ws_order_book_depth(
    book: &WsBookData,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderBookDepth> {
    let ts_event = millis_to_nanos(book.time)?;
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let mut bids: [BookOrder; DEPTH10_LEN] = [BookOrder::default(); DEPTH10_LEN];
    let mut asks: [BookOrder; DEPTH10_LEN] = [BookOrder::default(); DEPTH10_LEN];
    let mut bid_counts: [u32; DEPTH10_LEN] = [0; DEPTH10_LEN];
    let mut ask_counts: [u32; DEPTH10_LEN] = [0; DEPTH10_LEN];

    let raw_bids = book.levels.first().map_or(&[][..], |v| v.as_slice());
    let raw_asks = book.levels.get(1).map_or(&[][..], |v| v.as_slice());

    for (i, level) in raw_bids.iter().take(DEPTH10_LEN).enumerate() {
        let price = parse_price(level.px, instrument, "book.bid.px")?;
        let size = parse_quantity(level.sz, instrument, "book.bid.sz")?;
        bids[i] = BookOrder::new(OrderSide::Buy, price, size, 0);
        bid_counts[i] = level.n;
    }

    for bid in bids.iter_mut().skip(raw_bids.len().min(DEPTH10_LEN)) {
        *bid = BookOrder::new(
            OrderSide::Buy,
            Price::zero(price_precision),
            Quantity::zero(size_precision),
            0,
        );
    }

    for (i, level) in raw_asks.iter().take(DEPTH10_LEN).enumerate() {
        let price = parse_price(level.px, instrument, "book.ask.px")?;
        let size = parse_quantity(level.sz, instrument, "book.ask.sz")?;
        asks[i] = BookOrder::new(OrderSide::Sell, price, size, 0);
        ask_counts[i] = level.n;
    }

    for ask in asks.iter_mut().skip(raw_asks.len().min(DEPTH10_LEN)) {
        *ask = BookOrder::new(
            OrderSide::Sell,
            Price::zero(price_precision),
            Quantity::zero(size_precision),
            0,
        );
    }

    Ok(OrderBookDepth::new(
        instrument.id(),
        bids,
        asks,
        bid_counts,
        ask_counts,
        RecordFlag::F_SNAPSHOT as u8,
        0,
        ts_event,
        ts_init,
    ))
}

/// Parses a WebSocket BBO (best bid/offer) message into a [`QuoteTick`].
pub fn parse_ws_quote_tick(
    bbo: &WsBboData,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<QuoteTick> {
    let bid_level = bbo.bbo[0]
        .as_ref()
        .context("BBO message missing bid level")?;
    let ask_level = bbo.bbo[1]
        .as_ref()
        .context("BBO message missing ask level")?;

    let bid_price = parse_price(bid_level.px, instrument, "bbo.bid.px")?;
    let ask_price = parse_price(ask_level.px, instrument, "bbo.ask.px")?;
    let bid_size = parse_quantity(bid_level.sz, instrument, "bbo.bid.sz")?;
    let ask_size = parse_quantity(ask_level.sz, instrument, "bbo.ask.sz")?;

    let ts_event = millis_to_nanos(bbo.time)?;

    QuoteTick::new_checked(
        instrument.id(),
        bid_price,
        ask_price,
        bid_size,
        ask_size,
        ts_event,
        ts_init,
    )
    .context("failed to construct QuoteTick from Hyperliquid BBO message")
}

/// Parses a WebSocket candle message into a [`Bar`].
pub fn parse_ws_candle(
    candle: &CandleData,
    instrument: &InstrumentAny,
    bar_type: &BarType,
    ts_init: UnixNanos,
) -> anyhow::Result<Bar> {
    let open = parse_price(candle.o, instrument, "candle.o")?;
    let high = parse_price(candle.h, instrument, "candle.h")?;
    let low = parse_price(candle.l, instrument, "candle.l")?;
    let close = parse_price(candle.c, instrument, "candle.c")?;
    let volume = parse_quantity(candle.v, instrument, "candle.v")?;

    let ts_event = millis_to_nanos(candle.t)?;

    Ok(Bar::new(
        *bar_type, open, high, low, close, volume, ts_event, ts_init,
    ))
}

/// Parses a WebSocket order update message into an [`OrderStatusReport`].
///
/// This converts Hyperliquid order data from WebSocket into Nautilus order status reports.
/// Handles both regular and conditional orders (stop/limit-if-touched).
pub fn parse_ws_order_status_report(
    order: &WsOrderData,
    instrument: &InstrumentAny,
    account_id: AccountId,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderStatusReport> {
    let instrument_id = instrument.id();
    let venue_order_id = VenueOrderId::new(order.order.oid.to_string());
    let order_side = OrderSide::from(order.order.side);

    // Determine order type based on trigger info
    let is_conditional =
        is_conditional_order_data(order.order.trigger_px, order.order.tpsl.as_ref());
    let order_type = if is_conditional {
        if let (Some(is_market), Some(tpsl)) = (order.order.is_market, order.order.tpsl.as_ref()) {
            parse_trigger_order_type(is_market, tpsl)
        } else {
            OrderType::Limit // fallback
        }
    } else {
        OrderType::Limit // Regular limit order
    };

    let time_in_force = order
        .order
        .tif
        .map_or(TimeInForce::Gtc, hyperliquid_time_in_force_to_nautilus);
    let order_status = OrderStatus::from(order.status);

    // orig_sz is the original order quantity, sz is the remaining quantity
    let orig_qty = parse_quantity(order.order.orig_sz, instrument, "order.orig_sz")?;
    let remaining_qty = parse_quantity(order.order.sz, instrument, "order.sz")?;
    let filled_qty = orig_qty - orig_qty.min(remaining_qty);

    let price = parse_price(order.order.limit_px, instrument, "order.limitPx")?;

    let ts_accepted = millis_to_nanos(order.order.timestamp)?;
    let ts_last = millis_to_nanos(order.status_timestamp)?;

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

    if let Some(ref cloid) = order.order.cloid {
        report = report.with_client_order_id(ClientOrderId::new(cloid.as_str()));
    }

    if matches!(order.order.tif, Some(HyperliquidTimeInForce::Alo)) {
        report = report.with_post_only(true);
    }

    if let Some(reduce_only) = order.order.reduce_only {
        report = report.with_reduce_only(reduce_only);
    }

    if let Some(reason) = order.status.rejection_reason() {
        report = report.with_cancel_reason(reason.to_string());
    }

    report = report.with_price(price);

    if is_conditional && let Some(trigger_px) = order.order.trigger_px {
        let trigger_price = parse_price(trigger_px, instrument, "order.triggerPx")?;
        report = report.with_trigger_price(trigger_price);
    }

    Ok(report)
}

/// Parses a WebSocket fill message into a [`FillReport`].
///
/// This converts Hyperliquid fill data from WebSocket user events into Nautilus fill reports.
pub fn parse_ws_fill_report(
    fill: &WsFillData,
    instrument: &InstrumentAny,
    account_id: AccountId,
    ts_init: UnixNanos,
) -> anyhow::Result<FillReport> {
    let instrument_id = instrument.id();

    if let Some(liquidation) = fill.liquidation.as_ref() {
        log::warn!(
            "Liquidation fill: {} oid={} method={:?} mark_px={} liquidated_user={}",
            instrument_id,
            fill.oid,
            liquidation.method,
            liquidation.mark_px,
            liquidation
                .liquidated_user
                .as_deref()
                .unwrap_or("<unknown>"),
        );
    } else if matches!(fill.dir, HyperliquidFillDirection::AutoDeleveraging) {
        log::warn!(
            "Auto-deleveraging fill: {instrument_id} oid={} px={} sz={}",
            fill.oid,
            fill.px,
            fill.sz,
        );
    }

    let venue_order_id = VenueOrderId::new(fill.oid.to_string());
    let trade_id = make_fill_trade_id(
        &fill.hash,
        fill.oid,
        fill.px,
        fill.sz,
        fill.time,
        fill.start_position,
    );

    let order_side = OrderSide::from(fill.side);
    let last_qty = parse_quantity(fill.sz, instrument, "fill.sz")?;
    let last_px = parse_price(fill.px, instrument, "fill.px")?;
    let liquidity_side = if fill.crossed {
        LiquiditySide::Taker
    } else {
        LiquiditySide::Maker
    };

    let fee_amount = fill.fee;

    let commission_currency =
        crate::http::parse::resolve_fee_currency(fill.fee_token.as_str(), fee_amount, instrument)?;

    let commission = Money::from_decimal(fee_amount, commission_currency)
        .with_context(|| format!("Failed to create commission from fee='{}'", fill.fee))?;
    let ts_event = millis_to_nanos(fill.time)?;

    // No client order ID available in fill data directly
    let client_order_id = None;

    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 WebSocket ActiveAssetCtx message into mark price, index price, and funding rate updates.
///
/// This converts Hyperliquid asset context data into Nautilus price and funding rate updates.
/// Returns a tuple of (`MarkPriceUpdate`, `Option<IndexPriceUpdate>`, `Option<FundingRateUpdate>`).
/// Index price and funding rate are only present for perpetual contracts.
pub fn parse_ws_asset_context(
    ctx: &WsActiveAssetCtxData,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<(
    MarkPriceUpdate,
    Option<IndexPriceUpdate>,
    Option<FundingRateUpdate>,
)> {
    let instrument_id = instrument.id();

    match ctx {
        WsActiveAssetCtxData::Perp { coin: _, ctx } => {
            let mark_price = parse_price(ctx.shared.mark_px, instrument, "ctx.mark_px")?;
            let mark_price_update =
                MarkPriceUpdate::new(instrument_id, mark_price, ts_init, ts_init);

            let index_price = parse_price(ctx.oracle_px, instrument, "ctx.oracle_px")?;
            let index_price_update =
                IndexPriceUpdate::new(instrument_id, index_price, ts_init, ts_init);

            let funding_rate_update = FundingRateUpdate::new(
                instrument_id,
                ctx.funding,
                Some(60), // Hyperliquid exchanges funding hourly
                None,     // Hyperliquid doesn't provide next funding time in this message
                ts_init,
                ts_init,
            );

            Ok((
                mark_price_update,
                Some(index_price_update),
                Some(funding_rate_update),
            ))
        }
        WsActiveAssetCtxData::Spot { coin: _, ctx } => {
            let mark_price = parse_price(ctx.shared.mark_px, instrument, "ctx.mark_px")?;
            let mark_price_update =
                MarkPriceUpdate::new(instrument_id, mark_price, ts_init, ts_init);

            Ok((mark_price_update, None, None))
        }
    }
}

/// Parses an `activeAssetCtx` open interest string into an open interest custom data update.
///
/// The caller is responsible for restricting this to the perpetual branch; spot
/// `activeAssetCtx` payloads carry no open interest field.
pub fn parse_ws_open_interest(
    open_interest: Decimal,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<HyperliquidOpenInterest> {
    Ok(HyperliquidOpenInterest::new(
        instrument.id(),
        open_interest,
        ts_init,
        ts_init,
    ))
}

/// Converts Hyperliquid TWAP times to nanos.
///
/// History row `time` is seconds; `state.timestamp` and fill `time` are milliseconds.
fn venue_time_to_nanos(value: u64) -> anyhow::Result<UnixNanos> {
    if value < 100_000_000_000 {
        Ok(UnixNanos::from(value.checked_mul(1_000_000_000).context(
            "venue time seconds overflow converting to nanos",
        )?))
    } else {
        millis_to_nanos(value)
    }
}

/// Parses one `userTwapHistory` row into custom data.
///
/// Unknown coins leave `instrument_id` unset and do not fail the parse.
pub fn parse_ws_twap_history_row(
    row: &WsTwapHistoryData,
    user: &str,
    is_snapshot: bool,
    instrument: Option<&InstrumentAny>,
    ts_init: UnixNanos,
) -> anyhow::Result<HyperliquidTwapHistory> {
    let state: &TwapStateData = &row.state;
    let ts_event = venue_time_to_nanos(row.time)?;
    let state_timestamp = venue_time_to_nanos(state.timestamp)?;
    let envelope_user = if user.is_empty() {
        state.user.as_str()
    } else {
        user
    };

    Ok(HyperliquidTwapHistory::new(
        envelope_user.to_string(),
        row.twap_id,
        state.coin.to_string(),
        instrument.map(Instrument::id),
        OrderSide::from(state.side),
        state.sz,
        state.executed_sz,
        state.executed_ntl,
        state.minutes,
        state.reduce_only,
        state.randomize,
        row.status.status,
        row.status.description.clone(),
        state_timestamp,
        is_snapshot,
        ts_event,
        ts_init,
    ))
}

/// Parses one `userTwapSliceFills` item into custom data.
///
/// Unknown coins leave `instrument_id` unset and do not fail the parse.
pub fn parse_ws_twap_slice_fill(
    item: &WsTwapSliceFillData,
    user: &str,
    is_snapshot: bool,
    instrument: Option<&InstrumentAny>,
    ts_init: UnixNanos,
) -> anyhow::Result<HyperliquidTwapSliceFill> {
    let fill = &item.fill;
    let ts_event = millis_to_nanos(fill.time)?;

    Ok(HyperliquidTwapSliceFill::new(
        user.to_string(),
        item.twap_id,
        fill.coin.to_string(),
        instrument.map(Instrument::id),
        fill.px,
        fill.sz,
        OrderSide::from(fill.side),
        fill.hash.clone(),
        fill.oid,
        fill.tid,
        fill.crossed,
        fill.fee,
        fill.fee_token.to_string(),
        fill.dir.to_string(),
        fill.closed_pnl,
        is_snapshot,
        ts_event,
        ts_init,
    ))
}

#[cfg(test)]
mod tests {
    use std::str::FromStr;

    use nautilus_model::{
        identifiers::{InstrumentId, Symbol},
        instruments::CryptoPerpetual,
        types::currency::Currency,
    };
    use rstest::rstest;
    use rust_decimal_macros::dec;
    use ustr::Ustr;

    use super::*;
    use crate::{
        common::{
            consts::HYPERLIQUID_VENUE,
            enums::{
                HyperliquidFillDirection, HyperliquidLiquidationMethod,
                HyperliquidOrderStatus as HyperliquidOrderStatusEnum, HyperliquidSide,
                HyperliquidTimeInForce,
            },
        },
        websocket::messages::{
            CandleData, FillLiquidationData, PerpsAssetCtx, SharedAssetCtx, SpotAssetCtx,
            WsBasicOrderData, WsBookData, WsLevelData,
        },
    };

    fn create_test_instrument() -> InstrumentAny {
        let instrument_id = InstrumentId::new(Symbol::new("BTC-PERP"), *HYPERLIQUID_VENUE);

        InstrumentAny::CryptoPerpetual(
            CryptoPerpetual::builder()
                .instrument_id(instrument_id)
                .raw_symbol(Symbol::new("BTC-PERP"))
                .base_currency(Currency::from("BTC"))
                .quote_currency(Currency::from("USDC"))
                .settlement_currency(Currency::from("USDC"))
                .is_inverse(false)
                .price_precision(2)
                .size_precision(3)
                .price_increment(Price::from("0.01"))
                .size_increment(Quantity::from("0.001"))
                .ts_event(UnixNanos::default())
                .ts_init(UnixNanos::default())
                .build()
                .unwrap(),
        )
    }

    #[rstest]
    fn test_parse_ws_candle_preserves_open_event_and_receipt_initialization_timestamps() {
        let instrument = create_test_instrument();
        let bar_type = BarType::from("BTC-PERP.HYPERLIQUID-1-MINUTE-LAST-EXTERNAL");
        let candle = CandleData {
            t: 1_700_000_000_000,
            close_time: 1_700_000_059_999,
            s: Ustr::from("BTC"),
            i: Ustr::from("1m"),
            o: dec!(100.0),
            c: dec!(100.5),
            h: dec!(101.0),
            l: dec!(99.0),
            v: dec!(10.0),
            n: 42,
        };
        let receipt_timestamp = UnixNanos::from(1_700_000_060_123_000_000);

        let bar = parse_ws_candle(&candle, &instrument, &bar_type, receipt_timestamp).unwrap();

        assert_eq!(bar.ts_event, millis_to_nanos(candle.t).unwrap());
        assert_eq!(bar.ts_init, receipt_timestamp);
    }

    #[rstest]
    fn test_parse_ws_order_status_report_basic() {
        let instrument = create_test_instrument();
        let account_id = AccountId::new("HYPERLIQUID-001");
        let ts_init = UnixNanos::default();

        let order_data = WsOrderData {
            order: WsBasicOrderData {
                coin: Ustr::from("BTC"),
                side: HyperliquidSide::Buy,
                limit_px: dec!(50000.0),
                sz: dec!(0.5),
                oid: 12345,
                timestamp: 1704470400000,
                orig_sz: dec!(1.0),
                cloid: Some("test-order-1".to_string()),
                tif: Some(HyperliquidTimeInForce::Alo),
                reduce_only: Some(true),
                trigger_px: Some(dec!(0.0)),
                is_market: None,
                tpsl: None,
                trigger_activated: None,
                trailing_stop: None,
            },
            status: HyperliquidOrderStatusEnum::Open,
            status_timestamp: 1704470400000,
        };

        let result = parse_ws_order_status_report(&order_data, &instrument, account_id, ts_init);
        assert!(result.is_ok());

        let report = result.unwrap();
        assert_eq!(report.order_side, OrderSide::Buy.into());
        assert_eq!(report.order_type, OrderType::Limit);
        assert_eq!(report.order_status, OrderStatus::Accepted);
        assert_eq!(report.time_in_force, TimeInForce::Gtc);
        assert!(report.post_only);
        assert!(report.reduce_only);
        assert!(report.trigger_price.is_none());
    }

    #[rstest]
    #[case(
        HyperliquidOrderStatusEnum::BadAloPxRejected,
        "Post only order would have immediately matched"
    )]
    #[case(
        HyperliquidOrderStatusEnum::ReduceOnlyRejected,
        "Reduce only order would increase position."
    )]
    #[case(
        HyperliquidOrderStatusEnum::IocCancelRejected,
        "Order could not immediately match against any resting orders"
    )]
    fn test_parse_ws_rejection_preserves_venue_reason(
        #[case] status: HyperliquidOrderStatusEnum,
        #[case] expected_reason: &str,
    ) {
        let instrument = create_test_instrument();
        let order_data = WsOrderData {
            order: WsBasicOrderData {
                coin: Ustr::from("BTC"),
                side: HyperliquidSide::Buy,
                limit_px: dec!(50000.0),
                sz: dec!(1.0),
                oid: 12345,
                timestamp: 1704470400000,
                orig_sz: dec!(1.0),
                cloid: Some("test-rejection".to_string()),
                tif: Some(HyperliquidTimeInForce::Alo),
                reduce_only: Some(false),
                trigger_px: None,
                is_market: None,
                tpsl: None,
                trigger_activated: None,
                trailing_stop: None,
            },
            status,
            status_timestamp: 1704470400000,
        };

        let report = parse_ws_order_status_report(
            &order_data,
            &instrument,
            AccountId::new("HYPERLIQUID-001"),
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(report.order_status, OrderStatus::Rejected);
        assert_eq!(report.cancel_reason.as_deref(), Some(expected_reason));
    }

    #[rstest]
    fn test_parse_ws_fill_report_basic() {
        let instrument = create_test_instrument();
        let account_id = AccountId::new("HYPERLIQUID-001");
        let ts_init = UnixNanos::default();

        let fill_data = WsFillData {
            coin: Ustr::from("BTC"),
            px: dec!(50000.0),
            sz: dec!(0.1),
            side: HyperliquidSide::Buy,
            time: 1704470400000,
            start_position: dec!(0.0),
            dir: HyperliquidFillDirection::OpenLong,
            closed_pnl: dec!(0.0),
            hash: "0xabc123".to_string(),
            oid: 12345,
            crossed: true,
            fee: dec!(0.05),
            tid: 98765,
            liquidation: None,
            fee_token: Ustr::from("USDC"),
            builder_fee: None,
            cloid: Some("0xd211f1c27288259290850338d22132a0".to_string()),
            twap_id: None,
        };

        let result = parse_ws_fill_report(&fill_data, &instrument, account_id, ts_init);
        assert!(result.is_ok());

        let report = result.unwrap();
        assert_eq!(report.order_side, OrderSide::Buy);
        assert_eq!(report.liquidity_side, LiquiditySide::Taker);
    }

    #[rstest]
    fn test_parse_ws_fill_report_with_liquidation() {
        let instrument = create_test_instrument();
        let account_id = AccountId::new("HYPERLIQUID-001");
        let ts_init = UnixNanos::default();

        let fill_data = WsFillData {
            coin: Ustr::from("BTC"),
            px: dec!(50000.0),
            sz: dec!(0.1),
            side: HyperliquidSide::Sell,
            time: 1704470400000,
            start_position: dec!(0.1),
            dir: HyperliquidFillDirection::CloseLong,
            closed_pnl: dec!(-25.0),
            hash: "0xdef456".to_string(),
            oid: 54321,
            crossed: true,
            fee: dec!(0.0),
            tid: 12345,
            liquidation: Some(FillLiquidationData {
                liquidated_user: Some("0xuser".to_string()),
                mark_px: dec!(50000.0),
                method: HyperliquidLiquidationMethod::Market,
            }),
            fee_token: Ustr::from("USDC"),
            builder_fee: None,
            cloid: None,
            twap_id: None,
        };

        let report = parse_ws_fill_report(&fill_data, &instrument, account_id, ts_init).unwrap();

        // The fill is still emitted through the standard path; the liquidation
        // metadata is logged for observability rather than encoded on the report.
        assert_eq!(report.order_side, OrderSide::Sell);
        assert_eq!(report.liquidity_side, LiquiditySide::Taker);
        assert_eq!(report.venue_order_id.to_string(), "54321");
    }

    #[rstest]
    fn test_parse_ws_fill_report_outcome_round_trip() {
        use crate::http::{
            models::{OutcomeMarket, OutcomeMeta},
            parse::{create_instrument_from_def, parse_outcome_instruments},
        };

        let meta = OutcomeMeta {
            outcomes: vec![OutcomeMarket {
                outcome: 99,
                name: "BTC daily".to_string(),
                description: String::new(),
                side_specs: vec![],
            }],
            questions: vec![],
        };

        let defs = parse_outcome_instruments(&meta).unwrap();
        let instrument = create_instrument_from_def(&defs[0], UnixNanos::default()).unwrap();
        assert_eq!(instrument.id().symbol.as_str(), "99-YES-OUTCOME");

        let fill_data = WsFillData {
            coin: Ustr::from("#990"),
            px: dec!(0.4500),
            sz: dec!(1500.00),
            side: HyperliquidSide::Buy,
            time: 1_704_470_400_000,
            start_position: dec!(0.00),
            dir: HyperliquidFillDirection::OpenLong,
            closed_pnl: dec!(0.0),
            hash: "0xabc789".to_string(),
            oid: 42_42,
            crossed: true,
            fee: dec!(0.0),
            tid: 7777,
            liquidation: None,
            fee_token: Ustr::from("+990"),
            builder_fee: None,
            cloid: None,
            twap_id: None,
        };

        let report = parse_ws_fill_report(
            &fill_data,
            &instrument,
            AccountId::new("HYPERLIQUID-001"),
            UnixNanos::default(),
        )
        .unwrap();

        // Zero-fee outcome fills fall back to the instrument's quote currency
        // (USDH) instead of the unregistered side token, keeping downstream
        // OrderFilled events and persistence on a registered currency.
        assert_eq!(report.commission.currency.code, "USDH");
        assert!(report.commission.as_decimal().is_zero());
        assert_eq!(report.order_side, OrderSide::Buy);
    }

    #[rstest]
    fn test_parse_ws_order_book_deltas_snapshot_behavior() {
        let instrument = create_test_instrument();
        let ts_init = UnixNanos::default();

        let book = WsBookData {
            coin: Ustr::from("BTC"),
            levels: [
                vec![WsLevelData {
                    px: dec!(50000.0),
                    sz: dec!(1.0),
                    n: 1,
                }],
                vec![WsLevelData {
                    px: dec!(50001.0),
                    sz: dec!(2.0),
                    n: 1,
                }],
            ],
            time: 1_704_470_400_000,
        };

        let deltas = parse_ws_order_book_deltas(&book, &instrument, ts_init).unwrap();

        assert_eq!(deltas.deltas.len(), 3); // clear + bid + ask
        assert_eq!(deltas.deltas[0].action, BookAction::Clear);

        let bid_delta = &deltas.deltas[1];
        assert_eq!(bid_delta.action, BookAction::Add);
        assert_eq!(bid_delta.order.side, OrderSide::Buy.into());
        assert!(bid_delta.order.size.is_positive());
        assert_eq!(bid_delta.order.order_id, 0);

        let ask_delta = &deltas.deltas[2];
        assert_eq!(ask_delta.action, BookAction::Add);
        assert_eq!(ask_delta.order.side, OrderSide::Sell.into());
        assert!(ask_delta.order.size.is_positive());
        assert_eq!(ask_delta.order.order_id, 0);
    }

    #[rstest]
    fn test_parse_ws_order_book_depth_preserves_sparse_book() {
        let instrument = create_test_instrument();
        let ts_init = UnixNanos::from(123);

        let book = WsBookData {
            coin: Ustr::from("BTC"),
            levels: [
                vec![
                    WsLevelData {
                        px: dec!(100.00),
                        sz: dec!(1.0),
                        n: 2,
                    },
                    WsLevelData {
                        px: dec!(99.99),
                        sz: dec!(2.0),
                        n: 3,
                    },
                    WsLevelData {
                        px: dec!(99.98),
                        sz: dec!(3.0),
                        n: 1,
                    },
                ],
                vec![
                    WsLevelData {
                        px: dec!(100.01),
                        sz: dec!(1.5),
                        n: 1,
                    },
                    WsLevelData {
                        px: dec!(100.02),
                        sz: dec!(2.5),
                        n: 4,
                    },
                ],
            ],
            time: 1_704_470_400_000,
        };

        let depth = parse_ws_order_book_depth(&book, &instrument, ts_init).unwrap();

        let expected_bids = [("100.00", "1.000"), ("99.99", "2.000"), ("99.98", "3.000")];
        let expected_asks = [("100.01", "1.500"), ("100.02", "2.500")];

        assert_eq!(depth.instrument_id, instrument.id());
        assert_eq!(depth.bids.len(), expected_bids.len());
        assert_eq!(depth.asks.len(), expected_asks.len());
        assert_eq!(depth.bid_counts.as_slice(), &[2, 3, 1]);
        assert_eq!(depth.ask_counts.as_slice(), &[1, 4]);
        for (order, (price, size)) in depth.bids.iter().zip(expected_bids) {
            assert_eq!(order.side, Some(OrderSide::Buy));
            assert_eq!(order.price, Price::from(price));
            assert_eq!(order.size, Quantity::from(size));
            assert_eq!(order.order_id, 0);
        }

        for (order, (price, size)) in depth.asks.iter().zip(expected_asks) {
            assert_eq!(order.side, Some(OrderSide::Sell));
            assert_eq!(order.price, Price::from(price));
            assert_eq!(order.size, Quantity::from(size));
            assert_eq!(order.order_id, 0);
        }
        assert_eq!(depth.sequence, 0);
        assert_eq!(depth.ts_init, ts_init);

        // Snapshot flag set
        assert_eq!(depth.flags, RecordFlag::F_SNAPSHOT as u8);
        assert_eq!(
            depth.ts_event,
            UnixNanos::from(1_704_470_400_000 * 1_000_000)
        );
    }

    #[rstest]
    fn test_parse_ws_order_book_depth_truncates_beyond_10() {
        let instrument = create_test_instrument();
        let ts_init = UnixNanos::default();

        let mk_levels = |base: f64, n: usize| -> Vec<WsLevelData> {
            (0..n)
                .map(|i| WsLevelData {
                    px: Decimal::from_str(&format!("{:.2}", base - i as f64 * 0.01)).unwrap(),
                    sz: dec!(1.0),
                    n: 1,
                })
                .collect()
        };

        let book = WsBookData {
            coin: Ustr::from("BTC"),
            levels: [mk_levels(100.00, 15), mk_levels(100.50, 12)],
            time: 1_704_470_400_000,
        };

        let depth = parse_ws_order_book_depth(&book, &instrument, ts_init).unwrap();

        // Only first 10 on each side retained
        for i in 0..10 {
            assert!(
                !depth.bids[i].size.is_zero(),
                "bid slot {i} unexpectedly empty"
            );
            assert!(
                !depth.asks[i].size.is_zero(),
                "ask slot {i} unexpectedly empty"
            );
        }
    }

    #[rstest]
    fn test_parse_ws_asset_context_perp() {
        let instrument = create_test_instrument();
        let ts_init = UnixNanos::default();

        let ctx_data = WsActiveAssetCtxData::Perp {
            coin: Ustr::from("BTC"),
            ctx: PerpsAssetCtx {
                shared: SharedAssetCtx {
                    day_ntl_vlm: dec!(1000000.0),
                    prev_day_px: dec!(49000.0),
                    mark_px: dec!(50000.0),
                    mid_px: Some(dec!(50001.0)),
                    impact_pxs: Some(vec!["50000.0".to_string(), "50002.0".to_string()]),
                    day_base_vlm: Some(dec!(100.0)),
                },
                funding: dec!(0.0001),
                open_interest: dec!(100000.0),
                oracle_px: dec!(50005.0),
                premium: Some(dec!(-0.0001)),
            },
        };

        let result = parse_ws_asset_context(&ctx_data, &instrument, ts_init);
        assert!(result.is_ok());

        let (mark_price, index_price, funding_rate) = result.unwrap();

        assert_eq!(mark_price.instrument_id, instrument.id());
        assert_eq!(mark_price.value.as_f64(), 50_000.0);

        assert!(index_price.is_some());
        let index = index_price.unwrap();
        assert_eq!(index.instrument_id, instrument.id());
        assert_eq!(index.value.as_f64(), 50_005.0);

        assert!(funding_rate.is_some());
        let funding = funding_rate.unwrap();
        assert_eq!(funding.instrument_id, instrument.id());
        assert_eq!(funding.rate.to_string(), "0.0001");
        assert_eq!(funding.interval, Some(60));
    }

    #[rstest]
    fn test_parse_ws_asset_context_spot() {
        let instrument = create_test_instrument();
        let ts_init = UnixNanos::default();

        let ctx_data = WsActiveAssetCtxData::Spot {
            coin: Ustr::from("BTC"),
            ctx: SpotAssetCtx {
                shared: SharedAssetCtx {
                    day_ntl_vlm: dec!(1000000.0),
                    prev_day_px: dec!(49000.0),
                    mark_px: dec!(50000.0),
                    mid_px: Some(dec!(50001.0)),
                    impact_pxs: Some(vec!["50000.0".to_string(), "50002.0".to_string()]),
                    day_base_vlm: Some(dec!(100.0)),
                },
                circulating_supply: dec!(19000000.0),
            },
        };

        let result = parse_ws_asset_context(&ctx_data, &instrument, ts_init);
        assert!(result.is_ok());

        let (mark_price, index_price, funding_rate) = result.unwrap();

        assert_eq!(mark_price.instrument_id, instrument.id());
        assert_eq!(mark_price.value.as_f64(), 50_000.0);
        assert!(index_price.is_none());
        assert!(funding_rate.is_none());
    }

    /// Pins the direct `Decimal::from_str` path for the funding rate. An f64
    /// round-trip (the prior implementation) cannot represent these values
    /// exactly, so the parsed Decimal would diverge from the input string.
    #[rstest]
    #[case::positive_high_precision("0.0001234567890123456")]
    #[case::negative_high_precision("-0.0001234567890123456")]
    fn test_parse_ws_asset_context_perp_preserves_funding_precision(#[case] funding_str: &str) {
        let instrument = create_test_instrument();
        let ts_init = UnixNanos::default();

        let expected = Decimal::from_str(funding_str).unwrap();

        let ctx_data = WsActiveAssetCtxData::Perp {
            coin: Ustr::from("BTC"),
            ctx: PerpsAssetCtx {
                shared: SharedAssetCtx {
                    day_ntl_vlm: dec!(1000000.0),
                    prev_day_px: dec!(49000.0),
                    mark_px: dec!(50000.0),
                    mid_px: None,
                    impact_pxs: None,
                    day_base_vlm: None,
                },
                funding: Decimal::from_str(funding_str).unwrap(),
                open_interest: dec!(100000.0),
                oracle_px: dec!(50005.0),
                premium: None,
            },
        };

        let (_, _, funding_rate) = parse_ws_asset_context(&ctx_data, &instrument, ts_init).unwrap();

        let funding = funding_rate.expect("perp ctx must yield funding rate");
        assert_eq!(funding.rate, expected);
    }

    #[rstest]
    fn test_parse_ws_open_interest_perp() {
        let instrument = create_test_instrument();
        let ts_init = UnixNanos::default();

        let open_interest = parse_ws_open_interest(dec!(100000.0), &instrument, ts_init).unwrap();

        assert_eq!(open_interest.instrument_id, instrument.id());
        assert_eq!(open_interest.open_interest.to_string(), "100000.0");
        assert_eq!(open_interest.ts_event, ts_init);
        assert_eq!(open_interest.ts_init, ts_init);
    }

    #[rstest]
    #[case::round("100000.0")]
    #[case::precise("100000.123456789")]
    fn test_parse_ws_open_interest_preserves_precision(#[case] open_interest_str: &str) {
        let instrument = create_test_instrument();
        let ts_init = UnixNanos::default();

        let expected = Decimal::from_str(open_interest_str).unwrap();

        let open_interest = parse_ws_open_interest(
            Decimal::from_str(open_interest_str).unwrap(),
            &instrument,
            ts_init,
        )
        .unwrap();

        assert_eq!(open_interest.open_interest, expected);
    }

    #[rstest]
    fn test_parse_ws_twap_history_row_from_live_mainnet_fixture() {
        let fixture = include_str!("../../test_data/ws_user_twap_history.json");
        let msg: crate::websocket::messages::HyperliquidWsMessage =
            serde_json::from_str(fixture).expect("fixture should deserialize");
        let crate::websocket::messages::HyperliquidWsMessage::UserTwapHistory { data } = msg else {
            panic!("expected UserTwapHistory");
        };
        let ts_init = UnixNanos::from(99);
        let is_snapshot = data.is_snapshot.unwrap_or(false);

        let row =
            parse_ws_twap_history_row(&data.history[0], &data.user, is_snapshot, None, ts_init)
                .unwrap();

        assert!(row.is_snapshot);
        assert_eq!(row.user, data.user);
        assert_eq!(row.coin, "xyz:HOOD");
        assert_eq!(row.twap_id, Some(2081397));
        assert!(row.instrument_id.is_none());
        assert_eq!(row.side, OrderSide::Buy);
        assert_eq!(row.size.to_string(), "100.0");
        assert_eq!(row.executed_size.to_string(), "100.0");
        assert_eq!(row.minutes, 240);
        assert!(!row.randomize);
        assert!(!row.reduce_only);
        assert_eq!(
            row.status,
            crate::common::enums::HyperliquidTwapStatus::Finished
        );
        assert!(row.status_description.is_empty());
        // Live mainnet history.time is seconds (not milliseconds).
        assert_eq!(row.ts_event, UnixNanos::from(1_785_848_057_000_000_000));
        assert_eq!(row.ts_init, ts_init);
    }

    #[rstest]
    fn test_parse_ws_twap_slice_fill_from_live_mainnet_fixture() {
        let fixture = include_str!("../../test_data/ws_user_twap_slice_fills.json");
        let msg: crate::websocket::messages::HyperliquidWsMessage =
            serde_json::from_str(fixture).expect("fixture should deserialize");
        let crate::websocket::messages::HyperliquidWsMessage::UserTwapSliceFills { data } = msg
        else {
            panic!("expected UserTwapSliceFills");
        };
        let instrument = create_test_instrument();
        let ts_init = UnixNanos::from(99);
        let is_snapshot = data.is_snapshot.unwrap_or(false);

        let fill = parse_ws_twap_slice_fill(
            &data.twap_slice_fills[0],
            &data.user,
            is_snapshot,
            Some(&instrument),
            ts_init,
        )
        .unwrap();

        assert!(fill.is_snapshot);
        assert_eq!(fill.twap_id, 2_087_225);
        assert_eq!(
            fill.hash,
            "0x0000000000000000000000000000000000000000000000000000000000000000"
        );
        assert_eq!(fill.coin, "BTC");
        assert_eq!(fill.instrument_id, Some(instrument.id()));
        assert_eq!(fill.side, OrderSide::Buy);
        assert_eq!(fill.price.to_string(), "64597.0");
        assert!(fill.crossed);
    }
}