nautilus_data/engine/

mod.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 high-performance `DataEngine` for all environments.
//!
//! The `DataEngine` is the central component of the entire data stack.
//! The data engines primary responsibility is to orchestrate interactions between
//! the `DataClient` instances, and the rest of the platform. This includes sending
//! requests to, and receiving responses from, data endpoints via its registered
//! data clients.
//!
//! The engine employs a simple fan-in fan-out messaging pattern to execute
//! `DataCommand` type messages, and process `DataResponse` messages or market data
//! objects.
//!
//! Alternative implementations can be written on top of the generic engine - which
//! just need to override the `execute`, `process`, `send` and `receive` methods.

pub mod bar;
pub mod book;
mod commands;
pub mod config;
mod handlers;

#[cfg(feature = "defi")]
pub mod pool;

use std::{
    any::{Any, type_name},
    cell::{Ref, RefCell},
    collections::VecDeque,
    fmt::{Debug, Display},
    num::NonZeroUsize,
    rc::Rc,
};

use ahash::{AHashMap, AHashSet};
pub use bar::BarAggregatorSubscription;
use book::{
    BookSnapshotInfo, BookSnapshotInfos, BookSnapshotKey, BookSnapshotUnsubscribeResult,
    BookSnapshotter, BookUpdater,
};
pub(crate) use commands::{DeferredCommand, DeferredCommandQueue};
use config::DataEngineConfig;
use futures::future::join_all;
use handlers::{BarBarHandler, BarQuoteHandler, BarTradeHandler};
use indexmap::IndexMap;
use nautilus_common::{
    cache::Cache,
    clock::Clock,
    logging::{RECV, RES},
    messages::data::{
        DataCommand, DataResponse, ForwardPricesResponse, RequestCommand, RequestForwardPrices,
        SubscribeBars, SubscribeBookDeltas, SubscribeBookDepth10, SubscribeBookSnapshots,
        SubscribeCommand, SubscribeOptionChain, UnsubscribeBars, UnsubscribeBookDeltas,
        UnsubscribeBookDepth10, UnsubscribeBookSnapshots, UnsubscribeCommand,
        UnsubscribeInstrumentStatus, UnsubscribeOptionChain, UnsubscribeOptionGreeks,
        UnsubscribeQuotes,
    },
    msgbus::{
        self, ShareableMessageHandler, TypedHandler, TypedIntoHandler,
        switchboard::{self, MessagingSwitchboard},
    },
    runner::get_data_cmd_sender,
    timer::{TimeEvent, TimeEventCallback},
};
use nautilus_core::{
    UUID4, WeakCell,
    correctness::{
        FAILED, check_key_in_map, check_key_not_in_map, check_predicate_false, check_predicate_true,
    },
    datetime::millis_to_nanos_unchecked,
};
#[cfg(feature = "defi")]
use nautilus_model::defi::DefiData;
use nautilus_model::{
    data::{
        Bar, BarType, CustomData, Data, DataType, FundingRateUpdate, IndexPriceUpdate,
        InstrumentClose, InstrumentStatus, MarkPriceUpdate, OrderBookDelta, OrderBookDeltas,
        OrderBookDepth10, QuoteTick, TradeTick,
        option_chain::{OptionGreeks, StrikeRange},
    },
    enums::{
        AggregationSource, BarAggregation, BookType, MarketStatusAction, PriceType, RecordFlag,
    },
    identifiers::{ClientId, InstrumentId, OptionSeriesId, Venue},
    instruments::{Instrument, InstrumentAny, SyntheticInstrument},
    orderbook::OrderBook,
    types::Price,
};
#[cfg(feature = "streaming")]
use nautilus_persistence::backend::catalog::ParquetDataCatalog;
use ustr::Ustr;

#[cfg(feature = "defi")]
#[allow(unused_imports)] // Brings DeFi impl blocks into scope
use crate::defi::engine as _;
#[cfg(feature = "defi")]
use crate::engine::pool::PoolUpdater;
use crate::{
    aggregation::{
        BarAggregator, RenkoBarAggregator, TickBarAggregator, TickImbalanceBarAggregator,
        TickRunsBarAggregator, TimeBarAggregator, ValueBarAggregator, ValueImbalanceBarAggregator,
        ValueRunsBarAggregator, VolumeBarAggregator, VolumeImbalanceBarAggregator,
        VolumeRunsBarAggregator,
    },
    client::DataClientAdapter,
    option_chains::OptionChainManager,
};

/// Provides a high-performance `DataEngine` for all environments.
#[derive(Debug)]
pub struct DataEngine {
    pub(crate) clock: Rc<RefCell<dyn Clock>>,
    pub(crate) cache: Rc<RefCell<Cache>>,
    pub(crate) external_clients: AHashSet<ClientId>,
    clients: IndexMap<ClientId, DataClientAdapter>,
    default_client: Option<DataClientAdapter>,
    #[cfg(feature = "streaming")]
    catalogs: AHashMap<Ustr, ParquetDataCatalog>,
    routing_map: IndexMap<Venue, ClientId>,
    book_intervals: AHashMap<NonZeroUsize, BookSnapshotInfos>,
    book_snapshot_counts: IndexMap<BookSnapshotKey, usize>,
    book_deltas_subs: AHashSet<InstrumentId>,
    book_depth10_subs: AHashSet<InstrumentId>,
    book_updaters: AHashMap<InstrumentId, Rc<BookUpdater>>,
    book_snapshotters: AHashMap<NonZeroUsize, Rc<BookSnapshotter>>,
    bar_aggregators: IndexMap<BarType, Rc<RefCell<Box<dyn BarAggregator>>>>,
    bar_aggregator_handlers: AHashMap<BarType, Vec<BarAggregatorSubscription>>,
    option_chain_managers: AHashMap<OptionSeriesId, Rc<RefCell<OptionChainManager>>>,
    option_chain_instrument_index: AHashMap<InstrumentId, OptionSeriesId>,
    deferred_cmd_queue: DeferredCommandQueue,
    pending_option_chain_requests: AHashMap<UUID4, SubscribeOptionChain>,
    _synthetic_quote_feeds: AHashMap<InstrumentId, Vec<SyntheticInstrument>>,
    _synthetic_trade_feeds: AHashMap<InstrumentId, Vec<SyntheticInstrument>>,
    buffered_deltas_map: AHashMap<InstrumentId, OrderBookDeltas>,
    command_count: u64,
    data_count: u64,
    request_count: u64,
    response_count: u64,
    pub(crate) msgbus_priority: u8,
    pub(crate) config: DataEngineConfig,
    #[cfg(feature = "defi")]
    pub(crate) pool_updaters: AHashMap<InstrumentId, Rc<PoolUpdater>>,
    #[cfg(feature = "defi")]
    pub(crate) pool_updaters_pending: AHashSet<InstrumentId>,
    #[cfg(feature = "defi")]
    pub(crate) pool_snapshot_pending: AHashSet<InstrumentId>,
    #[cfg(feature = "defi")]
    pub(crate) pool_event_buffers: AHashMap<InstrumentId, Vec<DefiData>>,
}

impl DataEngine {
    /// Creates a new [`DataEngine`] instance.
    #[must_use]
    pub fn new(
        clock: Rc<RefCell<dyn Clock>>,
        cache: Rc<RefCell<Cache>>,
        config: Option<DataEngineConfig>,
    ) -> Self {
        let config = config.unwrap_or_default();

        let external_clients: AHashSet<ClientId> = config
            .external_clients
            .clone()
            .unwrap_or_default()
            .into_iter()
            .collect();

        Self {
            clock,
            cache,
            external_clients,
            clients: IndexMap::new(),
            default_client: None,
            #[cfg(feature = "streaming")]
            catalogs: AHashMap::new(),
            routing_map: IndexMap::new(),
            book_intervals: AHashMap::new(),
            book_snapshot_counts: IndexMap::new(),
            book_deltas_subs: AHashSet::new(),
            book_depth10_subs: AHashSet::new(),
            book_updaters: AHashMap::new(),
            book_snapshotters: AHashMap::new(),
            bar_aggregators: IndexMap::new(),
            bar_aggregator_handlers: AHashMap::new(),
            option_chain_managers: AHashMap::new(),
            option_chain_instrument_index: AHashMap::new(),
            deferred_cmd_queue: Rc::new(RefCell::new(VecDeque::new())),
            pending_option_chain_requests: AHashMap::new(),
            _synthetic_quote_feeds: AHashMap::new(),
            _synthetic_trade_feeds: AHashMap::new(),
            buffered_deltas_map: AHashMap::new(),
            command_count: 0,
            data_count: 0,
            request_count: 0,
            response_count: 0,
            msgbus_priority: 10, // High-priority for built-in component
            config,
            #[cfg(feature = "defi")]
            pool_updaters: AHashMap::new(),
            #[cfg(feature = "defi")]
            pool_updaters_pending: AHashSet::new(),
            #[cfg(feature = "defi")]
            pool_snapshot_pending: AHashSet::new(),
            #[cfg(feature = "defi")]
            pool_event_buffers: AHashMap::new(),
        }
    }

