nautilus_derive/websocket/
parse.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Parsers for Derive public WebSocket subscription payloads.
17
18use anyhow::Context;
19use nautilus_core::{
20    UnixNanos,
21    datetime::{NANOSECONDS_IN_MILLISECOND, NANOSECONDS_IN_SECOND},
22};
23use nautilus_model::{
24    data::{
25        Bar, BarType, BookOrder, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate,
26        OrderBookDelta, OrderBookDeltas, OrderBookDepth10, QuoteTick, TradeTick,
27        depth::DEPTH10_LEN, greeks::OptionGreekValues, option_chain::OptionGreeks,
28    },
29    enums::{AggressorSide, BarAggregation, BookAction, GreeksConvention, OrderSide, RecordFlag},
30    identifiers::{InstrumentId, TradeId},
31    types::{Price, Quantity},
32};
33use rust_decimal::prelude::ToPrimitive;
34use ustr::Ustr;
35
36use super::messages::{
37    DeriveOrderbookData, DeriveOrderbookLevel, DeriveOrderbookMsg, DerivePublicWsData,
38    DeriveTickerData, DeriveTickerMsg, DeriveTradesMsg, WsSubscriptionPayload,
39};
40use crate::{
41    common::{enums::DeriveOrderSide, parse::format_instrument_id},
42    http::models::{
43        DerivePublicCandle, DerivePublicFundingRate, DerivePublicTrade, DeriveTickerSnapshot,
44    },
45};
46
47/// Parses a Derive public subscription payload into a typed market data update.
48///
49/// # Errors
50///
51/// Returns an error when the channel is unsupported or `params.data` does not
52/// match the channel payload shape.
53pub fn parse_public_ws_data(payload: &WsSubscriptionPayload) -> anyhow::Result<DerivePublicWsData> {
54    let channel = payload.channel.as_str();
55
56    if channel.starts_with("orderbook.") {
57        return parse_orderbook_msg(payload).map(DerivePublicWsData::Orderbook);
58    }
59
60    if channel.starts_with("trades.") {
61        return parse_trades_msg(payload).map(DerivePublicWsData::Trades);
62    }
63
64    if channel.starts_with("ticker_slim.") || channel.starts_with("ticker.") {
65        return parse_ticker_msg(payload).map(|msg| DerivePublicWsData::Ticker(Box::new(msg)));
66    }
67
68    anyhow::bail!("unsupported Derive public WS channel `{}`", payload.channel)
69}
70
71/// Parses an order book subscription payload.
72///
73/// # Errors
74///
75/// Returns an error when `params.data` is not a Derive order book snapshot.
76pub fn parse_orderbook_msg(payload: &WsSubscriptionPayload) -> anyhow::Result<DeriveOrderbookMsg> {
77    let data = serde_json::from_str::<DeriveOrderbookData>(payload.data.get())
78        .context("failed to decode Derive orderbook data")?;
79    Ok(DeriveOrderbookMsg {
80        channel: Ustr::from(payload.channel.as_str()),
81        data,
82    })
83}
84
85/// Parses a public trades subscription payload.
86///
87/// # Errors
88///
89/// Returns an error when `params.data` is not a list of Derive public trades.
90pub fn parse_trades_msg(payload: &WsSubscriptionPayload) -> anyhow::Result<DeriveTradesMsg> {
91    let trades = serde_json::from_str::<Vec<DerivePublicTrade>>(payload.data.get())
92        .context("failed to decode Derive trades data")?;
93    Ok(DeriveTradesMsg {
94        channel: Ustr::from(payload.channel.as_str()),
95        trades,
96    })
97}
98
99/// Parses a ticker subscription payload.
100///
101/// # Errors
102///
103/// Returns an error when `params.data` is not a Derive ticker payload.
104pub fn parse_ticker_msg(payload: &WsSubscriptionPayload) -> anyhow::Result<DeriveTickerMsg> {
105    let mut data = serde_json::from_str::<DeriveTickerData>(payload.data.get())
106        .context("failed to decode Derive ticker data")?;
107    data.apply_channel_context(payload.channel.as_str())
108        .map_err(anyhow::Error::msg)?;
109    Ok(DeriveTickerMsg {
110        channel: Ustr::from(payload.channel.as_str()),
111        data,
112    })
113}
114
115/// Parses an order book snapshot message into Nautilus snapshot deltas.
116///
117/// Derive's grouped order book stream sends a full depth snapshot for the
118/// requested grouping and depth, so the output starts with a clear delta and
119/// marks the last add with `F_LAST`. The payload does not include a change ID,
120/// so this uses the feed timestamp as the snapshot sequence.
121///
122/// Pass price and size precision from the instrument definition rather than
123/// inferring them from the wire values, since Derive may trim trailing zeroes.
124///
125/// # Errors
126///
127/// Returns an error when a price, size, or timestamp cannot be converted.
128pub fn parse_orderbook_deltas(
129    msg: &DeriveOrderbookMsg,
130    price_precision: u8,
131    size_precision: u8,
132    ts_init: UnixNanos,
133) -> anyhow::Result<OrderBookDeltas> {
134    let instrument_id = msg.data.instrument_id();
135    let timestamp =
136        u64::try_from(msg.data.timestamp).context("negative Derive orderbook timestamp")?;
137    let ts_event = timestamp_millis_to_nanos(timestamp, "timestamp")?;
138    let sequence = timestamp;
139    let context = BookDeltaContext {
140        instrument_id,
141        sequence,
142        price_precision,
143        size_precision,
144        ts_event,
145        ts_init,
146    };
147
148    let mut deltas = Vec::with_capacity(1 + msg.data.bids.len() + msg.data.asks.len());
149    let clear_flags = if msg.data.bids.is_empty() && msg.data.asks.is_empty() {
150        RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
151    } else {
152        RecordFlag::F_SNAPSHOT as u8
153    };
154    deltas.push(OrderBookDelta::new_checked(
155        context.instrument_id,
156        BookAction::Clear,
157        BookOrder::default(),
158        clear_flags,
159        context.sequence,
160        context.ts_event,
161        context.ts_init,
162    )?);
163
164    for (idx, level) in msg.data.bids.iter().enumerate() {
165        push_level_delta(&mut deltas, &context, OrderSide::Buy, level, idx as u64)?;
166    }
167
168    let bid_count = msg.data.bids.len();
169    for (idx, level) in msg.data.asks.iter().enumerate() {
170        push_level_delta(
171            &mut deltas,
172            &context,
173            OrderSide::Sell,
174            level,
175            (bid_count + idx) as u64,
176        )?;
177    }
178
179    if let Some(last) = deltas.last_mut() {
180        last.flags |= RecordFlag::F_LAST as u8;
181    }
182
183    OrderBookDeltas::new_checked(context.instrument_id, deltas)
184}
185
186/// Parses an order book snapshot message into a fixed top-10 depth update.
187///
188/// Derive sends snapshots for the requested order book channel. Missing levels
189/// are filled with zero-size orders so the fixed arrays are always populated.
190///
191/// # Errors
192///
193/// Returns an error when a price, size, or timestamp cannot be converted.
194pub fn parse_orderbook_depth10(
195    msg: &DeriveOrderbookMsg,
196    price_precision: u8,
197    size_precision: u8,
198    ts_init: UnixNanos,
199) -> anyhow::Result<OrderBookDepth10> {
200    let instrument_id = msg.data.instrument_id();
201    let timestamp =
202        u64::try_from(msg.data.timestamp).context("negative Derive orderbook timestamp")?;
203    let ts_event = timestamp_millis_to_nanos(timestamp, "timestamp")?;
204
205    let mut bids = [BookOrder::default(); DEPTH10_LEN];
206    let mut asks = [BookOrder::default(); DEPTH10_LEN];
207    let mut bid_counts = [0; DEPTH10_LEN];
208    let mut ask_counts = [0; DEPTH10_LEN];
209
210    fill_depth_side(
211        &mut bids,
212        &mut bid_counts,
213        &msg.data.bids,
214        OrderSide::Buy,
215        price_precision,
216        size_precision,
217    )?;
218    fill_depth_side(
219        &mut asks,
220        &mut ask_counts,
221        &msg.data.asks,
222        OrderSide::Sell,
223        price_precision,
224        size_precision,
225    )?;
226
227    Ok(OrderBookDepth10::new(
228        instrument_id,
229        bids,
230        asks,
231        bid_counts,
232        ask_counts,
233        RecordFlag::F_SNAPSHOT as u8,
234        timestamp,
235        ts_event,
236        ts_init,
237    ))
238}
239
240/// Parses a public trade message into a Nautilus trade tick.
241///
242/// Pass price and size precision from the instrument definition rather than
243/// inferring them from the wire values, since Derive may trim trailing zeroes.
244///
245/// # Errors
246///
247/// Returns an error when price, size, or timestamp conversion fails.
248pub fn parse_trade_tick(
249    trade: &DerivePublicTrade,
250    price_precision: u8,
251    size_precision: u8,
252    ts_init: UnixNanos,
253) -> anyhow::Result<TradeTick> {
254    let instrument_id = format_instrument_id(trade.instrument_name.as_str());
255    let price = Price::from_decimal_dp(trade.trade_price, price_precision)
256        .with_context(|| format!("invalid trade price for {}", trade.instrument_name))?;
257    let size = Quantity::from_decimal_dp(trade.trade_amount, size_precision)
258        .with_context(|| format!("invalid trade amount for {}", trade.instrument_name))?;
259    let aggressor_side = match trade.direction {
260        DeriveOrderSide::Buy => AggressorSide::Buyer,
261        DeriveOrderSide::Sell => AggressorSide::Seller,
262    };
263    let trade_id = TradeId::new(&trade.trade_id);
264    let timestamp = u64::try_from(trade.timestamp).context("negative Derive trade timestamp")?;
265    let ts_event = timestamp_millis_to_nanos(timestamp, "timestamp")?;
266
267    TradeTick::new_checked(
268        instrument_id,
269        price,
270        size,
271        aggressor_side,
272        trade_id,
273        ts_event,
274        ts_init,
275    )
276}
277
278/// Parses a ticker message into a Nautilus top-of-book quote.
279///
280/// Pass price and size precision from the instrument definition rather than
281/// inferring them from the wire values, since Derive may trim trailing zeroes.
282///
283/// # Errors
284///
285/// Returns an error when price, size, or timestamp conversion fails.
286pub fn parse_ticker_quote(
287    msg: &DeriveTickerMsg,
288    price_precision: u8,
289    size_precision: u8,
290    ts_init: UnixNanos,
291) -> anyhow::Result<QuoteTick> {
292    let instrument_id = msg.data.instrument_id();
293    let instrument_name = msg.data.instrument_name().as_str();
294    let bid_price = Price::from_decimal_dp(msg.data.best_bid_price(), price_precision)
295        .with_context(|| format!("invalid bid price for {instrument_name}"))?;
296    let ask_price = Price::from_decimal_dp(msg.data.best_ask_price(), price_precision)
297        .with_context(|| format!("invalid ask price for {instrument_name}"))?;
298    let bid_size = Quantity::from_decimal_dp(msg.data.best_bid_amount(), size_precision)
299        .with_context(|| format!("invalid bid amount for {instrument_name}"))?;
300    let ask_size = Quantity::from_decimal_dp(msg.data.best_ask_amount(), size_precision)
301        .with_context(|| format!("invalid ask amount for {instrument_name}"))?;
302    let timestamp =
303        u64::try_from(msg.data.timestamp()).context("negative Derive ticker timestamp")?;
304    let ts_event = timestamp_millis_to_nanos(timestamp, "timestamp")?;
305
306    QuoteTick::new_checked(
307        instrument_id,
308        bid_price,
309        ask_price,
310        bid_size,
311        ask_size,
312        ts_event,
313        ts_init,
314    )
315}
316
317/// Parses a REST `public/get_tickers` snapshot into a Nautilus top-of-book quote.
318///
319/// # Errors
320///
321/// Returns an error when price, size, or timestamp conversion fails.
322pub fn parse_ticker_quote_from_rest(
323    ticker: &DeriveTickerSnapshot,
324    price_precision: u8,
325    size_precision: u8,
326    ts_init: UnixNanos,
327) -> anyhow::Result<QuoteTick> {
328    let instrument_id = format_instrument_id(ticker.instrument_name.as_str());
329    let instrument_name = ticker.instrument_name.as_str();
330    let bid_price = Price::from_decimal_dp(ticker.best_bid_price, price_precision)
331        .with_context(|| format!("invalid bid price for {instrument_name}"))?;
332    let ask_price = Price::from_decimal_dp(ticker.best_ask_price, price_precision)
333        .with_context(|| format!("invalid ask price for {instrument_name}"))?;
334    let bid_size = Quantity::from_decimal_dp(ticker.best_bid_amount, size_precision)
335        .with_context(|| format!("invalid bid amount for {instrument_name}"))?;
336    let ask_size = Quantity::from_decimal_dp(ticker.best_ask_amount, size_precision)
337        .with_context(|| format!("invalid ask amount for {instrument_name}"))?;
338    let timestamp = u64::try_from(ticker.timestamp).context("negative Derive ticker timestamp")?;
339    let ts_event = timestamp_millis_to_nanos(timestamp, "timestamp")?;
340
341    QuoteTick::new_checked(
342        instrument_id,
343        bid_price,
344        ask_price,
345        bid_size,
346        ask_size,
347        ts_event,
348        ts_init,
349    )
350}
351
352#[derive(Debug, Clone, Copy)]
353struct BookDeltaContext {
354    instrument_id: InstrumentId,
355    sequence: u64,
356    price_precision: u8,
357    size_precision: u8,
358    ts_event: UnixNanos,
359    ts_init: UnixNanos,
360}
361
362fn push_level_delta(
363    deltas: &mut Vec<OrderBookDelta>,
364    context: &BookDeltaContext,
365    side: OrderSide,
366    level: &DeriveOrderbookLevel,
367    order_id: u64,
368) -> anyhow::Result<()> {
369    if level.amount().is_zero() {
370        return Ok(());
371    }
372
373    let price = Price::from_decimal_dp(level.price(), context.price_precision)
374        .context("invalid Derive orderbook price")?;
375    let size = Quantity::from_decimal_dp(level.amount(), context.size_precision)
376        .context("invalid Derive orderbook amount")?;
377    let order = BookOrder::new(side, price, size, order_id);
378    deltas.push(OrderBookDelta::new_checked(
379        context.instrument_id,
380        BookAction::Add,
381        order,
382        RecordFlag::F_SNAPSHOT as u8,
383        context.sequence,
384        context.ts_event,
385        context.ts_init,
386    )?);
387    Ok(())
388}
389
390fn fill_depth_side(
391    orders: &mut [BookOrder; DEPTH10_LEN],
392    counts: &mut [u32; DEPTH10_LEN],
393    levels: &[DeriveOrderbookLevel],
394    side: OrderSide,
395    price_precision: u8,
396    size_precision: u8,
397) -> anyhow::Result<()> {
398    let mut index = 0;
399
400    for level in levels {
401        let price = Price::from_decimal_dp(level.price(), price_precision)
402            .context("invalid Derive orderbook price")?;
403        let size = Quantity::from_decimal_dp(level.amount(), size_precision)
404            .context("invalid Derive orderbook amount")?;
405
406        if size.is_zero() {
407            continue;
408        }
409
410        orders[index] = BookOrder::new(side, price, size, 0);
411        counts[index] = 1;
412        index += 1;
413
414        if index == DEPTH10_LEN {
415            break;
416        }
417    }
418
419    for order in orders.iter_mut().skip(index) {
420        *order = BookOrder::new(
421            side,
422            Price::zero(price_precision),
423            Quantity::zero(size_precision),
424            0,
425        );
426    }
427
428    Ok(())
429}
430
431fn timestamp_millis_to_nanos(value: u64, field: &str) -> anyhow::Result<UnixNanos> {
432    let nanos = value
433        .checked_mul(NANOSECONDS_IN_MILLISECOND)
434        .with_context(|| format!("Derive {field} overflows nanoseconds"))?;
435    Ok(UnixNanos::from(nanos))
436}
437
438fn ticker_ts_event(timestamp_ms: i64) -> anyhow::Result<UnixNanos> {
439    let timestamp = u64::try_from(timestamp_ms).context("negative Derive ticker timestamp")?;
440    timestamp_millis_to_nanos(timestamp, "timestamp")
441}
442
443/// Parses a ticker payload into a [`MarkPriceUpdate`].
444///
445/// # Errors
446///
447/// Returns an error when the ticker timestamp is negative or overflows.
448pub fn parse_mark_price(
449    msg: &DeriveTickerMsg,
450    price_precision: u8,
451    ts_init: UnixNanos,
452) -> anyhow::Result<Option<MarkPriceUpdate>> {
453    let instrument_id = msg.data.instrument_id();
454    let value = Price::from_decimal_dp(msg.data.mark_price(), price_precision)
455        .with_context(|| format!("invalid Derive mark price for {instrument_id}"))?;
456    let ts_event = ticker_ts_event(msg.data.timestamp())?;
457    Ok(Some(MarkPriceUpdate::new(
458        instrument_id,
459        value,
460        ts_event,
461        ts_init,
462    )))
463}
464
465/// Parses a ticker payload into an [`IndexPriceUpdate`].
466///
467/// # Errors
468///
469/// Returns an error when the ticker timestamp is negative or overflows.
470pub fn parse_index_price(
471    msg: &DeriveTickerMsg,
472    price_precision: u8,
473    ts_init: UnixNanos,
474) -> anyhow::Result<Option<IndexPriceUpdate>> {
475    let instrument_id = msg.data.instrument_id();
476    let value = Price::from_decimal_dp(msg.data.index_price(), price_precision)
477        .with_context(|| format!("invalid Derive index price for {instrument_id}"))?;
478    let ts_event = ticker_ts_event(msg.data.timestamp())?;
479    Ok(Some(IndexPriceUpdate::new(
480        instrument_id,
481        value,
482        ts_event,
483        ts_init,
484    )))
485}
486
487/// Parses a perpetual ticker payload into a [`FundingRateUpdate`].
488///
489/// Returns `Ok(None)` when the ticker does not carry funding.
490///
491/// # Errors
492///
493/// Returns an error when the ticker timestamp is negative or overflows.
494pub fn parse_funding_rate(
495    msg: &DeriveTickerMsg,
496    ts_init: UnixNanos,
497) -> anyhow::Result<Option<FundingRateUpdate>> {
498    let Some(rate) = msg.data.funding_rate() else {
499        return Ok(None);
500    };
501    let instrument_id = msg.data.instrument_id();
502    let ts_event = ticker_ts_event(msg.data.timestamp())?;
503    Ok(Some(FundingRateUpdate::new(
504        instrument_id,
505        rate,
506        None,
507        None,
508        ts_event,
509        ts_init,
510    )))
511}
512
513/// Parses a `public/get_funding_rate_history` record into a [`FundingRateUpdate`].
514///
515/// # Errors
516///
517/// Returns an error when the record timestamp is negative or overflows.
518pub fn parse_funding_rate_history_record(
519    record: &DerivePublicFundingRate,
520    instrument_id: InstrumentId,
521    interval: Option<u16>,
522    ts_init: UnixNanos,
523) -> anyhow::Result<FundingRateUpdate> {
524    let ts_event = ticker_ts_event(record.timestamp)?;
525    Ok(FundingRateUpdate::new(
526        instrument_id,
527        record.funding_rate,
528        interval,
529        None,
530        ts_event,
531        ts_init,
532    ))
533}
534
535/// Parses a `public/get_tradingview_chart_data` record into a Nautilus [`Bar`].
536///
537/// Pass price and size precision from the instrument definition rather than
538/// inferring them from the wire values. The Derive `timestamp_bucket` is the
539/// bucket start in UNIX seconds; the returned bar's `ts_event` marks that
540/// start (not the close).
541///
542/// # Errors
543///
544/// Returns an error when price, size, or timestamp conversion fails.
545pub fn parse_candle_record(
546    record: &DerivePublicCandle,
547    bar_type: BarType,
548    price_precision: u8,
549    size_precision: u8,
550    ts_init: UnixNanos,
551) -> anyhow::Result<Bar> {
552    let open = Price::from_decimal_dp(record.open_price, price_precision)
553        .context("invalid Derive candle open price")?;
554    let high = Price::from_decimal_dp(record.high_price, price_precision)
555        .context("invalid Derive candle high price")?;
556    let low = Price::from_decimal_dp(record.low_price, price_precision)
557        .context("invalid Derive candle low price")?;
558    let close = Price::from_decimal_dp(record.close_price, price_precision)
559        .context("invalid Derive candle close price")?;
560    let volume = Quantity::from_decimal_dp(record.volume_contracts, size_precision)
561        .context("invalid Derive candle volume")?;
562    let timestamp =
563        u64::try_from(record.timestamp_bucket).context("negative Derive candle timestamp")?;
564    let ts_event = timestamp_seconds_to_nanos(timestamp, "candle timestamp_bucket")?;
565
566    Bar::new_checked(bar_type, open, high, low, close, volume, ts_event, ts_init)
567        .context("failed to construct Bar from Derive candle record")
568}
569
570/// Maps a Nautilus bar aggregation and step to the Derive `period` enum value
571/// (bucket size in seconds).
572///
573/// Derive supports the following bucket sizes: 60, 300, 900, 1800, 3600,
574/// 14400, 28800, 86400, 604800.
575///
576/// # Errors
577///
578/// Returns an error if the aggregation or step has no Derive equivalent.
579pub fn bar_spec_to_derive_period(aggregation: BarAggregation, step: u64) -> anyhow::Result<u32> {
580    match aggregation {
581        BarAggregation::Minute => match step {
582            1 => Ok(60),
583            5 => Ok(300),
584            15 => Ok(900),
585            30 => Ok(1800),
586            _ => anyhow::bail!(
587                "Derive only supports minute intervals 1, 5, 15, 30 (use HOUR for >= 60)"
588            ),
589        },
590        BarAggregation::Hour => match step {
591            1 => Ok(3600),
592            4 => Ok(14400),
593            8 => Ok(28800),
594            _ => anyhow::bail!("Derive only supports hour intervals 1, 4, 8"),
595        },
596        BarAggregation::Day => {
597            if step != 1 {
598                anyhow::bail!("Derive only supports 1 DAY interval bars");
599            }
600            Ok(86400)
601        }
602        BarAggregation::Week => {
603            if step != 1 {
604                anyhow::bail!("Derive only supports 1 WEEK interval bars");
605            }
606            Ok(604800)
607        }
608        _ => anyhow::bail!("Derive does not support {aggregation:?} bars"),
609    }
610}
611
612fn timestamp_seconds_to_nanos(value: u64, field: &str) -> anyhow::Result<UnixNanos> {
613    let nanos = value
614        .checked_mul(NANOSECONDS_IN_SECOND)
615        .with_context(|| format!("Derive {field} overflows nanoseconds"))?;
616    Ok(UnixNanos::from(nanos))
617}
618
619/// Parses an option ticker payload into [`OptionGreeks`].
620///
621/// Returns `Ok(None)` when the ticker does not carry option pricing.
622///
623/// # Errors
624///
625/// Returns an error when the ticker timestamp is negative or overflows.
626pub fn parse_option_greeks(
627    msg: &DeriveTickerMsg,
628    ts_init: UnixNanos,
629) -> anyhow::Result<Option<OptionGreeks>> {
630    let Some(pricing) = msg.data.option_pricing() else {
631        return Ok(None);
632    };
633    let instrument_id = msg.data.instrument_id();
634    let ts_event = ticker_ts_event(msg.data.timestamp())?;
635    let to_f64 = |label: &str, value: rust_decimal::Decimal| {
636        value
637            .to_f64()
638            .ok_or_else(|| anyhow::anyhow!("Derive {label} cannot be represented as f64"))
639    };
640
641    Ok(Some(OptionGreeks {
642        instrument_id,
643        convention: GreeksConvention::BlackScholes,
644        greeks: OptionGreekValues {
645            delta: to_f64("delta", pricing.delta)?,
646            gamma: to_f64("gamma", pricing.gamma)?,
647            vega: to_f64("vega", pricing.vega)?,
648            theta: to_f64("theta", pricing.theta)?,
649            rho: to_f64("rho", pricing.rho)?,
650        },
651        mark_iv: Some(to_f64("iv", pricing.iv)?),
652        bid_iv: Some(to_f64("bid_iv", pricing.bid_iv)?),
653        ask_iv: Some(to_f64("ask_iv", pricing.ask_iv)?),
654        underlying_price: Some(to_f64("forward_price", pricing.forward_price)?),
655        open_interest: msg
656            .data
657            .stats()
658            .map(|s| to_f64("open_interest", s.open_interest))
659            .transpose()?,
660        ts_event,
661        ts_init,
662    }))
663}
664
665#[cfg(test)]
666mod tests {
667    use std::{path::PathBuf, str::FromStr};
668
669    use nautilus_model::{
670        enums::{AggressorSide, BookAction, OrderSide, RecordFlag},
671        identifiers::{InstrumentId, TradeId},
672        types::{Price, Quantity},
673    };
674    use rstest::rstest;
675    use rust_decimal::Decimal;
676    use serde_json::{Value, json};
677
678    use super::*;
679    use crate::websocket::messages::DeriveWsFrame;
680
681    const PRICE_PRECISION: u8 = 2;
682    const SIZE_PRECISION: u8 = 3;
683    const INVALID_PRECISION: u8 = u8::MAX;
684
685    fn data_path() -> PathBuf {
686        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("test_data")
687    }
688
689    fn load_json(filename: &str) -> Value {
690        let content = std::fs::read_to_string(data_path().join(filename))
691            .unwrap_or_else(|_| panic!("failed to read {filename}"));
692        serde_json::from_str(&content).expect("invalid json")
693    }
694
695    fn subscription_payload(frame: &Value) -> WsSubscriptionPayload {
696        match DeriveWsFrame::parse(&frame.to_string()).unwrap() {
697            DeriveWsFrame::Subscription(payload) => payload,
698            other => panic!("expected subscription frame, was {other:?}"),
699        }
700    }
701
702    fn subscription_data_payload(channel: &str, data: &Value) -> WsSubscriptionPayload {
703        subscription_payload(&json!({
704            "jsonrpc": "2.0",
705            "method": "subscription",
706            "params": {
707                "channel": channel,
708                "data": data
709            }
710        }))
711    }
712
713    fn orderbook_json(timestamp: i64, bids: &Value, asks: &Value) -> Value {
714        let mut value = load_json("perps/ws_orderbook_eth.json");
715        value["timestamp"] = json!(timestamp);
716        value["bids"] = bids.clone();
717        value["asks"] = asks.clone();
718        value
719    }
720
721    fn trade_json(timestamp: i64, direction: &str) -> Value {
722        trade_json_with_values(timestamp, direction, "3500.2", "0.25")
723    }
724
725    fn trade_json_with_values(
726        timestamp: i64,
727        direction: &str,
728        trade_price: &str,
729        trade_amount: &str,
730    ) -> Value {
731        let mut value = load_json("perps/ws_trade_eth.json");
732        value["direction"] = json!(direction);
733        value["timestamp"] = json!(timestamp);
734        value["trade_amount"] = json!(trade_amount);
735        value["trade_id"] = json!("trade-1");
736        value["trade_price"] = json!(trade_price);
737        value
738    }
739
740    fn ticker_json_with_timestamp(timestamp: i64) -> Value {
741        let mut value = load_json("perps/ws_ticker_eth.json");
742        value["best_ask_amount"] = json!("1.20");
743        value["best_ask_price"] = json!("3501.00");
744        value["best_bid_amount"] = json!("0.80");
745        value["best_bid_price"] = json!("3499.50");
746        value["timestamp"] = json!(timestamp);
747        value
748    }
749
750    fn ticker_json() -> Value {
751        ticker_json_with_timestamp(1_700_000_000_000)
752    }
753
754    fn price(value: &str) -> Price {
755        Price::from_decimal_dp(Decimal::from_str(value).unwrap(), PRICE_PRECISION).unwrap()
756    }
757
758    fn quantity(value: &str) -> Quantity {
759        Quantity::from_decimal_dp(Decimal::from_str(value).unwrap(), SIZE_PRECISION).unwrap()
760    }
761
762    #[rstest]
763    fn test_parse_public_orderbook_frame() {
764        let payload = subscription_data_payload(
765            "orderbook.ETH-PERP.1.10",
766            &orderbook_json(
767                1_700_000_000_000,
768                &json!([["3499.50", "1.20"], ["3499.00", "0.40"]]),
769                &json!([["3501.00", "0.80"]]),
770            ),
771        );
772
773        let msg = parse_orderbook_msg(&payload).unwrap();
774        let deltas =
775            parse_orderbook_deltas(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(123))
776                .unwrap();
777
778        assert_eq!(msg.channel.as_str(), "orderbook.ETH-PERP.1.10");
779        assert_eq!(
780            msg.data.instrument_id(),
781            InstrumentId::from("ETH-PERP.DERIVE")
782        );
783        assert_eq!(msg.data.bids[0].price().to_string(), "3499.50");
784        assert_eq!(deltas.instrument_id, InstrumentId::from("ETH-PERP.DERIVE"));
785        assert_eq!(deltas.deltas.len(), 4);
786        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
787        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy);
788        assert_eq!(deltas.deltas[1].order.price, price("3499.50"));
789        assert_eq!(deltas.deltas[1].order.size, quantity("1.20"));
790        assert_eq!(deltas.deltas[3].order.side, OrderSide::Sell);
791        assert_eq!(
792            deltas.deltas[3].flags,
793            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
794        );
795    }
796
797    #[rstest]
798    fn test_parse_public_trades_frame() {
799        let payload = subscription_data_payload(
800            "trades.perp.ETH",
801            &json!([trade_json(1_700_000_000_001, "buy")]),
802        );
803
804        let msg = parse_trades_msg(&payload).unwrap();
805        let tick = parse_trade_tick(
806            &msg.trades[0],
807            PRICE_PRECISION,
808            SIZE_PRECISION,
809            UnixNanos::from(456),
810        )
811        .unwrap();
812
813        assert_eq!(msg.channel.as_str(), "trades.perp.ETH");
814        assert_eq!(msg.trades.len(), 1);
815        assert_eq!(
816            format_instrument_id(msg.trades[0].instrument_name.as_str()),
817            InstrumentId::from("ETH-PERP.DERIVE")
818        );
819        assert_eq!(tick.instrument_id, InstrumentId::from("ETH-PERP.DERIVE"));
820        assert_eq!(tick.price, price("3500.2"));
821        assert_eq!(tick.size, quantity("0.25"));
822        assert_eq!(tick.aggressor_side, AggressorSide::Buyer);
823        assert_eq!(tick.trade_id, TradeId::from("trade-1"));
824        assert_eq!(tick.ts_event, UnixNanos::from(1_700_000_000_001_000_000));
825    }
826
827    #[rstest]
828    fn test_parse_public_ticker_frame() {
829        let payload = subscription_data_payload(
830            "ticker_slim.ETH-PERP.1000",
831            &load_json("perps/ws_ticker_slim_eth.json"),
832        );
833
834        let msg = parse_ticker_msg(&payload).unwrap();
835        let quote = parse_ticker_quote(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(789))
836            .unwrap();
837
838        assert_eq!(msg.channel.as_str(), "ticker_slim.ETH-PERP.1000");
839        assert_eq!(
840            msg.data.instrument_id(),
841            InstrumentId::from("ETH-PERP.DERIVE")
842        );
843        assert_eq!(msg.data.timestamp(), 1_779_953_796_714);
844        assert_eq!(quote.instrument_id, InstrumentId::from("ETH-PERP.DERIVE"));
845        assert_eq!(quote.bid_price, price("1992.36"));
846        assert_eq!(quote.ask_price, price("1992.37"));
847        assert_eq!(quote.bid_size, quantity("1.505"));
848        assert_eq!(quote.ask_size, quantity("1.505"));
849        assert_eq!(quote.ts_event, UnixNanos::from(1_779_953_796_714_000_000));
850    }
851
852    #[rstest]
853    fn test_parse_spot_orderbook_frame() {
854        let mut data = load_json("spot/ws_orderbook_eth.json");
855        data["bids"] = json!([["2050.0", "1.20"], ["2049.5", "0.40"]]);
856        data["asks"] = json!([["2051.0", "0.80"]]);
857        let payload = subscription_data_payload("orderbook.ETH-USDC.1.10", &data);
858
859        let msg = parse_orderbook_msg(&payload).unwrap();
860        let deltas =
861            parse_orderbook_deltas(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(123))
862                .unwrap();
863
864        assert_eq!(msg.channel.as_str(), "orderbook.ETH-USDC.1.10");
865        assert_eq!(
866            msg.data.instrument_id(),
867            InstrumentId::from("ETH-USDC.DERIVE")
868        );
869        assert_eq!(deltas.instrument_id, InstrumentId::from("ETH-USDC.DERIVE"));
870        assert_eq!(deltas.deltas.len(), 4);
871        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
872        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy);
873        assert_eq!(deltas.deltas[1].order.price, price("2050.0"));
874        assert_eq!(deltas.deltas[1].order.size, quantity("1.20"));
875        assert_eq!(deltas.deltas[3].order.side, OrderSide::Sell);
876        assert_eq!(
877            deltas.deltas[3].flags,
878            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
879        );
880    }
881
882    #[rstest]
883    fn test_parse_spot_trades_frame() {
884        let payload = subscription_data_payload(
885            "trades.erc20.ETH",
886            &json!([load_json("spot/ws_trade_eth.json")]),
887        );
888
889        let msg = parse_trades_msg(&payload).unwrap();
890        let tick = parse_trade_tick(
891            &msg.trades[0],
892            PRICE_PRECISION,
893            SIZE_PRECISION,
894            UnixNanos::from(456),
895        )
896        .unwrap();
897
898        assert_eq!(msg.channel.as_str(), "trades.erc20.ETH");
899        assert_eq!(msg.trades.len(), 1);
900        assert_eq!(tick.instrument_id, InstrumentId::from("ETH-USDC.DERIVE"));
901        assert_eq!(tick.price, price("2050"));
902        assert_eq!(tick.size, quantity("0.1"));
903        assert_eq!(tick.aggressor_side, AggressorSide::Seller);
904        assert_eq!(
905            tick.trade_id,
906            TradeId::from("0445f96a-10fb-4fdc-a0f9-eed94a2f32e1")
907        );
908    }
909
910    #[rstest]
911    fn test_parse_spot_ticker_slim_frame_handles_null_funding() {
912        let payload = subscription_data_payload(
913            "ticker_slim.ETH-USDC.1000",
914            &load_json("spot/ws_ticker_slim_eth.json"),
915        );
916
917        let msg = parse_ticker_msg(&payload).unwrap();
918        let quote = parse_ticker_quote(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(789))
919            .unwrap();
920
921        assert_eq!(msg.channel.as_str(), "ticker_slim.ETH-USDC.1000");
922        assert_eq!(
923            msg.data.instrument_id(),
924            InstrumentId::from("ETH-USDC.DERIVE")
925        );
926        assert_eq!(quote.instrument_id, InstrumentId::from("ETH-USDC.DERIVE"));
927
928        assert!(
929            parse_funding_rate(&msg, UnixNanos::from(789))
930                .unwrap()
931                .is_none()
932        );
933        let mark = parse_mark_price(&msg, PRICE_PRECISION, UnixNanos::from(789))
934            .unwrap()
935            .expect("spot slim ticker carries mark price");
936        let index = parse_index_price(&msg, PRICE_PRECISION, UnixNanos::from(789))
937            .unwrap()
938            .expect("spot slim ticker carries index price");
939        assert_eq!(mark.instrument_id, InstrumentId::from("ETH-USDC.DERIVE"));
940        assert_eq!(index.instrument_id, InstrumentId::from("ETH-USDC.DERIVE"));
941    }
942
943    #[rstest]
944    fn test_parse_public_ticker_direct_payload() {
945        let payload = subscription_data_payload(
946            "ticker.ETH-PERP.1000",
947            &ticker_json_with_timestamp(1_700_000_000_011),
948        );
949
950        let msg = parse_ticker_msg(&payload).unwrap();
951        let quote = parse_ticker_quote(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(790))
952            .unwrap();
953
954        assert_eq!(msg.channel.as_str(), "ticker.ETH-PERP.1000");
955        assert_eq!(msg.data.timestamp(), 1_700_000_000_011);
956        assert_eq!(
957            msg.data.instrument_id(),
958            InstrumentId::from("ETH-PERP.DERIVE")
959        );
960        assert_eq!(quote.instrument_id, InstrumentId::from("ETH-PERP.DERIVE"));
961        assert_eq!(quote.ts_event, UnixNanos::from(1_700_000_000_011_000_000));
962    }
963
964    #[rstest]
965    fn test_parse_ticker_quote_uses_supplied_precision_when_wire_scale_varies() {
966        let mut ticker = ticker_json_with_timestamp(1_700_000_000_012);
967        ticker["best_bid_price"] = json!("3500");
968        ticker["best_ask_price"] = json!("3501");
969        ticker["best_bid_amount"] = json!("1");
970        ticker["best_ask_amount"] = json!("2");
971        let payload = subscription_data_payload("ticker.ETH-PERP.1000", &ticker);
972
973        let msg = parse_ticker_msg(&payload).unwrap();
974        let quote = parse_ticker_quote(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(790))
975            .unwrap();
976
977        assert_eq!(quote.bid_price, price("3500"));
978        assert_eq!(quote.ask_price, price("3501"));
979        assert_eq!(quote.bid_size, quantity("1"));
980        assert_eq!(quote.ask_size, quantity("2"));
981        assert_eq!(quote.bid_price.precision, PRICE_PRECISION);
982        assert_eq!(quote.bid_size.precision, SIZE_PRECISION);
983    }
984
985    #[rstest]
986    fn test_parse_ticker_quote_from_rest_emits_quote() {
987        let ticker: DeriveTickerSnapshot =
988            serde_json::from_value(ticker_json_with_timestamp(1_700_000_000_013)).unwrap();
989
990        let quote = parse_ticker_quote_from_rest(
991            &ticker,
992            PRICE_PRECISION,
993            SIZE_PRECISION,
994            UnixNanos::from(791),
995        )
996        .unwrap();
997
998        assert_eq!(quote.instrument_id, InstrumentId::from("ETH-PERP.DERIVE"));
999        assert_eq!(quote.bid_price, price("3499.50"));
1000        assert_eq!(quote.ask_price, price("3501.00"));
1001        assert_eq!(quote.bid_size, quantity("0.80"));
1002        assert_eq!(quote.ask_size, quantity("1.20"));
1003        assert_eq!(quote.ts_event, UnixNanos::from(1_700_000_000_013_000_000));
1004    }
1005
1006    #[rstest]
1007    fn test_parse_ticker_quote_from_rest_rejects_negative_timestamp() {
1008        let mut value = ticker_json_with_timestamp(1_700_000_000_013);
1009        value["timestamp"] = json!(-1_i64);
1010        let ticker: DeriveTickerSnapshot = serde_json::from_value(value).unwrap();
1011
1012        let err = parse_ticker_quote_from_rest(
1013            &ticker,
1014            PRICE_PRECISION,
1015            SIZE_PRECISION,
1016            UnixNanos::from(791),
1017        )
1018        .expect_err("must reject negative timestamp");
1019        assert!(err.to_string().contains("negative Derive ticker timestamp"));
1020    }
1021
1022    #[rstest]
1023    fn test_parse_orderbook_deltas_empty_book_marks_clear_last() {
1024        let payload = subscription_data_payload(
1025            "orderbook.ETH-PERP.1.10",
1026            &orderbook_json(1_700_000_000_000, &json!([]), &json!([])),
1027        );
1028
1029        let msg = parse_orderbook_msg(&payload).unwrap();
1030        let deltas =
1031            parse_orderbook_deltas(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(123))
1032                .unwrap();
1033
1034        assert_eq!(deltas.deltas.len(), 1);
1035        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
1036        assert_eq!(
1037            deltas.deltas[0].flags,
1038            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1039        );
1040    }
1041
1042    #[rstest]
1043    fn test_parse_orderbook_deltas_skips_zero_size_levels() {
1044        let payload = subscription_data_payload(
1045            "orderbook.ETH-PERP.1.10",
1046            &orderbook_json(
1047                1_700_000_000_000,
1048                &json!([["3499.50", "0"], ["3499.00", "0.40"]]),
1049                &json!([["3501.00", "0"]]),
1050            ),
1051        );
1052
1053        let msg = parse_orderbook_msg(&payload).unwrap();
1054        let deltas =
1055            parse_orderbook_deltas(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(123))
1056                .unwrap();
1057
1058        assert_eq!(deltas.deltas.len(), 2);
1059        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy);
1060        assert_eq!(deltas.deltas[1].order.price, price("3499.00"));
1061        assert_eq!(deltas.deltas[1].order.size, quantity("0.40"));
1062        assert_eq!(deltas.deltas[1].order.order_id, 1);
1063        assert_eq!(
1064            deltas.deltas[1].flags,
1065            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1066        );
1067    }
1068
1069    #[rstest]
1070    fn test_parse_orderbook_deltas_uses_supplied_precision_when_wire_scale_varies() {
1071        let payload = subscription_data_payload(
1072            "orderbook.ETH-PERP.1.10",
1073            &orderbook_json(
1074                1_700_000_000_000,
1075                &json!([["3500", "1"]]),
1076                &json!([["3501", "2"]]),
1077            ),
1078        );
1079
1080        let msg = parse_orderbook_msg(&payload).unwrap();
1081        let deltas =
1082            parse_orderbook_deltas(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(123))
1083                .unwrap();
1084
1085        assert_eq!(deltas.deltas[1].order.price, price("3500"));
1086        assert_eq!(deltas.deltas[1].order.size, quantity("1"));
1087        assert_eq!(deltas.deltas[2].order.price, price("3501"));
1088        assert_eq!(deltas.deltas[2].order.size, quantity("2"));
1089        assert_eq!(deltas.deltas[1].order.price.precision, PRICE_PRECISION);
1090        assert_eq!(deltas.deltas[1].order.size.precision, SIZE_PRECISION);
1091    }
1092
1093    #[rstest]
1094    fn test_parse_orderbook_depth10_skips_zero_sizes_caps_and_zero_fills() {
1095        let bids = Value::Array(
1096            (0..12)
1097                .map(|i| {
1098                    let size = if i == 1 { "0" } else { "1" };
1099                    json!([format!("{}", 3500 - i), size])
1100                })
1101                .collect(),
1102        );
1103        let asks = json!([["3501", "2"], ["3502", "0"], ["3503", "3"]]);
1104        let payload = subscription_data_payload(
1105            "orderbook.ETH-PERP.1.10",
1106            &orderbook_json(1_700_000_000_000, &bids, &asks),
1107        );
1108
1109        let msg = parse_orderbook_msg(&payload).unwrap();
1110        let depth =
1111            parse_orderbook_depth10(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(123))
1112                .unwrap();
1113
1114        assert_eq!(depth.instrument_id, InstrumentId::from("ETH-PERP.DERIVE"));
1115        assert_eq!(depth.bids[0].price, price("3500"));
1116        assert_eq!(depth.bids[1].price, price("3498"));
1117        assert_eq!(depth.bids[9].price, price("3490"));
1118        assert_eq!(depth.bid_counts[0], 1);
1119        assert_eq!(depth.bid_counts[9], 1);
1120        assert_eq!(depth.asks[0].price, price("3501"));
1121        assert_eq!(depth.asks[1].price, price("3503"));
1122        assert_eq!(depth.asks[2].price, Price::zero(PRICE_PRECISION));
1123        assert_eq!(depth.asks[2].size, Quantity::zero(SIZE_PRECISION));
1124        assert_eq!(depth.ask_counts[0], 1);
1125        assert_eq!(depth.ask_counts[1], 1);
1126        assert_eq!(depth.ask_counts[2], 0);
1127        assert_eq!(depth.sequence, 1_700_000_000_000);
1128        assert_eq!(depth.flags, RecordFlag::F_SNAPSHOT as u8);
1129        assert_eq!(depth.ts_event, UnixNanos::from(1_700_000_000_000_000_000));
1130    }
1131
1132    #[rstest]
1133    fn test_parse_trade_tick_maps_sell_direction() {
1134        let payload = subscription_data_payload(
1135            "trades.perp.ETH",
1136            &json!([trade_json(1_700_000_000_001, "sell")]),
1137        );
1138
1139        let msg = parse_trades_msg(&payload).unwrap();
1140        let tick = parse_trade_tick(
1141            &msg.trades[0],
1142            PRICE_PRECISION,
1143            SIZE_PRECISION,
1144            UnixNanos::from(456),
1145        )
1146        .unwrap();
1147
1148        assert_eq!(tick.aggressor_side, AggressorSide::Seller);
1149    }
1150
1151    #[rstest]
1152    fn test_parse_trade_tick_uses_supplied_precision_when_wire_scale_varies() {
1153        let payload = subscription_data_payload(
1154            "trades.perp.ETH",
1155            &json!([trade_json_with_values(
1156                1_700_000_000_001,
1157                "buy",
1158                "3500",
1159                "1"
1160            )]),
1161        );
1162
1163        let msg = parse_trades_msg(&payload).unwrap();
1164        let tick = parse_trade_tick(
1165            &msg.trades[0],
1166            PRICE_PRECISION,
1167            SIZE_PRECISION,
1168            UnixNanos::from(456),
1169        )
1170        .unwrap();
1171
1172        assert_eq!(tick.price, price("3500"));
1173        assert_eq!(tick.size, quantity("1"));
1174        assert_eq!(tick.price.precision, PRICE_PRECISION);
1175        assert_eq!(tick.size.precision, SIZE_PRECISION);
1176    }
1177
1178    #[rstest]
1179    fn test_parse_public_ws_data_dispatches_orderbook_channel() {
1180        let payload = subscription_data_payload(
1181            "orderbook.ETH-PERP.1.10",
1182            &orderbook_json(1_700_000_000_000, &json!([]), &json!([])),
1183        );
1184
1185        let parsed = parse_public_ws_data(&payload).unwrap();
1186
1187        match parsed {
1188            DerivePublicWsData::Orderbook(msg) => {
1189                assert_eq!(msg.channel.as_str(), "orderbook.ETH-PERP.1.10");
1190                assert_eq!(
1191                    msg.data.instrument_id(),
1192                    InstrumentId::from("ETH-PERP.DERIVE")
1193                );
1194            }
1195            other => panic!("expected orderbook data, was {other:?}"),
1196        }
1197    }
1198
1199    #[rstest]
1200    fn test_parse_public_ws_data_dispatches_trades_channel() {
1201        let payload = subscription_data_payload("trades.perp.ETH", &json!([]));
1202
1203        let parsed = parse_public_ws_data(&payload).unwrap();
1204
1205        match parsed {
1206            DerivePublicWsData::Trades(msg) => assert!(msg.trades.is_empty()),
1207            other => panic!("expected trades data, was {other:?}"),
1208        }
1209    }
1210
1211    #[rstest]
1212    fn test_parse_public_ws_data_dispatches_ticker_channel() {
1213        let payload = subscription_data_payload(
1214            "ticker_slim.ETH-PERP.1000",
1215            &load_json("perps/ws_ticker_slim_eth.json"),
1216        );
1217
1218        let parsed = parse_public_ws_data(&payload).unwrap();
1219
1220        match parsed {
1221            DerivePublicWsData::Ticker(msg) => {
1222                assert_eq!(msg.channel.as_str(), "ticker_slim.ETH-PERP.1000");
1223                assert_eq!(
1224                    msg.data.instrument_id(),
1225                    InstrumentId::from("ETH-PERP.DERIVE")
1226                );
1227            }
1228            other => panic!("expected ticker data, was {other:?}"),
1229        }
1230    }
1231
1232    #[rstest]
1233    fn test_parse_orderbook_msg_rejects_malformed_payload() {
1234        let payload = subscription_data_payload(
1235            "orderbook.ETH-PERP.1.10",
1236            &json!({
1237                "instrument_name": "ETH-PERP",
1238                "timestamp": 1_700_000_000_000_i64,
1239                "bids": []
1240            }),
1241        );
1242
1243        let err = parse_orderbook_msg(&payload).expect_err("must reject malformed orderbook");
1244
1245        assert!(
1246            err.to_string()
1247                .contains("failed to decode Derive orderbook data")
1248        );
1249    }
1250
1251    #[rstest]
1252    fn test_parse_trades_msg_rejects_malformed_payload() {
1253        let payload = subscription_data_payload("trades.perp.ETH", &json!({}));
1254
1255        let err = parse_trades_msg(&payload).expect_err("must reject malformed trades");
1256
1257        assert!(
1258            err.to_string()
1259                .contains("failed to decode Derive trades data")
1260        );
1261    }
1262
1263    #[rstest]
1264    fn test_parse_ticker_msg_rejects_malformed_payload() {
1265        let payload = subscription_data_payload(
1266            "ticker.ETH-PERP.1000",
1267            &json!({
1268                "timestamp": 1_700_000_000_010_i64
1269            }),
1270        );
1271
1272        let err = parse_ticker_msg(&payload).expect_err("must reject malformed ticker");
1273
1274        assert!(
1275            err.to_string()
1276                .contains("failed to decode Derive ticker data")
1277        );
1278    }
1279
1280    #[rstest]
1281    #[case("ticker_slim.ETH-PERP")]
1282    #[case("ticker_slim..1000")]
1283    fn test_parse_ticker_msg_rejects_malformed_slim_channel(#[case] channel: &str) {
1284        let payload =
1285            subscription_data_payload(channel, &load_json("perps/ws_ticker_slim_eth.json"));
1286
1287        let err = parse_ticker_msg(&payload).expect_err("must reject malformed slim channel");
1288
1289        assert!(err.to_string().contains("invalid Derive ticker channel"));
1290    }
1291
1292    #[rstest]
1293    fn test_parse_orderbook_deltas_rejects_negative_timestamp() {
1294        let payload = subscription_data_payload(
1295            "orderbook.ETH-PERP.1.10",
1296            &orderbook_json(-1, &json!([]), &json!([])),
1297        );
1298
1299        let msg = parse_orderbook_msg(&payload).unwrap();
1300        let err =
1301            parse_orderbook_deltas(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(123))
1302                .expect_err("must reject negative orderbook timestamp");
1303
1304        assert!(
1305            err.to_string()
1306                .contains("negative Derive orderbook timestamp")
1307        );
1308    }
1309
1310    #[rstest]
1311    fn test_parse_orderbook_deltas_rejects_timestamp_overflow() {
1312        let payload = subscription_data_payload(
1313            "orderbook.ETH-PERP.1.10",
1314            &orderbook_json(i64::MAX, &json!([]), &json!([])),
1315        );
1316
1317        let msg = parse_orderbook_msg(&payload).unwrap();
1318        let err =
1319            parse_orderbook_deltas(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(123))
1320                .expect_err("must reject overflowing orderbook timestamp");
1321
1322        assert!(
1323            err.to_string()
1324                .contains("Derive timestamp overflows nanoseconds")
1325        );
1326    }
1327
1328    #[rstest]
1329    fn test_parse_orderbook_deltas_rejects_invalid_size_precision() {
1330        let payload = subscription_data_payload(
1331            "orderbook.ETH-PERP.1.10",
1332            &orderbook_json(
1333                1_700_000_000_000,
1334                &json!([["3500", "1"]]),
1335                &json!([["3501", "2"]]),
1336            ),
1337        );
1338
1339        let msg = parse_orderbook_msg(&payload).unwrap();
1340        let err = parse_orderbook_deltas(
1341            &msg,
1342            PRICE_PRECISION,
1343            INVALID_PRECISION,
1344            UnixNanos::from(123),
1345        )
1346        .expect_err("must reject invalid orderbook size precision");
1347
1348        assert!(err.to_string().contains("invalid Derive orderbook amount"));
1349    }
1350
1351    #[rstest]
1352    fn test_parse_trade_tick_rejects_negative_timestamp() {
1353        let payload = subscription_data_payload("trades.perp.ETH", &json!([trade_json(-1, "buy")]));
1354
1355        let msg = parse_trades_msg(&payload).unwrap();
1356        let err = parse_trade_tick(
1357            &msg.trades[0],
1358            PRICE_PRECISION,
1359            SIZE_PRECISION,
1360            UnixNanos::from(456),
1361        )
1362        .expect_err("must reject negative trade timestamp");
1363
1364        assert!(err.to_string().contains("negative Derive trade timestamp"));
1365    }
1366
1367    #[rstest]
1368    fn test_parse_trade_tick_rejects_timestamp_overflow() {
1369        let payload =
1370            subscription_data_payload("trades.perp.ETH", &json!([trade_json(i64::MAX, "buy")]));
1371
1372        let msg = parse_trades_msg(&payload).unwrap();
1373        let err = parse_trade_tick(
1374            &msg.trades[0],
1375            PRICE_PRECISION,
1376            SIZE_PRECISION,
1377            UnixNanos::from(456),
1378        )
1379        .expect_err("must reject overflowing trade timestamp");
1380
1381        assert!(
1382            err.to_string()
1383                .contains("Derive timestamp overflows nanoseconds")
1384        );
1385    }
1386
1387    #[rstest]
1388    fn test_parse_trade_tick_rejects_invalid_price_precision() {
1389        let payload = subscription_data_payload(
1390            "trades.perp.ETH",
1391            &json!([trade_json(1_700_000_000_001, "buy")]),
1392        );
1393
1394        let msg = parse_trades_msg(&payload).unwrap();
1395        let err = parse_trade_tick(
1396            &msg.trades[0],
1397            INVALID_PRECISION,
1398            SIZE_PRECISION,
1399            UnixNanos::from(456),
1400        )
1401        .expect_err("must reject invalid trade price precision");
1402
1403        assert!(err.to_string().contains("invalid trade price for ETH-PERP"));
1404    }
1405
1406    #[rstest]
1407    fn test_parse_ticker_quote_rejects_negative_timestamp() {
1408        let payload = subscription_data_payload(
1409            "ticker.ETH-PERP.1000",
1410            &json!({
1411                "timestamp": -1_i64,
1412                "instrument_ticker": ticker_json()
1413            }),
1414        );
1415
1416        let msg = parse_ticker_msg(&payload).unwrap();
1417        let err = parse_ticker_quote(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(789))
1418            .expect_err("must reject negative ticker timestamp");
1419
1420        assert!(err.to_string().contains("negative Derive ticker timestamp"));
1421    }
1422
1423    #[rstest]
1424    fn test_parse_ticker_quote_rejects_timestamp_overflow() {
1425        let payload = subscription_data_payload(
1426            "ticker.ETH-PERP.1000",
1427            &json!({
1428                "timestamp": i64::MAX,
1429                "instrument_ticker": ticker_json()
1430            }),
1431        );
1432
1433        let msg = parse_ticker_msg(&payload).unwrap();
1434        let err = parse_ticker_quote(&msg, PRICE_PRECISION, SIZE_PRECISION, UnixNanos::from(789))
1435            .expect_err("must reject overflowing ticker timestamp");
1436
1437        assert!(
1438            err.to_string()
1439                .contains("Derive timestamp overflows nanoseconds")
1440        );
1441    }
1442
1443    #[rstest]
1444    fn test_parse_public_ws_data_rejects_unknown_channel() {
1445        let payload = WsSubscriptionPayload {
1446            channel: Ustr::from("wallet.ETH"),
1447            data: serde_json::value::to_raw_value(&json!({})).unwrap(),
1448        };
1449
1450        let err = parse_public_ws_data(&payload).expect_err("must reject unknown channel");
1451
1452        assert!(
1453            err.to_string()
1454                .contains("unsupported Derive public WS channel")
1455        );
1456    }
1457
1458    fn option_ticker_json(timestamp: i64) -> Value {
1459        let mut value = load_json("options/http_ticker_eth_snapshot.json");
1460        value["timestamp"] = json!(timestamp);
1461        value
1462    }
1463
1464    fn perp_envelope_payload(timestamp: i64) -> WsSubscriptionPayload {
1465        subscription_data_payload(
1466            "ticker.ETH-PERP.1000",
1467            &json!({
1468                "timestamp": timestamp,
1469                "instrument_ticker": ticker_json_with_timestamp(timestamp),
1470            }),
1471        )
1472    }
1473
1474    fn option_envelope_payload(timestamp: i64) -> WsSubscriptionPayload {
1475        let mut option_data = option_ticker_json(timestamp);
1476        option_data["instrument_name"] = json!("ETH-20260627-3500-C");
1477        subscription_data_payload(
1478            "ticker.ETH-20260627-3500-C.1000",
1479            &json!({
1480                "timestamp": timestamp,
1481                "instrument_ticker": option_data,
1482            }),
1483        )
1484    }
1485
1486    fn slim_payload() -> WsSubscriptionPayload {
1487        subscription_data_payload(
1488            "ticker_slim.ETH-PERP.1000",
1489            &load_json("perps/ws_ticker_slim_eth.json"),
1490        )
1491    }
1492
1493    #[rstest]
1494    fn test_parse_mark_price_maps_slim_variant() {
1495        let msg = parse_ticker_msg(&slim_payload()).unwrap();
1496
1497        let update = parse_mark_price(&msg, PRICE_PRECISION, UnixNanos::from(789))
1498            .unwrap()
1499            .expect("slim ticker carries mark price");
1500
1501        assert_eq!(update.instrument_id, InstrumentId::from("ETH-PERP.DERIVE"));
1502        assert_eq!(update.value, price("1992.49"));
1503        assert_eq!(update.ts_event, UnixNanos::from(1_779_953_796_714_000_000));
1504        assert_eq!(update.ts_init, UnixNanos::from(789));
1505    }
1506
1507    #[rstest]
1508    fn test_parse_index_price_maps_slim_variant() {
1509        let msg = parse_ticker_msg(&slim_payload()).unwrap();
1510
1511        let update = parse_index_price(&msg, PRICE_PRECISION, UnixNanos::from(789))
1512            .unwrap()
1513            .expect("slim ticker carries index price");
1514
1515        assert_eq!(update.instrument_id, InstrumentId::from("ETH-PERP.DERIVE"));
1516        assert_eq!(update.value, price("1991.79"));
1517        assert_eq!(update.ts_event, UnixNanos::from(1_779_953_796_714_000_000));
1518        assert_eq!(update.ts_init, UnixNanos::from(789));
1519    }
1520
1521    #[rstest]
1522    fn test_parse_funding_rate_maps_slim_variant() {
1523        let msg = parse_ticker_msg(&slim_payload()).unwrap();
1524
1525        let update = parse_funding_rate(&msg, UnixNanos::from(789))
1526            .unwrap()
1527            .expect("slim ticker carries perp funding");
1528
1529        assert_eq!(update.instrument_id, InstrumentId::from("ETH-PERP.DERIVE"));
1530        assert_eq!(update.rate, Decimal::from_str("0.000012500").unwrap());
1531        assert_eq!(update.ts_event, UnixNanos::from(1_779_953_796_714_000_000));
1532        assert_eq!(update.ts_init, UnixNanos::from(789));
1533    }
1534
1535    #[rstest]
1536    fn test_parse_option_greeks_returns_none_for_slim_variant_without_option_pricing() {
1537        let msg = parse_ticker_msg(&slim_payload()).unwrap();
1538
1539        let result = parse_option_greeks(&msg, UnixNanos::from(789)).unwrap();
1540
1541        assert!(result.is_none());
1542    }
1543
1544    fn option_slim_payload(filename: &str, instrument_name: &str) -> WsSubscriptionPayload {
1545        subscription_data_payload(
1546            &format!("ticker_slim.{instrument_name}.1000"),
1547            &load_json(filename),
1548        )
1549    }
1550
1551    #[rstest]
1552    fn test_parse_option_greeks_maps_slim_variant() {
1553        let msg = parse_ticker_msg(&option_slim_payload(
1554            "options/ws_ticker_slim_eth_call.json",
1555            "ETH-20260612-1600-C",
1556        ))
1557        .unwrap();
1558
1559        let greeks = parse_option_greeks(&msg, UnixNanos::from(789))
1560            .unwrap()
1561            .expect("slim ticker carries option pricing");
1562
1563        assert_eq!(
1564            greeks.instrument_id,
1565            InstrumentId::from("ETH-20260612-1600-C.DERIVE")
1566        );
1567        assert_eq!(greeks.convention, GreeksConvention::BlackScholes);
1568        assert!((greeks.greeks.delta - 0.95222).abs() < 1e-9);
1569        assert!((greeks.greeks.gamma - 0.00036344).abs() < 1e-9);
1570        assert_eq!(greeks.mark_iv, Some(0.67698));
1571        assert_eq!(greeks.bid_iv, Some(0.0));
1572        assert_eq!(greeks.ask_iv, Some(0.88815));
1573        assert_eq!(greeks.underlying_price, Some(1992.6));
1574        assert_eq!(greeks.open_interest, Some(0.0));
1575        assert_eq!(greeks.ts_event, UnixNanos::from(1_779_953_796_231_000_000));
1576        assert_eq!(greeks.ts_init, UnixNanos::from(789));
1577    }
1578
1579    #[rstest]
1580    fn test_parse_option_greeks_maps_slim_put_variant() {
1581        let msg = parse_ticker_msg(&option_slim_payload(
1582            "options/ws_ticker_slim_eth_put.json",
1583            "ETH-20260612-1900-P",
1584        ))
1585        .unwrap();
1586
1587        let greeks = parse_option_greeks(&msg, UnixNanos::from(789))
1588            .unwrap()
1589            .expect("slim ticker carries put option pricing");
1590
1591        assert_eq!(
1592            greeks.instrument_id,
1593            InstrumentId::from("ETH-20260612-1900-P.DERIVE")
1594        );
1595        assert!((greeks.greeks.delta + 0.30438).abs() < 1e-9);
1596        assert!((greeks.greeks.gamma - 0.00169741).abs() < 1e-9);
1597        assert_eq!(greeks.mark_iv, Some(0.51012));
1598        assert_eq!(greeks.bid_iv, Some(0.48229));
1599        assert_eq!(greeks.ask_iv, Some(0.52063));
1600        assert_eq!(greeks.underlying_price, Some(1992.6));
1601        assert_eq!(greeks.open_interest, Some(42.13));
1602        assert_eq!(greeks.ts_event, UnixNanos::from(1_779_953_797_040_000_000));
1603        assert_eq!(greeks.ts_init, UnixNanos::from(789));
1604    }
1605
1606    #[rstest]
1607    fn test_parse_funding_rate_returns_none_for_option_payload() {
1608        let msg = parse_ticker_msg(&option_envelope_payload(1_700_000_000_010)).unwrap();
1609
1610        let result = parse_funding_rate(&msg, UnixNanos::from(789)).unwrap();
1611
1612        assert!(result.is_none());
1613    }
1614
1615    #[rstest]
1616    fn test_parse_option_greeks_returns_none_for_perp_payload() {
1617        let msg = parse_ticker_msg(&perp_envelope_payload(1_700_000_000_010)).unwrap();
1618
1619        let result = parse_option_greeks(&msg, UnixNanos::from(789)).unwrap();
1620
1621        assert!(result.is_none());
1622    }
1623
1624    #[rstest]
1625    fn test_parse_option_greeks_open_interest_none_when_stats_absent() {
1626        // Legacy full ticker payloads may omit `stats`. When the WS path
1627        // receives one without stats, `open_interest` must degrade to
1628        // None while the remaining greek fields still populate normally.
1629        let timestamp = 1_700_000_000_010_i64;
1630        let mut option_data = option_ticker_json(timestamp);
1631        option_data["instrument_name"] = json!("ETH-20260627-3500-C");
1632        option_data["stats"] = json!(null);
1633        let payload = subscription_data_payload(
1634            "ticker.ETH-20260627-3500-C.1000",
1635            &json!({
1636                "timestamp": timestamp,
1637                "instrument_ticker": option_data,
1638            }),
1639        );
1640        let msg = parse_ticker_msg(&payload).unwrap();
1641
1642        let greeks = parse_option_greeks(&msg, UnixNanos::from(789))
1643            .unwrap()
1644            .expect("option greeks present when option_pricing is set");
1645        assert!(greeks.open_interest.is_none());
1646        // The other greek fields must still be populated from option_pricing
1647        assert!((greeks.greeks.delta - 0.55).abs() < 1e-9);
1648        assert!(greeks.mark_iv.is_some());
1649        assert!(greeks.underlying_price.is_some());
1650    }
1651
1652    #[rstest]
1653    fn test_parse_mark_price_rejects_negative_timestamp() {
1654        let payload = subscription_data_payload(
1655            "ticker.ETH-PERP.1000",
1656            &json!({
1657                "timestamp": -1_i64,
1658                "instrument_ticker": ticker_json(),
1659            }),
1660        );
1661        let msg = parse_ticker_msg(&payload).unwrap();
1662
1663        let err = parse_mark_price(&msg, PRICE_PRECISION, UnixNanos::from(789))
1664            .expect_err("must reject negative ticker timestamp");
1665
1666        assert!(err.to_string().contains("negative Derive ticker timestamp"));
1667    }
1668
1669    #[rstest]
1670    fn test_parse_mark_price_rejects_timestamp_overflow() {
1671        let payload = subscription_data_payload(
1672            "ticker.ETH-PERP.1000",
1673            &json!({
1674                "timestamp": i64::MAX,
1675                "instrument_ticker": ticker_json(),
1676            }),
1677        );
1678        let msg = parse_ticker_msg(&payload).unwrap();
1679
1680        let err = parse_mark_price(&msg, PRICE_PRECISION, UnixNanos::from(789))
1681            .expect_err("must reject overflowing ticker timestamp");
1682
1683        assert!(
1684            err.to_string()
1685                .contains("Derive timestamp overflows nanoseconds")
1686        );
1687    }
1688
1689    #[rstest]
1690    fn test_parse_index_price_rejects_negative_timestamp() {
1691        let payload = subscription_data_payload(
1692            "ticker.ETH-PERP.1000",
1693            &json!({
1694                "timestamp": -1_i64,
1695                "instrument_ticker": ticker_json(),
1696            }),
1697        );
1698        let msg = parse_ticker_msg(&payload).unwrap();
1699
1700        let err = parse_index_price(&msg, PRICE_PRECISION, UnixNanos::from(789))
1701            .expect_err("must reject negative ticker timestamp");
1702
1703        assert!(err.to_string().contains("negative Derive ticker timestamp"));
1704    }
1705
1706    #[rstest]
1707    fn test_parse_index_price_rejects_timestamp_overflow() {
1708        let payload = subscription_data_payload(
1709            "ticker.ETH-PERP.1000",
1710            &json!({
1711                "timestamp": i64::MAX,
1712                "instrument_ticker": ticker_json(),
1713            }),
1714        );
1715        let msg = parse_ticker_msg(&payload).unwrap();
1716
1717        let err = parse_index_price(&msg, PRICE_PRECISION, UnixNanos::from(789))
1718            .expect_err("must reject overflowing ticker timestamp");
1719
1720        assert!(
1721            err.to_string()
1722                .contains("Derive timestamp overflows nanoseconds")
1723        );
1724    }
1725
1726    #[rstest]
1727    fn test_parse_funding_rate_rejects_negative_timestamp() {
1728        let payload = subscription_data_payload(
1729            "ticker.ETH-PERP.1000",
1730            &json!({
1731                "timestamp": -1_i64,
1732                "instrument_ticker": ticker_json(),
1733            }),
1734        );
1735        let msg = parse_ticker_msg(&payload).unwrap();
1736
1737        let err = parse_funding_rate(&msg, UnixNanos::from(789))
1738            .expect_err("must reject negative ticker timestamp");
1739
1740        assert!(err.to_string().contains("negative Derive ticker timestamp"));
1741    }
1742
1743    #[rstest]
1744    fn test_parse_funding_rate_rejects_timestamp_overflow() {
1745        let payload = subscription_data_payload(
1746            "ticker.ETH-PERP.1000",
1747            &json!({
1748                "timestamp": i64::MAX,
1749                "instrument_ticker": ticker_json(),
1750            }),
1751        );
1752        let msg = parse_ticker_msg(&payload).unwrap();
1753
1754        let err = parse_funding_rate(&msg, UnixNanos::from(789))
1755            .expect_err("must reject overflowing ticker timestamp");
1756
1757        assert!(
1758            err.to_string()
1759                .contains("Derive timestamp overflows nanoseconds")
1760        );
1761    }
1762
1763    #[rstest]
1764    fn test_parse_funding_rate_history_record_maps_fields() {
1765        let record = DerivePublicFundingRate {
1766            funding_rate: Decimal::from_str("0.00015").unwrap(),
1767            timestamp: 1_700_000_000_000,
1768        };
1769        let instrument_id = InstrumentId::from("ETH-PERP.DERIVE");
1770
1771        let update = parse_funding_rate_history_record(
1772            &record,
1773            instrument_id,
1774            Some(60),
1775            UnixNanos::from(789),
1776        )
1777        .unwrap();
1778
1779        assert_eq!(update.instrument_id, instrument_id);
1780        assert_eq!(update.rate, Decimal::from_str("0.00015").unwrap());
1781        assert_eq!(update.interval, Some(60));
1782        assert!(update.next_funding_ns.is_none());
1783        assert_eq!(update.ts_event, UnixNanos::from(1_700_000_000_000_000_000));
1784        assert_eq!(update.ts_init, UnixNanos::from(789));
1785    }
1786
1787    #[rstest]
1788    fn test_parse_funding_rate_history_record_rejects_negative_timestamp() {
1789        let record = DerivePublicFundingRate {
1790            funding_rate: Decimal::from_str("0.0001").unwrap(),
1791            timestamp: -1,
1792        };
1793        let err = parse_funding_rate_history_record(
1794            &record,
1795            InstrumentId::from("ETH-PERP.DERIVE"),
1796            None,
1797            UnixNanos::from(789),
1798        )
1799        .expect_err("must reject negative timestamp");
1800
1801        assert!(err.to_string().contains("negative Derive ticker timestamp"));
1802    }
1803
1804    #[rstest]
1805    fn test_parse_candle_record_maps_fields() {
1806        // `timestamp` and `timestamp_bucket` differ so a swap from `timestamp_bucket`
1807        // to `timestamp` in the parser would shift ts_event and fail the assertion.
1808        let record = DerivePublicCandle {
1809            open_price: Decimal::from_str("3500.0").unwrap(),
1810            high_price: Decimal::from_str("3501.5").unwrap(),
1811            low_price: Decimal::from_str("3499.0").unwrap(),
1812            close_price: Decimal::from_str("3501.0").unwrap(),
1813            volume_usd: Decimal::from_str("12345.6").unwrap(),
1814            volume_contracts: Decimal::from_str("3.527").unwrap(),
1815            timestamp: 1_700_000_007,
1816            timestamp_bucket: 1_700_000_000,
1817        };
1818        let bar_type = BarType::from("ETH-PERP.DERIVE-1-MINUTE-LAST-EXTERNAL");
1819
1820        let bar = parse_candle_record(
1821            &record,
1822            bar_type,
1823            PRICE_PRECISION,
1824            SIZE_PRECISION,
1825            UnixNanos::from(789),
1826        )
1827        .unwrap();
1828
1829        assert_eq!(bar.bar_type, bar_type);
1830        assert_eq!(bar.open, Price::from_str("3500.00").unwrap());
1831        assert_eq!(bar.high, Price::from_str("3501.50").unwrap());
1832        assert_eq!(bar.low, Price::from_str("3499.00").unwrap());
1833        assert_eq!(bar.close, Price::from_str("3501.00").unwrap());
1834        assert_eq!(bar.volume, Quantity::from_str("3.527").unwrap());
1835        assert_eq!(bar.ts_event, UnixNanos::from(1_700_000_000_000_000_000));
1836        assert_eq!(bar.ts_init, UnixNanos::from(789));
1837    }
1838
1839    #[rstest]
1840    fn test_parse_candle_record_rejects_negative_timestamp() {
1841        let record = DerivePublicCandle {
1842            open_price: Decimal::from_str("1").unwrap(),
1843            high_price: Decimal::from_str("1").unwrap(),
1844            low_price: Decimal::from_str("1").unwrap(),
1845            close_price: Decimal::from_str("1").unwrap(),
1846            volume_usd: Decimal::ZERO,
1847            volume_contracts: Decimal::ZERO,
1848            timestamp: 1_700_000_000,
1849            timestamp_bucket: -1,
1850        };
1851        let err = parse_candle_record(
1852            &record,
1853            BarType::from("ETH-PERP.DERIVE-1-MINUTE-LAST-EXTERNAL"),
1854            PRICE_PRECISION,
1855            SIZE_PRECISION,
1856            UnixNanos::from(789),
1857        )
1858        .expect_err("must reject negative timestamp");
1859
1860        assert!(err.to_string().contains("negative Derive candle timestamp"));
1861    }
1862
1863    #[rstest]
1864    fn test_parse_candle_record_rejects_timestamp_overflow() {
1865        let record = DerivePublicCandle {
1866            open_price: Decimal::from_str("1").unwrap(),
1867            high_price: Decimal::from_str("1").unwrap(),
1868            low_price: Decimal::from_str("1").unwrap(),
1869            close_price: Decimal::from_str("1").unwrap(),
1870            volume_usd: Decimal::ZERO,
1871            volume_contracts: Decimal::ZERO,
1872            timestamp: 1_700_000_000,
1873            timestamp_bucket: i64::MAX,
1874        };
1875        let err = parse_candle_record(
1876            &record,
1877            BarType::from("ETH-PERP.DERIVE-1-MINUTE-LAST-EXTERNAL"),
1878            PRICE_PRECISION,
1879            SIZE_PRECISION,
1880            UnixNanos::from(789),
1881        )
1882        .expect_err("must reject overflowing timestamp");
1883
1884        assert!(
1885            err.to_string()
1886                .contains("Derive candle timestamp_bucket overflows nanoseconds"),
1887            "{err}",
1888        );
1889    }
1890
1891    #[rstest]
1892    #[case(BarAggregation::Minute, 1, 60)]
1893    #[case(BarAggregation::Minute, 5, 300)]
1894    #[case(BarAggregation::Minute, 15, 900)]
1895    #[case(BarAggregation::Minute, 30, 1800)]
1896    #[case(BarAggregation::Hour, 1, 3600)]
1897    #[case(BarAggregation::Hour, 4, 14400)]
1898    #[case(BarAggregation::Hour, 8, 28800)]
1899    #[case(BarAggregation::Day, 1, 86400)]
1900    #[case(BarAggregation::Week, 1, 604800)]
1901    fn test_bar_spec_to_derive_period_maps_supported_intervals(
1902        #[case] aggregation: BarAggregation,
1903        #[case] step: u64,
1904        #[case] expected: u32,
1905    ) {
1906        assert_eq!(
1907            bar_spec_to_derive_period(aggregation, step).unwrap(),
1908            expected
1909        );
1910    }
1911
1912    #[rstest]
1913    #[case(BarAggregation::Minute, 2, "minute intervals")]
1914    #[case(BarAggregation::Hour, 2, "hour intervals")]
1915    #[case(BarAggregation::Day, 7, "1 DAY interval")]
1916    #[case(BarAggregation::Week, 2, "1 WEEK interval")]
1917    #[case(BarAggregation::Second, 1, "does not support")]
1918    fn test_bar_spec_to_derive_period_rejects_unsupported(
1919        #[case] aggregation: BarAggregation,
1920        #[case] step: u64,
1921        #[case] expected_msg: &str,
1922    ) {
1923        let err =
1924            bar_spec_to_derive_period(aggregation, step).expect_err("must reject unsupported spec");
1925        assert!(
1926            err.to_string().contains(expected_msg),
1927            "expected {expected_msg:?}, was {err}",
1928        );
1929    }
1930
1931    #[rstest]
1932    fn test_parse_option_greeks_rejects_negative_timestamp() {
1933        let mut option_data = option_ticker_json(1_700_000_000_000);
1934        option_data["instrument_name"] = json!("ETH-20260627-3500-C");
1935        let payload = subscription_data_payload(
1936            "ticker.ETH-20260627-3500-C.1000",
1937            &json!({
1938                "timestamp": -1_i64,
1939                "instrument_ticker": option_data,
1940            }),
1941        );
1942        let msg = parse_ticker_msg(&payload).unwrap();
1943
1944        let err = parse_option_greeks(&msg, UnixNanos::from(789))
1945            .expect_err("must reject negative ticker timestamp");
1946
1947        assert!(err.to_string().contains("negative Derive ticker timestamp"));
1948    }
1949
1950    #[rstest]
1951    fn test_parse_option_greeks_rejects_timestamp_overflow() {
1952        let mut option_data = option_ticker_json(1_700_000_000_000);
1953        option_data["instrument_name"] = json!("ETH-20260627-3500-C");
1954        let payload = subscription_data_payload(
1955            "ticker.ETH-20260627-3500-C.1000",
1956            &json!({
1957                "timestamp": i64::MAX,
1958                "instrument_ticker": option_data,
1959            }),
1960        );
1961        let msg = parse_ticker_msg(&payload).unwrap();
1962
1963        let err = parse_option_greeks(&msg, UnixNanos::from(789))
1964            .expect_err("must reject overflowing ticker timestamp");
1965
1966        assert!(
1967            err.to_string()
1968                .contains("Derive timestamp overflows nanoseconds")
1969        );
1970    }
1971}
nautilus_derive/websocket/parse.rs

nautilus_derive/websocket/
parse.rs
