nautilus_kraken/websocket/spot_v2/

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

//! WebSocket message parsers for converting Kraken streaming data to Nautilus domain models.

use anyhow::Context;
use chrono::{DateTime, Utc};
use nautilus_core::{UUID4, nanos::UnixNanos};
use nautilus_model::{
    data::{Bar, BarSpecification, BarType, BookOrder, OrderBookDelta, QuoteTick, TradeTick},
    enums::{
        AggregationSource, AggressorSide, BarAggregation, BookAction, LiquiditySide, OrderSide,
        OrderStatus, OrderType, PriceType, RecordFlag, TimeInForce, TriggerType,
    },
    identifiers::{AccountId, ClientOrderId, InstrumentId, TradeId, VenueOrderId},
    instruments::{Instrument, any::InstrumentAny},
    reports::{FillReport, OrderStatusReport},
    types::{Currency, Money, Price, Quantity},
};

use super::{
    enums::{KrakenExecType, KrakenLiquidityInd, KrakenWsOrderStatus},
    messages::{
        KrakenSpotWsMessage, KrakenWsBookData, KrakenWsBookLevel, KrakenWsExecutionData,
        KrakenWsOhlcData, KrakenWsOrderResponse, KrakenWsTickerData, KrakenWsTradeData,
    },
};
use crate::common::enums::{KrakenOrderSide, KrakenOrderType, KrakenTimeInForce};

/// Parses Kraken WebSocket ticker data into a Nautilus quote tick.
///
/// # Errors
///
/// Returns an error if:
/// - Bid or ask price/quantity cannot be parsed.
pub fn parse_quote_tick(
    ticker: &KrakenWsTickerData,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<QuoteTick> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let bid_price = Price::new_checked(ticker.bid, price_precision).with_context(|| {
        format!("Failed to construct bid Price with precision {price_precision}")
    })?;
    let bid_size = Quantity::new_checked(ticker.bid_qty, size_precision).with_context(|| {
        format!("Failed to construct bid Quantity with precision {size_precision}")
    })?;

    let ask_price = Price::new_checked(ticker.ask, price_precision).with_context(|| {
        format!("Failed to construct ask Price with precision {price_precision}")
    })?;
    let ask_size = Quantity::new_checked(ticker.ask_qty, size_precision).with_context(|| {
        format!("Failed to construct ask Quantity with precision {size_precision}")
    })?;

    let ts_event = datetime_to_nanos(ticker.timestamp, "ticker.timestamp")?;

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

/// Parses Kraken WebSocket trade data into a Nautilus trade tick.
///
/// # Errors
///
/// Returns an error if:
/// - Price or quantity cannot be parsed.
/// - Timestamp is invalid.
pub fn parse_trade_tick(
    trade: &KrakenWsTradeData,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<TradeTick> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let price = Price::new_checked(trade.price, price_precision)
        .with_context(|| format!("Failed to construct Price with precision {price_precision}"))?;
    let size = Quantity::new_checked(trade.qty, size_precision)
        .with_context(|| format!("Failed to construct Quantity with precision {size_precision}"))?;

    let aggressor = match trade.side {
        KrakenOrderSide::Buy => AggressorSide::Buyer,
        KrakenOrderSide::Sell => AggressorSide::Seller,
    };

    let trade_id = TradeId::new_checked(trade.trade_id.to_string())?;
    let ts_event = datetime_to_nanos(trade.timestamp, "trade.timestamp")?;

    TradeTick::new_checked(
        instrument_id,
        price,
        size,
        aggressor,
        trade_id,
        ts_event,
        ts_init,
    )
    .context("Failed to construct TradeTick from Kraken WebSocket trade")
}

/// Parses Kraken WebSocket book data into Nautilus order book deltas.
///
/// Returns a vector of deltas, one for each bid and ask level.
///
/// # Errors
///
/// Returns an error if:
/// - Price or quantity cannot be parsed.
/// - Timestamp is invalid.
pub fn parse_book_deltas(
    book: &KrakenWsBookData,
    instrument: &InstrumentAny,
    sequence: u64,
    is_snapshot: bool,
    ts_init: UnixNanos,
) -> anyhow::Result<Vec<OrderBookDelta>> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let ts_event = datetime_to_nanos(book.timestamp, "book.timestamp")?;

    let mut current_sequence = sequence;
    let mut deltas = Vec::new();

    if is_snapshot {
        deltas.push(OrderBookDelta::clear(
            instrument_id,
            current_sequence,
            ts_event,
            ts_init,
        ));
        current_sequence += 1;
    }

    if let Some(ref bids) = book.bids {
        parse_book_side_levels(
            bids,
            OrderSide::Buy,
            is_snapshot,
            instrument_id,
            price_precision,
            size_precision,
            ts_event,
            ts_init,
            &mut current_sequence,
            &mut deltas,
        )?;
    }

    if let Some(ref asks) = book.asks {
        parse_book_side_levels(
            asks,
            OrderSide::Sell,
            is_snapshot,
            instrument_id,
            price_precision,
            size_precision,
            ts_event,
            ts_init,
            &mut current_sequence,
            &mut deltas,
        )?;
    }

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

    Ok(deltas)
}

