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

nautilus_tardis/machine/
message.rs
