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, OrderBookDepth10, 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        // Subscribe to data for each instrument
89        for instrument_id in instrument_ids {
90            if self.config.subscribe_instrument {
91                self.subscribe_instrument(instrument_id, client_id, subscribe_params.clone());
92            }
93
94            if self.config.subscribe_book_deltas {
95                self.subscribe_book_deltas(
96                    instrument_id,
97                    self.config.book_type,
98                    None,
99                    client_id,
100                    self.config.manage_book,
101                    subscribe_params.clone(),
102                );
103
104                if self.config.manage_book {
105                    let book = OrderBook::new(instrument_id, self.config.book_type);
106                    self.books.insert(instrument_id, book);
107                }
108            }
109
110            if self.config.subscribe_book_at_interval {
111                self.subscribe_book_at_interval(
112                    instrument_id,
113                    self.config.book_type,
114                    self.config
115                        .book_depth
116                        .map(|depth| {
117                            NonZeroUsize::new(depth).ok_or_else(|| {
118                                ConfigError::range("book_depth", "must be positive, was 0")
119                            })
120                        })
121                        .transpose()?,
122                    NonZeroUsize::new(self.config.book_interval_ms).ok_or_else(|| {
123                        ConfigError::range("book_interval_ms", "must be positive, was 0")
124                    })?,
125                    client_id,
126                    subscribe_params.clone(),
127                );
128            }
129
130            if self.config.subscribe_book_depth {
131                self.subscribe_book_depth10(
132                    instrument_id,
133                    self.config.book_type,
134                    client_id,
135                    self.config.manage_book,
136                    subscribe_params.clone(),
137                );
138            }
139
140            if self.config.subscribe_quotes {
141                self.subscribe_quotes(instrument_id, client_id, subscribe_params.clone());
142            }
143
144            if self.config.subscribe_trades {
145                self.subscribe_trades(instrument_id, client_id, subscribe_params.clone());
146            }
147
148            if self.config.subscribe_mark_prices {
149                self.subscribe_mark_prices(instrument_id, client_id, subscribe_params.clone());
150            }
151
152            if self.config.subscribe_index_prices {
153                self.subscribe_index_prices(instrument_id, client_id, subscribe_params.clone());
154            }
155
156            if self.config.subscribe_funding_rates {
157                self.subscribe_funding_rates(instrument_id, client_id, subscribe_params.clone());
158            }
159
160            if self.config.subscribe_instrument_status {
161                self.subscribe_instrument_status(
162                    instrument_id,
163                    client_id,
164                    subscribe_params.clone(),
165                );
166            }
167
168            if self.config.subscribe_instrument_close {
169                self.subscribe_instrument_close(instrument_id, client_id, subscribe_params.clone());
170            }
171
172            if self.config.subscribe_option_greeks {
173                self.subscribe_option_greeks(instrument_id, client_id, subscribe_params.clone());
174            }
175
176            // Request historical quotes (default to last 1 hour)
177            if self.config.request_quotes {
178                let start = self.clock().utc_now() - SignedDuration::from_hours(1);
179
180                if let Err(e) = self.request_quotes(
181                    instrument_id,
182                    Some(start),
183                    None,
184                    None,
185                    client_id,
186                    request_params.clone(),
187                ) {
188                    log::error!("Failed to request quotes for {instrument_id}: {e}");
189                }
190            }
191
192            // Request order book snapshot if configured
193            if self.config.request_book_snapshot {
194                let _ = self.request_book_snapshot(
195                    instrument_id,
196                    self.config
197                        .book_depth
198                        .map(|depth| {
199                            NonZeroUsize::new(depth).ok_or_else(|| {
200                                ConfigError::range("book_depth", "must be positive, was 0")
201                            })
202                        })
203                        .transpose()?,
204                    client_id,
205                    request_params.clone(),
206                );
207            }
208
209            // TODO: Request book deltas when Rust data engine has RequestBookDeltas
210
211            // Request historical trades (default to last 1 hour)
212            if self.config.request_trades {
213                let start = self.clock().utc_now() - SignedDuration::from_hours(1);
214
215                if let Err(e) = self.request_trades(
216                    instrument_id,
217                    Some(start),
218                    None,
219                    None,
220                    client_id,
221                    request_params.clone(),
222                ) {
223                    log::error!("Failed to request trades for {instrument_id}: {e}");
224                }
225            }
226
227            // Request historical funding rates (default to last 7 days)
228            if self.config.request_funding_rates {
229                let start = self.clock().utc_now() - SignedDuration::from_hours(7 * 24);
230
231                if let Err(e) = self.request_funding_rates(
232                    instrument_id,
233                    Some(start),
234                    None,
235                    None,
236                    client_id,
237                    request_params.clone(),
238                ) {
239                    log::error!("Failed to request funding rates for {instrument_id}: {e}");
240                }
241            }
242        }
243
244        // Subscribe to bars
245        if let Some(bar_types) = self.config.bar_types.clone() {
246            for bar_type in bar_types {
247                if self.config.subscribe_bars {
248                    self.subscribe_bars(bar_type, client_id, subscribe_params.clone());
249                }
250
251                // Request historical bars (default to last 1 hour)
252                if self.config.request_bars {
253                    let start = self.clock().utc_now() - SignedDuration::from_hours(1);
254
255                    if let Err(e) = self.request_bars(
256                        bar_type,
257                        Some(start),
258                        None,
259                        None,
260                        client_id,
261                        request_params.clone(),
262                    ) {
263                        log::error!("Failed to request bars for {bar_type}: {e}");
264                    }
265                }
266            }
267        }
268
269        // Set up stats timer
270        if stats_interval_secs > 0 {
271            self.clock().set_timer(
272                "STATS-TIMER",
273                Duration::from_secs(stats_interval_secs),
274                None,
275                None,
276                None,
277                Some(true),
278                Some(false),
279            )?;
280        }
281
282        Ok(())
283    }
284
285    fn on_stop(&mut self) -> anyhow::Result<()> {
286        if !self.config.can_unsubscribe {
287            return Ok(());
288        }
289
290        let instrument_ids = self.config.instrument_ids.clone();
291        let client_id = self.config.client_id;
292        let subscribe_params = self.config.subscribe_params.clone();
293
294        for instrument_id in instrument_ids {
295            if self.config.subscribe_instrument {
296                self.unsubscribe_instrument(instrument_id, client_id, subscribe_params.clone());
297            }
298
299            if self.config.subscribe_book_deltas {
300                self.unsubscribe_book_deltas(instrument_id, client_id, subscribe_params.clone());
301            }
302
303            if self.config.subscribe_book_at_interval {
304                self.unsubscribe_book_at_interval(
305                    instrument_id,
306                    NonZeroUsize::new(self.config.book_interval_ms).ok_or_else(|| {
307                        ConfigError::range("book_interval_ms", "must be positive, was 0")
308                    })?,
309                    client_id,
310                    subscribe_params.clone(),
311                );
312            }
313
314            if self.config.subscribe_book_depth {
315                self.unsubscribe_book_depth10(instrument_id, client_id, subscribe_params.clone());
316            }
317
318            if self.config.subscribe_quotes {
319                self.unsubscribe_quotes(instrument_id, client_id, subscribe_params.clone());
320            }
321
322            if self.config.subscribe_trades {
323                self.unsubscribe_trades(instrument_id, client_id, subscribe_params.clone());
324            }
325
326            if self.config.subscribe_mark_prices {
327                self.unsubscribe_mark_prices(instrument_id, client_id, subscribe_params.clone());
328            }
329
330            if self.config.subscribe_index_prices {
331                self.unsubscribe_index_prices(instrument_id, client_id, subscribe_params.clone());
332            }
333
334            if self.config.subscribe_funding_rates {
335                self.unsubscribe_funding_rates(instrument_id, client_id, subscribe_params.clone());
336            }
337
338            if self.config.subscribe_instrument_status {
339                self.unsubscribe_instrument_status(
340                    instrument_id,
341                    client_id,
342                    subscribe_params.clone(),
343                );
344            }
345
346            if self.config.subscribe_instrument_close {
347                self.unsubscribe_instrument_close(
348                    instrument_id,
349                    client_id,
350                    subscribe_params.clone(),
351                );
352            }
353
354            if self.config.subscribe_option_greeks {
355                self.unsubscribe_option_greeks(instrument_id, client_id, subscribe_params.clone());
356            }
357        }
358
359        if let Some(bar_types) = self.config.bar_types.clone() {
360            for bar_type in bar_types {
361                if self.config.subscribe_bars {
362                    self.unsubscribe_bars(bar_type, client_id, subscribe_params.clone());
363                }
364            }
365        }
366
367        Ok(())
368    }
369
370    fn on_time_event(&mut self, _event: &TimeEvent) -> anyhow::Result<()> {
371        // Timer events are used by the actor but don't require specific handling
372        Ok(())
373    }
374
375    fn on_instrument(&mut self, instrument: &InstrumentAny) -> anyhow::Result<()> {
376        warn_if_implausible_unix_nanos("instrument", instrument.ts_event(), instrument.ts_init());
377
378        if self.config.log_data {
379            log_info!("{instrument:?}", color = LogColor::Cyan);
380        }
381        Ok(())
382    }
383
384    fn on_book(&mut self, book: &OrderBook) -> anyhow::Result<()> {
385        if self.config.log_data {
386            let levels = self.config.book_levels_to_print;
387            let instrument_id = book.instrument_id;
388            let book_str = book.pprint(levels, None);
389            log_info!("\n{instrument_id}\n{book_str}", color = LogColor::Cyan);
390        }
391
392        Ok(())
393    }
394
395    fn on_book_deltas(&mut self, deltas: &OrderBookDeltas) -> anyhow::Result<()> {
396        warn_if_implausible_unix_nanos("book deltas", deltas.ts_event, deltas.ts_init);
397
398        if self.config.manage_book {
399            if let Some(book) = self.books.get_mut(&deltas.instrument_id) {
400                book.apply_deltas(deltas)?;
401
402                if self.config.log_data {
403                    let levels = self.config.book_levels_to_print;
404                    let instrument_id = deltas.instrument_id;
405                    let book_str = book.pprint(levels, None);
406                    log_info!("\n{instrument_id}\n{book_str}", color = LogColor::Cyan);
407                }
408            }
409        } else if self.config.log_data {
410            log_info!("{deltas:?}", color = LogColor::Cyan);
411        }
412        Ok(())
413    }
414
415    fn on_book_depth(&mut self, depth: &OrderBookDepth10) -> anyhow::Result<()> {
416        warn_if_implausible_unix_nanos("book depth", depth.ts_event, depth.ts_init);
417
418        if self.config.log_data {
419            log_info!("{depth:?}", color = LogColor::Cyan);
420        }
421        Ok(())
422    }
423
424    fn on_quote(&mut self, quote: &QuoteTick) -> anyhow::Result<()> {
425        warn_if_implausible_unix_nanos("quote", quote.ts_event, quote.ts_init);
426
427        if self.config.log_data {
428            log_info!("{quote:?}", color = LogColor::Cyan);
429        }
430        Ok(())
431    }
432
433    fn on_trade(&mut self, trade: &TradeTick) -> anyhow::Result<()> {
434        warn_if_implausible_unix_nanos("trade", trade.ts_event, trade.ts_init);
435
436        if self.config.log_data {
437            log_info!("{trade:?}", color = LogColor::Cyan);
438        }
439        Ok(())
440    }
441
442    fn on_bar(&mut self, bar: &Bar) -> anyhow::Result<()> {
443        warn_if_implausible_unix_nanos("bar", bar.ts_event, bar.ts_init);
444
445        if self.config.log_data {
446            log_info!("{bar:?}", color = LogColor::Cyan);
447        }
448        Ok(())
449    }
450
451    fn on_mark_price(&mut self, mark_price: &MarkPriceUpdate) -> anyhow::Result<()> {
452        warn_if_implausible_unix_nanos("mark price", mark_price.ts_event, mark_price.ts_init);
453
454        if self.config.log_data {
455            log_info!("{mark_price:?}", color = LogColor::Cyan);
456        }
457        Ok(())
458    }
459
460    fn on_index_price(&mut self, index_price: &IndexPriceUpdate) -> anyhow::Result<()> {
461        warn_if_implausible_unix_nanos("index price", index_price.ts_event, index_price.ts_init);
462
463        if self.config.log_data {
464            log_info!("{index_price:?}", color = LogColor::Cyan);
465        }
466        Ok(())
467    }
468
469    fn on_funding_rate(&mut self, funding_rate: &FundingRateUpdate) -> anyhow::Result<()> {
470        warn_if_implausible_unix_nanos("funding rate", funding_rate.ts_event, funding_rate.ts_init);
471        warn_if_implausible_optional(
472            "funding rate",
473            "next_funding_ns",
474            funding_rate.next_funding_ns,
475        );
476
477        if self.config.log_data {
478            log_info!("{funding_rate:?}", color = LogColor::Cyan);
479        }
480        Ok(())
481    }
482
483    fn on_instrument_status(&mut self, data: &InstrumentStatus) -> anyhow::Result<()> {
484        warn_if_implausible_unix_nanos("instrument status", data.ts_event, data.ts_init);
485
486        if self.config.log_data {
487            log_info!("{data:?}", color = LogColor::Cyan);
488        }
489        Ok(())
490    }
491
492    fn on_instrument_close(&mut self, update: &InstrumentClose) -> anyhow::Result<()> {
493        warn_if_implausible_unix_nanos("instrument close", update.ts_event, update.ts_init);
494
495        if self.config.log_data {
496            log_info!("{update:?}", color = LogColor::Cyan);
497        }
498        Ok(())
499    }
500
501    fn on_option_greeks(&mut self, greeks: &OptionGreeks) -> anyhow::Result<()> {
502        warn_if_implausible_unix_nanos("option greeks", greeks.ts_event, greeks.ts_init);
503
504        if self.config.log_data {
505            log_info!("{greeks:?}", color = LogColor::Cyan);
506        }
507        Ok(())
508    }
509
510    fn on_historical_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
511        for trade in trades {
512            warn_if_implausible_unix_nanos("historical trade", trade.ts_event, trade.ts_init);
513        }
514
515        if self.config.log_data {
516            log_info!(
517                "Received {} historical trades",
518                trades.len(),
519                color = LogColor::Cyan
520            );
521
522            for trade in trades.iter().take(5) {
523                log_info!("  {trade:?}", color = LogColor::Cyan);
524            }
525
526            if trades.len() > 5 {
527                log_info!(
528                    "  ... and {} more trades",
529                    trades.len() - 5,
530                    color = LogColor::Cyan
531                );
532            }
533        }
534        Ok(())
535    }
536
537    fn on_historical_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
538        for quote in quotes {
539            warn_if_implausible_unix_nanos("historical quote", quote.ts_event, quote.ts_init);
540        }
541
542        if self.config.log_data {
543            log_info!(
544                "Received {} historical quotes",
545                quotes.len(),
546                color = LogColor::Cyan
547            );
548
549            for quote in quotes.iter().take(5) {
550                log_info!("  {quote:?}", color = LogColor::Cyan);
551            }
552
553            if quotes.len() > 5 {
554                log_info!(
555                    "  ... and {} more quotes",
556                    quotes.len() - 5,
557                    color = LogColor::Cyan
558                );
559            }
560        }
561        Ok(())
562    }
563
564    fn on_historical_funding_rates(
565        &mut self,
566        funding_rates: &[FundingRateUpdate],
567    ) -> anyhow::Result<()> {
568        for rate in funding_rates {
569            warn_if_implausible_unix_nanos("historical funding rate", rate.ts_event, rate.ts_init);
570            warn_if_implausible_optional(
571                "historical funding rate",
572                "next_funding_ns",
573                rate.next_funding_ns,
574            );
575        }
576
577        if self.config.log_data {
578            log_info!(
579                "Received {} historical funding rates",
580                funding_rates.len(),
581                color = LogColor::Cyan
582            );
583
584            for rate in funding_rates.iter().take(5) {
585                log_info!("  {rate:?}", color = LogColor::Cyan);
586            }
587
588            if funding_rates.len() > 5 {
589                log_info!(
590                    "  ... and {} more funding rates",
591                    funding_rates.len() - 5,
592                    color = LogColor::Cyan
593                );
594            }
595        }
596        Ok(())
597    }
598
599    fn on_historical_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
600        for bar in bars {
601            warn_if_implausible_unix_nanos("historical bar", bar.ts_event, bar.ts_init);
602        }
603
604        if self.config.log_data {
605            log_info!(
606                "Received {} historical bars",
607                bars.len(),
608                color = LogColor::Cyan
609            );
610
611            for bar in bars.iter().take(5) {
612                log_info!("  {bar:?}", color = LogColor::Cyan);
613            }
614
615            if bars.len() > 5 {
616                log_info!(
617                    "  ... and {} more bars",
618                    bars.len() - 5,
619                    color = LogColor::Cyan
620                );
621            }
622        }
623        Ok(())
624    }
625}
626
627impl DataTester {
628    /// Creates a new [`DataTester`] instance.
629    #[must_use]
630    pub fn new(config: DataTesterConfig) -> Self {
631        Self {
632            core: DataActorCore::new(config.base.clone()),
633            config,
634            books: AHashMap::new(),
635        }
636    }
637}
nautilus_testkit/testers/data/actor.rs

nautilus_testkit/testers/data/
actor.rs
