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

//! Functions translating raw OKX WebSocket frames into Nautilus data types.

use std::{str::FromStr, sync::LazyLock};

use ahash::{AHashMap, AHashSet};
use anyhow::Context;
use nautilus_common::cache::fifo::FifoCacheMap;
use nautilus_core::{UUID4, nanos::UnixNanos};
use nautilus_model::{
    data::{
        Bar, BarSpecification, BarType, BookOrder, Data, FundingRateUpdate, IndexPriceUpdate,
        InstrumentStatus, MarkPriceUpdate, OptionGreekValues, OrderBookDelta, OrderBookDeltas,
        OrderBookDepth, QuoteTick, TradeTick, option_chain::OptionGreeks,
    },
    enums::{
        AggregationSource, AggressorSide, BookAction, LiquiditySide, OrderSide, OrderStatus,
        OrderType, RecordFlag, TimeInForce, TrailingOffsetType, TriggerType,
    },
    events::{OrderAccepted, OrderCanceled, OrderExpired, OrderTriggered, OrderUpdated},
    identifiers::{
        AccountId, ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, VenueOrderId,
    },
    instruments::{Instrument, InstrumentAny},
    reports::{FillReport, OrderStatusReport},
    types::{Money, Price, Quantity},
};
use parking_lot::Mutex;
use rust_decimal::Decimal;
use ustr::Ustr;

use super::{
    enums::OKXWsChannel,
    messages::{
        OKXAlgoOrderMsg, OKXBookMsg, OKXCandleMsg, OKXIndexPriceMsg, OKXMarkPriceMsg,
        OKXOptionSummaryMsg, OKXOrderMsg, OKXRpiBookMsg, OKXTickerMsg, OKXTradeMsg, OrderBookEntry,
    },
};
use crate::{
    common::{
        consts::{OKX_POST_ONLY_CANCEL_REASON, OKX_POST_ONLY_CANCEL_SOURCE},
        enums::{
            OKXAlgoOrderStatus, OKXAlgoOrderType, OKXBookAction, OKXCandleConfirm, OKXGreeksType,
            OKXInstrumentStatus, OKXInstrumentType, OKXOrderCategory, OKXOrderStatus, OKXOrderType,
            OKXSide, OKXTargetCurrency, OKXTriggerType,
        },
        models::OKXInstrument,
        parse::{
            determine_order_type_with_alt, is_market_price, okx_channel_to_bar_spec,
            okx_status_to_market_action, parse_fee, parse_fee_currency, parse_funding_rate_msg,
            parse_instrument_any, parse_instrument_id, parse_message_vec,
            parse_millisecond_timestamp, parse_parent_client_order_id, parse_price, parse_quantity,
            parse_spread_order_status_report as parse_common_spread_order_status_report,
        },
    },
    http::models::OKXSpreadOrder,
    websocket::messages::{ExecutionReport, NautilusWsMessage, OKXFundingRateMsg},
};

/// Maximum terminal-order baselines retained for reconnect replay before the oldest is evicted.
const TERMINAL_BASELINE_CAPACITY: usize = 10_000;

/// Cumulative fee per venue order ID (`ordId`).
pub type FeeCache = FillBaselineCache<Money>;

/// Cumulative filled quantity per venue order ID (`ordId`).
pub type FilledQtyCache = FillBaselineCache<Quantity>;

/// Cumulative per-order baseline used to derive incremental fill values.
///
/// OKX reports `fee` and `accFillSz` cumulatively, so a fill's incremental commission derives
/// from the previous message for the same `ordId`, as does its quantity whenever `fillSz` is
/// absent. Losing an open order's baseline would charge its whole cumulative fee, and its whole
/// accumulated quantity, to the next fill.
///
/// Baselines are therefore split by order lifecycle. An order that can still fill keeps its
/// baseline in full. A terminal order keeps one only to absorb reconnect replays, so those are
/// held in a bounded cache that evicts oldest-first and cannot grow with session length.
#[derive(Debug)]
pub struct FillBaselineCache<T> {
    open: AHashMap<Ustr, T>,
    terminal: FifoCacheMap<Ustr, T, TERMINAL_BASELINE_CAPACITY>,
}

impl<T> Default for FillBaselineCache<T> {
    fn default() -> Self {
        Self::new()
    }
}

impl<T> FillBaselineCache<T> {
    /// Creates a new empty [`FillBaselineCache`].
    #[must_use]
    pub fn new() -> Self {
        Self {
            open: AHashMap::new(),
            terminal: FifoCacheMap::new(),
        }
    }

    /// Returns the cumulative baseline recorded for `ord_id`.
    #[must_use]
    pub fn get(&self, ord_id: &Ustr) -> Option<&T> {
        self.open.get(ord_id).or_else(|| self.terminal.get(ord_id))
    }

    /// Records the cumulative baseline for `ord_id`.
    ///
    /// Pass `is_terminal` for an order state the venue cannot follow with a new fill; its
    /// baseline then becomes evictable. An open order's baseline is retained in full.
    pub fn record(&mut self, ord_id: Ustr, value: T, is_terminal: bool) {
        if is_terminal {
            self.open.remove(&ord_id);
            self.terminal.insert(ord_id, value);
        } else if self.terminal.contains_key(&ord_id) {
            self.terminal.insert(ord_id, value);
        } else {
            self.open.insert(ord_id, value);
        }
    }

    /// Drops any baseline recorded for `ord_id`.
    pub fn remove(&mut self, ord_id: &Ustr) {
        self.open.remove(ord_id);
        self.terminal.remove(ord_id);
    }
}

/// Returns whether the venue order state admits no further fills.
#[must_use]
pub const fn is_terminal_order_state(state: OKXOrderStatus) -> bool {
    matches!(
        state,
        OKXOrderStatus::Filled | OKXOrderStatus::Canceled | OKXOrderStatus::MmpCanceled
    )
}

/// Extracts fee rates from a cached instrument.
///
/// Returns a tuple of (`margin_init`, `margin_maint`, `maker_fee`, `taker_fee`).
/// All values are None if the instrument type doesn't support fees.
pub(crate) fn extract_fees_from_cached_instrument(
    instrument: &InstrumentAny,
) -> (
    Option<Decimal>,
    Option<Decimal>,
    Option<Decimal>,
    Option<Decimal>,
) {
    match instrument {
        InstrumentAny::CurrencyPair(pair) => (
            Some(pair.margin_init),
            Some(pair.margin_maint),
            Some(pair.maker_fee),
            Some(pair.taker_fee),
        ),
        InstrumentAny::CryptoPerpetual(perp) => (
            Some(perp.margin_init),
            Some(perp.margin_maint),
            Some(perp.maker_fee),
            Some(perp.taker_fee),
        ),
        InstrumentAny::CryptoFuture(future) => (
            Some(future.margin_init),
            Some(future.margin_maint),
            Some(future.maker_fee),
            Some(future.taker_fee),
        ),
        InstrumentAny::CryptoOption(option) => (
            Some(option.margin_init),
            Some(option.margin_maint),
            Some(option.maker_fee),
            Some(option.taker_fee),
        ),
        _ => (None, None, None, None),
    }
}

/// Represents the result of parsing an OKX order message into a specific event.
#[derive(Debug, Clone)]
pub enum ParsedOrderEvent {
    /// Order has been accepted by the venue.
    Accepted(OrderAccepted),
    /// Order has been canceled.
    Canceled(OrderCanceled),
    /// Order has expired (e.g., GTD order reached expiration time).
    Expired(OrderExpired),
    /// Stop/algo order has been triggered.
    Triggered(OrderTriggered),
    /// Order has been modified (price, quantity, or venue order ID changed).
    Updated(OrderUpdated),
    /// Order fill event.
    Fill(FillReport),
    /// Status update that doesn't map to a specific event (for reconciliation/external orders).
    StatusOnly(Box<OrderStatusReport>),
    /// Duplicate message detected (e.g. reconnect replay with unchanged fill).
    /// The dispatcher should update caches but not emit any event.
    Skipped,
}

/// Snapshot of order state for detecting updates.
#[derive(Debug, Clone)]
pub struct OrderStateSnapshot {
    pub venue_order_id: VenueOrderId,
    pub quantity: Quantity,
    pub price: Option<Price>,
}

