nautilus_tardis/machine/
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
16use std::sync::Arc;
17
18use anyhow::Context;
19use jiff::Timestamp;
20use nautilus_core::UnixNanos;
21use nautilus_model::{
22    data::{
23        Bar, BarType, BookOrder, Data, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate,
24        OptionGreekValues, OptionGreeks, OrderBookDelta, OrderBookDeltas, OrderBookDepth,
25        QuoteTick, TradeTick,
26    },
27    enums::{AggregationSource, BookAction, GreeksConvention, OrderSide, RecordFlag},
28    identifiers::{InstrumentId, TradeId},
29    types::{Price, Quantity},
30};
31use rust_decimal::Decimal;
32
33use super::{
34    message::{
35        BarMsg, BookChangeMsg, BookLevel, BookSnapshotMsg, DerivativeTickerMsg, OptionSummaryMsg,
36        TradeMsg, WsMessage,
37    },
38    types::TardisInstrumentMiniInfo,
39};
40use crate::{
41    common::parse::{
42        derive_trade_id, normalize_amount, parse_aggressor_side, parse_bar_spec, parse_book_action,
43    },
44    config::BookSnapshotOutput,
45};
46
47fn timestamp_to_unix_nanos(timestamp: Timestamp, field: &str) -> anyhow::Result<UnixNanos> {
48    let nanos = u64::try_from(timestamp.as_nanosecond())
49        .with_context(|| format!("invalid timestamp: {field} is outside the UnixNanos range"))?;
50    Ok(UnixNanos::from(nanos))
51}
52
53#[must_use]
54pub fn parse_tardis_ws_message(
55    msg: WsMessage,
56    info: &Arc<TardisInstrumentMiniInfo>,
57    book_snapshot_output: &BookSnapshotOutput,
58) -> Option<Data> {
59    match msg {
60        WsMessage::BookChange(msg) => {
61            if msg.bids.is_empty() && msg.asks.is_empty() {
62                let exchange = msg.exchange;
63                let symbol = &msg.symbol;
64                log::error!("Invalid book change for {exchange} {symbol} (empty bids and asks)");
65                return None;
66            }
67
68            match parse_book_change_msg_as_deltas(
69                &msg,
70                info.price_precision,
71                info.size_precision,
72                info.instrument_id,
73            ) {
74                Ok(deltas) => Some(Data::BookDeltas(Box::new(deltas))),
75                Err(e) => {
76                    log::error!("Failed to parse book change message: {e}");
77                    None
78                }
79            }
80        }
81        WsMessage::BookSnapshot(msg) => match msg.depth {
82            1 => {
83                match parse_book_snapshot_msg_as_quote(
84                    &msg,
85                    info.price_precision,
86                    info.size_precision,
87                    info.instrument_id,
88                ) {
89                    Ok(quote) => Some(Data::Quote(quote)),
90                    Err(e) => {
91                        log::error!("Failed to parse book snapshot quote message: {e}");
92                        None
93                    }
94                }
95            }
96            _ => match book_snapshot_output {
97                BookSnapshotOutput::Depth => {
98                    match parse_book_snapshot_msg_as_depth(
99                        &msg,
100                        info.price_precision,
101                        info.size_precision,
102                        info.instrument_id,
103                    ) {
104                        Ok(depth) => Some(Data::BookDepth(Box::new(depth))),
105                        Err(e) => {
106                            log::error!("Failed to parse book snapshot as depth: {e}");
107                            None
108                        }
109                    }
110                }
111                BookSnapshotOutput::Deltas => {
112                    match parse_book_snapshot_msg_as_deltas(
113                        &msg,
114                        info.price_precision,
115                        info.size_precision,
116                        info.instrument_id,
117                    ) {
118                        Ok(deltas) => Some(Data::BookDeltas(Box::new(deltas))),
119                        Err(e) => {
120                            log::error!("Failed to parse book snapshot as deltas: {e}");
121                            None
122                        }
123                    }
124                }
125            },
126        },
127        WsMessage::Trade(msg) => {
128            match parse_trade_msg(
129                &msg,
130                info.price_precision,
131                info.size_precision,
132                info.instrument_id,
133            ) {
134                Ok(trade) => Some(Data::Trade(trade)),
135                Err(e) => {
136                    log::error!("Failed to parse trade message: {e}");
137                    None
138                }
139            }
140        }
141        WsMessage::TradeBar(msg) => {
142            match parse_bar_msg(
143                &msg,
144                info.price_precision,
145                info.size_precision,
146                info.instrument_id,
147            ) {
148                Ok(bar) => Some(Data::Bar(bar)),
149                Err(e) => {
150                    log::error!("Failed to parse bar message: {e}");
151                    None
152                }
153            }
154        }
155        WsMessage::OptionSummary(msg) => Some(Data::OptionGreeks(parse_option_summary_msg(
156            &msg,
157            info.instrument_id,
158        ))),
159        // Derivative ticker messages are handled through a separate callback path
160        // for FundingRateUpdate since they're not part of the Data enum.
161        WsMessage::DerivativeTicker(_) => None,
162        WsMessage::Disconnect(_) => None,
163    }
164}
165
166#[must_use]
167pub fn parse_tardis_ws_message_data(
168    msg: WsMessage,
169    info: &Arc<TardisInstrumentMiniInfo>,
170    book_snapshot_output: &BookSnapshotOutput,
171    extract_bbo_as_quotes: bool,
172) -> Vec<Data> {
173    match msg {
174        WsMessage::OptionSummary(msg) if extract_bbo_as_quotes => {
175            let mut data = Vec::with_capacity(2);
176
177            match parse_option_summary_msg_as_quote(
178                &msg,
179                info.price_precision,
180                info.size_precision,
181                info.instrument_id,
182            ) {
183                Ok(Some(quote)) => data.push(Data::Quote(quote)),
184                Ok(None) => {}
185                Err(e) => {
186                    log::error!("Failed to parse option summary quote message: {e}");
187                }
188            }
189
190            data.push(Data::OptionGreeks(parse_option_summary_msg(
191                &msg,
192                info.instrument_id,
193            )));
194            data
195        }
196        msg => parse_tardis_ws_message(msg, info, book_snapshot_output)
197            .into_iter()
198            .collect(),
199    }
200}
201
202/// Parses a Tardis option summary message into a Nautilus `OptionGreeks`.
203///
204/// Greeks absent from the exchange feed default to `0.0` (matching the existing exchange-greeks
205/// producers); implied volatilities, underlying price and open interest stay `None` when the
206/// exchange does not provide them.
207#[must_use]
208pub fn parse_option_summary_msg(
209    msg: &OptionSummaryMsg,
210    instrument_id: InstrumentId,
211) -> OptionGreeks {
212    OptionGreeks {
213        instrument_id,
214        convention: GreeksConvention::BlackScholes,
215        greeks: OptionGreekValues {
216            delta: msg.delta.unwrap_or(0.0),
217            gamma: msg.gamma.unwrap_or(0.0),
218            vega: msg.vega.unwrap_or(0.0),
219            theta: msg.theta.unwrap_or(0.0),
220            rho: msg.rho.unwrap_or(0.0),
221        },
222        mark_iv: msg.mark_iv,
223        bid_iv: msg.best_bid_iv,
224        ask_iv: msg.best_ask_iv,
225        underlying_price: msg.underlying_price,
226        open_interest: msg.open_interest,
227        ts_event: UnixNanos::from(msg.timestamp),
228        ts_init: UnixNanos::from(msg.local_timestamp),
229    }
230}
231
232/// Parses a Tardis option summary best bid/offer into a Nautilus `QuoteTick`.
233///
234/// Returns `Ok(None)` when any best bid/offer field is absent.
235///
236/// # Errors
237///
238/// Returns an error if a provided bid or ask price or size is invalid.
239pub fn parse_option_summary_msg_as_quote(
240    msg: &OptionSummaryMsg,
241    price_precision: u8,
242    size_precision: u8,
243    instrument_id: InstrumentId,
244) -> anyhow::Result<Option<QuoteTick>> {
245    let (Some(best_bid_price), Some(best_bid_amount), Some(best_ask_price), Some(best_ask_amount)) = (
246        msg.best_bid_price,
247        msg.best_bid_amount,
248        msg.best_ask_price,
249        msg.best_ask_amount,
250    ) else {
251        return Ok(None);
252    };
253
254    let bid_price = Price::from_decimal_dp(best_bid_price, price_precision)
255        .with_context(|| format!("invalid option summary bid price for message: {msg:?}"))?;
256    let ask_price = Price::from_decimal_dp(best_ask_price, price_precision)
257        .with_context(|| format!("invalid option summary ask price for message: {msg:?}"))?;
258    let bid_size = parse_non_zero_quantity(best_bid_amount, size_precision)
259        .with_context(|| format!("invalid option summary bid size for message: {msg:?}"))?;
260    let ask_size = parse_non_zero_quantity(best_ask_amount, size_precision)
261        .with_context(|| format!("invalid option summary ask size for message: {msg:?}"))?;
262
263    Ok(Some(QuoteTick::new(
264        instrument_id,
265        bid_price,
266        ask_price,
267        bid_size,
268        ask_size,
269        UnixNanos::from(msg.timestamp),
270        UnixNanos::from(msg.local_timestamp),
271    )))
272}
273
274/// Parse a Tardis WebSocket message specifically for funding rate updates.
275/// Returns `Some(FundingRateUpdate)` if the message contains funding rate data, `None` otherwise.
276#[must_use]
277pub fn parse_tardis_ws_message_funding_rate(
278    msg: WsMessage,
279    info: &Arc<TardisInstrumentMiniInfo>,
280) -> Option<FundingRateUpdate> {
281    match msg {
282        WsMessage::DerivativeTicker(msg) => {
283            match parse_derivative_ticker_msg(&msg, info.instrument_id) {
284                Ok(funding_rate) => funding_rate,
285                Err(e) => {
286                    log::error!("Failed to parse derivative ticker message for funding rate: {e}");
287                    None
288                }
289            }
290        }
291        _ => None, // Only derivative ticker messages can contain funding rates
292    }
293}
294
295/// Parse a book change message into order book deltas, returning an error if timestamps invalid.
296/// Parse a book change message into order book deltas.
297///
298/// # Errors
299///
300/// Returns an error if timestamp fields cannot be converted to nanoseconds.
301pub fn parse_book_change_msg_as_deltas(
302    msg: &BookChangeMsg,
303    price_precision: u8,
304    size_precision: u8,
305    instrument_id: InstrumentId,
306) -> anyhow::Result<OrderBookDeltas> {
307    parse_book_msg_as_deltas(
308        &msg.bids,
309        &msg.asks,
310        msg.is_snapshot,
311        price_precision,
312        size_precision,
313        instrument_id,
314        msg.timestamp,
315        msg.local_timestamp,
316    )
317}
318
319/// Parse a book snapshot message into order book deltas, returning an error if timestamps invalid.
320/// Parse a book snapshot message into order book deltas.
321///
322/// # Errors
323///
324/// Returns an error if timestamp fields cannot be converted to nanoseconds.
325pub fn parse_book_snapshot_msg_as_deltas(
326    msg: &BookSnapshotMsg,
327    price_precision: u8,
328    size_precision: u8,
329    instrument_id: InstrumentId,
330) -> anyhow::Result<OrderBookDeltas> {
331    parse_book_msg_as_deltas(
332        &msg.bids,
333        &msg.asks,
334        true,
335        price_precision,
336        size_precision,
337        instrument_id,
338        msg.timestamp,
339        msg.local_timestamp,
340    )
341}
342
343/// Parse a book snapshot message into an [`OrderBookDepth`].
344///
345/// # Errors
346///
347/// Returns an error if timestamp fields cannot be converted to nanoseconds, or a level
348/// price or amount is invalid.
349pub fn parse_book_snapshot_msg_as_depth(
350    msg: &BookSnapshotMsg,
351    price_precision: u8,
352    size_precision: u8,
353    instrument_id: InstrumentId,
354) -> anyhow::Result<OrderBookDepth> {
355    let ts_event = timestamp_to_unix_nanos(msg.timestamp, "event timestamp")?;
356    let ts_init = timestamp_to_unix_nanos(msg.local_timestamp, "init timestamp")?;
357
358    let mut bids = Vec::with_capacity(msg.bids.len());
359    let mut asks = Vec::with_capacity(msg.asks.len());
360    let mut bid_counts = Vec::with_capacity(msg.bids.len());
361    let mut ask_counts = Vec::with_capacity(msg.asks.len());
362
363    for level in &msg.bids {
364        let price = Price::from_decimal_dp(level.price, price_precision)?;
365        let size = Quantity::from_decimal_dp(level.amount, size_precision)?;
366        bids.push(BookOrder::new(OrderSide::Buy, price, size, 0));
367        bid_counts.push(1);
368    }
369
370    for level in &msg.asks {
371        let price = Price::from_decimal_dp(level.price, price_precision)?;
372        let size = Quantity::from_decimal_dp(level.amount, size_precision)?;
373        asks.push(BookOrder::new(OrderSide::Sell, price, size, 0));
374        ask_counts.push(1);
375    }
376
377    Ok(OrderBookDepth::new(
378        instrument_id,
379        bids,
380        asks,
381        bid_counts,
382        ask_counts,
383        RecordFlag::F_SNAPSHOT as u8,
384        0, // Sequence not available from Tardis
385        ts_event,
386        ts_init,
387    ))
388}
389
390/// Parse raw book levels into order book deltas, returning error for invalid timestamps.
391#[expect(clippy::too_many_arguments)]
392/// Parse raw book levels into order book deltas.
393///
394/// # Errors
395///
396/// Returns an error if timestamp fields cannot be converted to nanoseconds.
397pub fn parse_book_msg_as_deltas(
398    bids: &[BookLevel],
399    asks: &[BookLevel],
400    is_snapshot: bool,
401    price_precision: u8,
402    size_precision: u8,
403    instrument_id: InstrumentId,
404    timestamp: Timestamp,
405    local_timestamp: Timestamp,
406) -> anyhow::Result<OrderBookDeltas> {
407    let ts_event = timestamp_to_unix_nanos(timestamp, "event timestamp")?;
408    let ts_init = timestamp_to_unix_nanos(local_timestamp, "init timestamp")?;
409
410    let capacity = if is_snapshot {
411        bids.len() + asks.len() + 1
412    } else {
413        bids.len() + asks.len()
414    };
415    let mut deltas: Vec<OrderBookDelta> = Vec::with_capacity(capacity);
416
417    if is_snapshot {
418        deltas.push(OrderBookDelta::clear(instrument_id, 0, ts_event, ts_init));
419    }
420
421    for level in bids {
422        match parse_book_level(
423            instrument_id,
424            price_precision,
425            size_precision,
426            OrderSide::Buy,
427            level,
428            is_snapshot,
429            ts_event,
430            ts_init,
431        ) {
432            Ok(delta) => deltas.push(delta),
433            Err(e) => log::warn!("Skipping invalid bid level for {instrument_id}: {e}"),
434        }
435    }
436
437    for level in asks {
438        match parse_book_level(
439            instrument_id,
440            price_precision,
441            size_precision,
442            OrderSide::Sell,
443            level,
444            is_snapshot,
445            ts_event,
446            ts_init,
447        ) {
448            Ok(delta) => deltas.push(delta),
449            Err(e) => log::warn!("Skipping invalid ask level for {instrument_id}: {e}"),
450        }
451    }
452
453    if let Some(last_delta) = deltas.last_mut() {
454        last_delta.flags |= RecordFlag::F_LAST as u8;
455    }
456
457    Ok(OrderBookDeltas::new(instrument_id, deltas))
458}
459
460/// Parse a single book level into an order book delta.
461///
462/// # Errors
463///
464/// Returns an error if the price or normalized amount is invalid, or a non-delete action
465/// has a zero size after normalization.
466#[expect(clippy::too_many_arguments)]
467pub fn parse_book_level(
468    instrument_id: InstrumentId,
469    price_precision: u8,
470    size_precision: u8,
471    side: OrderSide,
472    level: &BookLevel,
473    is_snapshot: bool,
474    ts_event: UnixNanos,
475    ts_init: UnixNanos,
476) -> anyhow::Result<OrderBookDelta> {
477    let amount = normalize_amount(level.amount, size_precision);
478    let price = Price::from_decimal_dp(level.price, price_precision)?;
479    let size = Quantity::from_decimal_dp(amount, size_precision)?;
480    let action = parse_book_action(is_snapshot, size.as_f64());
481    let order_id = 0; // Not applicable for L2 data
482    let order = BookOrder::new(side, price, size, order_id);
483    let flags = if is_snapshot {
484        RecordFlag::F_SNAPSHOT as u8
485    } else {
486        0
487    };
488    let sequence = 0; // Not available
489
490    anyhow::ensure!(
491        !(action != BookAction::Delete && size.is_zero()),
492        "Invalid zero size for {action}"
493    );
494
495    Ok(OrderBookDelta::new(
496        instrument_id,
497        action,
498        order,
499        flags,
500        sequence,
501        ts_event,
502        ts_init,
503    ))
504}
505
506/// Parse a book snapshot message into a quote tick, returning an error on invalid data.
507/// Parse a book snapshot message into a quote tick.
508///
509/// # Errors
510///
511/// Returns an error if missing bid/ask levels or invalid prices or sizes.
512pub fn parse_book_snapshot_msg_as_quote(
513    msg: &BookSnapshotMsg,
514    price_precision: u8,
515    size_precision: u8,
516    instrument_id: InstrumentId,
517) -> anyhow::Result<QuoteTick> {
518    let ts_event = UnixNanos::from(msg.timestamp);
519    let ts_init = UnixNanos::from(msg.local_timestamp);
520
521    let best_bid = msg
522        .bids
523        .first()
524        .context("missing best bid level for quote message")?;
525    let bid_price = Price::from_decimal_dp(best_bid.price, price_precision)
526        .with_context(|| format!("invalid bid price for message: {msg:?}"))?;
527    let bid_size = parse_non_zero_quantity(best_bid.amount, size_precision)
528        .with_context(|| format!("Invalid bid size for message: {msg:?}"))?;
529
530    let best_ask = msg
531        .asks
532        .first()
533        .context("missing best ask level for quote message")?;
534    let ask_price = Price::from_decimal_dp(best_ask.price, price_precision)
535        .with_context(|| format!("invalid ask price for message: {msg:?}"))?;
536    let ask_size = parse_non_zero_quantity(best_ask.amount, size_precision)
537        .with_context(|| format!("Invalid ask size for message: {msg:?}"))?;
538
539    Ok(QuoteTick::new(
540        instrument_id,
541        bid_price,
542        ask_price,
543        bid_size,
544        ask_size,
545        ts_event,
546        ts_init,
547    ))
548}
549
550/// Parse a trade message into a trade tick, returning an error on invalid data.
551/// Parse a trade message into a trade tick.
552///
553/// # Errors
554///
555/// Returns an error if the trade price or size is invalid.
556pub fn parse_trade_msg(
557    msg: &TradeMsg,
558    price_precision: u8,
559    size_precision: u8,
560    instrument_id: InstrumentId,
561) -> anyhow::Result<TradeTick> {
562    let price = Price::from_decimal_dp(msg.price, price_precision)
563        .with_context(|| format!("invalid trade price in message: {msg:?}"))?;
564    let size = parse_non_zero_quantity(msg.amount, size_precision)
565        .with_context(|| format!("Invalid trade size in message: {msg:?}"))?;
566    let aggressor_side = parse_aggressor_side(&msg.side);
567    let ts_event = UnixNanos::from(msg.timestamp);
568    let ts_init = UnixNanos::from(msg.local_timestamp);
569    let trade_id = match msg.id.as_deref() {
570        Some(id) if !id.is_empty() => TradeId::new(id),
571        _ => {
572            let price = msg.price.normalize().to_string();
573            let amount = msg.amount.normalize().to_string();
574            derive_trade_id(msg.symbol, ts_event.as_u64(), &price, &amount, &msg.side)
575        }
576    };
577
578    Ok(TradeTick::new(
579        instrument_id,
580        price,
581        size,
582        aggressor_side,
583        trade_id,
584        ts_event,
585        ts_init,
586    ))
587}
588
589/// Parse a bar message into a Bar.
590///
591/// # Errors
592///
593/// Returns an error if the bar specification cannot be parsed, a price is invalid, the
594/// volume is invalid or rounds to zero, or the size precision is invalid.
595pub fn parse_bar_msg(
596    msg: &BarMsg,
597    price_precision: u8,
598    size_precision: u8,
599    instrument_id: InstrumentId,
600) -> anyhow::Result<Bar> {
601    let spec = parse_bar_spec(&msg.name)?;
602    let bar_type = BarType::new(instrument_id, spec, AggregationSource::External);
603
604    let open = Price::from_decimal_dp(msg.open, price_precision)?;
605    let high = Price::from_decimal_dp(msg.high, price_precision)?;
606    let low = Price::from_decimal_dp(msg.low, price_precision)?;
607    let close = Price::from_decimal_dp(msg.close, price_precision)?;
608    let volume = parse_non_zero_quantity(msg.volume, size_precision)?;
609    let ts_event = UnixNanos::from(msg.timestamp);
610    let ts_init = UnixNanos::from(msg.local_timestamp);
611
612    Ok(Bar::new(
613        bar_type, open, high, low, close, volume, ts_event, ts_init,
614    ))
615}
616
617/// Extracts event and init timestamps from a derivative ticker message.
618fn parse_derivative_ticker_timestamps(
619    msg: &DerivativeTickerMsg,
620) -> anyhow::Result<(UnixNanos, UnixNanos)> {
621    Ok((
622        timestamp_to_unix_nanos(msg.timestamp, "event timestamp")?,
623        timestamp_to_unix_nanos(msg.local_timestamp, "init timestamp")?,
624    ))
625}
626
627/// Parses a derivative ticker message into a funding rate update.
628///
629/// # Errors
630///
631/// Returns an error if timestamp conversion fails.
632pub fn parse_derivative_ticker_msg(
633    msg: &DerivativeTickerMsg,
634    instrument_id: InstrumentId,
635) -> anyhow::Result<Option<FundingRateUpdate>> {
636    let rate = match msg.funding_rate {
637        Some(rate) => rate,
638        None => return Ok(None),
639    };
640
641    let (ts_event, ts_init) = parse_derivative_ticker_timestamps(msg)?;
642    let next_funding_ns = msg
643        .funding_timestamp
644        .map(|ts| timestamp_to_unix_nanos(ts, "funding timestamp"))
645        .transpose()?;
646
647    Ok(Some(FundingRateUpdate::new(
648        instrument_id,
649        rate,
650        None,
651        next_funding_ns,
652        ts_event,
653        ts_init,
654    )))
655}
656
657/// Parses a derivative ticker message into a mark price update.
658///
659/// # Errors
660///
661/// Returns an error if timestamp conversion fails or the mark price is invalid.
662pub fn parse_derivative_ticker_mark_price(
663    msg: &DerivativeTickerMsg,
664    instrument_id: InstrumentId,
665    price_precision: u8,
666) -> anyhow::Result<Option<MarkPriceUpdate>> {
667    let mark_price = match msg.mark_price {
668        Some(p) => p,
669        None => return Ok(None),
670    };
671
672    let (ts_event, ts_init) = parse_derivative_ticker_timestamps(msg)?;
673    let price = Price::from_decimal_dp(mark_price, price_precision)?;
674
675    Ok(Some(MarkPriceUpdate::new(
676        instrument_id,
677        price,
678        ts_event,
679        ts_init,
680    )))
681}
682
683/// Parses a derivative ticker message into an index price update.
684///
685/// # Errors
686///
687/// Returns an error if timestamp conversion fails or the index price is invalid.
688pub fn parse_derivative_ticker_index_price(
689    msg: &DerivativeTickerMsg,
690    instrument_id: InstrumentId,
691    price_precision: u8,
692) -> anyhow::Result<Option<IndexPriceUpdate>> {
693    let index_price = match msg.index_price {
694        Some(p) => p,
695        None => return Ok(None),
696    };
697
698    let (ts_event, ts_init) = parse_derivative_ticker_timestamps(msg)?;
699    let price = Price::from_decimal_dp(index_price, price_precision)?;
700
701    Ok(Some(IndexPriceUpdate::new(
702        instrument_id,
703        price,
704        ts_event,
705        ts_init,
706    )))
707}
708
709fn parse_non_zero_quantity(amount: Decimal, precision: u8) -> anyhow::Result<Quantity> {
710    anyhow::ensure!(!amount.is_zero(), "value was zero");
711    let quantity = Quantity::from_decimal_dp(amount, precision)?;
712    anyhow::ensure!(
713        !quantity.is_zero(),
714        "value {amount} was zero after rounding to precision {precision}"
715    );
716    Ok(quantity)
717}
718
719#[cfg(test)]
720mod tests {
721    use nautilus_model::enums::AggressorSide;
722    use rstest::rstest;
723    use rust_decimal_macros::dec;
724
725    use super::*;
726    use crate::common::{
727        enums::TardisExchange, parse::parse_instrument_id, testing::load_test_json,
728    };
729
730    #[rstest]
731    fn test_parse_book_change_message() {
732        let json_data = load_test_json("book_change.json");
733        let msg: BookChangeMsg = serde_json::from_str(&json_data).unwrap();
734
735        let price_precision = 0;
736        let size_precision = 0;
737        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
738        let deltas =
739            parse_book_change_msg_as_deltas(&msg, price_precision, size_precision, instrument_id)
740                .unwrap();
741
742        assert_eq!(deltas.deltas.len(), 1);
743        assert_eq!(deltas.instrument_id, instrument_id);
744        assert_eq!(deltas.flags, RecordFlag::F_LAST as u8);
745        assert_eq!(deltas.sequence, 0);
746        assert_eq!(deltas.ts_event, UnixNanos::from(1571830193469000000));
747        assert_eq!(deltas.ts_init, UnixNanos::from(1571830193469000000));
748        assert_eq!(
749            deltas.deltas[0].instrument_id,
750            InstrumentId::from("XBTUSD.BITMEX")
751        );
752        assert_eq!(deltas.deltas[0].action, BookAction::Update);
753        assert_eq!(deltas.deltas[0].order.price, Price::from("7985"));
754        assert_eq!(deltas.deltas[0].order.size, Quantity::from(283318));
755        assert_eq!(deltas.deltas[0].order.order_id, 0);
756        assert_eq!(deltas.deltas[0].flags, RecordFlag::F_LAST as u8);
757        assert_eq!(deltas.deltas[0].sequence, 0);
758        assert_eq!(
759            deltas.deltas[0].ts_event,
760            UnixNanos::from(1571830193469000000)
761        );
762        assert_eq!(
763            deltas.deltas[0].ts_init,
764            UnixNanos::from(1571830193469000000)
765        );
766    }
767
768    #[rstest]
769    fn test_parse_book_snapshot_message_as_deltas() {
770        let json_data = load_test_json("book_snapshot.json");
771        let msg: BookSnapshotMsg = serde_json::from_str(&json_data).unwrap();
772
773        let price_precision = 1;
774        let size_precision = 0;
775        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
776        let deltas =
777            parse_book_snapshot_msg_as_deltas(&msg, price_precision, size_precision, instrument_id)
778                .unwrap();
779
780        let clear_delta = deltas.deltas[0];
781        let bid_delta = deltas.deltas[1];
782        let ask_delta = deltas.deltas[3];
783
784        assert_eq!(deltas.deltas.len(), 5);
785        assert_eq!(deltas.instrument_id, instrument_id);
786        assert_eq!(
787            deltas.flags,
788            RecordFlag::F_LAST as u8 + RecordFlag::F_SNAPSHOT as u8
789        );
790        assert_eq!(deltas.sequence, 0);
791        assert_eq!(deltas.ts_event, UnixNanos::from(1572010786950000000));
792        assert_eq!(deltas.ts_init, UnixNanos::from(1572010786961000000));
793
794        // CLEAR delta
795        assert_eq!(clear_delta.instrument_id, instrument_id);
796        assert_eq!(clear_delta.action, BookAction::Clear);
797        assert_eq!(clear_delta.flags, RecordFlag::F_SNAPSHOT as u8);
798        assert_eq!(clear_delta.sequence, 0);
799        assert_eq!(clear_delta.ts_event, UnixNanos::from(1572010786950000000));
800        assert_eq!(clear_delta.ts_init, UnixNanos::from(1572010786961000000));
801
802        // First bid delta
803        assert_eq!(bid_delta.instrument_id, instrument_id);
804        assert_eq!(bid_delta.action, BookAction::Add);
805        assert_eq!(bid_delta.order.side, OrderSide::Buy.into());
806        assert_eq!(bid_delta.order.price, Price::from("7633.5"));
807        assert_eq!(bid_delta.order.size, Quantity::from(1906067));
808        assert_eq!(bid_delta.order.order_id, 0);
809        assert_eq!(bid_delta.flags, RecordFlag::F_SNAPSHOT as u8);
810        assert_eq!(bid_delta.sequence, 0);
811        assert_eq!(bid_delta.ts_event, UnixNanos::from(1572010786950000000));
812        assert_eq!(bid_delta.ts_init, UnixNanos::from(1572010786961000000));
813
814        // First ask delta
815        assert_eq!(ask_delta.instrument_id, instrument_id);
816        assert_eq!(ask_delta.action, BookAction::Add);
817        assert_eq!(ask_delta.order.side, OrderSide::Sell.into());
818        assert_eq!(ask_delta.order.price, Price::from("7634.0"));
819        assert_eq!(ask_delta.order.size, Quantity::from(1467849));
820        assert_eq!(ask_delta.order.order_id, 0);
821        assert_eq!(ask_delta.flags, RecordFlag::F_SNAPSHOT as u8);
822        assert_eq!(ask_delta.sequence, 0);
823        assert_eq!(ask_delta.ts_event, UnixNanos::from(1572010786950000000));
824        assert_eq!(ask_delta.ts_init, UnixNanos::from(1572010786961000000));
825    }
826
827    #[rstest]
828    fn test_parse_book_snapshot_message_as_depth() {
829        let json_data = load_test_json("book_snapshot.json");
830        let msg: BookSnapshotMsg = serde_json::from_str(&json_data).unwrap();
831
832        let price_precision = 1;
833        let size_precision = 0;
834        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
835
836        let depth =
837            parse_book_snapshot_msg_as_depth(&msg, price_precision, size_precision, instrument_id)
838                .unwrap();
839
840        assert_eq!(depth.instrument_id, instrument_id);
841        assert_eq!(depth.flags, RecordFlag::F_SNAPSHOT as u8);
842        assert_eq!(depth.sequence, 0);
843        assert_eq!(depth.ts_event, UnixNanos::from(1572010786950000000));
844        assert_eq!(depth.ts_init, UnixNanos::from(1572010786961000000));
845
846        // Check first bid level
847        assert_eq!(depth.bids[0].side, OrderSide::Buy.into());
848        assert_eq!(depth.bids[0].price, Price::from("7633.5"));
849        assert_eq!(depth.bids[0].size, Quantity::from(1906067));
850        assert_eq!(depth.bids[0].order_id, 0);
851        assert_eq!(depth.bid_counts[0], 1);
852
853        // Check second bid level
854        assert_eq!(depth.bids[1].side, OrderSide::Buy.into());
855        assert_eq!(depth.bids[1].price, Price::from("7633.0"));
856        assert_eq!(depth.bids[1].size, Quantity::from(65319));
857        assert_eq!(depth.bid_counts[1], 1);
858
859        // Check first ask level
860        assert_eq!(depth.asks[0].side, OrderSide::Sell.into());
861        assert_eq!(depth.asks[0].price, Price::from("7634.0"));
862        assert_eq!(depth.asks[0].size, Quantity::from(1467849));
863        assert_eq!(depth.asks[0].order_id, 0);
864        assert_eq!(depth.ask_counts[0], 1);
865
866        // Check second ask level
867        assert_eq!(depth.asks[1].side, OrderSide::Sell.into());
868        assert_eq!(depth.asks[1].price, Price::from("7634.5"));
869        assert_eq!(depth.asks[1].size, Quantity::from(67939));
870        assert_eq!(depth.ask_counts[1], 1);
871
872        assert_eq!(depth.bids.len(), 2);
873        assert_eq!(depth.asks.len(), 2);
874        assert_eq!(depth.bid_counts.as_slice(), &[1; 2]);
875        assert_eq!(depth.ask_counts.as_slice(), &[1; 2]);
876        assert_eq!(depth.bids[1].order_id, 0);
877        assert_eq!(depth.asks[1].order_id, 0);
878    }
879
880    #[rstest]
881    fn test_parse_book_snapshot_message_as_depth_keeps_all_levels() {
882        let bids: Vec<BookLevel> = (0..25)
883            .map(|i| BookLevel {
884                price: dec!(7633.5) - Decimal::from(i) * dec!(0.5),
885                amount: Decimal::from(1000 + i),
886            })
887            .collect();
888        let asks: Vec<BookLevel> = (0..25)
889            .map(|i| BookLevel {
890                price: dec!(7634.0) + Decimal::from(i) * dec!(0.5),
891                amount: Decimal::from(2000 + i),
892            })
893            .collect();
894        let msg = BookSnapshotMsg {
895            symbol: ustr::Ustr::from("XBTUSD"),
896            exchange: TardisExchange::Bitmex,
897            name: "book_snapshot_25_100ms".to_string(),
898            depth: 25,
899            interval: 100,
900            bids,
901            asks,
902            timestamp: "2019-10-25T13:39:46.950Z".parse::<Timestamp>().unwrap(),
903            local_timestamp: "2019-10-25T13:39:46.961Z".parse::<Timestamp>().unwrap(),
904        };
905
906        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
907        let depth = parse_book_snapshot_msg_as_depth(&msg, 1, 0, instrument_id).unwrap();
908
909        assert_eq!(depth.instrument_id, instrument_id);
910        assert_eq!(depth.bids.len(), 25);
911        assert_eq!(depth.asks.len(), 25);
912        assert_eq!(depth.bid_counts.as_slice(), &[1; 25]);
913        assert_eq!(depth.ask_counts.as_slice(), &[1; 25]);
914        assert_eq!(depth.bids[0].price, Price::from("7633.5"));
915        assert_eq!(depth.bids[24].price, Price::from("7621.5"));
916        assert_eq!(depth.asks[0].price, Price::from("7634.0"));
917        assert_eq!(depth.asks[24].price, Price::from("7646.0"));
918        assert_eq!(depth.flags, RecordFlag::F_SNAPSHOT as u8);
919        assert_eq!(depth.sequence, 0);
920    }
921
922    #[rstest]
923    fn test_parse_book_snapshot_message_as_quote() {
924        let json_data = load_test_json("book_snapshot.json");
925        let msg: BookSnapshotMsg = serde_json::from_str(&json_data).unwrap();
926
927        let price_precision = 1;
928        let size_precision = 0;
929        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
930        let quote =
931            parse_book_snapshot_msg_as_quote(&msg, price_precision, size_precision, instrument_id)
932                .expect("Failed to parse book snapshot quote message");
933
934        assert_eq!(quote.instrument_id, instrument_id);
935        assert_eq!(quote.bid_price, Price::from("7633.5"));
936        assert_eq!(quote.bid_size, Quantity::from(1906067));
937        assert_eq!(quote.ask_price, Price::from("7634.0"));
938        assert_eq!(quote.ask_size, Quantity::from(1467849));
939        assert_eq!(quote.ts_event, UnixNanos::from(1572010786950000000));
940        assert_eq!(quote.ts_init, UnixNanos::from(1572010786961000000));
941    }
942
943    #[rstest]
944    fn test_parse_trade_message() {
945        let json_data = load_test_json("trade.json");
946        let msg: TradeMsg = serde_json::from_str(&json_data).unwrap();
947
948        let price_precision = 0;
949        let size_precision = 0;
950        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
951        let trade = parse_trade_msg(&msg, price_precision, size_precision, instrument_id)
952            .expect("Failed to parse trade message");
953
954        assert_eq!(trade.instrument_id, instrument_id);
955        assert_eq!(trade.price, Price::from("7996"));
956        assert_eq!(trade.size, Quantity::from(50));
957        assert_eq!(trade.aggressor_side, AggressorSide::Sell);
958        assert_eq!(trade.ts_event, UnixNanos::from(1571826769669000000));
959        assert_eq!(trade.ts_init, UnixNanos::from(1571826769740000000));
960    }
961
962    fn build_trade_msg_without_id() -> TradeMsg {
963        let json_data = load_test_json("trade.json");
964        let mut msg: TradeMsg = serde_json::from_str(&json_data).unwrap();
965        msg.id = None;
966        msg
967    }
968
969    #[rstest]
970    fn test_parse_trade_message_derives_trade_id_when_missing() {
971        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
972
973        let first = parse_trade_msg(&build_trade_msg_without_id(), 0, 0, instrument_id).unwrap();
974        let second = parse_trade_msg(&build_trade_msg_without_id(), 0, 0, instrument_id).unwrap();
975
976        assert_eq!(first.trade_id, second.trade_id, "derivation must be stable");
977        assert_eq!(first.trade_id.as_str().len(), 16);
978
979        let mut altered = build_trade_msg_without_id();
980        altered.price = dec!(7997.0);
981        let altered_trade = parse_trade_msg(&altered, 0, 0, instrument_id).unwrap();
982        assert_ne!(first.trade_id, altered_trade.trade_id);
983    }
984
985    #[rstest]
986    fn test_parse_trade_message_derives_trade_id_when_empty() {
987        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
988
989        let mut msg = build_trade_msg_without_id();
990        msg.id = Some(String::new());
991
992        let trade = parse_trade_msg(&msg, 0, 0, instrument_id).unwrap();
993        let fallback = parse_trade_msg(&build_trade_msg_without_id(), 0, 0, instrument_id).unwrap();
994        assert_eq!(trade.trade_id, fallback.trade_id);
995    }
996
997    #[rstest]
998    fn test_parse_bar_message() {
999        let json_data = load_test_json("bar.json");
1000        let msg: BarMsg = serde_json::from_str(&json_data).unwrap();
1001
1002        let price_precision = 1;
1003        let size_precision = 0;
1004        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
1005        let bar = parse_bar_msg(&msg, price_precision, size_precision, instrument_id).unwrap();
1006
1007        assert_eq!(
1008            bar.bar_type,
1009            BarType::from("XBTUSD.BITMEX-10-SECOND-LAST-EXTERNAL")
1010        );
1011        assert_eq!(bar.open, Price::from("7623.5"));
1012        assert_eq!(bar.high, Price::from("7623.5"));
1013        assert_eq!(bar.low, Price::from("7623"));
1014        assert_eq!(bar.close, Price::from("7623.5"));
1015        assert_eq!(bar.volume, Quantity::from(37034));
1016        assert_eq!(bar.ts_event, UnixNanos::from(1572009100000000000));
1017        assert_eq!(bar.ts_init, UnixNanos::from(1572009100369000000));
1018    }
1019
1020    #[rstest]
1021    fn test_parse_tardis_ws_message_book_snapshot_routes_to_depth() {
1022        let json_data = load_test_json("book_snapshot.json");
1023        let msg: BookSnapshotMsg = serde_json::from_str(&json_data).unwrap();
1024        let ws_msg = WsMessage::BookSnapshot(msg);
1025
1026        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
1027        let info = Arc::new(TardisInstrumentMiniInfo::new(
1028            instrument_id,
1029            None,
1030            TardisExchange::Bitmex,
1031            1,
1032            0,
1033        ));
1034
1035        let result = parse_tardis_ws_message(ws_msg, &info, &BookSnapshotOutput::Depth);
1036
1037        assert!(result.is_some());
1038        assert!(matches!(result.unwrap(), Data::BookDepth(_)));
1039    }
1040
1041    #[rstest]
1042    fn test_parse_tardis_ws_message_sparse_book_snapshot_routes_to_depth() {
1043        let json_data = r#"{
1044            "type": "book_snapshot",
1045            "symbol": "ETC",
1046            "exchange": "hyperliquid",
1047            "name": "book_snapshot_20_10s",
1048            "depth": 20,
1049            "interval": 10000,
1050            "bids": [{"price": 20.002, "amount": 5.81}],
1051            "asks": [{"price": 20.003, "amount": 162.45}, {}],
1052            "timestamp": "2025-03-03T10:48:10.000Z",
1053            "localTimestamp": "2025-03-03T10:48:10.596818Z"
1054        }"#;
1055        let msg: BookSnapshotMsg = serde_json::from_str(json_data).unwrap();
1056        let ws_msg = WsMessage::BookSnapshot(msg);
1057
1058        let instrument_id = InstrumentId::from("ETC.HYPERLIQUID");
1059        let info = Arc::new(TardisInstrumentMiniInfo::new(
1060            instrument_id,
1061            None,
1062            TardisExchange::Hyperliquid,
1063            3,
1064            2,
1065        ));
1066
1067        let result = parse_tardis_ws_message(ws_msg, &info, &BookSnapshotOutput::Depth);
1068
1069        assert!(result.is_some());
1070        assert!(matches!(result.unwrap(), Data::BookDepth(_)));
1071    }
1072
1073    #[rstest]
1074    fn test_parse_tardis_ws_message_book_snapshot_routes_to_deltas() {
1075        let json_data = load_test_json("book_snapshot.json");
1076        let msg: BookSnapshotMsg = serde_json::from_str(&json_data).unwrap();
1077        let ws_msg = WsMessage::BookSnapshot(msg);
1078
1079        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
1080        let info = Arc::new(TardisInstrumentMiniInfo::new(
1081            instrument_id,
1082            None,
1083            TardisExchange::Bitmex,
1084            1,
1085            0,
1086        ));
1087
1088        let result = parse_tardis_ws_message(ws_msg, &info, &BookSnapshotOutput::Deltas);
1089
1090        assert!(result.is_some());
1091        assert!(matches!(result.unwrap(), Data::BookDeltas(_)));
1092    }
1093
1094    #[rstest]
1095    fn test_parse_tardis_ws_message_mexc_snapshot_5000_routes_to_deltas() {
1096        let json_data = load_test_json("mexc_book_snapshot_5000.json");
1097        let msg: BookSnapshotMsg = serde_json::from_str(&json_data).unwrap();
1098        let ws_msg = WsMessage::BookSnapshot(msg);
1099
1100        let instrument_id = InstrumentId::from("BTCUSDT.MEXC");
1101        let info = Arc::new(TardisInstrumentMiniInfo::new(
1102            instrument_id,
1103            None,
1104            TardisExchange::Mexc,
1105            1,
1106            1,
1107        ));
1108
1109        let result = parse_tardis_ws_message(ws_msg, &info, &BookSnapshotOutput::Deltas);
1110
1111        let Some(Data::BookDeltas(deltas)) = result else {
1112            panic!("Expected Data::BookDeltas, was {result:?}");
1113        };
1114        assert_eq!(deltas.instrument_id, instrument_id);
1115        assert_eq!(deltas.deltas.len(), 7);
1116        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
1117        assert_eq!(deltas.deltas[1].order.price, Price::from("100.0"));
1118        assert_eq!(deltas.deltas[4].order.price, Price::from("101.0"));
1119        assert_eq!(
1120            deltas.deltas[6].flags,
1121            RecordFlag::F_LAST as u8 + RecordFlag::F_SNAPSHOT as u8
1122        );
1123    }
1124
1125    #[rstest]
1126    fn test_parse_tardis_ws_message_option_summary_routes_to_option_greeks() {
1127        let json_data = load_test_json("option_summary.json");
1128        let msg: OptionSummaryMsg = serde_json::from_str(&json_data).unwrap();
1129        let ts_event = UnixNanos::from(msg.timestamp);
1130        let ts_init = UnixNanos::from(msg.local_timestamp);
1131        let ws_msg = WsMessage::OptionSummary(msg);
1132
1133        let instrument_id = InstrumentId::from("BTC-28JUN24-70000-C.DERIBIT");
1134        let info = Arc::new(TardisInstrumentMiniInfo::new(
1135            instrument_id,
1136            None,
1137            TardisExchange::Deribit,
1138            4,
1139            1,
1140        ));
1141
1142        let result = parse_tardis_ws_message(ws_msg, &info, &BookSnapshotOutput::Deltas);
1143
1144        let Some(Data::OptionGreeks(greeks)) = result else {
1145            panic!("Expected Data::OptionGreeks, was {result:?}");
1146        };
1147        assert_eq!(greeks.instrument_id, instrument_id);
1148        assert_eq!(greeks.convention, GreeksConvention::BlackScholes);
1149        assert_eq!(greeks.greeks.delta, 0.25);
1150        assert_eq!(greeks.greeks.gamma, 0.00002);
1151        assert_eq!(greeks.greeks.vega, 45.5);
1152        assert_eq!(greeks.greeks.theta, -15.2);
1153        assert_eq!(greeks.greeks.rho, 0.05);
1154        assert_eq!(greeks.mark_iv, Some(0.565));
1155        assert_eq!(greeks.bid_iv, Some(0.55));
1156        assert_eq!(greeks.ask_iv, Some(0.58));
1157        assert_eq!(greeks.underlying_price, Some(63_500.0));
1158        assert_eq!(greeks.open_interest, Some(150.0));
1159        assert_eq!(greeks.ts_event, ts_event);
1160        assert_eq!(greeks.ts_init, ts_init);
1161    }
1162
1163    #[rstest]
1164    fn test_parse_tardis_ws_message_data_extracts_option_summary_quote_when_enabled() {
1165        let json_data = load_test_json("option_summary.json");
1166        let msg: OptionSummaryMsg = serde_json::from_str(&json_data).unwrap();
1167        let ts_event = UnixNanos::from(msg.timestamp);
1168        let ts_init = UnixNanos::from(msg.local_timestamp);
1169        let ws_msg = WsMessage::OptionSummary(msg);
1170
1171        let instrument_id = InstrumentId::from("BTC-28JUN24-70000-C.DERIBIT");
1172        let info = Arc::new(TardisInstrumentMiniInfo::new(
1173            instrument_id,
1174            None,
1175            TardisExchange::Deribit,
1176            4,
1177            1,
1178        ));
1179
1180        let data = parse_tardis_ws_message_data(ws_msg, &info, &BookSnapshotOutput::Deltas, true);
1181
1182        assert_eq!(data.len(), 2);
1183        let Data::Quote(quote) = &data[0] else {
1184            panic!("Expected first data item to be QuoteTick");
1185        };
1186        let Data::OptionGreeks(greeks) = &data[1] else {
1187            panic!("Expected second data item to be OptionGreeks");
1188        };
1189        assert_eq!(quote.instrument_id, instrument_id);
1190        assert_eq!(quote.bid_price, Price::from("0.035"));
1191        assert_eq!(quote.bid_size, Quantity::from("5"));
1192        assert_eq!(quote.ask_price, Price::from("0.04"));
1193        assert_eq!(quote.ask_size, Quantity::from("10"));
1194        assert_eq!(quote.ts_event, ts_event);
1195        assert_eq!(quote.ts_init, ts_init);
1196        assert_eq!(greeks.instrument_id, instrument_id);
1197        assert_eq!(greeks.ts_event, ts_event);
1198        assert_eq!(greeks.ts_init, ts_init);
1199    }
1200
1201    #[rstest]
1202    fn test_parse_tardis_ws_message_data_skips_option_summary_quote_when_disabled() {
1203        let json_data = load_test_json("option_summary.json");
1204        let msg: OptionSummaryMsg = serde_json::from_str(&json_data).unwrap();
1205        let ws_msg = WsMessage::OptionSummary(msg);
1206
1207        let instrument_id = InstrumentId::from("BTC-28JUN24-70000-C.DERIBIT");
1208        let info = Arc::new(TardisInstrumentMiniInfo::new(
1209            instrument_id,
1210            None,
1211            TardisExchange::Deribit,
1212            4,
1213            1,
1214        ));
1215
1216        let data = parse_tardis_ws_message_data(ws_msg, &info, &BookSnapshotOutput::Deltas, false);
1217
1218        assert_eq!(data.len(), 1);
1219        assert!(
1220            matches!(data[0], Data::OptionGreeks(_)),
1221            "Expected OptionGreeks, was {:?}",
1222            data[0]
1223        );
1224    }
1225
1226    #[rstest]
1227    fn test_parse_tardis_ws_message_data_skips_option_summary_quote_when_bbo_missing() {
1228        let json_data = load_test_json("option_summary.json");
1229        let mut msg: OptionSummaryMsg = serde_json::from_str(&json_data).unwrap();
1230        msg.best_ask_price = None;
1231        let ws_msg = WsMessage::OptionSummary(msg);
1232
1233        let instrument_id = InstrumentId::from("BTC-28JUN24-70000-C.DERIBIT");
1234        let info = Arc::new(TardisInstrumentMiniInfo::new(
1235            instrument_id,
1236            None,
1237            TardisExchange::Deribit,
1238            4,
1239            1,
1240        ));
1241
1242        let data = parse_tardis_ws_message_data(ws_msg, &info, &BookSnapshotOutput::Deltas, true);
1243
1244        assert_eq!(data.len(), 1);
1245        assert!(
1246            matches!(data[0], Data::OptionGreeks(_)),
1247            "Expected OptionGreeks, was {:?}",
1248            data[0]
1249        );
1250    }
1251
1252    #[rstest]
1253    fn test_parse_tardis_ws_message_data_keeps_option_summary_when_bbo_size_invalid() {
1254        let json_data = load_test_json("option_summary.json");
1255        let mut msg: OptionSummaryMsg = serde_json::from_str(&json_data).unwrap();
1256        let ts_event = UnixNanos::from(msg.timestamp);
1257        let ts_init = UnixNanos::from(msg.local_timestamp);
1258        msg.best_bid_amount = Some(Decimal::ZERO);
1259        let ws_msg = WsMessage::OptionSummary(msg);
1260
1261        let instrument_id = InstrumentId::from("BTC-28JUN24-70000-C.DERIBIT");
1262        let info = Arc::new(TardisInstrumentMiniInfo::new(
1263            instrument_id,
1264            None,
1265            TardisExchange::Deribit,
1266            4,
1267            1,
1268        ));
1269
1270        let data = parse_tardis_ws_message_data(ws_msg, &info, &BookSnapshotOutput::Deltas, true);
1271
1272        assert_eq!(data.len(), 1);
1273        let Data::OptionGreeks(greeks) = &data[0] else {
1274            panic!("Expected OptionGreeks, was {:?}", data[0]);
1275        };
1276        assert_eq!(greeks.instrument_id, instrument_id);
1277        assert_eq!(greeks.ts_event, ts_event);
1278        assert_eq!(greeks.ts_init, ts_init);
1279    }
1280
1281    #[rstest]
1282    fn test_parse_tardis_ws_message_data_keeps_option_summary_when_bbo_price_invalid() {
1283        let json_data = load_test_json("option_summary.json");
1284        let mut msg: OptionSummaryMsg = serde_json::from_str(&json_data).unwrap();
1285        let ts_event = UnixNanos::from(msg.timestamp);
1286        let ts_init = UnixNanos::from(msg.local_timestamp);
1287        msg.best_bid_price = Some(Decimal::MAX);
1288        let ws_msg = WsMessage::OptionSummary(msg);
1289
1290        let instrument_id = InstrumentId::from("BTC-28JUN24-70000-C.DERIBIT");
1291        let info = Arc::new(TardisInstrumentMiniInfo::new(
1292            instrument_id,
1293            None,
1294            TardisExchange::Deribit,
1295            4,
1296            1,
1297        ));
1298
1299        let data = parse_tardis_ws_message_data(ws_msg, &info, &BookSnapshotOutput::Deltas, true);
1300
1301        assert_eq!(data.len(), 1);
1302        let Data::OptionGreeks(greeks) = &data[0] else {
1303            panic!("Expected OptionGreeks, was {:?}", data[0]);
1304        };
1305        assert_eq!(greeks.instrument_id, instrument_id);
1306        assert_eq!(greeks.ts_event, ts_event);
1307        assert_eq!(greeks.ts_init, ts_init);
1308    }
1309
1310    #[rstest]
1311    fn test_parse_option_summary_msg_defaults_absent_fields() {
1312        let ts = "2024-01-15T10:30:00.123Z".parse::<Timestamp>().unwrap();
1313        let msg = OptionSummaryMsg {
1314            symbol: ustr::Ustr::from("BTC-28JUN24-70000-C"),
1315            exchange: TardisExchange::Deribit,
1316            option_type: "call".to_string(),
1317            strike_price: 70_000.0,
1318            expiration_date: ts,
1319            best_bid_price: None,
1320            best_bid_amount: None,
1321            best_bid_iv: None,
1322            best_ask_price: None,
1323            best_ask_amount: None,
1324            best_ask_iv: None,
1325            last_price: None,
1326            open_interest: None,
1327            mark_price: None,
1328            mark_iv: None,
1329            delta: None,
1330            gamma: None,
1331            vega: None,
1332            theta: None,
1333            rho: None,
1334            underlying_price: None,
1335            underlying_index: "BTC-USD".to_string(),
1336            timestamp: ts,
1337            local_timestamp: ts,
1338        };
1339
1340        let instrument_id = InstrumentId::from("BTC-28JUN24-70000-C.DERIBIT");
1341        let greeks = parse_option_summary_msg(&msg, instrument_id);
1342
1343        // Absent greeks default to 0.0; absent IVs, underlying, and open interest stay None.
1344        assert_eq!(greeks.greeks.delta, 0.0);
1345        assert_eq!(greeks.greeks.gamma, 0.0);
1346        assert_eq!(greeks.greeks.vega, 0.0);
1347        assert_eq!(greeks.greeks.theta, 0.0);
1348        assert_eq!(greeks.greeks.rho, 0.0);
1349        assert_eq!(greeks.mark_iv, None);
1350        assert_eq!(greeks.bid_iv, None);
1351        assert_eq!(greeks.ask_iv, None);
1352        assert_eq!(greeks.underlying_price, None);
1353        assert_eq!(greeks.open_interest, None);
1354    }
1355
1356    #[rstest]
1357    fn test_parse_derivative_ticker_funding_rate() {
1358        let json_data = load_test_json("derivative_ticker.json");
1359        let msg: DerivativeTickerMsg = serde_json::from_str(&json_data).unwrap();
1360
1361        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
1362
1363        let result = parse_derivative_ticker_msg(&msg, instrument_id).unwrap();
1364        assert!(result.is_some());
1365
1366        let funding = result.unwrap();
1367        assert_eq!(funding.instrument_id, instrument_id);
1368        assert_eq!(funding.rate.to_string(), "-0.00001568");
1369        assert!(funding.ts_event.as_u64() > 0);
1370        assert!(funding.ts_init.as_u64() > 0);
1371    }
1372
1373    #[rstest]
1374    fn test_parse_derivative_ticker_mark_price() {
1375        let json_data = load_test_json("derivative_ticker.json");
1376        let msg: DerivativeTickerMsg = serde_json::from_str(&json_data).unwrap();
1377
1378        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
1379        let price_precision = 2;
1380
1381        let result =
1382            parse_derivative_ticker_mark_price(&msg, instrument_id, price_precision).unwrap();
1383        assert!(result.is_some());
1384
1385        let mark = result.unwrap();
1386        assert_eq!(mark.instrument_id, instrument_id);
1387        assert_eq!(mark.value, Price::new(7987.56, price_precision));
1388        assert!(mark.ts_event.as_u64() > 0);
1389        assert!(mark.ts_init.as_u64() > 0);
1390    }
1391
1392    #[rstest]
1393    fn test_parse_derivative_ticker_index_price() {
1394        let json_data = load_test_json("derivative_ticker.json");
1395        let msg: DerivativeTickerMsg = serde_json::from_str(&json_data).unwrap();
1396
1397        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
1398        let price_precision = 2;
1399
1400        let result =
1401            parse_derivative_ticker_index_price(&msg, instrument_id, price_precision).unwrap();
1402        assert!(result.is_some());
1403
1404        let index = result.unwrap();
1405        assert_eq!(index.instrument_id, instrument_id);
1406        assert_eq!(index.value, Price::new(7989.28, price_precision));
1407        assert!(index.ts_event.as_u64() > 0);
1408        assert!(index.ts_init.as_u64() > 0);
1409    }
1410
1411    #[rstest]
1412    fn test_parse_derivative_ticker_missing_fields() {
1413        // Test with minimal data (only funding_rate, no mark/index)
1414        let json = r#"{
1415            "type": "derivative_ticker",
1416            "symbol": "BTCUSD",
1417            "exchange": "bitmex",
1418            "lastPrice": null,
1419            "openInterest": null,
1420            "fundingRate": 0.0001,
1421            "indexPrice": null,
1422            "markPrice": null,
1423            "timestamp": "2024-01-01T00:00:00.000Z",
1424            "localTimestamp": "2024-01-01T00:00:00.100Z"
1425        }"#;
1426        let msg: DerivativeTickerMsg = serde_json::from_str(json).unwrap();
1427
1428        let instrument_id = InstrumentId::from("BTCUSD.BITMEX");
1429
1430        let funding = parse_derivative_ticker_msg(&msg, instrument_id).unwrap();
1431        assert_eq!(funding.unwrap().next_funding_ns, None);
1432
1433        let mark = parse_derivative_ticker_mark_price(&msg, instrument_id, 1).unwrap();
1434        assert!(mark.is_none());
1435
1436        let index = parse_derivative_ticker_index_price(&msg, instrument_id, 1).unwrap();
1437        assert!(index.is_none());
1438    }
1439
1440    #[rstest]
1441    fn test_parse_mexc_futures_derivative_ticker() {
1442        let json_data = load_test_json("mexc_futures_derivative_ticker.json");
1443        let msg: DerivativeTickerMsg = serde_json::from_str(&json_data).unwrap();
1444
1445        let instrument_id = InstrumentId::from("BTC_USDT-PERP.MEXC");
1446
1447        let funding = parse_derivative_ticker_msg(&msg, instrument_id).unwrap();
1448        let mark = parse_derivative_ticker_mark_price(&msg, instrument_id, 1).unwrap();
1449        let index = parse_derivative_ticker_index_price(&msg, instrument_id, 1).unwrap();
1450
1451        assert_eq!(msg.exchange, TardisExchange::MexcFutures);
1452        assert_eq!(funding.unwrap().rate.to_string(), "0.0001");
1453        assert_eq!(mark.unwrap().value, Price::new(61235.2, 1));
1454        assert_eq!(index.unwrap().value, Price::new(61230.1, 1));
1455    }
1456
1457    #[rstest]
1458    fn test_parse_okex_xperp_derivative_ticker() {
1459        let json_data = load_test_json("okex_futures_xperp_derivative_ticker.json");
1460        let msg: DerivativeTickerMsg = serde_json::from_str(&json_data).unwrap();
1461
1462        let instrument_id = parse_instrument_id(&msg.exchange, msg.symbol);
1463        let funding = parse_derivative_ticker_msg(&msg, instrument_id)
1464            .unwrap()
1465            .unwrap();
1466        let mark = parse_derivative_ticker_mark_price(&msg, instrument_id, 1)
1467            .unwrap()
1468            .unwrap();
1469        let index = parse_derivative_ticker_index_price(&msg, instrument_id, 1)
1470            .unwrap()
1471            .unwrap();
1472
1473        // OKX X-Perps publish no predicted rate, so the funding timestamp is the only forward
1474        // reference carried through normalization
1475        assert_eq!(msg.exchange, TardisExchange::OkexFutures);
1476        assert_eq!(
1477            instrument_id,
1478            InstrumentId::from("BTC-USD_UM_XPERP-310404.OKEX")
1479        );
1480        assert_eq!(funding.rate, dec!(-0.0004050736802759));
1481        assert_eq!(
1482            funding.next_funding_ns,
1483            Some(UnixNanos::from("2026-08-10T16:00:00Z"))
1484        );
1485        assert_eq!(funding.interval, None);
1486        assert_eq!(
1487            funding.ts_event,
1488            UnixNanos::from("2026-08-10T12:00:13.061Z")
1489        );
1490        assert_eq!(funding.ts_init, UnixNanos::from("2026-08-10T12:00:13.093Z"));
1491        assert_eq!(mark.value, Price::new(65014.7, 1));
1492        assert_eq!(index.value, Price::new(65066.1, 1));
1493    }
1494
1495    #[rstest]
1496    fn test_parse_okex_usdc_derivative_ticker_across_index_migration() {
1497        let pre_json = load_test_json("okex_swap_derivative_ticker_pre_index_migration.json");
1498        let post_json = load_test_json("okex_swap_derivative_ticker_post_index_migration.json");
1499        let pre: DerivativeTickerMsg = serde_json::from_str(&pre_json).unwrap();
1500        let post: DerivativeTickerMsg = serde_json::from_str(&post_json).unwrap();
1501
1502        let pre_id = parse_instrument_id(&pre.exchange, pre.symbol);
1503        let post_id = parse_instrument_id(&post.exchange, post.symbol);
1504        let pre_index = parse_derivative_ticker_index_price(&pre, pre_id, 1)
1505            .unwrap()
1506            .unwrap();
1507        let post_index = parse_derivative_ticker_index_price(&post, post_id, 1)
1508            .unwrap()
1509            .unwrap();
1510        let pre_funding = parse_derivative_ticker_msg(&pre, pre_id).unwrap().unwrap();
1511        let post_funding = parse_derivative_ticker_msg(&post, post_id)
1512            .unwrap()
1513            .unwrap();
1514
1515        // Captured either side of the 2023-04-10T08:40Z index migration, which switches the index
1516        // feed Tardis reads from BTC-USD to BTC-USDC but leaves the contract symbol alone, so both
1517        // must still resolve to one instrument. The index prices are a month apart and only pin
1518        // each capture, they do not themselves demonstrate the switch.
1519        assert_eq!(pre.exchange, TardisExchange::OkexSwap);
1520        assert_eq!(post.exchange, TardisExchange::OkexSwap);
1521        assert_eq!(pre_id, InstrumentId::from("BTC-USDC-SWAP.OKEX"));
1522        assert_eq!(post_id, pre_id);
1523        assert_eq!(pre_index.value, Price::new(28379.9, 1));
1524        assert_eq!(post_index.value, Price::new(28525.1, 1));
1525        assert_eq!(pre_funding.rate, dec!(-0.0001546799749051));
1526        assert_eq!(post_funding.rate, dec!(0.0001309027042856));
1527        assert_eq!(
1528            pre_funding.next_funding_ns,
1529            Some(UnixNanos::from("2023-04-01T16:00:00Z"))
1530        );
1531        assert_eq!(
1532            post_funding.next_funding_ns,
1533            Some(UnixNanos::from("2023-05-01T16:00:00Z"))
1534        );
1535    }
1536
1537    fn load_exact_decimal_messages() -> Vec<WsMessage> {
1538        let json_data = load_test_json("exact_decimals.json");
1539        serde_json::from_str(&json_data).unwrap()
1540    }
1541
1542    #[rstest]
1543    fn test_parse_trade_msg_keeps_decimals_that_collapse_in_f64() {
1544        let messages = load_exact_decimal_messages();
1545
1546        let (WsMessage::Trade(first), WsMessage::Trade(second)) = (&messages[0], &messages[1])
1547        else {
1548            panic!("Expected trade messages, was {messages:?}");
1549        };
1550
1551        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
1552
1553        let first_trade = parse_trade_msg(first, 9, 9, instrument_id).unwrap();
1554        let second_trade = parse_trade_msg(second, 9, 9, instrument_id).unwrap();
1555
1556        let first_f64 = "100000000.123456789".parse::<f64>().unwrap();
1557        let second_f64 = "100000000.123456788".parse::<f64>().unwrap();
1558        assert_eq!(first_f64.to_bits(), second_f64.to_bits());
1559        assert_eq!(first_trade.price, Price::from("100000000.123456789"));
1560        assert_eq!(first_trade.size, Quantity::from("100000000.123456789"));
1561        assert_eq!(second_trade.price, Price::from("100000000.123456788"));
1562        assert_eq!(second_trade.size, Quantity::from("100000000.123456788"));
1563        assert_ne!(first_trade.trade_id, second_trade.trade_id);
1564    }
1565
1566    #[rstest]
1567    fn test_parse_trade_msg_rounds_exact_token_half_even() {
1568        let messages = load_exact_decimal_messages();
1569
1570        let WsMessage::Trade(msg) = &messages[2] else {
1571            panic!("Expected trade message, was {:?}", messages[2]);
1572        };
1573
1574        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
1575
1576        let trade = parse_trade_msg(msg, 2, 2, instrument_id).unwrap();
1577
1578        let token_f64 = "100000000.005000001".parse::<f64>().unwrap();
1579        let via_f64 = Price::from_decimal_dp(Decimal::try_from(token_f64).unwrap(), 2).unwrap();
1580        assert_eq!(via_f64, Price::from("100000000.00"));
1581        assert_eq!(trade.price, Price::from("100000000.01"));
1582        assert_eq!(trade.size, Quantity::from("100000000.01"));
1583    }
1584
1585    #[rstest]
1586    fn test_parse_book_change_msg_keeps_decimals_that_collapse_in_f64() {
1587        let messages = load_exact_decimal_messages();
1588
1589        let WsMessage::BookChange(msg) = &messages[3] else {
1590            panic!("Expected book change message, was {:?}", messages[3]);
1591        };
1592
1593        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
1594
1595        let deltas = parse_book_change_msg_as_deltas(msg, 9, 9, instrument_id).unwrap();
1596
1597        assert_eq!(deltas.deltas.len(), 2);
1598        assert_eq!(deltas.deltas[0].order.side, Some(OrderSide::Buy));
1599        assert_eq!(
1600            deltas.deltas[0].order.price,
1601            Price::from("100000000.123456789")
1602        );
1603        assert_eq!(
1604            deltas.deltas[0].order.size,
1605            Quantity::from("100000000.123456789")
1606        );
1607        assert_eq!(deltas.deltas[1].order.side, Some(OrderSide::Sell));
1608        assert_eq!(
1609            deltas.deltas[1].order.price,
1610            Price::from("100000000.123456788")
1611        );
1612        assert_eq!(
1613            deltas.deltas[1].order.size,
1614            Quantity::from("100000000.123456788")
1615        );
1616    }
1617
1618    #[rstest]
1619    fn test_parse_derivative_ticker_msg_keeps_exact_tokens() {
1620        let messages = load_exact_decimal_messages();
1621
1622        let WsMessage::DerivativeTicker(msg) = &messages[4] else {
1623            panic!("Expected derivative ticker message, was {:?}", messages[4]);
1624        };
1625
1626        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
1627
1628        let funding = parse_derivative_ticker_msg(msg, instrument_id)
1629            .unwrap()
1630            .unwrap();
1631        let mark = parse_derivative_ticker_mark_price(msg, instrument_id, 9)
1632            .unwrap()
1633            .unwrap();
1634        let index = parse_derivative_ticker_index_price(msg, instrument_id, 9)
1635            .unwrap()
1636            .unwrap();
1637
1638        assert_eq!(funding.rate, dec!(0.30000000000000004));
1639        assert_eq!(mark.value, Price::from("100000000.123456788"));
1640        assert_eq!(index.value, Price::from("100000000.123456789"));
1641    }
1642
1643    #[rstest]
1644    fn test_parse_book_change_msg_normalizes_float_noise_amount_to_delete() {
1645        let messages = load_exact_decimal_messages();
1646
1647        let WsMessage::BookChange(msg) = &messages[5] else {
1648            panic!("Expected book change message, was {:?}", messages[5]);
1649        };
1650
1651        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
1652
1653        let deltas = parse_book_change_msg_as_deltas(msg, 1, 0, instrument_id).unwrap();
1654
1655        assert_eq!(msg.bids[0].amount, dec!(0.0000000000000001110223024625));
1656        assert_eq!(deltas.deltas.len(), 1);
1657        assert_eq!(deltas.deltas[0].action, BookAction::Delete);
1658        assert_eq!(deltas.deltas[0].order.price, Price::from("7984.5"));
1659        assert_eq!(deltas.deltas[0].order.size, Quantity::from("0"));
1660    }
1661
1662    #[rstest]
1663    fn test_parse_trade_msg_derives_trade_id_from_decimal_value() {
1664        let messages = load_exact_decimal_messages();
1665
1666        let (WsMessage::Trade(padded), WsMessage::Trade(plain)) = (&messages[6], &messages[7])
1667        else {
1668            panic!("Expected trade messages, was {messages:?}");
1669        };
1670
1671        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
1672
1673        let padded_trade = parse_trade_msg(padded, 1, 0, instrument_id).unwrap();
1674        let plain_trade = parse_trade_msg(plain, 1, 0, instrument_id).unwrap();
1675        let csv_trade_id = derive_trade_id(
1676            ustr::Ustr::from("XBTUSD"),
1677            padded_trade.ts_event.as_u64(),
1678            &7996.5_f64.to_string(),
1679            &50.0_f64.to_string(),
1680            "sell",
1681        );
1682
1683        assert_eq!(padded.price.to_string(), "7996.50");
1684        assert_eq!(padded.amount.to_string(), "50.0");
1685        assert_eq!(padded_trade.trade_id, plain_trade.trade_id);
1686        assert_eq!(padded_trade.trade_id, csv_trade_id);
1687    }
1688
1689    #[rstest]
1690    fn test_parse_book_change_msg_truncates_amounts_and_skips_invalid_price() {
1691        let messages = load_exact_decimal_messages();
1692
1693        let WsMessage::BookChange(msg) = &messages[8] else {
1694            panic!("Expected book change message, was {:?}", messages[8]);
1695        };
1696
1697        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
1698
1699        let deltas = parse_book_change_msg_as_deltas(msg, 1, 2, instrument_id).unwrap();
1700
1701        assert_eq!(deltas.deltas.len(), 2);
1702        assert_eq!(deltas.deltas[0].order.price, Price::from("7984.5"));
1703        assert_eq!(deltas.deltas[0].order.size, Quantity::from("0.12"));
1704        assert_eq!(deltas.deltas[1].order.price, Price::from("7984.0"));
1705        assert_eq!(deltas.deltas[1].order.size, Quantity::from("3.00"));
1706        assert_eq!(deltas.deltas[1].flags, RecordFlag::F_LAST as u8);
1707    }
1708
1709    #[rstest]
1710    #[case(dec!(0.015), "0.02")]
1711    #[case(dec!(0.006), "0.01")]
1712    fn test_parse_non_zero_quantity_rounds_half_even(
1713        #[case] amount: Decimal,
1714        #[case] expected: &str,
1715    ) {
1716        let quantity = parse_non_zero_quantity(amount, 2).unwrap();
1717
1718        assert_eq!(quantity, Quantity::from(expected));
1719    }
1720
1721    #[rstest]
1722    #[case(dec!(0), "value was zero")]
1723    #[case(dec!(0.004), "value 0.004 was zero after rounding to precision 2")]
1724    #[case(dec!(0.005), "value 0.005 was zero after rounding to precision 2")]
1725    fn test_parse_non_zero_quantity_rejects_zero(#[case] amount: Decimal, #[case] expected: &str) {
1726        let error = parse_non_zero_quantity(amount, 2).unwrap_err();
1727
1728        assert_eq!(error.to_string(), expected);
1729    }
1730}
nautilus_tardis/machine/parse.rs

nautilus_tardis/machine/
parse.rs
