nautilus_databento/

data.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.
// -------------------------------------------------------------------------------------------------

//! Provides a unified data client that combines Databento's live streaming and historical data capabilities.
//!
//! This module implements a data client that manages connections to multiple Databento datasets,
//! handles live market data subscriptions, and provides access to historical data on demand.

use std::{
    path::PathBuf,
    str::FromStr,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use ahash::AHashMap;
use databento::{dbn, live::Subscription};
use indexmap::IndexMap;
use nautilus_common::{
    clients::DataClient,
    live::runner::get_data_event_sender,
    messages::{
        DataEvent, DataResponse,
        data::{
            BarsResponse, BookDeltasResponse, BookDepthResponse, InstrumentResponse,
            InstrumentsResponse, QuotesResponse, RequestBars, RequestBookDeltas, RequestBookDepth,
            RequestInstrument, RequestInstruments, RequestQuotes, RequestTrades,
            SubscribeBookDeltas, SubscribeInstrument, SubscribeInstrumentStatus, SubscribeQuotes,
            SubscribeTrades, TradesResponse, UnsubscribeBookDeltas, UnsubscribeInstrumentStatus,
            UnsubscribeQuotes, UnsubscribeTrades,
        },
    },
};
use nautilus_core::{
    AtomicMap, DurationNanos, Params, UnixNanos,
    datetime::datetime_to_unix_nanos,
    time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_live::task::TaskGroup;
use nautilus_model::{
    data::{CustomData, Data},
    enums::BarAggregation,
    identifiers::{ClientId, InstrumentId, Symbol, Venue},
    instruments::{Instrument, InstrumentAny},
};
use parking_lot::Mutex;
use tokio_util::sync::CancellationToken;

use crate::{
    common::{Credential, DATABENTO_VENUE},
    historical::{DatabentoHistoricalClient, RangeQueryParams},
    live::{DatabentoFeedHandler, DatabentoMessage, HandlerCommand},
    loader::DatabentoDataLoader,
    symbology::instrument_id_to_symbol_string,
    types::{Dataset, PublisherId},
};

const PRICE_PRECISION_PARAM: &str = "price_precision";
const SCHEMA_PARAM: &str = "schema";
const QUOTE_SCHEMAS: &[dbn::Schema] = &[
    dbn::Schema::Mbp1,
    dbn::Schema::Bbo1S,
    dbn::Schema::Bbo1M,
    dbn::Schema::Cmbp1,
    dbn::Schema::Cbbo1S,
    dbn::Schema::Cbbo1M,
    dbn::Schema::Tbbo,
    dbn::Schema::Tcbbo,
];
const TRADE_SCHEMAS: &[dbn::Schema] = &[
    dbn::Schema::Trades,
    dbn::Schema::Tbbo,
    dbn::Schema::Tcbbo,
    dbn::Schema::Mbp1,
    dbn::Schema::Cmbp1,
];

/// Configuration for the Databento data client.
#[derive(Debug, Clone)]
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.adapters.databento", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.databento")
)]
pub struct DatabentoDataClientConfig {
    /// Databento API credential.
    pub(crate) credential: Credential,
    /// Path to publishers.json file.
    pub publishers_filepath: PathBuf,
    /// Venue-to-dataset overrides applied on top of the publishers.json mappings.
    pub venue_dataset_map: IndexMap<String, String>,
    /// Whether to use exchange as venue for GLBX instruments.
    pub use_exchange_as_venue: bool,
    /// Whether to timestamp bars on close.
    pub bars_timestamp_on_close: bool,
    /// Reconnection timeout in minutes (None for infinite retries).
    pub reconnect_timeout_mins: Option<u64>,
}

#[cfg(feature = "python")]
nautilus_core::impl_pyo3_config_getters!(DatabentoDataClientConfig {
    publishers_filepath: PathBuf,
    use_exchange_as_venue: bool,
    bars_timestamp_on_close: bool,
    venue_dataset_map: IndexMap<String, String>,
});

impl DatabentoDataClientConfig {
    /// Creates a new [`DatabentoDataClientConfig`] instance.
    #[must_use]
    pub fn new(
        api_key: impl Into<String>,
        publishers_filepath: PathBuf,
        use_exchange_as_venue: bool,
        bars_timestamp_on_close: bool,
    ) -> Self {
        Self {
            credential: Credential::new(api_key),
            publishers_filepath,
            venue_dataset_map: IndexMap::new(),
            use_exchange_as_venue,
            bars_timestamp_on_close,
            reconnect_timeout_mins: Some(10), // Default: 10 minutes
        }
    }

    /// Returns the API key associated with this config.
    #[must_use]
    pub fn api_key(&self) -> &str {
        self.credential.api_key()
    }

    /// Returns a masked version of the API key for logging purposes.
    #[must_use]
    pub fn api_key_masked(&self) -> String {
        self.credential.api_key_masked()
    }
}

/// A Databento data client that combines live streaming and historical data functionality.
///
/// This client uses the existing `DatabentoFeedHandler` for live data subscriptions
/// and `DatabentoHistoricalClient` for historical data requests. It supports multiple
/// datasets simultaneously, with separate feed handlers per dataset.
#[cfg_attr(feature = "python", pyo3::pyclass)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.databento")
)]
#[derive(Debug)]
pub struct DatabentoDataClient {
    client_id: ClientId,
    config: DatabentoDataClientConfig,
    is_connected: AtomicBool,
    historical: DatabentoHistoricalClient,
    loader: DatabentoDataLoader,
    cmd_channels: Arc<Mutex<AHashMap<String, tokio::sync::mpsc::UnboundedSender<HandlerCommand>>>>,
    task_handles: TaskGroup,
    cancellation_token: CancellationToken,
    publisher_venue_map: Arc<IndexMap<PublisherId, Venue>>,
    symbol_venue_map: Arc<AtomicMap<Symbol, Venue>>,
    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
}

impl DatabentoDataClient {
    /// Creates a new [`DatabentoDataClient`] instance.
    ///
    /// # Errors
    ///
    /// Returns an error if client creation or publisher configuration loading fails.
    pub fn new(
        client_id: ClientId,
        config: DatabentoDataClientConfig,
        clock: &'static AtomicTime,
    ) -> anyhow::Result<Self> {
        let historical = DatabentoHistoricalClient::new(
            config.credential.clone(),
            config.publishers_filepath.clone(),
            clock,
            config.use_exchange_as_venue,
        )?;

        // Create data loader for venue-to-dataset mapping
        let mut loader = DatabentoDataLoader::new(Some(config.publishers_filepath.clone()))?;
        for (venue, dataset) in &config.venue_dataset_map {
            loader.set_dataset_for_venue(
                Dataset::from(dataset.as_str()),
                Venue::from(venue.as_str()),
            );
        }

        // Load publisher configuration
        let file_content = std::fs::read_to_string(&config.publishers_filepath)?;
        let publishers_vec: Vec<crate::types::DatabentoPublisher> =
            serde_json::from_str(&file_content)?;

        let publisher_venue_map = publishers_vec
            .into_iter()
            .map(|p| (p.publisher_id, Venue::from(p.venue.as_str())))
            .collect::<IndexMap<u16, Venue>>();

        let data_sender = get_data_event_sender();

        let task_handles = TaskGroup::new();

        Ok(Self {
            client_id,
            config,
            is_connected: AtomicBool::new(false),
            historical,
            loader,
            cmd_channels: Arc::new(Mutex::new(AHashMap::new())),
            cancellation_token: task_handles.cancellation_token(),
            task_handles,
            publisher_venue_map: Arc::new(publisher_venue_map),
            symbol_venue_map: Arc::new(AtomicMap::new()),
            data_sender,
        })
    }

