nautilus_tardis/
data.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Tardis data client for streaming replay or live data into the engine.
17
18use std::{
19    sync::{
20        Arc,
21        atomic::{AtomicBool, Ordering},
22    },
23    time::Duration,
24};
25
26use ahash::{AHashMap, AHashSet};
27use futures_util::{SinkExt, StreamExt};
28use nautilus_common::{
29    clients::DataClient,
30    live::{runner::get_data_event_sender, runtime::get_runtime},
31    messages::{
32        DataEvent,
33        data::{
34            subscribe::{SubscribeFundingRates, SubscribeIndexPrices, SubscribeMarkPrices},
35            unsubscribe::{UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeMarkPrices},
36        },
37    },
38};
39use nautilus_core::string::urlencoding;
40use nautilus_model::{
41    data::Data,
42    identifiers::{ClientId, Venue},
43};
44use tokio::{sync::mpsc::UnboundedSender, task::JoinHandle};
45use tokio_tungstenite::{connect_async, tungstenite};
46use tokio_util::sync::CancellationToken;
47
48use crate::{
49    common::{
50        consts::{
51            WS_HEARTBEAT_INTERVAL_SECS, WS_INITIAL_RECONNECT_DELAY_SECS,
52            WS_MAX_RECONNECT_DELAY_SECS,
53        },
54        enums::TardisDataType,
55        urls::resolve_ws_base_url,
56    },
57    config::{BookSnapshotOutput, TardisDataClientConfig},
58    http::TardisHttpClient,
59    machine::{
60        cache::DerivativeTickerCache,
61        client::determine_instrument_info,
62        message::WsMessage,
63        parse::{
64            parse_derivative_ticker_index_price, parse_derivative_ticker_mark_price,
65            parse_tardis_ws_message_data, parse_tardis_ws_message_funding_rate,
66        },
67        types::{TardisInstrumentKey, TardisInstrumentMiniInfo},
68    },
69};
70
71/// Tardis data client for streaming replay or live data into the platform.
72#[derive(Debug)]
73pub struct TardisDataClient {
74    client_id: ClientId,
75    config: TardisDataClientConfig,
76    is_connected: Arc<AtomicBool>,
77    cancellation_token: CancellationToken,
78    tasks: Vec<JoinHandle<()>>,
79    data_sender: UnboundedSender<DataEvent>,
80}
81
82impl TardisDataClient {
83    /// Creates a new [`TardisDataClient`] instance.
84    ///
85    /// # Errors
86    ///
87    /// Returns an error if the data event sender is not initialized.
88    pub fn new(client_id: ClientId, config: TardisDataClientConfig) -> anyhow::Result<Self> {
89        let data_sender = get_data_event_sender();
90
91        Ok(Self {
92            client_id,
93            config,
94            is_connected: Arc::new(AtomicBool::new(false)),
95            cancellation_token: CancellationToken::new(),
96            tasks: Vec::new(),
97            data_sender,
98        })
99    }
100
101    /// Returns `true` if the client is configured for live streaming mode.
102    fn is_stream_mode(&self) -> bool {
103        self.config.options.is_empty() && !self.config.stream_options.is_empty()
104    }
105
106    /// Builds the WebSocket URL for connecting to the Tardis Machine Server.
107    ///
108    /// Ensures `derivative_ticker` is included in the data types for each
109    /// option set so that mark price, index price, and funding rate events
110    /// are available without requiring manual configuration.
111    fn build_ws_url(&self, base_url: &str) -> anyhow::Result<String> {
112        let deriv = TardisDataType::DerivativeTicker.as_tardis_str();
113
114        if self.is_stream_mode() {
115            let mut options = self.config.stream_options.clone();
116            for opt in &mut options {
117                if !opt.data_types.iter().any(|dt| dt == deriv) {
118                    opt.data_types.push(deriv.to_string());
119                }
120            }
121            let options_json = serde_json::to_string(&options)?;
122            Ok(format!(
123                "{base_url}/ws-stream-normalized?options={}",
124                urlencoding::encode(&options_json)
125            ))
126        } else {
127            let mut options = self.config.options.clone();
128            for opt in &mut options {
129                if !opt.data_types.iter().any(|dt| dt == deriv) {
130                    opt.data_types.push(deriv.to_string());
131                }
132            }
133            let options_json = serde_json::to_string(&options)?;
134            Ok(format!(
135                "{base_url}/ws-replay-normalized?options={}",
136                urlencoding::encode(&options_json)
137            ))
138        }
139    }
140
141    /// Spawns the WebSocket message processing loop using an already-connected
142    /// stream. The initial handshake happens in `connect()` so callers get an
143    /// error if the first connection fails. In stream mode the spawned task
144    /// handles subsequent reconnections automatically.
145    fn spawn_ws_task(
146        &mut self,
147        ws_stream: tokio_tungstenite::WebSocketStream<
148            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
149        >,
150        url: String,
151        instrument_map: AHashMap<TardisInstrumentKey, Arc<TardisInstrumentMiniInfo>>,
152        book_snapshot_output: BookSnapshotOutput,
153        extract_bbo_as_quotes: bool,
154        is_stream_mode: bool,
155    ) {
156        let sender = self.data_sender.clone();
157        let cancel = self.cancellation_token.clone();
158        let connected = self.is_connected.clone();
159
160        let handle = get_runtime().spawn(async move {
161            let mut reconnect_delay = Duration::from_secs(WS_INITIAL_RECONNECT_DELAY_SECS);
162            let instrument_map = instrument_map;
163
164            // Process the initial (already-connected) stream
165            let should_reconnect = Self::run_ws_session(
166                ws_stream,
167                &cancel,
168                &sender,
169                &instrument_map,
170                &book_snapshot_output,
171                extract_bbo_as_quotes,
172            )
173            .await;
174
175            if !should_reconnect || !is_stream_mode || cancel.is_cancelled() {
176                connected.store(false, Ordering::Release);
177                return;
178            }
179
180            // Mark disconnected while reconnecting so health checks see the outage
181            connected.store(false, Ordering::Release);
182
183            // Reconnection loop (stream mode only)
184            loop {
185                log::warn!(
186                    "Stream disconnected, reconnecting in {}s",
187                    reconnect_delay.as_secs()
188                );
189
190                tokio::select! {
191                    () = tokio::time::sleep(reconnect_delay) => {}
192                    () = cancel.cancelled() => break,
193                }
194
195                reconnect_delay = std::cmp::min(
196                    reconnect_delay * 2,
197                    Duration::from_secs(WS_MAX_RECONNECT_DELAY_SECS),
198                );
199
200                // Reconnect WS first (critical path), then refresh instruments
201                let ws_result = tokio::select! {
202                    result = connect_async(&url) => Some(result),
203                    () = cancel.cancelled() => None,
204                };
205
206                let Some(ws_result) = ws_result else {
207                    break;
208                };
209
210                match ws_result {
211                    Ok((ws_stream, _)) => {
212                        log::info!("Reconnected to Tardis Machine");
213                        connected.store(true, Ordering::Release);
214                        reconnect_delay = Duration::from_secs(WS_INITIAL_RECONNECT_DELAY_SECS);
215
216                        let should_reconnect = Self::run_ws_session(
217                            ws_stream,
218                            &cancel,
219                            &sender,
220                            &instrument_map,
221                            &book_snapshot_output,
222                            extract_bbo_as_quotes,
223                        )
224                        .await;
225
226                        if !should_reconnect || cancel.is_cancelled() {
227                            break;
228                        }
229
230                        connected.store(false, Ordering::Release);
231                    }
232                    Err(e) => {
233                        if cancel.is_cancelled() {
234                            break;
235                        }
236
237                        log::warn!(
238                            "Failed to reconnect to Tardis Machine: {e}, retrying in {}s",
239                            reconnect_delay.as_secs()
240                        );
241                    }
242                }
243            }
244
245            connected.store(false, Ordering::Release);
246        });
247
248        self.tasks.push(handle);
249    }
250
251    /// Runs a single WebSocket session: starts heartbeat, processes messages,
252    /// and returns whether the caller should attempt reconnection.
253    async fn run_ws_session(
254        ws_stream: tokio_tungstenite::WebSocketStream<
255            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
256        >,
257        cancel: &CancellationToken,
258        sender: &UnboundedSender<DataEvent>,
259        instrument_map: &AHashMap<TardisInstrumentKey, Arc<TardisInstrumentMiniInfo>>,
260        book_snapshot_output: &BookSnapshotOutput,
261        extract_bbo_as_quotes: bool,
262    ) -> bool {
263        let (mut writer, mut reader) = ws_stream.split();
264
265        let heartbeat_token = cancel.child_token();
266        let heartbeat_signal = heartbeat_token.clone();
267
268        get_runtime().spawn(async move {
269            let mut interval =
270                tokio::time::interval(Duration::from_secs(WS_HEARTBEAT_INTERVAL_SECS));
271            loop {
272                tokio::select! {
273                    _ = interval.tick() => {
274                        log::trace!("Sending PING");
275
276                        if let Err(e) = writer.send(tungstenite::Message::Ping(vec![].into())).await {
277                            log::debug!("Heartbeat send failed: {e}");
278                            break;
279                        }
280                    }
281                    () = heartbeat_signal.cancelled() => break,
282                }
283            }
284        });
285
286        let should_reconnect = Self::run_ws_loop(
287            &mut reader,
288            cancel,
289            sender,
290            instrument_map,
291            book_snapshot_output,
292            extract_bbo_as_quotes,
293        )
294        .await;
295
296        heartbeat_token.cancel();
297        should_reconnect
298    }
299
300    /// Extracts and sends all data events from a `DerivativeTicker` message:
301    /// funding rate, mark price, and index price. Only emits events when values
302    /// change from the previous update. Returns `false` if the channel is broken
303    /// and the caller should exit the loop.
304    fn send_derivative_ticker_events(
305        ws_msg: &WsMessage,
306        info: &Arc<TardisInstrumentMiniInfo>,
307        sender: &UnboundedSender<DataEvent>,
308        cache: &mut DerivativeTickerCache,
309    ) -> bool {
310        if let Some(funding) = parse_tardis_ws_message_funding_rate(ws_msg.clone(), info)
311            && cache.should_emit_funding_rate(&funding)
312            && sender.send(DataEvent::FundingRate(funding)).is_err()
313        {
314            return false;
315        }
316
317        if let WsMessage::DerivativeTicker(msg) = ws_msg {
318            if let Ok(Some(mark_price)) =
319                parse_derivative_ticker_mark_price(msg, info.instrument_id, info.price_precision)
320                && cache.should_emit_mark_price(&mark_price)
321                && sender
322                    .send(DataEvent::Data(Data::MarkPriceUpdate(mark_price)))
323                    .is_err()
324            {
325                return false;
326            }
327
328            if let Ok(Some(index_price)) =
329                parse_derivative_ticker_index_price(msg, info.instrument_id, info.price_precision)
330                && cache.should_emit_index_price(&index_price)
331                && sender
332                    .send(DataEvent::Data(Data::IndexPriceUpdate(index_price)))
333                    .is_err()
334            {
335                return false;
336            }
337        }
338
339        true
340    }
341
342    /// Processes WebSocket messages until the stream ends, an error occurs, or
343    /// the cancellation token fires. Returns `true` if the caller should attempt
344    /// reconnection (stream mode only).
345    async fn run_ws_loop(
346        reader: &mut futures_util::stream::SplitStream<
347            tokio_tungstenite::WebSocketStream<
348                tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
349            >,
350        >,
351        cancel: &CancellationToken,
352        sender: &UnboundedSender<DataEvent>,
353        instrument_map: &AHashMap<TardisInstrumentKey, Arc<TardisInstrumentMiniInfo>>,
354        book_snapshot_output: &BookSnapshotOutput,
355        extract_bbo_as_quotes: bool,
356    ) -> bool {
357        let mut ticker_cache = DerivativeTickerCache::default();
358
359        loop {
360            let msg = tokio::select! {
361                msg = reader.next() => msg,
362                () = cancel.cancelled() => {
363                    log::debug!("Stream task cancelled");
364                    return false;
365                }
366            };
367
368            match msg {
369                Some(Ok(tungstenite::Message::Text(text))) => {
370                    match serde_json::from_str::<WsMessage>(&text) {
371                        Ok(ws_msg) => {
372                            if matches!(ws_msg, WsMessage::Disconnect(_)) {
373                                log::debug!("Received disconnect message");
374                                continue;
375                            }
376
377                            let info = determine_instrument_info(&ws_msg, instrument_map);
378
379                            if let Some(info) = info {
380                                if matches!(ws_msg, WsMessage::DerivativeTicker(_)) {
381                                    if !Self::send_derivative_ticker_events(
382                                        &ws_msg,
383                                        &info,
384                                        sender,
385                                        &mut ticker_cache,
386                                    ) {
387                                        return false;
388                                    }
389                                } else {
390                                    let data = parse_tardis_ws_message_data(
391                                        ws_msg,
392                                        &info,
393                                        book_snapshot_output,
394                                        extract_bbo_as_quotes,
395                                    );
396
397                                    for data in data {
398                                        if let Err(e) = sender.send(DataEvent::Data(data)) {
399                                            log::error!("Failed to send data event: {e}");
400                                            return false;
401                                        }
402                                    }
403                                }
404                            }
405                        }
406                        Err(e) => {
407                            log::error!("Failed to deserialize message: {e}");
408                        }
409                    }
410                }
411                Some(Ok(tungstenite::Message::Close(frame))) => {
412                    if let Some(frame) = frame {
413                        log::debug!("WebSocket closed: {} {}", frame.code, frame.reason);
414                    } else {
415                        log::debug!("WebSocket closed");
416                    }
417                    return true;
418                }
419                Some(Ok(_)) => {}
420                Some(Err(e)) => {
421                    log::warn!("WebSocket error: {e}");
422                    return true;
423                }
424                None => {
425                    log::debug!("Stream ended");
426                    return true;
427                }
428            }
429        }
430    }
431}
432
433#[async_trait::async_trait(?Send)]
434impl DataClient for TardisDataClient {
435    fn client_id(&self) -> ClientId {
436        self.client_id
437    }
438
439    fn venue(&self) -> Option<Venue> {
440        None // Tardis is multi-venue
441    }
442
443    fn start(&mut self) -> anyhow::Result<()> {
444        log::info!("Starting {}", self.client_id);
445        Ok(())
446    }
447
448    fn stop(&mut self) -> anyhow::Result<()> {
449        log::info!("Stopping {}", self.client_id);
450        self.cancellation_token.cancel();
451
452        for handle in self.tasks.drain(..) {
453            handle.abort();
454        }
455        self.is_connected.store(false, Ordering::Release);
456        Ok(())
457    }
458
459    fn reset(&mut self) -> anyhow::Result<()> {
460        self.cancellation_token.cancel();
461
462        for handle in self.tasks.drain(..) {
463            handle.abort();
464        }
465        self.cancellation_token = CancellationToken::new();
466        self.is_connected.store(false, Ordering::Release);
467        Ok(())
468    }
469
470    fn dispose(&mut self) -> anyhow::Result<()> {
471        self.stop()
472    }
473
474    fn is_connected(&self) -> bool {
475        self.is_connected.load(Ordering::Acquire)
476    }
477
478    fn is_disconnected(&self) -> bool {
479        !self.is_connected()
480    }
481
482    fn subscribe_mark_prices(&mut self, _cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
483        Ok(())
484    }
485
486    fn subscribe_index_prices(&mut self, _cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
487        Ok(())
488    }
489
490    fn subscribe_funding_rates(&mut self, _cmd: SubscribeFundingRates) -> anyhow::Result<()> {
491        Ok(())
492    }
493
494    fn unsubscribe_mark_prices(&mut self, _cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
495        Ok(())
496    }
497
498    fn unsubscribe_index_prices(&mut self, _cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
499        Ok(())
500    }
501
502    fn unsubscribe_funding_rates(&mut self, _cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
503        Ok(())
504    }
505
506    async fn connect(&mut self) -> anyhow::Result<()> {
507        if self.is_connected() {
508            return Ok(());
509        }
510
511        if self.config.options.is_empty() && self.config.stream_options.is_empty() {
512            anyhow::bail!("Either replay `options` or `stream_options` must be provided");
513        }
514
515        let is_stream_mode = self.is_stream_mode();
516        let book_snapshot_output = self.config.book_snapshot_output.clone();
517        let extract_bbo_as_quotes = self.config.extract_bbo_as_quotes;
518
519        let http_client = TardisHttpClient::new(
520            self.config.api_key.as_deref(),
521            None,
522            None,
523            self.config.normalize_symbols,
524            self.config.proxy_url.clone(),
525        )?;
526
527        let exchanges: AHashSet<_> = if is_stream_mode {
528            self.config
529                .stream_options
530                .iter()
531                .map(|opt| opt.exchange)
532                .collect()
533        } else {
534            self.config.options.iter().map(|opt| opt.exchange).collect()
535        };
536
537        let base_url = resolve_ws_base_url(self.config.tardis_ws_url.as_deref())?;
538        let (instrument_map, instruments) = http_client
539            .bootstrap_instruments(&exchanges)
540            .await
541            .map_err(|e| anyhow::anyhow!("Failed to bootstrap instruments: {e}"))?;
542
543        for instrument in instruments {
544            if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
545                log::error!("Failed to send instrument event: {e}");
546            }
547        }
548
549        let url = self.build_ws_url(&base_url)?;
550
551        let mode_label = if is_stream_mode { "stream" } else { "replay" };
552        log::info!("Connecting to Tardis Machine {mode_label}");
553        log::debug!("URL: {url}");
554
555        self.cancellation_token = CancellationToken::new();
556
557        let (ws_stream, _) = connect_async(&url)
558            .await
559            .map_err(|e| anyhow::anyhow!("Failed to connect to Tardis Machine: {e}"))?;
560
561        log::info!("Connected to Tardis Machine");
562
563        self.spawn_ws_task(
564            ws_stream,
565            url,
566            instrument_map,
567            book_snapshot_output,
568            extract_bbo_as_quotes,
569            is_stream_mode,
570        );
571        self.is_connected.store(true, Ordering::Release);
572
573        log::info!("Connected: {}", self.client_id);
574        Ok(())
575    }
576
577    async fn disconnect(&mut self) -> anyhow::Result<()> {
578        self.cancellation_token.cancel();
579        self.cancellation_token = CancellationToken::new();
580
581        let handles: Vec<_> = self.tasks.drain(..).collect();
582        if !handles.is_empty() {
583            for handle in handles {
584                if let Err(e) = handle.await {
585                    log::error!("Error joining task: {e}");
586                }
587            }
588            log::info!("Disconnected: {}", self.client_id);
589        }
590
591        self.is_connected.store(false, Ordering::Release);
592
593        Ok(())
594    }
595}
596
597#[cfg(test)]
598mod tests {
599    use chrono::NaiveDate;
600    use nautilus_common::live::runner::set_data_event_sender;
601    use rstest::rstest;
602
603    use super::*;
604    use crate::{
605        common::{consts::TARDIS_CLIENT_ID, enums::TardisExchange},
606        config::TardisDataClientConfig,
607        machine::types::ReplayNormalizedRequestOptions,
608    };
609
610    fn setup_test_env() {
611        use std::cell::OnceCell;
612
613        thread_local! {
614            static INIT: OnceCell<()> = const { OnceCell::new() };
615        }
616
617        INIT.with(|cell| {
618            cell.get_or_init(|| {
619                let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
620                set_data_event_sender(sender);
621            });
622        });
623    }
624
625    #[rstest]
626    fn test_build_ws_url_injects_derivative_ticker() {
627        setup_test_env();
628
629        let config = TardisDataClientConfig {
630            options: vec![ReplayNormalizedRequestOptions {
631                exchange: TardisExchange::BinanceFutures,
632                symbols: Some(vec!["BTCUSDT".to_string()]),
633                from: NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(),
634                to: NaiveDate::from_ymd_opt(2024, 1, 2).unwrap(),
635                data_types: vec!["trade".to_string()],
636                with_disconnect_messages: Some(false),
637            }],
638            ..Default::default()
639        };
640
641        let client = TardisDataClient::new(*TARDIS_CLIENT_ID, config).unwrap();
642        let url = client.build_ws_url("ws://localhost:8001").unwrap();
643
644        assert!(
645            url.contains("derivative_ticker"),
646            "URL should contain derivative_ticker but was: {url}"
647        );
648        assert!(url.contains("trade"), "URL should still contain trade");
649    }
650
651    #[rstest]
652    fn test_build_ws_url_does_not_duplicate_derivative_ticker() {
653        setup_test_env();
654
655        let config = TardisDataClientConfig {
656            options: vec![ReplayNormalizedRequestOptions {
657                exchange: TardisExchange::BinanceFutures,
658                symbols: Some(vec!["BTCUSDT".to_string()]),
659                from: NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(),
660                to: NaiveDate::from_ymd_opt(2024, 1, 2).unwrap(),
661                data_types: vec!["trade".to_string(), "derivative_ticker".to_string()],
662                with_disconnect_messages: Some(false),
663            }],
664            ..Default::default()
665        };
666
667        let client = TardisDataClient::new(*TARDIS_CLIENT_ID, config).unwrap();
668        let ws_url = client.build_ws_url("ws://localhost:8001").unwrap();
669
670        let decoded = urlencoding::decode(ws_url.split("options=").nth(1).unwrap()).unwrap();
671        let count = decoded.matches("derivative_ticker").count();
672        assert_eq!(count, 1, "derivative_ticker should appear exactly once");
673    }
674}
nautilus_tardis/data.rs

nautilus_tardis/
data.rs
