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

nautilus_tardis/machine/
message.rs