    /// Returns the API key associated with this client.
    #[must_use]
    pub fn api_key(&self) -> &str {
        self.config.api_key()
    }

    /// Returns a masked version of the API key for logging purposes.
    #[must_use]
    pub fn api_key_masked(&self) -> String {
        self.config.api_key_masked()
    }

    /// Gets the dataset for a given venue using the data loader.
    ///
    /// # Errors
    ///
    /// Returns an error if the venue-to-dataset mapping cannot be found.
    fn get_dataset_for_venue(&self, venue: Venue) -> anyhow::Result<String> {
        self.loader
            .get_dataset_for_venue(&venue)
            .map(ToString::to_string)
            .ok_or_else(|| anyhow::anyhow!("No dataset found for venue: {venue}"))
    }

    /// Gets or creates a feed handler for the specified dataset.
    fn get_or_create_feed_handler(&self, dataset: &str) -> bool {
        let mut channels = self.cmd_channels.lock();

        if !channels.contains_key(dataset) {
            log::debug!("Creating new feed handler for dataset: {dataset}");
            let cmd_tx = self.initialize_live_feed(dataset.to_string());
            channels.insert(dataset.to_string(), cmd_tx);

            log::debug!("Feed handler created for dataset: {dataset}, channel stored");
            return true;
        }

        false
    }

    fn send_subscription_to_dataset(
        &self,
        dataset: &str,
        price_precision: Option<(Symbol, u8)>,
        subscription: Subscription,
        start_after_subscribe: bool,
    ) -> anyhow::Result<()> {
        let tx = {
            let channels = self.cmd_channels.lock();
            channels
                .get(dataset)
                .cloned()
                .ok_or_else(|| anyhow::anyhow!("No feed handler found for dataset: {dataset}"))?
        };

        send_subscription_commands(
            &tx,
            dataset,
            price_precision,
            subscription,
            start_after_subscribe,
        )
    }

    fn send_close_to_active_feeds(&self) {
        let channels = self.cmd_channels.lock();
        for (dataset, tx) in channels.iter() {
            if let Err(e) = tx.send(HandlerCommand::Close) {
                log::warn!("Failed to send close command to dataset {dataset}: {e}");
            }
        }
    }

    fn clear_feed_channels(&self) {
        let mut channels = self.cmd_channels.lock();
        channels.clear();
    }

    fn abort_active_tasks(&self) {
        self.task_handles.begin_shutdown();
    }

    fn spawn_task<F>(&self, future: F)
    where
        F: std::future::Future<Output = ()> + Send + 'static,
    {
        if let Err(e) = self.task_handles.spawn(future) {
            log::debug!("Skipping Databento task after shutdown began: {e}");
        }
    }

    /// Initializes the live feed handler for streaming data.
    fn initialize_live_feed(
        &self,
        dataset: String,
    ) -> tokio::sync::mpsc::UnboundedSender<HandlerCommand> {
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
        let (msg_tx, msg_rx) = tokio::sync::mpsc::unbounded_channel();
        let feed_dataset = dataset.clone();
        let feed_channels = self.cmd_channels.clone();

        let mut feed_handler = DatabentoFeedHandler::new(
            self.config.credential.clone(),
            dataset,
            cmd_rx,
            msg_tx,
            (*self.publisher_venue_map).clone(),
            self.symbol_venue_map.clone(),
            self.config.use_exchange_as_venue,
            self.config.bars_timestamp_on_close,
            self.config.reconnect_timeout_mins,
        );

        let feed_future = async move {
            if let Err(e) = feed_handler.run().await {
                log::error!("Feed handler error: {e}");
            }
            feed_channels.lock().remove(&feed_dataset);
        };

        let cancellation_token = self.cancellation_token.clone();
        let data_sender = self.data_sender.clone();

        // Spawn message processing task with cancellation support
        let msg_future = async move {
            let mut msg_rx = msg_rx;

            loop {
                tokio::select! {
                    msg = msg_rx.recv() => {
                        match msg {
                            Some(DatabentoMessage::Data(data)) => {
                                log::debug!("Received data: {data:?}");
                                if let Err(e) = data_sender.send(DataEvent::Data(data)) {
                                    log::error!("Failed to send data event: {e}");
                                }
                            }
                            Some(DatabentoMessage::Instrument(instrument)) => {
                                log::debug!("Received instrument definition: {}", instrument.id());
                                if let Err(e) = data_sender.send(DataEvent::Instrument(*instrument)) {
                                    log::error!("Failed to send instrument: {e}");
                                }
                            }
                            Some(DatabentoMessage::Status(status)) => {
                                log::debug!("Received status: {status:?}");
                                if let Err(e) =
                                    data_sender.send(DataEvent::Data(Data::InstrumentStatus(status)))
                                {
                                    log::error!("Failed to send status data event: {e}");
                                }
                            }
                            Some(DatabentoMessage::Imbalance(imbalance)) => {
                                log::debug!("Received imbalance: {imbalance:?}");
                                let data = Data::Custom(CustomData::from_arc(Arc::new(imbalance)));
                                if let Err(e) = data_sender.send(DataEvent::Data(data)) {
                                    log::error!("Failed to send imbalance data event: {e}");
                                }
                            }
                            Some(DatabentoMessage::Statistics(statistics)) => {
                                log::debug!("Received statistics: {statistics:?}");
                                let data = Data::Custom(CustomData::from_arc(Arc::new(statistics)));
                                if let Err(e) = data_sender.send(DataEvent::Data(data)) {
                                    log::error!("Failed to send statistics data event: {e}");
                                }
                            }
                            Some(DatabentoMessage::SubscriptionAck(ack)) => {
                                log::debug!("Received subscription ack: {}", ack.message);
                            }
                            Some(DatabentoMessage::Error(error)) => {
                                log::error!("Feed handler error: {error}");
                            }
                            Some(DatabentoMessage::Close) => {
                                log::debug!("Feed handler closed");
                                break;
                            }
                            None => {
                                log::debug!("Message channel closed");
                                break;
                            }
                        }
                    }
                    () = cancellation_token.cancelled() => {
                        log::debug!("Message processing cancelled");
                        break;
                    }
                }
            }
        };

        if let Err(e) = self.task_handles.spawn(feed_future) {
            log::warn!("Skipping Databento feed task after shutdown began: {e}");
        }

        if let Err(e) = self.task_handles.spawn(msg_future) {
            log::warn!("Skipping Databento message task after shutdown began: {e}");
        }

        cmd_tx
    }
}