    /// Registers all message bus handlers for the data engine.
    pub fn register_msgbus_handlers(engine: &Rc<RefCell<Self>>) {
        let weak = WeakCell::from(Rc::downgrade(engine));

        let weak1 = weak.clone();
        msgbus::register_data_command_endpoint(
            MessagingSwitchboard::data_engine_execute(),
            TypedIntoHandler::from(move |cmd: DataCommand| {
                if let Some(rc) = weak1.upgrade() {
                    rc.borrow_mut().execute(cmd);
                }
            }),
        );

        msgbus::register_data_command_endpoint(
            MessagingSwitchboard::data_engine_queue_execute(),
            TypedIntoHandler::from(move |cmd: DataCommand| {
                get_data_cmd_sender().clone().execute(cmd);
            }),
        );

        // Register process handler (polymorphic - uses Any)
        let weak2 = weak.clone();
        msgbus::register_any(
            MessagingSwitchboard::data_engine_process(),
            ShareableMessageHandler::from_any(move |data: &dyn Any| {
                if let Some(rc) = weak2.upgrade() {
                    rc.borrow_mut().process(data);
                }
            }),
        );

        // Register process_data handler (typed - takes ownership)
        let weak3 = weak.clone();
        msgbus::register_data_endpoint(
            MessagingSwitchboard::data_engine_process_data(),
            TypedIntoHandler::from(move |data: Data| {
                if let Some(rc) = weak3.upgrade() {
                    rc.borrow_mut().process_data(data);
                }
            }),
        );

        // Register process_defi_data handler (typed - takes ownership)
        #[cfg(feature = "defi")]
        {
            let weak4 = weak.clone();
            msgbus::register_defi_data_endpoint(
                MessagingSwitchboard::data_engine_process_defi_data(),
                TypedIntoHandler::from(move |data: DefiData| {
                    if let Some(rc) = weak4.upgrade() {
                        rc.borrow_mut().process_defi_data(data);
                    }
                }),
            );
        }

        let weak5 = weak;
        msgbus::register_data_response_endpoint(
            MessagingSwitchboard::data_engine_response(),
            TypedIntoHandler::from(move |resp: DataResponse| {
                if let Some(rc) = weak5.upgrade() {
                    rc.borrow_mut().response(resp);
                }
            }),
        );
    }

    /// Returns the total count of data commands received by the engine.
    #[must_use]
    pub const fn command_count(&self) -> u64 {
        self.command_count
    }

    /// Returns the total count of data stream objects received by the engine.
    #[must_use]
    pub const fn data_count(&self) -> u64 {
        self.data_count
    }

    #[cfg(feature = "defi")]
    pub(crate) const fn increment_data_count(&mut self) {
        self.data_count += 1;
    }

    /// Returns the total count of data requests received by the engine.
    #[must_use]
    pub const fn request_count(&self) -> u64 {
        self.request_count
    }

    /// Returns the total count of data responses received by the engine.
    #[must_use]
    pub const fn response_count(&self) -> u64 {
        self.response_count
    }

    /// Returns whether an `OptionChainManager` exists for the given series.
    #[must_use]
    pub fn has_option_chain_manager(&self, series_id: &OptionSeriesId) -> bool {
        self.option_chain_managers.contains_key(series_id)
    }

    /// Returns the count of pending option-chain bootstrap requests.
    #[must_use]
    pub fn pending_option_chain_request_count(&self) -> usize {
        self.pending_option_chain_requests.len()
    }

