nautilus_testkit/testers/data/
actor.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::{num::NonZeroUsize, time::Duration};
17
18use ahash::{AHashMap, AHashSet};
19use jiff::SignedDuration;
20use nautilus_common::{
21    actor::{DataActor, DataActorCore},
22    config::ConfigError,
23    enums::LogColor,
24    log_info, nautilus_actor,
25    timer::TimeEvent,
26};
27use nautilus_model::{
28    data::{
29        Bar, FundingRateUpdate, IndexPriceUpdate, InstrumentClose, InstrumentStatus,
30        MarkPriceUpdate, OrderBookDeltas, OrderBookDepth, QuoteTick, TradeTick,
31        option_chain::OptionGreeks,
32    },
33    identifiers::InstrumentId,
34    instruments::{Instrument, InstrumentAny},
35    orderbook::OrderBook,
36};
37
38use super::config::DataTesterConfig;
39use crate::testers::timestamps::{warn_if_implausible_optional, warn_if_implausible_unix_nanos};
40
41/// A data tester actor for live testing market data subscriptions.
42///
43/// Subscribes to configured data types for specified instruments and logs
44/// received data to demonstrate the data flow. Useful for testing adapters
45/// and validating data connectivity.
46///
47/// This actor provides equivalent functionality to the Python `DataTester`
48/// in the test kit.
49#[derive(Debug)]
50pub struct DataTester {
51    pub(super) core: DataActorCore,
52    pub(super) config: DataTesterConfig,
53    pub(super) books: AHashMap<InstrumentId, OrderBook>,
54}
55
56nautilus_actor!(DataTester);
57
58impl DataActor for DataTester {
59    #[expect(
60        clippy::too_many_lines,
61        reason = "startup subscribes to each configured data scenario explicitly"
62    )]
63    fn on_start(&mut self) -> anyhow::Result<()> {
64        let instrument_ids = self.config.instrument_ids.clone();
65        let client_id = self.config.client_id;
66        let subscribe_params = self.config.subscribe_params.clone();
67        let request_params = self.config.request_params.clone();
68        let stats_interval_secs = self.config.stats_interval_secs;
69
70        // Request instruments if configured
71        if self.config.request_instruments {
72            let mut venues = AHashSet::new();
73            for instrument_id in &instrument_ids {
74                venues.insert(instrument_id.venue);
75            }
76
77            for venue in venues {
78                let _ = self.request_instruments(
79                    Some(venue),
80                    None,
81                    None,
82                    client_id,
83                    request_params.clone(),
84                );
85            }
86        }
87
88        let book_depth = self
89            .config
90            .book_depth
91            .map(|depth| {
92                NonZeroUsize::new(depth)
93                    .ok_or_else(|| ConfigError::range("book_depth", "must be positive, was 0"))
94            })
95            .transpose()?;
96
97        // Subscribe to data for each instrument
98        for instrument_id in instrument_ids {
99            if self.config.subscribe_instrument {
100                self.subscribe_instrument(instrument_id, client_id, subscribe_params.clone());
101            }
102
103            if self.config.subscribe_book_deltas {
104                self.subscribe_book_deltas(
105                    instrument_id,
106                    self.config.book_type,
107                    book_depth,
108                    client_id,
109                    self.config.manage_book,
110                    subscribe_params.clone(),
111                );
112
113                if self.config.manage_book {
114                    let book = OrderBook::new(instrument_id, self.config.book_type);
115                    self.books.insert(instrument_id, book);
116                }
117            }
118
119            if self.config.subscribe_book_at_interval {
120                self.subscribe_book_at_interval(
121                    instrument_id,
122                    self.config.book_type,
123                    book_depth,
124                    NonZeroUsize::new(self.config.book_interval_ms).ok_or_else(|| {
125                        ConfigError::range("book_interval_ms", "must be positive, was 0")
126                    })?,
127                    client_id,
128                    subscribe_params.clone(),
129                );
130            }
131
132            if self.config.subscribe_book_depth {
133                self.subscribe_book_depth(
134                    instrument_id,
135                    self.config.book_type,
136                    book_depth,
137                    client_id,
138                    self.config.manage_book
139                        && !self.config.subscribe_book_deltas
140                        && !self.config.subscribe_book_at_interval,
141                    subscribe_params.clone(),
142                );
143            }
144
145            if self.config.subscribe_quotes {
146                self.subscribe_quotes(instrument_id, client_id, subscribe_params.clone());
147            }
148
149            if self.config.subscribe_trades {
150                self.subscribe_trades(instrument_id, client_id, subscribe_params.clone());
151            }
152
153            if self.config.subscribe_mark_prices {
154                self.subscribe_mark_prices(instrument_id, client_id, subscribe_params.clone());
155            }
156
157            if self.config.subscribe_index_prices {
158                self.subscribe_index_prices(instrument_id, client_id, subscribe_params.clone());
159            }
160
161            if self.config.subscribe_funding_rates {
162                self.subscribe_funding_rates(instrument_id, client_id, subscribe_params.clone());
163            }
164
165            if self.config.subscribe_instrument_status {
166                self.subscribe_instrument_status(
167                    instrument_id,
168                    client_id,
169                    subscribe_params.clone(),
170                );
171            }
172
173            if self.config.subscribe_instrument_close {
174                self.subscribe_instrument_close(instrument_id, client_id, subscribe_params.clone());
175            }
176
177            if self.config.subscribe_option_greeks {
178                self.subscribe_option_greeks(instrument_id, client_id, subscribe_params.clone());
179            }
180
181            // Request historical quotes (default to last 1 hour)
182            if self.config.request_quotes {
183                let start = self.clock().utc_now() - SignedDuration::from_hours(1);
184
185                if let Err(e) = self.request_quotes(
186                    instrument_id,
187                    Some(start),
188                    None,
189                    None,
190                    client_id,
191                    request_params.clone(),
192                ) {
193                    log::error!("Failed to request quotes for {instrument_id}: {e}");
194                }
195            }
196
197            // Request order book snapshot if configured
198            if self.config.request_book_snapshot {
199                let _ = self.request_book_snapshot(
200                    instrument_id,
201                    self.config
202                        .book_depth
203                        .map(|depth| {
204                            NonZeroUsize::new(depth).ok_or_else(|| {
205                                ConfigError::range("book_depth", "must be positive, was 0")
206                            })
207                        })
208                        .transpose()?,
209                    client_id,
210                    request_params.clone(),
211                );
212            }
213
214            // TODO: Request book deltas when Rust data engine has RequestBookDeltas
215
216            // Request historical trades (default to last 1 hour)
217            if self.config.request_trades {
218                let start = self.clock().utc_now() - SignedDuration::from_hours(1);
219
220                if let Err(e) = self.request_trades(
221                    instrument_id,
222                    Some(start),
223                    None,
224                    None,
225                    client_id,
226                    request_params.clone(),
227                ) {
228                    log::error!("Failed to request trades for {instrument_id}: {e}");
229                }
230            }
231
232            // Request historical funding rates (default to last 7 days)
233            if self.config.request_funding_rates {
234                let start = self.clock().utc_now() - SignedDuration::from_hours(7 * 24);
235
236                if let Err(e) = self.request_funding_rates(
237                    instrument_id,
238                    Some(start),
239                    None,
240                    None,
241                    client_id,
242                    request_params.clone(),
243                ) {
244                    log::error!("Failed to request funding rates for {instrument_id}: {e}");
245                }
246            }
247        }
248
249        // Subscribe to bars
250        if let Some(bar_types) = self.config.bar_types.clone() {
251            for bar_type in bar_types {
252                if self.config.subscribe_bars {
253                    self.subscribe_bars(bar_type, client_id, subscribe_params.clone());
254                }
255
256                // Request historical bars (default to last 1 hour)
257                if self.config.request_bars {
258                    let start = self.clock().utc_now() - SignedDuration::from_hours(1);
259
260                    if let Err(e) = self.request_bars(
261                        bar_type,
262                        Some(start),
263                        None,
264                        None,
265                        client_id,
266                        request_params.clone(),
267                    ) {
268                        log::error!("Failed to request bars for {bar_type}: {e}");
269                    }
270                }
271            }
272        }
273
274        // Set up stats timer
275        if stats_interval_secs > 0 {
276            self.clock().set_timer(
277                "STATS-TIMER",
278                Duration::from_secs(stats_interval_secs),
279                None,
280                None,
281                None,
282                Some(true),
283                Some(false),
284            )?;
285        }
286
287        Ok(())
288    }
289
290    fn on_stop(&mut self) -> anyhow::Result<()> {
291        if !self.config.can_unsubscribe {
292            return Ok(());
293        }
294
295        let instrument_ids = self.config.instrument_ids.clone();
296        let client_id = self.config.client_id;
297        let subscribe_params = self.config.subscribe_params.clone();
298
299        for instrument_id in instrument_ids {
300            if self.config.subscribe_instrument {
301                self.unsubscribe_instrument(instrument_id, client_id, subscribe_params.clone());
302            }
303
304            if self.config.subscribe_book_deltas {
305                self.unsubscribe_book_deltas(instrument_id, client_id, subscribe_params.clone());
306            }
307
308            if self.config.subscribe_book_at_interval {
309                self.unsubscribe_book_at_interval(
310                    instrument_id,
311                    NonZeroUsize::new(self.config.book_interval_ms).ok_or_else(|| {
312                        ConfigError::range("book_interval_ms", "must be positive, was 0")
313                    })?,
314                    client_id,
315                    subscribe_params.clone(),
316                );
317            }
318
319            if self.config.subscribe_book_depth {
320                self.unsubscribe_book_depth(instrument_id, client_id, subscribe_params.clone());
321            }
322
323            if self.config.subscribe_quotes {
324                self.unsubscribe_quotes(instrument_id, client_id, subscribe_params.clone());
325            }
326
327            if self.config.subscribe_trades {
328                self.unsubscribe_trades(instrument_id, client_id, subscribe_params.clone());
329            }
330
331            if self.config.subscribe_mark_prices {
332                self.unsubscribe_mark_prices(instrument_id, client_id, subscribe_params.clone());
333            }
334
335            if self.config.subscribe_index_prices {
336                self.unsubscribe_index_prices(instrument_id, client_id, subscribe_params.clone());
337            }
338
339            if self.config.subscribe_funding_rates {
340                self.unsubscribe_funding_rates(instrument_id, client_id, subscribe_params.clone());
341            }
342
343            if self.config.subscribe_instrument_status {
344                self.unsubscribe_instrument_status(
345                    instrument_id,
346                    client_id,
347                    subscribe_params.clone(),
348                );
349            }
350
351            if self.config.subscribe_instrument_close {
352                self.unsubscribe_instrument_close(
353                    instrument_id,
354                    client_id,
355                    subscribe_params.clone(),
356                );
357            }
358
359            if self.config.subscribe_option_greeks {
360                self.unsubscribe_option_greeks(instrument_id, client_id, subscribe_params.clone());
361            }
362        }
363
364        if let Some(bar_types) = self.config.bar_types.clone() {
365            for bar_type in bar_types {
366                if self.config.subscribe_bars {
367                    self.unsubscribe_bars(bar_type, client_id, subscribe_params.clone());
368                }
369            }
370        }
371
372        Ok(())
373    }
374
375    fn on_time_event(&mut self, _event: &TimeEvent) -> anyhow::Result<()> {
376        // Timer events are used by the actor but don't require specific handling
377        Ok(())
378    }
379
380    fn on_instrument(&mut self, instrument: &InstrumentAny) -> anyhow::Result<()> {
381        warn_if_implausible_unix_nanos("instrument", instrument.ts_event(), instrument.ts_init());
382
383        if self.config.log_data {
384            log_info!("{instrument:?}", color = LogColor::Cyan);
385        }
386        Ok(())
387    }
388
389    fn on_book(&mut self, book: &OrderBook) -> anyhow::Result<()> {
390        if self.config.log_data {
391            let levels = self.config.book_levels_to_print;
392            let instrument_id = book.instrument_id;
393            let book_str = book.pprint(levels, None);
394            log_info!("\n{instrument_id}\n{book_str}", color = LogColor::Cyan);
395        }
396
397        Ok(())
398    }
399
400    fn on_book_deltas(&mut self, deltas: &OrderBookDeltas) -> anyhow::Result<()> {
401        warn_if_implausible_unix_nanos("book deltas", deltas.ts_event, deltas.ts_init);
402
403        if self.config.manage_book {
404            if let Some(book) = self.books.get_mut(&deltas.instrument_id) {
405                book.apply_deltas(deltas)?;
406
407                if self.config.log_data {
408                    let levels = self.config.book_levels_to_print;
409                    let instrument_id = deltas.instrument_id;
410                    let book_str = book.pprint(levels, None);
411                    log_info!("\n{instrument_id}\n{book_str}", color = LogColor::Cyan);
412                }
413            }
414        } else if self.config.log_data {
415            log_info!("{deltas:?}", color = LogColor::Cyan);
416        }
417        Ok(())
418    }
419
420    fn on_book_depth(&mut self, depth: &OrderBookDepth) -> anyhow::Result<()> {
421        warn_if_implausible_unix_nanos("book depth", depth.ts_event, depth.ts_init);
422
423        if self.config.log_data {
424            log_info!("{depth:?}", color = LogColor::Cyan);
425        }
426        Ok(())
427    }
428
429    fn on_quote(&mut self, quote: &QuoteTick) -> anyhow::Result<()> {
430        warn_if_implausible_unix_nanos("quote", quote.ts_event, quote.ts_init);
431
432        if self.config.log_data {
433            log_info!("{quote:?}", color = LogColor::Cyan);
434        }
435        Ok(())
436    }
437
438    fn on_trade(&mut self, trade: &TradeTick) -> anyhow::Result<()> {
439        warn_if_implausible_unix_nanos("trade", trade.ts_event, trade.ts_init);
440
441        if self.config.log_data {
442            log_info!("{trade:?}", color = LogColor::Cyan);
443        }
444        Ok(())
445    }
446
447    fn on_bar(&mut self, bar: &Bar) -> anyhow::Result<()> {
448        warn_if_implausible_unix_nanos("bar", bar.ts_event, bar.ts_init);
449
450        if self.config.log_data {
451            log_info!("{bar:?}", color = LogColor::Cyan);
452        }
453        Ok(())
454    }
455
456    fn on_mark_price(&mut self, mark_price: &MarkPriceUpdate) -> anyhow::Result<()> {
457        warn_if_implausible_unix_nanos("mark price", mark_price.ts_event, mark_price.ts_init);
458
459        if self.config.log_data {
460            log_info!("{mark_price:?}", color = LogColor::Cyan);
461        }
462        Ok(())
463    }
464
465    fn on_index_price(&mut self, index_price: &IndexPriceUpdate) -> anyhow::Result<()> {
466        warn_if_implausible_unix_nanos("index price", index_price.ts_event, index_price.ts_init);
467
468        if self.config.log_data {
469            log_info!("{index_price:?}", color = LogColor::Cyan);
470        }
471        Ok(())
472    }
473
474    fn on_funding_rate(&mut self, funding_rate: &FundingRateUpdate) -> anyhow::Result<()> {
475        warn_if_implausible_unix_nanos("funding rate", funding_rate.ts_event, funding_rate.ts_init);
476        warn_if_implausible_optional(
477            "funding rate",
478            "next_funding_ns",
479            funding_rate.next_funding_ns,
480        );
481
482        if self.config.log_data {
483            log_info!("{funding_rate:?}", color = LogColor::Cyan);
484        }
485        Ok(())
486    }
487
488    fn on_instrument_status(&mut self, data: &InstrumentStatus) -> anyhow::Result<()> {
489        warn_if_implausible_unix_nanos("instrument status", data.ts_event, data.ts_init);
490
491        if self.config.log_data {
492            log_info!("{data:?}", color = LogColor::Cyan);
493        }
494        Ok(())
495    }
496
497    fn on_instrument_close(&mut self, update: &InstrumentClose) -> anyhow::Result<()> {
498        warn_if_implausible_unix_nanos("instrument close", update.ts_event, update.ts_init);
499
500        if self.config.log_data {
501            log_info!("{update:?}", color = LogColor::Cyan);
502        }
503        Ok(())
504    }
505
506    fn on_option_greeks(&mut self, greeks: &OptionGreeks) -> anyhow::Result<()> {
507        warn_if_implausible_unix_nanos("option greeks", greeks.ts_event, greeks.ts_init);
508
509        if self.config.log_data {
510            log_info!("{greeks:?}", color = LogColor::Cyan);
511        }
512        Ok(())
513    }
514
515    fn on_historical_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
516        for trade in trades {
517            warn_if_implausible_unix_nanos("historical trade", trade.ts_event, trade.ts_init);
518        }
519
520        if self.config.log_data {
521            log_info!(
522                "Received {} historical trades",
523                trades.len(),
524                color = LogColor::Cyan
525            );
526
527            for trade in trades.iter().take(5) {
528                log_info!("  {trade:?}", color = LogColor::Cyan);
529            }
530
531            if trades.len() > 5 {
532                log_info!(
533                    "  ... and {} more trades",
534                    trades.len() - 5,
535                    color = LogColor::Cyan
536                );
537            }
538        }
539        Ok(())
540    }
541
542    fn on_historical_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
543        for quote in quotes {
544            warn_if_implausible_unix_nanos("historical quote", quote.ts_event, quote.ts_init);
545        }
546
547        if self.config.log_data {
548            log_info!(
549                "Received {} historical quotes",
550                quotes.len(),
551                color = LogColor::Cyan
552            );
553
554            for quote in quotes.iter().take(5) {
555                log_info!("  {quote:?}", color = LogColor::Cyan);
556            }
557
558            if quotes.len() > 5 {
559                log_info!(
560                    "  ... and {} more quotes",
561                    quotes.len() - 5,
562                    color = LogColor::Cyan
563                );
564            }
565        }
566        Ok(())
567    }
568
569    fn on_historical_funding_rates(
570        &mut self,
571        funding_rates: &[FundingRateUpdate],
572    ) -> anyhow::Result<()> {
573        for rate in funding_rates {
574            warn_if_implausible_unix_nanos("historical funding rate", rate.ts_event, rate.ts_init);
575            warn_if_implausible_optional(
576                "historical funding rate",
577                "next_funding_ns",
578                rate.next_funding_ns,
579            );
580        }
581
582        if self.config.log_data {
583            log_info!(
584                "Received {} historical funding rates",
585                funding_rates.len(),
586                color = LogColor::Cyan
587            );
588
589            for rate in funding_rates.iter().take(5) {
590                log_info!("  {rate:?}", color = LogColor::Cyan);
591            }
592
593            if funding_rates.len() > 5 {
594                log_info!(
595                    "  ... and {} more funding rates",
596                    funding_rates.len() - 5,
597                    color = LogColor::Cyan
598                );
599            }
600        }
601        Ok(())
602    }
603
604    fn on_historical_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
605        for bar in bars {
606            warn_if_implausible_unix_nanos("historical bar", bar.ts_event, bar.ts_init);
607        }
608
609        if self.config.log_data {
610            log_info!(
611                "Received {} historical bars",
612                bars.len(),
613                color = LogColor::Cyan
614            );
615
616            for bar in bars.iter().take(5) {
617                log_info!("  {bar:?}", color = LogColor::Cyan);
618            }
619
620            if bars.len() > 5 {
621                log_info!(
622                    "  ... and {} more bars",
623                    bars.len() - 5,
624                    color = LogColor::Cyan
625                );
626            }
627        }
628        Ok(())
629    }
630}
631
632impl DataTester {
633    /// Creates a new [`DataTester`] instance.
634    #[must_use]
635    pub fn new(config: DataTesterConfig) -> Self {
636        Self {
637            core: DataActorCore::new(config.base.clone()),
638            config,
639            books: AHashMap::new(),
640        }
641    }
642}
nautilus_testkit/testers/data/actor.rs

nautilus_testkit/testers/data/
actor.rs
