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, sender::EventSender},
    messages::{
        DataEvent, DataResponse,
        data::{
            BarsResponse, BookDeltasResponse, BookDepthResponse, InstrumentResponse,
            InstrumentsResponse, QuotesResponse, RequestBars, RequestBookDeltas, RequestBookDepth,
            RequestInstrument, RequestInstruments, RequestQuotes, RequestTrades, SubscribeBars,
            SubscribeBookDeltas, SubscribeBookDepth, SubscribeCustomData, SubscribeInstrument,
            SubscribeInstrumentStatus, SubscribeQuotes, SubscribeTrades, TradesResponse,
            UnsubscribeBars, UnsubscribeBookDeltas, UnsubscribeBookDepth, UnsubscribeCustomData,
            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::{BarType, CustomData, Data, DataType, custom::CustomDataTrait},
    enums::{BarAggregation, PriceType},
    identifiers::{ClientId, InstrumentId, Symbol, Venue},
    instruments::{Instrument, InstrumentAny},
};
use parking_lot::Mutex;
use tokio_util::sync::CancellationToken;

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

const PRICE_PRECISION_PARAM: &str = "price_precision";
const SCHEMA_PARAM: &str = "schema";
const MBP10_DEPTH: usize = 10;
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,
    /// Override for the Databento Historical API base URL.
    pub historical_base_url: Option<String>,
    /// Override for the Databento live gateway address.
    pub live_gateway_addr: Option<String>,
    /// 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,
    historical_base_url: Option<String>,
    live_gateway_addr: Option<String>,
    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
            historical_base_url: None,
            live_gateway_addr: None,
        }
    }

    /// 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: EventSender<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 = match &config.historical_base_url {
            Some(base_url) => DatabentoHistoricalClient::new_with_base_url(
                config.credential.clone(),
                config.publishers_filepath.clone(),
                clock,
                config.use_exchange_as_venue,
                base_url,
            )?,
            None => 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 subscribe_schema(
        &self,
        instrument_id: InstrumentId,
        schema: dbn::Schema,
    ) -> anyhow::Result<()> {
        let symbol = instrument_id.symbol.to_string();
        // Records resolve to the underlying contract, so a continuous or parent symbol would
        // publish on topics that its subscriber does not listen on
        anyhow::ensure!(
            infer_symbology_type(&symbol) == dbn::SType::RawSymbol,
            "Unsupported symbol {symbol} for a Databento {schema} subscription, only raw symbols \
             are available (not continuous or parent symbols)",
        );

        let dataset = self.get_dataset_for_venue(instrument_id.venue)?;
        let start_after_subscribe = self.get_or_create_feed_handler(&dataset);

        self.symbol_venue_map
            .insert(instrument_id.symbol, instrument_id.venue);

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

        self.send_subscription_to_dataset(&dataset, None, 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,
        );

        if let Some(addr) = &self.config.live_gateway_addr {
            feed_handler = feed_handler.with_gateway_addr(addr.clone());
        }

        let feed_future = async move {
            // run() logs its terminal error and forwards it as `DatabentoMessage::Error`
            let _ = feed_handler.run().await;
            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 instrument_id = imbalance.instrument_id;
                                let data = Data::Custom(custom_data_for_instrument(Arc::new(imbalance), instrument_id));
                                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 instrument_id = statistics.instrument_id;
                                let data = Data::Custom(custom_data_for_instrument(Arc::new(statistics), instrument_id));
                                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 `DatabentoStatistics` or `DatabentoImbalance` for the instrument given by the
    /// identifier of the data type.
    ///
    /// # Errors
    ///
    /// Returns an error if the data type is not supported, has metadata, has a missing or invalid
    /// instrument ID identifier, or the subscription request fails.
    fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
        let schema = custom_data_schema(&cmd.data_type)?;
        let instrument_id = custom_data_instrument_id(&cmd.data_type)?;
        self.subscribe_schema(instrument_id, schema)
    }

    /// Subscribes to 10-level order book depth snapshots for the specified instrument.
    ///
    /// # Errors
    ///
    /// Returns an error if the requested depth is not 10, the symbol is not a raw symbol, or the
    /// subscription request fails.
    fn subscribe_book_depth(&mut self, cmd: SubscribeBookDepth) -> anyhow::Result<()> {
        if let Some(depth) = cmd.depth {
            anyhow::ensure!(
                depth.get() == MBP10_DEPTH,
                "Unsupported book depth {depth} for Databento `mbp-10`, only a depth of {MBP10_DEPTH} is available",
            );
        }

        self.subscribe_schema(cmd.instrument_id, dbn::Schema::Mbp10)
    }

    /// Subscribes to OHLCV bars for the specified bar type.
    ///
    /// # Errors
    ///
    /// Returns an error if the bar type has no matching OHLCV schema or the subscription request
    /// fails.
    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
        let schema = ohlcv_schema(cmd.bar_type)?;
        self.subscribe_schema(cmd.bar_type.instrument_id(), schema)
    }

    /// 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::debug!(
            "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::debug!(
            "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::debug!(
            "Databento does not support granular unsubscribing - ignoring unsubscribe request for {}",
            cmd.instrument_id
        );

        Ok(())
    }

    fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> 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::debug!(
            "Databento does not support granular unsubscribing - ignoring unsubscribe request for {}",
            cmd.data_type
        );

        Ok(())
    }

    fn unsubscribe_book_depth(&mut self, cmd: &UnsubscribeBookDepth) -> 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::debug!(
            "Databento does not support granular unsubscribing - ignoring unsubscribe request for {}",
            cmd.instrument_id
        );

        Ok(())
    }

    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> 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::debug!(
            "Databento does not support granular unsubscribing - ignoring unsubscribe request for {}",
            cmd.bar_type
        );

        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::debug!(
            "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_depth(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_start = start_of_utc_day(start_nanos);
    let query_params =
        instrument_query_params(dataset.to_string(), instrument_id, query_start, 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: &EventSender<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 custom_data_for_instrument(
    data: Arc<dyn CustomDataTrait>,
    instrument_id: InstrumentId,
) -> CustomData {
    let data_type = DataType::new(data.type_name(), None, Some(instrument_id.to_string()));
    CustomData::new(data, data_type)
}

fn custom_data_schema(data_type: &DataType) -> anyhow::Result<dbn::Schema> {
    match data_type.type_name() {
        name if name == DatabentoStatistics::type_name_static() => Ok(dbn::Schema::Statistics),
        name if name == DatabentoImbalance::type_name_static() => Ok(dbn::Schema::Imbalance),
        name => anyhow::bail!(
            "Unsupported custom data type for Databento live subscription: {name}, expected {} or {}",
            DatabentoStatistics::type_name_static(),
            DatabentoImbalance::type_name_static(),
        ),
    }
}

fn custom_data_instrument_id(data_type: &DataType) -> anyhow::Result<InstrumentId> {
    // Records are published without metadata, so metadata would change the subscribed topic
    anyhow::ensure!(
        data_type
            .metadata()
            .is_none_or(|metadata| metadata.is_empty()),
        "{} subscriptions do not support metadata, received {}",
        data_type.type_name(),
        data_type.metadata_str(),
    );

    let identifier = data_type.identifier().ok_or_else(|| {
        anyhow::anyhow!(
            "{} subscriptions require an instrument ID identifier",
            data_type.type_name()
        )
    })?;

    InstrumentId::from_str(identifier)
        .map_err(|e| anyhow::anyhow!("Invalid instrument ID identifier `{identifier}`: {e}"))
}

fn ohlcv_schema(bar_type: BarType) -> anyhow::Result<dbn::Schema> {
    let spec = bar_type.spec();
    anyhow::ensure!(
        spec.price_type == PriceType::Last,
        "Unsupported bar price type for Databento OHLCV: {spec}, only LAST is emitted",
    );
    anyhow::ensure!(
        spec.step.get() == 1,
        "Unsupported bar step for Databento OHLCV: {spec}, only a step of 1 is available",
    );

    ohlcv_schema_from_aggregation(spec.aggregation)
}

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::{num::NonZeroUsize, path::PathBuf};

    use databento::dbn::{
        MappingInterval, SymbolMapping,
        decode::{DbnMetadata, DecodeRecord, dbn::Decoder},
        encode::EncodeRecord,
    };
    use nautilus_common::{
        live::runner::replace_data_event_sender,
        msgbus::{self, switchboard::get_custom_topic},
    };
    use nautilus_core::UUID4;
    use nautilus_model::{
        data::OrderBookDelta,
        enums::BookType,
        identifiers::{ClientId, InstrumentId},
        instruments::{CurrencyPair, InstrumentAny},
        types::{Currency, Price, Quantity},
    };
    use rstest::rstest;
    use serde_json::json;
    use time::macros::date;
    use tokio::{
        io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
        net::TcpListener,
    };

    use super::*;
    use crate::enums::{DatabentoStatisticType, DatabentoStatisticUpdateAction};

    #[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"
        );
    }

    #[rstest]
    fn test_config_base_url_overrides_default_to_none() {
        let config = DatabentoDataClientConfig::new(
            "32-character-with-lots-of-filler",
            PathBuf::from("test_publishers.json"),
            true,
            true,
        );

        assert!(config.historical_base_url.is_none());
        assert!(config.live_gateway_addr.is_none());
    }

    #[rstest]
    fn test_historical_base_url_override_invalid_errors() {
        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.historical_base_url = Some("not a url".to_string());

        let result = DatabentoDataClient::new(
            ClientId::from("DATABENTO-TEST"),
            config,
            get_atomic_clock_realtime(),
        );

        assert!(result.is_err());
    }

    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 subscribe_book_depth_cmd(depth: Option<NonZeroUsize>) -> SubscribeBookDepth {
        SubscribeBookDepth::new(
            InstrumentId::from("ESM4.GLBX"),
            BookType::L2_MBP,
            Some(ClientId::from("DATABENTO-TEST")),
            None,
            UUID4::new(),
            UnixNanos::default(),
            depth,
            true,
            None,
            None,
        )
    }

    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]
    #[case::intraday(1_782_302_400_000_000_001)]
    #[case::midnight(1_782_259_200_000_000_000)]
    #[case::day_end(1_782_345_599_999_999_999)]
    #[tokio::test]
    async fn test_seed_price_precision_includes_midnight(#[case] start: u64) {
        let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("test_data/test_data.definition.futures_contract.dbn.zst");
        let mut decoder = Decoder::from_zstd_file(path).unwrap();
        let mut metadata = decoder.metadata().clone();
        let record = decoder
            .decode_record::<dbn::InstrumentDefMsg>()
            .unwrap()
            .unwrap();
        metadata.symbols = vec!["ESU6".to_string()];

        metadata.mappings = vec![SymbolMapping {
            raw_symbol: "ESU6".to_string(),
            intervals: vec![MappingInterval {
                start_date: date!(2026 - 06 - 24),
                end_date: date!(2026 - 06 - 25),
                symbol: record.hd.instrument_id.to_string(),
            }],
        }];

        let mut body = Vec::new();
        {
            let mut encoder = dbn::encode::dbn::Encoder::with_zstd(&mut body, &metadata).unwrap();
            encoder.encode_record(record).unwrap();
            encoder.flush().unwrap();
        }

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base_url = format!("http://{}/", listener.local_addr().unwrap());

        let server = tokio::spawn(serve_historical_response(listener, body));

        let client = DatabentoHistoricalClient::new_with_base_url(
            Credential::new("32-character-with-lots-of-filler"),
            PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("publishers.json"),
            get_atomic_clock_realtime(),
            false,
            &base_url,
        )
        .unwrap();
        let instrument_id = InstrumentId::from("ESU6.GLBX");
        let end = UnixNanos::from(1_782_345_600_000_000_123);

        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            seed_price_precision_if_needed(
                &client,
                "GLBX.MDP3",
                instrument_id,
                start.into(),
                Some(end),
                None,
            ),
        )
        .await
        .unwrap();
        let request = server.await.unwrap();
        let fields: AHashMap<_, _> = request
            .split('&')
            .map(|field| field.split_once('=').unwrap())
            .collect();

        assert_eq!(fields["dataset"], "GLBX.MDP3");
        assert_eq!(fields["schema"], "definition");
        assert_eq!(fields["symbols"], "ESU6");
        assert_eq!(fields["start"], "1782259200000000000");
        assert_eq!(fields["end"], end.to_string());
        assert_eq!(client.price_precision(instrument_id.symbol), Some(2));
    }

    async fn serve_historical_response(listener: TcpListener, body: Vec<u8>) -> String {
        let (stream, _) = listener.accept().await.unwrap();
        let mut stream = BufReader::new(stream);
        let mut line = String::new();
        stream.read_line(&mut line).await.unwrap();
        assert_eq!(line, "POST /v0/timeseries.get_range HTTP/1.1\r\n");
        let mut content_length = None;

        loop {
            line.clear();
            stream.read_line(&mut line).await.unwrap();

            if line == "\r\n" {
                break;
            }

            if let Some((name, value)) = line.split_once(':')
                && name.eq_ignore_ascii_case("content-length")
            {
                content_length = Some(value.trim().parse::<usize>().unwrap());
            }
        }

        let mut request = vec![0; content_length.unwrap()];
        stream.read_exact(&mut request).await.unwrap();
        let headers = format!(
            "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
            body.len(),
        );
        stream.write_all(headers.as_bytes()).await.unwrap();
        stream.write_all(&body).await.unwrap();
        stream.shutdown().await.unwrap();
        String::from_utf8(request).unwrap()
    }

    #[rstest]
    #[case::explicit(Some(5), None)]
    #[case::cached(None, Some(3))]
    #[tokio::test]
    async fn test_seed_price_precision_skips_known_precision(
        #[case] price_precision: Option<u8>,
        #[case] cached_precision: Option<u8>,
    ) {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base_url = format!("http://{}/", listener.local_addr().unwrap());

        let client = DatabentoHistoricalClient::new_with_base_url(
            Credential::new("32-character-with-lots-of-filler"),
            PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("publishers.json"),
            get_atomic_clock_realtime(),
            false,
            &base_url,
        )
        .unwrap();
        let instrument_id = InstrumentId::from("ESU6.GLBX");
        if let Some(precision) = cached_precision {
            client.set_price_precision(instrument_id.symbol, precision);
        }

        tokio::select! {
            () = seed_price_precision_if_needed(
                &client,
                "GLBX.MDP3",
                instrument_id,
                1_706_443_200_000_000_001.into(),
                Some(1_706_486_400_000_000_123.into()),
                price_precision,
            ) => {}
            result = listener.accept() => panic!("Unexpected definition request: {result:?}"),
            () = tokio::time::sleep(std::time::Duration::from_secs(5)) => {
                panic!("Precision seeding did not finish");
            }
        }

        assert_eq!(
            client.price_precision(instrument_id.symbol),
            cached_precision
        );
    }

    #[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]
    #[case::second("ESM4.GLBX-1-SECOND-LAST-EXTERNAL", dbn::Schema::Ohlcv1S)]
    #[case::minute("ESM4.GLBX-1-MINUTE-LAST-EXTERNAL", dbn::Schema::Ohlcv1M)]
    #[case::hour("ESM4.GLBX-1-HOUR-LAST-EXTERNAL", dbn::Schema::Ohlcv1H)]
    #[case::day("ESM4.GLBX-1-DAY-LAST-EXTERNAL", dbn::Schema::Ohlcv1D)]
    fn test_ohlcv_schema_maps_aggregation(#[case] bar_type: &str, #[case] expected: dbn::Schema) {
        let schema = ohlcv_schema(BarType::from(bar_type)).unwrap();

        assert_eq!(schema, expected);
    }

    #[rstest]
    #[case::step("ESM4.GLBX-5-MINUTE-LAST-EXTERNAL")]
    #[case::aggregation("ESM4.GLBX-100-TICK-LAST-EXTERNAL")]
    #[case::bid("ESM4.GLBX-1-MINUTE-BID-EXTERNAL")]
    #[case::mid("ESM4.GLBX-1-MINUTE-MID-EXTERNAL")]
    fn test_ohlcv_schema_rejects_unsupported_bar_type(#[case] bar_type: &str) {
        assert!(ohlcv_schema(BarType::from(bar_type)).is_err());
    }

    #[rstest]
    #[case::statistics("DatabentoStatistics", dbn::Schema::Statistics)]
    #[case::imbalance("DatabentoImbalance", dbn::Schema::Imbalance)]
    fn test_subscribe_custom_data_sends_schema_subscription(
        #[case] type_name: &str,
        #[case] expected: dbn::Schema,
    ) {
        let mut client = test_data_client();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        client
            .cmd_channels
            .lock()
            .insert("GLBX.MDP3".to_string(), tx);
        let command = SubscribeCustomData::new(
            Some(ClientId::from("DATABENTO-TEST")),
            None,
            DataType::new(type_name, None, Some("ESM4.GLBX".to_string())),
            UUID4::new(),
            UnixNanos::default(),
            None,
            None,
        );

        client.subscribe(command).unwrap();

        let HandlerCommand::Subscribe(subscription) = rx.try_recv().unwrap() else {
            panic!("expected subscription");
        };
        assert_eq!(subscription.schema, expected);
        assert_eq!(
            subscription.symbols,
            databento::Symbols::Symbols(vec!["ESM4".to_string()])
        );
        assert!(rx.try_recv().is_err());
    }

    #[rstest]
    #[case::unsupported_type("DatabentoOther", Some("ESM4.GLBX"))]
    #[case::missing_identifier("DatabentoStatistics", None)]
    #[case::invalid_identifier("DatabentoStatistics", Some("ESM4"))]
    fn test_subscribe_custom_data_rejects_invalid_data_type(
        #[case] type_name: &str,
        #[case] identifier: Option<&str>,
    ) {
        let mut client = test_data_client();
        let command = SubscribeCustomData::new(
            Some(ClientId::from("DATABENTO-TEST")),
            None,
            DataType::new(type_name, None, identifier.map(str::to_string)),
            UUID4::new(),
            UnixNanos::default(),
            None,
            None,
        );

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

    #[rstest]
    fn test_custom_data_for_instrument_matches_subscribed_topic() {
        let instrument_id = InstrumentId::from("ESM4.GLBX");
        let statistics = DatabentoStatistics::new(
            instrument_id,
            DatabentoStatisticType::SettlementPrice,
            DatabentoStatisticUpdateAction::Added,
            Some(Price::from("5000.25")),
            None,
            1,
            2,
            3,
            UnixNanos::from(4),
            5,
            UnixNanos::from(6),
            UnixNanos::from(7),
            UnixNanos::from(8),
        );
        let subscribed = DataType::new("DatabentoStatistics", None, Some("ESM4.GLBX".to_string()));

        let custom = custom_data_for_instrument(Arc::new(statistics), instrument_id);

        assert_eq!(custom.data_type, subscribed);
        assert_eq!(
            get_custom_topic(&custom.data_type),
            get_custom_topic(&subscribed)
        );
        assert_eq!(custom.data_type.identifier(), Some("ESM4.GLBX"));
    }

    #[rstest]
    fn test_subscribe_schema_sends_raw_symbol() {
        let client = test_data_client();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        client
            .cmd_channels
            .lock()
            .insert("GLBX.MDP3".to_string(), tx);

        client
            .subscribe_schema(InstrumentId::from("ESM4.GLBX"), dbn::Schema::Ohlcv1M)
            .unwrap();

        let HandlerCommand::Subscribe(subscription) = rx.try_recv().unwrap() else {
            panic!("expected subscription");
        };
        assert_eq!(subscription.stype_in, dbn::SType::RawSymbol);
        assert_eq!(
            subscription.symbols,
            databento::Symbols::Symbols(vec!["ESM4".to_string()])
        );
    }

    #[rstest]
    #[case::continuous("ES.c.0.GLBX")]
    #[case::parent("ES.FUT.GLBX")]
    #[case::instrument_id("12345.GLBX")]
    fn test_subscribe_schema_rejects_non_raw_symbols(#[case] instrument_id: &str) {
        let client = test_data_client();

        let result = client.subscribe_schema(InstrumentId::from(instrument_id), dbn::Schema::Mbp10);

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

    #[rstest]
    fn test_subscribe_custom_data_rejects_metadata() {
        let mut client = test_data_client();
        let mut metadata = Params::new();
        metadata.insert("venue".to_string(), json!("GLBX"));
        let command = SubscribeCustomData::new(
            Some(ClientId::from("DATABENTO-TEST")),
            None,
            DataType::new(
                "DatabentoStatistics",
                Some(metadata),
                Some("ESM4.GLBX".to_string()),
            ),
            UUID4::new(),
            UnixNanos::default(),
            None,
            None,
        );

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

    #[rstest]
    fn test_subscribed_custom_data_reaches_subscriber() {
        let instrument_id = InstrumentId::from("ESM4.GLBX");
        let subscribed = DataType::new("DatabentoStatistics", None, Some("ESM4.GLBX".to_string()));
        let other = DataType::new("DatabentoStatistics", None, Some("ESU4.GLBX".to_string()));
        let handler = msgbus::stubs::get_message_saving_handler::<CustomData>(None);
        msgbus::subscribe_any(get_custom_topic(&subscribed).into(), handler.clone(), None);
        let statistics = DatabentoStatistics::new(
            instrument_id,
            DatabentoStatisticType::SettlementPrice,
            DatabentoStatisticUpdateAction::Added,
            Some(Price::from("5000.25")),
            None,
            1,
            2,
            3,
            UnixNanos::from(4),
            5,
            UnixNanos::from(6),
            UnixNanos::from(7),
            UnixNanos::from(8),
        );
        let custom = custom_data_for_instrument(Arc::new(statistics), instrument_id);

        msgbus::publish_any(get_custom_topic(&custom.data_type), &custom);
        msgbus::publish_any(get_custom_topic(&other), &custom);

        let received = msgbus::stubs::get_saved_messages::<CustomData>(&handler);
        assert_eq!(received, vec![custom]);
    }

    #[rstest]
    fn test_subscribe_book_depth_accepts_depth_ten() {
        let mut client = test_data_client();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        client
            .cmd_channels
            .lock()
            .insert("GLBX.MDP3".to_string(), tx);

        client
            .subscribe_book_depth(subscribe_book_depth_cmd(NonZeroUsize::new(10)))
            .unwrap();

        let HandlerCommand::Subscribe(subscription) = rx.try_recv().unwrap() else {
            panic!("expected subscription");
        };
        assert_eq!(subscription.schema, dbn::Schema::Mbp10);
    }

    #[rstest]
    #[case::five(5)]
    #[case::twenty(20)]
    fn test_subscribe_book_depth_rejects_other_depths(#[case] depth: usize) {
        let mut client = test_data_client();

        let result =
            client.subscribe_book_depth(subscribe_book_depth_cmd(NonZeroUsize::new(depth)));

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

    #[rstest]
    fn test_subscribe_bars_sends_ohlcv_subscription() {
        let mut client = test_data_client();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        client
            .cmd_channels
            .lock()
            .insert("GLBX.MDP3".to_string(), tx);
        let command = SubscribeBars::new(
            BarType::from("ESM4.GLBX-1-MINUTE-LAST-EXTERNAL"),
            Some(ClientId::from("DATABENTO-TEST")),
            None,
            UUID4::new(),
            UnixNanos::default(),
            None,
            None,
        );

        client.subscribe_bars(command).unwrap();

        let HandlerCommand::Subscribe(subscription) = rx.try_recv().unwrap() else {
            panic!("expected subscription");
        };
        assert_eq!(subscription.schema, dbn::Schema::Ohlcv1M);
        assert_eq!(
            subscription.symbols,
            databento::Symbols::Symbols(vec!["ESM4".to_string()])
        );
        assert!(rx.try_recv().is_err());
    }

    #[rstest]
    fn test_subscribe_book_depth_sends_mbp10_subscription() {
        let mut client = test_data_client();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        client
            .cmd_channels
            .lock()
            .insert("GLBX.MDP3".to_string(), tx);
        client
            .subscribe_book_depth(subscribe_book_depth_cmd(None))
            .unwrap();

        let HandlerCommand::Subscribe(subscription) = rx.try_recv().unwrap() else {
            panic!("expected subscription");
        };
        assert_eq!(subscription.schema, dbn::Schema::Mbp10);
        assert_eq!(
            subscription.symbols,
            databento::Symbols::Symbols(vec!["ESM4".to_string()])
        );
        assert!(rx.try_recv().is_err());
    }

    #[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)
        ));
    }
}