nautilus_tardis/machine/
message.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 jiff::Timestamp;
17use nautilus_core::serialization::{
18    deserialize_decimal_token_borrowed, deserialize_optional_decimal_token_borrowed,
19};
20use rust_decimal::Decimal;
21use serde::{Deserialize, Deserializer, Serialize, de::Error};
22use serde_json::value::RawValue;
23use ustr::Ustr;
24
25use crate::common::{enums::TardisExchange, parse::deserialize_uppercase};
26
27/// Represents a single level in the order book (bid or ask).
28#[derive(Debug, Clone, Deserialize, Serialize)]
29pub struct BookLevel {
30    /// The price at this level.
31    pub price: Decimal,
32    /// The amount at this level.
33    pub amount: Decimal,
34}
35
36/// Represents a Tardis WebSocket message for book changes.
37#[derive(Debug, Clone, Deserialize, Serialize)]
38#[serde(rename_all = "camelCase")]
39pub struct BookChangeMsg {
40    /// The symbol as provided by the exchange.
41    #[serde(deserialize_with = "deserialize_uppercase")]
42    pub symbol: Ustr,
43    /// The exchange ID.
44    pub exchange: TardisExchange,
45    /// Indicates whether this is an initial order book snapshot.
46    pub is_snapshot: bool,
47    /// Updated bids, with price and amount levels.
48    #[serde(deserialize_with = "deserialize_book_levels")]
49    pub bids: Vec<BookLevel>,
50    /// Updated asks, with price and amount levels.
51    #[serde(deserialize_with = "deserialize_book_levels")]
52    pub asks: Vec<BookLevel>,
53    /// The order book update timestamp provided by the exchange (ISO 8601 format).
54    pub timestamp: Timestamp,
55    /// The local timestamp when the message was received.
56    pub local_timestamp: Timestamp,
57}
58
59/// Represents a Tardis WebSocket message for book snapshots.
60#[derive(Debug, Clone, Deserialize, Serialize)]
61#[serde(rename_all = "camelCase")]
62pub struct BookSnapshotMsg {
63    /// The symbol as provided by the exchange.
64    #[serde(deserialize_with = "deserialize_uppercase")]
65    pub symbol: Ustr,
66    /// The exchange ID.
67    pub exchange: TardisExchange,
68    /// The name of the snapshot, e.g., `book_snapshot_{depth}_{interval}{time_unit}`.
69    pub name: String,
70    /// The requested number of levels (top bids/asks).
71    pub depth: u32,
72    /// The requested snapshot interval in milliseconds.
73    pub interval: u32,
74    /// The top bids price-amount levels.
75    #[serde(deserialize_with = "deserialize_book_levels")]
76    pub bids: Vec<BookLevel>,
77    /// The top asks price-amount levels.
78    #[serde(deserialize_with = "deserialize_book_levels")]
79    pub asks: Vec<BookLevel>,
80    /// The snapshot timestamp based on the last book change message processed timestamp.
81    pub timestamp: Timestamp,
82    /// The local timestamp when the message was received.
83    pub local_timestamp: Timestamp,
84}
85
86/// Represents a Tardis WebSocket message for trades.
87#[derive(Debug, Clone, Deserialize, Serialize)]
88#[serde(tag = "type")]
89#[serde(rename_all = "camelCase")]
90pub struct TradeMsg {
91    /// The symbol as provided by the exchange.
92    #[serde(deserialize_with = "deserialize_uppercase")]
93    pub symbol: Ustr,
94    /// The exchange ID.
95    pub exchange: TardisExchange,
96    /// The trade ID provided by the exchange (optional).
97    pub id: Option<String>,
98    /// The trade price as provided by the exchange.
99    #[serde(deserialize_with = "deserialize_decimal_token_borrowed")]
100    pub price: Decimal,
101    /// The trade amount as provided by the exchange.
102    #[serde(deserialize_with = "deserialize_decimal_token_borrowed")]
103    pub amount: Decimal,
104    /// The liquidity taker side (aggressor) for the trade.
105    pub side: String,
106    /// The trade timestamp provided by the exchange.
107    pub timestamp: Timestamp,
108    /// The local timestamp when the message was received.
109    pub local_timestamp: Timestamp,
110}
111
112/// Derivative instrument ticker info sourced from real-time ticker & instrument channels.
113#[derive(Debug, Clone, Serialize, Deserialize)]
114#[serde(rename_all = "camelCase")]
115pub struct DerivativeTickerMsg {
116    /// The symbol as provided by the exchange.
117    #[serde(deserialize_with = "deserialize_uppercase")]
118    pub symbol: Ustr,
119    /// The exchange ID.
120    pub exchange: TardisExchange,
121    /// The last instrument price if provided by exchange.
122    pub last_price: Option<f64>,
123    /// The last open interest if provided by exchange.
124    pub open_interest: Option<f64>,
125    /// The last funding rate if provided by exchange.
126    #[serde(
127        default,
128        deserialize_with = "deserialize_optional_decimal_token_borrowed"
129    )]
130    pub funding_rate: Option<Decimal>,
131    /// The timestamp of the next funding if provided by exchange.
132    pub funding_timestamp: Option<Timestamp>,
133    /// The last index price if provided by exchange.
134    #[serde(
135        default,
136        deserialize_with = "deserialize_optional_decimal_token_borrowed"
137    )]
138    pub index_price: Option<Decimal>,
139    /// The last mark price if provided by exchange.
140    #[serde(
141        default,
142        deserialize_with = "deserialize_optional_decimal_token_borrowed"
143    )]
144    pub mark_price: Option<Decimal>,
145    /// The message timestamp provided by exchange.
146    pub timestamp: Timestamp,
147    /// The local timestamp when the message was received.
148    pub local_timestamp: Timestamp,
149}
150
151/// Option summary info sourced from the options instrument channel, carrying exchange-provided
152/// greeks, implied volatilities, mark and underlying prices, and best bid/ask for a single option.
153#[derive(Debug, Clone, Serialize, Deserialize)]
154#[serde(rename_all = "camelCase")]
155pub struct OptionSummaryMsg {
156    /// The symbol as provided by the exchange.
157    #[serde(deserialize_with = "deserialize_uppercase")]
158    pub symbol: Ustr,
159    /// The exchange ID.
160    pub exchange: TardisExchange,
161    /// The option type, either `put` or `call`.
162    pub option_type: String,
163    /// The option strike price.
164    pub strike_price: f64,
165    /// The option expiration date provided by the exchange.
166    pub expiration_date: Timestamp,
167    /// The best bid price if provided by the exchange.
168    #[serde(
169        default,
170        deserialize_with = "deserialize_optional_decimal_token_borrowed"
171    )]
172    pub best_bid_price: Option<Decimal>,
173    /// The best bid amount if provided by the exchange.
174    #[serde(
175        default,
176        deserialize_with = "deserialize_optional_decimal_token_borrowed"
177    )]
178    pub best_bid_amount: Option<Decimal>,
179    /// The best bid implied volatility if provided by the exchange.
180    #[serde(rename = "bestBidIV")]
181    pub best_bid_iv: Option<f64>,
182    /// The best ask price if provided by the exchange.
183    #[serde(
184        default,
185        deserialize_with = "deserialize_optional_decimal_token_borrowed"
186    )]
187    pub best_ask_price: Option<Decimal>,
188    /// The best ask amount if provided by the exchange.
189    #[serde(
190        default,
191        deserialize_with = "deserialize_optional_decimal_token_borrowed"
192    )]
193    pub best_ask_amount: Option<Decimal>,
194    /// The best ask implied volatility if provided by the exchange.
195    #[serde(rename = "bestAskIV")]
196    pub best_ask_iv: Option<f64>,
197    /// The last trade price if provided by the exchange.
198    pub last_price: Option<f64>,
199    /// The open interest if provided by the exchange.
200    pub open_interest: Option<f64>,
201    /// The mark price if provided by the exchange.
202    pub mark_price: Option<f64>,
203    /// The mark implied volatility if provided by the exchange.
204    #[serde(rename = "markIV")]
205    pub mark_iv: Option<f64>,
206    /// The option delta if provided by the exchange.
207    pub delta: Option<f64>,
208    /// The option gamma if provided by the exchange.
209    pub gamma: Option<f64>,
210    /// The option vega if provided by the exchange.
211    pub vega: Option<f64>,
212    /// The option theta if provided by the exchange.
213    pub theta: Option<f64>,
214    /// The option rho if provided by the exchange.
215    pub rho: Option<f64>,
216    /// The underlying price if provided by the exchange.
217    pub underlying_price: Option<f64>,
218    /// The underlying index name.
219    pub underlying_index: String,
220    /// The message timestamp provided by the exchange.
221    pub timestamp: Timestamp,
222    /// The local timestamp when the message was received.
223    pub local_timestamp: Timestamp,
224}
225
226/// Trades data in aggregated form, known as OHLC, candlesticks, klines etc. Not only most common
227/// time based aggregation is supported, but volume and tick count based as well. Bars are computed
228/// from tick-by-tick raw trade data, if in given interval no trades happened, there is no bar produced.
229#[derive(Debug, Clone, Serialize, Deserialize)]
230#[serde(rename_all = "camelCase")]
231pub struct BarMsg {
232    /// The symbol as provided by the exchange.
233    #[serde(deserialize_with = "deserialize_uppercase")]
234    pub symbol: Ustr,
235    /// The exchange ID.
236    pub exchange: TardisExchange,
237    /// name with format `trade_bar`_{interval}
238    pub name: String,
239    /// The requested trade bar interval.
240    pub interval: u64,
241    /// The open price.
242    #[serde(deserialize_with = "deserialize_decimal_token_borrowed")]
243    pub open: Decimal,
244    /// The high price.
245    #[serde(deserialize_with = "deserialize_decimal_token_borrowed")]
246    pub high: Decimal,
247    /// The low price.
248    #[serde(deserialize_with = "deserialize_decimal_token_borrowed")]
249    pub low: Decimal,
250    /// The close price.
251    #[serde(deserialize_with = "deserialize_decimal_token_borrowed")]
252    pub close: Decimal,
253    /// The total volume traded in given interval.
254    #[serde(deserialize_with = "deserialize_decimal_token_borrowed")]
255    pub volume: Decimal,
256    /// The buy volume traded in given interval.
257    pub buy_volume: f64,
258    /// The sell volume traded in given interval.
259    pub sell_volume: f64,
260    /// The trades count in given interval.
261    pub trades: u64,
262    /// The volume weighted average price.
263    pub vwap: f64,
264    /// The timestamp of first trade for given bar.
265    pub open_timestamp: Timestamp,
266    /// The timestamp of last trade for given bar.
267    pub close_timestamp: Timestamp,
268    /// The end of interval period timestamp.
269    pub timestamp: Timestamp,
270    /// The message arrival timestamp that triggered given bar computation.
271    pub local_timestamp: Timestamp,
272}
273
274/// Message that marks events when real-time WebSocket connection that was used to collect the
275/// historical data got disconnected.
276#[derive(Debug, Clone, Serialize, Deserialize)]
277#[serde(rename_all = "camelCase")]
278pub struct DisconnectMsg {
279    /// The exchange ID.
280    pub exchange: TardisExchange,
281    /// The message arrival timestamp that triggered given bar computation (ISO 8601 format).
282    pub local_timestamp: Timestamp,
283}
284
285/// A Tardis Machine Server message type.
286///
287/// Deserializes from any `serde_json` input. The message structs read decimals from borrowed JSON
288/// tokens, so deserializing one directly requires borrowed input such as `serde_json::from_str`
289/// or `serde_json::from_slice`.
290#[allow(missing_docs)]
291#[derive(Debug, Clone, Serialize)]
292#[serde(rename_all = "snake_case", tag = "type")]
293pub enum WsMessage {
294    BookChange(BookChangeMsg),
295    BookSnapshot(BookSnapshotMsg),
296    Trade(TradeMsg),
297    TradeBar(BarMsg),
298    DerivativeTicker(DerivativeTickerMsg),
299    OptionSummary(OptionSummaryMsg),
300    Disconnect(DisconnectMsg),
301}
302
303impl<'de> Deserialize<'de> for WsMessage {
304    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
305        let raw = Box::<RawValue>::deserialize(deserializer)?;
306        decode_ws_message(raw.get()).map_err(D::Error::custom)
307    }
308}
309
310// Dispatches on `type` before parsing the message directly: serde's internally tagged enum
311// buffering cannot carry the raw numeric tokens that exact decimal fields read.
312pub(crate) fn decode_ws_message(text: &str) -> serde_json::Result<WsMessage> {
313    let header: WsMessageHeader = serde_json::from_str(text)?;
314
315    match header.kind {
316        WsMessageKind::BookChange => serde_json::from_str(text).map(WsMessage::BookChange),
317        WsMessageKind::BookSnapshot => serde_json::from_str(text).map(WsMessage::BookSnapshot),
318        WsMessageKind::Trade => serde_json::from_str(text).map(WsMessage::Trade),
319        WsMessageKind::TradeBar => serde_json::from_str(text).map(WsMessage::TradeBar),
320        WsMessageKind::DerivativeTicker => {
321            serde_json::from_str(text).map(WsMessage::DerivativeTicker)
322        }
323        WsMessageKind::OptionSummary => serde_json::from_str(text).map(WsMessage::OptionSummary),
324        WsMessageKind::Disconnect => serde_json::from_str(text).map(WsMessage::Disconnect),
325    }
326}
327
328#[derive(Deserialize)]
329struct WsMessageHeader {
330    #[serde(rename = "type")]
331    kind: WsMessageKind,
332}
333
334#[derive(Deserialize)]
335#[serde(rename_all = "snake_case")]
336enum WsMessageKind {
337    BookChange,
338    BookSnapshot,
339    Trade,
340    TradeBar,
341    DerivativeTicker,
342    OptionSummary,
343    Disconnect,
344}
345
346#[derive(Debug, Deserialize)]
347struct RawBookLevel {
348    #[serde(
349        default,
350        deserialize_with = "deserialize_optional_decimal_token_borrowed"
351    )]
352    price: Option<Decimal>,
353    #[serde(
354        default,
355        deserialize_with = "deserialize_optional_decimal_token_borrowed"
356    )]
357    amount: Option<Decimal>,
358}
359
360fn deserialize_book_levels<'de, D>(deserializer: D) -> Result<Vec<BookLevel>, D::Error>
361where
362    D: Deserializer<'de>,
363{
364    Vec::<RawBookLevel>::deserialize(deserializer)?
365        .into_iter()
366        .filter_map(|level| match (level.price, level.amount) {
367            (Some(price), Some(amount)) => Some(Ok(BookLevel { price, amount })),
368            (None, None) => None,
369            (None, Some(_)) => Some(Err(D::Error::custom("book level missing price"))),
370            (Some(_), None) => Some(Err(D::Error::custom("book level missing amount"))),
371        })
372        .collect()
373}
374
375#[cfg(test)]
376mod tests {
377    use rstest::rstest;
378    use rust_decimal_macros::dec;
379
380    use super::*;
381    use crate::common::testing::load_test_json;
382
383    #[rstest]
384    fn test_parse_book_change_message() {
385        let json_data = load_test_json("book_change.json");
386        let message: BookChangeMsg = serde_json::from_str(&json_data).unwrap();
387
388        assert_eq!(message.symbol, "XBTUSD");
389        assert_eq!(message.exchange, TardisExchange::Bitmex);
390        assert!(!message.is_snapshot);
391        assert!(message.bids.is_empty());
392        assert_eq!(message.asks.len(), 1);
393        assert_eq!(message.asks[0].price, dec!(7985.0));
394        assert_eq!(message.asks[0].amount, dec!(283318.0));
395        assert_eq!(
396            message.timestamp,
397            "2019-10-23T11:29:53.469Z".parse::<Timestamp>().unwrap()
398        );
399        assert_eq!(
400            message.local_timestamp,
401            "2019-10-23T11:29:53.469Z".parse::<Timestamp>().unwrap()
402        );
403    }
404
405    #[rstest]
406    fn test_parse_book_change_message_skips_empty_book_levels() {
407        let json_data = r#"{
408            "type": "book_change",
409            "symbol": "XBTUSD",
410            "exchange": "bitmex",
411            "isSnapshot": false,
412            "bids": [{"price": 7984, "amount": 100}, {}],
413            "asks": [{}],
414            "timestamp": "2019-10-23T11:29:53.469Z",
415            "localTimestamp": "2019-10-23T11:29:53.469Z"
416        }"#;
417
418        let message: BookChangeMsg = serde_json::from_str(json_data).unwrap();
419
420        assert_eq!(message.bids.len(), 1);
421        assert!(message.asks.is_empty());
422        assert_eq!(message.bids[0].price, dec!(7984.0));
423        assert_eq!(message.bids[0].amount, dec!(100.0));
424    }
425
426    #[rstest]
427    #[case(r#"[{"price": 7984}]"#, "book level missing amount")]
428    #[case(r#"[{"amount": 100}]"#, "book level missing price")]
429    fn test_parse_book_change_message_rejects_partial_book_level(
430        #[case] bids: &str,
431        #[case] error_message: &str,
432    ) {
433        let json_data = format!(
434            r#"{{
435                "type": "book_change",
436                "symbol": "XBTUSD",
437                "exchange": "bitmex",
438                "isSnapshot": false,
439                "bids": {bids},
440                "asks": [],
441                "timestamp": "2019-10-23T11:29:53.469Z",
442                "localTimestamp": "2019-10-23T11:29:53.469Z"
443            }}"#
444        );
445
446        let error = serde_json::from_str::<BookChangeMsg>(&json_data).unwrap_err();
447
448        assert!(error.to_string().contains(error_message));
449    }
450
451    #[rstest]
452    fn test_parse_book_snapshot_message() {
453        let json_data = load_test_json("book_snapshot.json");
454        let message: BookSnapshotMsg = serde_json::from_str(&json_data).unwrap();
455
456        assert_eq!(message.symbol, "XBTUSD");
457        assert_eq!(message.exchange, TardisExchange::Bitmex);
458        assert_eq!(message.name, "book_snapshot_2_50ms");
459        assert_eq!(message.depth, 2);
460        assert_eq!(message.interval, 50);
461        assert_eq!(message.bids.len(), 2);
462        assert_eq!(message.asks.len(), 2);
463        assert_eq!(message.bids[0].price, dec!(7633.5));
464        assert_eq!(message.bids[0].amount, dec!(1906067.0));
465        assert_eq!(message.asks[0].price, dec!(7634.0));
466        assert_eq!(message.asks[0].amount, dec!(1467849.0));
467        assert_eq!(
468            message.timestamp,
469            "2019-10-25T13:39:46.950Z".parse::<Timestamp>().unwrap(),
470        );
471        assert_eq!(
472            message.local_timestamp,
473            "2019-10-25T13:39:46.961Z".parse::<Timestamp>().unwrap()
474        );
475    }
476
477    #[rstest]
478    fn test_parse_book_snapshot_message_skips_empty_book_levels() {
479        let json_data = r#"{
480            "type": "book_snapshot",
481            "symbol": "ETC",
482            "exchange": "hyperliquid",
483            "name": "book_snapshot_20_10s",
484            "depth": 20,
485            "interval": 10000,
486            "bids": [{"price": 20.002, "amount": 5.81}],
487            "asks": [{"price": 20.003, "amount": 162.45}, {}],
488            "timestamp": "2025-03-03T10:48:10.000Z",
489            "localTimestamp": "2025-03-03T10:48:10.596818Z"
490        }"#;
491
492        let message: BookSnapshotMsg = serde_json::from_str(json_data).unwrap();
493
494        assert_eq!(message.symbol, "ETC");
495        assert_eq!(message.exchange, TardisExchange::Hyperliquid);
496        assert_eq!(message.bids.len(), 1);
497        assert_eq!(message.asks.len(), 1);
498        assert_eq!(message.asks[0].price, dec!(20.003));
499        assert_eq!(message.asks[0].amount, dec!(162.45));
500    }
501
502    #[rstest]
503    fn test_parse_book_snapshot_message_rejects_partial_book_level() {
504        let json_data = r#"{
505            "type": "book_snapshot",
506            "symbol": "ETC",
507            "exchange": "hyperliquid",
508            "name": "book_snapshot_20_10s",
509            "depth": 20,
510            "interval": 10000,
511            "bids": [{"price": 20.002}],
512            "asks": [],
513            "timestamp": "2025-03-03T10:48:10.000Z",
514            "localTimestamp": "2025-03-03T10:48:10.596818Z"
515        }"#;
516
517        let error = serde_json::from_str::<BookSnapshotMsg>(json_data).unwrap_err();
518
519        assert!(error.to_string().contains("book level missing amount"));
520    }
521
522    #[rstest]
523    fn test_parse_trade_message() {
524        let json_data = load_test_json("trade.json");
525        let message: TradeMsg = serde_json::from_str(&json_data).unwrap();
526
527        assert_eq!(message.symbol, "XBTUSD");
528        assert_eq!(message.exchange, TardisExchange::Bitmex);
529        assert_eq!(
530            message.id,
531            Some("282a0445-0e3a-abeb-f403-11003204ea1b".to_string())
532        );
533        assert_eq!(message.price, dec!(7996.0));
534        assert_eq!(message.amount, dec!(50.0));
535        assert_eq!(message.side, "sell");
536        assert_eq!(
537            message.timestamp,
538            "2019-10-23T10:32:49.669Z".parse::<Timestamp>().unwrap()
539        );
540        assert_eq!(
541            message.local_timestamp,
542            "2019-10-23T10:32:49.740Z".parse::<Timestamp>().unwrap()
543        );
544    }
545
546    #[rstest]
547    fn test_parse_derivative_ticker_message() {
548        let json_data = load_test_json("derivative_ticker.json");
549        let message: DerivativeTickerMsg = serde_json::from_str(&json_data).unwrap();
550
551        assert_eq!(message.symbol, "BTC-PERPETUAL");
552        assert_eq!(message.exchange, TardisExchange::Deribit);
553        assert_eq!(message.last_price, Some(7_987.5));
554        assert_eq!(message.open_interest, Some(84_129_491.0));
555        assert_eq!(message.funding_rate, Some(dec!(-0.00001568)));
556        assert_eq!(message.index_price, Some(dec!(7989.28)));
557        assert_eq!(message.mark_price, Some(dec!(7987.56)));
558        assert_eq!(
559            message.timestamp,
560            "2019-10-23T11:34:29.302Z".parse::<Timestamp>().unwrap()
561        );
562        assert_eq!(
563            message.local_timestamp,
564            "2019-10-23T11:34:29.416Z".parse::<Timestamp>().unwrap()
565        );
566    }
567
568    #[rstest]
569    fn test_parse_option_summary_message() {
570        let json_data = load_test_json("option_summary.json");
571        let message: OptionSummaryMsg = serde_json::from_str(&json_data).unwrap();
572
573        assert_eq!(message.symbol, "BTC-28JUN24-70000-C");
574        assert_eq!(message.exchange, TardisExchange::Deribit);
575        assert_eq!(message.option_type, "call");
576        assert_eq!(message.strike_price, 70_000.0);
577        assert_eq!(message.best_bid_iv, Some(0.55));
578        assert_eq!(message.best_ask_iv, Some(0.58));
579        assert_eq!(message.mark_iv, Some(0.565));
580        assert_eq!(message.delta, Some(0.25));
581        assert_eq!(message.gamma, Some(0.000_02));
582        assert_eq!(message.vega, Some(45.5));
583        assert_eq!(message.theta, Some(-15.2));
584        assert_eq!(message.rho, Some(0.05));
585        assert_eq!(message.underlying_price, Some(63_500.0));
586        assert_eq!(message.underlying_index, "BTC-USD");
587        assert_eq!(message.open_interest, Some(150.0));
588        assert_eq!(
589            message.timestamp,
590            "2024-01-15T10:30:00.123Z".parse::<Timestamp>().unwrap()
591        );
592        assert_eq!(
593            message.local_timestamp,
594            "2024-01-15T10:30:00.234Z".parse::<Timestamp>().unwrap()
595        );
596    }
597
598    #[rstest]
599    fn test_parse_bar_message() {
600        let json_data = load_test_json("bar.json");
601        let message: BarMsg = serde_json::from_str(&json_data).unwrap();
602
603        assert_eq!(message.symbol, "XBTUSD");
604        assert_eq!(message.exchange, TardisExchange::Bitmex);
605        assert_eq!(message.name, "trade_bar_10000ms");
606        assert_eq!(message.interval, 10_000);
607        assert_eq!(message.open, dec!(7623.5));
608        assert_eq!(message.high, dec!(7623.5));
609        assert_eq!(message.low, dec!(7623.0));
610        assert_eq!(message.close, dec!(7623.5));
611        assert_eq!(message.volume, dec!(37034.0));
612        assert_eq!(message.buy_volume, 24_244.0);
613        assert_eq!(message.sell_volume, 12_790.0);
614        assert_eq!(message.trades, 9);
615        assert_eq!(message.vwap, 7_623.327320840309);
616        assert_eq!(
617            message.open_timestamp,
618            "2019-10-25T13:11:31.574Z".parse::<Timestamp>().unwrap()
619        );
620        assert_eq!(
621            message.close_timestamp,
622            "2019-10-25T13:11:39.212Z".parse::<Timestamp>().unwrap()
623        );
624        assert_eq!(
625            message.local_timestamp,
626            "2019-10-25T13:11:40.369Z".parse::<Timestamp>().unwrap()
627        );
628        assert_eq!(
629            message.timestamp,
630            "2019-10-25T13:11:40.000Z".parse::<Timestamp>().unwrap()
631        );
632    }
633
634    #[rstest]
635    fn test_parse_disconnect_message() {
636        let json_data = load_test_json("disconnect.json");
637        let message: DisconnectMsg = serde_json::from_str(&json_data).unwrap();
638
639        assert_eq!(message.exchange, TardisExchange::Deribit);
640        assert_eq!(
641            message.local_timestamp,
642            "2019-10-23T11:34:29.416Z".parse::<Timestamp>().unwrap()
643        );
644    }
645
646    #[rstest]
647    #[case("book_change.json", "book_change")]
648    #[case("book_snapshot.json", "book_snapshot")]
649    #[case("trade.json", "trade")]
650    #[case("derivative_ticker.json", "derivative_ticker")]
651    #[case("option_summary.json", "option_summary")]
652    #[case("bar.json", "trade_bar")]
653    #[case("disconnect.json", "disconnect")]
654    fn test_ws_message_deserialize_matches_decode(#[case] fixture: &str, #[case] kind: &str) {
655        let json_data = load_test_json(fixture);
656        let message = decode_ws_message(&json_data).unwrap();
657
658        let decoded_kind = match &message {
659            WsMessage::BookChange(_) => "book_change",
660            WsMessage::BookSnapshot(_) => "book_snapshot",
661            WsMessage::Trade(_) => "trade",
662            WsMessage::TradeBar(_) => "trade_bar",
663            WsMessage::DerivativeTicker(_) => "derivative_ticker",
664            WsMessage::OptionSummary(_) => "option_summary",
665            WsMessage::Disconnect(_) => "disconnect",
666        };
667
668        let decoded = serde_json::to_value(message).unwrap();
669        let direct: WsMessage = serde_json::from_str(&json_data).unwrap();
670        let value: serde_json::Value = serde_json::from_str(&json_data).unwrap();
671        let buffered: WsMessage = serde_json::from_value(value).unwrap();
672
673        assert_eq!(decoded_kind, kind);
674        assert_eq!(serde_json::to_value(direct).unwrap(), decoded);
675        assert_eq!(serde_json::to_value(buffered).unwrap(), decoded);
676    }
677
678    #[rstest]
679    #[case(r#"{"type":"quote","symbol":"XBTUSD"}"#, "unknown variant `quote`")]
680    #[case(r#"{"symbol":"XBTUSD"}"#, "missing field `type`")]
681    fn test_decode_ws_message_rejects_invalid_type(#[case] text: &str, #[case] expected: &str) {
682        let error = decode_ws_message(text).unwrap_err();
683
684        assert!(error.to_string().contains(expected));
685    }
686}
nautilus_tardis/machine/message.rs

nautilus_tardis/machine/
message.rs