/// Parses an OKX order message into a specific order event.
///
/// This function determines the appropriate event type based on:
/// - Current order status from OKX
/// - Whether there's a new fill
/// - Whether the order was updated (price/quantity change)
/// - Whether it's an algo order that triggered
///
/// # Errors
///
/// Returns an error if parsing order identifiers or numeric fields fails.
#[expect(clippy::too_many_arguments)]
pub fn parse_order_event(
    msg: &OKXOrderMsg,
    client_order_id: ClientOrderId,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    instrument: &InstrumentAny,
    previous_fee: Option<Money>,
    previous_filled_qty: Option<Quantity>,
    previous_state: Option<&OrderStateSnapshot>,
    ts_init: UnixNanos,
) -> anyhow::Result<ParsedOrderEvent> {
    let venue_order_id = VenueOrderId::new(msg.ord_id);
    let instrument_id = instrument.id();

    let has_new_fill = (!msg.fill_sz.is_empty() && msg.fill_sz != "0")
        || !msg.trade_id.is_empty()
        || has_acc_fill_sz_increased(
            msg.acc_fill_sz.as_deref(),
            previous_filled_qty,
            instrument.size_precision(),
        );

    warn_unrecognized_order_state(msg);

    // Check for order updates, but skip when other events take precedence:
    // - Fill events: fill data must be processed, update detection secondary
    // - Terminal states: handled by specific branches below
    let skip_update_check = has_new_fill
        || matches!(
            msg.state,
            OKXOrderStatus::Filled | OKXOrderStatus::Canceled | OKXOrderStatus::MmpCanceled
        );

    if !skip_update_check
        && let Some(prev) = previous_state
        && is_order_updated_excluding_venue_id_for_live(msg, prev, instrument)?
    {
        let ts_event = parse_millisecond_timestamp(msg.u_time);
        let quantity = parse_quantity(&msg.sz, instrument.size_precision())?;
        let price = if is_market_price(&msg.px) {
            None
        } else {
            Some(parse_price(&msg.px, instrument.price_precision())?)
        };

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

    match msg.state {
        OKXOrderStatus::Filled | OKXOrderStatus::PartiallyFilled | OKXOrderStatus::Unknown
            if has_new_fill =>
        {
            match parse_fill_report(
                msg,
                instrument,
                account_id,
                previous_fee,
                previous_filled_qty,
                ts_init,
            )? {
                Some(mut report) => {
                    report.client_order_id = Some(client_order_id);
                    Ok(ParsedOrderEvent::Fill(report))
                }
                None => Ok(ParsedOrderEvent::Skipped),
            }
        }
        OKXOrderStatus::Live => {
            let ts_event = parse_millisecond_timestamp(msg.c_time);
            Ok(ParsedOrderEvent::Accepted(OrderAccepted::new(
                trader_id,
                strategy_id,
                instrument_id,
                client_order_id,
                venue_order_id,
                account_id,
                UUID4::new(),
                ts_event,
                ts_init,
                false, // reconciliation
            )))
        }
        OKXOrderStatus::Canceled | OKXOrderStatus::MmpCanceled => {
            let ts_event = parse_millisecond_timestamp(msg.u_time);

            if is_order_expired_by_reason(msg) {
                Ok(ParsedOrderEvent::Expired(OrderExpired::new(
                    trader_id,
                    strategy_id,
                    instrument_id,
                    client_order_id,
                    UUID4::new(),
                    ts_event,
                    ts_init,
                    false,
                    Some(venue_order_id),
                    Some(account_id),
                )))
            } else {
                Ok(ParsedOrderEvent::Canceled(OrderCanceled::new(
                    trader_id,
                    strategy_id,
                    instrument_id,
                    client_order_id,
                    UUID4::new(),
                    ts_event,
                    ts_init,
                    false,
                    Some(venue_order_id),
                    Some(account_id),
                    None,
                )))
            }
        }
        _ => {
            // PartiallyFilled without new fill or other states - use status report
            parse_order_status_report(msg, instrument, account_id, ts_init)
                .map(|r| ParsedOrderEvent::StatusOnly(Box::new(r)))
        }
    }
}

/// Parses an OKX spread order message into a specific order event.
///
/// # Errors
///
/// Returns an error if parsing order identifiers or numeric fields fails.
#[expect(clippy::too_many_arguments)]
pub fn parse_spread_order_event(
    msg: &OKXSpreadOrder,
    client_order_id: ClientOrderId,
    account_id: AccountId,
    trader_id: TraderId,
    strategy_id: StrategyId,
    instrument: &InstrumentAny,
    previous_filled_qty: Option<Quantity>,
    previous_state: Option<&OrderStateSnapshot>,
    ts_init: UnixNanos,
) -> anyhow::Result<ParsedOrderEvent> {
    let venue_order_id = VenueOrderId::new(msg.ord_id.as_str());
    let instrument_id = instrument.id();
    let has_new_fill = (!msg.fill_sz.is_empty() && msg.fill_sz != "0")
        || !msg.trade_id.is_empty()
        || has_acc_fill_sz_increased(
            Some(msg.acc_fill_sz.as_str()),
            previous_filled_qty,
            instrument.size_precision(),
        );
    let skip_update_check = has_new_fill
        || matches!(
            msg.state,
            OKXOrderStatus::Filled | OKXOrderStatus::Canceled | OKXOrderStatus::MmpCanceled
        );

    if !skip_update_check
        && let Some(prev) = previous_state
        && is_spread_order_updated_excluding_venue_id_for_live(msg, prev, instrument)?
    {
        let ts_event = msg.u_time.map_or(ts_init, parse_millisecond_timestamp);
        let quantity = parse_quantity(&msg.sz, instrument.size_precision())?;
        let price = if is_market_price(&msg.px) {
            None
        } else {
            Some(parse_price(&msg.px, instrument.price_precision())?)
        };

        return Ok(ParsedOrderEvent::Updated(OrderUpdated::new(
            trader_id,
            strategy_id,
            instrument_id,
            client_order_id,
            quantity,
            UUID4::new(),
            ts_event,
            ts_init,
            false,
            Some(venue_order_id),
            Some(account_id),
            price,
            None,
            None,
            false,
        )));
    }

    match msg.state {
        OKXOrderStatus::Filled | OKXOrderStatus::PartiallyFilled if has_new_fill => {
            match parse_spread_order_fill_report(
                msg,
                instrument,
                account_id,
                previous_filled_qty,
                ts_init,
            )? {
                Some(report) => Ok(ParsedOrderEvent::Fill(report)),
                None => Ok(ParsedOrderEvent::Skipped),
            }
        }
        OKXOrderStatus::Live => {
            let ts_event = msg.c_time.map_or(ts_init, parse_millisecond_timestamp);
            Ok(ParsedOrderEvent::Accepted(OrderAccepted::new(
                trader_id,
                strategy_id,
                instrument_id,
                client_order_id,
                venue_order_id,
                account_id,
                UUID4::new(),
                ts_event,
                ts_init,
                false,
            )))
        }
        OKXOrderStatus::Canceled | OKXOrderStatus::MmpCanceled => {
            let ts_event = msg
                .u_time
                .or(msg.c_time)
                .map_or(ts_init, parse_millisecond_timestamp);
            Ok(ParsedOrderEvent::Canceled(OrderCanceled::new(
                trader_id,
                strategy_id,
                instrument_id,
                client_order_id,
                UUID4::new(),
                ts_event,
                ts_init,
                false,
                Some(venue_order_id),
                Some(account_id),
                None,
            )))
        }
        _ => parse_common_spread_order_status_report(
            msg,
            account_id,
            instrument.id(),
            instrument.price_precision(),
            instrument.size_precision(),
            ts_init,
        )
        .map(|report| ParsedOrderEvent::StatusOnly(Box::new(report))),
    }
}

/// Case-insensitive substring check.
#[inline]
/// Builds a deterministic synthesized `TradeId` for fills where OKX omits
/// the venue `trade_id`. Hashes the immutable fill fields with FNV-1a so
/// the result fits inside the 36-character `TradeId` cap and stays stable
/// across reconnect replays of the same physical fill.
fn synthesize_trade_id(msg: &OKXOrderMsg) -> String {
    const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
    const FNV_PRIME: u64 = 0x0100_0000_01b3;

    let mut hasher: u64 = FNV_OFFSET;
    let mut update = |bytes: &[u8]| {
        for byte in bytes {
            hasher ^= u64::from(*byte);
            hasher = hasher.wrapping_mul(FNV_PRIME);
        }
        // Field separator so that "ab" + "c" doesn't hash to the same as "a" + "bc".
        hasher ^= 0xff;
        hasher = hasher.wrapping_mul(FNV_PRIME);
    };

    update(msg.ord_id.as_bytes());
    update(msg.fill_time.to_string().as_bytes());
    update(msg.fill_sz.as_bytes());
    update(msg.fill_px.as_bytes());
    update(msg.acc_fill_sz.as_deref().unwrap_or("").as_bytes());

    format!("synth-{hasher:016x}")
}

fn contains_ignore_ascii_case(haystack: &str, needle: &str) -> bool {
    haystack
        .as_bytes()
        .windows(needle.len())
        .any(|window| window.eq_ignore_ascii_case(needle.as_bytes()))
}

/// Determines if a Canceled order is actually an Expired order based on cancel reason.
///
/// OKX's numeric `cancelSource` enumeration is not fully documented, so
/// unmapped codes log once via [`log_unknown_cancel_source`] to surface
/// candidates for the mapping.
fn is_order_expired_by_reason(msg: &OKXOrderMsg) -> bool {
    if let Some(ref reason) = msg.cancel_source_reason
        && (contains_ignore_ascii_case(reason, "expir")
            || contains_ignore_ascii_case(reason, "gtd")
            || contains_ignore_ascii_case(reason, "timeout"))
    {
        return true;
    }

    if let Some(ref source) = msg.cancel_source
        && (source == "5" || source == "time_expired" || source == "gtd_expired")
    {
        return true;
    }

    log_unknown_cancel_source(msg);
    false
}

const MAX_TRACKED_CANCEL_SOURCES: usize = 64;

/// Logs each unseen `(cancel_source, cancel_source_reason)` pair once,
/// capped at [`MAX_TRACKED_CANCEL_SOURCES`].
fn log_unknown_cancel_source(msg: &OKXOrderMsg) {
    static SEEN: LazyLock<Mutex<AHashSet<String>>> = LazyLock::new(|| Mutex::new(AHashSet::new()));
    log_unknown_cancel_source_inner(msg, &SEEN, MAX_TRACKED_CANCEL_SOURCES);
}

/// Test-injectable variant. Returns `true` when a new entry was recorded.
fn log_unknown_cancel_source_inner(
    msg: &OKXOrderMsg,
    seen: &Mutex<AHashSet<String>>,
    max_tracked: usize,
) -> bool {
    let source = msg.cancel_source.as_deref().unwrap_or("");
    let reason = msg.cancel_source_reason.as_deref().unwrap_or("");

    if source.is_empty() && reason.is_empty() {
        return false;
    }

    if matches!(source, "5" | "31" | "time_expired" | "gtd_expired") {
        return false;
    }

    let key = format!("{source}|{reason}");
    let mut seen = seen.lock();

    if seen.len() >= max_tracked {
        return false;
    }

    if seen.insert(key) {
        log::debug!("Observed unmapped OKX cancelSource: source='{source}', reason='{reason}'");
        true
    } else {
        false
    }
}

/// Checks if order parameters have been updated compared to previous state.
///
/// For Live status, `venue_id` changes are ignored because algo triggers create new IDs
/// but should emit `OrderAccepted`, not `OrderUpdated`. Price/quantity changes still count.
fn is_order_updated_excluding_venue_id_for_live(
    msg: &OKXOrderMsg,
    previous: &OrderStateSnapshot,
    instrument: &InstrumentAny,
) -> anyhow::Result<bool> {
    // For non-Live statuses, venue_id change indicates amendment
    if msg.state != OKXOrderStatus::Live {
        let current_venue_id = VenueOrderId::new(msg.ord_id);
        if previous.venue_order_id != current_venue_id {
            return Ok(true);
        }
    }

    let current_qty = parse_quantity(&msg.sz, instrument.size_precision())?;
    if previous.quantity != current_qty {
        return Ok(true);
    }

    // Price change only applies to limit orders
    if !is_market_price(&msg.px) {
        let current_price = parse_price(&msg.px, instrument.price_precision())?;

        if let Some(prev_price) = previous.price
            && prev_price != current_price
        {
            return Ok(true);
        }
    }

    Ok(false)
}

fn is_spread_order_updated_excluding_venue_id_for_live(
    msg: &OKXSpreadOrder,
    previous: &OrderStateSnapshot,
    instrument: &InstrumentAny,
) -> anyhow::Result<bool> {
    if msg.state != OKXOrderStatus::Live {
        let current_venue_id = VenueOrderId::new(msg.ord_id.as_str());
        if previous.venue_order_id != current_venue_id {
            return Ok(true);
        }
    }

    let current_qty = parse_quantity(&msg.sz, instrument.size_precision())?;
    if previous.quantity != current_qty {
        return Ok(true);
    }

    if !is_market_price(&msg.px) {
        let current_price = parse_price(&msg.px, instrument.price_precision())?;

        if let Some(prev_price) = previous.price
            && prev_price != current_price
        {
            return Ok(true);
        }
    }

    Ok(false)
}

/// Parses vector of OKX book messages into Nautilus order book deltas.
///
/// # Errors
///
/// Returns an error if any underlying book message cannot be parsed.
pub fn parse_book_msg_vec(
    data: Vec<OKXBookMsg>,
    instrument_id: &InstrumentId,
    price_precision: u8,
    size_precision: u8,
    action: OKXBookAction,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<Data>> {
    let mut deltas = Vec::with_capacity(data.len());

    for msg in data {
        deltas.push(Data::BookDeltas(Box::new(parse_book_msg(
            &msg,
            *instrument_id,
            price_precision,
            size_precision,
            &action,
            ts_init,
        )?)));
    }

    Ok(deltas)
}

/// Parses RPI book messages into Nautilus order book deltas.
///
/// # Errors
///
/// Returns an error if any RPI book message cannot be represented at the instrument precision.
pub fn parse_rpi_book_msg_vec(
    data: Vec<OKXRpiBookMsg>,
    instrument_id: &InstrumentId,
    price_precision: u8,
    size_precision: u8,
    action: OKXBookAction,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<Data>> {
    let mut deltas = Vec::with_capacity(data.len());

    for msg in data {
        deltas.push(Data::BookDeltas(Box::new(parse_rpi_book_msg(
            &msg,
            *instrument_id,
            price_precision,
            size_precision,
            &action,
            ts_init,
        )?)));
    }

    Ok(deltas)
}

/// Parses vector of OKX ticker messages into Nautilus quote ticks.
///
/// # Errors
///
/// Returns an error if any ticker message fails to parse.
pub fn parse_ticker_msg_vec(
    data: serde_json::Value,
    instrument_id: &InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<Data>> {
    parse_message_vec(
        data,
        |msg| {
            parse_ticker_msg(
                msg,
                *instrument_id,
                price_precision,
                size_precision,
                ts_init,
            )
        },
        Data::Quote,
    )
}

/// Parses vector of OKX book messages into Nautilus quote ticks.
///
/// # Errors
///
/// Returns an error if any quote message fails to parse.
pub fn parse_quote_msg_vec(
    data: serde_json::Value,
    instrument_id: &InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<Data>> {
    parse_message_vec(
        data,
        |msg| {
            parse_quote_msg(
                msg,
                *instrument_id,
                price_precision,
                size_precision,
                ts_init,
            )
        },
        Data::Quote,
    )
}

/// Parses vector of OKX trade messages into Nautilus trade ticks.
///
/// # Errors
///
/// Returns an error if any trade message fails to parse.
pub fn parse_trade_msg_vec(
    data: serde_json::Value,
    instrument_id: &InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<Data>> {
    parse_message_vec(
        data,
        |msg| {
            parse_trade_msg(
                msg,
                *instrument_id,
                price_precision,
                size_precision,
                ts_init,
            )
        },
        Data::Trade,
    )
}

/// Parses vector of OKX mark price messages into Nautilus mark price updates.
///
/// # Errors
///
/// Returns an error if any mark price message fails to parse.
pub fn parse_mark_price_msg_vec(
    data: serde_json::Value,
    instrument_id: &InstrumentId,
    price_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<Data>> {
    parse_message_vec(
        data,
        |msg| parse_mark_price_msg(msg, *instrument_id, price_precision, ts_init),
        Data::MarkPrice,
    )
}

/// Parses vector of OKX index price messages into Nautilus index price updates.
///
/// # Errors
///
/// Returns an error if any index price message fails to parse.
pub fn parse_index_price_msg_vec(
    data: serde_json::Value,
    instrument_id: &InstrumentId,
    price_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<Data>> {
    parse_message_vec(
        data,
        |msg| parse_index_price_msg(msg, *instrument_id, price_precision, ts_init),
        Data::IndexPrice,
    )
}

/// Parses vector of OKX funding rate messages into Nautilus funding rate updates.
/// Includes caching to filter out duplicate funding rates.
///
/// # Errors
///
/// Returns an error if any funding rate message fails to parse.
pub fn parse_funding_rate_msg_vec(
    data: serde_json::Value,
    instrument_id: &InstrumentId,
    ts_init: UnixNanos,
    funding_cache: &mut AHashMap<Ustr, (Ustr, u64)>,
) -> anyhow::Result<Vec<FundingRateUpdate>> {
    let msgs: Vec<OKXFundingRateMsg> = serde_json::from_value(data)?;

    let mut result = Vec::with_capacity(msgs.len());

    for msg in &msgs {
        let cache_key = (msg.funding_rate, msg.funding_time);

        if let Some(cached) = funding_cache.get(&msg.inst_id)
            && *cached == cache_key
        {
            continue; // Skip duplicate
        }

        // New or changed funding rate, update cache and parse
        funding_cache.insert(msg.inst_id, cache_key);
        let funding_rate = parse_funding_rate_msg(msg, *instrument_id, ts_init)?;
        result.push(funding_rate);
    }

    Ok(result)
}

/// Parses vector of OKX candle messages into Nautilus bars.
///
/// # Errors
///
/// Returns an error if candle messages cannot be deserialized or parsed.
pub fn parse_candle_msg_vec(
    data: serde_json::Value,
    instrument_id: &InstrumentId,
    price_precision: u8,
    size_precision: u8,
    spec: BarSpecification,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<Data>> {
    let msgs: Vec<OKXCandleMsg> = serde_json::from_value(data)?;
    let bar_type = BarType::new(*instrument_id, spec, AggregationSource::External);
    let mut bars = Vec::with_capacity(msgs.len());

    for msg in msgs {
        // Only process completed candles to avoid duplicate/partial bars
        if msg.confirm == OKXCandleConfirm::Closed {
            let bar = parse_candle_msg(&msg, bar_type, price_precision, size_precision, ts_init)?;
            bars.push(Data::Bar(bar));
        }
    }

    Ok(bars)
}

/// Parses vector of OKX book messages into Nautilus depth updates.
///
/// # Errors
///
/// Returns an error if any book depth message fails to parse.
pub fn parse_book_depth_msg_vec(
    data: Vec<OKXBookMsg>,
    instrument_id: &InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<Data>> {
    let mut depth_updates = Vec::with_capacity(data.len());

    for msg in data {
        let depth = parse_book_depth_msg(
            &msg,
            *instrument_id,
            price_precision,
            size_precision,
            ts_init,
        )?;
        depth_updates.push(Data::BookDepth(Box::new(depth)));
    }

    Ok(depth_updates)
}

/// Parses an OKX book message into Nautilus order book deltas.
///
/// # Errors
///
/// Returns an error if bid or ask levels contain values that cannot be parsed.
pub fn parse_book_msg(
    msg: &OKXBookMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    action: &OKXBookAction,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderBookDeltas> {
    let is_snapshot = action == &OKXBookAction::Snapshot;

    let flags = if is_snapshot {
        RecordFlag::F_SNAPSHOT as u8
    } else {
        0
    };
    let ts_event = parse_millisecond_timestamp(msg.ts);
    let mut deltas = Vec::with_capacity(msg.asks.len() + msg.bids.len() + usize::from(is_snapshot));

    if is_snapshot {
        deltas.push(OrderBookDelta::clear(
            instrument_id,
            msg.seq_id,
            ts_event,
            ts_init,
        ));
    }

    for bid in &msg.bids {
        let book_action = match action {
            OKXBookAction::Snapshot => BookAction::Add,
            _ => match bid.size.as_str() {
                "0" => BookAction::Delete,
                _ => BookAction::Update,
            },
        };
        let price = parse_price(&bid.price, price_precision)?;
        let size = parse_quantity(&bid.size, size_precision)?;
        let order_id = 0; // TBD
        let order = BookOrder::new(OrderSide::Buy, price, size, order_id);
        let delta = OrderBookDelta::new(
            instrument_id,
            book_action,
            order,
            flags,
            msg.seq_id,
            ts_event,
            ts_init,
        );
        deltas.push(delta);
    }

    for ask in &msg.asks {
        let book_action = match action {
            OKXBookAction::Snapshot => BookAction::Add,
            _ => match ask.size.as_str() {
                "0" => BookAction::Delete,
                _ => BookAction::Update,
            },
        };
        let price = parse_price(&ask.price, price_precision)?;
        let size = parse_quantity(&ask.size, size_precision)?;
        let order_id = 0; // TBD
        let order = BookOrder::new(OrderSide::Sell, price, size, order_id);
        let delta = OrderBookDelta::new(
            instrument_id,
            book_action,
            order,
            flags,
            msg.seq_id,
            ts_event,
            ts_init,
        );
        deltas.push(delta);
    }

    // The data engine only publishes a buffered batch once a delta closes the event
    if let Some(last) = deltas.last_mut() {
        last.flags |= RecordFlag::F_LAST as u8;
    }

    OrderBookDeltas::new_checked(instrument_id, deltas)
}

/// Parses an RPI book message into Nautilus order book deltas.
///
/// # Errors
///
/// Returns an error if a price or total quantity cannot be represented at the instrument precision.
pub fn parse_rpi_book_msg(
    msg: &OKXRpiBookMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    action: &OKXBookAction,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderBookDeltas> {
    let is_snapshot = action == &OKXBookAction::Snapshot;

    let flags = if is_snapshot {
        RecordFlag::F_SNAPSHOT as u8
    } else {
        0
    };
    let ts_event = parse_millisecond_timestamp(msg.ts);
    let mut deltas = Vec::with_capacity(msg.asks.len() + msg.bids.len() + usize::from(is_snapshot));

    if is_snapshot {
        deltas.push(OrderBookDelta::clear(
            instrument_id,
            msg.seq_id,
            ts_event,
            ts_init,
        ));
    }

    for bid in &msg.bids {
        let book_action = if action == &OKXBookAction::Snapshot {
            BookAction::Add
        } else if bid.1.is_zero() {
            BookAction::Delete
        } else {
            BookAction::Update
        };
        let price = Price::from_decimal_dp(bid.0, price_precision)?;
        let size = Quantity::from_decimal_dp(bid.1, size_precision)?;
        let order = BookOrder::new(OrderSide::Buy, price, size, 0);
        deltas.push(OrderBookDelta::new(
            instrument_id,
            book_action,
            order,
            flags,
            msg.seq_id,
            ts_event,
            ts_init,
        ));
    }

    for ask in &msg.asks {
        let book_action = if action == &OKXBookAction::Snapshot {
            BookAction::Add
        } else if ask.1.is_zero() {
            BookAction::Delete
        } else {
            BookAction::Update
        };
        let price = Price::from_decimal_dp(ask.0, price_precision)?;
        let size = Quantity::from_decimal_dp(ask.1, size_precision)?;
        let order = BookOrder::new(OrderSide::Sell, price, size, 0);
        deltas.push(OrderBookDelta::new(
            instrument_id,
            book_action,
            order,
            flags,
            msg.seq_id,
            ts_event,
            ts_init,
        ));
    }

    // The data engine only publishes a buffered batch once a delta closes the event
    if let Some(last) = deltas.last_mut() {
        last.flags |= RecordFlag::F_LAST as u8;
    }

    OrderBookDeltas::new_checked(instrument_id, deltas)
}

/// Parses an OKX book message into a Nautilus quote tick.
///
/// # Errors
///
/// Returns an error if any quote levels contain values that cannot be parsed.
pub fn parse_quote_msg(
    msg: &OKXBookMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<QuoteTick> {
    let best_bid: &OrderBookEntry = msg
        .bids
        .first()
        .ok_or_else(|| anyhow::anyhow!("Empty bids array for {instrument_id}"))?;
    let best_ask: &OrderBookEntry = msg
        .asks
        .first()
        .ok_or_else(|| anyhow::anyhow!("Empty asks array for {instrument_id}"))?;

    let bid_price = parse_price(&best_bid.price, price_precision)?;
    let ask_price = parse_price(&best_ask.price, price_precision)?;
    let bid_size = parse_quantity(&best_bid.size, size_precision)?;
    let ask_size = parse_quantity(&best_ask.size, size_precision)?;
    let ts_event = parse_millisecond_timestamp(msg.ts);

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

/// Parses an OKX book message into a Nautilus [`OrderBookDepth`].
///
/// Uses venue order counts for the snapshot's non-empty price levels.
///
/// # Errors
///
/// Returns an error if price, size, or order count fields cannot be parsed for any level.
pub fn parse_book_depth_msg(
    msg: &OKXBookMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderBookDepth> {
    let mut bids = Vec::with_capacity(msg.bids.len());
    let mut asks = Vec::with_capacity(msg.asks.len());
    let mut bid_counts = Vec::with_capacity(msg.bids.len());
    let mut ask_counts = Vec::with_capacity(msg.asks.len());

    for (levels, side, orders, counts) in [
        (&msg.bids, OrderSide::Buy, &mut bids, &mut bid_counts),
        (&msg.asks, OrderSide::Sell, &mut asks, &mut ask_counts),
    ] {
        for level in levels {
            orders.push(BookOrder::new(
                side,
                parse_price(&level.price, price_precision)?,
                parse_quantity(&level.size, size_precision)?,
                0,
            ));
            counts.push(
                level
                    .orders_count
                    .parse::<u32>()
                    .context("invalid OKX order count")?,
            );
        }
    }

    let ts_event = parse_millisecond_timestamp(msg.ts);

    Ok(OrderBookDepth::new(
        instrument_id,
        bids,
        asks,
        bid_counts,
        ask_counts,
        RecordFlag::F_SNAPSHOT as u8,
        msg.seq_id, // Use sequence ID for OKX L2 books
        ts_event,
        ts_init,
    ))
}

/// Parses an OKX ticker message into a Nautilus quote tick.
///
/// # Errors
///
/// Returns an error if bid or ask values cannot be parsed from the message.
pub fn parse_ticker_msg(
    msg: &OKXTickerMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<QuoteTick> {
    let bid_price = parse_price(&msg.bid_px, price_precision)?;
    let ask_price = parse_price(&msg.ask_px, price_precision)?;
    let bid_size = parse_quantity(&msg.bid_sz, size_precision)?;
    let ask_size = parse_quantity(&msg.ask_sz, size_precision)?;
    let ts_event = parse_millisecond_timestamp(msg.ts);

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

/// Parses an OKX trade message into a Nautilus trade tick.
///
/// # Errors
///
/// Returns an error if trade prices or sizes cannot be parsed.
pub fn parse_trade_msg(
    msg: &OKXTradeMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<TradeTick> {
    let price = parse_price(&msg.px, price_precision)?;
    let size = parse_quantity(&msg.sz, size_precision)?;
    let aggressor_side: AggressorSide = msg.side.into();
    let trade_id = TradeId::new(&msg.trade_id);
    let ts_event = parse_millisecond_timestamp(msg.ts);

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

/// Parses an OKX mark price message into a Nautilus mark price update.
///
/// # Errors
///
/// Returns an error if the mark price fails to parse.
pub fn parse_mark_price_msg(
    msg: &OKXMarkPriceMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<MarkPriceUpdate> {
    let price = parse_price(&msg.mark_px, price_precision)?;
    let ts_event = parse_millisecond_timestamp(msg.ts);

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

/// Parses an OKX index price message into a Nautilus index price update.
///
/// # Errors
///
/// Returns an error if the index price fails to parse.
pub fn parse_index_price_msg(
    msg: &OKXIndexPriceMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<IndexPriceUpdate> {
    let price = parse_price(&msg.idx_px, price_precision)?;
    let ts_event = parse_millisecond_timestamp(msg.ts);

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

/// Parses an OKX candle message into a Nautilus bar.
///
/// # Errors
///
/// Returns an error if candle price or volume fields cannot be parsed.
pub fn parse_candle_msg(
    msg: &OKXCandleMsg,
    bar_type: BarType,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
) -> anyhow::Result<Bar> {
    let open = parse_price(&msg.o, price_precision)?;
    let high = parse_price(&msg.h, price_precision)?;
    let low = parse_price(&msg.l, price_precision)?;
    let close = parse_price(&msg.c, price_precision)?;
    let volume = parse_quantity(&msg.vol, size_precision)?;
    let ts_event = parse_millisecond_timestamp(msg.ts);

    Bar::new_checked(bar_type, open, high, low, close, volume, ts_event, ts_init)
}

/// Parses vector of OKX order messages into Nautilus execution reports.
///
/// # Errors
///
/// Returns an error if any contained order messages cannot be parsed.
pub fn parse_order_msg_vec(
    data: &[OKXOrderMsg],
    account_id: AccountId,
    instruments: &AHashMap<Ustr, InstrumentAny>,
    fee_cache: &mut FeeCache,
    filled_qty_cache: &mut FilledQtyCache,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<ExecutionReport>> {
    let mut order_reports = Vec::with_capacity(data.len());

    for msg in data {
        match parse_order_msg(
            msg,
            account_id,
            instruments,
            fee_cache,
            filled_qty_cache,
            ts_init,
        ) {
            Ok(report) => {
                order_reports.push(report);

                if let Some(instrument) = instruments.get(&msg.inst_id) {
                    update_fee_fill_caches(msg, instrument, fee_cache, filled_qty_cache);
                }
            }
            Err(e) => log::error!("Failed to parse execution report from message: {e}"),
        }
    }

    Ok(order_reports)
}

/// Updates fee and fill caches from a raw OKX order message.
///
/// Call after parsing each message so subsequent messages in the same batch
/// see the correct cumulative fee and filled quantity.
pub fn update_fee_fill_caches(
    msg: &OKXOrderMsg,
    instrument: &InstrumentAny,
    fee_cache: &mut FeeCache,
    filled_qty_cache: &mut FilledQtyCache,
) {
    let is_terminal = is_terminal_order_state(msg.state);

    if let Some(ref fee_str) = msg.fee
        && !fee_str.is_empty()
    {
        let fee_dec = Decimal::from_str(fee_str).unwrap_or_default();
        let fee_ccy = parse_fee_currency(msg.fee_ccy.as_str(), fee_dec, || {
            format!("update_fee_fill_caches ord_id={}", msg.ord_id)
        });

        if let Ok(total_fee) = crate::common::parse::parse_fee(Some(fee_str.as_str()), fee_ccy) {
            fee_cache.record(msg.ord_id, total_fee, is_terminal);
        }
    }

    if let Some(ref acc_fill_sz) = msg.acc_fill_sz
        && !acc_fill_sz.is_empty()
        && acc_fill_sz != "0"
        && let Ok(qty) = parse_quantity(acc_fill_sz, instrument.size_precision())
    {
        filled_qty_cache.record(msg.ord_id, qty, is_terminal);
    }
}

/// Checks if `acc_fill_sz` has increased compared to the previous filled quantity.
fn has_acc_fill_sz_increased(
    acc_fill_sz: Option<&str>,
    previous_filled_qty: Option<Quantity>,
    size_precision: u8,
) -> bool {
    if let Some(acc_str) = acc_fill_sz {
        if acc_str.is_empty() || acc_str == "0" {
            return false;
        }

        if let Ok(current_filled) = parse_quantity(acc_str, size_precision) {
            if let Some(prev_qty) = previous_filled_qty {
                return current_filled > prev_qty;
            }
            return !current_filled.is_zero();
        }
    }
    false
}

/// Parses a single OKX order message into an [`ExecutionReport`].
///
/// # Errors
///
/// Returns an error if the instrument cannot be found or if parsing the
/// underlying order payload fails.
pub fn parse_order_msg(
    msg: &OKXOrderMsg,
    account_id: AccountId,
    instruments: &AHashMap<Ustr, InstrumentAny>,
    fee_cache: &FeeCache,
    filled_qty_cache: &FilledQtyCache,
    ts_init: UnixNanos,
) -> anyhow::Result<ExecutionReport> {
    let instrument = instruments
        .get(&msg.inst_id)
        .ok_or_else(|| anyhow::anyhow!("No instrument found for inst_id: {}", msg.inst_id))?;

    let previous_fee = fee_cache.get(&msg.ord_id).copied();
    let previous_filled_qty = filled_qty_cache.get(&msg.ord_id).copied();

    let has_new_fill = (!msg.fill_sz.is_empty() && msg.fill_sz != "0")
        || !msg.trade_id.is_empty()
        || has_acc_fill_sz_increased(
            msg.acc_fill_sz.as_deref(),
            previous_filled_qty,
            instrument.size_precision(),
        );

    warn_unrecognized_order_state(msg);

    match msg.state {
        OKXOrderStatus::Filled | OKXOrderStatus::PartiallyFilled | OKXOrderStatus::Unknown
            if has_new_fill =>
        {
            match parse_fill_report(
                msg,
                instrument,
                account_id,
                previous_fee,
                previous_filled_qty,
                ts_init,
            )? {
                Some(report) => Ok(ExecutionReport::Fill(report)),
                None => parse_order_status_report(msg, instrument, account_id, ts_init)
                    .map(ExecutionReport::Order),
            }
        }
        _ => parse_order_status_report(msg, instrument, account_id, ts_init)
            .map(ExecutionReport::Order),
    }
}

/// Logs a warning when an order message carries a state this build does not
/// recognize. Fill data on the message is still processed; only the status
/// classification is skipped.
fn warn_unrecognized_order_state(msg: &OKXOrderMsg) {
    if msg.state == OKXOrderStatus::Unknown {
        log::warn!(
            "Unrecognized order state: order_id={}, inst_id={}, processing any fill data and skipping status classification",
            msg.ord_id,
            msg.inst_id,
        );
    }
}

/// Parses a single OKX spread order message into an [`ExecutionReport`].
///
/// # Errors
///
/// Returns an error if the instrument cannot be found or if parsing the
/// underlying order payload fails.
pub fn parse_spread_order_msg(
    msg: &OKXSpreadOrder,
    account_id: AccountId,
    instruments: &AHashMap<Ustr, InstrumentAny>,
    filled_qty_cache: &FilledQtyCache,
    ts_init: UnixNanos,
) -> anyhow::Result<ExecutionReport> {
    let instrument = instruments
        .get(&msg.sprd_id)
        .ok_or_else(|| anyhow::anyhow!("No instrument found for sprd_id: {}", msg.sprd_id))?;
    let previous_filled_qty = filled_qty_cache.get(&msg.ord_id).copied();
    let has_new_fill = (!msg.fill_sz.is_empty() && msg.fill_sz != "0")
        || !msg.trade_id.is_empty()
        || has_acc_fill_sz_increased(
            Some(msg.acc_fill_sz.as_str()),
            previous_filled_qty,
            instrument.size_precision(),
        );

    match msg.state {
        OKXOrderStatus::Filled | OKXOrderStatus::PartiallyFilled if has_new_fill => {
            match parse_spread_order_fill_report(
                msg,
                instrument,
                account_id,
                previous_filled_qty,
                ts_init,
            )? {
                Some(report) => Ok(ExecutionReport::Fill(report)),
                None => parse_common_spread_order_status_report(
                    msg,
                    account_id,
                    instrument.id(),
                    instrument.price_precision(),
                    instrument.size_precision(),
                    ts_init,
                )
                .map(ExecutionReport::Order),
            }
        }
        _ => parse_common_spread_order_status_report(
            msg,
            account_id,
            instrument.id(),
            instrument.price_precision(),
            instrument.size_precision(),
            ts_init,
        )
        .map(ExecutionReport::Order),
    }
}

/// Parses an OKX algo order message into a Nautilus execution report.
///
/// # Errors
///
/// Returns an error if the instrument cannot be found or if message fields
/// fail to parse.
pub fn parse_algo_order_msg(
    msg: &OKXAlgoOrderMsg,
    account_id: AccountId,
    instruments: &AHashMap<Ustr, InstrumentAny>,
    ts_init: UnixNanos,
) -> anyhow::Result<Option<ExecutionReport>> {
    // Skip unsupported algo types (iceberg, smart_iceberg, twap, chase); their
    // triggered child orders still arrive on the regular orders channel
    if matches!(
        msg.ord_type,
        OKXAlgoOrderType::Iceberg
            | OKXAlgoOrderType::SmartIceberg
            | OKXAlgoOrderType::Twap
            | OKXAlgoOrderType::Chase
    ) {
        log::debug!("Skipping unsupported algo order type: {:?}", msg.ord_type);
        return Ok(None);
    }

    if msg.ord_type == OKXAlgoOrderType::Other {
        log::warn!(
            "Skipping algo order with unrecognized order type: algo_id={}, inst_id={}",
            msg.algo_id,
            msg.inst_id,
        );
        return Ok(None);
    }

    if msg.state == OKXAlgoOrderStatus::Unknown {
        log::warn!(
            "Skipping algo order with unrecognized state: algo_id={}, inst_id={}",
            msg.algo_id,
            msg.inst_id,
        );
        return Ok(None);
    }

    let inst = instruments
        .get(&msg.inst_id)
        .ok_or_else(|| anyhow::anyhow!("No instrument found for inst_id: {}", msg.inst_id))?;

    parse_algo_order_status_report(msg, inst, account_id, ts_init)
        .map(ExecutionReport::Order)
        .map(Some)
}

/// Parses an OKX algo order message into a Nautilus order status report.
///
/// # Errors
///
/// Returns an error if any order identifiers or numeric fields cannot be
/// parsed.
pub fn parse_algo_order_status_report(
    msg: &OKXAlgoOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderStatusReport> {
    let client_order_id = parse_parent_client_order_id(Some(&msg.algo_cl_ord_id), &msg.cl_ord_id);

    // For algo orders that haven't triggered, ord_id will be empty, use algo_id instead
    let venue_order_id = if msg.ord_id.is_empty() {
        VenueOrderId::new(msg.algo_id.as_str())
    } else {
        VenueOrderId::new(msg.ord_id.as_str())
    };

    let order_side = OrderSide::from(msg.side);

    let algo_fields = parse_algo_order_fields(msg)?;

    let status: OrderStatus = msg
        .state
        .try_into()
        .map_err(|e| anyhow::anyhow!("Unsupported OKX algo order status: {e}"))?;

    let quantity = parse_algo_order_quantity(msg, instrument)?;

    let filled_qty = if msg.state == OKXAlgoOrderStatus::Filled
        && !msg.actual_sz.is_empty()
        && msg.actual_sz != "0"
    {
        parse_quantity(msg.actual_sz.as_str(), instrument.size_precision())?
    } else {
        Quantity::zero(instrument.size_precision())
    };

    // Parse limit price if it exists (not -1)
    let price = if is_market_price(algo_fields.ord_px) {
        None
    } else {
        Some(parse_price(
            algo_fields.ord_px,
            instrument.price_precision(),
        )?)
    };

    let trigger_type = match algo_fields.trigger_px_type {
        OKXTriggerType::Last => TriggerType::LastPrice,
        OKXTriggerType::Mark => TriggerType::MarkPrice,
        OKXTriggerType::Index => TriggerType::IndexPrice,
        OKXTriggerType::None => TriggerType::Default,
    };

    let ts_accepted = parse_millisecond_timestamp(msg.c_time);
    let ts_last = parse_millisecond_timestamp(msg.u_time);

    let mut report = OrderStatusReport::new(
        account_id,
        instrument.id(),
        client_order_id,
        venue_order_id,
        order_side.into(),
        algo_fields.order_type,
        TimeInForce::Gtc,
        status,
        quantity,
        filled_qty,
        ts_accepted,
        ts_last,
        ts_init,
        None,
    );

    if !algo_fields.trigger_px.is_empty() {
        report.trigger_price = Some(parse_price(
            algo_fields.trigger_px,
            instrument.price_precision(),
        )?);
    }

    report.trigger_type = Some(trigger_type);

    if let Some(limit_price) = price {
        report.price = Some(limit_price);
    }

    if algo_fields.order_type == OrderType::TrailingStopMarket {
        if !msg.callback_ratio.is_empty() {
            // OKX ratio is e.g. "0.01" for 1%, convert to basis points
            let ratio = Decimal::from_str(&msg.callback_ratio)?;
            report.trailing_offset = Some(ratio * Decimal::new(10_000, 0));
            report.trailing_offset_type = Some(TrailingOffsetType::BasisPoints);
        } else if !msg.callback_spread.is_empty() {
            report.trailing_offset = Some(Decimal::from_str(&msg.callback_spread)?);
            report.trailing_offset_type = Some(TrailingOffsetType::Price);
        }

        if !msg.active_px.is_empty() {
            report.activation_price =
                Some(parse_price(&msg.active_px, instrument.price_precision())?);
        }
    }

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

    Ok(report)
}

struct AlgoOrderFields<'a> {
    order_type: OrderType,
    trigger_px: &'a str,
    trigger_px_type: OKXTriggerType,
    ord_px: &'a str,
}

fn parse_algo_order_fields(msg: &OKXAlgoOrderMsg) -> anyhow::Result<AlgoOrderFields<'_>> {
    match msg.ord_type {
        OKXAlgoOrderType::MoveOrderStop => Ok(AlgoOrderFields {
            order_type: OrderType::TrailingStopMarket,
            trigger_px: msg.trigger_px.as_str(),
            trigger_px_type: msg.trigger_px_type,
            ord_px: msg.ord_px.as_str(),
        }),
        OKXAlgoOrderType::Conditional | OKXAlgoOrderType::Oco => {
            if msg.tp_trigger_px.is_empty() {
                let (trigger_px, trigger_px_type, ord_px) = if msg.sl_trigger_px.is_empty() {
                    (
                        msg.trigger_px.as_str(),
                        msg.trigger_px_type,
                        msg.ord_px.as_str(),
                    )
                } else {
                    (
                        msg.sl_trigger_px.as_str(),
                        msg.sl_trigger_px_type,
                        msg.sl_ord_px.as_str(),
                    )
                };

                Ok(AlgoOrderFields {
                    order_type: if is_market_price(ord_px) {
                        OrderType::StopMarket
                    } else {
                        OrderType::StopLimit
                    },
                    trigger_px,
                    trigger_px_type,
                    ord_px,
                })
            } else {
                let ord_px = msg.tp_ord_px.as_str();
                Ok(AlgoOrderFields {
                    order_type: if is_market_price(ord_px) {
                        OrderType::MarketIfTouched
                    } else {
                        OrderType::LimitIfTouched
                    },
                    trigger_px: msg.tp_trigger_px.as_str(),
                    trigger_px_type: msg.tp_trigger_px_type,
                    ord_px,
                })
            }
        }
        OKXAlgoOrderType::Trigger => Ok(AlgoOrderFields {
            order_type: if is_market_price(&msg.ord_px) {
                OrderType::StopMarket
            } else {
                OrderType::StopLimit
            },
            trigger_px: msg.trigger_px.as_str(),
            trigger_px_type: msg.trigger_px_type,
            ord_px: msg.ord_px.as_str(),
        }),
        _ => anyhow::bail!("Unsupported algo order type: {:?}", msg.ord_type),
    }
}

fn parse_algo_order_quantity(
    msg: &OKXAlgoOrderMsg,
    instrument: &InstrumentAny,
) -> anyhow::Result<Quantity> {
    if !msg.sz.is_empty() {
        return parse_quantity(msg.sz.as_str(), instrument.size_precision());
    }

    if !msg.close_fraction.is_empty()
        || !msg.sl_trigger_px.is_empty()
        || !msg.tp_trigger_px.is_empty()
    {
        return Ok(Quantity::zero(instrument.size_precision()));
    }

    anyhow::bail!("Missing sz for algo order {}", msg.algo_id)
}

/// Parses an OKX order message into a Nautilus order status report.
///
/// # Errors
///
/// Returns an error if order metadata or numeric values cannot be parsed.
pub fn parse_order_status_report(
    msg: &OKXOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderStatusReport> {
    let client_order_id =
        parse_parent_client_order_id(msg.algo_cl_ord_id.as_deref(), &msg.cl_ord_id);
    let venue_order_id = VenueOrderId::new(msg.ord_id);
    let order_side = OrderSide::from(msg.side);

    let okx_order_type = msg.ord_type;

    // Determine order type based on presence of limit price for certain OKX order types
    let order_type = match okx_order_type {
        OKXOrderType::Trigger => {
            if is_market_price(&msg.px) {
                OrderType::StopMarket
            } else {
                OrderType::StopLimit
            }
        }
        OKXOrderType::Fok | OKXOrderType::Ioc | OKXOrderType::OptimalLimitIoc => {
            determine_order_type_with_alt(
                okx_order_type,
                &msg.px,
                msg.px_vol.as_deref().unwrap_or(""),
                msg.px_usd.as_deref().unwrap_or(""),
            )?
        }
        other => other
            .try_into()
            .map_err(|e| anyhow::anyhow!("Unsupported OKX order type: {e}"))?,
    };
    let order_status: OrderStatus = msg
        .state
        .try_into()
        .map_err(|e| anyhow::anyhow!("Unsupported OKX order status: {e}"))?;

    let time_in_force = match okx_order_type {
        OKXOrderType::Fok | OKXOrderType::OpFok => TimeInForce::Fok,
        OKXOrderType::Ioc | OKXOrderType::OptimalLimitIoc => TimeInForce::Ioc,
        _ => TimeInForce::Gtc,
    };

    let size_precision = instrument.size_precision();

    // Parse quantities based on target currency
    // OKX always returns acc_fill_sz in base currency, but sz depends on tgt_ccy

    // Determine if this is a quote-quantity order
    // Method 1: Explicit tgt_ccy field set to QuoteCcy
    let is_quote_qty_explicit = msg.tgt_ccy == Some(OKXTargetCurrency::QuoteCcy);

    // Method 2: Use OKX defaults when tgt_ccy is None (old orders or missing field)
    // OKX API defaults for SPOT market orders: BUY orders use quote_ccy, SELL orders use base_ccy
    // Note: tgtCcy only applies to SPOT market orders (not limit orders)
    // For limit orders, sz is always in base currency regardless of side
    let is_quote_qty_heuristic = msg.tgt_ccy.is_none()
        && (msg.inst_type == OKXInstrumentType::Spot || msg.inst_type == OKXInstrumentType::Margin)
        && msg.side == OKXSide::Buy
        && order_type == OrderType::Market;

    let (quantity, filled_qty) = if is_quote_qty_explicit || is_quote_qty_heuristic {
        // Quote-quantity order: sz is in quote currency, need to convert to base
        let sz_quote_dec = Decimal::from_str(&msg.sz).map_err(|e| {
            anyhow::anyhow!("Failed to parse sz='{}' as quote quantity: {}", msg.sz, e)
        })?;

        // Determine the price to use for conversion
        // Priority: 1) limit price (px) for limit orders, 2) avg_px for market orders
        let conversion_price_dec =
            if !is_market_price(&msg.px) {
                // Limit order: use the limit price (msg.px)
                Some(
                    Decimal::from_str(&msg.px)
                        .map_err(|e| anyhow::anyhow!("Failed to parse px='{}': {}", msg.px, e))?,
                )
            } else if !msg.avg_px.is_empty() && msg.avg_px != "0" {
                // Market order with fills: use average fill price
                Some(Decimal::from_str(&msg.avg_px).map_err(|e| {
                    anyhow::anyhow!("Failed to parse avg_px='{}': {}", msg.avg_px, e)
                })?)
            } else {
                None
            };

        // Convert quote quantity to base: quantity_base = sz_quote / price
        let quantity_base = if let Some(price) = conversion_price_dec {
            if price.is_zero() {
                parse_quantity(&msg.sz, size_precision)?
            } else {
                Quantity::from_decimal_dp(sz_quote_dec / price, size_precision)?
            }
        } else {
            // No price available, can't convert - use sz as-is temporarily
            // This will be corrected once the order gets filled and price is available
            parse_quantity(&msg.sz, size_precision)?
        };

        let filled_qty = parse_quantity(msg.acc_fill_sz.as_deref().unwrap_or(""), size_precision)?;

        (quantity_base, filled_qty)
    } else {
        // Base-quantity order: both sz and acc_fill_sz are in base currency
        let quantity = parse_quantity(&msg.sz, size_precision)?;
        let filled_qty = parse_quantity(msg.acc_fill_sz.as_deref().unwrap_or(""), size_precision)?;

        (quantity, filled_qty)
    };

    // For quote-quantity orders marked as FILLED, adjust quantity to match filled_qty
    // to avoid precision mismatches from quote-to-base conversion
    let (quantity, filled_qty) = if (is_quote_qty_explicit || is_quote_qty_heuristic)
        && msg.state == OKXOrderStatus::Filled
        && filled_qty.is_positive()
    {
        (filled_qty, filled_qty)
    } else {
        (quantity, filled_qty)
    };

    let ts_accepted = parse_millisecond_timestamp(msg.c_time);
    let ts_last = parse_millisecond_timestamp(msg.u_time);

    let is_liquidation = matches!(
        msg.category,
        OKXOrderCategory::FullLiquidation | OKXOrderCategory::PartialLiquidation
    );

    let is_adl = msg.category == OKXOrderCategory::Adl;

    if is_liquidation {
        log::warn!(
            "Liquidation order status update: order_id={}, category={:?}, inst_id={}, state={:?}",
            msg.ord_id.as_str(),
            msg.category,
            msg.inst_id.as_str(),
            msg.state,
        );
    }

    if is_adl {
        log::warn!(
            "ADL (Auto-Deleveraging) order status update: order_id={}, inst_id={}, state={:?}",
            msg.ord_id.as_str(),
            msg.inst_id.as_str(),
            msg.state,
        );
    }

    let mut report = OrderStatusReport::new(
        account_id,
        instrument.id(),
        client_order_id,
        venue_order_id,
        order_side.into(),
        order_type,
        time_in_force,
        order_status,
        quantity,
        filled_qty,
        ts_accepted,
        ts_last,
        ts_init,
        None, // Generate UUID4 automatically
    );

    let price_precision = instrument.price_precision();

    if okx_order_type == OKXOrderType::Trigger {
        // For triggered orders coming through regular orders channel,
        // set the price if it's a stop-limit order
        if !is_market_price(&msg.px)
            && let Ok(price) = parse_price(&msg.px, price_precision)
        {
            report = report.with_price(price);
        }
    } else {
        // For regular orders, use px field
        if !is_market_price(&msg.px)
            && let Ok(price) = parse_price(&msg.px, price_precision)
        {
            report = report.with_price(price);
        }
    }

    if !msg.avg_px.is_empty()
        && let Ok(decimal) = Decimal::from_str(&msg.avg_px)
    {
        report.avg_px = Some(decimal);
    }

    if matches!(
        msg.ord_type,
        OKXOrderType::PostOnly | OKXOrderType::Rpi | OKXOrderType::MmpAndPostOnly
    ) || matches!(
        msg.cancel_source.as_deref(),
        Some(source) if source == OKX_POST_ONLY_CANCEL_SOURCE
    ) || matches!(
        msg.cancel_source_reason.as_deref(),
        Some(reason) if reason.contains("POST_ONLY")
    ) {
        report = report.with_post_only(true);
    }

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

    let mut linked_ids = Vec::new();

    if let Some(algo_cl_ord_id) = msg
        .algo_cl_ord_id
        .as_ref()
        .filter(|value| !value.is_empty())
    {
        let algo_client_id = ClientOrderId::new(algo_cl_ord_id.as_str());
        if report.client_order_id != Some(algo_client_id) {
            linked_ids.push(algo_client_id);
        }
    }

    if let Some(attach_algo_cl_ord_id) = msg
        .attach_algo_cl_ord_id
        .as_ref()
        .filter(|value| !value.is_empty())
    {
        let attach_client_id = ClientOrderId::new(attach_algo_cl_ord_id.as_str());
        if report.client_order_id != Some(attach_client_id)
            && !linked_ids.contains(&attach_client_id)
        {
            linked_ids.push(attach_client_id);
        }
    }

    for attach_algo in &msg.attach_algo_ords {
        if attach_algo.attach_algo_cl_ord_id.is_empty() {
            continue;
        }

        let attach_client_id = ClientOrderId::new(attach_algo.attach_algo_cl_ord_id.as_str());
        if report.client_order_id != Some(attach_client_id)
            && !linked_ids.contains(&attach_client_id)
        {
            linked_ids.push(attach_client_id);
        }
    }

    if !linked_ids.is_empty() {
        report = report.with_linked_order_ids(linked_ids);
    }

    if let Some(reason) = msg
        .cancel_source_reason
        .as_ref()
        .filter(|reason| !reason.is_empty())
    {
        report = report.with_cancel_reason(reason.clone());
    } else if let Some(source) = msg
        .cancel_source
        .as_ref()
        .filter(|source| !source.is_empty())
    {
        let reason = if source == OKX_POST_ONLY_CANCEL_SOURCE {
            OKX_POST_ONLY_CANCEL_REASON.to_string()
        } else {
            format!("cancel_source={source}")
        };
        report = report.with_cancel_reason(reason);
    }

    Ok(report)
}

fn parse_spread_order_fill_report(
    msg: &OKXSpreadOrder,
    instrument: &InstrumentAny,
    _account_id: AccountId,
    previous_filled_qty: Option<Quantity>,
    _ts_init: UnixNanos,
) -> anyhow::Result<Option<FillReport>> {
    let size_precision = instrument.size_precision();
    if !msg.fill_sz.is_empty() && msg.fill_sz != "0" {
        parse_quantity(&msg.fill_sz, size_precision).map_err(|e| {
            anyhow::anyhow!("Failed to parse spread fill_sz='{}': {e}", msg.fill_sz)
        })?;
    } else if !msg.acc_fill_sz.is_empty() && msg.acc_fill_sz != "0" {
        let current_filled = parse_quantity(&msg.acc_fill_sz, size_precision).map_err(|e| {
            anyhow::anyhow!(
                "Failed to parse spread acc_fill_sz='{}': {e}",
                msg.acc_fill_sz
            )
        })?;

        if let Some(prev_qty) = previous_filled_qty {
            if current_filled < prev_qty {
                anyhow::bail!(
                    "Cumulative spread fill went backwards: acc_fill_sz='{}' < previous_filled_qty={} \
                     (possible stale data after reconnect)",
                    msg.acc_fill_sz,
                    prev_qty
                );
            }

            if (current_filled - prev_qty).is_zero() {
                log::debug!(
                    "Skipping duplicate spread fill: acc_fill_sz='{}' unchanged from previous={}",
                    msg.acc_fill_sz,
                    prev_qty
                );
                return Ok(None);
            }
        }
    } else {
        anyhow::bail!(
            "Cannot determine spread fill quantity: fill_sz='{}' and acc_fill_sz='{}'",
            msg.fill_sz,
            msg.acc_fill_sz
        );
    }

    anyhow::bail!(
        "missing fee for spread fill report sprd_id={}; OKX sprd-orders updates omit fee",
        msg.sprd_id
    )
}

/// Parses an OKX order message into a Nautilus fill report.
///
/// # Errors
///
/// Returns an error if order quantities, prices, or fees cannot be parsed.
pub fn parse_fill_report(
    msg: &OKXOrderMsg,
    instrument: &InstrumentAny,
    account_id: AccountId,
    previous_fee: Option<Money>,
    previous_filled_qty: Option<Quantity>,
    ts_init: UnixNanos,
) -> anyhow::Result<Option<FillReport>> {
    let client_order_id =
        parse_parent_client_order_id(msg.algo_cl_ord_id.as_deref(), &msg.cl_ord_id);
    let venue_order_id = VenueOrderId::new(msg.ord_id);

    // OKX may not provide a `trade_id` (some algo trigger payloads, manual
    // settlements). Derive a deterministic id from the immutable fill fields
    // via FNV-1a so reconnect replays of the same fill collapse to one event
    // in the downstream `WsDispatchState::check_and_insert_trade` dedup. A
    // random UUID4 here would defeat dedup, since each replay would mint a
    // new id. The hash output keeps the synthesized id within `TradeId`'s
    // 36-character cap.
    let trade_id = if msg.trade_id.is_empty() {
        let synthetic = synthesize_trade_id(msg);
        TradeId::new(&synthetic)
    } else {
        TradeId::new(&msg.trade_id)
    };

    let order_side = OrderSide::from(msg.side);

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

    let price_str = if !msg.fill_px.is_empty() {
        &msg.fill_px
    } else if !msg.avg_px.is_empty() {
        &msg.avg_px
    } else {
        &msg.px
    };
    let last_px = parse_price(price_str, price_precision).map_err(|e| {
        anyhow::anyhow!(
            "Failed to parse price (fill_px='{}', avg_px='{}', px='{}'): {}",
            msg.fill_px,
            msg.avg_px,
            msg.px,
            e
        )
    })?;

    // OKX provides fillSz (incremental fill) or accFillSz (cumulative total)
    // If fillSz is provided, use it directly as the incremental fill quantity
    let last_qty = if !msg.fill_sz.is_empty() && msg.fill_sz != "0" {
        parse_quantity(&msg.fill_sz, size_precision)
            .map_err(|e| anyhow::anyhow!("Failed to parse fill_sz='{}': {e}", msg.fill_sz))?
    } else if let Some(ref acc_fill_sz) = msg.acc_fill_sz {
        // If fillSz is missing but accFillSz is available, calculate incremental fill
        if !acc_fill_sz.is_empty() && acc_fill_sz != "0" {
            let current_filled = parse_quantity(acc_fill_sz, size_precision)
                .map_err(|e| anyhow::anyhow!("Failed to parse acc_fill_sz='{acc_fill_sz}': {e}"))?;

            // Calculate incremental fill as: current_total - previous_total
            if let Some(prev_qty) = previous_filled_qty {
                if current_filled < prev_qty {
                    anyhow::bail!(
                        "Cumulative fill went backwards: acc_fill_sz='{acc_fill_sz}' < previous_filled_qty={prev_qty} \
                         (possible stale data after reconnect)"
                    );
                }
                let incremental = current_filled - prev_qty;
                if incremental.is_zero() {
                    log::debug!(
                        "Skipping duplicate fill: acc_fill_sz='{acc_fill_sz}' unchanged from previous={prev_qty}"
                    );
                    return Ok(None);
                }
                incremental
            } else {
                // First fill, use accumulated as incremental
                current_filled
            }
        } else {
            anyhow::bail!(
                "Cannot determine fill quantity: fill_sz is empty/zero and acc_fill_sz is empty/zero"
            );
        }
    } else {
        anyhow::bail!(
            "Cannot determine fill quantity: fill_sz='{}' and acc_fill_sz is None",
            msg.fill_sz
        );
    };

    let fee_str = msg
        .fee
        .as_deref()
        .filter(|fee| !fee.trim().is_empty())
        .ok_or_else(|| anyhow::anyhow!("missing fee for fill report inst_id={}", msg.inst_id))?;
    let fee_dec = Decimal::from_str(fee_str)
        .map_err(|e| anyhow::anyhow!("Failed to parse fee '{fee_str}': {e}"))?;

    let fee_currency = parse_fee_currency(msg.fee_ccy.as_str(), fee_dec, || {
        format!("fill report for inst_id={}", msg.inst_id)
    });

    // OKX sends fees as negative numbers (e.g., "-2.5" for a $2.5 charge), parse_fee negates to positive
    let total_fee = parse_fee(Some(fee_str), fee_currency)
        .map_err(|e| anyhow::anyhow!("Failed to parse fee={:?}: {}", msg.fee, e))?;

    // OKX sends cumulative fees, so we subtract the previous total to get this fill's fee
    let commission = if let Some(previous_fee) = previous_fee {
        if total_fee.currency == previous_fee.currency {
            let incremental = total_fee - previous_fee;

            if incremental < Money::zero(fee_currency) {
                log::debug!(
                    "Negative incremental fee detected - likely a maker rebate or fee refund: order_id={}, total_fee={}, previous_fee={}, incremental={}",
                    msg.ord_id.as_str(),
                    total_fee,
                    previous_fee,
                    incremental,
                );
            }

            // Skip corruption check when previous is negative (rebate), as transitions from
            // rebate to charge legitimately have incremental > total (e.g., -1 → +2 gives +3)
            if previous_fee >= Money::zero(fee_currency)
                && total_fee > Money::zero(fee_currency)
                && incremental > total_fee
            {
                log::error!(
                    "Incremental fee exceeds total fee - likely fee cache corruption, using total fee as fallback: order_id={}, total_fee={}, previous_fee={}, incremental={}",
                    msg.ord_id.as_str(),
                    total_fee,
                    previous_fee,
                    incremental,
                );
                total_fee
            } else {
                incremental
            }
        } else {
            log::warn!(
                "Fee currency changed from {} to {} for order_id={}, using total fee as commission",
                previous_fee.currency.code,
                total_fee.currency.code,
                msg.ord_id.as_str(),
            );
            total_fee
        }
    } else {
        total_fee
    };

    let liquidity_side: LiquiditySide = msg.exec_type.into();
    let ts_event = parse_millisecond_timestamp(msg.fill_time);

    let is_liquidation = matches!(
        msg.category,
        OKXOrderCategory::FullLiquidation | OKXOrderCategory::PartialLiquidation
    );

    let is_adl = msg.category == OKXOrderCategory::Adl;

    if is_liquidation {
        log::warn!(
            "Liquidation order detected: order_id={}, category={:?}, inst_id={}, side={:?}, fill_sz={}, fill_px={}",
            msg.ord_id.as_str(),
            msg.category,
            msg.inst_id.as_str(),
            msg.side,
            msg.fill_sz,
            msg.fill_px,
        );
    }

    if is_adl {
        log::warn!(
            "ADL (Auto-Deleveraging) order detected: order_id={}, inst_id={}, side={:?}, fill_sz={}, fill_px={}",
            msg.ord_id.as_str(),
            msg.inst_id.as_str(),
            msg.side,
            msg.fill_sz,
            msg.fill_px,
        );
    }

    let report = FillReport::new(
        account_id,
        instrument.id(),
        venue_order_id,
        trade_id,
        order_side,
        last_qty,
        last_px,
        commission,
        liquidity_side,
        client_order_id,
        None,
        ts_event,
        ts_init,
        None, // Generate UUID4 automatically
    );

    Ok(Some(report))
}

/// Parses an option summary payload into [`OptionGreeks`].
///
/// Selects Black-Scholes (`delta_bs`, `gamma_bs`, `vega_bs`, `theta_bs`) or
/// price-adjusted (`delta`, `gamma`, `vega`, `theta`) greeks based on `greeks_type`.
/// BS greeks align with what Deribit and Bybit provide; PA greeks are denominated
/// in the underlying/coin units and match OKX's native contract convention.
///
/// # Errors
///
/// Returns an error if any of the greeks or volatility fields cannot be parsed as f64.
pub fn parse_option_summary_greeks(
    msg: &OKXOptionSummaryMsg,
    instrument_id: &InstrumentId,
    greeks_type: OKXGreeksType,
    ts_init: UnixNanos,
) -> anyhow::Result<OptionGreeks> {
    let ts_event = UnixNanos::from(msg.ts * 1_000_000);

    let (delta_s, gamma_s, vega_s, theta_s, delta_ctx, gamma_ctx, vega_ctx, theta_ctx) =
        match greeks_type {
            OKXGreeksType::Bs => (
                &msg.delta_bs,
                &msg.gamma_bs,
                &msg.vega_bs,
                &msg.theta_bs,
                "invalid delta_bs",
                "invalid gamma_bs",
                "invalid vega_bs",
                "invalid theta_bs",
            ),
            OKXGreeksType::Pa => (
                &msg.delta,
                &msg.gamma,
                &msg.vega,
                &msg.theta,
                "invalid delta (pa)",
                "invalid gamma (pa)",
                "invalid vega (pa)",
                "invalid theta (pa)",
            ),
        };

    let delta: f64 = delta_s.parse().context(delta_ctx)?;
    let gamma: f64 = gamma_s.parse().context(gamma_ctx)?;
    let vega: f64 = vega_s.parse().context(vega_ctx)?;
    let theta: f64 = theta_s.parse().context(theta_ctx)?;

    let bid_iv: f64 = msg.bid_vol.parse().context("invalid bid_vol")?;
    let ask_iv: f64 = msg.ask_vol.parse().context("invalid ask_vol")?;
    let mark_iv: f64 = msg.mark_vol.parse().context("invalid mark_vol")?;

    let underlying_price = msg
        .fwd_px
        .as_deref()
        .filter(|s| !s.is_empty())
        .map(str::parse::<f64>)
        .transpose()
        .context("invalid fwd_px")?;

    Ok(OptionGreeks {
        instrument_id: *instrument_id,
        convention: greeks_type.into(),
        greeks: OptionGreekValues {
            delta,
            gamma,
            vega,
            theta,
            rho: 0.0, // OKX does not provide rho
        },
        mark_iv: Some(mark_iv),
        bid_iv: Some(bid_iv),
        ask_iv: Some(ask_iv),
        underlying_price,
        open_interest: None,
        ts_event,
        ts_init,
    })
}

/// Parses OKX WebSocket message payloads into Nautilus data structures.
///
/// # Errors
///
/// Returns an error if the payload cannot be deserialized or if downstream
/// parsing routines fail.
///
/// # Panics
///
/// Panics only in the case where `okx_channel_to_bar_spec(channel)` returns
/// `None` after a prior `is_some` check - an unreachable scenario indicating a
/// logic error.
#[expect(clippy::too_many_arguments)]
pub fn parse_ws_message_data(
    channel: &OKXWsChannel,
    data: serde_json::Value,
    instrument_id: &InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_init: UnixNanos,
    funding_cache: &mut AHashMap<Ustr, (Ustr, u64)>,
    instruments_cache: &AHashMap<Ustr, InstrumentAny>,
) -> anyhow::Result<Option<NautilusWsMessage>> {
    match channel {
        OKXWsChannel::Instruments => {
            if let Ok(msg) = serde_json::from_value::<OKXInstrument>(data) {
                let inst_key = msg.inst_id;
                let cached_instrument = instruments_cache.get(&inst_key);
                let (margin_init, margin_maint, maker_fee, taker_fee) = cached_instrument.map_or(
                    (None, None, None, None),
                    extract_fees_from_cached_instrument,
                );
                let instrument_id =
                    cached_instrument.map_or_else(|| parse_instrument_id(inst_key), Instrument::id);

                let status_action = okx_status_to_market_action(msg.state);
                let status = InstrumentStatus::new(
                    instrument_id,
                    status_action,
                    ts_init,
                    ts_init,
                    None,
                    None,
                    Some(matches!(msg.state, OKXInstrumentStatus::Live)),
                    None,
                    None,
                );

                match parse_instrument_any(
                    &msg,
                    margin_init,
                    margin_maint,
                    maker_fee,
                    taker_fee,
                    ts_init,
                ) {
                    Ok(Some(inst_any)) => Ok(Some(NautilusWsMessage::Instrument(
                        Box::new(inst_any),
                        Some(status),
                    ))),
                    Ok(None) => {
                        log::warn!("Empty instrument payload: {msg:?}");
                        Ok(Some(NautilusWsMessage::InstrumentStatus(status)))
                    }
                    Err(e) => {
                        log::warn!("Failed to parse instrument {inst_key}: {e}");
                        Ok(Some(NautilusWsMessage::InstrumentStatus(status)))
                    }
                }
            } else {
                anyhow::bail!("Failed to deserialize instrument payload")
            }
        }
        OKXWsChannel::BboTbt => {
            let data_vec = parse_quote_msg_vec(
                data,
                instrument_id,
                price_precision,
                size_precision,
                ts_init,
            )?;
            Ok(Some(NautilusWsMessage::Data(data_vec)))
        }
        OKXWsChannel::Tickers => {
            let data_vec = parse_ticker_msg_vec(
                data,
                instrument_id,
                price_precision,
                size_precision,
                ts_init,
            )?;
            Ok(Some(NautilusWsMessage::Data(data_vec)))
        }
        OKXWsChannel::Trades | OKXWsChannel::SprdPublicTrades => {
            let data_vec = parse_trade_msg_vec(
                data,
                instrument_id,
                price_precision,
                size_precision,
                ts_init,
            )?;
            Ok(Some(NautilusWsMessage::Data(data_vec)))
        }
        OKXWsChannel::MarkPrice => {
            let data_vec = parse_mark_price_msg_vec(data, instrument_id, price_precision, ts_init)?;
            Ok(Some(NautilusWsMessage::Data(data_vec)))
        }
        OKXWsChannel::IndexTickers => {
            let data_vec =
                parse_index_price_msg_vec(data, instrument_id, price_precision, ts_init)?;
            Ok(Some(NautilusWsMessage::Data(data_vec)))
        }
        OKXWsChannel::FundingRate => {
            let data_vec = parse_funding_rate_msg_vec(data, instrument_id, ts_init, funding_cache)?;
            Ok(Some(NautilusWsMessage::FundingRates(data_vec)))
        }
        OKXWsChannel::EventContractMarkets => Ok(Some(NautilusWsMessage::Raw(data))),
        channel if okx_channel_to_bar_spec(channel).is_some() => {
            let bar_spec = okx_channel_to_bar_spec(channel).expect("bar_spec checked above");
            let data_vec = parse_candle_msg_vec(
                data,
                instrument_id,
                price_precision,
                size_precision,
                bar_spec,
                ts_init,
            )?;
            Ok(Some(NautilusWsMessage::Data(data_vec)))
        }
        OKXWsChannel::Books
        | OKXWsChannel::BooksTbt
        | OKXWsChannel::Books5
        | OKXWsChannel::Books50Tbt => {
            if let Ok(book_msgs) = serde_json::from_value::<Vec<OKXBookMsg>>(data) {
                let data_vec = parse_book_depth_msg_vec(
                    book_msgs,
                    instrument_id,
                    price_precision,
                    size_precision,
                    ts_init,
                )?;
                Ok(Some(NautilusWsMessage::Data(data_vec)))
            } else {
                anyhow::bail!("Failed to deserialize Books channel data as Vec<OKXBookMsg>")
            }
        }
        _ => {
            log::warn!("Unsupported channel for message parsing: {channel:?}");
            Ok(None)
        }
    }
}

#[cfg(test)]
mod tests {
    use ahash::AHashMap;
    use nautilus_core::nanos::UnixNanos;
    use nautilus_model::{
        data::bar::BAR_SPEC_1_DAY_LAST,
        enums::{AccountType, BookType, GreeksConvention},
        identifiers::{ClientOrderId, Symbol, VenueOrderId},
        instruments::CryptoPerpetual,
        orderbook::OrderBook,
        types::Currency,
    };
    use rstest::rstest;
    use rust_decimal::Decimal;
    use rust_decimal_macros::dec;
    use serde_json::Value;
    use ustr::Ustr;

    use super::*;
    use crate::{
        OKXPositionSide,
        common::{
            enums::{
                OKXAlgoOrderStatus, OKXExecType, OKXInstrumentType, OKXMarginMode, OKXOrderType,
                OKXPriceType, OKXQuickMarginType, OKXSelfTradePreventionMode, OKXSide,
                OKXTradeMode,
            },
            parse::parse_account_state,
            testing::load_test_json,
        },
        http::models::OKXAccount,
        websocket::messages::{
            OKXAlgoOrderMsg, OKXAttachedAlgoOrd, OKXLiquidationWarningMsg, OKXWebSocketArg,
            OKXWsFrame,
        },
    };

    fn is_order_updated(
        msg: &OKXOrderMsg,
        previous: &OrderStateSnapshot,
        instrument: &InstrumentAny,
    ) -> anyhow::Result<bool> {
        let current_venue_id = VenueOrderId::new(msg.ord_id);

        if previous.venue_order_id != current_venue_id {
            return Ok(true);
        }

        let current_qty = parse_quantity(&msg.sz, instrument.size_precision())?;
        if previous.quantity != current_qty {
            return Ok(true);
        }

        if !is_market_price(&msg.px) {
            let current_price = parse_price(&msg.px, instrument.price_precision())?;

            if let Some(prev_price) = previous.price
                && prev_price != current_price
            {
                return Ok(true);
            }
        }

        Ok(false)
    }

    fn create_stub_instrument() -> CryptoPerpetual {
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap()
    }

    #[rstest]
    fn open_order_baselines_survive_unrelated_order_churn() {
        let instrument = InstrumentAny::CryptoPerpetual(create_stub_instrument());
        let mut fee_cache = FeeCache::new();
        let mut filled_qty_cache = FilledQtyCache::new();

        let mut resting = create_stub_order_msg("1.0", Some("1.0".to_string()), "ord-resting", "");
        resting.state = OKXOrderStatus::PartiallyFilled;
        update_fee_fill_caches(&resting, &instrument, &mut fee_cache, &mut filled_qty_cache);

        for i in 0..=TERMINAL_BASELINE_CAPACITY {
            let mut msg =
                create_stub_order_msg("1.0", Some("1.0".to_string()), &format!("ord-{i}"), "");
            msg.state = OKXOrderStatus::Filled;
            update_fee_fill_caches(&msg, &instrument, &mut fee_cache, &mut filled_qty_cache);
        }

        let resting_key = Ustr::from("ord-resting");

        assert_eq!(fee_cache.get(&resting_key).unwrap().as_decimal(), dec!(1.0));
        assert_eq!(
            filled_qty_cache.get(&resting_key).unwrap().as_decimal(),
            dec!(1.0)
        );
        assert_eq!(fee_cache.open.len(), 1);
        assert_eq!(filled_qty_cache.open.len(), 1);

        // Demotion is what bounds the open map, so prove the migration and not just the
        // classification: a settled order must leave `open` and expose its newer baseline.
        let mut settled = resting.clone();
        settled.state = OKXOrderStatus::Filled;
        settled.fee = Some("-2.5".to_string());
        settled.acc_fill_sz = Some("2.0".to_string());
        update_fee_fill_caches(&settled, &instrument, &mut fee_cache, &mut filled_qty_cache);

        assert_eq!(fee_cache.open.len(), 0);
        assert_eq!(filled_qty_cache.open.len(), 0);
        assert_eq!(fee_cache.get(&resting_key).unwrap().as_decimal(), dec!(2.5));
        assert_eq!(
            filled_qty_cache.get(&resting_key).unwrap().as_decimal(),
            dec!(2.0)
        );
    }

    #[rstest]
    #[case(OKXOrderStatus::Filled)]
    #[case(OKXOrderStatus::Canceled)]
    #[case(OKXOrderStatus::MmpCanceled)]
    fn terminal_order_baselines_stay_terminal_after_late_updates(
        #[case] terminal_state: OKXOrderStatus,
        #[values(OKXOrderStatus::Live, OKXOrderStatus::PartiallyFilled)] late_state: OKXOrderStatus,
    ) {
        let instrument = InstrumentAny::CryptoPerpetual(create_stub_instrument());
        let mut fee_cache = FeeCache::new();
        let mut filled_qty_cache = FilledQtyCache::new();
        let mut msg = create_stub_order_msg("1.0", Some("2.0".to_string()), "ord-late", "");
        msg.state = terminal_state;
        update_fee_fill_caches(&msg, &instrument, &mut fee_cache, &mut filled_qty_cache);

        msg.state = late_state;
        msg.fee = Some("-0.5".to_string());
        msg.acc_fill_sz = Some("1.0".to_string());
        update_fee_fill_caches(&msg, &instrument, &mut fee_cache, &mut filled_qty_cache);

        let key = msg.ord_id;
        let fee = Money::from_decimal(dec!(0.5), Currency::USDT()).unwrap();
        let quantity = Quantity::from("1.00000000");

        assert_eq!(fee_cache.open.len(), 0);
        assert_eq!(filled_qty_cache.open.len(), 0);
        assert_eq!(fee_cache.terminal.get(&key), Some(&fee));
        assert_eq!(filled_qty_cache.terminal.get(&key), Some(&quantity));
        assert_eq!(fee_cache.get(&key), Some(&fee));
        assert_eq!(filled_qty_cache.get(&key), Some(&quantity));
    }

    #[rstest]
    fn terminal_order_baselines_evict_oldest_first() {
        let instrument = InstrumentAny::CryptoPerpetual(create_stub_instrument());
        let mut fee_cache = FeeCache::new();
        let mut filled_qty_cache = FilledQtyCache::new();

        for i in 0..=TERMINAL_BASELINE_CAPACITY {
            let mut msg =
                create_stub_order_msg("1.0", Some("1.0".to_string()), &format!("ord-{i}"), "");
            msg.state = OKXOrderStatus::Filled;
            update_fee_fill_caches(&msg, &instrument, &mut fee_cache, &mut filled_qty_cache);
        }

        let oldest = Ustr::from("ord-0");
        let newest = Ustr::from(&format!("ord-{TERMINAL_BASELINE_CAPACITY}"));

        assert_eq!(fee_cache.get(&oldest), None);
        assert_eq!(filled_qty_cache.get(&oldest), None);
        assert_eq!(fee_cache.get(&newest).unwrap().as_decimal(), dec!(1.0));
        assert_eq!(
            filled_qty_cache.get(&newest).unwrap().as_decimal(),
            dec!(1.0)
        );
        assert_eq!(fee_cache.open.len(), 0);
    }

    fn create_stub_order_msg(
        fill_sz: &str,
        acc_fill_sz: Option<String>,
        order_id: &str,
        trade_id: &str,
    ) -> OKXOrderMsg {
        OKXOrderMsg {
            acc_fill_sz,
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_order_1".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("-1.0".to_string()),
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "50000.0".to_string(),
            fill_sz: fill_sz.to_string(),
            fill_time: 1_746_947_317_402,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from(order_id),
            ord_type: OKXOrderType::Market,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::PartiallyFilled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Taker,
            sz: "0.03".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: trade_id.to_string(),
            u_time: 1_746_947_317_402,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        }
    }

    #[rstest]
    #[case::standard(false)]
    #[case::rpi(true)]
    fn snapshot_replaces_existing_book(#[case] rpi: bool, #[values(false, true)] empty: bool) {
        let fixture = if rpi {
            "ws_books_rpi_snapshot.json"
        } else {
            "ws_books_snapshot.json"
        };

        let mut frame: Value = serde_json::from_str(&load_test_json(fixture)).unwrap();
        let instrument_id = InstrumentId::from("BTC-USDT.OKX");

        let parse = |frame: &Value| {
            if rpi {
                parse_rpi_book_msg(
                    &serde_json::from_value(frame["data"][0].clone()).unwrap(),
                    instrument_id,
                    7,
                    3,
                    &OKXBookAction::Snapshot,
                    UnixNanos::from(123),
                )
                .unwrap()
            } else {
                parse_book_msg(
                    &serde_json::from_value(frame["data"][0].clone()).unwrap(),
                    instrument_id,
                    2,
                    1,
                    &OKXBookAction::Snapshot,
                    UnixNanos::from(123),
                )
                .unwrap()
            }
        };

        let mut actual = OrderBook::new(instrument_id, BookType::L2_MBP);
        actual.apply_deltas(&parse(&frame)).unwrap();
        for side in ["bids", "asks"] {
            let levels = frame["data"][0][side].as_array_mut().unwrap();
            if empty {
                levels.clear();
            } else {
                levels.remove(0);
            }
        }

        let snapshot = parse(&frame);
        let mut expected = OrderBook::new(instrument_id, BookType::L2_MBP);
        expected.apply_deltas(&snapshot).unwrap();
        actual.apply_deltas(&snapshot).unwrap();

        assert_eq!(actual.bids_as_map(None), expected.bids_as_map(None));
        assert_eq!(actual.asks_as_map(None), expected.asks_as_map(None));

        let expected_clear_flags = if empty {
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        } else {
            RecordFlag::F_SNAPSHOT as u8
        };

        assert_eq!(snapshot.deltas[0].action, BookAction::Clear);
        assert_eq!(snapshot.deltas[0].flags, expected_clear_flags);
        assert_eq!(snapshot.deltas[0].sequence, snapshot.sequence);
        assert_eq!(snapshot.deltas[0].ts_event, snapshot.ts_event);
        assert_eq!(snapshot.deltas[0].ts_init, UnixNanos::from(123));
    }

    #[rstest]
    #[case::snapshot("ws_books_snapshot.json", OKXBookAction::Snapshot)]
    #[case::update("ws_books_update.json", OKXBookAction::Update)]
    #[case::rpi_snapshot("ws_books_rpi_snapshot.json", OKXBookAction::Snapshot)]
    #[case::rpi_update("ws_books_rpi_update.json", OKXBookAction::Update)]
    fn book_deltas_flag_only_the_final_delta_last(
        #[case] fixture: &str,
        #[case] action: OKXBookAction,
    ) {
        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let frame: OKXWsFrame = serde_json::from_str(&load_test_json(fixture)).unwrap();

        let deltas = match frame {
            OKXWsFrame::BookData { data, .. } => {
                parse_book_msg(&data[0], instrument_id, 2, 1, &action, UnixNanos::default())
            }
            OKXWsFrame::RpiBookData { data, .. } => {
                parse_rpi_book_msg(&data[0], instrument_id, 7, 3, &action, UnixNanos::default())
            }
            _ => panic!("Expected a book frame"),
        }
        .unwrap();

        let last_count = deltas
            .deltas
            .iter()
            .filter(|delta| RecordFlag::F_LAST.matches(delta.flags))
            .count();

        assert_eq!(last_count, 1, "exactly one delta must close the event");
        assert!(RecordFlag::F_LAST.matches(deltas.deltas.last().unwrap().flags));
        assert!(RecordFlag::F_LAST.matches(deltas.flags));
    }

    #[rstest]
    fn empty_book_snapshot_flags_the_clear_delta_last() {
        let instrument_id = InstrumentId::from("BTC-USDT.OKX");

        let msg = OKXBookMsg {
            bids: Vec::new(),
            asks: Vec::new(),
            ts: 1_597_026_383_085,
            checksum: None,
            prev_seq_id: None,
            seq_id: 123_456,
        };

        let deltas = parse_book_msg(
            &msg,
            instrument_id,
            2,
            1,
            &OKXBookAction::Snapshot,
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(deltas.deltas.len(), 1);
        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
        assert_eq!(
            deltas.flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
    }

    #[rstest]
    fn test_parse_books_snapshot() {
        let json_data = load_test_json("ws_books_snapshot.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let (okx_books, action): (Vec<OKXBookMsg>, OKXBookAction) = match msg {
            OKXWsFrame::BookData { data, action, .. } => (data, action),
            _ => panic!("Expected a `BookData` variant"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let deltas = parse_book_msg(
            &okx_books[0],
            instrument_id,
            2,
            1,
            &action,
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(deltas.instrument_id, instrument_id);
        assert_eq!(deltas.deltas.len(), 17);
        assert_eq!(
            deltas.flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
        assert_eq!(deltas.sequence, 123_456);
        assert_eq!(deltas.ts_event, UnixNanos::from(1_597_026_383_085_000_000));
        assert_eq!(deltas.ts_init, UnixNanos::default());

        // Verify some individual deltas are parsed correctly
        assert!(!deltas.deltas.is_empty());
        // Snapshot should have both bid and ask deltas
        assert!(
            deltas
                .deltas
                .iter()
                .any(|d| d.order.side == OrderSide::Buy.into()),
            "Should have bid deltas"
        );
        assert!(
            deltas
                .deltas
                .iter()
                .any(|d| d.order.side == OrderSide::Sell.into()),
            "Should have ask deltas"
        );
    }

    #[rstest]
    fn test_parse_books_update() {
        let json_data = load_test_json("ws_books_update.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let (okx_books, action): (Vec<OKXBookMsg>, OKXBookAction) = match msg {
            OKXWsFrame::BookData { data, action, .. } => (data, action),
            _ => panic!("Expected a `BookData` variant"),
        };

        let deltas = parse_book_msg(
            &okx_books[0],
            instrument_id,
            2,
            1,
            &action,
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(deltas.instrument_id, instrument_id);
        assert_eq!(deltas.deltas.len(), 16);
        assert_eq!(deltas.flags, RecordFlag::F_LAST as u8);
        assert_eq!(deltas.sequence, 123_457);
        assert_eq!(deltas.ts_event, UnixNanos::from(1_597_026_383_085_000_000));
        assert_eq!(deltas.ts_init, UnixNanos::default());

        // Verify some individual deltas are parsed correctly
        assert!(!deltas.deltas.is_empty());
        // Update should also have both bid and ask deltas
        assert!(
            deltas
                .deltas
                .iter()
                .any(|d| d.order.side == OrderSide::Buy.into()),
            "Should have bid deltas"
        );
        assert!(
            deltas
                .deltas
                .iter()
                .any(|d| d.order.side == OrderSide::Sell.into()),
            "Should have ask deltas"
        );
    }

    #[rstest]
    fn test_parse_rpi_books_update_uses_total_quantity_and_sequence() {
        let json_data = load_test_json("ws_books_rpi_update.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let (data, action) = match msg {
            OKXWsFrame::RpiBookData { data, action, .. } => (data, action),
            _ => panic!("Expected an RPI book update"),
        };
        let instrument_id = InstrumentId::from("OMI-USD.OKX");

        let deltas =
            parse_rpi_book_msg(&data[0], instrument_id, 7, 3, &action, UnixNanos::from(123))
                .unwrap();

        assert_eq!(deltas.instrument_id, instrument_id);
        assert_eq!(deltas.deltas.len(), 2);
        assert_eq!(deltas.flags, RecordFlag::F_LAST as u8);
        assert_eq!(deltas.sequence, 1_082_831_230);
        assert_eq!(deltas.ts_event, UnixNanos::from(1_785_406_443_903_000_000));
        assert_eq!(deltas.ts_init, UnixNanos::from(123));
        assert_eq!(deltas.deltas[0].action, BookAction::Delete);
        assert_eq!(deltas.deltas[0].order.side, OrderSide::Sell.into());
        assert_eq!(deltas.deltas[0].order.price, Price::from("0.0001617"));
        assert_eq!(deltas.deltas[0].order.size, Quantity::from("0"));
        assert_eq!(deltas.deltas[1].action, BookAction::Update);
        assert_eq!(deltas.deltas[1].order.side, OrderSide::Sell.into());
        assert_eq!(deltas.deltas[1].order.price, Price::from("0.0001625"));
        assert_eq!(deltas.deltas[1].order.size, Quantity::from("12324367.786"));
    }

    #[rstest]
    fn test_parse_tickers() {
        let json_data = load_test_json("ws_tickers.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let okx_tickers: Vec<OKXTickerMsg> = match msg {
            OKXWsFrame::Data { data, .. } => serde_json::from_value(data).unwrap(),
            _ => panic!("Expected a `Data` variant"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let trade =
            parse_ticker_msg(&okx_tickers[0], instrument_id, 2, 1, UnixNanos::default()).unwrap();

        assert_eq!(trade.instrument_id, InstrumentId::from("BTC-USDT.OKX"));
        assert_eq!(trade.bid_price, Price::from("8888.88"));
        assert_eq!(trade.ask_price, Price::from("9999.99"));
        assert_eq!(trade.bid_size, Quantity::from(5));
        assert_eq!(trade.ask_size, Quantity::from(11));
        assert_eq!(trade.ts_event, UnixNanos::from(1_597_026_383_085_000_000));
        assert_eq!(trade.ts_init, UnixNanos::default());
    }

    #[rstest]
    fn test_parse_quotes() {
        let json_data = load_test_json("ws_bbo_tbt.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let okx_quotes: Vec<OKXBookMsg> = match msg {
            OKXWsFrame::Data { data, .. } => serde_json::from_value(data).unwrap(),
            _ => panic!("Expected a `Data` variant"),
        };
        let instrument_id = InstrumentId::from("BTC-USDT.OKX");

        let quote =
            parse_quote_msg(&okx_quotes[0], instrument_id, 2, 1, UnixNanos::default()).unwrap();

        assert_eq!(quote.instrument_id, InstrumentId::from("BTC-USDT.OKX"));
        assert_eq!(quote.bid_price, Price::from("8476.97"));
        assert_eq!(quote.ask_price, Price::from("8476.98"));
        assert_eq!(quote.bid_size, Quantity::from(256));
        assert_eq!(quote.ask_size, Quantity::from(415));
        assert_eq!(quote.ts_event, UnixNanos::from(1_597_026_383_085_000_000));
        assert_eq!(quote.ts_init, UnixNanos::default());
    }

    #[rstest]
    fn test_parse_trades() {
        let json_data = load_test_json("ws_trades.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let okx_trades: Vec<OKXTradeMsg> = match msg {
            OKXWsFrame::Data { data, .. } => serde_json::from_value(data).unwrap(),
            _ => panic!("Expected a `Data` variant"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let trade =
            parse_trade_msg(&okx_trades[0], instrument_id, 1, 8, UnixNanos::default()).unwrap();

        assert_eq!(trade.instrument_id, InstrumentId::from("BTC-USDT.OKX"));
        assert_eq!(trade.price, Price::from("42219.9"));
        assert_eq!(trade.size, Quantity::from("0.12060306"));
        assert_eq!(trade.aggressor_side, AggressorSide::Buy);
        assert_eq!(trade.trade_id, TradeId::from("130639474"));
        assert_eq!(trade.ts_event, UnixNanos::from(1_630_048_897_897_000_000));
        assert_eq!(trade.ts_init, UnixNanos::default());
    }

    #[rstest]
    fn test_parse_candle() {
        let json_data = load_test_json("ws_candle.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let okx_candles: Vec<OKXCandleMsg> = match msg {
            OKXWsFrame::Data { data, .. } => serde_json::from_value(data).unwrap(),
            _ => panic!("Expected a `Data` variant"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let bar_type = BarType::new(
            instrument_id,
            BAR_SPEC_1_DAY_LAST,
            AggregationSource::External,
        );
        let bar = parse_candle_msg(&okx_candles[0], bar_type, 2, 0, UnixNanos::default()).unwrap();

        assert_eq!(bar.bar_type, bar_type);
        assert_eq!(bar.open, Price::from("8533.02"));
        assert_eq!(bar.high, Price::from("8553.74"));
        assert_eq!(bar.low, Price::from("8527.17"));
        assert_eq!(bar.close, Price::from("8548.26"));
        assert_eq!(bar.volume, Quantity::from(45247));
        assert_eq!(bar.ts_event, UnixNanos::from(1_597_026_383_085_000_000));
        assert_eq!(bar.ts_init, UnixNanos::default());
    }

    #[rstest]
    fn test_parse_funding_rate() {
        let json_data = load_test_json("ws_funding_rate.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();

        let okx_funding_rates: Vec<crate::websocket::messages::OKXFundingRateMsg> = match msg {
            OKXWsFrame::Data { data, .. } => serde_json::from_value(data).unwrap(),
            _ => panic!("Expected a `Data` variant"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let funding_rate =
            parse_funding_rate_msg(&okx_funding_rates[0], instrument_id, UnixNanos::default())
                .unwrap();

        assert_eq!(funding_rate.instrument_id, instrument_id);
        assert_eq!(funding_rate.rate, dec!(0.0001));
        assert_eq!(funding_rate.interval, Some(8 * 60));
        assert_eq!(
            funding_rate.next_funding_ns,
            Some(UnixNanos::from(1_744_590_349_506_000_000))
        );
        assert_eq!(
            funding_rate.ts_event,
            UnixNanos::from(1_744_590_349_506_000_000)
        );
        assert_eq!(funding_rate.ts_init, UnixNanos::default());
    }

    #[rstest]
    fn test_parse_book_vec() {
        let json_data = load_test_json("ws_books_snapshot.json");
        let event: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let (msgs, action): (Vec<OKXBookMsg>, OKXBookAction) = match event {
            OKXWsFrame::BookData { data, action, .. } => (data, action),
            _ => panic!("Expected BookData"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let deltas_vec =
            parse_book_msg_vec(msgs, &instrument_id, 8, 1, action, UnixNanos::default()).unwrap();

        assert_eq!(deltas_vec.len(), 1);

        if let Data::BookDeltas(d) = &deltas_vec[0] {
            assert_eq!(d.sequence, 123_456);
        } else {
            panic!("Expected Deltas");
        }
    }

    #[rstest]
    fn test_parse_ticker_vec() {
        let json_data = load_test_json("ws_tickers.json");
        let event: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let data_val: serde_json::Value = match event {
            OKXWsFrame::Data { data, .. } => data,
            _ => panic!("Expected Data"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let quotes_vec =
            parse_ticker_msg_vec(data_val, &instrument_id, 8, 1, UnixNanos::default()).unwrap();

        assert_eq!(quotes_vec.len(), 1);

        if let Data::Quote(q) = &quotes_vec[0] {
            assert_eq!(q.bid_price, Price::from("8888.88000000"));
            assert_eq!(q.ask_price, Price::from("9999.99"));
        } else {
            panic!("Expected Quote");
        }
    }

    #[rstest]
    fn test_parse_trade_vec() {
        let json_data = load_test_json("ws_trades.json");
        let event: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let data_val: serde_json::Value = match event {
            OKXWsFrame::Data { data, .. } => data,
            _ => panic!("Expected Data"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let trades_vec =
            parse_trade_msg_vec(data_val, &instrument_id, 8, 1, UnixNanos::default()).unwrap();

        assert_eq!(trades_vec.len(), 1);

        if let Data::Trade(t) = &trades_vec[0] {
            assert_eq!(t.trade_id, TradeId::new("130639474"));
        } else {
            panic!("Expected Trade");
        }
    }

    #[rstest]
    fn test_parse_candle_vec() {
        let json_data = load_test_json("ws_candle.json");
        let event: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let data_val: serde_json::Value = match event {
            OKXWsFrame::Data { data, .. } => data,
            _ => panic!("Expected Data"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let bars_vec = parse_candle_msg_vec(
            data_val,
            &instrument_id,
            2,
            1,
            BAR_SPEC_1_DAY_LAST,
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(bars_vec.len(), 1);

        if let Data::Bar(b) = &bars_vec[0] {
            assert_eq!(b.open, Price::from("8533.02"));
        } else {
            panic!("Expected Bar");
        }
    }

    #[rstest]
    fn test_parse_book_message() {
        let json_data = load_test_json("ws_bbo_tbt.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let (okx_books, arg): (Vec<OKXBookMsg>, OKXWebSocketArg) = match msg {
            OKXWsFrame::Data { data, arg, .. } => (serde_json::from_value(data).unwrap(), arg),
            _ => panic!("Expected a `Data` variant"),
        };

        assert_eq!(arg.channel, OKXWsChannel::BboTbt);
        assert_eq!(arg.inst_id.as_ref().unwrap(), &Ustr::from("BTC-USDT"));
        assert_eq!(arg.inst_type, None);
        assert_eq!(okx_books.len(), 1);

        let book_msg = &okx_books[0];

        // Check asks
        assert_eq!(book_msg.asks.len(), 1);
        let ask = &book_msg.asks[0];
        assert_eq!(ask.price, "8476.98");
        assert_eq!(ask.size, "415");
        assert_eq!(ask.liquidated_orders_count, "0");
        assert_eq!(ask.orders_count, "13");

        // Check bids
        assert_eq!(book_msg.bids.len(), 1);
        let bid = &book_msg.bids[0];
        assert_eq!(bid.price, "8476.97");
        assert_eq!(bid.size, "256");
        assert_eq!(bid.liquidated_orders_count, "0");
        assert_eq!(bid.orders_count, "12");
        assert_eq!(book_msg.ts, 1_597_026_383_085);
        assert_eq!(book_msg.seq_id, 123_456);
        assert_eq!(book_msg.checksum, None);
        assert_eq!(book_msg.prev_seq_id, None);
    }

    #[rstest]
    fn test_parse_ws_account_message() {
        let json_data = load_test_json("ws_account.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();

        let OKXWsFrame::Data { data, .. } = msg else {
            panic!("Expected OKXWsFrame::Data");
        };

        let accounts: Vec<OKXAccount> = serde_json::from_value(data).unwrap();

        assert_eq!(accounts.len(), 1);
        let account = &accounts[0];

        assert_eq!(account.total_eq, "100.56089404807182");
        assert_eq!(account.details.len(), 3);

        let usdt_detail = &account.details[0];
        assert_eq!(usdt_detail.ccy, "USDT");
        assert_eq!(usdt_detail.avail_bal, "100.52768569797846");
        assert_eq!(usdt_detail.cash_bal, "100.52768569797846");

        let btc_detail = &account.details[1];
        assert_eq!(btc_detail.ccy, "BTC");
        assert_eq!(btc_detail.avail_bal, "0.0000000051");

        let eth_detail = &account.details[2];
        assert_eq!(eth_detail.ccy, "ETH");
        assert_eq!(eth_detail.avail_bal, "0.000000185");

        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();
        let account_state = parse_account_state(account, account_id, AccountType::Margin, ts_init);

        assert!(account_state.is_ok());
        let state = account_state.unwrap();
        assert_eq!(state.account_id, account_id);
        assert_eq!(state.balances.len(), 3);
    }

    #[rstest]
    fn test_parse_ws_account_message_empty_balance() {
        // GH-3772: OKX returns empty strings and empty details for zero-balance accounts
        let json_data = load_test_json("ws_account_empty.json");
        let msg: OKXWsFrame = serde_json::from_str(&json_data).unwrap();

        let OKXWsFrame::Data { data, .. } = msg else {
            panic!("Expected OKXWsFrame::Data");
        };

        let accounts: Vec<OKXAccount> = serde_json::from_value(data).unwrap();
        assert_eq!(accounts.len(), 1);

        let account = &accounts[0];
        assert!(account.details.is_empty());
        assert_eq!(account.total_eq, "0");

        let account_id = AccountId::new("OKX-001");
        let account_state = parse_account_state(
            account,
            account_id,
            AccountType::Margin,
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(account_state.account_id, account_id);
        assert_eq!(account_state.margins.len(), 0);
        assert_eq!(account_state.balances.len(), 1);

        let balance = &account_state.balances[0];
        assert_eq!(balance.total, Money::new(0.0, Currency::USD()));
        assert_eq!(balance.free, Money::new(0.0, Currency::USD()));
        assert_eq!(balance.locked, Money::new(0.0, Currency::USD()));
    }

    #[rstest]
    fn test_parse_order_msg() {
        let json_data = load_test_json("ws_orders.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();

        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();

        // Create a mock instrument for testing
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();

        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let ts_init = UnixNanos::default();
        let mut fee_cache = FeeCache::new();
        let mut filled_qty_cache = FilledQtyCache::new();

        let result = parse_order_msg_vec(
            &data,
            account_id,
            &instruments,
            &mut fee_cache,
            &mut filled_qty_cache,
            ts_init,
        );

        assert!(result.is_ok());
        let order_reports = result.unwrap();
        assert_eq!(order_reports.len(), 1);

        // Verify the parsed order report
        let report = &order_reports[0];

        if let ExecutionReport::Fill(fill_report) = report {
            assert_eq!(fill_report.account_id, account_id);
            assert_eq!(fill_report.instrument_id, instrument_id);
            assert_eq!(
                fill_report.client_order_id,
                Some(ClientOrderId::new("001BTCUSDT20250106001"))
            );
            assert_eq!(
                fill_report.venue_order_id,
                VenueOrderId::new("2497956918703120384")
            );
            assert_eq!(fill_report.trade_id, TradeId::from("1518905529"));
            assert_eq!(fill_report.order_side, OrderSide::Buy);
            assert_eq!(fill_report.last_px, Price::from("103698.90"));
            assert_eq!(fill_report.last_qty, Quantity::from("0.03000000"));
            assert_eq!(fill_report.liquidity_side, LiquiditySide::Maker);
        } else {
            panic!("Expected Fill report for filled order");
        }
    }

    #[rstest]
    fn test_parse_order_status_report() {
        let json_data = load_test_json("ws_orders.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();
        let order_msg = &data[0];

        let account_id = AccountId::new("OKX-001");
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();

        let ts_init = UnixNanos::default();

        let result = parse_order_status_report(
            order_msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            ts_init,
        );

        assert!(result.is_ok());
        let order_status_report = result.unwrap();

        assert_eq!(order_status_report.account_id, account_id);
        assert_eq!(order_status_report.instrument_id, instrument_id);
        assert_eq!(
            order_status_report.client_order_id,
            Some(ClientOrderId::new("001BTCUSDT20250106001"))
        );
        assert_eq!(
            order_status_report.venue_order_id,
            VenueOrderId::new("2497956918703120384")
        );
        assert_eq!(order_status_report.order_side, OrderSide::Buy.into());
        assert_eq!(order_status_report.order_status, OrderStatus::Filled);
        assert_eq!(order_status_report.quantity, Quantity::from("0.03000000"));
        assert_eq!(order_status_report.filled_qty, Quantity::from("0.03000000"));
    }

    #[rstest]
    fn test_parse_fill_report() {
        let json_data = load_test_json("ws_orders.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();
        let order_msg = &data[0];

        let account_id = AccountId::new("OKX-001");
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();

        let ts_init = UnixNanos::default();

        let result = parse_fill_report(
            order_msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            None,
            None,
            ts_init,
        );

        assert!(result.is_ok());
        let fill_report = result.unwrap().unwrap();

        assert_eq!(fill_report.account_id, account_id);
        assert_eq!(fill_report.instrument_id, instrument_id);
        assert_eq!(
            fill_report.client_order_id,
            Some(ClientOrderId::new("001BTCUSDT20250106001"))
        );
        assert_eq!(
            fill_report.venue_order_id,
            VenueOrderId::new("2497956918703120384")
        );
        assert_eq!(fill_report.trade_id, TradeId::from("1518905529"));
        assert_eq!(fill_report.order_side, OrderSide::Buy);
        assert_eq!(fill_report.last_px, Price::from("103698.90"));
        assert_eq!(fill_report.last_qty, Quantity::from("0.03000000"));
        assert_eq!(fill_report.liquidity_side, LiquiditySide::Maker);
    }

    #[rstest]
    fn test_parse_fill_report_rejects_missing_fee() {
        let instrument = create_stub_instrument();
        let mut msg = create_stub_order_msg("0.01", None, "ord-1", "trade-1");
        msg.fee = None;

        let error = parse_fill_report(
            &msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            AccountId::new("OKX-001"),
            None,
            None,
            UnixNanos::default(),
        )
        .unwrap_err();

        assert!(error.to_string().contains("missing fee"));
    }

    #[rstest]
    fn test_parse_book_depth_msg() {
        let json_data = load_test_json("ws_books_snapshot.json");
        let event: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let msgs: Vec<OKXBookMsg> = match event {
            OKXWsFrame::BookData { data, .. } => data,
            _ => panic!("Expected BookData"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let depth =
            parse_book_depth_msg(&msgs[0], instrument_id, 2, 0, UnixNanos::default()).unwrap();

        let expected_bids = [
            ("8476.97", "256"),
            ("8475.55", "101"),
            ("8475.54", "100"),
            ("8475.30", "1"),
            ("8447.32", "6"),
            ("8447.02", "246"),
            ("8446.83", "24"),
            ("8446.00", "95"),
        ];
        let expected_asks = [
            ("8476.98", "415"),
            ("8477.00", "7"),
            ("8477.34", "85"),
            ("8477.56", "1"),
            ("8505.84", "8"),
            ("8506.37", "85"),
            ("8506.49", "2"),
            ("8506.96", "100"),
        ];

        assert_eq!(depth.instrument_id, instrument_id);
        assert_eq!(depth.sequence, 123_456);
        assert_eq!(depth.ts_event, UnixNanos::from(1_597_026_383_085_000_000));
        assert_eq!(depth.ts_init, UnixNanos::default());
        assert_eq!(depth.flags, RecordFlag::F_SNAPSHOT as u8);
        assert_eq!(depth.bids.len(), expected_bids.len());
        assert_eq!(depth.asks.len(), expected_asks.len());
        assert_eq!(depth.bid_counts.as_slice(), &[12, 1, 1, 1, 1, 1, 1, 3]);
        assert_eq!(depth.ask_counts.as_slice(), &[13, 2, 1, 1, 1, 1, 1, 2]);
        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);
        }
    }

    #[rstest]
    fn test_parse_book_depth_msg_vec() {
        let json_data = load_test_json("ws_books_snapshot.json");
        let event: OKXWsFrame = serde_json::from_str(&json_data).unwrap();
        let msgs: Vec<OKXBookMsg> = match event {
            OKXWsFrame::BookData { data, .. } => data,
            _ => panic!("Expected BookData"),
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let depth_vec =
            parse_book_depth_msg_vec(msgs, &instrument_id, 2, 0, UnixNanos::default()).unwrap();

        assert_eq!(depth_vec.len(), 1);

        if let Data::BookDepth(d) = &depth_vec[0] {
            assert_eq!(d.instrument_id, instrument_id);
            assert_eq!(d.sequence, 123_456);
            assert_eq!(d.bids[0].price, Price::from("8476.97"));
            assert_eq!(d.asks[0].price, Price::from("8476.98"));
        } else {
            panic!("Expected Depth");
        }
    }

    #[rstest]
    fn test_parse_fill_report_with_fee_cache() {
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();

        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        // First fill: 0.01 BTC out of 0.03 BTC total (1/3)
        let order_msg_1 = OKXOrderMsg {
            acc_fill_sz: Some("0.01".to_string()),
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_order_1".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("-1.0".to_string()), // Total fee so far
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "50000.0".to_string(),
            fill_sz: "0.01".to_string(),
            fill_time: 1_746_947_317_402,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("1234567890"),
            ord_type: OKXOrderType::Market,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::PartiallyFilled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Maker,
            sz: "0.03".to_string(), // Total order size
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: "trade_1".to_string(),
            u_time: 1_746_947_317_402,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let fill_report_1 = parse_fill_report(
            &order_msg_1,
            &InstrumentAny::CryptoPerpetual(instrument.clone()),
            account_id,
            None,
            None,
            ts_init,
        )
        .unwrap()
        .unwrap();

        // First fill should get the full fee since there's no previous fee
        assert_eq!(fill_report_1.commission, Money::new(1.0, Currency::USDT()));

        // Second fill: 0.02 BTC more, now 0.03 BTC total (completely filled)
        let order_msg_2 = OKXOrderMsg {
            acc_fill_sz: Some("0.03".to_string()),
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_order_1".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("-3.0".to_string()), // Same total fee
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "50000.0".to_string(),
            fill_sz: "0.02".to_string(),
            fill_time: 1_746_947_317_403,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("1234567890"),
            ord_type: OKXOrderType::Market,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::Filled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Maker,
            sz: "0.03".to_string(), // Same total order size
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: "trade_2".to_string(),
            u_time: 1_746_947_317_403,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let fill_report_2 = parse_fill_report(
            &order_msg_2,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            Some(fill_report_1.commission),
            Some(fill_report_1.last_qty),
            ts_init,
        )
        .unwrap()
        .unwrap();

        // Second fill should get total_fee - previous_fee = 3.0 - 1.0 = 2.0
        assert_eq!(fill_report_2.commission, Money::new(2.0, Currency::USDT()));

        // Test passed - fee was correctly split proportionally
    }

    #[rstest]
    fn test_parse_fill_report_with_maker_rebates() {
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();

        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        // First fill: maker rebate of $0.5 (OKX sends as "0.5", parse_fee makes it -0.5)
        let order_msg_1 = OKXOrderMsg {
            acc_fill_sz: Some("0.01".to_string()),
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_order_rebate".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("0.5".to_string()), // Rebate: positive value from OKX
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "50000.0".to_string(),
            fill_sz: "0.01".to_string(),
            fill_time: 1_746_947_317_402,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("rebate_order_123"),
            ord_type: OKXOrderType::Market,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::PartiallyFilled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Maker,
            sz: "0.02".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: "trade_rebate_1".to_string(),
            u_time: 1_746_947_317_402,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let fill_report_1 = parse_fill_report(
            &order_msg_1,
            &InstrumentAny::CryptoPerpetual(instrument.clone()),
            account_id,
            None,
            None,
            ts_init,
        )
        .unwrap()
        .unwrap();

        // First fill gets the full rebate (negative commission)
        assert_eq!(fill_report_1.commission, Money::new(-0.5, Currency::USDT()));

        // Second fill: another maker rebate of $0.3, cumulative now $0.8
        let order_msg_2 = OKXOrderMsg {
            acc_fill_sz: Some("0.02".to_string()),
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_order_rebate".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("0.8".to_string()), // Cumulative rebate
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "50000.0".to_string(),
            fill_sz: "0.01".to_string(),
            fill_time: 1_746_947_317_403,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("rebate_order_123"),
            ord_type: OKXOrderType::Market,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::Filled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Maker,
            sz: "0.02".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: "trade_rebate_2".to_string(),
            u_time: 1_746_947_317_403,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let fill_report_2 = parse_fill_report(
            &order_msg_2,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            Some(fill_report_1.commission),
            Some(fill_report_1.last_qty),
            ts_init,
        )
        .unwrap()
        .unwrap();

        // Second fill: incremental = -0.8 - (-0.5) = -0.3
        assert_eq!(fill_report_2.commission, Money::new(-0.3, Currency::USDT()));
    }

    #[rstest]
    fn test_parse_fill_report_rebate_to_charge_transition() {
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();

        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        // First fill: maker rebate of $1.0
        let order_msg_1 = OKXOrderMsg {
            acc_fill_sz: Some("0.01".to_string()),
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_order_transition".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("1.0".to_string()), // Rebate from OKX
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "50000.0".to_string(),
            fill_sz: "0.01".to_string(),
            fill_time: 1_746_947_317_402,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("transition_order_456"),
            ord_type: OKXOrderType::Market,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::PartiallyFilled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Maker,
            sz: "0.02".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: "trade_transition_1".to_string(),
            u_time: 1_746_947_317_402,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let fill_report_1 = parse_fill_report(
            &order_msg_1,
            &InstrumentAny::CryptoPerpetual(instrument.clone()),
            account_id,
            None,
            None,
            ts_init,
        )
        .unwrap()
        .unwrap();

        // First fill gets rebate (negative)
        assert_eq!(fill_report_1.commission, Money::new(-1.0, Currency::USDT()));

        // Second fill: taker charge of $5.0, net cumulative is now $2.0 charge
        // This is the edge case: incremental = 2.0 - (-1.0) = 3.0, which exceeds total (2.0)
        // But it's legitimate, not corruption
        let order_msg_2 = OKXOrderMsg {
            acc_fill_sz: Some("0.02".to_string()),
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_order_transition".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("-2.0".to_string()), // Now a charge (negative from OKX)
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "50000.0".to_string(),
            fill_sz: "0.01".to_string(),
            fill_time: 1_746_947_317_403,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("transition_order_456"),
            ord_type: OKXOrderType::Market,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::Filled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Taker,
            sz: "0.02".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: "trade_transition_2".to_string(),
            u_time: 1_746_947_317_403,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let fill_report_2 = parse_fill_report(
            &order_msg_2,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            Some(fill_report_1.commission),
            Some(fill_report_1.last_qty),
            ts_init,
        )
        .unwrap()
        .unwrap();

        // Second fill: incremental = 2.0 - (-1.0) = 3.0
        // This should NOT trigger corruption detection because previous was negative
        assert_eq!(fill_report_2.commission, Money::new(3.0, Currency::USDT()));
    }

    #[rstest]
    fn test_parse_fill_report_negative_incremental() {
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();

        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        // First fill: charge of $2.0
        let order_msg_1 = OKXOrderMsg {
            acc_fill_sz: Some("0.01".to_string()),
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_order_neg_inc".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("-2.0".to_string()),
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "50000.0".to_string(),
            fill_sz: "0.01".to_string(),
            fill_time: 1_746_947_317_402,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("neg_inc_order_789"),
            ord_type: OKXOrderType::Market,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::PartiallyFilled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Taker,
            sz: "0.02".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: "trade_neg_inc_1".to_string(),
            u_time: 1_746_947_317_402,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let fill_report_1 = parse_fill_report(
            &order_msg_1,
            &InstrumentAny::CryptoPerpetual(instrument.clone()),
            account_id,
            None,
            None,
            ts_init,
        )
        .unwrap()
        .unwrap();

        assert_eq!(fill_report_1.commission, Money::new(2.0, Currency::USDT()));

        // Second fill: charge reduced to $1.5 total (refund or maker rebate on this fill)
        // Incremental = 1.5 - 2.0 = -0.5 (negative incremental triggers debug log)
        let order_msg_2 = OKXOrderMsg {
            acc_fill_sz: Some("0.02".to_string()),
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_order_neg_inc".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("-1.5".to_string()), // Total reduced
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "50000.0".to_string(),
            fill_sz: "0.01".to_string(),
            fill_time: 1_746_947_317_403,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("neg_inc_order_789"),
            ord_type: OKXOrderType::Market,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::Filled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Maker,
            sz: "0.02".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: "trade_neg_inc_2".to_string(),
            u_time: 1_746_947_317_403,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let fill_report_2 = parse_fill_report(
            &order_msg_2,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            Some(fill_report_1.commission),
            Some(fill_report_1.last_qty),
            ts_init,
        )
        .unwrap()
        .unwrap();

        // Incremental is negative: 1.5 - 2.0 = -0.5
        assert_eq!(fill_report_2.commission, Money::new(-0.5, Currency::USDT()));
    }

    #[rstest]
    fn test_parse_fill_report_fee_currency_change_no_panic() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        // First fill charged in USDT
        let previous_fee = Money::new(1.0, Currency::USDT());

        // Second fill charged in BTC (fee currency changed)
        let mut order_msg =
            create_stub_order_msg("0.01", Some("0.02".to_string()), "1234567890", "trade_2");
        order_msg.fee = Some("-0.00005".to_string());
        order_msg.fee_ccy = Ustr::from("BTC");

        let result = parse_fill_report(
            &order_msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            Some(previous_fee),
            Some(Quantity::from("0.01")),
            ts_init,
        );

        let fill_report = result.unwrap().unwrap();
        assert_eq!(fill_report.commission.currency, Currency::BTC());
    }

    #[rstest]
    fn test_parse_fill_report_empty_fill_sz_first_fill() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        let order_msg =
            create_stub_order_msg("", Some("0.01".to_string()), "1234567890", "trade_1");

        let fill_report = parse_fill_report(
            &order_msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            None,
            None,
            ts_init,
        )
        .unwrap()
        .unwrap();

        assert_eq!(fill_report.last_qty, Quantity::from("0.01"));
    }

    #[rstest]
    fn test_parse_fill_report_empty_fill_sz_subsequent_fills() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        let order_msg_1 =
            create_stub_order_msg("", Some("0.01".to_string()), "1234567890", "trade_1");

        let fill_report_1 = parse_fill_report(
            &order_msg_1,
            &InstrumentAny::CryptoPerpetual(instrument.clone()),
            account_id,
            None,
            None,
            ts_init,
        )
        .unwrap()
        .unwrap();

        assert_eq!(fill_report_1.last_qty, Quantity::from("0.01"));

        let order_msg_2 =
            create_stub_order_msg("", Some("0.03".to_string()), "1234567890", "trade_2");

        let fill_report_2 = parse_fill_report(
            &order_msg_2,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            Some(fill_report_1.commission),
            Some(fill_report_1.last_qty),
            ts_init,
        )
        .unwrap()
        .unwrap();

        assert_eq!(fill_report_2.last_qty, Quantity::from("0.02"));
    }

    #[rstest]
    fn test_parse_fill_report_error_both_empty() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        let order_msg = create_stub_order_msg("", Some(String::new()), "1234567890", "trade_1");

        let result = parse_fill_report(
            &order_msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            None,
            None,
            ts_init,
        );

        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("Cannot determine fill quantity"));
        assert!(err_msg.contains("empty/zero"));
    }

    #[rstest]
    fn test_parse_fill_report_error_acc_fill_sz_none() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        let order_msg = create_stub_order_msg("", None, "1234567890", "trade_1");

        let result = parse_fill_report(
            &order_msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            None,
            None,
            ts_init,
        );

        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("Cannot determine fill quantity"));
        assert!(err_msg.contains("acc_fill_sz is None"));
    }

    #[rstest]
    fn test_parse_fill_report_error_acc_fill_sz_less_than_previous() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        // acc_fill_sz (0.01) < previous_filled_qty (0.03) - stale data after reconnect
        let order_msg =
            create_stub_order_msg("", Some("0.01".to_string()), "1234567890", "trade_2");

        let result = parse_fill_report(
            &order_msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            account_id,
            None,
            Some(Quantity::from("0.03")),
            ts_init,
        );

        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("Cumulative fill went backwards"));
    }

    #[rstest]
    fn test_parse_order_msg_acc_fill_sz_only_update() {
        // Test that we emit fill reports when OKX only updates acc_fill_sz without fill_sz or trade_id
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let fee_cache = FeeCache::new();
        let mut filled_qty_cache = FilledQtyCache::new();

        // First update: acc_fill_sz = 0.01, no fill_sz, no trade_id
        let msg_1 = create_stub_order_msg("", Some("0.01".to_string()), "1234567890", "");

        let report_1 = parse_order_msg(
            &msg_1,
            account_id,
            &instruments,
            &fee_cache,
            &filled_qty_cache,
            ts_init,
        )
        .unwrap();

        // Should generate a fill report (not a status report)
        assert!(matches!(report_1, ExecutionReport::Fill(_)));
        if let ExecutionReport::Fill(fill) = &report_1 {
            assert_eq!(fill.last_qty, Quantity::from("0.01"));
        }

        // Update cache
        filled_qty_cache.record(Ustr::from("1234567890"), Quantity::from("0.01"), false);

        // Second update: acc_fill_sz increased to 0.03, still no fill_sz or trade_id
        let msg_2 = create_stub_order_msg("", Some("0.03".to_string()), "1234567890", "");

        let report_2 = parse_order_msg(
            &msg_2,
            account_id,
            &instruments,
            &fee_cache,
            &filled_qty_cache,
            ts_init,
        )
        .unwrap();

        // Should still generate a fill report for the incremental 0.02
        assert!(matches!(report_2, ExecutionReport::Fill(_)));
        if let ExecutionReport::Fill(fill) = &report_2 {
            assert_eq!(fill.last_qty, Quantity::from("0.02"));
        }
    }

    #[rstest]
    fn test_parse_book_depth_msg_partial_levels() {
        let book_msg = OKXBookMsg {
            asks: vec![
                OrderBookEntry {
                    price: "8476.98".to_string(),
                    size: "415".to_string(),
                    liquidated_orders_count: "0".to_string(),
                    orders_count: "13".to_string(),
                },
                OrderBookEntry {
                    price: "8477.00".to_string(),
                    size: "7".to_string(),
                    liquidated_orders_count: "0".to_string(),
                    orders_count: "2".to_string(),
                },
            ],
            bids: vec![OrderBookEntry {
                price: "8476.97".to_string(),
                size: "256".to_string(),
                liquidated_orders_count: "0".to_string(),
                orders_count: "12".to_string(),
            }],
            ts: 1_597_026_383_085,
            checksum: None,
            prev_seq_id: None,
            seq_id: 123_456,
        };

        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let depth =
            parse_book_depth_msg(&book_msg, instrument_id, 2, 0, UnixNanos::default()).unwrap();

        assert_eq!(depth.instrument_id, instrument_id);
        assert_eq!(depth.bids.len(), 1);
        assert_eq!(depth.asks.len(), 2);
        assert_eq!(depth.bids[0].price, Price::from("8476.97"));
        assert_eq!(depth.bids[0].size, Quantity::from("256"));
        assert_eq!(depth.bids[0].side, Some(OrderSide::Buy));
        assert_eq!(depth.bids[0].order_id, 0);
        assert_eq!(depth.bid_counts.as_slice(), &[12]);
        assert_eq!(depth.ask_counts.as_slice(), &[13, 2]);
        for (order, (price, size)) in depth
            .asks
            .iter()
            .zip([("8476.98", "415"), ("8477.00", "7")])
        {
            assert_eq!(order.price, Price::from(price));
            assert_eq!(order.size, Quantity::from(size));
            assert_eq!(order.side, Some(OrderSide::Sell));
            assert_eq!(order.order_id, 0);
        }
        assert_eq!(depth.sequence, 123_456);
        assert_eq!(depth.flags, RecordFlag::F_SNAPSHOT as u8);
        assert_eq!(depth.ts_event, UnixNanos::from(1_597_026_383_085_000_000));
        assert_eq!(depth.ts_init, UnixNanos::default());
    }

    #[rstest]
    fn test_parse_algo_order_msg_stop_market() {
        let json_data = load_test_json("ws_orders_algo.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXAlgoOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        // Test first algo order (stop market sell)
        let msg = &data[0];
        assert_eq!(msg.algo_id, "706620792746729472");
        assert_eq!(msg.algo_cl_ord_id, "STOP001BTCUSDT20250120");
        assert_eq!(msg.state, OKXAlgoOrderStatus::Live);
        assert_eq!(msg.ord_px, "-1"); // Market order indicator

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();

        // Create mock instrument
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(0.into())
            .ts_init(0.into())
            .build()
            .unwrap();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let result = parse_algo_order_msg(msg, account_id, &instruments, UnixNanos::default());

        let report = result.unwrap().unwrap();

        if let ExecutionReport::Order(status_report) = report {
            assert_eq!(status_report.order_type, OrderType::StopMarket);
            assert_eq!(status_report.order_side, OrderSide::Sell.into());
            assert_eq!(status_report.quantity, Quantity::from("0.01000000"));
            assert_eq!(status_report.trigger_price, Some(Price::from("95000.00")));
            assert_eq!(status_report.trigger_type, Some(TriggerType::LastPrice));
            assert_eq!(status_report.price, None); // No limit price for market orders
        } else {
            panic!("Expected Order report");
        }
    }

    #[rstest]
    fn test_parse_algo_order_msg_stop_limit() {
        let json_data = load_test_json("ws_orders_algo.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXAlgoOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        // Test second algo order (stop limit buy)
        let msg = &data[1];
        assert_eq!(msg.algo_id, "706620792746729473");
        assert_eq!(msg.state, OKXAlgoOrderStatus::Live);
        assert_eq!(msg.ord_px, "106000"); // Limit price

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();

        // Create mock instrument
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(0.into())
            .ts_init(0.into())
            .build()
            .unwrap();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let result = parse_algo_order_msg(msg, account_id, &instruments, UnixNanos::default());

        let report = result.unwrap().unwrap();

        if let ExecutionReport::Order(status_report) = report {
            assert_eq!(status_report.order_type, OrderType::StopLimit);
            assert_eq!(status_report.order_side, OrderSide::Buy.into());
            assert_eq!(status_report.quantity, Quantity::from("0.02000000"));
            assert_eq!(status_report.trigger_price, Some(Price::from("105000.00")));
            assert_eq!(status_report.trigger_type, Some(TriggerType::MarkPrice));
            assert_eq!(status_report.price, Some(Price::from("106000.00"))); // Has limit price
        } else {
            panic!("Expected Order report");
        }
    }

    #[rstest]
    fn test_parse_triggered_algo_order_preserves_parent_identity() {
        let json_data = load_test_json("ws_orders_algo.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXAlgoOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();
        let msg = &data[2];

        assert_eq!(msg.state, OKXAlgoOrderStatus::OrderPlaced);
        assert_eq!(msg.algo_cl_ord_id, "STOP003BTCUSDT20250120");
        assert_eq!(msg.cl_ord_id, "706620792746729474_0");
        assert_eq!(msg.actual_sz, "0.01");

        let account_id = AccountId::new("OKX-001");
        let instrument = create_stub_instrument();
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let report = parse_algo_order_msg(msg, account_id, &instruments, UnixNanos::default())
            .unwrap()
            .unwrap();
        let ExecutionReport::Order(report) = report else {
            panic!("Expected Order report");
        };

        assert_eq!(
            report.client_order_id,
            Some(ClientOrderId::from("STOP003BTCUSDT20250120"))
        );
        assert_eq!(
            report.venue_order_id,
            VenueOrderId::from("706620792746729999")
        );
        assert_eq!(report.order_status, OrderStatus::Triggered);
        assert_eq!(report.filled_qty, Quantity::from("0.00000000"));
    }

    #[rstest]
    fn test_parse_filled_algo_order_uses_actual_quantity() {
        let json_data = load_test_json("ws_orders_algo.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXAlgoOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();
        let msg = &data[4];

        assert_eq!(msg.state, OKXAlgoOrderStatus::Filled);
        assert_eq!(msg.actual_sz, "0.005");

        let instrument = create_stub_instrument();
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let report = parse_algo_order_msg(
            msg,
            AccountId::new("OKX-001"),
            &instruments,
            UnixNanos::default(),
        )
        .unwrap()
        .unwrap();
        let ExecutionReport::Order(report) = report else {
            panic!("Expected Order report");
        };

        assert_eq!(report.order_status, OrderStatus::Filled);
        assert_eq!(report.filled_qty, Quantity::from("0.00500000"));
    }

    #[rstest]
    fn test_parse_trigger_order_from_regular_channel() {
        let json_data = load_test_json("ws_orders_trigger.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        // Test triggered order that came through regular orders channel
        let msg = &data[0];
        assert_eq!(msg.ord_type, OKXOrderType::Trigger);
        assert_eq!(msg.state, OKXOrderStatus::Filled);

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();

        // Create mock instrument
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(0.into())
            .ts_init(0.into())
            .build()
            .unwrap();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let mut fee_cache = FeeCache::new();
        let mut filled_qty_cache = FilledQtyCache::new();

        let result = parse_order_msg_vec(
            std::slice::from_ref(msg),
            account_id,
            &instruments,
            &mut fee_cache,
            &mut filled_qty_cache,
            UnixNanos::default(),
        );

        assert!(result.is_ok());
        let reports = result.unwrap();
        assert_eq!(reports.len(), 1);

        if let ExecutionReport::Fill(fill_report) = &reports[0] {
            assert_eq!(fill_report.order_side, OrderSide::Sell);
            assert_eq!(fill_report.last_qty, Quantity::from("0.01000000"));
            assert_eq!(fill_report.last_px, Price::from("101950.00"));
        } else {
            panic!("Expected Fill report for filled trigger order");
        }
    }

    #[rstest]
    fn test_parse_liquidation_order() {
        let json_data = load_test_json("ws_orders_liquidation.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        // Test liquidation order
        let msg = &data[0];
        assert_eq!(msg.category, OKXOrderCategory::FullLiquidation);
        assert_eq!(msg.state, OKXOrderStatus::Filled);
        assert_eq!(msg.inst_id, "BTC-USDT-SWAP");

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();

        // Create mock instrument
        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(0.into())
            .ts_init(0.into())
            .build()
            .unwrap();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );
        let mut fee_cache = FeeCache::new();
        let mut filled_qty_cache = FilledQtyCache::new();

        let result = parse_order_msg_vec(
            std::slice::from_ref(msg),
            account_id,
            &instruments,
            &mut fee_cache,
            &mut filled_qty_cache,
            UnixNanos::default(),
        );

        assert!(result.is_ok());
        let reports = result.unwrap();
        assert_eq!(reports.len(), 1);

        // Verify it's a fill report for a liquidation
        if let ExecutionReport::Fill(fill_report) = &reports[0] {
            assert_eq!(fill_report.order_side, OrderSide::Sell);
            assert_eq!(fill_report.last_qty, Quantity::from("0.50000000"));
            assert_eq!(fill_report.last_px, Price::from("40000.00"));
            assert_eq!(fill_report.liquidity_side, LiquiditySide::Taker);
        } else {
            panic!("Expected Fill report for liquidation order");
        }
    }

    #[rstest]
    fn test_parse_adl_order() {
        let json_data = load_test_json("ws_orders_adl.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        // Test ADL order
        let msg = &data[0];
        assert_eq!(msg.category, OKXOrderCategory::Adl);
        assert_eq!(msg.state, OKXOrderStatus::Filled);
        assert_eq!(msg.inst_id, "ETH-USDT-SWAP");

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();

        // Create mock instrument
        let instrument_id = InstrumentId::from("ETH-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("ETH-USDT-SWAP"))
            .base_currency(Currency::ETH())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(0.into())
            .ts_init(0.into())
            .build()
            .unwrap();
        instruments.insert(
            Ustr::from("ETH-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let mut fee_cache = FeeCache::new();
        let mut filled_qty_cache = FilledQtyCache::new();

        let result = parse_order_msg_vec(
            std::slice::from_ref(msg),
            account_id,
            &instruments,
            &mut fee_cache,
            &mut filled_qty_cache,
            UnixNanos::default(),
        );

        assert!(result.is_ok());
        let reports = result.unwrap();
        assert_eq!(reports.len(), 1);

        // Verify it's a fill report for ADL
        if let ExecutionReport::Fill(fill_report) = &reports[0] {
            assert_eq!(fill_report.order_side, OrderSide::Buy);
            assert_eq!(fill_report.last_qty, Quantity::from("0.30000000"));
            assert_eq!(fill_report.last_px, Price::from("41000.00"));
            assert_eq!(fill_report.liquidity_side, LiquiditySide::Taker);
        } else {
            panic!("Expected Fill report for ADL order");
        }
    }

    #[rstest]
    fn test_parse_unknown_category_graceful_fallback() {
        // Test that unknown/future category values deserialize as Other instead of failing
        let json_with_unknown_category = r#"{
            "category": "some_future_category_we_dont_know"
        }"#;

        let result: Result<serde_json::Value, _> = serde_json::from_str(json_with_unknown_category);
        result.unwrap();

        // Test deserialization of the category field directly
        let category_result: Result<OKXOrderCategory, _> =
            serde_json::from_str(r#""some_future_category""#);
        assert!(category_result.is_ok());
        assert_eq!(category_result.unwrap(), OKXOrderCategory::Other);

        // Verify known categories still work
        let normal: OKXOrderCategory = serde_json::from_str(r#""normal""#).unwrap();
        assert_eq!(normal, OKXOrderCategory::Normal);

        let twap: OKXOrderCategory = serde_json::from_str(r#""twap""#).unwrap();
        assert_eq!(twap, OKXOrderCategory::Twap);
    }

    #[rstest]
    fn test_parse_partial_liquidation_order() {
        // Create a test message with partial liquidation category
        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();

        let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
        let instrument = CryptoPerpetual::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
            .base_currency(Currency::BTC())
            .quote_currency(Currency::USDT())
            .settlement_currency(Currency::USDT())
            .is_inverse(false)
            .price_precision(2)
            .size_precision(8)
            .price_increment(Price::from("0.01"))
            .size_increment(Quantity::from("0.00000001"))
            .ts_event(0.into())
            .ts_init(0.into())
            .build()
            .unwrap();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let partial_liq_msg = OKXOrderMsg {
            acc_fill_sz: Some("0.25".to_string()),
            algo_id: None,
            avg_px: "39000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::PartialLiquidation,
            ccy: Ustr::from("USDT"),
            cl_ord_id: String::new(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("-9.75".to_string()),
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: "39000.0".to_string(),
            fill_sz: "0.25".to_string(),
            fill_time: 1_746_947_317_402,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "10.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("2497956918703120888"),
            ord_type: OKXOrderType::Market,
            pnl: "-2500".to_string(),
            pos_side: OKXPositionSide::Long,
            px: String::new(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Sell,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::Filled,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Taker,
            sz: "0.25".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: "1518905888".to_string(),
            u_time: 1_746_947_317_402,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let fee_cache = FeeCache::new();
        let filled_qty_cache = FilledQtyCache::new();
        let result = parse_order_msg(
            &partial_liq_msg,
            account_id,
            &instruments,
            &fee_cache,
            &filled_qty_cache,
            UnixNanos::default(),
        );

        assert!(result.is_ok());
        let report = result.unwrap();

        // Verify it's a fill report for partial liquidation
        if let ExecutionReport::Fill(fill_report) = report {
            assert_eq!(fill_report.order_side, OrderSide::Sell);
            assert_eq!(fill_report.last_qty, Quantity::from("0.25000000"));
            assert_eq!(fill_report.last_px, Price::from("39000.00"));
        } else {
            panic!("Expected Fill report for partial liquidation order");
        }
    }

    #[rstest]
    fn test_parse_mmp_canceled_order_message() {
        use nautilus_model::instruments::stubs::crypto_option_btc_deribit;

        let json_data = load_test_json("ws_orders_mmp_canceled.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        let msg = &data[0];
        assert_eq!(msg.state, OKXOrderStatus::MmpCanceled);
        assert_eq!(msg.ord_type, OKXOrderType::MmpAndPostOnly);

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();
        let instrument_id = InstrumentId::from("BTC-USD-250328-92000-C.OKX");
        let mut option =
            crypto_option_btc_deribit(3, 1, Price::from("0.001"), Quantity::from("0.1"));
        option.id = instrument_id;
        option.raw_symbol = Symbol::from("BTC-USD-250328-92000-C");
        instruments.insert(
            Ustr::from("BTC-USD-250328-92000-C"),
            InstrumentAny::CryptoOption(option),
        );

        let fee_cache = FeeCache::new();
        let filled_qty_cache = FilledQtyCache::new();
        let report = parse_order_msg(
            msg,
            account_id,
            &instruments,
            &fee_cache,
            &filled_qty_cache,
            UnixNanos::default(),
        )
        .unwrap();

        match report {
            ExecutionReport::Order(report) => {
                // MMP cancels map to Canceled with the venue reason preserved
                assert_eq!(report.order_status, OrderStatus::Canceled);
                assert_eq!(report.instrument_id, instrument_id);
                assert!(
                    report
                        .cancel_reason
                        .as_ref()
                        .is_some_and(|reason| reason.as_str().contains("market maker protection")),
                );
            }
            other => panic!("Expected Order report for MMP-canceled order, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_order_msg_unknown_state_preserves_fill() {
        let json_data = load_test_json("ws_orders_unknown_state_fill.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        // Unknown states fall back to `Unknown` instead of failing deserialization
        let msg = &data[0];
        assert_eq!(msg.state, OKXOrderStatus::Unknown);

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(create_stub_instrument()),
        );

        let fee_cache = FeeCache::new();
        let filled_qty_cache = FilledQtyCache::new();
        let report = parse_order_msg(
            msg,
            account_id,
            &instruments,
            &fee_cache,
            &filled_qty_cache,
            UnixNanos::default(),
        )
        .unwrap();

        // Fill data is processed even when the state itself is unrecognized
        match report {
            ExecutionReport::Fill(fill_report) => {
                assert_eq!(fill_report.order_side, OrderSide::Buy);
                assert_eq!(fill_report.last_qty, Quantity::from("0.25000000"));
                assert_eq!(fill_report.last_px, Price::from("40000.00"));
                assert_eq!(fill_report.liquidity_side, LiquiditySide::Taker);
            }
            other => panic!("Expected Fill report for unknown-state order, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_order_msg_unknown_state_without_fill_errors() {
        let json_data = load_test_json("ws_orders_unknown_state_fill.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        let mut msg = data[0].clone();
        msg.fill_sz = "0".to_string();
        msg.trade_id = String::new();
        msg.acc_fill_sz = Some("0".to_string());

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(create_stub_instrument()),
        );

        let fee_cache = FeeCache::new();
        let filled_qty_cache = FilledQtyCache::new();
        let result = parse_order_msg(
            &msg,
            account_id,
            &instruments,
            &fee_cache,
            &filled_qty_cache,
            UnixNanos::default(),
        );

        // Without fill data there is nothing safe to emit for an unrecognized state
        assert!(result.is_err());
    }

    #[rstest]
    fn test_parse_order_msg_unknown_order_type_preserves_fill() {
        let json_data = load_test_json("ws_orders_unknown_ord_type_fill.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        // Unknown order types fall back to `Other` instead of failing deserialization
        let msg = &data[0];
        assert_eq!(msg.ord_type, OKXOrderType::Other);

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(create_stub_instrument()),
        );

        let fee_cache = FeeCache::new();
        let filled_qty_cache = FilledQtyCache::new();
        let report = parse_order_msg(
            msg,
            account_id,
            &instruments,
            &fee_cache,
            &filled_qty_cache,
            UnixNanos::default(),
        )
        .unwrap();

        match report {
            ExecutionReport::Fill(fill_report) => {
                assert_eq!(fill_report.order_side, OrderSide::Sell);
                assert_eq!(fill_report.last_qty, Quantity::from("0.25000000"));
                assert_eq!(fill_report.last_px, Price::from("40000.00"));
            }
            other => panic!("Expected Fill report for unknown-ord-type order, was {other:?}"),
        }
    }

    #[rstest]
    fn test_deserialize_liquidation_warning_message() {
        let json_data = load_test_json("ws_liquidation_warning.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXLiquidationWarningMsg> =
            serde_json::from_value(payload["data"].clone()).unwrap();

        assert_eq!(data.len(), 1);
        let warning = &data[0];
        assert_eq!(warning.inst_id, Ustr::from("BTC-USDT-SWAP"));
        assert_eq!(warning.inst_type, OKXInstrumentType::Swap);
        assert_eq!(warning.mgn_mode, OKXMarginMode::Cross);
        assert_eq!(warning.pos_side, OKXPositionSide::Long);
        assert_eq!(warning.pos, "0.5");
        assert_eq!(warning.mgn_ratio, "0.62");
        assert_eq!(warning.mark_px, "41250.5");
        assert_eq!(warning.c_time, 1_622_559_930_237);
        assert_eq!(warning.u_time, 1_788_000_000_001);
        assert_eq!(warning.p_time.as_deref(), Some("1788000000002"));
    }

    #[rstest]
    fn test_websocket_instrument_update_preserves_cached_fees() {
        use nautilus_model::{identifiers::InstrumentId, instruments::InstrumentAny};

        use crate::common::{models::OKXInstrument, parse::parse_instrument_any};

        let ts_init = UnixNanos::default();

        // Create initial instrument with fees (simulating HTTP load)
        // These values are already in Nautilus format (HTTP client negates OKX values)
        let initial_fees = (
            Some(Decimal::new(8, 4)),  // Nautilus: 0.0008 (commission)
            Some(Decimal::new(10, 4)), // Nautilus: 0.0010 (commission)
        );

        // Deserialize initial instrument from JSON
        let initial_inst_json = serde_json::json!({
            "instType": "SPOT",
            "instId": "BTC-USD",
            "baseCcy": "BTC",
            "quoteCcy": "USD",
            "settleCcy": "",
            "ctVal": "",
            "ctMult": "",
            "ctValCcy": "",
            "optType": "",
            "stk": "",
            "listTime": "1733454000000",
            "expTime": "",
            "lever": "",
            "tickSz": "0.1",
            "lotSz": "0.00000001",
            "minSz": "0.00001",
            "ctType": "linear",
            "alias": "",
            "state": "live",
            "maxLmtSz": "9999999999",
            "maxMktSz": "1000000",
            "maxTwapSz": "9999999999.0000000000000000",
            "maxIcebergSz": "9999999999.0000000000000000",
            "maxTriggerSz": "9999999999.0000000000000000",
            "maxStopSz": "1000000",
            "uly": "",
            "instFamily": "",
            "ruleType": "normal",
            "maxLmtAmt": "20000000",
            "maxMktAmt": "1000000"
        });

        let initial_inst: OKXInstrument = serde_json::from_value(initial_inst_json)
            .expect("Failed to deserialize initial instrument");

        // Parse initial instrument with fees
        let parsed_initial = parse_instrument_any(
            &initial_inst,
            None,
            None,
            initial_fees.0,
            initial_fees.1,
            ts_init,
        )
        .expect("Failed to parse initial instrument")
        .expect("Initial instrument should not be None");

        // Verify fees were applied
        if let InstrumentAny::CurrencyPair(ref pair) = parsed_initial {
            assert_eq!(pair.maker_fee, dec!(0.0008));
            assert_eq!(pair.taker_fee, dec!(0.0010));
        } else {
            panic!("Expected CurrencyPair instrument");
        }

        // Build instrument cache with the initial instrument
        let mut instruments_cache = AHashMap::new();
        instruments_cache.insert(Ustr::from("BTC-USD"), parsed_initial);

        // Create WebSocket update message (same structure as initial, simulating a WebSocket update)
        let ws_update = serde_json::json!({
            "instType": "SPOT",
            "instId": "BTC-USD",
            "baseCcy": "BTC",
            "quoteCcy": "USD",
            "settleCcy": "",
            "ctVal": "",
            "ctMult": "",
            "ctValCcy": "",
            "optType": "",
            "stk": "",
            "listTime": "1733454000000",
            "expTime": "",
            "lever": "",
            "tickSz": "0.1",
            "lotSz": "0.00000001",
            "minSz": "0.00001",
            "ctType": "linear",
            "alias": "",
            "state": "live",
            "maxLmtSz": "9999999999",
            "maxMktSz": "1000000",
            "maxTwapSz": "9999999999.0000000000000000",
            "maxIcebergSz": "9999999999.0000000000000000",
            "maxTriggerSz": "9999999999.0000000000000000",
            "maxStopSz": "1000000",
            "uly": "",
            "instFamily": "",
            "ruleType": "normal",
            "maxLmtAmt": "20000000",
            "maxMktAmt": "1000000"
        });

        let instrument_id = InstrumentId::from("BTC-USD.OKX");
        let mut funding_cache = AHashMap::new();

        // Parse WebSocket update with cache
        let result = parse_ws_message_data(
            &OKXWsChannel::Instruments,
            ws_update,
            &instrument_id,
            2,
            8,
            ts_init,
            &mut funding_cache,
            &instruments_cache,
        )
        .expect("Failed to parse WebSocket instrument update");

        // Verify the update preserves the cached fees
        if let Some(NautilusWsMessage::Instrument(boxed_inst, _status)) = result {
            if let InstrumentAny::CurrencyPair(pair) = *boxed_inst {
                assert_eq!(
                    pair.maker_fee,
                    Decimal::new(8, 4),
                    "Maker fee should be preserved from cache"
                );
                assert_eq!(
                    pair.taker_fee,
                    Decimal::new(10, 4),
                    "Taker fee should be preserved from cache"
                );
            } else {
                panic!("Expected CurrencyPair instrument from WebSocket update");
            }
        } else {
            panic!("Expected Instrument message from WebSocket update");
        }
    }

    #[rstest]
    #[case::fok_order(OKXOrderType::Fok, TimeInForce::Fok)]
    #[case::ioc_order(OKXOrderType::Ioc, TimeInForce::Ioc)]
    #[case::optimal_limit_ioc_order(OKXOrderType::OptimalLimitIoc, TimeInForce::Ioc)]
    #[case::market_order(OKXOrderType::Market, TimeInForce::Gtc)]
    #[case::limit_order(OKXOrderType::Limit, TimeInForce::Gtc)]
    fn test_parse_time_in_force_from_ord_type(
        #[case] okx_ord_type: OKXOrderType,
        #[case] expected_tif: TimeInForce,
    ) {
        let time_in_force = match okx_ord_type {
            OKXOrderType::Fok | OKXOrderType::OpFok => TimeInForce::Fok,
            OKXOrderType::Ioc | OKXOrderType::OptimalLimitIoc => TimeInForce::Ioc,
            _ => TimeInForce::Gtc,
        };

        assert_eq!(
            time_in_force, expected_tif,
            "OKXOrderType::{okx_ord_type:?} should parse to TimeInForce::{expected_tif:?}"
        );
    }

    #[rstest]
    fn test_deserialize_fok_order_message() {
        let json_data = load_test_json("ws_orders_fok.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].ord_type, OKXOrderType::Fok);
        assert_eq!(data[0].cl_ord_id, "FOK-TEST-001");
        assert_eq!(data[0].inst_id, Ustr::from("BTC-USDT"));
    }

    #[rstest]
    fn test_deserialize_ioc_order_message() {
        let json_data = load_test_json("ws_orders_ioc.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].ord_type, OKXOrderType::Ioc);
        assert_eq!(data[0].cl_ord_id, "IOC-TEST-001");
        assert_eq!(data[0].inst_id, Ustr::from("BTC-USDT"));
    }

    #[rstest]
    fn test_deserialize_optimal_limit_ioc_order_message() {
        let json_data = load_test_json("ws_orders_optimal_limit_ioc.json");
        let ws_msg: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(ws_msg["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].ord_type, OKXOrderType::OptimalLimitIoc);
        assert_eq!(data[0].cl_ord_id, "OPTIMAL-IOC-TEST-001");
        assert_eq!(data[0].inst_id, Ustr::from("BTC-USDT-SWAP"));
    }

    #[rstest]
    fn test_deserialize_regular_order_message() {
        let json_data = load_test_json("ws_orders.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].inst_id, Ustr::from("BTC-USDT-SWAP"));
        assert_eq!(data[0].state, OKXOrderStatus::Filled);
        assert_eq!(data[0].category, OKXOrderCategory::Normal);
        assert_eq!(data[0].rebate.as_deref(), Some("0"));
        assert_eq!(data[0].rebate_ccy.as_deref(), Some("USDT"));
        assert_eq!(data[0].stp_mode, OKXSelfTradePreventionMode::CancelMaker);
        assert!(data[0].linked_algo_ord.is_some());
        assert_eq!(data[0].tag.as_deref(), Some(""));
        assert_eq!(data[0].source.as_deref(), Some(""));
    }

    #[rstest]
    fn test_deserialize_algo_order_message() {
        let json_data = load_test_json("ws_orders_algo.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXAlgoOrderMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].inst_id, Ustr::from("BTC-USDT-SWAP"));
    }

    #[rstest]
    fn test_deserialize_algo_order_missing_trigger_px_type() {
        // algo-advance channel messages omit triggerPxType
        let json = r#"{
            "algoId": "123",
            "algoClOrdId": "cl_1",
            "clOrdId": "",
            "ordId": "",
            "instId": "BTC-USDT-SWAP",
            "instType": "SWAP",
            "ordType": "move_order_stop",
            "state": "live",
            "side": "sell",
            "posSide": "long",
            "sz": "0.01",
            "triggerPx": "95000",
            "ordPx": "-1",
            "tdMode": "cross",
            "lever": "",
            "reduceOnly": "false",
            "actualPx": "",
            "actualSz": "",
            "notionalUsd": "",
            "cTime": "1706000000000",
            "uTime": "1706000001000",
            "triggerTime": "",
            "tag": "",
            "callbackRatio": "0.01",
            "callbackSpread": "",
            "activePx": ""
        }"#;

        let msg: OKXAlgoOrderMsg = serde_json::from_str(json).unwrap();

        assert_eq!(msg.trigger_px_type, OKXTriggerType::None);
        assert_eq!(msg.ord_type, OKXAlgoOrderType::MoveOrderStop);
    }

    #[rstest]
    fn test_deserialize_liquidation_order_message() {
        let json_data = load_test_json("ws_orders_liquidation.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].category, OKXOrderCategory::FullLiquidation);
    }

    #[rstest]
    fn test_deserialize_adl_order_message() {
        let json_data = load_test_json("ws_orders_adl.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].category, OKXOrderCategory::Adl);
    }

    #[rstest]
    fn test_deserialize_trigger_order_message() {
        let json_data = load_test_json("ws_orders_trigger.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXOrderMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].ord_type, OKXOrderType::Trigger);
        assert_eq!(data[0].category, OKXOrderCategory::Normal);
    }

    #[rstest]
    fn test_deserialize_book_snapshot_message() {
        let json_data = load_test_json("ws_books_snapshot.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let action: Option<OKXBookAction> =
            serde_json::from_value(payload["action"].clone()).unwrap();
        let data: Vec<OKXBookMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(action, Some(OKXBookAction::Snapshot));
        assert!(!data[0].asks.is_empty());
        assert!(!data[0].bids.is_empty());
    }

    #[rstest]
    fn test_deserialize_book_update_message() {
        let json_data = load_test_json("ws_books_update.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let action: Option<OKXBookAction> =
            serde_json::from_value(payload["action"].clone()).unwrap();
        let data: Vec<OKXBookMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(action, Some(OKXBookAction::Update));
        assert!(!data[0].asks.is_empty());
        assert!(!data[0].bids.is_empty());
    }

    #[rstest]
    fn test_deserialize_ticker_message() {
        let json_data = load_test_json("ws_tickers.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXTickerMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].inst_id, Ustr::from("BTC-USDT"));
        assert_eq!(data[0].last_px, "9999.99");
    }

    #[rstest]
    fn test_deserialize_candle_message() {
        let json_data = load_test_json("ws_candle.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXCandleMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert!(!data[0].o.is_empty());
        assert!(!data[0].h.is_empty());
        assert!(!data[0].l.is_empty());
        assert!(!data[0].c.is_empty());
    }

    #[rstest]
    fn test_deserialize_funding_rate_message() {
        let json_data = load_test_json("ws_funding_rate.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXFundingRateMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].inst_id, Ustr::from("BTC-USDT-SWAP"));
    }

    #[rstest]
    fn test_deserialize_bbo_tbt_message() {
        let json_data = load_test_json("ws_bbo_tbt.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXBookMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert!(!data[0].asks.is_empty());
        assert!(!data[0].bids.is_empty());
    }

    #[rstest]
    fn test_deserialize_trade_message() {
        let json_data = load_test_json("ws_trades.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXTradeMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert!(!data.is_empty());
        assert_eq!(data[0].inst_id, Ustr::from("BTC-USD"));
    }

    fn create_order_msg_for_event_test(
        state: OKXOrderStatus,
        cl_ord_id: &str,
        ord_id: &str,
        px: &str,
        sz: &str,
    ) -> OKXOrderMsg {
        OKXOrderMsg {
            acc_fill_sz: Some("0".to_string()),
            algo_id: None,
            avg_px: "50000.0".to_string(),
            c_time: 1_746_947_317_401,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: cl_ord_id.to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: Some("0".to_string()),
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: String::new(),
            fill_sz: String::new(),
            fill_time: 0,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: "2.0".to_string(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from(ord_id),
            ord_type: OKXOrderType::Limit,
            pnl: "0".to_string(),
            pos_side: OKXPositionSide::Long,
            px: px.to_string(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Taker,
            sz: sz.to_string(),
            tag: None,
            td_mode: OKXTradeMode::Isolated,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: String::new(),
            u_time: 1_746_947_317_402,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        }
    }

    fn create_spread_order_msg_for_event_test(
        state: OKXOrderStatus,
        cl_ord_id: &str,
        ord_id: &str,
        px: &str,
        sz: &str,
    ) -> OKXSpreadOrder {
        OKXSpreadOrder {
            sprd_id: Ustr::from("BTC-USDT_BTC-USDT-SWAP"),
            ord_id: Ustr::from(ord_id),
            cl_ord_id: Ustr::from(cl_ord_id),
            tag: String::new(),
            side: OKXSide::Buy,
            ord_type: OKXOrderType::Limit,
            sz: sz.to_string(),
            px: px.to_string(),
            avg_px: String::new(),
            state,
            acc_fill_sz: "0".to_string(),
            pending_fill_sz: "0".to_string(),
            pending_settle_sz: "0".to_string(),
            canceled_sz: "0".to_string(),
            fill_sz: String::new(),
            fill_px: String::new(),
            trade_id: Ustr::default(),
            cancel_source: String::new(),
            req_id: String::new(),
            amend_result: String::new(),
            code: String::new(),
            msg: String::new(),
            c_time: Some(1_746_947_317_401),
            u_time: Some(1_746_947_317_402),
        }
    }

    #[rstest]
    fn test_deserialize_spread_orders_message() {
        let json_data = load_test_json("ws_sprd_orders.json");
        let frame: OKXWsFrame = serde_json::from_str(&json_data).unwrap();

        match frame {
            OKXWsFrame::Data { arg, data } => {
                let orders: Vec<OKXSpreadOrder> = serde_json::from_value(data).unwrap();

                assert_eq!(arg.channel, OKXWsChannel::SprdOrders);
                assert_eq!(orders.len(), 1);
                assert_eq!(orders[0].sprd_id, Ustr::from("BCH-USDT_BCH-USDT-SWAP"));
                assert_eq!(orders[0].ord_id, Ustr::from("3386544889978159104"));
                assert_eq!(orders[0].state, OKXOrderStatus::Live);
            }
            other => panic!("Expected Data, was {other:?}"),
        }
    }

    #[rstest]
    fn test_synthesize_trade_id_is_deterministic_and_under_36_chars() {
        let mut msg = create_order_msg_for_event_test(
            OKXOrderStatus::Filled,
            "client-1",
            "venue-1",
            "50000.0",
            "0.001",
        );
        msg.fill_px = "50000.0".to_string();
        msg.fill_sz = "0.001".to_string();
        msg.fill_time = 1_746_947_317_500;
        msg.acc_fill_sz = Some("0.001".to_string());

        let id1 = synthesize_trade_id(&msg);
        let id2 = synthesize_trade_id(&msg);

        assert_eq!(id1, id2, "synthesized id must be deterministic");
        assert!(
            id1.len() <= 36,
            "synthesized id must fit in TradeId, was {}",
            id1.len()
        );
        assert!(id1.starts_with("synth-"));
    }

    #[rstest]
    fn test_synthesize_trade_id_changes_with_fill_fields() {
        let mut msg = create_order_msg_for_event_test(
            OKXOrderStatus::Filled,
            "client-1",
            "venue-1",
            "50000.0",
            "0.001",
        );
        msg.fill_px = "50000.0".to_string();
        msg.fill_sz = "0.001".to_string();
        msg.fill_time = 1_746_947_317_500;
        msg.acc_fill_sz = Some("0.001".to_string());

        let baseline = synthesize_trade_id(&msg);

        msg.fill_sz = "0.002".to_string();
        let different_size = synthesize_trade_id(&msg);
        assert_ne!(baseline, different_size);

        msg.fill_sz = "0.001".to_string();
        msg.fill_time = 1_746_947_317_999;
        let different_time = synthesize_trade_id(&msg);
        assert_ne!(baseline, different_time);
    }

    #[rstest]
    fn test_empty_trade_id_fill_deduped_across_replays() {
        use crate::websocket::dispatch::WsDispatchState;

        // Two identical fill messages with no venue trade_id - the dedup in
        // `WsDispatchState::check_and_insert_trade` must suppress the replay.
        // Regression lock for the empty-trade_id UUID fabrication bug: if
        // `synthesize_trade_id` drifts back to a non-deterministic id, the
        // second `check_and_insert_trade` would return false (not a dupe)
        // and this test fails.
        let mut msg = create_order_msg_for_event_test(
            OKXOrderStatus::Filled,
            "client-1",
            "venue-1",
            "50000.0",
            "0.001",
        );
        msg.trade_id = String::new();
        msg.fill_px = "50000.0".to_string();
        msg.fill_sz = "0.001".to_string();
        msg.fill_time = 1_746_947_317_500;
        msg.acc_fill_sz = Some("0.001".to_string());

        let first_id = TradeId::new(synthesize_trade_id(&msg));
        let second_id = TradeId::new(synthesize_trade_id(&msg));
        assert_eq!(first_id, second_id, "synthesized id must survive replay");

        let state = WsDispatchState::default();
        assert!(
            !state.check_and_insert_trade(first_id),
            "first insert is not a duplicate"
        );
        assert!(
            state.check_and_insert_trade(second_id),
            "replayed fill with empty trade_id must dedup"
        );
    }

    #[rstest]
    fn test_parse_order_event_live_returns_accepted() {
        let instrument = create_stub_instrument();
        let msg = create_order_msg_for_event_test(
            OKXOrderStatus::Live,
            "test_client_123",
            "venue_456",
            "50000.0",
            "0.01",
        );

        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let result = parse_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            None,
            ts_init,
        );

        assert!(result.is_ok());
        match result.unwrap() {
            ParsedOrderEvent::Accepted(accepted) => {
                assert_eq!(accepted.client_order_id, client_order_id);
                assert_eq!(accepted.venue_order_id, VenueOrderId::new("venue_456"));
                assert_eq!(accepted.trader_id, trader_id);
                assert_eq!(accepted.strategy_id, strategy_id);
            }
            other => panic!("Expected Accepted, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_spread_order_event_live_returns_accepted() {
        let instrument = create_stub_instrument();
        let msg = create_spread_order_msg_for_event_test(
            OKXOrderStatus::Live,
            "test_client_123",
            "venue_456",
            "1.0",
            "0.01",
        );
        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let result = parse_spread_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            ts_init,
        );

        match result.unwrap() {
            ParsedOrderEvent::Accepted(accepted) => {
                assert_eq!(accepted.client_order_id, client_order_id);
                assert_eq!(accepted.venue_order_id, VenueOrderId::new("venue_456"));
                assert_eq!(accepted.account_id, account_id);
            }
            other => panic!("Expected Accepted, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_spread_order_event_canceled_returns_canceled() {
        let instrument = create_stub_instrument();
        let msg = create_spread_order_msg_for_event_test(
            OKXOrderStatus::Canceled,
            "test_client_123",
            "venue_456",
            "1.0",
            "0.01",
        );
        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let result = parse_spread_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            ts_init,
        );

        match result.unwrap() {
            ParsedOrderEvent::Canceled(canceled) => {
                assert_eq!(canceled.client_order_id, client_order_id);
                assert_eq!(
                    canceled.venue_order_id,
                    Some(VenueOrderId::new("venue_456"))
                );
                assert_eq!(canceled.account_id, Some(account_id));
            }
            other => panic!("Expected Canceled, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_spread_order_event_filled_returns_fill() {
        let instrument = create_stub_instrument();
        let mut msg = create_spread_order_msg_for_event_test(
            OKXOrderStatus::Filled,
            "test_client_123",
            "venue_456",
            "1.0",
            "0.01",
        );
        msg.fill_sz = "0.01".to_string();
        msg.fill_px = "1.0".to_string();
        msg.trade_id = Ustr::from("trade_789");
        msg.acc_fill_sz = "0.01".to_string();

        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let result = parse_spread_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            ts_init,
        );

        let error = result.unwrap_err();
        assert!(error.to_string().contains("missing fee"));
    }

    #[rstest]
    fn test_parse_spread_order_fill_report_uses_incremental_acc_fill_sz() {
        let instrument = create_stub_instrument();
        let mut msg = create_spread_order_msg_for_event_test(
            OKXOrderStatus::PartiallyFilled,
            "test_client_123",
            "venue_456",
            "1.0",
            "0.03",
        );
        msg.acc_fill_sz = "0.03".to_string();
        msg.fill_sz = String::new();
        msg.fill_px = String::new();

        let error = parse_spread_order_fill_report(
            &msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            AccountId::new("OKX-001"),
            Some(Quantity::from("0.01000000")),
            UnixNanos::from(1_000_000_000),
        )
        .unwrap_err();

        assert!(error.to_string().contains("missing fee"));
        assert!(error.to_string().contains("sprd-orders updates omit fee"));
    }

    #[rstest]
    fn test_parse_spread_order_fill_report_skips_duplicate_acc_fill_sz() {
        let instrument = create_stub_instrument();
        let mut msg = create_spread_order_msg_for_event_test(
            OKXOrderStatus::PartiallyFilled,
            "test_client_123",
            "venue_456",
            "1.0",
            "0.01",
        );
        msg.acc_fill_sz = "0.01".to_string();
        msg.fill_sz = String::new();
        msg.fill_px = String::new();

        let result = parse_spread_order_fill_report(
            &msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            AccountId::new("OKX-001"),
            Some(Quantity::from("0.01000000")),
            UnixNanos::from(1_000_000_000),
        )
        .unwrap();

        assert!(result.is_none());
    }

    #[rstest]
    fn test_parse_spread_order_fill_report_rejects_regressed_acc_fill_sz() {
        let instrument = create_stub_instrument();
        let mut msg = create_spread_order_msg_for_event_test(
            OKXOrderStatus::PartiallyFilled,
            "test_client_123",
            "venue_456",
            "1.0",
            "0.01",
        );
        msg.acc_fill_sz = "0.01".to_string();
        msg.fill_sz = String::new();
        msg.fill_px = String::new();

        let error = parse_spread_order_fill_report(
            &msg,
            &InstrumentAny::CryptoPerpetual(instrument),
            AccountId::new("OKX-001"),
            Some(Quantity::from("0.03000000")),
            UnixNanos::from(1_000_000_000),
        )
        .unwrap_err();

        assert!(
            error
                .to_string()
                .contains("Cumulative spread fill went backwards")
        );
    }

    #[rstest]
    fn test_parse_order_event_live_with_price_change_returns_updated() {
        let instrument = create_stub_instrument();
        let msg = create_order_msg_for_event_test(
            OKXOrderStatus::Live,
            "test_client_123",
            "venue_456",
            "51000.0",
            "0.01",
        );

        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let previous_state = OrderStateSnapshot {
            venue_order_id: VenueOrderId::new("venue_456"),
            quantity: Quantity::from("0.01000000"),
            price: Some(Price::from("50000.00")),
        };

        let result = parse_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            Some(&previous_state),
            ts_init,
        );

        assert!(result.is_ok());
        match result.unwrap() {
            ParsedOrderEvent::Updated(updated) => {
                assert_eq!(updated.client_order_id, client_order_id);
                assert_eq!(updated.price, Some(Price::from("51000.00")));
            }
            other => panic!("Expected Updated, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_order_event_live_with_quantity_change_returns_updated() {
        let instrument = create_stub_instrument();
        let msg = create_order_msg_for_event_test(
            OKXOrderStatus::Live,
            "test_client_123",
            "venue_456",
            "50000.0",
            "0.02",
        );

        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let previous_state = OrderStateSnapshot {
            venue_order_id: VenueOrderId::new("venue_456"),
            quantity: Quantity::from("0.01000000"),
            price: Some(Price::from("50000.00")),
        };

        let result = parse_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            Some(&previous_state),
            ts_init,
        );

        assert!(result.is_ok());
        match result.unwrap() {
            ParsedOrderEvent::Updated(updated) => {
                assert_eq!(updated.client_order_id, client_order_id);
                assert_eq!(updated.quantity, Quantity::from("0.02000000"));
            }
            other => panic!("Expected Updated, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_order_event_canceled_returns_canceled() {
        let instrument = create_stub_instrument();
        let msg = create_order_msg_for_event_test(
            OKXOrderStatus::Canceled,
            "test_client_123",
            "venue_456",
            "50000.0",
            "0.01",
        );

        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let result = parse_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            None,
            ts_init,
        );

        assert!(result.is_ok());
        match result.unwrap() {
            ParsedOrderEvent::Canceled(canceled) => {
                assert_eq!(canceled.client_order_id, client_order_id);
                assert_eq!(
                    canceled.venue_order_id,
                    Some(VenueOrderId::new("venue_456"))
                );
            }
            other => panic!("Expected Canceled, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_order_event_canceled_with_expiry_reason_returns_expired() {
        let instrument = create_stub_instrument();
        let mut msg = create_order_msg_for_event_test(
            OKXOrderStatus::Canceled,
            "test_client_123",
            "venue_456",
            "50000.0",
            "0.01",
        );
        msg.cancel_source_reason = Some("GTD order expired".to_string());

        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let result = parse_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            None,
            ts_init,
        );

        assert!(result.is_ok());
        match result.unwrap() {
            ParsedOrderEvent::Expired(expired) => {
                assert_eq!(expired.client_order_id, client_order_id);
                assert_eq!(expired.venue_order_id, Some(VenueOrderId::new("venue_456")));
            }
            other => panic!("Expected Expired, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_order_event_filled_with_fill_data_returns_fill() {
        let instrument = create_stub_instrument();
        let mut msg = create_order_msg_for_event_test(
            OKXOrderStatus::Filled,
            "test_client_123",
            "venue_456",
            "50000.0",
            "0.01",
        );
        msg.fill_sz = "0.01".to_string();
        msg.fill_px = "50000.0".to_string();
        msg.trade_id = "trade_789".to_string();
        msg.acc_fill_sz = Some("0.01".to_string());

        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let result = parse_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            None,
            ts_init,
        );

        assert!(result.is_ok());
        match result.unwrap() {
            ParsedOrderEvent::Fill(fill) => {
                assert_eq!(fill.client_order_id, Some(client_order_id));
                assert_eq!(fill.venue_order_id, VenueOrderId::new("venue_456"));
                assert_eq!(fill.trade_id, TradeId::from("trade_789"));
            }
            other => panic!("Expected Fill, was {other:?}"),
        }
    }

    #[rstest]
    fn test_is_order_expired_by_reason_gtd_in_reason() {
        let mut msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        msg.cancel_source_reason = Some("GTD order expired".to_string());
        assert!(is_order_expired_by_reason(&msg));
    }

    #[rstest]
    fn test_is_order_expired_by_reason_timeout_in_reason() {
        let mut msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        msg.cancel_source_reason = Some("Order timeout".to_string());
        assert!(is_order_expired_by_reason(&msg));
    }

    #[rstest]
    fn test_is_order_expired_by_reason_expir_in_reason() {
        let mut msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        msg.cancel_source_reason = Some("Expiration reached".to_string());
        assert!(is_order_expired_by_reason(&msg));
    }

    #[rstest]
    fn test_is_order_expired_by_reason_source_code_5() {
        let mut msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        msg.cancel_source = Some("5".to_string());
        assert!(is_order_expired_by_reason(&msg));
    }

    #[rstest]
    fn test_is_order_expired_by_reason_source_time_expired() {
        let mut msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        msg.cancel_source = Some("time_expired".to_string());
        assert!(is_order_expired_by_reason(&msg));
    }

    #[rstest]
    fn test_is_order_expired_by_reason_false_for_user_cancel() {
        let mut msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        msg.cancel_source_reason = Some("User canceled".to_string());
        msg.cancel_source = Some("1".to_string());
        assert!(!is_order_expired_by_reason(&msg));
    }

    #[rstest]
    fn test_is_order_expired_by_reason_false_when_no_reason() {
        let msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        assert!(!is_order_expired_by_reason(&msg));
    }

    fn fresh_cancel_source_seen() -> Mutex<AHashSet<String>> {
        Mutex::new(AHashSet::new())
    }

    #[rstest]
    fn test_log_unknown_cancel_source_records_first_observation() {
        let mut msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        msg.cancel_source = Some("99".to_string());
        msg.cancel_source_reason = Some("Unknown reason".to_string());

        let seen = fresh_cancel_source_seen();
        assert!(log_unknown_cancel_source_inner(&msg, &seen, 8));
        assert_eq!(seen.lock().len(), 1);
    }

    #[rstest]
    fn test_log_unknown_cancel_source_dedups_repeat_pair() {
        let mut msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        msg.cancel_source = Some("99".to_string());
        msg.cancel_source_reason = Some("Unknown reason".to_string());

        let seen = fresh_cancel_source_seen();
        assert!(log_unknown_cancel_source_inner(&msg, &seen, 8));
        assert!(!log_unknown_cancel_source_inner(&msg, &seen, 8));
        assert_eq!(seen.lock().len(), 1);
    }

    #[rstest]
    #[case::post_only("31")]
    #[case::known_expired("5")]
    #[case::sentinel_time("time_expired")]
    #[case::sentinel_gtd("gtd_expired")]
    fn test_log_unknown_cancel_source_skips_known_sources(#[case] source: &str) {
        let mut msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        msg.cancel_source = Some(source.to_string());

        let seen = fresh_cancel_source_seen();
        assert!(!log_unknown_cancel_source_inner(&msg, &seen, 8));
        assert!(seen.lock().is_empty());
    }

    #[rstest]
    fn test_log_unknown_cancel_source_skips_when_source_and_reason_empty() {
        let msg =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        let seen = fresh_cancel_source_seen();
        assert!(!log_unknown_cancel_source_inner(&msg, &seen, 8));
        assert!(seen.lock().is_empty());
    }

    #[rstest]
    fn test_log_unknown_cancel_source_respects_capacity_cap() {
        let cap = 4;
        let seen = fresh_cancel_source_seen();

        for i in 0..cap {
            let mut msg = create_order_msg_for_event_test(
                OKXOrderStatus::Canceled,
                "test",
                "123",
                "100",
                "1",
            );
            msg.cancel_source = Some(format!("novel_{i}"));
            assert!(log_unknown_cancel_source_inner(&msg, &seen, cap));
        }

        let mut overflow =
            create_order_msg_for_event_test(OKXOrderStatus::Canceled, "test", "123", "100", "1");
        overflow.cancel_source = Some("novel_overflow".to_string());
        assert!(!log_unknown_cancel_source_inner(&overflow, &seen, cap));
        assert_eq!(seen.lock().len(), cap);
    }

    // Regression test: PartiallyFilled order with price change should emit Updated, not StatusOnly
    #[rstest]
    fn test_parse_order_event_partially_filled_with_price_change_returns_updated() {
        let instrument = create_stub_instrument();
        let msg = create_order_msg_for_event_test(
            OKXOrderStatus::PartiallyFilled,
            "test_client_123",
            "venue_456",
            "51000.0",
            "0.01",
        );

        let client_order_id = ClientOrderId::new("test_client_123");
        let account_id = AccountId::new("OKX-001");
        let trader_id = TraderId::new("TRADER-001");
        let strategy_id = StrategyId::new("STRATEGY-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        let previous_state = OrderStateSnapshot {
            venue_order_id: VenueOrderId::new("venue_456"),
            quantity: Quantity::from("0.01000000"),
            price: Some(Price::from("50000.00")),
        };

        let result = parse_order_event(
            &msg,
            client_order_id,
            account_id,
            trader_id,
            strategy_id,
            &InstrumentAny::CryptoPerpetual(instrument),
            None,
            None,
            Some(&previous_state),
            ts_init,
        );

        assert!(result.is_ok());
        match result.unwrap() {
            ParsedOrderEvent::Updated(updated) => {
                assert_eq!(updated.client_order_id, client_order_id);
                assert_eq!(updated.price, Some(Price::from("51000.00")));
            }
            other => {
                panic!("Expected Updated for PartiallyFilled with price change, was {other:?}")
            }
        }
    }

    #[rstest]
    fn test_is_order_updated_price_change() {
        let instrument = create_stub_instrument();
        let msg = create_order_msg_for_event_test(
            OKXOrderStatus::Live,
            "test",
            "venue_123",
            "51000.0",
            "0.01",
        );

        let previous = OrderStateSnapshot {
            venue_order_id: VenueOrderId::new("venue_123"),
            quantity: Quantity::from("0.01000000"),
            price: Some(Price::from("50000.00")),
        };

        let result = is_order_updated(&msg, &previous, &InstrumentAny::CryptoPerpetual(instrument));
        assert!(result.is_ok());
        assert!(result.unwrap());
    }

    #[rstest]
    fn test_is_order_updated_quantity_change() {
        let instrument = create_stub_instrument();
        let msg = create_order_msg_for_event_test(
            OKXOrderStatus::Live,
            "test",
            "venue_123",
            "50000.0",
            "0.02", // New quantity
        );

        let previous = OrderStateSnapshot {
            venue_order_id: VenueOrderId::new("venue_123"),
            quantity: Quantity::from("0.01000000"), // Old quantity
            price: Some(Price::from("50000.00")),
        };

        let result = is_order_updated(&msg, &previous, &InstrumentAny::CryptoPerpetual(instrument));
        assert!(result.is_ok());
        assert!(result.unwrap());
    }

    #[rstest]
    fn test_is_order_updated_venue_id_change() {
        let instrument = create_stub_instrument();
        let msg = create_order_msg_for_event_test(
            OKXOrderStatus::Live,
            "test",
            "venue_456", // New venue ID
            "50000.0",
            "0.01",
        );

        let previous = OrderStateSnapshot {
            venue_order_id: VenueOrderId::new("venue_123"), // Old venue ID
            quantity: Quantity::from("0.01000000"),
            price: Some(Price::from("50000.00")),
        };

        let result = is_order_updated(&msg, &previous, &InstrumentAny::CryptoPerpetual(instrument));
        assert!(result.is_ok());
        assert!(result.unwrap());
    }

    #[rstest]
    fn test_is_order_updated_no_change() {
        let instrument = create_stub_instrument();
        let msg = create_order_msg_for_event_test(
            OKXOrderStatus::Live,
            "test",
            "venue_123",
            "50000.0",
            "0.01",
        );

        let previous = OrderStateSnapshot {
            venue_order_id: VenueOrderId::new("venue_123"),
            quantity: Quantity::from("0.01000000"),
            price: Some(Price::from("50000.00")),
        };

        let result = is_order_updated(&msg, &previous, &InstrumentAny::CryptoPerpetual(instrument));
        assert!(result.is_ok());
        assert!(!result.unwrap());
    }

    #[rstest]
    fn test_parse_order_status_report_ts_last_and_ts_init_ordering() {
        let instrument = create_stub_instrument();
        let inst = InstrumentAny::CryptoPerpetual(instrument);
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::from(999_000_000_000u64);

        let msg = OKXOrderMsg {
            acc_fill_sz: Some("0".to_string()),
            algo_id: None,
            avg_px: String::new(),
            c_time: 1_706_000_000_000, // ~2024-01-23 in ms
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "test_ts_order".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: None,
            attach_algo_ords: Vec::new(),
            outcome: None,
            fee: None,
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: String::new(),
            fill_sz: String::new(),
            fill_time: 0,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: String::new(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("123456"),
            ord_type: OKXOrderType::Limit,
            pnl: String::new(),
            pos_side: OKXPositionSide::Long,
            px: "50000.00".to_string(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::Live,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Taker,
            sz: "0.01".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Cross,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: String::new(),
            u_time: 1_706_000_001_000, // 1 second later in ms
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let report = parse_order_status_report(&msg, &inst, account_id, ts_init).unwrap();

        assert_eq!(
            report.ts_accepted,
            UnixNanos::from(1_706_000_000_000_u64 * 1_000_000)
        );
        assert_eq!(
            report.ts_last,
            UnixNanos::from(1_706_000_001_000_u64 * 1_000_000)
        );
        assert_eq!(report.ts_init, ts_init);
    }

    #[rstest]
    fn test_parse_order_status_report_preserves_attached_tp_sl_child_ids() {
        let instrument = create_stub_instrument();
        let inst = InstrumentAny::CryptoPerpetual(instrument);
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::default();

        let msg = OKXOrderMsg {
            acc_fill_sz: Some("0".to_string()),
            algo_id: None,
            avg_px: String::new(),
            c_time: 1_706_000_000_000,
            cancel_source: None,
            cancel_source_reason: None,
            category: OKXOrderCategory::Normal,
            ccy: Ustr::from("USDT"),
            cl_ord_id: "O-attached-entry".to_string(),
            algo_cl_ord_id: None,
            attach_algo_cl_ord_id: Some("O-attached-sl".to_string()),
            attach_algo_ords: vec![
                OKXAttachedAlgoOrd {
                    attach_algo_id: "algo-sl".to_string(),
                    attach_algo_cl_ord_id: "O-attached-sl".to_string(),
                    sl_trigger_px: "1500".to_string(),
                    sl_ord_px: "-1".to_string(),
                    sl_trigger_px_type: Some(OKXTriggerType::Last),
                    tp_trigger_px: String::new(),
                    tp_ord_px: String::new(),
                    tp_trigger_px_type: None,
                    callback_ratio: String::new(),
                    callback_spread: String::new(),
                    active_px: String::new(),
                },
                OKXAttachedAlgoOrd {
                    attach_algo_id: "algo-tp".to_string(),
                    attach_algo_cl_ord_id: "O-attached-tp".to_string(),
                    sl_trigger_px: String::new(),
                    sl_ord_px: String::new(),
                    sl_trigger_px_type: None,
                    tp_trigger_px: "2500".to_string(),
                    tp_ord_px: "-1".to_string(),
                    tp_trigger_px_type: Some(OKXTriggerType::Last),
                    callback_ratio: String::new(),
                    callback_spread: String::new(),
                    active_px: String::new(),
                },
            ],
            outcome: None,
            fee: None,
            fee_ccy: Ustr::from("USDT"),
            fill_fee: None,
            fill_fee_ccy: None,
            fill_mark_px: None,
            fill_mark_vol: None,
            fill_px_vol: None,
            fill_px_usd: None,
            fill_fwd_px: None,
            fill_notional_usd: None,
            fill_pnl: None,
            fill_px: String::new(),
            fill_sz: String::new(),
            fill_time: 0,
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            is_tp_limit: None,
            lever: String::new(),
            linked_algo_ord: None,
            notional_usd: None,
            ord_id: Ustr::from("123456"),
            ord_type: OKXOrderType::Limit,
            pnl: String::new(),
            pos_side: OKXPositionSide::Long,
            px: "2000.00".to_string(),
            px_type: OKXPriceType::None,
            px_usd: None,
            px_vol: None,
            quick_mgn_type: OKXQuickMarginType::None,
            rebate: None,
            rebate_ccy: None,
            reduce_only: "false".to_string(),
            side: OKXSide::Buy,
            sl_ord_px: None,
            sl_trigger_px: None,
            sl_trigger_px_type: None,
            source: None,
            state: OKXOrderStatus::Live,
            stp_id: None,
            stp_mode: OKXSelfTradePreventionMode::None,
            exec_type: OKXExecType::Taker,
            sz: "0.01".to_string(),
            tag: None,
            td_mode: OKXTradeMode::Cross,
            tgt_ccy: None,
            tp_ord_px: None,
            tp_trigger_px: None,
            tp_trigger_px_type: None,
            trade_id: String::new(),
            u_time: 1_706_000_001_000,
            amend_result: None,
            req_id: None,
            code: None,
            msg: None,
        };

        let report = parse_order_status_report(&msg, &inst, account_id, ts_init).unwrap();
        let linked_order_ids = report
            .linked_order_ids
            .expect("expected linked child order ids");

        assert_eq!(linked_order_ids.len(), 2);
        assert!(linked_order_ids.contains(&ClientOrderId::from("O-attached-sl")));
        assert!(linked_order_ids.contains(&ClientOrderId::from("O-attached-tp")));
    }

    #[rstest]
    fn test_parse_algo_order_timestamps_converted_from_ms_to_ns() {
        let instrument = create_stub_instrument();
        let inst = InstrumentAny::CryptoPerpetual(instrument);
        let account_id = AccountId::new("OKX-001");
        let ts_init = UnixNanos::from(999_000_000_000u64);

        let msg = OKXAlgoOrderMsg {
            algo_id: "algo_1".to_string(),
            algo_cl_ord_id: "algo_cl_1".to_string(),
            cl_ord_id: String::new(),
            ord_id: String::new(),
            ord_id_list: Vec::new(),
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            ord_type: OKXAlgoOrderType::Trigger,
            state: OKXAlgoOrderStatus::Live,
            side: OKXSide::Buy,
            pos_side: OKXPositionSide::Long,
            sz: "0.01".to_string(),
            trigger_px: "45000.00".to_string(),
            trigger_px_type: OKXTriggerType::Last,
            sl_trigger_px: String::new(),
            sl_ord_px: String::new(),
            sl_trigger_px_type: OKXTriggerType::None,
            tp_trigger_px: String::new(),
            tp_ord_px: String::new(),
            tp_trigger_px_type: OKXTriggerType::None,
            ord_px: "-1".to_string(),
            td_mode: OKXTradeMode::Cross,
            lever: String::new(),
            reduce_only: "false".to_string(),
            close_fraction: String::new(),
            actual_px: String::new(),
            actual_sz: String::new(),
            notional_usd: String::new(),
            c_time: 1_706_000_000_000,
            u_time: 1_706_000_001_000,
            trigger_time: String::new(),
            fail_code: String::new(),
            tag: String::new(),
            callback_ratio: String::new(),
            callback_spread: String::new(),
            active_px: String::new(),
            ccy: None,
            tgt_ccy: None,
            fee: None,
            fee_ccy: None,
            advance_ord_type: None,
        };

        let report = parse_algo_order_status_report(&msg, &inst, account_id, ts_init).unwrap();

        let expected_accepted_ns = 1_706_000_000_000_u64 * 1_000_000;
        let expected_last_ns = 1_706_000_001_000_u64 * 1_000_000;
        assert_eq!(report.ts_accepted, UnixNanos::from(expected_accepted_ns));
        assert_eq!(report.ts_last, UnixNanos::from(expected_last_ns));
        assert_eq!(report.ts_init, ts_init);
    }

    fn stub_algo_order_msg(ord_type: OKXAlgoOrderType) -> OKXAlgoOrderMsg {
        OKXAlgoOrderMsg {
            algo_id: "algo_1".to_string(),
            algo_cl_ord_id: "algo_cl_1".to_string(),
            cl_ord_id: String::new(),
            ord_id: String::new(),
            ord_id_list: Vec::new(),
            inst_id: Ustr::from("BTC-USDT-SWAP"),
            inst_type: OKXInstrumentType::Swap,
            ord_type,
            state: OKXAlgoOrderStatus::Live,
            side: OKXSide::Sell,
            pos_side: OKXPositionSide::Long,
            sz: "0.01".to_string(),
            trigger_px: "95000.00".to_string(),
            trigger_px_type: OKXTriggerType::Last,
            sl_trigger_px: String::new(),
            sl_ord_px: String::new(),
            sl_trigger_px_type: OKXTriggerType::None,
            tp_trigger_px: String::new(),
            tp_ord_px: String::new(),
            tp_trigger_px_type: OKXTriggerType::None,
            ord_px: "-1".to_string(),
            td_mode: OKXTradeMode::Cross,
            lever: String::new(),
            reduce_only: "false".to_string(),
            close_fraction: String::new(),
            actual_px: String::new(),
            actual_sz: String::new(),
            notional_usd: String::new(),
            c_time: 1_706_000_000_000,
            u_time: 1_706_000_001_000,
            trigger_time: String::new(),
            fail_code: String::new(),
            tag: String::new(),
            callback_ratio: String::new(),
            callback_spread: String::new(),
            active_px: String::new(),
            ccy: None,
            tgt_ccy: None,
            fee: None,
            fee_ccy: None,
            advance_ord_type: None,
        }
    }

    #[rstest]
    fn test_parse_algo_order_trailing_stop_with_callback_ratio() {
        let instrument = create_stub_instrument();
        let inst = InstrumentAny::CryptoPerpetual(instrument);
        let account_id = AccountId::new("OKX-001");

        let mut msg = stub_algo_order_msg(OKXAlgoOrderType::MoveOrderStop);
        msg.callback_ratio = "0.01".to_string(); // 1% = 100 basis points

        let report =
            parse_algo_order_status_report(&msg, &inst, account_id, UnixNanos::default()).unwrap();

        assert_eq!(report.order_type, OrderType::TrailingStopMarket);
        assert_eq!(report.trailing_offset, Some(dec!(100)));
        assert_eq!(
            report.trailing_offset_type,
            Some(TrailingOffsetType::BasisPoints),
        );
        assert_eq!(report.trigger_price, Some(Price::from("95000.00")));
    }

    #[rstest]
    fn test_parse_algo_order_trailing_stop_captures_activation_price() {
        let instrument = create_stub_instrument();
        let inst = InstrumentAny::CryptoPerpetual(instrument);
        let account_id = AccountId::new("OKX-001");

        let mut msg = stub_algo_order_msg(OKXAlgoOrderType::MoveOrderStop);
        msg.callback_ratio = "0.01".to_string();
        msg.active_px = "94000.5".to_string();

        let report =
            parse_algo_order_status_report(&msg, &inst, account_id, UnixNanos::default()).unwrap();

        assert_eq!(report.order_type, OrderType::TrailingStopMarket);
        assert_eq!(report.activation_price, Some(Price::from("94000.50")));
    }

    #[rstest]
    fn test_parse_algo_order_trailing_stop_with_callback_spread() {
        let instrument = create_stub_instrument();
        let inst = InstrumentAny::CryptoPerpetual(instrument);
        let account_id = AccountId::new("OKX-001");

        let mut msg = stub_algo_order_msg(OKXAlgoOrderType::MoveOrderStop);
        msg.callback_spread = "50.5".to_string();

        let report =
            parse_algo_order_status_report(&msg, &inst, account_id, UnixNanos::default()).unwrap();

        assert_eq!(report.order_type, OrderType::TrailingStopMarket);
        assert_eq!(report.trailing_offset, Some(dec!(50.5)));
        assert_eq!(report.trailing_offset_type, Some(TrailingOffsetType::Price),);
    }

    #[rstest]
    fn test_parse_algo_order_unsupported_type_skipped() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let msg = stub_algo_order_msg(OKXAlgoOrderType::Iceberg);

        let result = parse_algo_order_msg(&msg, account_id, &instruments, UnixNanos::default());

        assert!(result.unwrap().is_none());
    }

    #[rstest]
    fn test_parse_algo_order_chase_type_skipped() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let msg = stub_algo_order_msg(OKXAlgoOrderType::Chase);

        let result = parse_algo_order_msg(&msg, account_id, &instruments, UnixNanos::default());

        assert!(result.unwrap().is_none());
    }

    #[rstest]
    fn test_parse_algo_order_unknown_type_skipped() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let msg = stub_algo_order_msg(OKXAlgoOrderType::Other);

        let result = parse_algo_order_msg(&msg, account_id, &instruments, UnixNanos::default());

        assert!(result.unwrap().is_none());
    }

    #[rstest]
    fn test_parse_algo_order_unknown_state_skipped() {
        let instrument = create_stub_instrument();
        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(instrument),
        );

        let mut msg = stub_algo_order_msg(OKXAlgoOrderType::Trigger);
        msg.state = OKXAlgoOrderStatus::Unknown;

        let result = parse_algo_order_msg(&msg, account_id, &instruments, UnixNanos::default());

        assert!(result.unwrap().is_none());
    }

    #[rstest]
    fn test_deserialize_algo_order_states_message() {
        let json_data = load_test_json("ws_orders_algo_states.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXAlgoOrderMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        assert_eq!(data.len(), 6);
        assert_eq!(data[0].state, OKXAlgoOrderStatus::Effective);
        assert_eq!(data[1].state, OKXAlgoOrderStatus::PartiallyEffective);
        assert_eq!(data[2].state, OKXAlgoOrderStatus::Pause);
        assert_eq!(data[3].state, OKXAlgoOrderStatus::OrderFailed);
        assert_eq!(data[4].state, OKXAlgoOrderStatus::PartiallyFailed);
        assert_eq!(data[5].ord_type, OKXAlgoOrderType::Chase);
    }

    #[rstest]
    fn test_parse_algo_order_states_map_to_nautilus_status() {
        let json_data = load_test_json("ws_orders_algo_states.json");
        let payload: serde_json::Value = serde_json::from_str(&json_data).unwrap();
        let data: Vec<OKXAlgoOrderMsg> = serde_json::from_value(payload["data"].clone()).unwrap();

        let account_id = AccountId::new("OKX-001");
        let mut instruments = AHashMap::new();
        instruments.insert(
            Ustr::from("BTC-USDT-SWAP"),
            InstrumentAny::CryptoPerpetual(create_stub_instrument()),
        );

        let expected = [
            (0_usize, OrderStatus::Triggered), // effective
            (1, OrderStatus::Triggered),       // partially_effective
            (2, OrderStatus::Accepted),        // pause
            (3, OrderStatus::Rejected),        // order_failed
            (4, OrderStatus::Rejected),        // partially_failed
        ];

        for (idx, expected_status) in expected {
            let report =
                parse_algo_order_msg(&data[idx], account_id, &instruments, UnixNanos::default())
                    .unwrap()
                    .unwrap_or_else(|| panic!("Expected report for fixture index {idx}"));

            match report {
                ExecutionReport::Order(report) => {
                    assert_eq!(report.order_status, expected_status);
                }
                other => panic!("Expected Order report for fixture index {idx}, was {other:?}"),
            }
        }

        // Chase orders are unsupported and skipped; their triggered child orders
        // still arrive on the regular orders channel
        let chase =
            parse_algo_order_msg(&data[5], account_id, &instruments, UnixNanos::default()).unwrap();
        assert!(chase.is_none());
    }

    #[rstest]
    fn test_parse_algo_order_missing_trigger_px_type_defaults() {
        let instrument = create_stub_instrument();
        let inst = InstrumentAny::CryptoPerpetual(instrument);
        let account_id = AccountId::new("OKX-001");

        let mut msg = stub_algo_order_msg(OKXAlgoOrderType::MoveOrderStop);
        msg.trigger_px_type = OKXTriggerType::None;
        msg.callback_ratio = "0.005".to_string();

        let report =
            parse_algo_order_status_report(&msg, &inst, account_id, UnixNanos::default()).unwrap();

        assert_eq!(report.trigger_type, Some(TriggerType::Default));
        assert_eq!(report.order_type, OrderType::TrailingStopMarket);
    }

    #[rstest]
    fn test_parse_algo_order_close_fraction_stop_market_without_sz() {
        let instrument = create_stub_instrument();
        let inst = InstrumentAny::CryptoPerpetual(instrument);
        let account_id = AccountId::new("OKX-001");

        let mut msg = stub_algo_order_msg(OKXAlgoOrderType::Conditional);
        msg.sz = String::new();
        msg.trigger_px = String::new();
        msg.trigger_px_type = OKXTriggerType::None;
        msg.ord_px = String::new();
        msg.sl_trigger_px = "50000".to_string();
        msg.sl_ord_px = "-1".to_string();
        msg.sl_trigger_px_type = OKXTriggerType::Last;
        msg.close_fraction = "1".to_string();
        msg.reduce_only = "true".to_string();

        let report =
            parse_algo_order_status_report(&msg, &inst, account_id, UnixNanos::default()).unwrap();

        assert_eq!(report.order_type, OrderType::StopMarket);
        assert_eq!(report.trigger_price, Some(Price::from("50000.00")));
        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
        assert_eq!(report.price, None);
        assert_eq!(report.quantity, Quantity::zero(inst.size_precision()));
        assert!(report.reduce_only);
    }

    #[rstest]
    fn test_parse_algo_order_close_fraction_market_if_touched_without_sz() {
        let instrument = create_stub_instrument();
        let inst = InstrumentAny::CryptoPerpetual(instrument);
        let account_id = AccountId::new("OKX-001");

        let mut msg = stub_algo_order_msg(OKXAlgoOrderType::Conditional);
        msg.sz = String::new();
        msg.trigger_px = String::new();
        msg.trigger_px_type = OKXTriggerType::None;
        msg.ord_px = String::new();
        msg.sl_trigger_px = String::new();
        msg.sl_ord_px = String::new();
        msg.tp_trigger_px = "50000".to_string();
        msg.tp_ord_px = "-1".to_string();
        msg.tp_trigger_px_type = OKXTriggerType::Last;
        msg.close_fraction = "1".to_string();
        msg.reduce_only = "true".to_string();
        msg.side = OKXSide::Buy;

        let report =
            parse_algo_order_status_report(&msg, &inst, account_id, UnixNanos::default()).unwrap();

        assert_eq!(report.order_type, OrderType::MarketIfTouched);
        assert_eq!(report.trigger_price, Some(Price::from("50000.00")));
        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
        assert_eq!(report.price, None);
        assert_eq!(report.quantity, Quantity::zero(inst.size_precision()));
        assert!(report.reduce_only);
    }

    fn stub_book_entry(price: &str, size: &str) -> OrderBookEntry {
        OrderBookEntry {
            price: price.to_string(),
            size: size.to_string(),
            liquidated_orders_count: "0".to_string(),
            orders_count: "1".to_string(),
        }
    }

    fn stub_book_msg(bids: Vec<OrderBookEntry>, asks: Vec<OrderBookEntry>) -> OKXBookMsg {
        OKXBookMsg {
            bids,
            asks,
            ts: 1_706_000_000_000,
            seq_id: 1,
            prev_seq_id: Some(0),
            checksum: None,
        }
    }

    #[rstest]
    fn test_parse_quote_msg_empty_bids_returns_error() {
        let msg = stub_book_msg(vec![], vec![stub_book_entry("50000.00", "1.0")]);

        let result = parse_quote_msg(
            &msg,
            InstrumentId::from("BTC-USDT.OKX"),
            2,
            8,
            UnixNanos::default(),
        );
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("Empty bids"));
    }

    #[rstest]
    fn test_parse_quote_msg_empty_asks_returns_error() {
        let msg = stub_book_msg(vec![stub_book_entry("50000.00", "1.0")], vec![]);

        let result = parse_quote_msg(
            &msg,
            InstrumentId::from("BTC-USDT.OKX"),
            2,
            8,
            UnixNanos::default(),
        );
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("Empty asks"));
    }

    #[rstest]
    fn test_quote_cache_complete_bbo_tbt_message() {
        use nautilus_common::cache::quote::QuoteCache;

        let mut cache = QuoteCache::new();
        let instrument_id = InstrumentId::from("BTC-USD-260327-75000-C.OKX");
        let msg = stub_book_msg(
            vec![stub_book_entry("0.0035", "100")],
            vec![stub_book_entry("0.0040", "200")],
        );

        let bid_price = Some(parse_price(&msg.bids[0].price, 4).unwrap());
        let bid_size = Some(parse_quantity(&msg.bids[0].size, 0).unwrap());
        let ask_price = Some(parse_price(&msg.asks[0].price, 4).unwrap());
        let ask_size = Some(parse_quantity(&msg.asks[0].size, 0).unwrap());
        let ts_event = parse_millisecond_timestamp(msg.ts);

        let quote = cache
            .process(
                instrument_id,
                bid_price,
                ask_price,
                bid_size,
                ask_size,
                ts_event,
                UnixNanos::default(),
            )
            .unwrap();

        assert_eq!(quote.bid_price, Price::from("0.0035"));
        assert_eq!(quote.ask_price, Price::from("0.0040"));
        assert_eq!(quote.bid_size, Quantity::from(100));
        assert_eq!(quote.ask_size, Quantity::from(200));
    }

    #[rstest]
    fn test_quote_cache_empty_bids_uses_cached_value() {
        use nautilus_common::cache::quote::QuoteCache;

        let mut cache = QuoteCache::new();
        let instrument_id = InstrumentId::from("BTC-USD-260327-80000-C.OKX");

        cache
            .process(
                instrument_id,
                Some(Price::from("0.0010")),
                Some(Price::from("0.0015")),
                Some(Quantity::from(50)),
                Some(Quantity::from(75)),
                UnixNanos::default(),
                UnixNanos::default(),
            )
            .unwrap();

        let msg = stub_book_msg(vec![], vec![stub_book_entry("0.0020", "100")]);
        let ask_price = Some(parse_price(&msg.asks[0].price, 4).unwrap());
        let ask_size = Some(parse_quantity(&msg.asks[0].size, 0).unwrap());
        let ts_event = parse_millisecond_timestamp(msg.ts);

        let quote = cache
            .process(
                instrument_id,
                None,
                ask_price,
                None,
                ask_size,
                ts_event,
                UnixNanos::default(),
            )
            .unwrap();

        assert_eq!(quote.bid_price, Price::from("0.0010"));
        assert_eq!(quote.bid_size, Quantity::from(50));
        assert_eq!(quote.ask_price, Price::from("0.0020"));
        assert_eq!(quote.ask_size, Quantity::from(100));
    }

    #[rstest]
    fn test_quote_cache_empty_asks_uses_cached_value() {
        use nautilus_common::cache::quote::QuoteCache;

        let mut cache = QuoteCache::new();
        let instrument_id = InstrumentId::from("BTC-USD-260327-79000-P.OKX");

        cache
            .process(
                instrument_id,
                Some(Price::from("0.0010")),
                Some(Price::from("0.0015")),
                Some(Quantity::from(50)),
                Some(Quantity::from(75)),
                UnixNanos::default(),
                UnixNanos::default(),
            )
            .unwrap();

        let msg = stub_book_msg(vec![stub_book_entry("0.0012", "60")], vec![]);
        let bid_price = Some(parse_price(&msg.bids[0].price, 4).unwrap());
        let bid_size = Some(parse_quantity(&msg.bids[0].size, 0).unwrap());
        let ts_event = parse_millisecond_timestamp(msg.ts);

        let quote = cache
            .process(
                instrument_id,
                bid_price,
                None,
                bid_size,
                None,
                ts_event,
                UnixNanos::default(),
            )
            .unwrap();

        assert_eq!(quote.bid_price, Price::from("0.0012"));
        assert_eq!(quote.bid_size, Quantity::from(60));
        assert_eq!(quote.ask_price, Price::from("0.0015"));
        assert_eq!(quote.ask_size, Quantity::from(75));
    }

    #[rstest]
    fn test_quote_cache_both_sides_empty_no_cache_returns_error() {
        use nautilus_common::cache::quote::QuoteCache;

        let mut cache = QuoteCache::new();
        let instrument_id = InstrumentId::from("BTC-USD-260327-80000-C.OKX");

        let result = cache.process(
            instrument_id,
            None,
            None,
            None,
            None,
            UnixNanos::default(),
            UnixNanos::default(),
        );

        result.unwrap_err();
    }

    #[rstest]
    fn test_quote_cache_both_sides_empty_with_cache_returns_cached() {
        use nautilus_common::cache::quote::QuoteCache;

        let mut cache = QuoteCache::new();
        let instrument_id = InstrumentId::from("BTC-USD-260327-80000-C.OKX");

        cache
            .process(
                instrument_id,
                Some(Price::from("0.0010")),
                Some(Price::from("0.0015")),
                Some(Quantity::from(50)),
                Some(Quantity::from(75)),
                UnixNanos::default(),
                UnixNanos::default(),
            )
            .unwrap();

        let quote = cache
            .process(
                instrument_id,
                None,
                None,
                None,
                None,
                UnixNanos::from(1_706_000_000_000_000_000_u64),
                UnixNanos::from(1_706_000_000_000_000_000_u64),
            )
            .unwrap();

        assert_eq!(quote.bid_price, Price::from("0.0010"));
        assert_eq!(quote.ask_price, Price::from("0.0015"));
        assert_eq!(
            quote.ts_event,
            UnixNanos::from(1_706_000_000_000_000_000_u64)
        );
    }

    #[rstest]
    fn test_parse_instruments_channel_produces_status() {
        use nautilus_model::{enums::MarketStatusAction, identifiers::InstrumentId};

        use crate::common::{models::OKXInstrument, parse::parse_instrument_any};

        let ts_init = UnixNanos::default();

        // Build a cached instrument with fees
        let inst_json = serde_json::json!({
            "instType": "SPOT",
            "instId": "BTC-USD",
            "baseCcy": "BTC",
            "quoteCcy": "USD",
            "settleCcy": "",
            "ctVal": "",
            "ctMult": "",
            "ctValCcy": "",
            "optType": "",
            "stk": "",
            "listTime": "1733454000000",
            "expTime": "",
            "lever": "",
            "tickSz": "0.1",
            "lotSz": "0.00000001",
            "minSz": "0.00001",
            "ctType": "",
            "state": "live",
            "ruleType": "normal",
            "maxLmtSz": "9999999999",
            "maxMktSz": "1000000",
            "maxLmtAmt": "20000000",
            "maxMktAmt": "1000000",
            "maxTwapSz": "9999999999",
            "maxIcebergSz": "9999999999",
            "maxTriggerSz": "9999999999",
            "maxStopSz": "1000000",
            "uly": "",
            "instFamily": ""
        });
        let initial: OKXInstrument = serde_json::from_value(inst_json).unwrap();
        let parsed = parse_instrument_any(&initial, None, None, None, None, ts_init)
            .unwrap()
            .unwrap();

        let mut instruments_cache = AHashMap::new();
        instruments_cache.insert(Ustr::from("BTC-USD"), parsed);

        let ws_data = serde_json::json!({
            "instType": "SPOT",
            "instId": "BTC-USD",
            "baseCcy": "BTC",
            "quoteCcy": "USD",
            "settleCcy": "",
            "ctVal": "",
            "ctMult": "",
            "ctValCcy": "",
            "optType": "",
            "stk": "",
            "listTime": "1733454000000",
            "expTime": "",
            "lever": "",
            "tickSz": "0.1",
            "lotSz": "0.00000001",
            "minSz": "0.00001",
            "ctType": "",
            "state": "live",
            "ruleType": "normal",
            "maxLmtSz": "9999999999",
            "maxMktSz": "1000000",
            "maxLmtAmt": "20000000",
            "maxMktAmt": "1000000",
            "maxTwapSz": "9999999999",
            "maxIcebergSz": "9999999999",
            "maxTriggerSz": "9999999999",
            "maxStopSz": "1000000",
            "uly": "",
            "instFamily": ""
        });

        let instrument_id = InstrumentId::from("BTC-USD.OKX");
        let mut funding_cache = AHashMap::new();

        let result = parse_ws_message_data(
            &OKXWsChannel::Instruments,
            ws_data,
            &instrument_id,
            2,
            8,
            ts_init,
            &mut funding_cache,
            &instruments_cache,
        )
        .expect("Failed to parse instruments channel");

        match result {
            Some(NautilusWsMessage::Instrument(inst, status)) => {
                assert_eq!(inst.id(), InstrumentId::from("BTC-USD.OKX"));
                let status = status.expect("Expected InstrumentStatus");
                assert_eq!(status.action, MarketStatusAction::Trading);
                assert_eq!(status.is_trading, Some(true));
            }
            other => panic!("Expected Instrument with status, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_instruments_channel_returns_status_when_definition_invalid() {
        use nautilus_model::{enums::MarketStatusAction, identifiers::InstrumentId};

        let ts_init = UnixNanos::default();
        let ws_data = serde_json::json!({
            "instType": "SPOT",
            "instId": "USDG-SGD",
            "baseCcy": "USDG",
            "quoteCcy": "SGD",
            "settleCcy": "",
            "ctVal": "",
            "ctMult": "",
            "ctValCcy": "",
            "optType": "",
            "stk": "",
            "listTime": "1733454000000",
            "expTime": "",
            "lever": "",
            "tickSz": "",
            "lotSz": "0.00000001",
            "minSz": "0.00001",
            "ctType": "",
            "state": "live",
            "ruleType": "normal",
            "maxLmtSz": "9999999999",
            "maxMktSz": "1000000",
            "maxLmtAmt": "20000000",
            "maxMktAmt": "1000000",
            "maxTwapSz": "9999999999",
            "maxIcebergSz": "9999999999",
            "maxTriggerSz": "9999999999",
            "maxStopSz": "1000000",
            "uly": "",
            "instFamily": ""
        });

        let mut funding_cache = AHashMap::new();
        let instruments_cache = AHashMap::new();
        let result = parse_ws_message_data(
            &OKXWsChannel::Instruments,
            ws_data,
            &InstrumentId::from("BTC-USD.OKX"),
            2,
            8,
            ts_init,
            &mut funding_cache,
            &instruments_cache,
        )
        .expect("Failed to parse instruments channel");

        match result {
            Some(NautilusWsMessage::InstrumentStatus(status)) => {
                assert_eq!(status.instrument_id, InstrumentId::from("USDG-SGD.OKX"));
                assert_eq!(status.action, MarketStatusAction::Trading);
                assert_eq!(status.is_trading, Some(true));
            }
            other => panic!("Expected InstrumentStatus, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_instruments_channel_suspend_status() {
        use nautilus_model::{enums::MarketStatusAction, identifiers::InstrumentId};

        use crate::common::{models::OKXInstrument, parse::parse_instrument_any};

        let ts_init = UnixNanos::default();

        let inst_json = serde_json::json!({
            "instType": "SPOT",
            "instId": "BTC-USD",
            "baseCcy": "BTC",
            "quoteCcy": "USD",
            "settleCcy": "",
            "ctVal": "",
            "ctMult": "",
            "ctValCcy": "",
            "optType": "",
            "stk": "",
            "listTime": "1733454000000",
            "expTime": "",
            "lever": "",
            "tickSz": "0.1",
            "lotSz": "0.00000001",
            "minSz": "0.00001",
            "ctType": "",
            "state": "live",
            "ruleType": "normal",
            "maxLmtSz": "9999999999",
            "maxMktSz": "1000000",
            "maxLmtAmt": "20000000",
            "maxMktAmt": "1000000",
            "maxTwapSz": "9999999999",
            "maxIcebergSz": "9999999999",
            "maxTriggerSz": "9999999999",
            "maxStopSz": "1000000",
            "uly": "",
            "instFamily": ""
        });
        let initial: OKXInstrument = serde_json::from_value(inst_json).unwrap();
        let parsed = parse_instrument_any(&initial, None, None, None, None, ts_init)
            .unwrap()
            .unwrap();

        let mut instruments_cache = AHashMap::new();
        instruments_cache.insert(Ustr::from("BTC-USD"), parsed);

        // WS update with suspend state
        let ws_data = serde_json::json!({
            "instType": "SPOT",
            "instId": "BTC-USD",
            "baseCcy": "BTC",
            "quoteCcy": "USD",
            "settleCcy": "",
            "ctVal": "",
            "ctMult": "",
            "ctValCcy": "",
            "optType": "",
            "stk": "",
            "listTime": "1733454000000",
            "expTime": "",
            "lever": "",
            "tickSz": "0.1",
            "lotSz": "0.00000001",
            "minSz": "0.00001",
            "ctType": "",
            "state": "suspend",
            "ruleType": "normal",
            "maxLmtSz": "9999999999",
            "maxMktSz": "1000000",
            "maxLmtAmt": "20000000",
            "maxMktAmt": "1000000",
            "maxTwapSz": "9999999999",
            "maxIcebergSz": "9999999999",
            "maxTriggerSz": "9999999999",
            "maxStopSz": "1000000",
            "uly": "",
            "instFamily": ""
        });

        let instrument_id = InstrumentId::from("BTC-USD.OKX");
        let mut funding_cache = AHashMap::new();

        let result = parse_ws_message_data(
            &OKXWsChannel::Instruments,
            ws_data,
            &instrument_id,
            2,
            8,
            ts_init,
            &mut funding_cache,
            &instruments_cache,
        )
        .expect("Failed to parse instruments channel");

        match result {
            Some(NautilusWsMessage::Instrument(_, status)) => {
                let status = status.expect("Expected InstrumentStatus");
                assert_eq!(status.action, MarketStatusAction::Suspend);
                assert_eq!(status.is_trading, Some(false));
            }
            other => panic!("Expected Instrument with status, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_option_summary_greeks() {
        let json_str = load_test_json("ws_opt_summary.json");
        let msgs: Vec<OKXOptionSummaryMsg> =
            serde_json::from_str(&json_str).expect("Failed to deserialize opt-summary fixture");
        assert_eq!(msgs.len(), 2);

        let instrument_id = InstrumentId::from("BTC-USD-250328-92000-C.OKX");
        let ts_init = UnixNanos::from(1_711_612_900_000_000_000u64);
        let greeks =
            parse_option_summary_greeks(&msgs[0], &instrument_id, OKXGreeksType::Bs, ts_init)
                .expect("parse failed");

        assert_eq!(greeks.instrument_id, instrument_id);
        assert!((greeks.greeks.delta - 0.5312).abs() < 1e-10);
        assert!((greeks.greeks.gamma - 0.000_013_4).abs() < 1e-15);
        assert!((greeks.greeks.vega - 0.0038).abs() < 1e-10);
        assert!((greeks.greeks.theta - (-0.0015)).abs() < 1e-10);
        assert!((greeks.greeks.rho - 0.0).abs() < 1e-10);
        assert!((greeks.mark_iv.unwrap() - 0.53).abs() < 1e-10);
        assert!((greeks.bid_iv.unwrap() - 0.52).abs() < 1e-10);
        assert!((greeks.ask_iv.unwrap() - 0.55).abs() < 1e-10);
        assert!((greeks.underlying_price.unwrap() - 92150.50).abs() < 1e-10);
        assert!(greeks.open_interest.is_none());
        assert_eq!(greeks.convention, GreeksConvention::BlackScholes);
        assert_eq!(
            greeks.ts_event,
            UnixNanos::from(1_711_612_800_000_000_000u64)
        );
        assert_eq!(greeks.ts_init, ts_init);
    }

    #[rstest]
    fn test_option_summary_msg_deserializes_with_uppercase_bs_alias() {
        let json = r#"{
            "instId": "BTC-USD-250328-92000-C",
            "uly": "BTC-USD",
            "delta": "0.52",
            "gamma": "0.00001",
            "theta": "-0.001",
            "vega": "0.003",
            "deltaBS": "0.53",
            "gammaBS": "0.00002",
            "thetaBS": "-0.002",
            "vegaBS": "0.004",
            "realVol": "0.45",
            "bidVol": "0.50",
            "askVol": "0.55",
            "markVol": "0.52",
            "lever": "10.0",
            "ts": "1711612800000"
        }"#;
        let msg: OKXOptionSummaryMsg =
            serde_json::from_str(json).expect("deltaBS alias failed to deserialize");
        assert_eq!(msg.delta_bs, "0.53");
        assert_eq!(msg.gamma_bs, "0.00002");
        assert_eq!(msg.theta_bs, "-0.002");
        assert_eq!(msg.vega_bs, "0.004");
    }

    #[rstest]
    fn test_parse_option_summary_greeks_put() {
        let json_str = load_test_json("ws_opt_summary.json");
        let msgs: Vec<OKXOptionSummaryMsg> =
            serde_json::from_str(&json_str).expect("Failed to deserialize opt-summary fixture");

        let instrument_id = InstrumentId::from("BTC-USD-250328-92000-P.OKX");
        let ts_init = UnixNanos::from(1_711_612_900_000_000_000u64);
        let greeks =
            parse_option_summary_greeks(&msgs[1], &instrument_id, OKXGreeksType::Bs, ts_init)
                .expect("parse failed");

        assert!((greeks.greeks.delta - (-0.4688)).abs() < 1e-10);
    }

    #[rstest]
    fn test_parse_option_summary_greeks_pa() {
        let json_str = load_test_json("ws_opt_summary.json");
        let msgs: Vec<OKXOptionSummaryMsg> =
            serde_json::from_str(&json_str).expect("Failed to deserialize opt-summary fixture");
        assert_eq!(msgs.len(), 2);

        let instrument_id = InstrumentId::from("BTC-USD-250328-92000-C.OKX");
        let ts_init = UnixNanos::from(1_711_612_900_000_000_000u64);
        let greeks =
            parse_option_summary_greeks(&msgs[0], &instrument_id, OKXGreeksType::Pa, ts_init)
                .expect("parse failed");

        assert_eq!(greeks.instrument_id, instrument_id);
        assert!((greeks.greeks.delta - 0.5234).abs() < 1e-10);
        assert!((greeks.greeks.gamma - 0.000_012_3).abs() < 1e-15);
        assert!((greeks.greeks.vega - 0.0034).abs() < 1e-10);
        assert!((greeks.greeks.theta - (-0.0012)).abs() < 1e-10);
        assert!((greeks.greeks.rho - 0.0).abs() < 1e-10);
        assert!((greeks.mark_iv.unwrap() - 0.53).abs() < 1e-10);
        assert!((greeks.bid_iv.unwrap() - 0.52).abs() < 1e-10);
        assert!((greeks.ask_iv.unwrap() - 0.55).abs() < 1e-10);
        assert!((greeks.underlying_price.unwrap() - 92150.50).abs() < 1e-10);
        assert_eq!(greeks.convention, GreeksConvention::PriceAdjusted);
    }

    #[rstest]
    fn test_parse_option_summary_greeks_pa_put() {
        let json_str = load_test_json("ws_opt_summary.json");
        let msgs: Vec<OKXOptionSummaryMsg> =
            serde_json::from_str(&json_str).expect("Failed to deserialize opt-summary fixture");

        let instrument_id = InstrumentId::from("BTC-USD-250328-92000-P.OKX");
        let ts_init = UnixNanos::from(1_711_612_900_000_000_000u64);
        let greeks =
            parse_option_summary_greeks(&msgs[1], &instrument_id, OKXGreeksType::Pa, ts_init)
                .expect("parse failed");

        assert!((greeks.greeks.delta - (-0.4766)).abs() < 1e-10);
    }

    #[rstest]
    fn test_option_greeks_filtering_only_subscribed_instruments() {
        use ahash::AHashSet;

        let json_str = load_test_json("ws_opt_summary.json");
        let msgs: Vec<OKXOptionSummaryMsg> =
            serde_json::from_str(&json_str).expect("Failed to deserialize");

        let call_id = InstrumentId::from("BTC-USD-250328-92000-C.OKX");
        let put_id = InstrumentId::from("BTC-USD-250328-92000-P.OKX");
        let ts_init = UnixNanos::from(1_711_612_900_000_000_000u64);

        // Subscribe to CALL only
        let mut subs = AHashSet::new();
        subs.insert(call_id);

        let mut results = Vec::new();

        for msg in &msgs {
            let inst_id_str = format!("{}.OKX", msg.inst_id);
            let instrument_id = InstrumentId::from(inst_id_str.as_str());
            if !subs.contains(&instrument_id) {
                continue;
            }

            if let Ok(greeks) =
                parse_option_summary_greeks(msg, &instrument_id, OKXGreeksType::Bs, ts_init)
            {
                results.push(greeks);
            }
        }

        assert_eq!(results.len(), 1);
        assert_eq!(results[0].instrument_id, call_id);
        assert!((results[0].greeks.delta - 0.5312).abs() < 1e-10);

        // Now subscribe to both
        subs.insert(put_id);

        let mut results = Vec::new();

        for msg in &msgs {
            let inst_id_str = format!("{}.OKX", msg.inst_id);
            let instrument_id = InstrumentId::from(inst_id_str.as_str());
            if !subs.contains(&instrument_id) {
                continue;
            }

            if let Ok(greeks) =
                parse_option_summary_greeks(msg, &instrument_id, OKXGreeksType::Bs, ts_init)
            {
                results.push(greeks);
            }
        }

        assert_eq!(results.len(), 2);
    }

    #[rstest]
    fn test_option_greeks_unsubscribed_instrument_filtered_out() {
        use ahash::AHashSet;

        let json_str = load_test_json("ws_opt_summary.json");
        let msgs: Vec<OKXOptionSummaryMsg> =
            serde_json::from_str(&json_str).expect("Failed to deserialize");

        let ts_init = UnixNanos::default();

        // Empty subscription set
        let subs: AHashSet<InstrumentId> = AHashSet::new();

        let mut results = Vec::new();

        for msg in &msgs {
            let inst_id_str = format!("{}.OKX", msg.inst_id);
            let instrument_id = InstrumentId::from(inst_id_str.as_str());
            if !subs.contains(&instrument_id) {
                continue;
            }

            if let Ok(greeks) =
                parse_option_summary_greeks(msg, &instrument_id, OKXGreeksType::Bs, ts_init)
            {
                results.push(greeks);
            }
        }

        assert!(results.is_empty());
    }

    #[rstest]
    fn test_option_greeks_family_dedup_subscribe_count() {
        use crate::common::parse::extract_inst_family;

        let mut family_subs: AHashMap<Ustr, usize> = AHashMap::new();

        let call_id = InstrumentId::from("BTC-USD-250328-92000-C.OKX");
        let put_id = InstrumentId::from("BTC-USD-250328-92000-P.OKX");
        let other_id = InstrumentId::from("BTC-USD-250328-80000-C.OKX");

        // Subscribe first instrument: count goes to 1 (triggers WS subscribe)
        let family = extract_inst_family(call_id.symbol.inner().as_str()).unwrap();
        let count = family_subs.entry(family).or_default();
        *count += 1;
        assert_eq!(*count, 1);
        let should_subscribe_ws = *count == 1;
        assert!(should_subscribe_ws);

        // Subscribe second instrument in same family: count goes to 2 (no WS subscribe)
        let family = extract_inst_family(put_id.symbol.inner().as_str()).unwrap();
        let count = family_subs.entry(family).or_default();
        *count += 1;
        assert_eq!(*count, 2);
        let should_subscribe_ws = *count == 1;
        assert!(!should_subscribe_ws);

        // Subscribe third instrument in same family: count goes to 3
        let family = extract_inst_family(other_id.symbol.inner().as_str()).unwrap();
        let count = family_subs.entry(family).or_default();
        *count += 1;
        assert_eq!(*count, 3);

        // Unsubscribe one: count goes to 2 (no WS unsubscribe)
        let family = extract_inst_family(call_id.symbol.inner().as_str()).unwrap();
        if let Some(count) = family_subs.get_mut(&family) {
            *count = count.saturating_sub(1);
            assert_eq!(*count, 2);
            let should_unsubscribe_ws = *count == 0;
            assert!(!should_unsubscribe_ws);
        }

        // Unsubscribe second: count goes to 1
        let family = extract_inst_family(put_id.symbol.inner().as_str()).unwrap();
        if let Some(count) = family_subs.get_mut(&family) {
            *count = count.saturating_sub(1);
            assert_eq!(*count, 1);
        }

        // Unsubscribe last: count goes to 0 (triggers WS unsubscribe)
        let family = extract_inst_family(other_id.symbol.inner().as_str()).unwrap();
        if let Some(count) = family_subs.get_mut(&family) {
            *count = count.saturating_sub(1);
            assert_eq!(*count, 0);
            let should_unsubscribe_ws = *count == 0;
            assert!(should_unsubscribe_ws);
        }
    }

    #[rstest]
    fn test_parse_event_contract_markets_returns_raw_message() {
        let data = serde_json::json!([
            {
                "seriesId": "BTC-ABOVE-DAILY",
                "eventId": "BTC-ABOVE-DAILY-260224-1600",
                "instId": "BTC-ABOVE-DAILY-260224-1600-65000",
                "listTime": "1769697132335",
                "fixTime": "",
                "expTime": "1769697132335",
                "state": "live",
                "outcome": "0",
                "floorStrike": "120000",
                "capStrike": "",
                "settleValue": "",
                "disputed": false,
                "hitDir": ""
            },
            {
                "seriesId": "BTC-HIT-MONTHLY",
                "eventId": "BTC-HIT-MONTHLY-260831-1600",
                "instId": "BTC-HIT-MONTHLY-260831-1600-37500",
                "listTime": "1785513600000",
                "fixTime": "",
                "expTime": "1788192000000",
                "state": "live",
                "outcome": "0",
                "floorStrike": "37500",
                "capStrike": "",
                "settleValue": "",
                "disputed": false,
                "hitDir": "dn"
            }
        ]);
        let instrument_id = InstrumentId::from("BTC-ABOVE-DAILY-260224-1600-65000.OKX");
        let mut funding_cache = AHashMap::new();
        let instruments_cache = AHashMap::new();

        let result = parse_ws_message_data(
            &OKXWsChannel::EventContractMarkets,
            data.clone(),
            &instrument_id,
            2,
            2,
            UnixNanos::default(),
            &mut funding_cache,
            &instruments_cache,
        )
        .unwrap();

        match result {
            Some(NautilusWsMessage::Raw(raw)) => assert_eq!(raw, data),
            _ => panic!("Expected raw event contract market payload"),
        }
    }

    #[rstest]
    fn test_parse_ws_message_data_spread_public_trades() {
        // sprd-public-trades keys the spread as `sprdId` and omits `count`; the
        // instrument is resolved from the channel arg, so the trade still parses.
        let data = serde_json::json!([{
            "sprdId": "ETH-USD-260925_ETH-USD-261225",
            "tradeId": "3392538740127301632",
            "px": "16.9",
            "sz": "100",
            "side": "sell",
            "ts": "1780047866507"
        }]);
        let instrument_id = InstrumentId::from("ETH-USD-260925_ETH-USD-261225.OKX");
        let mut funding_cache = AHashMap::new();
        let instruments_cache = AHashMap::new();

        let result = parse_ws_message_data(
            &OKXWsChannel::SprdPublicTrades,
            data,
            &instrument_id,
            1,
            0,
            UnixNanos::default(),
            &mut funding_cache,
            &instruments_cache,
        )
        .unwrap();

        let Some(NautilusWsMessage::Data(data_vec)) = result else {
            panic!("expected Data variant, was {result:?}");
        };
        assert_eq!(data_vec.len(), 1);
        let Data::Trade(trade) = &data_vec[0] else {
            panic!("expected Data::Trade, was {:?}", data_vec[0]);
        };
        assert_eq!(trade.instrument_id, instrument_id);
        assert_eq!(trade.price.as_decimal(), dec!(16.9));
        assert_eq!(trade.size.as_decimal(), dec!(100));
        assert_eq!(trade.aggressor_side, AggressorSide::Sell);
    }

    #[rstest]
    fn test_parse_spread_books5_snapshot_with_three_element_levels() {
        // sprd-books5 pushes a full snapshot with 3-element `[price, size, count]`
        // levels; it is parsed as a snapshot (F_SNAPSHOT), not an incremental update.
        let msg: OKXBookMsg = serde_json::from_value(serde_json::json!({
            "asks": [["16.7", "100", "1"]],
            "bids": [["16.65", "100", "1"]],
            "ts": "1780044924909",
            "seqId": 1_779_935_772_619_784_u64,
        }))
        .unwrap();
        let instrument_id = InstrumentId::from("ETH-USD-260925_ETH-USD-261225.OKX");

        let deltas = parse_book_msg(
            &msg,
            instrument_id,
            2,
            0,
            &OKXBookAction::Snapshot,
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(deltas.instrument_id, instrument_id);
        assert_eq!(
            deltas.flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
        let bid = deltas
            .deltas
            .iter()
            .find(|d| d.order.side == OrderSide::Buy.into())
            .expect("should have a bid delta");
        let ask = deltas
            .deltas
            .iter()
            .find(|d| d.order.side == OrderSide::Sell.into())
            .expect("should have an ask delta");
        assert_eq!(bid.order.price.as_decimal(), dec!(16.65));
        assert_eq!(ask.order.price.as_decimal(), dec!(16.7));
    }
}