nautilus_testkit/testers/data/

actor.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

use std::{num::NonZeroUsize, time::Duration};

use ahash::{AHashMap, AHashSet};
use jiff::SignedDuration;
use nautilus_common::{
    actor::{DataActor, DataActorCore},
    config::ConfigError,
    enums::LogColor,
    log_info, nautilus_actor,
    timer::TimeEvent,
};
use nautilus_model::{
    data::{
        Bar, FundingRateUpdate, IndexPriceUpdate, InstrumentClose, InstrumentStatus,
        MarkPriceUpdate, OrderBookDeltas, OrderBookDepth, QuoteTick, TradeTick,
        option_chain::OptionGreeks,
    },
    identifiers::InstrumentId,
    instruments::{Instrument, InstrumentAny},
    orderbook::OrderBook,
};

use super::config::DataTesterConfig;
use crate::testers::timestamps::{warn_if_implausible_optional, warn_if_implausible_unix_nanos};

/// A data tester actor for live testing market data subscriptions.
///
/// Subscribes to configured data types for specified instruments and logs
/// received data to demonstrate the data flow. Useful for testing adapters
/// and validating data connectivity.
///
/// This actor provides equivalent functionality to the Python `DataTester`
/// in the test kit.
#[derive(Debug)]
pub struct DataTester {
    pub(super) core: DataActorCore,
    pub(super) config: DataTesterConfig,
    pub(super) books: AHashMap<InstrumentId, OrderBook>,
}

nautilus_actor!(DataTester);