#[expect(clippy::too_many_arguments)]
fn parse_book_side_levels(
    levels: &[KrakenWsBookLevel],
    side: OrderSide,
    is_snapshot: bool,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
    current_sequence: &mut u64,
    deltas: &mut Vec<OrderBookDelta>,
) -> anyhow::Result<()> {
    for level in levels {
        let Some(delta) = parse_book_level(
            level,
            side,
            is_snapshot,
            instrument_id,
            price_precision,
            size_precision,
            *current_sequence,
            ts_event,
            ts_init,
        )?
        else {
            continue;
        };
        deltas.push(delta);
        *current_sequence += 1;
    }

    Ok(())
}

#[expect(clippy::too_many_arguments)]
fn parse_book_level(
    level: &KrakenWsBookLevel,
    side: OrderSide,
    is_snapshot: bool,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    sequence: u64,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> anyhow::Result<Option<OrderBookDelta>> {
    let price = Price::new_checked(level.price, price_precision)
        .with_context(|| format!("Failed to construct Price with precision {price_precision}"))?;
    let size = Quantity::new_checked(level.qty, size_precision)
        .with_context(|| format!("Failed to construct Quantity with precision {size_precision}"))?;

    let action = if is_snapshot {
        if size.raw == 0 {
            return Ok(None);
        }
        BookAction::Add
    } else if size.raw == 0 {
        BookAction::Delete
    } else {
        BookAction::Update
    };

    let order_id = price.raw as u64;
    let order = BookOrder::new(side, price, size, order_id);
    let mut flags = RecordFlag::F_MBP as u8;
    if is_snapshot {
        flags |= RecordFlag::F_SNAPSHOT as u8;
    }

    Ok(Some(OrderBookDelta::new(
        instrument_id,
        action,
        order,
        flags,
        sequence,
        ts_event,
        ts_init,
    )))
}

pub(super) fn datetime_to_nanos(value: DateTime<Utc>, field: &str) -> anyhow::Result<UnixNanos> {
    let nanos = value
        .timestamp_nanos_opt()
        .with_context(|| format!("Failed to convert {field}='{value}' to nanoseconds"))?;
    Ok(UnixNanos::from(nanos as u64))
}

/// Parses Kraken WebSocket OHLC data into a Nautilus bar.
///
/// The bar's `ts_event` is computed as `interval_begin` + `interval` minutes.
///
/// # Errors
///
/// Returns an error if:
/// - Price or quantity values cannot be parsed.
/// - The interval cannot be converted to a valid bar specification.
pub fn parse_ws_bar(
    ohlc: &KrakenWsOhlcData,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<Bar> {
    let instrument_id = instrument.id();
    let price_precision = instrument.price_precision();
    let size_precision = instrument.size_precision();

    let open = Price::new_checked(ohlc.open, price_precision)?;
    let high = Price::new_checked(ohlc.high, price_precision)?;
    let low = Price::new_checked(ohlc.low, price_precision)?;
    let close = Price::new_checked(ohlc.close, price_precision)?;
    let volume = Quantity::new_checked(ohlc.volume, size_precision)?;

    let bar_spec = interval_to_bar_spec(ohlc.interval)?;
    let bar_type = BarType::new(instrument_id, bar_spec, AggregationSource::External);

    // Compute bar close time: interval_begin + interval minutes
    let interval_secs = i64::from(ohlc.interval) * 60;
    let close_time = ohlc.interval_begin + chrono::Duration::seconds(interval_secs);
    let ts_event = UnixNanos::from(close_time.timestamp_nanos_opt().unwrap_or(0) as u64);

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

/// Converts a Kraken OHLC interval (minutes) to a Nautilus bar specification.
fn interval_to_bar_spec(interval: u32) -> anyhow::Result<BarSpecification> {
    let (step, aggregation) = match interval {
        1 => (1, BarAggregation::Minute),
        5 => (5, BarAggregation::Minute),
        15 => (15, BarAggregation::Minute),
        30 => (30, BarAggregation::Minute),
        60 => (1, BarAggregation::Hour),
        240 => (4, BarAggregation::Hour),
        1440 => (1, BarAggregation::Day),
        10080 => (1, BarAggregation::Week),
        21600 => (15, BarAggregation::Day), // 21600 min = 360 hours = 15 days
        _ => anyhow::bail!("Unsupported Kraken OHLC interval: {interval}"),
    };

    Ok(BarSpecification::new(step, aggregation, PriceType::Last))
}

/// Parses Kraken execution type and order status to Nautilus order status.
fn parse_order_status(
    exec_type: KrakenExecType,
    order_status: Option<KrakenWsOrderStatus>,
) -> OrderStatus {
    match exec_type {
        KrakenExecType::Canceled => return OrderStatus::Canceled,
        KrakenExecType::Expired => return OrderStatus::Expired,
        KrakenExecType::Filled => return OrderStatus::Filled,
        KrakenExecType::Trade => {
            return match order_status {
                Some(KrakenWsOrderStatus::Filled) => OrderStatus::Filled,
                Some(KrakenWsOrderStatus::PartiallyFilled) | None => OrderStatus::PartiallyFilled,
                Some(status) => status.into(),
            };
        }
        _ => {}
    }

    match order_status {
        Some(status) => status.into(),
        None => OrderStatus::Accepted,
    }
}

/// Parses Kraken order type to Nautilus order type.
fn parse_order_type(order_type: Option<KrakenOrderType>) -> OrderType {
    match order_type {
        Some(KrakenOrderType::Market) => OrderType::Market,
        Some(KrakenOrderType::Limit) => OrderType::Limit,
        Some(KrakenOrderType::StopLoss) => OrderType::StopMarket,
        Some(KrakenOrderType::TakeProfit) => OrderType::MarketIfTouched,
        Some(KrakenOrderType::StopLossLimit) => OrderType::StopLimit,
        Some(KrakenOrderType::TakeProfitLimit) => OrderType::LimitIfTouched,
        // Trailing stops lack offset fields in WS reports, map to non-trailing equivalents
        Some(KrakenOrderType::TrailingStop) => OrderType::StopMarket,
        Some(KrakenOrderType::TrailingStopLimit) => OrderType::StopLimit,
        Some(KrakenOrderType::SettlePosition) => OrderType::Market,
        None => OrderType::Limit,
    }
}

/// Parses Kraken order side to Nautilus order side.
fn parse_order_side(side: Option<KrakenOrderSide>) -> OrderSide {
    match side {
        Some(KrakenOrderSide::Buy) => OrderSide::Buy,
        Some(KrakenOrderSide::Sell) => OrderSide::Sell,
        None => OrderSide::Buy,
    }
}

/// Parses Kraken time-in-force to Nautilus time-in-force.
fn parse_time_in_force(
    time_in_force: Option<KrakenTimeInForce>,
    post_only: Option<bool>,
) -> TimeInForce {
    // Handle post_only flag
    if post_only == Some(true) {
        return TimeInForce::Gtc;
    }

    match time_in_force {
        Some(KrakenTimeInForce::GoodTilCancelled) => TimeInForce::Gtc,
        Some(KrakenTimeInForce::ImmediateOrCancel) => TimeInForce::Ioc,
        Some(KrakenTimeInForce::GoodTilDate) => TimeInForce::Gtd,
        Some(KrakenTimeInForce::FillOrKill) => TimeInForce::Fok,
        None => TimeInForce::Gtc,
    }
}

fn parse_liquidity_side(liquidity_ind: Option<KrakenLiquidityInd>) -> LiquiditySide {
    liquidity_ind.map_or(LiquiditySide::NoLiquiditySide, Into::into)
}

/// Parses a Kraken WebSocket execution message into an [`OrderStatusReport`].
///
/// # Errors
///
/// Returns an error if required fields are missing or cannot be parsed.
pub fn parse_ws_order_status_report(
    exec: &KrakenWsExecutionData,
    instrument: &InstrumentAny,
    account_id: AccountId,
    cached_order_qty: Option<f64>,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderStatusReport> {
    let instrument_id = instrument.id();
    let venue_order_id = VenueOrderId::new(&exec.order_id);
    let order_side = parse_order_side(exec.side);
    let order_type = parse_order_type(exec.order_type);
    let time_in_force = parse_time_in_force(exec.time_in_force, exec.post_only);
    let order_status = parse_order_status(exec.exec_type, exec.order_status);

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

    // Quantity fallback: order_qty -> cached -> cum_qty -> last_qty (for trade snapshots)
    let last_qty = exec
        .last_qty
        .map(|qty| Quantity::new_checked(qty, size_precision))
        .transpose()
        .context("Failed to parse last_qty")?;

    let filled_qty = exec
        .cum_qty
        .map(|qty| Quantity::new_checked(qty, size_precision))
        .transpose()
        .context("Failed to parse cum_qty")?
        .or(last_qty)
        .unwrap_or_else(|| Quantity::zero(size_precision));

    let quantity = exec
        .order_qty
        .or(cached_order_qty)
        .map(|qty| Quantity::new_checked(qty, size_precision))
        .transpose()
        .context("Failed to parse order_qty")?
        .unwrap_or(filled_qty);

    let ts_event = datetime_to_nanos(exec.timestamp, "execution.timestamp")?;

    let mut report = OrderStatusReport::new(
        account_id,
        instrument_id,
        None, // client_order_id set below if present
        venue_order_id,
        order_side,
        order_type,
        time_in_force,
        order_status,
        quantity,
        filled_qty,
        ts_event,
        ts_event,
        ts_init,
        Some(UUID4::new()),
    );

    if let Some(ref cl_ord_id) = exec.cl_ord_id
        && !cl_ord_id.is_empty()
    {
        report = report.with_client_order_id(ClientOrderId::new(cl_ord_id));
    }

    // Price fallback: limit_price -> avg_price -> last_price
    // Note: pending_new messages may not include any price fields, which is fine for
    // orders we submitted (engine already has the price from submission)
    let price_value = exec
        .limit_price
        .filter(|&p| p > 0.0)
        .or(exec.avg_price.filter(|&p| p > 0.0))
        .or(exec.last_price.filter(|&p| p > 0.0));

    if let Some(px) = price_value {
        let price =
            Price::new_checked(px, price_precision).context("Failed to parse order price")?;
        report = report.with_price(price);
    }

    // avg_px fallback: avg_price -> cum_cost / cum_qty -> last_price (for single trades/snapshots)
    let avg_px = exec
        .avg_price
        .filter(|&p| p > 0.0)
        .or_else(|| match (exec.cum_cost, exec.cum_qty) {
            (Some(cost), Some(qty)) if qty > 0.0 => Some(cost / qty),
            _ => None,
        })
        .or_else(|| exec.last_price.filter(|&p| p > 0.0));

    if let Some(avg_price) = avg_px {
        report = report.with_avg_px(avg_price)?;
    }

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

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

    if let Some(ref reason) = exec.reason
        && !reason.is_empty()
    {
        report = report.with_cancel_reason(reason.clone());
    }

    // Set trigger type for conditional orders (WebSocket doesn't provide trigger field)
    let is_conditional = matches!(
        order_type,
        OrderType::StopMarket
            | OrderType::StopLimit
            | OrderType::MarketIfTouched
            | OrderType::LimitIfTouched
    );

    if is_conditional {
        report = report.with_trigger_type(TriggerType::Default);
    }

    Ok(report)
}

/// Parses a Kraken WebSocket trade execution into a [`FillReport`].
///
/// This should only be called when exec_type is "trade".
///
/// # Errors
///
/// Returns an error if required fields are missing or cannot be parsed.
pub fn parse_ws_fill_report(
    exec: &KrakenWsExecutionData,
    instrument: &InstrumentAny,
    account_id: AccountId,
    ts_init: UnixNanos,
) -> anyhow::Result<FillReport> {
    let instrument_id = instrument.id();
    let venue_order_id = VenueOrderId::new(&exec.order_id);

    let exec_id = exec
        .exec_id
        .as_ref()
        .context("Missing exec_id for trade execution")?;
    let trade_id =
        TradeId::new_checked(exec_id).context("Invalid exec_id in Kraken trade execution")?;

    let order_side = parse_order_side(exec.side);

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

    let last_qty = exec
        .last_qty
        .map(|qty| Quantity::new_checked(qty, size_precision))
        .transpose()
        .context("Failed to parse last_qty")?
        .context("Missing last_qty for trade execution")?;

    let last_px = exec
        .last_price
        .map(|px| Price::new_checked(px, price_precision))
        .transpose()
        .context("Failed to parse last_price")?
        .context("Missing last_price for trade execution")?;

    let liquidity_side = parse_liquidity_side(exec.liquidity_ind);

    // Calculate commission from fees array
    let commission = if let Some(ref fees) = exec.fees {
        if let Some(fee) = fees.first() {
            let currency = Currency::get_or_create_crypto(&fee.asset);
            Money::new(fee.qty.abs(), currency)
        } else {
            Money::zero(instrument.quote_currency())
        }
    } else {
        Money::zero(instrument.quote_currency())
    };

    let ts_event = datetime_to_nanos(exec.timestamp, "execution.timestamp")?;

    let client_order_id = exec
        .cl_ord_id
        .as_ref()
        .filter(|s| !s.is_empty())
        .map(ClientOrderId::new);

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

/// Parses a raw WebSocket JSON string and returns [`KrakenSpotWsMessage::OrderResponse`] if the
/// message is an order-method response envelope, or `Ok(None)` for unrecognised messages.
///
/// # Errors
///
/// Returns an error if the message appears to be an order response but cannot be deserialized.
pub fn parse_order_response(text: &str) -> anyhow::Result<Option<KrakenSpotWsMessage>> {
    let value: serde_json::Value =
        serde_json::from_str(text).with_context(|| format!("Failed to parse JSON: {text}"))?;

    let method_str = match value.get("method").and_then(|m| m.as_str()) {
        Some(s) => s.to_owned(),
        None => return Ok(None),
    };

    if !matches!(
        method_str.as_str(),
        "add_order" | "amend_order" | "cancel_order" | "batch_add"
    ) {
        return Ok(None);
    }

    let response: KrakenWsOrderResponse = serde_json::from_value(value).with_context(|| {
        format!("Failed to deserialize order response for method '{method_str}'")
    })?;
    Ok(Some(KrakenSpotWsMessage::OrderResponse(response)))
}

#[cfg(test)]
mod tests {
    use nautilus_model::{identifiers::Symbol, types::Currency};
    use rstest::rstest;
    use ustr::Ustr;

    use super::*;
    use crate::{common::consts::KRAKEN_VENUE, websocket::spot_v2::messages::KrakenWsMessage};

    const TS: UnixNanos = UnixNanos::new(1_700_000_000_000_000_000);

    #[rstest]
    fn test_parse_time_in_force_fok() {
        assert_eq!(
            parse_time_in_force(Some(KrakenTimeInForce::FillOrKill), None),
            TimeInForce::Fok
        );
    }

    fn load_test_json(filename: &str) -> String {
        let path = format!("test_data/{filename}");
        std::fs::read_to_string(&path)
            .unwrap_or_else(|e| panic!("Failed to load test data from {path}: {e}"))
    }

    fn create_mock_instrument() -> InstrumentAny {
        use nautilus_model::instruments::currency_pair::CurrencyPair;

        let instrument_id = InstrumentId::new(Symbol::new("BTC/USD"), *KRAKEN_VENUE);
        InstrumentAny::CurrencyPair(CurrencyPair::new(
            instrument_id,
            Symbol::new("XBTUSDT"),
            Currency::BTC(),
            Currency::USDT(),
            1, // price_precision
            8, // size_precision
            Price::from("0.1"),
            Quantity::from("0.00000001"),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None, // info
            TS,
            TS,
        ))
    }

    #[rstest]
    fn test_parse_quote_tick() {
        let json = load_test_json("ws_ticker_snapshot.json");
        let message: KrakenWsMessage = serde_json::from_str(&json).unwrap();
        let ticker: KrakenWsTickerData = serde_json::from_value(message.data[0].clone()).unwrap();

        let instrument = create_mock_instrument();
        let quote_tick = parse_quote_tick(&ticker, &instrument, TS).unwrap();

        assert_eq!(quote_tick.instrument_id, instrument.id());
        assert!(quote_tick.bid_price.as_f64() > 0.0);
        assert!(quote_tick.ask_price.as_f64() > 0.0);
        assert!(quote_tick.bid_size.as_f64() > 0.0);
        assert!(quote_tick.ask_size.as_f64() > 0.0);
        assert_eq!(
            quote_tick.ts_event,
            UnixNanos::from(1_671_960_659_123_456_000)
        );
        assert_eq!(quote_tick.ts_init, TS);
    }

    #[rstest]
    fn test_parse_trade_tick() {
        let json = load_test_json("ws_trade_update.json");
        let message: KrakenWsMessage = serde_json::from_str(&json).unwrap();
        let trade: KrakenWsTradeData = serde_json::from_value(message.data[0].clone()).unwrap();

        let instrument = create_mock_instrument();
        let trade_tick = parse_trade_tick(&trade, &instrument, TS).unwrap();

        assert_eq!(trade_tick.instrument_id, instrument.id());
        assert!(trade_tick.price.as_f64() > 0.0);
        assert!(trade_tick.size.as_f64() > 0.0);
        assert!(matches!(
            trade_tick.aggressor_side,
            AggressorSide::Buyer | AggressorSide::Seller
        ));
        assert_eq!(
            trade_tick.ts_event,
            UnixNanos::from(1_696_613_755_440_295_000)
        );
        assert_eq!(trade_tick.ts_init, TS);
    }

    #[rstest]
    fn test_parse_book_deltas_snapshot() {
        let json = load_test_json("ws_book_snapshot.json");
        let message: KrakenWsMessage = serde_json::from_str(&json).unwrap();
        let book: KrakenWsBookData = serde_json::from_value(message.data[0].clone()).unwrap();

        let instrument = create_mock_instrument();
        let deltas = parse_book_deltas(&book, &instrument, 1, true, TS).unwrap();

        assert!(!deltas.is_empty());

        let bid_count = deltas
            .iter()
            .filter(|d| d.order.side == OrderSide::Buy)
            .count();
        let ask_count = deltas
            .iter()
            .filter(|d| d.order.side == OrderSide::Sell)
            .count();

        assert!(bid_count > 0);
        assert!(ask_count > 0);

        let first_delta = &deltas[0];
        assert_eq!(first_delta.instrument_id, instrument.id());
        assert_eq!(first_delta.action, BookAction::Clear);
        assert!(RecordFlag::F_SNAPSHOT.matches(first_delta.flags));
        assert!(!RecordFlag::F_LAST.matches(first_delta.flags));

        assert!(deltas[1..].iter().all(|d| d.action == BookAction::Add));
        assert!(
            deltas[1..]
                .iter()
                .all(|d| RecordFlag::F_MBP.matches(d.flags))
        );
        assert!(
            deltas[1..]
                .iter()
                .all(|d| RecordFlag::F_SNAPSHOT.matches(d.flags))
        );
        assert!(RecordFlag::F_LAST.matches(deltas.last().unwrap().flags));

        let expected_ts_event = UnixNanos::from(1_696_613_755_440_295_000);
        assert!(deltas.iter().all(|d| d.ts_event == expected_ts_event));
        assert!(deltas.iter().all(|d| d.ts_init == TS));
    }

    #[rstest]
    fn test_parse_book_deltas_update() {
        let json = load_test_json("ws_book_update.json");
        let message: KrakenWsMessage = serde_json::from_str(&json).unwrap();
        let book: KrakenWsBookData = serde_json::from_value(message.data[0].clone()).unwrap();

        let instrument = create_mock_instrument();
        let deltas = parse_book_deltas(&book, &instrument, 1, false, TS).unwrap();

        assert!(!deltas.is_empty());

        let first_delta = &deltas[0];
        assert_eq!(first_delta.instrument_id, instrument.id());
        assert_eq!(first_delta.action, BookAction::Update);
        assert_eq!(first_delta.order.side, OrderSide::Buy);
        assert!(first_delta.order.price.as_f64() > 0.0);
        assert!(RecordFlag::F_MBP.matches(first_delta.flags));
        assert!(RecordFlag::F_LAST.matches(first_delta.flags));
        assert!(!RecordFlag::F_SNAPSHOT.matches(first_delta.flags));

        let expected_ts_event = UnixNanos::from(1_696_613_755_440_295_000);
        assert!(deltas.iter().all(|d| d.ts_event == expected_ts_event));
        assert!(deltas.iter().all(|d| d.ts_init == TS));
    }

    #[rstest]
    fn test_parse_book_deltas_snapshot_skips_zero_qty_levels() {
        let book = KrakenWsBookData {
            symbol: Ustr::from("BTC/USD"),
            bids: Some(vec![KrakenWsBookLevel {
                price: 100.0,
                qty: 0.0,
            }]),
            asks: Some(vec![KrakenWsBookLevel {
                price: 101.0,
                qty: 2.0,
            }]),
            checksum: Some(0),
            timestamp: "2024-01-01T00:00:00Z".parse().unwrap(),
        };

        let instrument = create_mock_instrument();
        let deltas = parse_book_deltas(&book, &instrument, 7, true, TS).unwrap();

        assert_eq!(deltas.len(), 2);
        assert_eq!(deltas[0].action, BookAction::Clear);
        assert_eq!(deltas[0].sequence, 7);
        assert!(RecordFlag::F_SNAPSHOT.matches(deltas[0].flags));
        assert!(!RecordFlag::F_LAST.matches(deltas[0].flags));

        let add = &deltas[1];
        assert_eq!(add.action, BookAction::Add);
        assert_eq!(add.sequence, 8);
        assert_eq!(add.order.side, OrderSide::Sell);
        assert_eq!(add.order.price, Price::from("101.0"));
        assert!(RecordFlag::F_MBP.matches(add.flags));
        assert!(RecordFlag::F_SNAPSHOT.matches(add.flags));
        assert!(RecordFlag::F_LAST.matches(add.flags));
    }

    #[rstest]
    fn test_parse_book_deltas_update_zero_qty_deletes_level() {
        let book = KrakenWsBookData {
            symbol: Ustr::from("BTC/USD"),
            bids: Some(vec![KrakenWsBookLevel {
                price: 100.0,
                qty: 0.0,
            }]),
            asks: Some(vec![]),
            checksum: Some(0),
            timestamp: "2024-01-01T00:00:00Z".parse().unwrap(),
        };

        let instrument = create_mock_instrument();
        let deltas = parse_book_deltas(&book, &instrument, 11, false, TS).unwrap();

        assert_eq!(deltas.len(), 1);
        let delete = &deltas[0];
        assert_eq!(delete.action, BookAction::Delete);
        assert_eq!(delete.sequence, 11);
        assert_eq!(delete.order.side, OrderSide::Buy);
        assert_eq!(delete.order.price, Price::from("100.0"));
        assert_eq!(delete.order.size.raw, 0);
        assert!(RecordFlag::F_MBP.matches(delete.flags));
        assert!(RecordFlag::F_LAST.matches(delete.flags));
        assert!(!RecordFlag::F_SNAPSHOT.matches(delete.flags));
    }

    #[rstest]
    fn test_datetime_to_nanos() {
        let dt = "2023-10-06T17:35:55.440295Z"
            .parse::<DateTime<Utc>>()
            .unwrap();
        let result = datetime_to_nanos(dt, "test").unwrap();
        assert_eq!(result, UnixNanos::from(1_696_613_755_440_295_000));
    }

    #[rstest]
    fn test_datetime_to_nanos_out_of_range_errors() {
        let dt = "1500-01-01T00:00:00Z".parse::<DateTime<Utc>>().unwrap();
        let result = datetime_to_nanos(dt, "test");
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("test"));
    }

    #[rstest]
    fn test_parse_ws_bar() {
        let json = load_test_json("ws_ohlc_update.json");
        let message: KrakenWsMessage = serde_json::from_str(&json).unwrap();
        let ohlc: KrakenWsOhlcData = serde_json::from_value(message.data[0].clone()).unwrap();

        let instrument = create_mock_instrument();
        let bar = parse_ws_bar(&ohlc, &instrument, TS).unwrap();

        assert_eq!(bar.bar_type.instrument_id(), instrument.id());
        assert!(bar.open.as_f64() > 0.0);
        assert!(bar.high.as_f64() > 0.0);
        assert!(bar.low.as_f64() > 0.0);
        assert!(bar.close.as_f64() > 0.0);
        assert!(bar.volume.as_f64() > 0.0);

        let spec = bar.bar_type.spec();
        assert_eq!(spec.step.get(), 1);
        assert_eq!(spec.aggregation, BarAggregation::Minute);
        assert_eq!(spec.price_type, PriceType::Last);

        // Verify ts_event is computed as interval_begin + interval (close time)
        // interval_begin is 2023-10-04T16:25:00Z, interval is 1 minute, so close is 16:26:00Z
        let expected_close = ohlc.interval_begin + chrono::Duration::minutes(1);
        let expected_ts_event =
            UnixNanos::from(expected_close.timestamp_nanos_opt().unwrap() as u64);
        assert_eq!(bar.ts_event, expected_ts_event);
    }

    #[rstest]
    fn test_interval_to_bar_spec() {
        let test_cases = [
            (1, 1, BarAggregation::Minute),
            (5, 5, BarAggregation::Minute),
            (15, 15, BarAggregation::Minute),
            (30, 30, BarAggregation::Minute),
            (60, 1, BarAggregation::Hour),
            (240, 4, BarAggregation::Hour),
            (1440, 1, BarAggregation::Day),
            (10080, 1, BarAggregation::Week),
            (21600, 15, BarAggregation::Day), // 21600 min = 15 days
        ];

        for (interval, expected_step, expected_aggregation) in test_cases {
            let spec = interval_to_bar_spec(interval).unwrap();
            assert_eq!(
                spec.step.get(),
                expected_step,
                "Failed for interval {interval}"
            );
            assert_eq!(
                spec.aggregation, expected_aggregation,
                "Failed for interval {interval}"
            );
            assert_eq!(spec.price_type, PriceType::Last);
        }
    }

    #[rstest]
    fn test_interval_to_bar_spec_invalid() {
        let result = interval_to_bar_spec(999);
        assert!(result.is_err());
    }

    #[rstest]
    fn test_parse_order_response_envelope_returns_order_response_variant() {
        use crate::websocket::spot_v2::enums::KrakenWsMethod;

        let raw = load_test_json("ws_add_order_response_success.json");
        let parsed = parse_order_response(&raw).expect("parse ok");
        match parsed {
            Some(KrakenSpotWsMessage::OrderResponse(resp)) => {
                assert_eq!(resp.method, KrakenWsMethod::AddOrder);
                assert_eq!(resp.req_id, Some(42));
                assert!(resp.success);
            }
            other => panic!("expected OrderResponse, was {other:?}"),
        }
    }

    #[rstest]
    fn test_parse_order_response_returns_none_for_non_order_method() {
        let json = r#"{"method":"subscribe","req_id":1,"success":true}"#;
        let result = parse_order_response(json).expect("parse ok");
        assert!(result.is_none());
    }

    #[rstest]
    fn test_parse_order_response_returns_none_for_data_message() {
        let json = r#"{"channel":"ticker","type":"snapshot","data":[]}"#;
        let result = parse_order_response(json).expect("parse ok");
        assert!(result.is_none());
    }
}