Skip to main content

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