impl DataActor for DataTester {
    #[expect(
        clippy::too_many_lines,
        reason = "startup subscribes to each configured data scenario explicitly"
    )]
    fn on_start(&mut self) -> anyhow::Result<()> {
        let instrument_ids = self.config.instrument_ids.clone();
        let client_id = self.config.client_id;
        let subscribe_params = self.config.subscribe_params.clone();
        let request_params = self.config.request_params.clone();
        let stats_interval_secs = self.config.stats_interval_secs;
        let request_end = self.config.request_end;
        let request_anchor = request_end.unwrap_or_else(|| self.clock().utc_now());
        let request_start = self
            .config
            .request_start
            .unwrap_or_else(|| request_anchor - SignedDuration::from_hours(1));
        let request_limit = self.config.request_limit;

        // Request instruments if configured
        if self.config.request_instruments {
            let mut venues = AHashSet::new();
            for instrument_id in &instrument_ids {
                venues.insert(instrument_id.venue);
            }

            for venue in venues {
                let _ = self.request_instruments(
                    Some(venue),
                    None,
                    None,
                    client_id,
                    request_params.clone(),
                );
            }
        }

        let book_depth = self
            .config
            .book_depth
            .map(|depth| {
                NonZeroUsize::new(depth)
                    .ok_or_else(|| ConfigError::range("book_depth", "must be positive, was 0"))
            })
            .transpose()?;

        // Subscribe to data for each instrument
        for instrument_id in instrument_ids {
            if self.config.subscribe_instrument {
                self.subscribe_instrument(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_book_deltas {
                self.subscribe_book_deltas(
                    instrument_id,
                    self.config.book_type,
                    book_depth,
                    client_id,
                    self.config.manage_book,
                    subscribe_params.clone(),
                );

                if self.config.manage_book {
                    let book = OrderBook::new(instrument_id, self.config.book_type);
                    self.books.insert(instrument_id, book);
                }
            }

            if self.config.subscribe_book_at_interval {
                self.subscribe_book_at_interval(
                    instrument_id,
                    self.config.book_type,
                    book_depth,
                    NonZeroUsize::new(self.config.book_interval_ms).ok_or_else(|| {
                        ConfigError::range("book_interval_ms", "must be positive, was 0")
                    })?,
                    client_id,
                    subscribe_params.clone(),
                );
            }

            if self.config.subscribe_book_depth {
                self.subscribe_book_depth(
                    instrument_id,
                    self.config.book_type,
                    book_depth,
                    client_id,
                    self.config.manage_book
                        && !self.config.subscribe_book_deltas
                        && !self.config.subscribe_book_at_interval,
                    subscribe_params.clone(),
                );
            }

            if self.config.subscribe_quotes {
                self.subscribe_quotes(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_trades {
                self.subscribe_trades(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_mark_prices {
                self.subscribe_mark_prices(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_index_prices {
                self.subscribe_index_prices(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_funding_rates {
                self.subscribe_funding_rates(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_instrument_status {
                self.subscribe_instrument_status(
                    instrument_id,
                    client_id,
                    subscribe_params.clone(),
                );
            }

            if self.config.subscribe_instrument_close {
                self.subscribe_instrument_close(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_option_greeks {
                self.subscribe_option_greeks(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.request_quotes
                && let Err(e) = self.request_quotes(
                    instrument_id,
                    Some(request_start),
                    request_end,
                    request_limit,
                    client_id,
                    request_params.clone(),
                )
            {
                log::error!("Failed to request quotes for {instrument_id}: {e}");
            }

            // Request order book snapshot if configured
            if self.config.request_book_snapshot {
                let _ = self.request_book_snapshot(
                    instrument_id,
                    self.config
                        .book_depth
                        .map(|depth| {
                            NonZeroUsize::new(depth).ok_or_else(|| {
                                ConfigError::range("book_depth", "must be positive, was 0")
                            })
                        })
                        .transpose()?,
                    client_id,
                    request_params.clone(),
                );
            }

            // TODO: Request book deltas when Rust data engine has RequestBookDeltas

            if self.config.request_trades
                && let Err(e) = self.request_trades(
                    instrument_id,
                    Some(request_start),
                    request_end,
                    request_limit,
                    client_id,
                    request_params.clone(),
                )
            {
                log::error!("Failed to request trades for {instrument_id}: {e}");
            }

            if self.config.request_funding_rates {
                let start = self
                    .config
                    .request_start
                    .unwrap_or_else(|| request_anchor - SignedDuration::from_hours(7 * 24));

                if let Err(e) = self.request_funding_rates(
                    instrument_id,
                    Some(start),
                    request_end,
                    request_limit,
                    client_id,
                    request_params.clone(),
                ) {
                    log::error!("Failed to request funding rates for {instrument_id}: {e}");
                }
            }
        }

        // Subscribe to bars
        if let Some(bar_types) = self.config.bar_types.clone() {
            for bar_type in bar_types {
                if self.config.subscribe_bars {
                    self.subscribe_bars(bar_type, client_id, subscribe_params.clone());
                }

                if self.config.request_bars
                    && let Err(e) = self.request_bars(
                        bar_type,
                        Some(request_start),
                        request_end,
                        request_limit,
                        client_id,
                        request_params.clone(),
                    )
                {
                    log::error!("Failed to request bars for {bar_type}: {e}");
                }
            }
        }

        // Set up stats timer
        if stats_interval_secs > 0 {
            self.clock().set_timer(
                "STATS-TIMER",
                Duration::from_secs(stats_interval_secs),
                None,
                None,
                None,
                Some(true),
                Some(false),
            )?;
        }

        Ok(())
    }

    fn on_stop(&mut self) -> anyhow::Result<()> {
        if !self.config.can_unsubscribe {
            return Ok(());
        }

        let instrument_ids = self.config.instrument_ids.clone();
        let client_id = self.config.client_id;
        let subscribe_params = self.config.subscribe_params.clone();

        for instrument_id in instrument_ids {
            if self.config.subscribe_instrument {
                self.unsubscribe_instrument(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_book_deltas {
                self.unsubscribe_book_deltas(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_book_at_interval {
                self.unsubscribe_book_at_interval(
                    instrument_id,
                    NonZeroUsize::new(self.config.book_interval_ms).ok_or_else(|| {
                        ConfigError::range("book_interval_ms", "must be positive, was 0")
                    })?,
                    client_id,
                    subscribe_params.clone(),
                );
            }

            if self.config.subscribe_book_depth {
                self.unsubscribe_book_depth(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_quotes {
                self.unsubscribe_quotes(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_trades {
                self.unsubscribe_trades(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_mark_prices {
                self.unsubscribe_mark_prices(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_index_prices {
                self.unsubscribe_index_prices(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_funding_rates {
                self.unsubscribe_funding_rates(instrument_id, client_id, subscribe_params.clone());
            }

            if self.config.subscribe_instrument_status {
                self.unsubscribe_instrument_status(
                    instrument_id,
                    client_id,
                    subscribe_params.clone(),
                );
            }

            if self.config.subscribe_instrument_close {
                self.unsubscribe_instrument_close(
                    instrument_id,
                    client_id,
                    subscribe_params.clone(),
                );
            }

            if self.config.subscribe_option_greeks {
                self.unsubscribe_option_greeks(instrument_id, client_id, subscribe_params.clone());
            }
        }

        if let Some(bar_types) = self.config.bar_types.clone() {
            for bar_type in bar_types {
                if self.config.subscribe_bars {
                    self.unsubscribe_bars(bar_type, client_id, subscribe_params.clone());
                }
            }
        }

        Ok(())
    }

    fn on_time_event(&mut self, _event: &TimeEvent) -> anyhow::Result<()> {
        // Timer events are used by the actor but don't require specific handling
        Ok(())
    }

    fn on_instrument(&mut self, instrument: &InstrumentAny) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("instrument", instrument.ts_event(), instrument.ts_init());

        if self.config.log_data {
            log_info!("{instrument:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_book(&mut self, book: &OrderBook) -> anyhow::Result<()> {
        if self.config.log_data {
            let levels = self.config.book_levels_to_print;
            let instrument_id = book.instrument_id;
            let book_str = book.pprint(levels, None);
            log_info!("\n{instrument_id}\n{book_str}", color = LogColor::Cyan);
        }

        Ok(())
    }

    fn on_book_deltas(&mut self, deltas: &OrderBookDeltas) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("book deltas", deltas.ts_event, deltas.ts_init);

        if self.config.manage_book {
            if let Some(book) = self.books.get_mut(&deltas.instrument_id) {
                book.apply_deltas(deltas)?;

                if self.config.log_data {
                    let levels = self.config.book_levels_to_print;
                    let instrument_id = deltas.instrument_id;
                    let book_str = book.pprint(levels, None);
                    log_info!("\n{instrument_id}\n{book_str}", color = LogColor::Cyan);
                }
            }
        } else if self.config.log_data {
            log_info!("{deltas:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_book_depth(&mut self, depth: &OrderBookDepth) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("book depth", depth.ts_event, depth.ts_init);

        if self.config.log_data {
            log_info!("{depth:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_quote(&mut self, quote: &QuoteTick) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("quote", quote.ts_event, quote.ts_init);

        if self.config.log_data {
            log_info!("{quote:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_trade(&mut self, trade: &TradeTick) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("trade", trade.ts_event, trade.ts_init);

        if self.config.log_data {
            log_info!("{trade:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_bar(&mut self, bar: &Bar) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("bar", bar.ts_event, bar.ts_init);

        if self.config.log_data {
            log_info!("{bar:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_mark_price(&mut self, mark_price: &MarkPriceUpdate) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("mark price", mark_price.ts_event, mark_price.ts_init);

        if self.config.log_data {
            log_info!("{mark_price:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_index_price(&mut self, index_price: &IndexPriceUpdate) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("index price", index_price.ts_event, index_price.ts_init);

        if self.config.log_data {
            log_info!("{index_price:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_funding_rate(&mut self, funding_rate: &FundingRateUpdate) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("funding rate", funding_rate.ts_event, funding_rate.ts_init);
        warn_if_implausible_optional(
            "funding rate",
            "next_funding_ns",
            funding_rate.next_funding_ns,
        );

        if self.config.log_data {
            log_info!("{funding_rate:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_instrument_status(&mut self, data: &InstrumentStatus) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("instrument status", data.ts_event, data.ts_init);

        if self.config.log_data {
            log_info!("{data:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_instrument_close(&mut self, update: &InstrumentClose) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("instrument close", update.ts_event, update.ts_init);

        if self.config.log_data {
            log_info!("{update:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_option_greeks(&mut self, greeks: &OptionGreeks) -> anyhow::Result<()> {
        warn_if_implausible_unix_nanos("option greeks", greeks.ts_event, greeks.ts_init);

        if self.config.log_data {
            log_info!("{greeks:?}", color = LogColor::Cyan);
        }
        Ok(())
    }

    fn on_historical_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
        for trade in trades {
            warn_if_implausible_unix_nanos("historical trade", trade.ts_event, trade.ts_init);
        }

        if self.config.log_data {
            log_info!(
                "Received {} historical trades",
                trades.len(),
                color = LogColor::Cyan
            );

            for trade in trades.iter().take(5) {
                log_info!("  {trade:?}", color = LogColor::Cyan);
            }

            if trades.len() > 5 {
                log_info!(
                    "  ... and {} more trades",
                    trades.len() - 5,
                    color = LogColor::Cyan
                );
            }
        }
        Ok(())
    }

    fn on_historical_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
        for quote in quotes {
            warn_if_implausible_unix_nanos("historical quote", quote.ts_event, quote.ts_init);
        }

        if self.config.log_data {
            log_info!(
                "Received {} historical quotes",
                quotes.len(),
                color = LogColor::Cyan
            );

            for quote in quotes.iter().take(5) {
                log_info!("  {quote:?}", color = LogColor::Cyan);
            }

            if quotes.len() > 5 {
                log_info!(
                    "  ... and {} more quotes",
                    quotes.len() - 5,
                    color = LogColor::Cyan
                );
            }
        }
        Ok(())
    }

    fn on_historical_funding_rates(
        &mut self,
        funding_rates: &[FundingRateUpdate],
    ) -> anyhow::Result<()> {
        for rate in funding_rates {
            warn_if_implausible_unix_nanos("historical funding rate", rate.ts_event, rate.ts_init);
            warn_if_implausible_optional(
                "historical funding rate",
                "next_funding_ns",
                rate.next_funding_ns,
            );
        }

        if self.config.log_data {
            log_info!(
                "Received {} historical funding rates",
                funding_rates.len(),
                color = LogColor::Cyan
            );

            for rate in funding_rates.iter().take(5) {
                log_info!("  {rate:?}", color = LogColor::Cyan);
            }

            if funding_rates.len() > 5 {
                log_info!(
                    "  ... and {} more funding rates",
                    funding_rates.len() - 5,
                    color = LogColor::Cyan
                );
            }
        }
        Ok(())
    }

    fn on_historical_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
        for bar in bars {
            warn_if_implausible_unix_nanos("historical bar", bar.ts_event, bar.ts_init);
        }

        if self.config.log_data {
            log_info!(
                "Received {} historical bars",
                bars.len(),
                color = LogColor::Cyan
            );

            for bar in bars.iter().take(5) {
                log_info!("  {bar:?}", color = LogColor::Cyan);
            }

            if bars.len() > 5 {
                log_info!(
                    "  ... and {} more bars",
                    bars.len() - 5,
                    color = LogColor::Cyan
                );
            }
        }
        Ok(())
    }
}

impl DataTester {
    /// Creates a new [`DataTester`] instance.
    #[must_use]
    pub fn new(config: DataTesterConfig) -> Self {
        Self {
            core: DataActorCore::new(config.base.clone()),
            config,
            books: AHashMap::new(),
        }
    }
}