#[async_trait::async_trait(?Send)]
impl DataClient for DatabentoDataClient {
    /// Returns the client identifier.
    fn client_id(&self) -> ClientId {
        self.client_id
    }

    /// Returns the venue associated with this client (None for multi-venue clients).
    fn venue(&self) -> Option<Venue> {
        None
    }

    /// Starts the data client.
    ///
    /// # Errors
    ///
    /// Returns an error if the client fails to start.
    fn start(&mut self) -> anyhow::Result<()> {
        log::debug!("Starting");
        Ok(())
    }

    /// Stops the data client and cancels all active subscriptions.
    ///
    /// # Errors
    ///
    /// Returns an error if the client fails to stop cleanly.
    fn stop(&mut self) -> anyhow::Result<()> {
        log::debug!("Stopping");

        self.send_close_to_active_feeds();
        self.clear_feed_channels();
        self.cancellation_token.cancel();
        self.abort_active_tasks();
        self.is_connected.store(false, Ordering::Relaxed);

        Ok(())
    }

    fn reset(&mut self) -> anyhow::Result<()> {
        log::debug!("Resetting");
        self.send_close_to_active_feeds();
        self.clear_feed_channels();
        self.abort_active_tasks();
        self.is_connected.store(false, Ordering::Relaxed);
        Ok(())
    }

    fn dispose(&mut self) -> anyhow::Result<()> {
        log::debug!("Disposing");
        self.stop()
    }

    async fn connect(&mut self) -> anyhow::Result<()> {
        log::debug!("Connecting...");

        if !self.task_handles.is_open() {
            self.task_handles
                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
                .await
                .map_err(|e| anyhow::anyhow!("Failed to terminate Databento tasks: {e}"))?;
            self.task_handles
                .start_generation()
                .map_err(|e| anyhow::anyhow!("Failed to start Databento task generation: {e}"))?;
            self.cancellation_token = self.task_handles.cancellation_token();
        }

        self.is_connected.store(true, Ordering::Relaxed);

        log::info!("Connected");
        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        log::debug!("Disconnecting...");

        self.send_close_to_active_feeds();
        self.clear_feed_channels();
        self.task_handles.begin_shutdown();

        let tasks_result = self
            .task_handles
            .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
            .await
            .map_err(|e| anyhow::anyhow!("Failed to terminate Databento tasks: {e}"));

        self.is_connected.store(false, Ordering::Relaxed);

        log::info!("Disconnected");
        tasks_result
    }

    /// Returns whether the client is currently connected.
    fn is_connected(&self) -> bool {
        self.is_connected.load(Ordering::Relaxed)
    }

    fn is_disconnected(&self) -> bool {
        !self.is_connected()
    }

    /// Subscribes to instrument definition data for the specified instrument.
    ///
    /// # Errors
    ///
    /// Returns an error if the subscription request fails.
    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
        let dataset = self.get_dataset_for_venue(cmd.instrument_id.venue)?;
        let start_after_subscribe = self.get_or_create_feed_handler(&dataset);

        self.symbol_venue_map
            .insert(cmd.instrument_id.symbol, cmd.instrument_id.venue);
        let symbol = cmd.instrument_id.symbol.to_string();

        let subscription = Subscription::builder()
            .schema(databento::dbn::Schema::Definition)
            .symbols(symbol)
            .build();

        self.send_subscription_to_dataset(&dataset, None, subscription, start_after_subscribe)?;