    /// Returns a read-only reference to the engines clock.
    #[must_use]
    pub fn get_clock(&self) -> Ref<'_, dyn Clock> {
        self.clock.borrow()
    }

    /// Returns a read-only reference to the engines cache.
    #[must_use]
    pub fn get_cache(&self) -> Ref<'_, Cache> {
        self.cache.borrow()
    }

    /// Returns the `Rc<RefCell<Cache>>` used by this engine.
    #[must_use]
    pub fn cache_rc(&self) -> Rc<RefCell<Cache>> {
        Rc::clone(&self.cache)
    }

    /// Registers the `catalog` with the engine with an optional specific `name`.
    ///
    /// # Panics
    ///
    /// Panics if a catalog with the same `name` has already been registered.
    #[cfg(feature = "streaming")]
    pub fn register_catalog(&mut self, catalog: ParquetDataCatalog, name: Option<&str>) {
        let name = Ustr::from(name.unwrap_or("catalog_0"));

        check_key_not_in_map(&name, &self.catalogs, "name", "catalogs").expect(FAILED);

        self.catalogs.insert(name, catalog);
        log::info!("Registered catalog <{name}>");
    }

    /// Registers the `client` with the engine with an optional venue `routing`.
    ///
    ///
    /// # Panics
    ///
    /// Panics if a client with the same client ID has already been registered.
    pub fn register_client(&mut self, client: DataClientAdapter, routing: Option<Venue>) {
        let client_id = client.client_id();

        if let Some(default_client) = &self.default_client {
            check_predicate_false(
                default_client.client_id() == client.client_id(),
                "client_id already registered as default client",
            )
            .expect(FAILED);
        }

        check_key_not_in_map(&client_id, &self.clients, "client_id", "clients").expect(FAILED);

        if let Some(routing) = routing {
            self.routing_map.insert(routing, client_id);
            log::debug!("Set client {client_id} routing for {routing}");
        }

        if client.venue.is_none() && self.default_client.is_none() {
            self.default_client = Some(client);
            log::debug!("Registered client {client_id} for default routing");
        } else {
            self.clients.insert(client_id, client);
            log::debug!("Registered client {client_id}");
        }
    }

    /// Deregisters the client for the `client_id`.
    ///
    /// # Panics
    ///
    /// Panics if the client ID has not been registered.
    pub fn deregister_client(&mut self, client_id: &ClientId) {
        check_key_in_map(client_id, &self.clients, "client_id", "clients").expect(FAILED);

        self.clients.shift_remove(client_id);
        log::info!("Deregistered client {client_id}");
    }

    /// Registers the data `client` with the engine as the default routing client.
    ///
    /// When a specific venue routing cannot be found, this client will receive messages.
    ///
    /// # Warnings
    ///
    /// Any existing default routing client will be overwritten.
    ///
    /// # Panics
    ///
    /// Panics if a default client has already been registered.
    pub fn register_default_client(&mut self, client: DataClientAdapter) {
        check_predicate_true(
            self.default_client.is_none(),
            "default client already registered",
        )
        .expect(FAILED);

        let client_id = client.client_id();

        self.default_client = Some(client);
        log::debug!("Registered default client {client_id}");
    }

    /// Starts all registered data clients and re-arms bar aggregator timers.
    pub fn start(&mut self) {
        for client in self.get_clients_mut() {
            if let Err(e) = client.start() {
                log::error!("{e}");
            }
        }

        for aggregator in self.bar_aggregators.values() {
            if aggregator.borrow().bar_type().spec().is_time_aggregated() {
                aggregator
                    .borrow_mut()
                    .start_timer(Some(aggregator.clone()));
            }
        }
    }

    /// Stops all registered data clients and bar aggregator timers.
    pub fn stop(&mut self) {
        for client in self.get_clients_mut() {
            if let Err(e) = client.stop() {
                log::error!("{e}");
            }
        }

        for aggregator in self.bar_aggregators.values() {
            aggregator.borrow_mut().stop();
        }
    }

    /// Resets all registered data clients and clears bar aggregator state.
    pub fn reset(&mut self) {
        for client in self.get_clients_mut() {
            if let Err(e) = client.reset() {
                log::error!("{e}");
            }
        }

        let bar_types: Vec<BarType> = self.bar_aggregators.keys().copied().collect();
        for bar_type in bar_types {
            if let Err(e) = self.stop_bar_aggregator(bar_type) {
                log::error!("Error stopping bar aggregator during reset for {bar_type}: {e}");
            }
        }

        self.command_count = 0;
        self.data_count = 0;
        self.request_count = 0;
        self.response_count = 0;
    }

    /// Disposes the engine, stopping all clients and canceling any timers.
    pub fn dispose(&mut self) {
        for client in self.get_clients_mut() {
            if let Err(e) = client.dispose() {
                log::error!("{e}");
            }
        }

        self.clock.borrow_mut().cancel_timers();
    }

    /// Connects all registered data clients concurrently.
    ///
    /// Connection failures are logged but do not prevent the node from running.
    pub async fn connect(&mut self) {
        let futures: Vec<_> = self
            .get_clients_mut()
            .into_iter()
            .map(|client| client.connect())
            .collect();

        let results = join_all(futures).await;

        for error in results.into_iter().filter_map(Result::err) {
            log::error!("Failed to connect data client: {error}");
        }
    }

    /// Disconnects all registered data clients concurrently.
    ///
    /// # Errors
    ///
    /// Returns an error if any client fails to disconnect.
    pub async fn disconnect(&mut self) -> anyhow::Result<()> {
        let futures: Vec<_> = self
            .get_clients_mut()
            .into_iter()
            .map(|client| client.disconnect())
            .collect();

        let results = join_all(futures).await;
        let errors: Vec<_> = results.into_iter().filter_map(Result::err).collect();

        if errors.is_empty() {
            Ok(())
        } else {
            let error_msgs: Vec<_> = errors.iter().map(|e| e.to_string()).collect();
            anyhow::bail!(
                "Failed to disconnect data clients: {}",
                error_msgs.join("; ")
            )
        }
    }

    /// Returns `true` if all registered data clients are currently connected.
    #[must_use]
    pub fn check_connected(&self) -> bool {
        self.get_clients()
            .iter()
            .all(|client| client.is_connected())
    }

    /// Returns `true` if all registered data clients are currently disconnected.
    #[must_use]
    pub fn check_disconnected(&self) -> bool {
        self.get_clients()
            .iter()
            .all(|client| !client.is_connected())
    }

    /// Returns connection status for each registered client.
    #[must_use]
    pub fn client_connection_status(&self) -> Vec<(ClientId, bool)> {
        self.get_clients()
            .into_iter()
            .map(|client| (client.client_id(), client.is_connected()))
            .collect()
    }

    /// Returns a list of all registered client IDs, including the default client if set.
    #[must_use]
    pub fn registered_clients(&self) -> Vec<ClientId> {
        self.get_clients()
            .into_iter()
            .map(|client| client.client_id())
            .collect()
    }

    // -- SUBSCRIPTIONS ---------------------------------------------------------------------------

    pub(crate) fn collect_subscriptions<F, T>(&self, get_subs: F) -> Vec<T>
    where
        F: Fn(&DataClientAdapter) -> &AHashSet<T>,
        T: Clone,
    {
        self.get_clients()
            .into_iter()
            .flat_map(get_subs)
            .cloned()
            .collect()
    }

    #[must_use]
    pub fn get_clients(&self) -> Vec<&DataClientAdapter> {
        let (default_opt, clients_map) = (&self.default_client, &self.clients);
        let mut clients: Vec<&DataClientAdapter> = clients_map.values().collect();

        if let Some(default) = default_opt {
            clients.push(default);
        }

        clients
    }

    #[must_use]
    pub fn get_clients_mut(&mut self) -> Vec<&mut DataClientAdapter> {
        let (default_opt, clients_map) = (&mut self.default_client, &mut self.clients);
        let mut clients: Vec<&mut DataClientAdapter> = clients_map.values_mut().collect();

        if let Some(default) = default_opt {
            clients.push(default);
        }

        clients
    }

    pub fn get_client(
        &mut self,
        client_id: Option<&ClientId>,
        venue: Option<&Venue>,
    ) -> Option<&mut DataClientAdapter> {
        if let Some(client_id) = client_id {
            // Explicit ID: first look in registered clients
            if let Some(client) = self.clients.get_mut(client_id) {
                return Some(client);
            }

            // Then check if it matches the default client
            if let Some(default) = self.default_client.as_mut()
                && default.client_id() == *client_id
            {
                return Some(default);
            }

            // Unknown explicit client
            return None;
        }

        if let Some(v) = venue {
            // Route by venue if mapped client still registered
            if let Some(client_id) = self.routing_map.get(v) {
                return self.clients.get_mut(client_id);
            }
        }

        // Fallback to default client
        self.get_default_client()
    }

    const fn get_default_client(&mut self) -> Option<&mut DataClientAdapter> {
        self.default_client.as_mut()
    }

    /// Returns all custom data types currently subscribed across all clients.
    #[must_use]
    pub fn subscribed_custom_data(&self) -> Vec<DataType> {
        self.collect_subscriptions(|client| &client.subscriptions_custom)
    }

    /// Returns all instrument IDs currently subscribed across all clients.
    #[must_use]
    pub fn subscribed_instruments(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_instrument)
    }

    /// Returns all instrument IDs for which book delta subscriptions exist.
    #[must_use]
    pub fn subscribed_book_deltas(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_book_deltas)
    }

    /// Returns all instrument IDs for which book depth10 subscriptions exist.
    #[must_use]
    pub fn subscribed_book_depth10(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_book_depth10)
    }

    /// Returns all instrument IDs for which book snapshot subscriptions exist.
    #[must_use]
    pub fn subscribed_book_snapshots(&self) -> Vec<InstrumentId> {
        self.book_snapshot_counts
            .keys()
            .map(|(instrument_id, _)| *instrument_id)
            .collect()
    }

    /// Returns all instrument IDs for which quote subscriptions exist.
    #[must_use]
    pub fn subscribed_quotes(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_quotes)
    }

    /// Returns all instrument IDs for which trade subscriptions exist.
    #[must_use]
    pub fn subscribed_trades(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_trades)
    }

    /// Returns all bar types currently subscribed across all clients.
    #[must_use]
    pub fn subscribed_bars(&self) -> Vec<BarType> {
        self.collect_subscriptions(|client| &client.subscriptions_bars)
    }

    /// Returns all instrument IDs for which mark price subscriptions exist.
    #[must_use]
    pub fn subscribed_mark_prices(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_mark_prices)
    }

    /// Returns all instrument IDs for which index price subscriptions exist.
    #[must_use]
    pub fn subscribed_index_prices(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_index_prices)
    }

    /// Returns all instrument IDs for which funding rate subscriptions exist.
    #[must_use]
    pub fn subscribed_funding_rates(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_funding_rates)
    }

    /// Returns all instrument IDs for which status subscriptions exist.
    #[must_use]
    pub fn subscribed_instrument_status(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_instrument_status)
    }

    /// Returns all instrument IDs for which instrument close subscriptions exist.
    #[must_use]
    pub fn subscribed_instrument_close(&self) -> Vec<InstrumentId> {
        self.collect_subscriptions(|client| &client.subscriptions_instrument_close)
    }

    // -- COMMANDS --------------------------------------------------------------------------------

    /// Executes a `DataCommand` by delegating to subscribe, unsubscribe, or request handlers.
    ///
    /// Errors during execution are logged.
    pub fn execute(&mut self, cmd: DataCommand) {
        match &cmd {
            DataCommand::Subscribe(_) | DataCommand::Unsubscribe(_) => self.command_count += 1,
            DataCommand::Request(_) => self.request_count += 1,
            #[cfg(feature = "defi")]
            DataCommand::DefiRequest(_) => self.request_count += 1,
            #[cfg(feature = "defi")]
            DataCommand::DefiSubscribe(_) | DataCommand::DefiUnsubscribe(_) => {
                self.command_count += 1;
            }
            _ => {}
        }

        if let Err(e) = match cmd {
            DataCommand::Subscribe(c) => self.execute_subscribe(c),
            DataCommand::Unsubscribe(c) => self.execute_unsubscribe(&c),
            DataCommand::Request(c) => self.execute_request(c),
            #[cfg(feature = "defi")]
            DataCommand::DefiRequest(c) => self.execute_defi_request(c),
            #[cfg(feature = "defi")]
            DataCommand::DefiSubscribe(c) => self.execute_defi_subscribe(c),
            #[cfg(feature = "defi")]
            DataCommand::DefiUnsubscribe(c) => self.execute_defi_unsubscribe(&c),
            _ => {
                log::warn!("Unhandled DataCommand variant");
                Ok(())
            }
        } {
            log::error!("{e}");
        }
    }

    /// Handles a subscribe command, updating internal state and forwarding to the client.
    ///
    /// # Errors
    ///
    /// Returns an error if the subscription is invalid (e.g., synthetic instrument for book data),
    /// or if the underlying client operation fails.
    pub fn execute_subscribe(&mut self, cmd: SubscribeCommand) -> anyhow::Result<()> {
        // Update internal engine state
        match &cmd {
            SubscribeCommand::BookDeltas(cmd) => self.subscribe_book_deltas(cmd)?,
            SubscribeCommand::BookDepth10(cmd) => self.subscribe_book_depth10(cmd)?,
            SubscribeCommand::BookSnapshots(cmd) => {
                // Handles client forwarding internally (forwards as BookDeltas)
                return self.subscribe_book_snapshots(cmd);
            }
            SubscribeCommand::Bars(cmd) => self.subscribe_bars(cmd)?,
            SubscribeCommand::OptionChain(cmd) => {
                self.subscribe_option_chain(cmd);
                return Ok(());
            }
            SubscribeCommand::Instrument(cmd) if cmd.instrument_id.is_synthetic() => {
                anyhow::bail!("Cannot subscribe for synthetic instrument `Instrument` data");
            }
            SubscribeCommand::InstrumentStatus(cmd) if cmd.instrument_id.is_synthetic() => {
                anyhow::bail!("Cannot subscribe for synthetic instrument `InstrumentStatus` data");
            }
            SubscribeCommand::InstrumentClose(cmd) if cmd.instrument_id.is_synthetic() => {
                anyhow::bail!("Cannot subscribe for synthetic instrument `InstrumentClose` data");
            }
            SubscribeCommand::OptionGreeks(cmd) if cmd.instrument_id.is_synthetic() => {
                anyhow::bail!("Cannot subscribe for synthetic instrument `OptionGreeks` data");
            }
            _ => {} // Do nothing else
        }

        if let Some(client_id) = cmd.client_id()
            && self.external_clients.contains(client_id)
        {
            if self.config.debug {
                log::debug!("Skipping subscribe command for external client {client_id}: {cmd:?}",);
            }
            return Ok(());
        }

        if let Some(client) = self.get_client(cmd.client_id(), cmd.venue()) {
            client.execute_subscribe(cmd);
        } else {
            log::error!(
                "Cannot handle command: no client found for client_id={:?}, venue={:?}",
                cmd.client_id(),
                cmd.venue(),
            );
        }

        Ok(())
    }

    /// Handles an unsubscribe command, updating internal state and forwarding to the client.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying client operation fails.
    pub fn execute_unsubscribe(&mut self, cmd: &UnsubscribeCommand) -> anyhow::Result<()> {
        match &cmd {
            UnsubscribeCommand::BookDeltas(cmd) if !self.unsubscribe_book_deltas(cmd) => {
                return Ok(());
            }
            UnsubscribeCommand::BookDepth10(cmd) if !self.unsubscribe_book_depth10(cmd) => {
                return Ok(());
            }
            UnsubscribeCommand::BookSnapshots(cmd) => {
                // Handles client forwarding internally (forwards as BookDeltas)
                self.unsubscribe_book_snapshots(cmd);
                return Ok(());
            }
            UnsubscribeCommand::Bars(cmd) => self.unsubscribe_bars(cmd),
            UnsubscribeCommand::OptionChain(cmd) => {
                self.unsubscribe_option_chain(cmd);
                return Ok(());
            }
            UnsubscribeCommand::Instrument(cmd) if cmd.instrument_id.is_synthetic() => {
                anyhow::bail!("Cannot unsubscribe from synthetic instrument `Instrument` data");
            }
            UnsubscribeCommand::InstrumentStatus(cmd) if cmd.instrument_id.is_synthetic() => {
                anyhow::bail!(
                    "Cannot unsubscribe from synthetic instrument `InstrumentStatus` data"
                );
            }
            UnsubscribeCommand::InstrumentClose(cmd) if cmd.instrument_id.is_synthetic() => {
                anyhow::bail!(
                    "Cannot unsubscribe from synthetic instrument `InstrumentClose` data"
                );
            }
            UnsubscribeCommand::OptionGreeks(cmd) if cmd.instrument_id.is_synthetic() => {
                anyhow::bail!("Cannot unsubscribe from synthetic instrument `OptionGreeks` data");
            }
            _ => {} // Do nothing else
        }

        if let Some(client_id) = cmd.client_id()
            && self.external_clients.contains(client_id)
        {
            if self.config.debug {
                log::debug!(
                    "Skipping unsubscribe command for external client {client_id}: {cmd:?}",
                );
            }
            return Ok(());
        }

        if let Some(client) = self.get_client(cmd.client_id(), cmd.venue()) {
            client.execute_unsubscribe(cmd);
        } else {
            log::error!(
                "Cannot handle command: no client found for client_id={:?}, venue={:?}",
                cmd.client_id(),
                cmd.venue(),
            );
        }

        Ok(())
    }

    /// Sends a [`RequestCommand`] to a suitable data client implementation.
    ///
    /// # Errors
    ///
    /// Returns an error if no client is found for the given client ID or venue,
    /// or if the client fails to process the request.
    pub fn execute_request(&mut self, req: RequestCommand) -> anyhow::Result<()> {
        // Skip requests for external clients
        if let Some(cid) = req.client_id()
            && self.external_clients.contains(cid)
        {
            if self.config.debug {
                log::debug!("Skipping data request for external client {cid}: {req:?}");
            }
            return Ok(());
        }

        if let Some(client) = self.get_client(req.client_id(), req.venue()) {
            match req {
                RequestCommand::Data(req) => client.request_data(req),
                RequestCommand::Instrument(req) => client.request_instrument(req),
                RequestCommand::Instruments(req) => client.request_instruments(req),
                RequestCommand::BookSnapshot(req) => client.request_book_snapshot(req),
                RequestCommand::BookDepth(req) => client.request_book_depth(req),
                RequestCommand::Quotes(req) => client.request_quotes(req),
                RequestCommand::Trades(req) => client.request_trades(req),
                RequestCommand::FundingRates(req) => client.request_funding_rates(req),
                RequestCommand::ForwardPrices(req) => client.request_forward_prices(req),
                RequestCommand::Bars(req) => client.request_bars(req),
            }
        } else {
            anyhow::bail!(
                "Cannot handle request: no client found for {:?} {:?}",
                req.client_id(),
                req.venue()
            );
        }
    }

    /// Processes a dynamically-typed data message.
    ///
    /// Currently supports `InstrumentAny` and `FundingRateUpdate`; unrecognized types are logged as errors.
    pub fn process(&mut self, data: &dyn Any) {
        self.data_count += 1;
        // TODO: Eventually these can be added to the `Data` enum (C/Cython blocking), process here for now
        if let Some(instrument) = data.downcast_ref::<InstrumentAny>() {
            self.handle_instrument(instrument);
        } else if let Some(funding_rate) = data.downcast_ref::<FundingRateUpdate>() {
            self.handle_funding_rate(*funding_rate);
        } else if let Some(status) = data.downcast_ref::<InstrumentStatus>() {
            self.handle_instrument_status(*status);
        } else if let Some(option_greeks) = data.downcast_ref::<OptionGreeks>() {
            self.cache.borrow_mut().add_option_greeks(*option_greeks);
            let topic = switchboard::get_option_greeks_topic(option_greeks.instrument_id);
            msgbus::publish_option_greeks(topic, option_greeks);
            self.drain_deferred_commands();
        } else {
            log::error!("Cannot process data {data:?}, type is unrecognized");
        }

        // TODO: Add custom data handling here
    }

    /// Processes a `Data` enum instance, dispatching to appropriate handlers.
    pub fn process_data(&mut self, data: Data) {
        self.data_count += 1;

        match data {
            Data::Delta(delta) => self.handle_delta(delta),
            Data::Deltas(deltas) => self.handle_deltas(deltas.into_inner()),
            Data::Depth10(depth) => self.handle_depth10(*depth),
            Data::Quote(quote) => {
                self.handle_quote(quote);
                self.drain_deferred_commands();
            }
            Data::Trade(trade) => self.handle_trade(trade),
            Data::Bar(bar) => self.handle_bar(bar),
            Data::MarkPriceUpdate(mark_price) => {
                self.handle_mark_price(mark_price);
                self.drain_deferred_commands();
            }
            Data::IndexPriceUpdate(index_price) => {
                self.handle_index_price(index_price);
                self.drain_deferred_commands();
            }
            Data::InstrumentStatus(status) => {
                self.handle_instrument_status(status);
                self.drain_deferred_commands();
            }
            Data::InstrumentClose(close) => self.handle_instrument_close(close),
            Data::Custom(custom) => self.handle_custom_data(&custom),
        }
    }

    /// Processes a `DataResponse`, handling and publishing the response message.
    #[expect(clippy::needless_pass_by_value)] // Required by message bus dispatch
    pub fn response(&mut self, resp: DataResponse) {
        log::debug!("{RECV}{RES} {resp:?}");

        self.response_count += 1;
        let correlation_id = *resp.correlation_id();

        match &resp {
            DataResponse::Instrument(r) => {
                self.handle_instrument_response(r.data.clone());
            }
            DataResponse::Instruments(r) => {
                self.handle_instruments(&r.data);
            }
            DataResponse::Quotes(r) => {
                if !log_if_empty_response(&r.data, &r.instrument_id, &correlation_id) {
                    self.handle_quotes(&r.data);
                }
            }
            DataResponse::Trades(r) => {
                if !log_if_empty_response(&r.data, &r.instrument_id, &correlation_id) {
                    self.handle_trades(&r.data);
                }
            }
            DataResponse::FundingRates(r) => {
                if !log_if_empty_response(&r.data, &r.instrument_id, &correlation_id) {
                    self.handle_funding_rates(&r.data);
                }
            }
            DataResponse::Bars(r) => {
                if !log_if_empty_response(&r.data, &r.bar_type, &correlation_id) {
                    self.handle_bars(&r.data);
                }
            }
            DataResponse::Book(r) => self.handle_book_response(&r.data),
            DataResponse::ForwardPrices(r) => {
                return self.handle_forward_prices_response(&correlation_id, r);
            }
            _ => todo!("Handle other response types"),
        }

        msgbus::send_response(&correlation_id, &resp);
    }

    // -- DATA HANDLERS ---------------------------------------------------------------------------

    fn handle_instrument(&mut self, instrument: &InstrumentAny) {
        log::debug!("Handling instrument: {}", instrument.id());

        if let Err(e) = self
            .cache
            .as_ref()
            .borrow_mut()
            .add_instrument(instrument.clone())
        {
            log_error_on_cache_insert(&e);
        }

        let topic = switchboard::get_instrument_topic(instrument.id());
        log::debug!("Publishing instrument to topic: {topic}");
        msgbus::publish_any(topic, instrument);

        self.update_option_chains(instrument);
    }

    fn update_option_chains(&mut self, instrument: &InstrumentAny) {
        let Some(underlying) = instrument.underlying() else {
            return;
        };
        let Some(expiration_ns) = instrument.expiration_ns() else {
            return;
        };
        let Some(strike) = instrument.strike_price() else {
            return;
        };
        let Some(kind) = instrument.option_kind() else {
            return;
        };

        let venue = instrument.id().venue;
        let settlement = instrument.settlement_currency().code;
        let series_id = OptionSeriesId::new(venue, underlying, settlement, expiration_ns);

        // Clone Rc to release borrow on self.option_chain_managers before accessing self.clients
        let Some(manager_rc) = self.option_chain_managers.get(&series_id).cloned() else {
            return;
        };

        let clock = self.clock.clone();
        let client = self.get_client(None, Some(&venue));

        if manager_rc
            .borrow_mut()
            .add_instrument(instrument.id(), strike, kind, client, &clock)
        {
            self.option_chain_instrument_index
                .insert(instrument.id(), series_id);
        }
    }

    fn handle_delta(&mut self, delta: OrderBookDelta) {
        let deltas = if self.config.buffer_deltas {
            if let Some(buffered_deltas) = self.buffered_deltas_map.get_mut(&delta.instrument_id) {
                buffered_deltas.deltas.push(delta);
                buffered_deltas.flags = delta.flags;
                buffered_deltas.sequence = delta.sequence;
                buffered_deltas.ts_event = delta.ts_event;
                buffered_deltas.ts_init = delta.ts_init;
            } else {
                let buffered_deltas = OrderBookDeltas::new(delta.instrument_id, vec![delta]);
                self.buffered_deltas_map
                    .insert(delta.instrument_id, buffered_deltas);
            }

            if !RecordFlag::F_LAST.matches(delta.flags) {
                return; // Not the last delta for event
            }

            self.buffered_deltas_map
                .remove(&delta.instrument_id)
                .expect("buffered deltas exist")
        } else {
            OrderBookDeltas::new(delta.instrument_id, vec![delta])
        };

        let topic = switchboard::get_book_deltas_topic(deltas.instrument_id);
        msgbus::publish_deltas(topic, &deltas);
    }

    fn handle_deltas(&mut self, deltas: OrderBookDeltas) {
        if self.config.buffer_deltas {
            let instrument_id = deltas.instrument_id;

            for delta in deltas.deltas {
                if let Some(buffered_deltas) = self.buffered_deltas_map.get_mut(&instrument_id) {
                    buffered_deltas.deltas.push(delta);
                    buffered_deltas.flags = delta.flags;
                    buffered_deltas.sequence = delta.sequence;
                    buffered_deltas.ts_event = delta.ts_event;
                    buffered_deltas.ts_init = delta.ts_init;
                } else {
                    let buffered_deltas = OrderBookDeltas::new(instrument_id, vec![delta]);
                    self.buffered_deltas_map
                        .insert(instrument_id, buffered_deltas);
                }

                if RecordFlag::F_LAST.matches(delta.flags) {
                    let deltas_to_publish = self
                        .buffered_deltas_map
                        .remove(&instrument_id)
                        .expect("buffered deltas exist");
                    let topic = switchboard::get_book_deltas_topic(instrument_id);
                    msgbus::publish_deltas(topic, &deltas_to_publish);
                }
            }
        } else {
            let topic = switchboard::get_book_deltas_topic(deltas.instrument_id);
            msgbus::publish_deltas(topic, &deltas);
        }
    }

    fn handle_depth10(&self, depth: OrderBookDepth10) {
        let topic = switchboard::get_book_depth10_topic(depth.instrument_id);
        msgbus::publish_depth10(topic, &depth);
    }

    fn handle_quote(&self, quote: QuoteTick) {
        if let Err(e) = self.cache.as_ref().borrow_mut().add_quote(quote) {
            log_error_on_cache_insert(&e);
        }

        // TODO: Handle synthetics

        let topic = switchboard::get_quotes_topic(quote.instrument_id);
        msgbus::publish_quote(topic, &quote);
    }

    fn handle_trade(&self, trade: TradeTick) {
        if let Err(e) = self.cache.as_ref().borrow_mut().add_trade(trade) {
            log_error_on_cache_insert(&e);
        }

        // TODO: Handle synthetics

        let topic = switchboard::get_trades_topic(trade.instrument_id);
        msgbus::publish_trade(topic, &trade);
    }

    fn handle_bar(&self, bar: Bar) {
        // TODO: Handle additional bar logic
        if self.config.validate_data_sequence
            && let Some(last_bar) = self.cache.as_ref().borrow().bar(&bar.bar_type)
        {
            if bar.ts_event < last_bar.ts_event {
                log::warn!(
                    "Bar {bar} was prior to last bar `ts_event` {}",
                    last_bar.ts_event
                );
                return; // Bar is out of sequence
            }

            if bar.ts_init < last_bar.ts_init {
                log::warn!(
                    "Bar {bar} was prior to last bar `ts_init` {}",
                    last_bar.ts_init
                );
                return; // Bar is out of sequence
            }
            // TODO: Implement `bar.is_revision` logic
        }

        if let Err(e) = self.cache.as_ref().borrow_mut().add_bar(bar) {
            log_error_on_cache_insert(&e);
        }

        let topic = switchboard::get_bars_topic(bar.bar_type);
        msgbus::publish_bar(topic, &bar);
    }

    fn handle_mark_price(&self, mark_price: MarkPriceUpdate) {
        if let Err(e) = self.cache.as_ref().borrow_mut().add_mark_price(mark_price) {
            log_error_on_cache_insert(&e);
        }

        let topic = switchboard::get_mark_price_topic(mark_price.instrument_id);
        msgbus::publish_mark_price(topic, &mark_price);
    }

    fn handle_index_price(&self, index_price: IndexPriceUpdate) {
        if let Err(e) = self
            .cache
            .as_ref()
            .borrow_mut()
            .add_index_price(index_price)
        {
            log_error_on_cache_insert(&e);
        }

        let topic = switchboard::get_index_price_topic(index_price.instrument_id);
        msgbus::publish_index_price(topic, &index_price);
    }

    /// Handles a funding rate update by adding it to the cache and publishing to the message bus.
    pub fn handle_funding_rate(&mut self, funding_rate: FundingRateUpdate) {
        if let Err(e) = self
            .cache
            .as_ref()
            .borrow_mut()
            .add_funding_rate(funding_rate)
        {
            log_error_on_cache_insert(&e);
        }

        let topic = switchboard::get_funding_rate_topic(funding_rate.instrument_id);
        msgbus::publish_funding_rate(topic, &funding_rate);
    }

    fn handle_instrument_status(&mut self, status: InstrumentStatus) {
        if let Err(e) = self
            .cache
            .as_ref()
            .borrow_mut()
            .add_instrument_status(status)
        {
            log_error_on_cache_insert(&e);
        }

        let topic = switchboard::get_instrument_status_topic(status.instrument_id);
        msgbus::publish_any(topic, &status);

        // Check if this instrument belongs to an option chain before expiring
        if self
            .option_chain_instrument_index
            .contains_key(&status.instrument_id)
            && matches!(
                status.action,
                MarketStatusAction::Close | MarketStatusAction::NotAvailableForTrading
            )
        {
            self.expire_option_chain_instrument(status.instrument_id);
        }
    }

    /// Removes a settled/expired instrument from its option chain manager.
    ///
    /// Looks up the owning series via the reverse index, delegates removal to
    /// the manager (which unregisters msgbus handlers and pushes deferred wire
    /// unsubscribes), then drains those commands. When the series catalog
    /// becomes empty, the entire manager is torn down.
    fn expire_option_chain_instrument(&mut self, instrument_id: InstrumentId) {
        let Some(series_id) = self.option_chain_instrument_index.remove(&instrument_id) else {
            return;
        };

        let Some(manager_rc) = self.option_chain_managers.get(&series_id).cloned() else {
            return;
        };

        let series_empty = manager_rc
            .borrow_mut()
            .handle_instrument_expired(&instrument_id);

        // Drain deferred unsubscribe commands pushed by the manager
        self.drain_deferred_commands();

        log::info!(
            "Expired instrument {instrument_id} from option chain {series_id} (series_empty={series_empty})",
        );

        if series_empty {
            manager_rc.borrow_mut().teardown(&self.clock);
            self.option_chain_managers.remove(&series_id);

            log::info!("Torn down empty option chain manager for {series_id}");
        }
    }

    fn handle_instrument_close(&self, close: InstrumentClose) {
        let topic = switchboard::get_instrument_close_topic(close.instrument_id);
        msgbus::publish_any(topic, &close);
    }

    fn handle_custom_data(&self, custom: &CustomData) {
        log::debug!("Processing custom data: {}", custom.data.type_name());
        let topic = switchboard::get_custom_topic(&custom.data_type);
        msgbus::publish_any(topic, custom);
    }

    /// Drains deferred subscribe/unsubscribe commands pushed by option chain
    /// managers (or any other component) and executes them against the appropriate
    /// data client.
    fn drain_deferred_commands(&mut self) {
        // Loop because expire_series pushes Unsubscribe commands; converges in <= 3 iterations
        loop {
            let commands: VecDeque<DeferredCommand> =
                std::mem::take(&mut *self.deferred_cmd_queue.borrow_mut());

            if commands.is_empty() {
                break;
            }

            for cmd in commands {
                match cmd {
                    DeferredCommand::Subscribe(sub) => {
                        let client = self.get_client(sub.client_id(), sub.venue());
                        if let Some(client) = client {
                            client.execute_subscribe(sub);
                        }
                    }
                    DeferredCommand::Unsubscribe(unsub) => {
                        let client = self.get_client(unsub.client_id(), unsub.venue());
                        if let Some(client) = client {
                            client.execute_unsubscribe(&unsub);
                        }
                    }
                    DeferredCommand::ExpireSeries(series_id) => {
                        self.expire_series(series_id);
                    }
                }
            }
        }
    }

    /// Proactively expires all instruments for a series and tears down the manager.
    ///
    /// `handle_instrument_expired` removes each instrument from the aggregator and pushes
    /// deferred unsubscribe commands. `teardown` then cancels the snapshot timer and clears
    /// the handler lists (the aggregator is already empty at that point).
    fn expire_series(&mut self, series_id: OptionSeriesId) {
        let Some(manager_rc) = self.option_chain_managers.get(&series_id).cloned() else {
            return;
        };

        let instrument_ids: Vec<InstrumentId> = self
            .option_chain_instrument_index
            .iter()
            .filter(|(_, sid)| **sid == series_id)
            .map(|(id, _)| *id)
            .collect();

        for id in &instrument_ids {
            self.option_chain_instrument_index.remove(id);
            manager_rc.borrow_mut().handle_instrument_expired(id);
        }

        manager_rc.borrow_mut().teardown(&self.clock);
        self.option_chain_managers.remove(&series_id);

        log::info!("Proactively torn down expired option chain {series_id}");
    }

    // -- SUBSCRIPTION HANDLERS -------------------------------------------------------------------

    fn subscribe_book_deltas(&mut self, cmd: &SubscribeBookDeltas) -> anyhow::Result<()> {
        if cmd.instrument_id.is_synthetic() {
            anyhow::bail!("Cannot subscribe for synthetic instrument `OrderBookDelta` data");
        }

        self.book_deltas_subs.insert(cmd.instrument_id);
        self.setup_book_updater(&cmd.instrument_id, cmd.book_type, true, cmd.managed)?;

        Ok(())
    }

    fn subscribe_book_depth10(&mut self, cmd: &SubscribeBookDepth10) -> anyhow::Result<()> {
        if cmd.instrument_id.is_synthetic() {
            anyhow::bail!("Cannot subscribe for synthetic instrument `OrderBookDepth10` data");
        }

        self.book_depth10_subs.insert(cmd.instrument_id);
        self.setup_book_updater(&cmd.instrument_id, cmd.book_type, false, cmd.managed)?;

        Ok(())
    }

    fn subscribe_book_snapshots(&mut self, cmd: &SubscribeBookSnapshots) -> anyhow::Result<()> {
        if cmd.instrument_id.is_synthetic() {
            anyhow::bail!("Cannot subscribe for synthetic instrument `OrderBookDelta` data");
        }

        let had_snapshots = self.has_book_snapshot_subscriptions(&cmd.instrument_id);
        let inserted = self.increment_book_snapshot_subscription(cmd);

        if inserted && !had_snapshots && !self.book_deltas_subs.contains(&cmd.instrument_id) {
            self.setup_book_updater(&cmd.instrument_id, cmd.book_type, false, true)?;
        }

        if had_snapshots || self.book_deltas_subs.contains(&cmd.instrument_id) {
            return Ok(());
        }

        if let Some(client_id) = cmd.client_id.as_ref()
            && self.external_clients.contains(client_id)
        {
            if self.config.debug {
                log::debug!("Skipping subscribe command for external client {client_id}: {cmd:?}",);
            }
            return Ok(());
        }

        log::debug!(
            "Forwarding BookSnapshots as BookDeltas for {}, client_id={:?}, venue={:?}",
            cmd.instrument_id,
            cmd.client_id,
            cmd.venue,
        );

        if let Some(client) = self.get_client(cmd.client_id.as_ref(), cmd.venue.as_ref()) {
            let deltas_cmd = SubscribeBookDeltas::new(
                cmd.instrument_id,
                cmd.book_type,
                cmd.client_id,
                cmd.venue,
                UUID4::new(),
                cmd.ts_init,
                cmd.depth,
                true, // managed
                Some(cmd.command_id),
                cmd.params.clone(),
            );
            log::debug!(
                "Calling client.execute_subscribe for BookDeltas: {}",
                cmd.instrument_id
            );
            client.execute_subscribe(SubscribeCommand::BookDeltas(deltas_cmd));
        } else {
            log::error!(
                "Cannot handle command: no client found for client_id={:?}, venue={:?}",
                cmd.client_id,
                cmd.venue,
            );
        }

        Ok(())
    }

    fn subscribe_bars(&mut self, cmd: &SubscribeBars) -> anyhow::Result<()> {
        match cmd.bar_type.aggregation_source() {
            AggregationSource::Internal => {
                if !self.bar_aggregators.contains_key(&cmd.bar_type.standard()) {
                    self.start_bar_aggregator(cmd.bar_type)?;
                }
            }
            AggregationSource::External => {
                if cmd.bar_type.instrument_id().is_synthetic() {
                    anyhow::bail!(
                        "Cannot subscribe for externally aggregated synthetic instrument bar data"
                    );
                }
            }
        }

        Ok(())
    }

    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> bool {
        if !self.book_deltas_subs.contains(&cmd.instrument_id) {
            log::warn!("Cannot unsubscribe from `OrderBookDeltas` data: not subscribed");
            return false;
        }

        self.book_deltas_subs.remove(&cmd.instrument_id);
        self.maintain_book_updater(&cmd.instrument_id);

        // Snapshot subscriptions reuse the deltas feed.
        // Keep the client subscribed until the last snapshot consumer is gone.
        !self.has_book_snapshot_subscriptions(&cmd.instrument_id)
    }

    fn unsubscribe_book_depth10(&mut self, cmd: &UnsubscribeBookDepth10) -> bool {
        if !self.book_depth10_subs.contains(&cmd.instrument_id) {
            log::warn!("Cannot unsubscribe from `OrderBookDepth10` data: not subscribed");
            return false;
        }

        self.book_depth10_subs.remove(&cmd.instrument_id);
        self.maintain_book_updater(&cmd.instrument_id);

        true
    }

    fn unsubscribe_book_snapshots(&mut self, cmd: &UnsubscribeBookSnapshots) {
        match self.decrement_book_snapshot_subscription(cmd.instrument_id, cmd.interval_ms) {
            BookSnapshotUnsubscribeResult::NotSubscribed => {
                log::warn!("Cannot unsubscribe from `OrderBook` snapshots: not subscribed");
                return;
            }
            BookSnapshotUnsubscribeResult::Decremented => return,
            BookSnapshotUnsubscribeResult::Removed => {}
        }

        if self.has_book_snapshot_subscriptions(&cmd.instrument_id) {
            return;
        }

        self.maintain_book_updater(&cmd.instrument_id);

        if self.book_deltas_subs.contains(&cmd.instrument_id) {
            return;
        }

        if let Some(client_id) = cmd.client_id.as_ref()
            && self.external_clients.contains(client_id)
        {
            return;
        }

        if let Some(client) = self.get_client(cmd.client_id.as_ref(), cmd.venue.as_ref()) {
            let deltas_cmd = UnsubscribeBookDeltas::new(
                cmd.instrument_id,
                cmd.client_id,
                cmd.venue,
                UUID4::new(),
                cmd.ts_init,
                Some(cmd.command_id),
                cmd.params.clone(),
            );
            client.execute_unsubscribe(&UnsubscribeCommand::BookDeltas(deltas_cmd));
        }
    }

    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) {
        let bar_type = cmd.bar_type;

        // Don't remove aggregator if other exact-topic subscribers still exist
        let topic = switchboard::get_bars_topic(bar_type.standard());
        if msgbus::exact_subscriber_count_bars(topic) > 0 {
            return;
        }

        if self.bar_aggregators.contains_key(&bar_type.standard())
            && let Err(e) = self.stop_bar_aggregator(bar_type)
        {
            log::error!("Error stopping bar aggregator for {bar_type}: {e}");
        }

        // After stopping a composite, check if the source aggregator is now orphaned
        if bar_type.is_composite() {
            let source_type = bar_type.composite();
            let source_topic = switchboard::get_bars_topic(source_type);
            if msgbus::exact_subscriber_count_bars(source_topic) == 0
                && self.bar_aggregators.contains_key(&source_type)
                && let Err(e) = self.stop_bar_aggregator(source_type)
            {
                log::error!("Error stopping source bar aggregator for {source_type}: {e}");
            }
        }
    }

    fn subscribe_option_chain(&mut self, cmd: &SubscribeOptionChain) {
        let series_id = cmd.series_id;

        // Handle edits to existing subscriptions by tearing down and re-setting up the OptionChainManager.
        if let Some(old) = self.option_chain_managers.remove(&series_id) {
            log::info!("Re-subscribing option chain for {series_id}, tearing down previous");
            let all_ids = old.borrow().all_instrument_ids();
            let old_venue = old.borrow().venue();
            old.borrow_mut().teardown(&self.clock);
            self.forward_option_chain_unsubscribes(&all_ids, old_venue, cmd.client_id);
        }

        // Drain any stale pending forward price requests for this series
        self.pending_option_chain_requests
            .retain(|_, pending_cmd| pending_cmd.series_id != series_id);

        // For ATM-based strike ranges, request forward prices from the adapter
        // to enable instant bootstrap without waiting for the first WebSocket tick.
        if !matches!(cmd.strike_range, StrikeRange::Fixed(_)) {
            // Extract client_id first to avoid borrow conflicts
            let resolved_client_id = self
                .get_client(cmd.client_id.as_ref(), Some(&series_id.venue))
                .map(|c| c.client_id);

            if let Some(client_id) = resolved_client_id {
                let request_id = UUID4::new();
                let ts_init = self.clock.borrow().timestamp_ns();

                // Pick any one option instrument at this expiry from cache
                // to enable single-instrument forward price fetch (1 HTTP call)
                let sample_instrument_id = {
                    let cache = self.cache.borrow();
                    cache
                        .instruments(&series_id.venue, Some(&series_id.underlying))
                        .iter()
                        .find(|i| {
                            i.expiration_ns() == Some(series_id.expiration_ns)
                                && i.settlement_currency().code == series_id.settlement_currency
                        })
                        .map(|i| i.id())
                };

                let request = RequestForwardPrices::new(
                    series_id.venue,
                    series_id.underlying,
                    sample_instrument_id,
                    Some(client_id),
                    request_id,
                    ts_init,
                    None,
                );

                self.pending_option_chain_requests
                    .insert(request_id, cmd.clone());

                let req_cmd = RequestCommand::ForwardPrices(request);
                if let Err(e) = self.execute_request(req_cmd) {
                    log::warn!("Failed to request forward prices for {series_id}: {e}");
                    let cmd = self
                        .pending_option_chain_requests
                        .remove(&request_id)
                        .expect("just inserted");
                    self.create_option_chain_manager(&cmd, None);
                }

                return;
            }
        }

        self.create_option_chain_manager(cmd, None);
    }

    /// Creates and stores an `OptionChainManager` for the given subscription.
    fn create_option_chain_manager(
        &mut self,
        cmd: &SubscribeOptionChain,
        initial_atm_price: Option<Price>,
    ) {
        let series_id = cmd.series_id;
        let cache = self.cache.clone();
        let clock = self.clock.clone();
        let priority = self.msgbus_priority;
        let deferred_cmd_queue = self.deferred_cmd_queue.clone();

        let manager_rc = {
            let client = self.get_client(cmd.client_id.as_ref(), Some(&series_id.venue));
            OptionChainManager::create_and_setup(
                series_id,
                &cache,
                cmd,
                &clock,
                priority,
                client,
                initial_atm_price,
                deferred_cmd_queue,
            )
        };

        // Index all instruments for reverse lookup
        for id in manager_rc.borrow().all_instrument_ids() {
            self.option_chain_instrument_index.insert(id, series_id);
        }

        self.option_chain_managers.insert(series_id, manager_rc);
    }

    fn unsubscribe_option_chain(&mut self, cmd: &UnsubscribeOptionChain) {
        let series_id = cmd.series_id;

        let Some(manager_rc) = self.option_chain_managers.remove(&series_id) else {
            log::warn!("Cannot unsubscribe option chain for {series_id}: not subscribed");
            return;
        };

        // Extract info before teardown
        let all_ids = manager_rc.borrow().all_instrument_ids();
        let venue = manager_rc.borrow().venue();

        // Remove all instruments from reverse index
        for id in &all_ids {
            self.option_chain_instrument_index.remove(id);
        }

        manager_rc.borrow_mut().teardown(&self.clock);

        // Forward wire-level unsubscribes to the data client
        self.forward_option_chain_unsubscribes(&all_ids, venue, cmd.client_id);

        log::info!("Unsubscribed option chain for {series_id}");
    }

    /// Forwards wire-level unsubscribe commands for all option chain instruments.
    fn forward_option_chain_unsubscribes(
        &mut self,
        instrument_ids: &[InstrumentId],
        venue: Venue,
        client_id: Option<ClientId>,
    ) {
        let ts_init = self.clock.borrow().timestamp_ns();

        let Some(client) = self.get_client(client_id.as_ref(), Some(&venue)) else {
            log::error!(
                "Cannot forward option chain unsubscribes: no client found for venue={venue}",
            );
            return;
        };

        for instrument_id in instrument_ids {
            client.execute_unsubscribe(&UnsubscribeCommand::Quotes(UnsubscribeQuotes::new(
                *instrument_id,
                client_id,
                Some(venue),
                UUID4::new(),
                ts_init,
                None,
                None,
            )));
            client.execute_unsubscribe(&UnsubscribeCommand::OptionGreeks(
                UnsubscribeOptionGreeks::new(
                    *instrument_id,
                    client_id,
                    Some(venue),
                    UUID4::new(),
                    ts_init,
                    None,
                    None,
                ),
            ));
            client.execute_unsubscribe(&UnsubscribeCommand::InstrumentStatus(
                UnsubscribeInstrumentStatus::new(
                    *instrument_id,
                    client_id,
                    Some(venue),
                    UUID4::new(),
                    ts_init,
                    None,
                    None,
                ),
            ));
        }
    }

    fn maintain_book_updater(&mut self, instrument_id: &InstrumentId) {
        let Some(updater) = self.book_updaters.get(instrument_id) else {
            return;
        };

        let has_deltas = self.book_deltas_subs.contains(instrument_id);
        let has_depth10 = self.book_depth10_subs.contains(instrument_id);
        let has_snapshots = self.has_book_snapshot_subscriptions(instrument_id);

        let deltas_topic = switchboard::get_book_deltas_topic(*instrument_id);
        let depth_topic = switchboard::get_book_depth10_topic(*instrument_id);
        let deltas_handler: TypedHandler<OrderBookDeltas> = TypedHandler::new(updater.clone());
        let depth_handler: TypedHandler<OrderBookDepth10> = TypedHandler::new(updater.clone());

        if !has_deltas && !has_snapshots {
            msgbus::unsubscribe_book_deltas(deltas_topic.into(), &deltas_handler);
        }

        if !has_depth10 {
            msgbus::unsubscribe_book_depth10(depth_topic.into(), &depth_handler);
        }

        if !has_deltas && !has_depth10 && !has_snapshots {
            self.book_updaters.remove(instrument_id);
            log::debug!("Removed BookUpdater for instrument ID {instrument_id}");
        }
    }

    fn has_book_snapshot_subscriptions(&self, instrument_id: &InstrumentId) -> bool {
        self.book_snapshot_counts
            .keys()
            .any(|(id, _)| id == instrument_id)
    }

    fn increment_book_snapshot_subscription(&mut self, cmd: &SubscribeBookSnapshots) -> bool {
        let key = (cmd.instrument_id, cmd.interval_ms);

        if let Some(count) = self.book_snapshot_counts.get_mut(&key) {
            *count += 1;
            return false;
        }

        self.book_snapshot_counts.insert(key, 1);

        let snapshot_infos = if let Some(snapshot_infos) = self.book_intervals.get(&cmd.interval_ms)
        {
            snapshot_infos.clone()
        } else {
            let snapshot_infos = Rc::new(RefCell::new(IndexMap::new()));
            self.book_intervals
                .insert(cmd.interval_ms, snapshot_infos.clone());
            self.schedule_book_snapshotter(cmd.interval_ms, snapshot_infos.clone());
            snapshot_infos
        };

        let topic = switchboard::get_book_snapshots_topic(cmd.instrument_id, cmd.interval_ms);
        let snap_info = BookSnapshotInfo {
            instrument_id: cmd.instrument_id,
            venue: cmd.instrument_id.venue,
            is_composite: cmd.instrument_id.symbol.is_composite(),
            root: Ustr::from(cmd.instrument_id.symbol.root()),
            topic,
            interval_ms: cmd.interval_ms,
        };

        snapshot_infos
            .borrow_mut()
            .insert(cmd.instrument_id, snap_info);

        true
    }

    fn decrement_book_snapshot_subscription(
        &mut self,
        instrument_id: InstrumentId,
        interval_ms: NonZeroUsize,
    ) -> BookSnapshotUnsubscribeResult {
        let key = (instrument_id, interval_ms);

        let Some(count) = self.book_snapshot_counts.get_mut(&key) else {
            return BookSnapshotUnsubscribeResult::NotSubscribed;
        };

        if *count > 1 {
            *count -= 1;
            return BookSnapshotUnsubscribeResult::Decremented;
        }

        self.book_snapshot_counts.shift_remove(&key);

        let remove_interval = if let Some(snapshot_infos) = self.book_intervals.get(&interval_ms) {
            let mut snapshot_infos = snapshot_infos.borrow_mut();
            snapshot_infos.shift_remove(&instrument_id);
            snapshot_infos.is_empty()
        } else {
            false
        };

        if remove_interval {
            self.book_intervals.remove(&interval_ms);

            if let Some(snapshotter) = self.book_snapshotters.remove(&interval_ms) {
                let timer_name = snapshotter.timer_name;
                let mut clock = self.clock.borrow_mut();
                if clock.timer_exists(&timer_name) {
                    clock.cancel_timer(&timer_name);
                }
            }
        }

        BookSnapshotUnsubscribeResult::Removed
    }

    fn schedule_book_snapshotter(
        &mut self,
        interval_ms: NonZeroUsize,
        snapshot_infos: BookSnapshotInfos,
    ) {
        let interval_ns = millis_to_nanos_unchecked(interval_ms.get() as f64);
        let now_ns = self.clock.borrow().timestamp_ns().as_u64();
        let start_time_ns = now_ns - (now_ns % interval_ns) + interval_ns;

        let snapshotter = Rc::new(BookSnapshotter::new(
            interval_ms,
            snapshot_infos,
            self.cache.clone(),
        ));
        let timer_name = snapshotter.timer_name;
        let snapshotter_callback = snapshotter.clone();
        let callback_fn: Rc<dyn Fn(TimeEvent)> =
            Rc::new(move |event| snapshotter_callback.snapshot(event));
        let callback = TimeEventCallback::from(callback_fn);

        self.clock
            .borrow_mut()
            .set_timer_ns(
                &timer_name,
                interval_ns,
                Some(start_time_ns.into()),
                None,
                Some(callback),
                None,
                None,
            )
            .expect(FAILED);

        self.book_snapshotters.insert(interval_ms, snapshotter);
    }

    // -- RESPONSE HANDLERS -----------------------------------------------------------------------

    fn handle_instrument_response(&self, instrument: InstrumentAny) {
        let mut cache = self.cache.as_ref().borrow_mut();
        if let Err(e) = cache.add_instrument(instrument) {
            log_error_on_cache_insert(&e);
        }
    }

    fn handle_instruments(&self, instruments: &[InstrumentAny]) {
        // TODO: Improve by adding bulk update methods to cache and database
        let mut cache = self.cache.as_ref().borrow_mut();
        for instrument in instruments {
            if let Err(e) = cache.add_instrument(instrument.clone()) {
                log_error_on_cache_insert(&e);
            }
        }
    }

    fn handle_quotes(&self, quotes: &[QuoteTick]) {
        if let Err(e) = self.cache.as_ref().borrow_mut().add_quotes(quotes) {
            log_error_on_cache_insert(&e);
        }
    }

    fn handle_trades(&self, trades: &[TradeTick]) {
        if let Err(e) = self.cache.as_ref().borrow_mut().add_trades(trades) {
            log_error_on_cache_insert(&e);
        }
    }

    fn handle_funding_rates(&self, funding_rates: &[FundingRateUpdate]) {
        if let Err(e) = self
            .cache
            .as_ref()
            .borrow_mut()
            .add_funding_rates(funding_rates)
        {
            log_error_on_cache_insert(&e);
        }
    }

    fn handle_bars(&self, bars: &[Bar]) {
        if let Err(e) = self.cache.as_ref().borrow_mut().add_bars(bars) {
            log_error_on_cache_insert(&e);
        }
    }

    fn handle_book_response(&self, book: &OrderBook) {
        log::debug!("Adding order book {} to cache", book.instrument_id);

        if let Err(e) = self
            .cache
            .as_ref()
            .borrow_mut()
            .add_order_book(book.clone())
        {
            log_error_on_cache_insert(&e);
        }
    }

    /// Handles a `ForwardPricesResponse` by extracting the forward price
    /// for the pending option chain and creating the manager with instant bootstrap.
    fn handle_forward_prices_response(
        &mut self,
        correlation_id: &UUID4,
        resp: &ForwardPricesResponse,
    ) {
        let Some(cmd) = self.pending_option_chain_requests.remove(correlation_id) else {
            log::debug!(
                "No pending option chain request for correlation_id={correlation_id}, ignoring"
            );
            return;
        };

        let series_id = cmd.series_id;

        // Find a forward price that matches an instrument in this series.
        // We look up each forward price instrument in the cache to match by expiry and currency.
        let cache = self.cache.borrow();
        let mut best_price: Option<Price> = None;

        for fp in &resp.data {
            // Check if any cached instrument with this id belongs to our series
            if let Some(instrument) = cache.instrument(&fp.instrument_id)
                && let Some(expiration) = instrument.expiration_ns()
                && expiration == series_id.expiration_ns
                && instrument.settlement_currency().code == series_id.settlement_currency
            {
                match Price::from_decimal(fp.forward_price) {
                    Ok(price) => best_price = Some(price),
                    Err(e) => log::warn!("Invalid forward price for {}: {e}", fp.instrument_id),
                }
                break;
            }
        }
        drop(cache);

        if let Some(price) = best_price {
            log::info!("Forward price for {series_id}: {price} (instant bootstrap)",);
        } else {
            log::info!(
                "No matching forward price found for {series_id}, will bootstrap from live data",
            );
        }

        self.create_option_chain_manager(&cmd, best_price);
    }

    // -- INTERNAL --------------------------------------------------------------------------------

    fn setup_book_updater(
        &mut self,
        instrument_id: &InstrumentId,
        book_type: BookType,
        only_deltas: bool,
        managed: bool,
    ) -> anyhow::Result<()> {
        let mut cache = self.cache.borrow_mut();
        if managed && !cache.has_order_book(instrument_id) {
            let book = OrderBook::new(*instrument_id, book_type);
            log::debug!("Created {book}");
            cache.add_order_book(book)?;
        }

        // Reuse existing BookUpdater or create a new one
        let updater = self
            .book_updaters
            .entry(*instrument_id)
            .or_insert_with(|| Rc::new(BookUpdater::new(instrument_id, self.cache.clone())))
            .clone();

        // Subscribe to deltas (typed router handles duplicates)
        let topic = switchboard::get_book_deltas_topic(*instrument_id);
        let deltas_handler = TypedHandler::new(updater.clone());
        msgbus::subscribe_book_deltas(topic.into(), deltas_handler, Some(self.msgbus_priority));

        // Subscribe to depth10 if not only_deltas
        if !only_deltas {
            let topic = switchboard::get_book_depth10_topic(*instrument_id);
            let depth_handler = TypedHandler::new(updater);
            msgbus::subscribe_book_depth10(topic.into(), depth_handler, Some(self.msgbus_priority));
        }

        Ok(())
    }

    fn create_bar_aggregator(
        &self,
        instrument: &InstrumentAny,
        bar_type: BarType,
    ) -> Box<dyn BarAggregator> {
        let cache = self.cache.clone();

        let handler = move |bar: Bar| {
            if let Err(e) = cache.as_ref().borrow_mut().add_bar(bar) {
                log_error_on_cache_insert(&e);
            }

            let topic = switchboard::get_bars_topic(bar.bar_type);
            msgbus::publish_bar(topic, &bar);
        };

        let clock = self.clock.clone();
        let config = self.config.clone();

        let price_precision = instrument.price_precision();
        let size_precision = instrument.size_precision();

        if bar_type.spec().is_time_aggregated() {
            // Get time_bars_origin_offset from config
            let time_bars_origin_offset = config
                .time_bars_origins
                .get(&bar_type.spec().aggregation)
                .map(|duration| chrono::TimeDelta::from_std(*duration).unwrap_or_default());

            Box::new(TimeBarAggregator::new(
                bar_type,
                price_precision,
                size_precision,
                clock,
                handler,
                config.time_bars_build_with_no_updates,
                config.time_bars_timestamp_on_close,
                config.time_bars_interval_type,
                time_bars_origin_offset,
                config.time_bars_build_delay,
                config.time_bars_skip_first_non_full_bar,
            ))
        } else {
            match bar_type.spec().aggregation {
                BarAggregation::Tick => Box::new(TickBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    handler,
                )) as Box<dyn BarAggregator>,
                BarAggregation::TickImbalance => Box::new(TickImbalanceBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    handler,
                )) as Box<dyn BarAggregator>,
                BarAggregation::TickRuns => Box::new(TickRunsBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    handler,
                )) as Box<dyn BarAggregator>,
                BarAggregation::Volume => Box::new(VolumeBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    handler,
                )) as Box<dyn BarAggregator>,
                BarAggregation::VolumeImbalance => Box::new(VolumeImbalanceBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    handler,
                )) as Box<dyn BarAggregator>,
                BarAggregation::VolumeRuns => Box::new(VolumeRunsBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    handler,
                )) as Box<dyn BarAggregator>,
                BarAggregation::Value => Box::new(ValueBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    handler,
                )) as Box<dyn BarAggregator>,
                BarAggregation::ValueImbalance => Box::new(ValueImbalanceBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    handler,
                )) as Box<dyn BarAggregator>,
                BarAggregation::ValueRuns => Box::new(ValueRunsBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    handler,
                )) as Box<dyn BarAggregator>,
                BarAggregation::Renko => Box::new(RenkoBarAggregator::new(
                    bar_type,
                    price_precision,
                    size_precision,
                    instrument.price_increment(),
                    handler,
                )) as Box<dyn BarAggregator>,
                other => unreachable!(
                    "Unsupported internal bar aggregation dispatch for {other:?}; update `create_bar_aggregator`"
                ),
            }
        }
    }

    fn start_bar_aggregator(&mut self, bar_type: BarType) -> anyhow::Result<()> {
        // Get the instrument for this bar type
        let instrument = {
            let cache = self.cache.borrow();
            cache
                .instrument(&bar_type.instrument_id())
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "Cannot start bar aggregation: no instrument found for {}",
                        bar_type.instrument_id(),
                    )
                })?
                .clone()
        };

        // Use standard form of bar type as key
        let bar_key = bar_type.standard();

        // Create or retrieve aggregator in Rc<RefCell>
        let aggregator = if let Some(rc) = self.bar_aggregators.get(&bar_key) {
            rc.clone()
        } else {
            let agg = self.create_bar_aggregator(&instrument, bar_type);
            let rc = Rc::new(RefCell::new(agg));
            self.bar_aggregators.insert(bar_key, rc.clone());
            rc
        };

        // Subscribe to underlying data topics
        let mut subscriptions = Vec::new();

        if bar_type.is_composite() {
            let topic = switchboard::get_bars_topic(bar_type.composite());
            let handler = TypedHandler::new(BarBarHandler::new(&aggregator, bar_key));
            msgbus::subscribe_bars(topic.into(), handler.clone(), Some(self.msgbus_priority));
            subscriptions.push(BarAggregatorSubscription::Bar { topic, handler });
        } else if bar_type.spec().price_type == PriceType::Last {
            let topic = switchboard::get_trades_topic(bar_type.instrument_id());
            let handler = TypedHandler::new(BarTradeHandler::new(&aggregator, bar_key));
            msgbus::subscribe_trades(topic.into(), handler.clone(), Some(self.msgbus_priority));
            subscriptions.push(BarAggregatorSubscription::Trade { topic, handler });
        } else {
            // Warn if imbalance/runs aggregation is wired to quotes (needs aggressor_side from trades)
            if matches!(
                bar_type.spec().aggregation,
                BarAggregation::TickImbalance
                    | BarAggregation::VolumeImbalance
                    | BarAggregation::ValueImbalance
                    | BarAggregation::TickRuns
                    | BarAggregation::VolumeRuns
                    | BarAggregation::ValueRuns
            ) {
                log::warn!(
                    "Bar type {bar_type} uses imbalance/runs aggregation which requires trade \
                     data with `aggressor_side`, but `price_type` is not LAST so it will receive \
                     quote data: bars will not emit correctly",
                );
            }

            let topic = switchboard::get_quotes_topic(bar_type.instrument_id());
            let handler = TypedHandler::new(BarQuoteHandler::new(&aggregator, bar_key));
            msgbus::subscribe_quotes(topic.into(), handler.clone(), Some(self.msgbus_priority));
            subscriptions.push(BarAggregatorSubscription::Quote { topic, handler });
        }

        self.bar_aggregator_handlers.insert(bar_key, subscriptions);

        // Setup time bar aggregator if needed (matches Cython _setup_bar_aggregator)
        self.setup_bar_aggregator(bar_type, false)?;

        aggregator.borrow_mut().set_is_running(true);

        Ok(())
    }

    /// Sets up a bar aggregator, matching Cython _setup_bar_aggregator logic.
    ///
    /// This method handles historical mode, message bus subscriptions, and time bar aggregator setup.
    fn setup_bar_aggregator(&self, bar_type: BarType, historical: bool) -> anyhow::Result<()> {
        let bar_key = bar_type.standard();
        let aggregator = self.bar_aggregators.get(&bar_key).ok_or_else(|| {
            anyhow::anyhow!("Cannot setup bar aggregator: no aggregator found for {bar_type}")
        })?;

        // Set historical mode and handler
        let handler: Box<dyn FnMut(Bar)> = if historical {
            // Historical handler - process_historical equivalent
            let cache = self.cache.clone();
            Box::new(move |bar: Bar| {
                if let Err(e) = cache.as_ref().borrow_mut().add_bar(bar) {
                    log_error_on_cache_insert(&e);
                }
                // In historical mode, bars are processed but not published to message bus
            })
        } else {
            // Regular handler - process equivalent
            let cache = self.cache.clone();
            Box::new(move |bar: Bar| {
                if let Err(e) = cache.as_ref().borrow_mut().add_bar(bar) {
                    log_error_on_cache_insert(&e);
                }
                let topic = switchboard::get_bars_topic(bar.bar_type);
                msgbus::publish_bar(topic, &bar);
            })
        };

        aggregator
            .borrow_mut()
            .set_historical_mode(historical, handler);

        // For TimeBarAggregator, set clock and start timer
        if bar_type.spec().is_time_aggregated() {
            use nautilus_common::clock::TestClock;

            if historical {
                // Each aggregator gets its own independent clock
                let test_clock = Rc::new(RefCell::new(TestClock::new()));
                aggregator.borrow_mut().set_clock(test_clock);
                // Set weak reference for historical mode (start_timer called later from preprocess_historical_events)
                // Store weak reference so start_timer can use it when called later
                let aggregator_weak = Rc::downgrade(aggregator);
                aggregator.borrow_mut().set_aggregator_weak(aggregator_weak);
            } else {
                aggregator.borrow_mut().set_clock(self.clock.clone());
                aggregator
                    .borrow_mut()
                    .start_timer(Some(aggregator.clone()));
            }
        }

        Ok(())
    }

    fn stop_bar_aggregator(&mut self, bar_type: BarType) -> anyhow::Result<()> {
        let aggregator = self
            .bar_aggregators
            .shift_remove(&bar_type.standard())
            .ok_or_else(|| {
                anyhow::anyhow!("Cannot stop bar aggregator: no aggregator to stop for {bar_type}")
            })?;

        aggregator.borrow_mut().stop();

        // Unsubscribe any registered message handlers
        let bar_key = bar_type.standard();
        if let Some(subs) = self.bar_aggregator_handlers.remove(&bar_key) {
            for sub in subs {
                match sub {
                    BarAggregatorSubscription::Bar { topic, handler } => {
                        msgbus::unsubscribe_bars(topic.into(), &handler);
                    }
                    BarAggregatorSubscription::Trade { topic, handler } => {
                        msgbus::unsubscribe_trades(topic.into(), &handler);
                    }
                    BarAggregatorSubscription::Quote { topic, handler } => {
                        msgbus::unsubscribe_quotes(topic.into(), &handler);
                    }
                }
            }
        }

        Ok(())
    }
}

#[inline(always)]
fn log_error_on_cache_insert<T: Display>(e: &T) {
    log::error!("Error on cache insert: {e}");
}

#[inline(always)]
fn log_if_empty_response<T, I: Display>(data: &[T], id: &I, correlation_id: &UUID4) -> bool {
    if data.is_empty() {
        let name = type_name::<T>();
        let short_name = name.rsplit("::").next().unwrap_or(name);
        log::warn!("Received empty {short_name} response for {id} {correlation_id}");
        return true;
    }
    false
}