        Ok(())
    }

    /// Subscribes to quote tick data for the specified instruments.
    ///
    /// # Errors
    ///
    /// Returns an error if the subscription request fails.
    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
        let dataset = self.get_dataset_for_venue(cmd.instrument_id.venue)?;
        let symbol = cmd.instrument_id.symbol.to_string();
        let price_precision = price_precision_from_params(cmd.params.as_ref())?
            .map(|precision| (cmd.instrument_id.symbol, precision));
        let schema = schema_from_params(cmd.params.as_ref(), dbn::Schema::Mbp1, QUOTE_SCHEMAS)?;

        let subscription = Subscription::builder()
            .schema(schema)
            .symbols(symbol)
            .build();

        let start_after_subscribe = self.get_or_create_feed_handler(&dataset);
        self.symbol_venue_map
            .insert(cmd.instrument_id.symbol, cmd.instrument_id.venue);

        self.send_subscription_to_dataset(
            &dataset,
            price_precision,
            subscription,
            start_after_subscribe,
        )?;

        Ok(())
    }

    /// Subscribes to trade tick data for the specified instruments.
    ///
    /// # Errors
    ///
    /// Returns an error if the subscription request fails.
    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
        let dataset = self.get_dataset_for_venue(cmd.instrument_id.venue)?;
        let symbol = cmd.instrument_id.symbol.to_string();
        let price_precision = price_precision_from_params(cmd.params.as_ref())?
            .map(|precision| (cmd.instrument_id.symbol, precision));
        let schema = schema_from_params(cmd.params.as_ref(), dbn::Schema::Trades, TRADE_SCHEMAS)?;

        let subscription = Subscription::builder()
            .schema(schema)
            .symbols(symbol)
            .build();

        let start_after_subscribe = self.get_or_create_feed_handler(&dataset);
        self.symbol_venue_map
            .insert(cmd.instrument_id.symbol, cmd.instrument_id.venue);

        self.send_subscription_to_dataset(
            &dataset,
            price_precision,
            subscription,
            start_after_subscribe,
        )?;

        Ok(())
    }

    /// Subscribes to order book delta updates for the specified instruments.
    ///
    /// # Errors
    ///
    /// Returns an error if the subscription request fails.
    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
        let dataset = self.get_dataset_for_venue(cmd.instrument_id.venue)?;
        let start_after_subscribe = self.get_or_create_feed_handler(&dataset);

        self.symbol_venue_map
            .insert(cmd.instrument_id.symbol, cmd.instrument_id.venue);
        let symbol = cmd.instrument_id.symbol.to_string();

        let subscription = Subscription::builder()
            .schema(databento::dbn::Schema::Mbo) // Market by order for book deltas
            .symbols(symbol)
            .build();

        self.send_subscription_to_dataset(&dataset, None, subscription, start_after_subscribe)?;

        Ok(())
    }

    /// Subscribes to instrument status updates for the specified instruments.
    ///
    /// # Errors
    ///
    /// Returns an error if the subscription request fails.
    fn subscribe_instrument_status(
        &mut self,
        cmd: SubscribeInstrumentStatus,
    ) -> anyhow::Result<()> {
        let dataset = self.get_dataset_for_venue(cmd.instrument_id.venue)?;
        let start_after_subscribe = self.get_or_create_feed_handler(&dataset);

        self.symbol_venue_map
            .insert(cmd.instrument_id.symbol, cmd.instrument_id.venue);
        let symbol = cmd.instrument_id.symbol.to_string();

        let subscription = Subscription::builder()
            .schema(databento::dbn::Schema::Status)
            .symbols(symbol)
            .build();

        self.send_subscription_to_dataset(&dataset, None, subscription, start_after_subscribe)?;

        Ok(())
    }

    // Unsubscribe methods
    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
        // Note: Databento live API doesn't support granular unsubscribing.
        // The feed handler manages subscriptions and can handle reconnections
        // with the appropriate subscription state.
        log::warn!(
            "Databento does not support granular unsubscribing - ignoring unsubscribe request for {}",
            cmd.instrument_id
        );

        Ok(())
    }

    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
        // Note: Databento live API doesn't support granular unsubscribing.
        // The feed handler manages subscriptions and can handle reconnections
        // with the appropriate subscription state.
        log::warn!(
            "Databento does not support granular unsubscribing - ignoring unsubscribe request for {}",
            cmd.instrument_id
        );

        Ok(())
    }

    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
        // Note: Databento live API doesn't support granular unsubscribing.
        // The feed handler manages subscriptions and can handle reconnections
        // with the appropriate subscription state.
        log::warn!(
            "Databento does not support granular unsubscribing - ignoring unsubscribe request for {}",
            cmd.instrument_id
        );

        Ok(())
    }

    fn unsubscribe_instrument_status(
        &mut self,
        cmd: &UnsubscribeInstrumentStatus,
    ) -> anyhow::Result<()> {
        // Note: Databento live API doesn't support granular unsubscribing.
        // The feed handler manages subscriptions and can handle reconnections
        // with the appropriate subscription state.
        log::warn!(
            "Databento does not support granular unsubscribing - ignoring unsubscribe request for {}",
            cmd.instrument_id
        );

        Ok(())
    }

    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
        log::debug!("Request instruments: {request:?}");

        let historical_client = self.historical.clone();
        let data_sender = self.data_sender.clone();
        let dataset = request
            .venue
            .map(|venue| self.get_dataset_for_venue(venue))
            .transpose()?
            .unwrap_or_else(|| "GLBX.MDP3".to_string());
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let venue = request.venue.unwrap_or(*DATABENTO_VENUE);
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let request_params = request.params;
        let (query_start, query_end) = resolve_request_time_range(start_nanos, end_nanos);

        self.spawn_task(async move {
            let query_params = instruments_query_params(dataset, query_start, query_end);

            match historical_client.get_range_instruments(query_params).await {
                Ok(instruments) => {
                    log::debug!("Retrieved {} instruments", instruments.len());

                    let response = DataResponse::Instruments(InstrumentsResponse::new(
                        request_id,
                        client_id,
                        venue,
                        instruments,
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    if let Err(e) = data_sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send instruments response: {e}");
                    }
                }
                Err(e) => {
                    log::error!("Failed to request instruments: {e}");
                    let response = DataResponse::Instruments(InstrumentsResponse::new(
                        request_id,
                        client_id,
                        venue,
                        Vec::new(),
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    send_data_response(&data_sender, response, "empty instruments");
                }
            }
        });

        Ok(())
    }

    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
        log::debug!("Request instrument: {request:?}");

        let dataset = self.get_dataset_for_venue(request.instrument_id.venue)?;
        let historical_client = self.historical.clone();
        let data_sender = self.data_sender.clone();
        let instrument_id = request.instrument_id;
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let request_params = request.params;
        let (query_start, query_end) = resolve_request_time_range(start_nanos, end_nanos);

        self.spawn_task(async move {
            let query_params =
                instrument_query_params(dataset, instrument_id, query_start, query_end);

            match historical_client.get_range_instruments(query_params).await {
                Ok(instruments) => {
                    let instrument = requested_instrument(instruments, instrument_id);

                    let Some(instrument) = instrument else {
                        log::error!("Instrument not found: {instrument_id}");
                        return;
                    };

                    let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
                        request_id,
                        client_id,
                        instrument.id(),
                        instrument,
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    )));

                    if let Err(e) = data_sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send instrument response: {e}");
                    }
                }
                Err(e) => {
                    log::error!("Failed to request instrument {instrument_id}: {e}");
                }
            }
        });

        Ok(())
    }

    fn request_quotes(&self, request: RequestQuotes) -> anyhow::Result<()> {
        log::debug!("Request quotes: {request:?}");

        let historical_client = self.historical.clone();
        let data_sender = self.data_sender.clone();
        let dataset = self.get_dataset_for_venue(request.instrument_id.venue)?;
        let instrument_id = request.instrument_id;
        let symbols = historical_client.prepare_symbols_from_instrument_ids(&[instrument_id]);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let limit = request.limit.map(|limit| limit.get() as u64);
        let request_params = request.params;
        let price_precision = price_precision_from_params(request_params.as_ref())?;
        let schema = schema_from_params(request_params.as_ref(), dbn::Schema::Mbp1, QUOTE_SCHEMAS)?
            .to_string();
        let (query_start, query_end) = resolve_request_time_range(start_nanos, end_nanos);

        self.spawn_task(async move {
            seed_price_precision_if_needed(
                &historical_client,
                dataset.as_str(),
                instrument_id,
                query_start,
                query_end,
                price_precision,
            )
            .await;

            let params = RangeQueryParams {
                dataset,
                symbols,
                start: query_start,
                end: query_end,
                limit,
                price_precision,
            };

            match historical_client
                .get_range_quotes(params, Some(schema))
                .await
            {
                Ok(quotes) => {
                    log::debug!("Retrieved {} quotes", quotes.len());
                    let response = DataResponse::Quotes(QuotesResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        quotes,
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    if let Err(e) = data_sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send quotes response: {e}");
                    }
                }
                Err(e) => {
                    log::error!("Failed to request quotes: {e}");
                    let response = DataResponse::Quotes(QuotesResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        Vec::new(),
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    send_data_response(&data_sender, response, "empty quotes");
                }
            }
        });

        Ok(())
    }

    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
        log::debug!("Request trades: {request:?}");

        let historical_client = self.historical.clone();
        let data_sender = self.data_sender.clone();
        let dataset = self.get_dataset_for_venue(request.instrument_id.venue)?;
        let instrument_id = request.instrument_id;
        let symbols = historical_client.prepare_symbols_from_instrument_ids(&[instrument_id]);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let limit = request.limit.map(|limit| limit.get() as u64);
        let request_params = request.params;
        let price_precision = price_precision_from_params(request_params.as_ref())?;
        let schema =
            schema_from_params(request_params.as_ref(), dbn::Schema::Trades, TRADE_SCHEMAS)?
                .to_string();
        let (query_start, query_end) = resolve_request_time_range(start_nanos, end_nanos);

        self.spawn_task(async move {
            seed_price_precision_if_needed(
                &historical_client,
                dataset.as_str(),
                instrument_id,
                query_start,
                query_end,
                price_precision,
            )
            .await;

            let params = RangeQueryParams {
                dataset,
                symbols,
                start: query_start,
                end: query_end,
                limit,
                price_precision,
            };

            match historical_client
                .get_range_trades(params, Some(schema))
                .await
            {
                Ok(trades) => {
                    log::debug!("Retrieved {} trades", trades.len());
                    let response = DataResponse::Trades(TradesResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        trades,
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    if let Err(e) = data_sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send trades response: {e}");
                    }
                }
                Err(e) => {
                    log::error!("Failed to request trades: {e}");
                    let response = DataResponse::Trades(TradesResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        Vec::new(),
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    send_data_response(&data_sender, response, "empty trades");
                }
            }
        });

        Ok(())
    }

    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
        log::debug!("Request bars: {request:?}");

        let historical_client = self.historical.clone();
        let data_sender = self.data_sender.clone();
        let instrument_id = request.bar_type.instrument_id();
        let dataset = self.get_dataset_for_venue(instrument_id.venue)?;
        let symbols = historical_client.prepare_symbols_from_instrument_ids(&[instrument_id]);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let bar_type = request.bar_type;
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let limit = request.limit.map(|limit| limit.get() as u64);
        let request_params = request.params;
        let price_precision = price_precision_from_params(request_params.as_ref())?;
        let timestamp_on_close = self.config.bars_timestamp_on_close;
        let (query_start, query_end) = resolve_request_time_range(start_nanos, end_nanos);

        self.spawn_task(async move {
            seed_price_precision_if_needed(
                &historical_client,
                dataset.as_str(),
                instrument_id,
                query_start,
                query_end,
                price_precision,
            )
            .await;

            let params = RangeQueryParams {
                dataset,
                symbols,
                start: query_start,
                end: query_end,
                limit,
                price_precision,
            };

            let aggregation = match bar_type.spec().aggregation {
                BarAggregation::Second => BarAggregation::Second,
                BarAggregation::Minute => BarAggregation::Minute,
                BarAggregation::Hour => BarAggregation::Hour,
                BarAggregation::Day => BarAggregation::Day,
                _ => {
                    log::error!(
                        "Unsupported bar aggregation: {:?}",
                        bar_type.spec().aggregation
                    );
                    let response = DataResponse::Bars(BarsResponse::new(
                        request_id,
                        client_id,
                        bar_type,
                        Vec::new(),
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    send_data_response(&data_sender, response, "empty bars");
                    return;
                }
            };

            match historical_client
                .get_range_bars(params, aggregation, timestamp_on_close)
                .await
            {
                Ok(bars) => {
                    log::debug!("Retrieved {} bars", bars.len());
                    let response = DataResponse::Bars(BarsResponse::new(
                        request_id,
                        client_id,
                        bar_type,
                        bars,
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    if let Err(e) = data_sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send bars response: {e}");
                    }
                }
                Err(e) => {
                    log::error!("Failed to request bars: {e}");
                    let response = DataResponse::Bars(BarsResponse::new(
                        request_id,
                        client_id,
                        bar_type,
                        Vec::new(),
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    send_data_response(&data_sender, response, "empty bars");
                }
            }
        });

        Ok(())
    }

    fn request_book_depth(&self, request: RequestBookDepth) -> anyhow::Result<()> {
        log::debug!("Request book depth: {request:?}");

        let historical_client = self.historical.clone();
        let data_sender = self.data_sender.clone();
        let dataset = self.get_dataset_for_venue(request.instrument_id.venue)?;
        let instrument_id = request.instrument_id;
        let symbols = historical_client.prepare_symbols_from_instrument_ids(&[instrument_id]);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let limit = request.limit.map(|limit| limit.get() as u64);
        let depth = request.depth.map(|depth| depth.get());
        let request_params = request.params;
        let price_precision = price_precision_from_params(request_params.as_ref())?;
        let (query_start, query_end) = resolve_request_time_range(start_nanos, end_nanos);

        self.spawn_task(async move {
            seed_price_precision_if_needed(
                &historical_client,
                dataset.as_str(),
                instrument_id,
                query_start,
                query_end,
                price_precision,
            )
            .await;

            let params = RangeQueryParams {
                dataset,
                symbols,
                start: query_start,
                end: query_end,
                limit,
                price_precision,
            };

            match historical_client
                .get_range_order_book_depth10(params, depth)
                .await
            {
                Ok(depths) => {
                    log::debug!("Retrieved {} order book depths", depths.len());
                    let response = DataResponse::BookDepth(BookDepthResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        depths,
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    send_data_response(&data_sender, response, "book depth");
                }
                Err(e) => {
                    log::error!("Failed to request order book depths: {e}");
                    let response = DataResponse::BookDepth(BookDepthResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        Vec::new(),
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    send_data_response(&data_sender, response, "empty book depth");
                }
            }
        });

        Ok(())
    }

    fn request_book_deltas(&self, request: RequestBookDeltas) -> anyhow::Result<()> {
        log::debug!("Request book deltas: {request:?}");

        let historical_client = self.historical.clone();
        let data_sender = self.data_sender.clone();
        let dataset = self.get_dataset_for_venue(request.instrument_id.venue)?;
        let instrument_id = request.instrument_id;
        let symbols = historical_client.prepare_symbols_from_instrument_ids(&[instrument_id]);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let limit = request.limit.map(|limit| limit.get() as u64);
        let request_params = request.params;
        let price_precision = price_precision_from_params(request_params.as_ref())?;
        let (query_start, query_end) = resolve_request_time_range(start_nanos, end_nanos);

        self.spawn_task(async move {
            seed_price_precision_if_needed(
                &historical_client,
                dataset.as_str(),
                instrument_id,
                query_start,
                query_end,
                price_precision,
            )
            .await;

            let params = RangeQueryParams {
                dataset,
                symbols,
                start: query_start,
                end: query_end,
                limit,
                price_precision,
            };

            match historical_client.get_range_order_book_deltas(params).await {
                Ok(deltas) => {
                    log::debug!("Retrieved {} order book deltas", deltas.len());
                    let response = BookDeltasResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        deltas,
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    );

                    for response in partition_book_deltas_response(response) {
                        send_data_response(
                            &data_sender,
                            DataResponse::BookDeltas(response),
                            "book deltas",
                        );
                    }
                }
                Err(e) => {
                    log::error!("Failed to request order book deltas: {e}");
                    let response = DataResponse::BookDeltas(BookDeltasResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        Vec::new(),
                        start_nanos,
                        end_nanos,
                        get_atomic_clock_realtime().get_time_ns(),
                        request_params,
                    ));

                    send_data_response(&data_sender, response, "empty book deltas");
                }
            }
        });

        Ok(())
    }
}

fn instruments_query_params(
    dataset: String,
    start_nanos: UnixNanos,
    end_nanos: Option<UnixNanos>,
) -> RangeQueryParams {
    RangeQueryParams {
        dataset,
        symbols: vec!["ALL_SYMBOLS".to_string()],
        start: start_nanos,
        end: end_nanos,
        limit: None,
        price_precision: None,
    }
}

fn instrument_query_params(
    dataset: String,
    instrument_id: InstrumentId,
    start_nanos: UnixNanos,
    end_nanos: Option<UnixNanos>,
) -> RangeQueryParams {
    RangeQueryParams {
        dataset,
        symbols: vec![instrument_id_to_symbol_string(
            instrument_id,
            &mut AHashMap::new(),
        )],
        start: start_nanos,
        end: end_nanos,
        limit: None,
        price_precision: None,
    }
}

fn resolve_request_time_range(
    start_nanos: Option<UnixNanos>,
    end_nanos: Option<UnixNanos>,
) -> (UnixNanos, Option<UnixNanos>) {
    let mut end = end_nanos.unwrap_or_else(|| get_atomic_clock_realtime().get_time_ns());
    let mut start = start_nanos.unwrap_or_else(|| start_of_utc_day(end));

    if start > end {
        start = end;
    }

    if start == end {
        if end.is_zero() {
            end += DurationNanos::new(1);
        } else {
            start -= DurationNanos::new(1);
        }
    }

    (start, Some(end))
}

fn start_of_utc_day(timestamp: UnixNanos) -> UnixNanos {
    timestamp.floor(DurationNanos::from_days(1))
}

async fn seed_price_precision_if_needed(
    historical_client: &DatabentoHistoricalClient,
    dataset: &str,
    instrument_id: InstrumentId,
    start_nanos: UnixNanos,
    end_nanos: Option<UnixNanos>,
    price_precision: Option<u8>,
) {
    if price_precision.is_some()
        || historical_client
            .price_precision(instrument_id.symbol)
            .is_some()
    {
        return;
    }

    let query_params =
        instrument_query_params(dataset.to_string(), instrument_id, start_nanos, end_nanos);

    if let Err(e) = historical_client.get_range_instruments(query_params).await {
        log::warn!("Failed to seed price precision for {instrument_id}: {e}");
    }
}

fn send_data_response(
    data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
    response: DataResponse,
    label: &str,
) {
    if let Err(e) = data_sender.send(DataEvent::Response(response)) {
        log::error!("Failed to send {label} response: {e}");
    }
}

fn partition_book_deltas_response(mut response: BookDeltasResponse) -> Vec<BookDeltasResponse> {
    if response.data.is_empty() {
        return vec![response];
    }

    let mut partitions = IndexMap::new();

    for delta in std::mem::take(&mut response.data) {
        partitions
            .entry(delta.instrument_id)
            .or_insert_with(Vec::new)
            .push(delta);
    }

    partitions
        .into_iter()
        .map(|(instrument_id, data)| {
            let mut child = response.clone();
            child.instrument_id = instrument_id;
            child.data = data;
            child
        })
        .collect()
}

fn requested_instrument(
    instruments: Vec<InstrumentAny>,
    instrument_id: InstrumentId,
) -> Option<InstrumentAny> {
    instruments
        .into_iter()
        .rev()
        .find(|instrument| instrument.id() == instrument_id)
}

fn price_precision_from_params(params: Option<&Params>) -> anyhow::Result<Option<u8>> {
    let Some(price_precision) = params.and_then(|params| params.get_u64(PRICE_PRECISION_PARAM))
    else {
        return Ok(None);
    };

    Ok(Some(u8::try_from(price_precision).map_err(|_| {
        anyhow::anyhow!(
            "`{PRICE_PRECISION_PARAM}` must be less than or equal to {}",
            u8::MAX
        )
    })?))
}

fn schema_from_params(
    params: Option<&Params>,
    default_schema: dbn::Schema,
    allowed_schemas: &[dbn::Schema],
) -> anyhow::Result<dbn::Schema> {
    let schema = if let Some(schema) = params.and_then(|params| params.get_str(SCHEMA_PARAM)) {
        dbn::Schema::from_str(schema)?
    } else {
        default_schema
    };

    if allowed_schemas.contains(&schema) {
        return Ok(schema);
    }

    let allowed = allowed_schemas
        .iter()
        .map(dbn::Schema::as_str)
        .collect::<Vec<_>>()
        .join(", ");
    anyhow::bail!(
        "Invalid `{SCHEMA_PARAM}` '{}'. Must be one of: {allowed}",
        schema.as_str()
    );
}

fn send_subscription_commands(
    tx: &tokio::sync::mpsc::UnboundedSender<HandlerCommand>,
    dataset: &str,
    price_precision: Option<(Symbol, u8)>,
    subscription: Subscription,
    start_after_subscribe: bool,
) -> anyhow::Result<()> {
    if let Some((symbol, precision)) = price_precision {
        tx.send(HandlerCommand::SetPricePrecision(symbol, precision))
            .map_err(|e| anyhow::anyhow!("Failed to send command to dataset {dataset}: {e}"))?;
    }

    tx.send(HandlerCommand::Subscribe(subscription))
        .map_err(|e| anyhow::anyhow!("Failed to send command to dataset {dataset}: {e}"))?;

    if start_after_subscribe {
        tx.send(HandlerCommand::Start)
            .map_err(|e| anyhow::anyhow!("Failed to send command to dataset {dataset}: {e}"))?;
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use std::path::PathBuf;

    use nautilus_common::live::runner::replace_data_event_sender;
    use nautilus_core::UUID4;
    use nautilus_model::{
        data::OrderBookDelta,
        identifiers::{ClientId, InstrumentId},
        instruments::{CurrencyPair, InstrumentAny},
        types::{Currency, Price, Quantity},
    };
    use rstest::rstest;
    use serde_json::json;

    use super::*;

    #[derive(Clone, Copy)]
    enum SubscribeKind {
        Quotes,
        Trades,
    }

    fn currency_pair(instrument_id: &str) -> InstrumentAny {
        currency_pair_with_ts_init(instrument_id, UnixNanos::default())
    }

    fn test_data_client() -> DatabentoDataClient {
        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
        replace_data_event_sender(sender);

        let config = DatabentoDataClientConfig::new(
            "32-character-with-lots-of-filler",
            PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("publishers.json"),
            true,
            true,
        );
        DatabentoDataClient::new(
            ClientId::from("DATABENTO-TEST"),
            config,
            get_atomic_clock_realtime(),
        )
        .expect("test client should initialize")
    }

    #[rstest]
    #[tokio::test]
    async fn test_stop_closes_task_admission_until_disconnect_drains() {
        let mut client = test_data_client();

        client
            .task_handles
            .spawn(async { std::future::pending::<()>().await })
            .unwrap();
        client.is_connected.store(true, Ordering::Relaxed);

        client.stop().unwrap();

        assert!(!client.task_handles.is_open());
        assert!(client.is_disconnected());

        client.disconnect().await.unwrap();
        assert!(client.task_handles.is_empty());
    }

    #[rstest]
    #[case("EQUS", "EQUS.PLUS")] // overrides the apply_default EQUS -> EQUS.MINI mapping
    #[case("GLBX", "EQUS.MINI")] // overrides the apply_default GLBX -> GLBX.MDP3 mapping
    fn test_venue_dataset_map_overrides_default(#[case] venue: &str, #[case] dataset: &str) {
        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
        replace_data_event_sender(sender);

        let mut config = DatabentoDataClientConfig::new(
            "32-character-with-lots-of-filler",
            PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("publishers.json"),
            true,
            true,
        );
        config.venue_dataset_map = IndexMap::from([(venue.to_string(), dataset.to_string())]);

        let client = DatabentoDataClient::new(
            ClientId::from("DATABENTO-TEST"),
            config,
            get_atomic_clock_realtime(),
        )
        .expect("test client should initialize");

        assert_eq!(
            client.get_dataset_for_venue(Venue::from(venue)).unwrap(),
            dataset
        );

        // The override is targeted: an unrelated venue keeps its default.
        assert_eq!(
            client.get_dataset_for_venue(Venue::from("XCBO")).unwrap(),
            "OPRA.PILLAR"
        );
    }

    fn subscribe_quotes_cmd(params: Option<Params>) -> SubscribeQuotes {
        SubscribeQuotes::new(
            InstrumentId::from("ESM4.GLBX"),
            Some(ClientId::from("DATABENTO-TEST")),
            None,
            UUID4::new(),
            UnixNanos::default(),
            None,
            params,
        )
    }

    fn subscribe_trades_cmd(params: Option<Params>) -> SubscribeTrades {
        SubscribeTrades::new(
            InstrumentId::from("ESM4.GLBX"),
            Some(ClientId::from("DATABENTO-TEST")),
            None,
            UUID4::new(),
            UnixNanos::default(),
            None,
            params,
        )
    }

    fn currency_pair_with_ts_init(instrument_id: &str, ts_init: UnixNanos) -> InstrumentAny {
        let instrument_id = InstrumentId::from(instrument_id);
        InstrumentAny::CurrencyPair(
            CurrencyPair::builder()
                .instrument_id(instrument_id)
                .raw_symbol(instrument_id.symbol)
                .base_currency(Currency::from("BTC"))
                .quote_currency(Currency::from("USDT"))
                .price_precision(2)
                .size_precision(6)
                .price_increment(Price::from("0.01"))
                .size_increment(Quantity::from("0.000001"))
                .ts_event(UnixNanos::default())
                .ts_init(ts_init)
                .build()
                .unwrap(),
        )
    }

    #[rstest]
    fn test_instruments_query_params_requests_all_symbols() {
        let start = UnixNanos::from(1_000_000_000);
        let end = UnixNanos::from(2_000_000_000);

        let params = instruments_query_params("GLBX.MDP3".to_string(), start, Some(end));

        assert_eq!(params.dataset, "GLBX.MDP3");
        assert_eq!(params.symbols, vec!["ALL_SYMBOLS"]);
        assert_eq!(params.start, start);
        assert_eq!(params.end, Some(end));
        assert_eq!(params.limit, None);
        assert_eq!(params.price_precision, None);
    }

    #[rstest]
    fn test_instrument_query_params_requests_single_symbol() {
        let instrument_id = InstrumentId::from("ESM4.GLBX");

        let start = UnixNanos::from(1_000_000_000);
        let end = UnixNanos::from(2_000_000_000);

        let params =
            instrument_query_params("GLBX.MDP3".to_string(), instrument_id, start, Some(end));

        assert_eq!(params.dataset, "GLBX.MDP3");
        assert_eq!(params.symbols, vec!["ESM4"]);
        assert_eq!(params.start, start);
        assert_eq!(params.end, Some(end));
        assert_eq!(params.limit, None);
        assert_eq!(params.price_precision, None);
    }

    #[rstest]
    fn test_resolve_request_time_range_defaults_to_end_day() {
        let end = UnixNanos::from(1_706_443_200_000_000_001);

        let (start, resolved_end) = resolve_request_time_range(None, Some(end));

        assert_eq!(start, UnixNanos::from(1_706_400_000_000_000_000));
        assert_eq!(resolved_end, Some(end));
    }

    #[rstest]
    fn test_resolve_request_time_range_makes_empty_interval_non_empty() {
        let end = UnixNanos::from(1_706_443_200_000_000_001);

        let (start, resolved_end) = resolve_request_time_range(Some(end), Some(end));

        assert_eq!(start, end - DurationNanos::new(1));
        assert_eq!(resolved_end, Some(end));
    }

    #[rstest]
    fn test_requested_instrument_filters_exact_id() {
        let requested_id = InstrumentId::from("BTCUSDT.BINANCE");
        let instruments = vec![
            currency_pair("ETHUSDT.BINANCE"),
            currency_pair("BTCUSDT.BINANCE"),
        ];

        let instrument = requested_instrument(instruments, requested_id).expect("instrument");

        assert_eq!(instrument.id(), requested_id);
    }

    #[rstest]
    fn test_requested_instrument_returns_latest_matching_id() {
        let requested_id = InstrumentId::from("BTCUSDT.BINANCE");
        let instruments = vec![
            currency_pair_with_ts_init("BTCUSDT.BINANCE", UnixNanos::from(1)),
            currency_pair_with_ts_init("BTCUSDT.BINANCE", UnixNanos::from(2)),
        ];

        let instrument = requested_instrument(instruments, requested_id).expect("instrument");

        assert_eq!(instrument.ts_init(), UnixNanos::from(2));
    }

    #[rstest]
    fn test_requested_instrument_returns_none_on_miss() {
        let instruments = vec![currency_pair("ETHUSDT.BINANCE")];

        let instrument = requested_instrument(instruments, InstrumentId::from("BTCUSDT.BINANCE"));

        assert!(instrument.is_none());
    }

    #[rstest]
    fn test_price_precision_from_params() {
        let mut params = Params::new();
        params.insert(PRICE_PRECISION_PARAM.to_string(), json!(5));

        let price_precision = price_precision_from_params(Some(&params)).unwrap();

        assert_eq!(price_precision, Some(5));
    }

    #[rstest]
    fn test_price_precision_from_params_rejects_out_of_range_value() {
        let mut params = Params::new();
        params.insert(
            PRICE_PRECISION_PARAM.to_string(),
            json!(u64::from(u8::MAX) + 1),
        );

        let result = price_precision_from_params(Some(&params));

        assert!(result.is_err());
    }

    #[rstest]
    fn test_partition_book_deltas_response_by_child_instrument() {
        let correlation_id = UUID4::new();
        let client_id = ClientId::from("DATABENTO-TEST");
        let parent = InstrumentId::from("ES.FUT.GLBX");
        let child_a = InstrumentId::from("ESM6.GLBX");
        let child_b = InstrumentId::from("ESU6.GLBX");
        let deltas = vec![
            OrderBookDelta::clear(child_a, 1, UnixNanos::from(1_000), UnixNanos::from(1_000)),
            OrderBookDelta::clear(child_b, 2, UnixNanos::from(2_000), UnixNanos::from(2_000)),
            OrderBookDelta::clear(child_a, 3, UnixNanos::from(3_000), UnixNanos::from(3_000)),
        ];
        let mut params = Params::new();
        params.insert(PRICE_PRECISION_PARAM.to_string(), json!(5));
        let response = BookDeltasResponse::new(
            correlation_id,
            client_id,
            parent,
            deltas.clone(),
            Some(UnixNanos::from(500)),
            Some(UnixNanos::from(4_000)),
            UnixNanos::from(5_000),
            Some(params.clone()),
        );

        let responses = partition_book_deltas_response(response);

        assert_eq!(responses.len(), 2);
        assert_eq!(responses[0].correlation_id, correlation_id);
        assert_eq!(responses[1].correlation_id, correlation_id);
        assert_eq!(responses[0].client_id, client_id);
        assert_eq!(responses[1].client_id, client_id);
        assert_eq!(responses[0].instrument_id, child_a);
        assert_eq!(responses[1].instrument_id, child_b);
        assert_eq!(responses[0].data, vec![deltas[0], deltas[2]]);
        assert_eq!(responses[1].data, vec![deltas[1]]);
        assert_eq!(responses[0].start, Some(UnixNanos::from(500)));
        assert_eq!(responses[1].start, Some(UnixNanos::from(500)));
        assert_eq!(responses[0].end, Some(UnixNanos::from(4_000)));
        assert_eq!(responses[1].end, Some(UnixNanos::from(4_000)));
        assert_eq!(responses[0].ts_init, UnixNanos::from(5_000));
        assert_eq!(responses[1].ts_init, UnixNanos::from(5_000));
        assert_eq!(responses[0].params, Some(params.clone()));
        assert_eq!(responses[1].params, Some(params));
    }

    #[rstest]
    fn test_partition_book_deltas_response_preserves_homogeneous_response() {
        let correlation_id = UUID4::new();
        let client_id = ClientId::from("DATABENTO-TEST");
        let instrument_id = InstrumentId::from("ESM6.GLBX");
        let delta = OrderBookDelta::clear(
            instrument_id,
            1,
            UnixNanos::from(1_000),
            UnixNanos::from(1_000),
        );
        let response = BookDeltasResponse::new(
            correlation_id,
            client_id,
            instrument_id,
            vec![delta],
            Some(UnixNanos::from(500)),
            Some(UnixNanos::from(1_500)),
            UnixNanos::from(2_000),
            None,
        );

        let responses = partition_book_deltas_response(response);

        assert_eq!(responses.len(), 1);
        assert_eq!(responses[0].correlation_id, correlation_id);
        assert_eq!(responses[0].client_id, client_id);
        assert_eq!(responses[0].instrument_id, instrument_id);
        assert_eq!(responses[0].data, vec![delta]);
        assert_eq!(responses[0].start, Some(UnixNanos::from(500)));
        assert_eq!(responses[0].end, Some(UnixNanos::from(1_500)));
        assert_eq!(responses[0].ts_init, UnixNanos::from(2_000));
        assert_eq!(responses[0].params, None);
    }

    #[rstest]
    fn test_partition_book_deltas_response_preserves_empty_parent_response() {
        let correlation_id = UUID4::new();
        let parent = InstrumentId::from("ES.FUT.GLBX");
        let response = BookDeltasResponse::new(
            correlation_id,
            ClientId::from("DATABENTO-TEST"),
            parent,
            Vec::new(),
            None,
            None,
            UnixNanos::from(1_000),
            None,
        );

        let responses = partition_book_deltas_response(response);

        assert_eq!(responses.len(), 1);
        let response = &responses[0];
        assert_eq!(response.correlation_id, correlation_id);
        assert_eq!(response.instrument_id, parent);
        assert!(response.data.is_empty());
    }

    #[rstest]
    fn test_schema_from_params_returns_default() {
        let schema = schema_from_params(None, dbn::Schema::Mbp1, QUOTE_SCHEMAS).unwrap();

        assert_eq!(schema, dbn::Schema::Mbp1);
    }

    #[rstest]
    fn test_schema_from_params_accepts_allowed_value() {
        let mut params = Params::new();
        params.insert(SCHEMA_PARAM.to_string(), json!("tbbo"));

        let schema = schema_from_params(Some(&params), dbn::Schema::Mbp1, QUOTE_SCHEMAS).unwrap();

        assert_eq!(schema, dbn::Schema::Tbbo);
    }

    #[rstest]
    fn test_schema_from_params_rejects_disallowed_value() {
        let mut params = Params::new();
        params.insert(SCHEMA_PARAM.to_string(), json!("mbo"));

        let result = schema_from_params(Some(&params), dbn::Schema::Mbp1, QUOTE_SCHEMAS);

        assert!(result.is_err());
    }

    #[rstest]
    #[case::quotes(SubscribeKind::Quotes)]
    #[case::trades(SubscribeKind::Trades)]
    fn test_invalid_subscribe_params_do_not_create_feed_handler(#[case] kind: SubscribeKind) {
        let mut client = test_data_client();
        let mut params = Params::new();
        params.insert(SCHEMA_PARAM.to_string(), json!("definition"));

        let result = match kind {
            SubscribeKind::Quotes => client.subscribe_quotes(subscribe_quotes_cmd(Some(params))),
            SubscribeKind::Trades => client.subscribe_trades(subscribe_trades_cmd(Some(params))),
        };

        assert!(result.is_err());
        assert!(client.cmd_channels.lock().is_empty());
    }

    #[rstest]
    fn test_send_subscription_commands_starts_after_subscribe() {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let subscription = Subscription::builder()
            .schema(dbn::Schema::Mbp1)
            .symbols(vec!["ESM4"])
            .build();

        send_subscription_commands(
            &tx,
            "GLBX.MDP3",
            Some((Symbol::from("ESM4"), 2)),
            subscription,
            true,
        )
        .unwrap();

        assert!(matches!(
            rx.try_recv().unwrap(),
            HandlerCommand::SetPricePrecision(symbol, 2) if symbol == Symbol::from("ESM4")
        ));
        assert!(matches!(
            rx.try_recv().unwrap(),
            HandlerCommand::Subscribe(sub) if sub.schema == dbn::Schema::Mbp1
        ));
        assert!(matches!(rx.try_recv().unwrap(), HandlerCommand::Start));
    }

    #[rstest]
    fn test_send_subscription_commands_without_precision_or_start() {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let subscription = Subscription::builder()
            .schema(dbn::Schema::Mbp1)
            .symbols(vec!["ESM4"])
            .build();

        send_subscription_commands(&tx, "GLBX.MDP3", None, subscription, false).unwrap();

        assert!(matches!(
            rx.try_recv().unwrap(),
            HandlerCommand::Subscribe(sub) if sub.schema == dbn::Schema::Mbp1
        ));
        assert!(matches!(
            rx.try_recv(),
            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
        ));
    }
}