nautilus_interactive_brokers/data/

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

//! Core data client implementation for Interactive Brokers.

use std::{
    collections::HashMap,
    fmt::Debug,
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use ahash::AHashMap;
use anyhow::Context;
use ibapi::{
    contracts::{Contract, Currency as IBCurrency, Exchange as IBExchange, SecurityType, Symbol},
    market_data::IgnoreSize,
    prelude::{StreamExt, SubscriptionItemStreamExt},
};
use nautilus_common::{
    clients::DataClient,
    live::{runner::get_data_event_sender, sender::EventSender},
    messages::{
        DataEvent, DataResponse,
        data::{
            BarsResponse, InstrumentResponse, InstrumentsResponse, QuotesResponse, RequestBars,
            RequestInstrument, RequestInstruments, RequestQuotes, RequestTrades, SubscribeBars,
            SubscribeBookDeltas, SubscribeIndexPrices, SubscribeOptionGreeks, SubscribeQuotes,
            SubscribeTrades, TradesResponse, UnsubscribeBars, UnsubscribeBookDeltas,
            UnsubscribeIndexPrices, UnsubscribeOptionGreeks, UnsubscribeQuotes, UnsubscribeTrades,
        },
    },
};
use nautilus_core::{
    UnixNanos,
    params::Params,
    time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_live::task::{TaskGroup, TaskJoinOutcome, TaskSlot, finish_task};
use nautilus_model::{
    data::BarType,
    enums::BookType,
    identifiers::{ClientId, InstrumentId, Venue},
    instruments::{Instrument, any::InstrumentAny},
};
use tokio_util::sync::CancellationToken;

use super::{
    cache::{OptionGreeksCache, QuoteCache},
    convert::{
        apply_bar_price_magnifier, apply_price_magnifier, bar_request_segments,
        bar_type_to_ib_bar_size, calculate_duration_segments, extend_historical_tick_batch,
        ib_bar_to_nautilus_bar, jiff_to_ib_datetime,
        price_type_to_ib_realtime_what_to_show_for_security,
        price_type_to_ib_what_to_show_for_security, retain_historical_ticks_in_range,
        should_continue_historical_tick_pagination,
    },
    parse::log_tick_parse_error,
    streams::{
        DataFarmConnectionState, StreamConfig, handle_historical_bars_subscription,
        handle_index_price_subscription, handle_market_depth_subscription,
        handle_option_greeks_subscription, handle_quote_subscription,
        handle_realtime_bars_subscription, handle_tick_by_tick_quote_subscription,
        handle_trade_subscription, monitor_data_farm_notices,
    },
};
use crate::{
    common::{
        consts::IB_VENUE, contracts::parse_contract_from_json, shared_client::SharedClientHandle,
        symbology::is_crypto_contract,
    },
    config::{InteractiveBrokersDataClientConfig, MarketDataType},
    data_types::register_ib_custom_data,
    providers::instruments::InteractiveBrokersInstrumentProvider,
};

/// Interactive Brokers data client.
///
/// This client provides market data functionality using the `rust-ibapi` library.
/// It manages subscriptions, handles historical data requests, and streams
/// market data to NautilusTrader.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.adapters.interactive_brokers")
)]
pub struct InteractiveBrokersDataClient {
    client_id: ClientId,
    config: InteractiveBrokersDataClientConfig,
    instrument_provider: Arc<InteractiveBrokersInstrumentProvider>,
    is_connected: Arc<AtomicBool>,
    cancellation_token: CancellationToken,
    session_tasks: TaskGroup,
    command_tasks: TaskGroup,
    data_sender: EventSender<DataEvent>,
    subscriptions: Arc<Mutex<AHashMap<SubscriptionKey, SubscriptionInfo>>>,
    option_greeks_subscriptions: Arc<Mutex<AHashMap<InstrumentId, CancellationToken>>>,
    quote_cache: Arc<tokio::sync::Mutex<QuoteCache>>,
    option_greeks_cache: Arc<tokio::sync::Mutex<OptionGreeksCache>>,
    clock: &'static AtomicTime,
    ib_client: Option<SharedClientHandle>,
    last_bars: Arc<tokio::sync::Mutex<AHashMap<String, ibapi::market_data::realtime::Bar>>>,
    bar_timeout_tasks: Arc<tokio::sync::Mutex<AHashMap<String, TaskSlot<()>>>>,
    data_farm_state: Arc<DataFarmConnectionState>,
}

/// Information about an active subscription.
#[derive(Debug)]
struct SubscriptionInfo {
    /// Cancellation token for this specific subscription.
    cancellation_token: CancellationToken,
    /// Unique id distinguishing this registration from later ones for the same key.
    id: u64,
}

/// Type of subscription.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum SubscriptionType {
    /// Quote subscription.
    Quotes,
    /// Index price subscription.
    IndexPrices,
    /// Trade subscription.
    Trades,
    /// Bar subscription, one stream per bar type.
    Bars(BarType),
    /// Order book delta subscription.
    BookDeltas,
}

type SubscriptionKey = (InstrumentId, SubscriptionType);

static SUBSCRIPTION_ID_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

fn next_subscription_id() -> u64 {
    SUBSCRIPTION_ID_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
}

/// Removes a subscription's registry entry when its stream task ends permanently,
/// so a later re-subscribe is not blocked by a dead registration. The id check
/// keeps a finishing task from removing a newer registration for the same key.
struct SubscriptionGuard {
    subscriptions: Arc<Mutex<AHashMap<SubscriptionKey, SubscriptionInfo>>>,
    key: SubscriptionKey,
    id: u64,
}

impl SubscriptionGuard {
    fn release(self) {
        // Dropping performs the removal; a panicking stream task reaches the same
        // path through Drop, so a dead registration can never block a re-subscribe.
    }
}

impl Drop for SubscriptionGuard {
    fn drop(&mut self) {
        if let Ok(mut subscriptions) = self.subscriptions.lock()
            && subscriptions
                .get(&self.key)
                .is_some_and(|info| info.id == self.id)
        {
            subscriptions.remove(&self.key);
        }
    }
}

fn lock<T>(mutex: &Mutex<T>) -> anyhow::Result<std::sync::MutexGuard<'_, T>> {
    mutex
        .lock()
        .map_err(|_| anyhow::anyhow!("Failed to lock IB data client state"))
}

fn parse_start_ns(params: Option<&nautilus_core::Params>) -> Option<UnixNanos> {
    params
        .and_then(|params| params.get_u64("start_ns"))
        .or_else(|| {
            params
                .and_then(|params| params.get_str("start_ns"))
                .and_then(|value| value.parse::<u64>().ok())
        })
        .map(UnixNanos::from)
}

fn parse_bool_param_value(value: &str) -> bool {
    matches!(value, "true" | "True" | "1")
}

fn params_to_string_filters(params: Option<&Params>) -> Option<HashMap<String, String>> {
    let filters: HashMap<String, String> = params?
        .iter()
        .filter_map(|(key, value)| value.as_str().map(|value| (key.clone(), value.to_string())))
        .collect();
    (!filters.is_empty()).then_some(filters)
}

fn datetime_to_unix_nanos(dt: jiff::Timestamp) -> UnixNanos {
    UnixNanos::from(u64::try_from(dt.as_nanosecond()).unwrap_or_default())
}

fn request_trading_hours(use_regular_trading_hours: bool) -> ibapi::market_data::TradingHours {
    if use_regular_trading_hours {
        ibapi::market_data::TradingHours::Regular
    } else {
        ibapi::market_data::TradingHours::Extended
    }
}

const HISTORICAL_TICK_DEFAULT_LIMIT: usize = 10_000;

impl InteractiveBrokersDataClient {
    /// Create a new `InteractiveBrokersDataClient`.
    ///
    /// # Errors
    ///
    /// Returns an error if client creation fails.
    pub fn new(
        client_id: ClientId,
        config: InteractiveBrokersDataClientConfig,
        instrument_provider: Arc<InteractiveBrokersInstrumentProvider>,
    ) -> anyhow::Result<Self> {
        let clock = get_atomic_clock_realtime();
        config.validate()?;
        register_ib_custom_data();
        let data_sender = get_data_event_sender();

        let session_tasks = TaskGroup::new();
        let command_tasks = TaskGroup::new();

        Ok(Self {
            client_id,
            config,
            instrument_provider,
            is_connected: Arc::new(AtomicBool::new(false)),
            cancellation_token: session_tasks.cancellation_token(),
            session_tasks,
            command_tasks,
            data_sender,
            subscriptions: Arc::new(Mutex::new(AHashMap::new())),
            option_greeks_subscriptions: Arc::new(Mutex::new(AHashMap::new())),
            quote_cache: Arc::new(tokio::sync::Mutex::new(QuoteCache::new())),
            option_greeks_cache: Arc::new(tokio::sync::Mutex::new(OptionGreeksCache::new())),
            clock,
            ib_client: None,
            last_bars: Arc::new(tokio::sync::Mutex::new(AHashMap::new())),
            bar_timeout_tasks: Arc::new(tokio::sync::Mutex::new(AHashMap::new())),
            data_farm_state: Arc::new(DataFarmConnectionState::default()),
        })
    }

    fn cancel_active_subscriptions(&self) -> anyhow::Result<()> {
        {
            let mut subscriptions = lock(&self.subscriptions)?;
            for subscription in subscriptions.values() {
                subscription.cancellation_token.cancel();
            }

            subscriptions.clear();
        }

        {
            let mut subscriptions = lock(&self.option_greeks_subscriptions)?;
            for cancellation_token in subscriptions.values() {
                cancellation_token.cancel();
            }

            subscriptions.clear();
        }

        Ok(())
    }

    fn spawn_command<F>(&self, future: F)
    where
        F: std::future::Future<Output = ()> + Send + 'static,
    {
        if let Err(e) = self.command_tasks.spawn(future) {
            tracing::warn!("Skipping IB data command after shutdown began: {e}");
        }
    }

    async fn finish_tasks(&self) -> anyhow::Result<()> {
        let (session_result, command_result) = tokio::join!(
            self.session_tasks
                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
            self.command_tasks
                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
        );
        self.finish_bar_timeout_tasks().await;
        session_result.context("failed to finish IB data session tasks")?;
        command_result.context("failed to finish IB data command tasks")?;
        Ok(())
    }

    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
        if !self.session_tasks.is_open() || !self.command_tasks.is_open() {
            self.session_tasks.begin_shutdown();
            self.command_tasks.begin_shutdown();
            self.finish_tasks().await?;
            self.session_tasks
                .start_generation()
                .context("failed to start IB data session task generation")?;
            self.command_tasks
                .start_generation()
                .context("failed to start IB data command task generation")?;
            self.cancellation_token = self.session_tasks.cancellation_token();
        }

        Ok(())
    }

    async fn finish_bar_timeout_tasks(&self) {
        let mut tasks = self.bar_timeout_tasks.lock().await;
        let bar_types = tasks.keys().cloned().collect::<Vec<_>>();
        for bar_type in bar_types {
            let handle = tasks
                .get_mut(&bar_type)
                .expect("bar timeout task key collected from map");

            match finish_task(handle, Duration::ZERO, Duration::from_secs(1)).await {
                None => {}
                Some(TaskJoinOutcome::Completed(()) | TaskJoinOutcome::Aborted) => {}
                Some(TaskJoinOutcome::Failed(e)) => {
                    tracing::warn!("IB bar timeout task join failed: {e}");
                }
                Some(TaskJoinOutcome::Incomplete) => {
                    tracing::warn!("IB bar timeout task did not stop within one second");
                }
            }

            if handle.is_none() {
                tasks.remove(&bar_type);
            }
        }
    }

    /// Get a reference to the IB client if connected.
    /// This is used internally for provider method calls.
    #[allow(dead_code)] // Library API - may be used by other modules or PyO3 bindings
    pub(crate) fn get_ib_client(&self) -> Option<&Arc<ibapi::Client>> {
        self.ib_client.as_ref().map(|h| h.as_arc())
    }

    /// Get a reference to the instrument provider.
    #[allow(dead_code)] // Library API - may be used by other modules or PyO3 bindings
    pub(crate) fn instrument_provider(&self) -> Arc<InteractiveBrokersInstrumentProvider> {
        Arc::clone(&self.instrument_provider)
    }

    /// Batch load multiple instrument IDs using the internal IB client.
    ///
    /// This method calls the provider's batch_load with the data client's IB client.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The client is not connected.
    /// - The provider batch_load fails.
    pub async fn batch_load_instruments(
        &self,
        instrument_ids: Vec<InstrumentId>,
    ) -> anyhow::Result<Vec<InstrumentId>> {
        log::debug!(
            "Batch loading {} IB instruments through data client",
            instrument_ids.len()
        );
        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        let loaded = self
            .instrument_provider
            .batch_load(client, instrument_ids, None)
            .await?;
        log::debug!("Batch loaded {} IB instruments", loaded.len());
        Ok(loaded)
    }

    /// Fetch option chain for an underlying contract with expiry filtering.
    ///
    /// This method calls the provider's fetch_option_chain_by_range with the data client's IB client.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The client is not connected.
    /// - The provider method fails.
    pub async fn fetch_option_chain_by_range(
        &self,
        underlying_symbol: &str,
        exchange: Option<&str>,
        currency: Option<&str>,
        expiry_min: Option<&str>,
        expiry_max: Option<&str>,
    ) -> anyhow::Result<usize> {
        log::debug!(
            "Fetching IB option chain by range (symbol={underlying_symbol}, exchange={exchange:?}, currency={currency:?}, expiry_min={expiry_min:?}, expiry_max={expiry_max:?})"
        );
        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        let underlying = Contract {
            contract_id: 0,
            symbol: Symbol::from(underlying_symbol.to_string()),
            security_type: SecurityType::Stock,
            last_trade_date_or_contract_month: String::new(),
            strike: f64::MAX,
            right: None,
            multiplier: String::new(),
            exchange: IBExchange::from(exchange.unwrap_or("SMART")),
            currency: IBCurrency::from(currency.unwrap_or("USD")),
            local_symbol: String::new(),
            primary_exchange: IBExchange::from(""),
            trading_class: String::new(),
            include_expired: false,
            security_id_type: None,
            security_id: String::new(),
            combo_legs_description: String::new(),
            combo_legs: Vec::new(),
            delta_neutral_contract: None,
            issuer_id: String::new(),
            description: String::new(),
            last_trade_date: None,
        };

        let count = self
            .instrument_provider
            .fetch_option_chain_by_range(client, &underlying, expiry_min, expiry_max, None)
            .await?;
        log::debug!("Fetched {count} IB option instruments for {underlying_symbol}");
        Ok(count)
    }

    /// Fetch futures chain for a given underlying symbol.
    ///
    /// This method calls the provider's fetch_futures_chain with the data client's IB client.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The client is not connected.
    /// - The provider method fails.
    pub async fn fetch_futures_chain(
        &self,
        symbol: &str,
        exchange: Option<&str>,
        currency: Option<&str>,
        min_expiry_days: Option<u32>,
        max_expiry_days: Option<u32>,
    ) -> anyhow::Result<usize> {
        log::debug!(
            "Fetching IB futures chain (symbol={symbol}, exchange={exchange:?}, currency={currency:?}, min_days={min_expiry_days:?}, max_days={max_expiry_days:?})"
        );
        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        let count = self
            .instrument_provider
            .fetch_futures_chain(
                client,
                symbol,
                exchange.unwrap_or(""),
                currency.unwrap_or("USD"),
                None,
                false,
                min_expiry_days,
                max_expiry_days,
            )
            .await?;
        log::debug!("Fetched {count} IB futures instruments for {symbol}");
        Ok(count)
    }

    /// Fetch BAG (spread) contract details.
    ///
    /// This method calls the provider's fetch_bag_contract with the data client's IB client.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The client is not connected.
    /// - The provider method fails.
    pub async fn fetch_bag_contract(
        &self,
        bag_contract: &ibapi::contracts::Contract,
    ) -> anyhow::Result<usize> {
        log::debug!(
            "Fetching IB BAG contract details (contract_id={}, exchange={}, symbol={})",
            bag_contract.contract_id,
            bag_contract.exchange.as_str(),
            bag_contract.symbol.as_str()
        );
        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        let count = self
            .instrument_provider
            .fetch_bag_contract(client, bag_contract)
            .await?;
        log::debug!("Fetched {count} BAG instruments");
        Ok(count)
    }
}

impl Debug for InteractiveBrokersDataClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct(stringify!(InteractiveBrokersDataClient))
            .field("client_id", &self.client_id)
            .field("config", &self.config)
            .field("is_connected", &self.is_connected.load(Ordering::Relaxed))
            .field("has_ib_client", &self.ib_client.is_some())
            .finish_non_exhaustive()
    }
}

#[async_trait::async_trait(?Send)]
impl DataClient for InteractiveBrokersDataClient {
    fn client_id(&self) -> ClientId {
        self.client_id
    }

    fn venue(&self) -> Option<Venue> {
        // Interactive Brokers is a multi-venue adapter (SMART, IDEALPRO, ZEROHASH, CME, etc.),
        // so the data client must register for default routing rather than a single venue:
        // subscriptions and requests are routed by the command's venue (derived from the
        // instrument ID), which would otherwise never match and be dropped by the data engine.
        // Mirrors the other multi-venue adapters (Tardis, Databento).
        None
    }

    fn start(&mut self) -> anyhow::Result<()> {
        tracing::info!(
            client_id = %self.client_id,
            "Starting Interactive Brokers data client"
        );
        Ok(())
    }

    fn stop(&mut self) -> anyhow::Result<()> {
        tracing::info!(
            "Stopping Interactive Brokers data client {id}",
            id = self.client_id
        );
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        self.cancel_active_subscriptions()?;
        self.is_connected.store(false, Ordering::Relaxed);

        self.clear_bar_tracking_state();

        Ok(())
    }

    fn reset(&mut self) -> anyhow::Result<()> {
        tracing::debug!(
            "Resetting Interactive Brokers data client {id}",
            id = self.client_id
        );
        self.cancel_active_subscriptions()?;
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        self.is_connected.store(false, Ordering::Relaxed);
        self.clear_bar_tracking_state();
        self.quote_cache = Arc::new(tokio::sync::Mutex::new(QuoteCache::new()));
        self.option_greeks_cache = Arc::new(tokio::sync::Mutex::new(OptionGreeksCache::new()));
        self.cancellation_token = CancellationToken::new();

        Ok(())
    }

    fn dispose(&mut self) -> anyhow::Result<()> {
        self.stop()
    }

    async fn connect(&mut self) -> anyhow::Result<()> {
        tracing::debug!("Connecting Interactive Brokers data client...");

        self.prepare_task_groups().await?;

        let handle = crate::common::shared_client::get_or_connect(
            &self.config.host,
            self.config.port,
            self.config.client_id,
            self.config.connection_timeout,
        )
        .await
        .context("Failed to connect to IB Gateway/TWS")?;

        let client = Arc::clone(handle.as_arc());

        tracing::info!(
            "Connected to IB Gateway/TWS at {}:{} (client_id: {})",
            self.config.host,
            self.config.port,
            self.config.client_id
        );

        // Set market data type if not default
        let ib_data_type = (self.config.market_data_type != MarketDataType::Realtime)
            .then(|| self.config.market_data_type.into());

        if let Some(ib_data_type) = ib_data_type {
            client
                .switch_market_data_type(ib_data_type)
                .await
                .context("Failed to switch market data type")?;
            tracing::info!("Set market data type to {:?}", self.config.market_data_type);
        }

        // Initialize provider and load instruments from cache/config if configured
        tracing::debug!("Initializing IB data instrument provider");

        if let Err(e) = self
            .instrument_provider
            .initialize_with_client(client.as_ref())
            .await
        {
            if !self.config.instrument_provider.load_ids.is_empty()
                || !self.config.instrument_provider.load_contracts.is_empty()
            {
                return Err(e).context("Failed to load configured IB instruments on startup");
            }

            tracing::warn!("Failed to load instruments on startup: {}", e);
        }

        self.ib_client = Some(handle);

        let data_farm_state = Arc::clone(&self.data_farm_state);
        let is_connected = Arc::clone(&self.is_connected);
        let cancellation_token = self.cancellation_token.child_token();
        let clock = self.clock;

        if let Err(e) = self.session_tasks.spawn(async move {
            if let Err(e) = monitor_data_farm_notices(
                client,
                data_farm_state,
                ib_data_type,
                clock,
                cancellation_token,
                is_connected,
            )
            .await
            {
                tracing::warn!("IB data farm notice monitor stopped: {e:?}");
            }
        }) {
            self.session_tasks.begin_shutdown();
            self.command_tasks.begin_shutdown();
            self.ib_client = None;

            if let Err(teardown_error) = self.finish_tasks().await {
                return Err(anyhow::Error::new(e)
                    .context(format!("IB data startup teardown failed: {teardown_error}")));
            }

            return Err(anyhow::Error::new(e).context("failed to register IB data farm monitor"));
        }

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

        let instrument_count = self.instrument_provider.count();
        if instrument_count > 0 {
            tracing::debug!(
                "Data client connected with {} instruments in provider cache",
                instrument_count
            );

            for instrument in self.instrument_provider.get_all() {
                if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
                    tracing::warn!("Failed to publish startup-loaded instrument: {e}");
                    break;
                }
            }
        }

        tracing::info!("Connected Interactive Brokers data client");
        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        tracing::debug!("Disconnecting Interactive Brokers data client...");

        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        self.cancel_active_subscriptions()?;
        self.ib_client = None;
        let tasks_result = self.finish_tasks().await;
        self.is_connected.store(false, Ordering::Relaxed);
        tracing::info!("Disconnected Interactive Brokers data client");
        tasks_result
    }

    fn is_connected(&self) -> bool {
        self.is_connected.load(Ordering::Relaxed)
    }

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

    // Subscription handlers
    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
        tracing::debug!("Subscribing to quotes for {}", cmd.instrument_id);

        if self.has_subscription(cmd.instrument_id, SubscriptionType::Quotes)? {
            tracing::debug!(
                "Quote subscription already active for {}",
                cmd.instrument_id
            );
            return Ok(());
        }

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        // Get instrument from provider
        let instrument = self
            .instrument_provider
            .find(&cmd.instrument_id)
            .context(format!(
                "Instrument {} not found in provider",
                cmd.instrument_id
            ))?;

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

        // Convert instrument_id to IB contract
        let contract = self
            .instrument_provider
            .resolve_contract_for_instrument(cmd.instrument_id)
            .context("Failed to convert instrument_id to IB contract")?;

        // Check if contract is BAG (spread) or if batch_quotes parameter is set
        // BAG contracts have SecurityType::Spread or combo_legs populated
        let is_bag = matches!(
            contract.security_type,
            ibapi::contracts::SecurityType::Spread
        ) || !contract.combo_legs.is_empty();
        let batch_quotes = cmd
            .params
            .as_ref()
            .and_then(|params| params.get_str("batch_quotes"))
            .map_or(self.config.batch_quotes, parse_bool_param_value);

        let use_market_data = is_bag || batch_quotes;

        let instrument_id = cmd.instrument_id;
        let data_sender = self.data_sender.clone();
        let quote_cache = Arc::clone(&self.quote_cache);
        let clock = self.clock;

        // Get price magnifier from instrument provider
        let price_magnifier = self.instrument_provider.get_price_magnifier(&instrument_id) as f64;

        let subscription_token = self.cancellation_token.child_token();

        // Record the subscription before spawning so a fast-failing task cannot
        // finish before its registry entry exists.
        let subscription_id = next_subscription_id();
        lock(&self.subscriptions)?.insert(
            (cmd.instrument_id, SubscriptionType::Quotes),
            SubscriptionInfo {
                cancellation_token: subscription_token.clone(),
                id: subscription_id,
            },
        );

        let guard = SubscriptionGuard {
            subscriptions: Arc::clone(&self.subscriptions),
            key: (cmd.instrument_id, SubscriptionType::Quotes),
            id: subscription_id,
        };

        // Spawn subscription task
        let client_clone = client.as_arc().clone();
        let ignore_size_updates = self.config.ignore_quote_tick_size_updates;
        let data_farm_state = Arc::clone(&self.data_farm_state);
        let stream_config = StreamConfig::new(self.client_id, &self.config);

        let task = async move {
            if use_market_data {
                // Use market_data (reqMktData) for BAG contracts or when batch_quotes is requested
                tracing::debug!(
                    "Using market_data subscription for {} (BAG: {}, batch_quotes: {})",
                    instrument_id,
                    is_bag,
                    batch_quotes
                );

                if let Err(e) = handle_quote_subscription(
                    client_clone,
                    contract,
                    instrument_id,
                    price_precision,
                    size_precision,
                    data_sender,
                    quote_cache,
                    clock,
                    subscription_token,
                    ignore_size_updates,
                    Arc::clone(&data_farm_state),
                    stream_config,
                )
                .await
                {
                    tracing::error!("Quote subscription error for {}: {:?}", instrument_id, e);
                }
            } else {
                // Try tick_by_tick_bid_ask first for regular contracts (better performance)
                // Fallback to market_data if it fails (e.g., for BAG contracts not detected upfront)
                tracing::debug!(
                    "Attempting tick_by_tick_bid_ask subscription for {}",
                    instrument_id
                );

                match handle_tick_by_tick_quote_subscription(
                    client_clone.clone(),
                    contract.clone(),
                    instrument_id,
                    price_precision,
                    size_precision,
                    data_sender.clone(),
                    clock,
                    subscription_token.clone(),
                    price_magnifier,
                    Arc::clone(&data_farm_state),
                    stream_config,
                )
                .await
                {
                    Ok(()) => {
                        // Success - subscription is active
                    }
                    Err(e) => {
                        tracing::warn!(
                            "tick_by_tick_bid_ask failed for {} (may be BAG contract), falling back to market_data: {:?}",
                            instrument_id,
                            e
                        );

                        // Fallback to market_data (reqMktData) - works for BAG contracts
                        if let Err(fallback_err) = handle_quote_subscription(
                            client_clone,
                            contract,
                            instrument_id,
                            price_precision,
                            size_precision,
                            data_sender,
                            quote_cache,
                            clock,
                            subscription_token,
                            ignore_size_updates,
                            Arc::clone(&data_farm_state),
                            stream_config,
                        )
                        .await
                        {
                            tracing::error!(
                                "Quote subscription fallback also failed for {}: {:?}",
                                instrument_id,
                                fallback_err
                            );
                        } else {
                            tracing::debug!(
                                "Successfully subscribed to {} using market_data fallback",
                                instrument_id
                            );
                        }
                    }
                }
            }

            guard.release();
        };

        self.session_tasks
            .spawn(task)
            .context("failed to register IB data subscription task")?;

        tracing::debug!(
            "Quote subscription started for {} (method: {})",
            cmd.instrument_id,
            if use_market_data {
                "market_data"
            } else {
                "tick_by_tick_bid_ask"
            }
        );
        Ok(())
    }

    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
        tracing::debug!("Subscribing to index prices for {}", cmd.instrument_id);

        if self.has_subscription(cmd.instrument_id, SubscriptionType::IndexPrices)? {
            tracing::debug!(
                "Index price subscription already active for {}",
                cmd.instrument_id
            );
            return Ok(());
        }

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        let instrument = self
            .instrument_provider
            .find(&cmd.instrument_id)
            .context(format!(
                "Instrument {} not found in provider",
                cmd.instrument_id
            ))?;

        let contract = self
            .instrument_provider
            .resolve_contract_for_instrument(cmd.instrument_id)
            .context("Failed to convert instrument_id to IB contract")?;

        if !matches!(contract.security_type, SecurityType::Index) {
            tracing::warn!(
                "Index price subscription not supported for security type {:?} on {}",
                contract.security_type,
                cmd.instrument_id
            );
            return Ok(());
        }

        let price_precision = instrument.price_precision();
        let price_magnifier = self
            .instrument_provider
            .get_price_magnifier(&cmd.instrument_id);
        let instrument_id = cmd.instrument_id;
        let data_sender = self.data_sender.clone();
        let clock = self.clock;

        let subscription_token = self.cancellation_token.child_token();

        let subscription_id = next_subscription_id();
        lock(&self.subscriptions)?.insert(
            (cmd.instrument_id, SubscriptionType::IndexPrices),
            SubscriptionInfo {
                cancellation_token: subscription_token.clone(),
                id: subscription_id,
            },
        );

        let guard = SubscriptionGuard {
            subscriptions: Arc::clone(&self.subscriptions),
            key: (cmd.instrument_id, SubscriptionType::IndexPrices),
            id: subscription_id,
        };

        let client_clone = client.as_arc().clone();
        let data_farm_state = Arc::clone(&self.data_farm_state);
        let stream_config = StreamConfig::new(self.client_id, &self.config);

        let task = async move {
            if let Err(e) = handle_index_price_subscription(
                client_clone,
                contract,
                instrument_id,
                price_precision,
                price_magnifier,
                data_sender,
                clock,
                subscription_token,
                data_farm_state,
                stream_config,
            )
            .await
            {
                tracing::error!(
                    "Index price subscription error for {}: {:?}",
                    instrument_id,
                    e
                );
            }

            guard.release();
        };

        self.session_tasks
            .spawn(task)
            .context("failed to register IB data subscription task")?;

        tracing::debug!("Index price subscription started for {}", cmd.instrument_id);
        Ok(())
    }

    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
        tracing::debug!("Subscribing to option greeks for {}", cmd.instrument_id);

        if lock(&self.option_greeks_subscriptions)?.contains_key(&cmd.instrument_id) {
            tracing::debug!(
                "Option greeks subscription already active for {}",
                cmd.instrument_id
            );
            return Ok(());
        }

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        let instrument = self
            .instrument_provider
            .find(&cmd.instrument_id)
            .context(format!(
                "Instrument {} not found in provider",
                cmd.instrument_id
            ))?;

        if !matches!(
            instrument,
            InstrumentAny::OptionContract(_)
                | InstrumentAny::FuturesContract(_)
                | InstrumentAny::CryptoOption(_)
        ) && !matches!(
            self.instrument_provider
                .resolve_contract_for_instrument(cmd.instrument_id)?
                .security_type,
            SecurityType::Option | SecurityType::FuturesOption
        ) {
            tracing::warn!(
                "Option greeks subscription is only supported for option instruments: {}",
                cmd.instrument_id
            );
            return Ok(());
        }

        let contract = self
            .instrument_provider
            .resolve_contract_for_instrument(cmd.instrument_id)
            .context("Failed to convert instrument_id to IB contract")?;

        let instrument_id = cmd.instrument_id;
        let data_sender = self.data_sender.clone();
        let option_greeks_cache = Arc::clone(&self.option_greeks_cache);
        let clock = self.clock;
        let subscription_token = self.cancellation_token.child_token();
        let subscription_token_clone = subscription_token.clone();
        let client_clone = client.as_arc().clone();
        let data_farm_state = Arc::clone(&self.data_farm_state);
        let stream_config = StreamConfig::new(self.client_id, &self.config);

        let task = async move {
            if let Err(e) = handle_option_greeks_subscription(
                client_clone,
                contract,
                instrument_id,
                data_sender,
                option_greeks_cache,
                clock,
                subscription_token_clone,
                data_farm_state,
                stream_config,
            )
            .await
            {
                tracing::error!(
                    "Option greeks subscription error for {}: {:?}",
                    instrument_id,
                    e
                );
            }
        };

        self.session_tasks
            .spawn(task)
            .context("failed to register IB data subscription task")?;

        let mut subscriptions = lock(&self.option_greeks_subscriptions)?;
        subscriptions.insert(cmd.instrument_id, subscription_token);

        tracing::debug!(
            "Option greeks subscription started for {}",
            cmd.instrument_id
        );
        Ok(())
    }

    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
        tracing::debug!("Unsubscribing from quotes for {}", cmd.instrument_id);

        let mut subscriptions = lock(&self.subscriptions)?;
        if let Some(sub_info) = subscriptions.remove(&(cmd.instrument_id, SubscriptionType::Quotes))
        {
            sub_info.cancellation_token.cancel();
            tracing::debug!("Unsubscribed from quotes for {}", cmd.instrument_id);
        } else {
            tracing::warn!(
                "No active quote subscription found for {}",
                cmd.instrument_id
            );
        }

        // Clear quote cache for this instrument
        {
            // Quote cache doesn't have per-instrument clear, but we can clear all
            // In practice, the cache will naturally expire as new quotes arrive
        }

        Ok(())
    }

    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
        tracing::debug!("Unsubscribing from index prices for {}", cmd.instrument_id);

        let mut subscriptions = lock(&self.subscriptions)?;

        if let Some(sub_info) =
            subscriptions.remove(&(cmd.instrument_id, SubscriptionType::IndexPrices))
        {
            sub_info.cancellation_token.cancel();
            tracing::debug!("Unsubscribed from index prices for {}", cmd.instrument_id);
        } else {
            tracing::warn!(
                "No active index price subscription found for {}",
                cmd.instrument_id
            );
        }

        Ok(())
    }

    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
        tracing::debug!("Unsubscribing from option greeks for {}", cmd.instrument_id);

        let mut subscriptions = lock(&self.option_greeks_subscriptions)?;
        if let Some(subscription_token) = subscriptions.remove(&cmd.instrument_id) {
            subscription_token.cancel();
            tracing::debug!("Unsubscribed from option greeks for {}", cmd.instrument_id);
        } else {
            tracing::warn!(
                "No active option greeks subscription found for {}",
                cmd.instrument_id
            );
        }

        Ok(())
    }

    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
        tracing::debug!("Subscribing to trades for {}", cmd.instrument_id);

        if self.has_subscription(cmd.instrument_id, SubscriptionType::Trades)? {
            tracing::debug!(
                "Trade subscription already active for {}",
                cmd.instrument_id
            );
            return Ok(());
        }

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        // Get instrument from provider
        let instrument = self
            .instrument_provider
            .find(&cmd.instrument_id)
            .context(format!(
                "Instrument {} not found in provider",
                cmd.instrument_id
            ))?;

        // Check if instrument is a CurrencyPair (IB doesn't support trades for CurrencyPair)
        if matches!(instrument, InstrumentAny::CurrencyPair(_)) {
            tracing::error!(
                "Interactive Brokers does not support trades for CurrencyPair instruments: {}",
                cmd.instrument_id
            );
            return Ok(());
        }

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

        // Convert instrument_id to IB contract
        let contract = self
            .instrument_provider
            .resolve_contract_for_instrument(cmd.instrument_id)
            .context("Failed to convert instrument_id to IB contract")?;

        let instrument_id = cmd.instrument_id;
        let data_sender = self.data_sender.clone();
        let clock = self.clock;

        // Create subscription-specific cancellation token
        let subscription_token = self.cancellation_token.child_token();

        let subscription_id = next_subscription_id();
        lock(&self.subscriptions)?.insert(
            (cmd.instrument_id, SubscriptionType::Trades),
            SubscriptionInfo {
                cancellation_token: subscription_token.clone(),
                id: subscription_id,
            },
        );

        let guard = SubscriptionGuard {
            subscriptions: Arc::clone(&self.subscriptions),
            key: (cmd.instrument_id, SubscriptionType::Trades),
            id: subscription_id,
        };

        // Spawn subscription task
        let client_clone = client.as_arc().clone();
        let data_farm_state = Arc::clone(&self.data_farm_state);
        let stream_config = StreamConfig::new(self.client_id, &self.config);

        let task = async move {
            if let Err(e) = handle_trade_subscription(
                client_clone,
                contract,
                instrument_id,
                price_precision,
                size_precision,
                data_sender,
                clock,
                subscription_token,
                data_farm_state,
                stream_config,
            )
            .await
            {
                tracing::error!("Trade subscription error for {}: {:?}", instrument_id, e);
            }

            guard.release();
        };

        self.session_tasks
            .spawn(task)
            .context("failed to register IB data subscription task")?;

        tracing::debug!("Trade subscription started for {}", cmd.instrument_id);
        Ok(())
    }

    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
        tracing::debug!("Unsubscribing from trades for {}", cmd.instrument_id);

        let mut subscriptions = lock(&self.subscriptions)?;
        if let Some(sub_info) = subscriptions.remove(&(cmd.instrument_id, SubscriptionType::Trades))
        {
            sub_info.cancellation_token.cancel();
            tracing::debug!("Unsubscribed from trades for {}", cmd.instrument_id);
        } else {
            tracing::warn!(
                "No active trade subscription found for {}",
                cmd.instrument_id
            );
        }

        Ok(())
    }

    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
        tracing::debug!("Subscribing to bars for {}", cmd.bar_type);

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        // Get instrument from provider
        let instrument_id = cmd.bar_type.instrument_id();
        if self.has_subscription(instrument_id, SubscriptionType::Bars(cmd.bar_type))? {
            tracing::debug!("Bar subscription already active for {}", cmd.bar_type);
            return Ok(());
        }

        let instrument = self
            .instrument_provider
            .find(&instrument_id)
            .context(format!("Instrument {instrument_id} not found in provider"))?;

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

        // Convert instrument_id to IB contract
        let contract = self
            .instrument_provider
            .resolve_contract_for_instrument(instrument_id)
            .context("Failed to convert instrument_id to IB contract")?;

        let bar_type = cmd.bar_type;
        let bar_type_str = bar_type.to_string();
        let data_sender = self.data_sender.clone();
        let clock = self.clock;
        let last_bars = Arc::clone(&self.last_bars);
        let bar_timeout_tasks = Arc::clone(&self.bar_timeout_tasks);
        let handle_revised_bars = self.config.handle_revised_bars;
        let use_rth = self.config.use_regular_trading_hours;
        let start_ns = parse_start_ns(cmd.params.as_ref());
        // Crypto (ZEROHASH/PAXOS) trade-price bars must request AGGTRADES, not
        // TRADES (TWS rejects TRADES for crypto, error 10299) - on BOTH the
        // realtime (reqRealTimeBars) and historical (reqHistoricalData) paths, per
        // the Java engine's whatToShowFor rule. Capture the flag before `contract`
        // is moved into the subscription task below.
        let is_crypto = is_crypto_contract(&contract);

        // Create subscription-specific cancellation token
        let subscription_token = self.cancellation_token.child_token();

        let subscription_id = next_subscription_id();
        lock(&self.subscriptions)?.insert(
            (instrument_id, SubscriptionType::Bars(bar_type)),
            SubscriptionInfo {
                cancellation_token: subscription_token.clone(),
                id: subscription_id,
            },
        );

        let guard = SubscriptionGuard {
            subscriptions: Arc::clone(&self.subscriptions),
            key: (instrument_id, SubscriptionType::Bars(bar_type)),
            id: subscription_id,
        };

        // Spawn subscription task
        let client_clone = client.as_arc().clone();
        let data_farm_state = Arc::clone(&self.data_farm_state);
        let stream_config = StreamConfig::new(self.client_id, &self.config);

        let task = async move {
            let result = if bar_type.spec().timedelta().as_secs() == 5 {
                handle_realtime_bars_subscription(
                    client_clone,
                    contract,
                    bar_type,
                    bar_type_str,
                    instrument_id,
                    price_type_to_ib_realtime_what_to_show_for_security(
                        bar_type.spec().price_type,
                        is_crypto,
                    ),
                    price_precision,
                    size_precision,
                    data_sender,
                    clock,
                    last_bars,
                    bar_timeout_tasks,
                    handle_revised_bars,
                    use_rth,
                    subscription_token,
                    Arc::clone(&data_farm_state),
                    stream_config,
                )
                .await
            } else {
                handle_historical_bars_subscription(
                    client_clone,
                    contract,
                    bar_type,
                    price_type_to_ib_what_to_show_for_security(
                        bar_type.spec().price_type,
                        is_crypto,
                    ),
                    price_precision,
                    size_precision,
                    use_rth,
                    start_ns,
                    data_sender,
                    handle_revised_bars,
                    clock,
                    subscription_token,
                    Arc::clone(&data_farm_state),
                    stream_config,
                )
                .await
            };

            if let Err(e) = result {
                tracing::error!("Bars subscription error for {}: {:?}", bar_type, e);
            }

            guard.release();
        };

        self.session_tasks
            .spawn(task)
            .context("failed to register IB data subscription task")?;

        tracing::debug!("Real-time bars subscription started for {}", bar_type);
        Ok(())
    }

    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
        tracing::debug!("Unsubscribing from bars for {}", cmd.bar_type);

        let instrument_id = cmd.bar_type.instrument_id();
        let mut subscriptions = lock(&self.subscriptions)?;

        if let Some(sub_info) =
            subscriptions.remove(&(instrument_id, SubscriptionType::Bars(cmd.bar_type)))
        {
            sub_info.cancellation_token.cancel();
            tracing::debug!("Unsubscribed from bars for {}", cmd.bar_type);
        } else {
            tracing::warn!("No active bar subscription found for {}", cmd.bar_type);
        }

        Ok(())
    }

    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
        tracing::debug!("Subscribing to book deltas for {}", cmd.instrument_id);

        if self.has_subscription(cmd.instrument_id, SubscriptionType::BookDeltas)? {
            tracing::debug!(
                "Book delta subscription already active for {}",
                cmd.instrument_id
            );
            return Ok(());
        }

        // Validate book type (IB doesn't support L3_MBO)
        if cmd.book_type == BookType::L3_MBO {
            tracing::error!(
                "Cannot subscribe to order book deltas: L3_MBO data is not published by Interactive Brokers. Valid book types are L1_MBP, L2_MBP"
            );
            return Ok(());
        }

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        // Get instrument from provider
        let instrument = self
            .instrument_provider
            .find(&cmd.instrument_id)
            .context(format!(
                "Instrument {} not found in provider",
                cmd.instrument_id
            ))?;

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

        // Convert instrument_id to IB contract
        let contract = self
            .instrument_provider
            .resolve_contract_for_instrument(cmd.instrument_id)
            .context("Failed to convert instrument_id to IB contract")?;

        let instrument_id = cmd.instrument_id;
        let data_sender = self.data_sender.clone();
        let clock = self.clock;

        // Create subscription-specific cancellation token
        let subscription_token = self.cancellation_token.child_token();

        // Get depth from command or default to 20 (Python default)
        let depth_rows = cmd.depth.map_or(20, |d| d.get() as i32);

        // Get is_smart_depth from params or default to true
        let is_smart_depth = cmd
            .params
            .as_ref()
            .and_then(|params| params.get_str("is_smart_depth"))
            .is_none_or(parse_bool_param_value);

        let subscription_id = next_subscription_id();
        lock(&self.subscriptions)?.insert(
            (cmd.instrument_id, SubscriptionType::BookDeltas),
            SubscriptionInfo {
                cancellation_token: subscription_token.clone(),
                id: subscription_id,
            },
        );

        let guard = SubscriptionGuard {
            subscriptions: Arc::clone(&self.subscriptions),
            key: (cmd.instrument_id, SubscriptionType::BookDeltas),
            id: subscription_id,
        };

        // Spawn subscription task
        let client_clone = client.as_arc().clone();
        let data_farm_state = Arc::clone(&self.data_farm_state);
        let stream_config = StreamConfig::new(self.client_id, &self.config);

        let task = async move {
            if let Err(e) = handle_market_depth_subscription(
                client_clone,
                contract,
                instrument_id,
                price_precision,
                size_precision,
                depth_rows,
                is_smart_depth,
                data_sender,
                clock,
                subscription_token,
                data_farm_state,
                stream_config,
            )
            .await
            {
                tracing::error!(
                    "Market depth subscription error for {}: {:?}",
                    instrument_id,
                    e
                );
            }

            guard.release();
        };

        self.session_tasks
            .spawn(task)
            .context("failed to register IB data subscription task")?;

        tracing::debug!(
            "Market depth subscription started for {}",
            cmd.instrument_id
        );
        Ok(())
    }

    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
        tracing::debug!("Unsubscribing from book deltas for {}", cmd.instrument_id);

        let mut subscriptions = lock(&self.subscriptions)?;

        if let Some(sub_info) =
            subscriptions.remove(&(cmd.instrument_id, SubscriptionType::BookDeltas))
        {
            sub_info.cancellation_token.cancel();
            tracing::debug!("Unsubscribed from book deltas for {}", cmd.instrument_id);
        } else {
            tracing::warn!(
                "No active book delta subscription found for {}",
                cmd.instrument_id
            );
        }

        Ok(())
    }

    // Request handlers
    fn request_instrument(&self, cmd: RequestInstrument) -> anyhow::Result<()> {
        tracing::debug!("Requesting instrument: {}", cmd.instrument_id);

        if cmd.start.is_some() {
            tracing::warn!(
                "Requesting instrument {} with specified `start` which has no effect",
                cmd.instrument_id
            );
        }

        if cmd.end.is_some() {
            tracing::warn!(
                "Requesting instrument {} with specified `end` which has no effect",
                cmd.instrument_id
            );
        }

        // Check if force_instrument_update is requested
        let force_update = cmd
            .params
            .as_ref()
            .and_then(|params| params.get_str("force_instrument_update"))
            .is_some_and(parse_bool_param_value);

        // Get instrument from provider (or load if not found or force_update)
        let instrument =
            if force_update || self.instrument_provider.find(&cmd.instrument_id).is_none() {
                // Need to load instrument - spawn async task
                let client = self
                    .ib_client
                    .as_ref()
                    .context("IB client not connected. Call connect() first")?;
                let instrument_provider = Arc::clone(&self.instrument_provider);
                let instrument_id = cmd.instrument_id;
                let data_sender = self.data_sender.clone();
                let clock = self.clock;
                let request_id = cmd.request_id;
                let client_id = cmd.client_id.unwrap_or(self.client_id);
                let params = cmd.params.clone();
                let start_nanos = cmd.start.map(datetime_to_unix_nanos);
                let end_nanos = cmd.end.map(datetime_to_unix_nanos);

                let client_clone = client.as_arc().clone();

                self.spawn_command(async move {
                    let filters = params_to_string_filters(params.as_ref());

                    if let Err(e) = instrument_provider
                        .fetch_contract_details(&client_clone, instrument_id, force_update, filters)
                        .await
                    {
                        tracing::error!(
                            "Failed to fetch contract details for {}: {:?}",
                            instrument_id,
                            e
                        );
                        return;
                    }

                    if let Some(instrument) = instrument_provider.find(&instrument_id) {
                        let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
                            request_id,
                            client_id,
                            instrument_id,
                            instrument,
                            start_nanos,
                            end_nanos,
                            clock.get_time_ns(),
                            params,
                        )));

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

                // Return early, response will be sent async
                return Ok(());
            } else {
                // Instrument already in provider
                self.instrument_provider
                    .find(&cmd.instrument_id)
                    .context(format!(
                        "Instrument {} not found in provider",
                        cmd.instrument_id
                    ))?
            };

        let start_nanos = cmd.start.map(datetime_to_unix_nanos);
        let end_nanos = cmd.end.map(datetime_to_unix_nanos);

        let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
            cmd.request_id,
            cmd.client_id.unwrap_or(self.client_id),
            cmd.instrument_id,
            instrument,
            start_nanos,
            end_nanos,
            self.clock.get_time_ns(),
            cmd.params,
        )));

        if let Err(e) = self.data_sender.send(DataEvent::Response(response)) {
            tracing::error!("Failed to send instrument response: {e}");
        }

        Ok(())
    }

    fn request_instruments(&self, cmd: RequestInstruments) -> anyhow::Result<()> {
        tracing::debug!("Requesting all instruments for venue: {:?}", cmd.venue);

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        // Check for force_instrument_update
        let force_update = cmd
            .params
            .as_ref()
            .and_then(|params| params.get_str("force_instrument_update"))
            .is_some_and(parse_bool_param_value);

        // Check if ib_contracts parameter is provided for batch loading
        let mut contract_specs_to_load: Vec<serde_json::Value> = Vec::new();

        if let Some(params) = &cmd.params
            && let Some(ib_contracts_value) = params.get("ib_contracts")
        {
            match ib_contracts_value {
                serde_json::Value::Array(contract_specs) => {
                    tracing::debug!(
                        "Parsed {} structured contract specs from ib_contracts",
                        contract_specs.len()
                    );
                    contract_specs_to_load = contract_specs.clone();
                }
                serde_json::Value::String(ib_contracts_json_str) => {
                    match serde_json::from_str::<serde_json::Value>(ib_contracts_json_str) {
                        Ok(serde_json::Value::Array(contract_specs)) => {
                            tracing::debug!(
                                "Parsed {} contract specs from ib_contracts JSON",
                                contract_specs.len()
                            );
                            log::debug!("Parsed ib_contracts payload: {ib_contracts_json_str}");
                            contract_specs_to_load = contract_specs;
                        }
                        Ok(value) => {
                            tracing::warn!(
                                "Expected ib_contracts JSON array, received {}. Continuing without contracts",
                                value
                            );
                        }
                        Err(e) => {
                            tracing::warn!(
                                "Failed to parse ib_contracts JSON: {}. Continuing without contracts",
                                e
                            );
                        }
                    }
                }
                value => {
                    tracing::warn!(
                        "Expected ib_contracts array or JSON string, received {}. Continuing without contracts",
                        value
                    );
                }
            }
        }

        // If force_update is requested or we need to batch load, spawn async task
        let instrument_provider = Arc::clone(&self.instrument_provider);
        let client_clone = client.as_arc().clone();
        let data_sender = self.data_sender.clone();
        let clock = self.clock;
        let request_id = cmd.request_id;
        let client_id = cmd.client_id.unwrap_or(self.client_id);
        let venue = cmd.venue.unwrap_or(*IB_VENUE);
        let params = cmd.params.clone();
        let start_nanos = cmd.start.map(datetime_to_unix_nanos);
        let end_nanos = cmd.end.map(datetime_to_unix_nanos);

        // Handle batch loading if contracts are provided or force_update is requested
        if !contract_specs_to_load.is_empty() || force_update {
            let contract_specs_to_load_clone = contract_specs_to_load;
            let return_loaded_only = !contract_specs_to_load_clone.is_empty();

            self.spawn_command(async move {
                let mut loaded_instrument_ids = Vec::new();

                // Load instruments from contracts if provided
                if !contract_specs_to_load_clone.is_empty() {
                    for contract_spec in contract_specs_to_load_clone {
                        let contract =
                            match parse_contract_from_json(&contract_spec)
                            {
                                Ok(contract) => contract,
                                Err(e) => {
                                    tracing::warn!(
                                        "Failed to parse IB contract spec {:?}: {}",
                                        contract_spec,
                                        e
                                    );
                                    continue;
                                }
                            };

                        log::debug!(
                            "Loading instrument from IB contract spec (sec_type={:?}, symbol={}, local_symbol={}, exchange={}, expiry={})",
                            contract.security_type,
                            contract.symbol.as_str(),
                            contract.local_symbol.as_str(),
                            contract.exchange.as_str(),
                            contract.last_trade_date_or_contract_month.as_str()
                        );

                        match instrument_provider
                            .load_contract_spec(&client_clone, &contract, Some(&contract_spec))
                            .await
                        {
                            Ok(mut instrument_ids) => {
                                loaded_instrument_ids.append(&mut instrument_ids);
                            }
                            Err(e) => {
                                tracing::warn!(
                                    "Failed to load IB contract spec {:?}: {}",
                                    contract_spec,
                                    e
                                );
                            }
                        }
                    }
                }

                // If force_update, also reload all existing instruments
                if force_update && !return_loaded_only {
                    let all_instrument_ids: Vec<InstrumentId> = instrument_provider
                        .get_all()
                        .into_iter()
                        .map(|inst| inst.id())
                        .collect();

                    if !all_instrument_ids.is_empty()
                        && let Ok(mut reloaded_ids) = instrument_provider
                            .batch_load(&client_clone, all_instrument_ids, None)
                            .await
                    {
                        loaded_instrument_ids.append(&mut reloaded_ids);
                    }
                }

                let instruments = if return_loaded_only {
                    instrument_provider.find_all(&loaded_instrument_ids)
                } else {
                    Vec::new()
                };

                let instruments_count = instruments.len();

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

                if let Err(e) = data_sender.send(DataEvent::Response(response)) {
                    tracing::error!("Failed to send instruments response: {e}");
                } else {
                    tracing::debug!(
                        "Successfully sent {} instruments response (loaded {} new instruments)",
                        instruments_count,
                        loaded_instrument_ids.len()
                    );
                }
            });
        } else {
            let response = DataResponse::Instruments(InstrumentsResponse::new(
                cmd.request_id,
                cmd.client_id.unwrap_or(self.client_id),
                venue,
                Vec::new(),
                start_nanos,
                end_nanos,
                self.clock.get_time_ns(),
                cmd.params,
            ));

            if let Err(e) = self.data_sender.send(DataEvent::Response(response)) {
                tracing::error!("Failed to send instruments response: {e}");
            } else {
                tracing::debug!("Successfully sent empty instruments response");
            }
        }

        Ok(())
    }

    fn request_quotes(&self, cmd: RequestQuotes) -> anyhow::Result<()> {
        tracing::debug!("Requesting quotes for {}", cmd.instrument_id);

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        // Get instrument from provider
        let instrument = self
            .instrument_provider
            .find(&cmd.instrument_id)
            .context(format!(
                "Instrument {} not found in provider",
                cmd.instrument_id
            ))?;

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

        // Convert instrument_id to IB contract
        let contract = self
            .instrument_provider
            .resolve_contract_for_instrument(cmd.instrument_id)
            .context("Failed to convert instrument_id to IB contract")?;

        let limit = cmd
            .limit
            .map(|limit| limit.get())
            .or_else(|| cmd.start.is_none().then_some(HISTORICAL_TICK_DEFAULT_LIMIT));
        let number_of_ticks = limit.unwrap_or(1000).min(1000) as i32;

        let instrument_id = cmd.instrument_id;
        let data_sender = self.data_sender.clone();
        let clock = self.clock;
        let request_id = cmd.request_id;
        let client_id = cmd.client_id.unwrap_or(self.client_id);
        let params = cmd.params.clone();
        let start_nanos = cmd.start.map(datetime_to_unix_nanos);
        let end_nanos = cmd.end.map(datetime_to_unix_nanos);

        // Spawn async task to handle the request with pagination
        let client_clone = client.as_arc().clone();
        let start_nanos_clone = start_nanos;
        let end_nanos_clone = end_nanos;
        let cmd_start = cmd.start;
        let cmd_end = cmd.end;
        let trading_hours = request_trading_hours(self.config.use_regular_trading_hours);
        let price_magnifier = self.instrument_provider.get_price_magnifier(&instrument_id);

        self.spawn_command(async move {
            let mut all_quotes = Vec::new();
            // Work backwards from end_date, updating end to the earliest tick received
            let mut current_end_date = cmd_end;
            if current_end_date.is_none() {
                current_end_date = Some(jiff::Timestamp::now());
            }

            let current_start_date = cmd_start;

            loop {
                if !should_continue_historical_tick_pagination(
                    current_start_date,
                    current_end_date,
                    all_quotes.len(),
                    limit,
                ) {
                    break;
                }

                let current_end_ib = current_end_date.as_ref().map(jiff_to_ib_datetime);

                // Make request for this batch
                let mut builder = client_clone
                    .historical_ticks(&contract, number_of_ticks)
                    .trading_hours(trading_hours);

                // IB accepts start or end on reqHistoricalTicks, not both; backward
                // pagination anchors on the end and filters the start client-side.
                if let Some(end) = current_end_ib {
                    builder = builder.ending(end);
                }

                match builder.bid_ask(IgnoreSize::No).await {
                    Ok(subscription) => {
                        let mut subscription = subscription.filter_data();
                        let mut batch_quotes = Vec::new();

                        while let Some(tick_result) = subscription.next().await {
                            let tick = match tick_result {
                                Ok(tick) => tick,
                                Err(e) => {
                                    tracing::warn!("Historical quote ticks stream error: {e:?}");
                                    continue;
                                }
                            };

                            let ts_event =
                                super::convert::ib_timestamp_to_unix_nanos(&tick.timestamp);
                            let ts_init = clock.get_time_ns();

                            match super::parse::parse_quote_tick(
                                instrument_id,
                                Some(apply_price_magnifier(tick.price_bid, price_magnifier)),
                                Some(apply_price_magnifier(tick.price_ask, price_magnifier)),
                                tick.size_bid,
                                tick.size_ask,
                                price_precision,
                                size_precision,
                                ts_event,
                                ts_init,
                            ) {
                                Ok(quote_tick) => batch_quotes.push(quote_tick),
                                Err(e) => {
                                    tracing::warn!("Failed to parse quote tick: {:?}", e);
                                }
                            }
                        }

                        if !extend_historical_tick_batch(
                            &mut all_quotes,
                            batch_quotes,
                            current_start_date,
                            &mut current_end_date,
                            start_nanos_clone,
                            end_nanos_clone,
                            limit,
                            |quote| quote.ts_event,
                        ) {
                            break;
                        }
                    }
                    Err(e) => {
                        tracing::error!(
                            "Historical quotes request failed for {}: {:?}",
                            instrument_id,
                            e
                        );
                        break;
                    }
                }
            }

            retain_historical_ticks_in_range(
                &mut all_quotes,
                start_nanos_clone,
                end_nanos_clone,
                |quote| quote.ts_event,
            );

            all_quotes.sort_by_key(|q| q.ts_event);
            if let Some(limit) = limit
                && all_quotes.len() > limit
            {
                all_quotes = all_quotes.split_off(all_quotes.len() - limit);
            }

            let quotes_count = all_quotes.len();
            let response = DataResponse::Quotes(QuotesResponse::new(
                request_id,
                client_id,
                instrument_id,
                all_quotes,
                start_nanos_clone,
                end_nanos_clone,
                clock.get_time_ns(),
                params,
            ));

            if let Err(e) = data_sender.send(DataEvent::Response(response)) {
                tracing::error!("Failed to send quotes response: {e}");
            } else {
                tracing::debug!(
                    "Successfully sent {} quotes for {}",
                    quotes_count,
                    instrument_id
                );
            }
        });

        Ok(())
    }

    fn request_trades(&self, cmd: RequestTrades) -> anyhow::Result<()> {
        tracing::debug!("Requesting trades for {}", cmd.instrument_id);

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        // Get instrument from provider
        let instrument = self
            .instrument_provider
            .find(&cmd.instrument_id)
            .context(format!(
                "Instrument {} not found in provider",
                cmd.instrument_id
            ))?;

        // Check if instrument is a CurrencyPair (IB doesn't support trades for CurrencyPair)
        if matches!(instrument, InstrumentAny::CurrencyPair(_)) {
            tracing::error!(
                "Interactive Brokers does not support trades for CurrencyPair instruments: {}",
                cmd.instrument_id
            );
            return Ok(());
        }

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

        // Convert instrument_id to IB contract
        let contract = self
            .instrument_provider
            .resolve_contract_for_instrument(cmd.instrument_id)
            .context("Failed to convert instrument_id to IB contract")?;

        let limit = cmd
            .limit
            .map(|limit| limit.get())
            .or_else(|| cmd.start.is_none().then_some(HISTORICAL_TICK_DEFAULT_LIMIT));
        let number_of_ticks = limit.unwrap_or(1000).min(1000) as i32;

        let instrument_id = cmd.instrument_id;
        let data_sender = self.data_sender.clone();
        let clock = self.clock;
        let request_id = cmd.request_id;
        let client_id = cmd.client_id.unwrap_or(self.client_id);
        let params = cmd.params.clone();
        let start_nanos = cmd.start.map(datetime_to_unix_nanos);
        let end_nanos = cmd.end.map(datetime_to_unix_nanos);

        // Spawn async task to handle the request with pagination
        let client_clone = client.as_arc().clone();
        let start_nanos_clone = start_nanos;
        let end_nanos_clone = end_nanos;
        let cmd_start = cmd.start;
        let cmd_end = cmd.end;
        let trading_hours = request_trading_hours(self.config.use_regular_trading_hours);
        let price_magnifier = self.instrument_provider.get_price_magnifier(&instrument_id);

        self.spawn_command(async move {
            let mut all_trades = Vec::new();
            // Work backwards from end_date, updating end to the earliest tick received
            let mut current_end_date = cmd_end;
            if current_end_date.is_none() {
                current_end_date = Some(jiff::Timestamp::now());
            }

            let current_start_date = cmd_start;

            loop {
                if !should_continue_historical_tick_pagination(
                    current_start_date,
                    current_end_date,
                    all_trades.len(),
                    limit,
                ) {
                    break;
                }

                let current_end_ib = current_end_date.as_ref().map(jiff_to_ib_datetime);

                // Make request for this batch
                let mut builder = client_clone
                    .historical_ticks(&contract, number_of_ticks)
                    .trading_hours(trading_hours);

                // IB accepts start or end on reqHistoricalTicks, not both; backward
                // pagination anchors on the end and filters the start client-side.
                if let Some(end) = current_end_ib {
                    builder = builder.ending(end);
                }

                match builder.trade().await {
                    Ok(subscription) => {
                        let mut subscription = subscription.filter_data();
                        let mut batch_trades = Vec::new();

                        while let Some(tick_result) = subscription.next().await {
                            let tick = match tick_result {
                                Ok(tick) => tick,
                                Err(e) => {
                                    tracing::warn!("Historical trade ticks stream error: {e:?}");
                                    continue;
                                }
                            };

                            let ts_event =
                                super::convert::ib_timestamp_to_unix_nanos(&tick.timestamp);
                            let ts_init = clock.get_time_ns();

                            // Generate trade ID from exchange and special conditions if available
                            let trade_id = None;

                            let Some(size) = tick.size else {
                                tracing::warn!(
                                    "Skipping historical trade tick with no size for {}",
                                    instrument_id
                                );
                                continue;
                            };

                            match super::parse::parse_trade_tick(
                                instrument_id,
                                apply_price_magnifier(tick.price, price_magnifier),
                                size,
                                price_precision,
                                size_precision,
                                ts_event,
                                ts_init,
                                trade_id,
                            ) {
                                Ok(trade_tick) => batch_trades.push(trade_tick),
                                Err(e) => {
                                    log_tick_parse_error("trade", instrument_id, &e);
                                }
                            }
                        }

                        if !extend_historical_tick_batch(
                            &mut all_trades,
                            batch_trades,
                            current_start_date,
                            &mut current_end_date,
                            start_nanos_clone,
                            end_nanos_clone,
                            limit,
                            |trade| trade.ts_event,
                        ) {
                            break;
                        }
                    }
                    Err(e) => {
                        tracing::error!(
                            "Historical trades request failed for {}: {:?}",
                            instrument_id,
                            e
                        );
                        break;
                    }
                }
            }

            retain_historical_ticks_in_range(
                &mut all_trades,
                start_nanos_clone,
                end_nanos_clone,
                |trade| trade.ts_event,
            );

            all_trades.sort_by_key(|t| t.ts_event);
            if let Some(limit) = limit
                && all_trades.len() > limit
            {
                all_trades = all_trades.split_off(all_trades.len() - limit);
            }

            let trades_count = all_trades.len();
            let response = DataResponse::Trades(TradesResponse::new(
                request_id,
                client_id,
                instrument_id,
                all_trades,
                start_nanos_clone,
                end_nanos_clone,
                clock.get_time_ns(),
                params,
            ));

            if let Err(e) = data_sender.send(DataEvent::Response(response)) {
                tracing::error!("Failed to send trades response: {e}");
            } else {
                tracing::debug!(
                    "Successfully sent {} trades for {}",
                    trades_count,
                    instrument_id
                );
            }
        });

        Ok(())
    }

    fn request_bars(&self, cmd: RequestBars) -> anyhow::Result<()> {
        tracing::debug!("Requesting bars for {}", cmd.bar_type);

        // Validate bar spec (only time-aggregated bars are supported)
        if !cmd.bar_type.spec().is_time_aggregated() {
            tracing::error!(
                "Cannot request {} bars: only time bars are aggregated by Interactive Brokers",
                cmd.bar_type
            );
            return Ok(());
        }

        let client = self
            .ib_client
            .as_ref()
            .context("IB client not connected. Call connect() first")?;

        // Get instrument from provider
        let instrument_id = cmd.bar_type.instrument_id();
        let instrument = self
            .instrument_provider
            .find(&instrument_id)
            .context(format!("Instrument {instrument_id} not found in provider"))?;

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

        // Convert instrument_id to IB contract
        let contract = self
            .instrument_provider
            .resolve_contract_for_instrument(instrument_id)
            .context("Failed to convert instrument_id to IB contract")?;

        // Convert bar type to IB formats
        let ib_bar_size = bar_type_to_ib_bar_size(&cmd.bar_type)
            .context("Failed to convert bar type to IB bar size")?;
        // Crypto trade-price bars require AGGTRADES (TWS rejects TRADES for crypto,
        // error 10299); mirror the Java engine's whatToShowFor rule.
        let is_crypto = is_crypto_contract(&contract);
        let ib_what_to_show =
            price_type_to_ib_what_to_show_for_security(cmd.bar_type.spec().price_type, is_crypto);

        // Calculate segments to break down the request if needed.
        // Omit the end date for continuous futures (IB error 10339).
        let is_continuous_future = contract.security_type == SecurityType::ContinuousFuture;
        let segments = calculate_duration_segments(cmd.start, cmd.end, None);
        let segments = bar_request_segments(segments, is_continuous_future);

        let bar_type = cmd.bar_type;
        let data_sender = self.data_sender.clone();
        let clock = self.clock;
        let request_id = cmd.request_id;
        let client_id = cmd.client_id.unwrap_or(self.client_id);
        let params = cmd.params.clone();
        let start_nanos = cmd.start.map(datetime_to_unix_nanos);
        let end_nanos = cmd.end.map(datetime_to_unix_nanos);
        let limit = cmd.limit.map(|limit| limit.get());
        let price_magnifier = self.instrument_provider.get_price_magnifier(&instrument_id);

        // Spawn async task to handle the request with segmentation
        let client_clone = client.as_arc().clone();
        let trading_hours = request_trading_hours(self.config.use_regular_trading_hours);

        self.spawn_command(async move {
            let mut all_bars = Vec::new();

            for (seg_end, seg_duration) in segments {
                let mut request = client_clone
                    .historical_data(&contract, ib_bar_size)
                    .duration(seg_duration)
                    .what_to_show(ib_what_to_show)
                    .trading_hours(trading_hours);

                if let Some(end) = seg_end {
                    request = request.ending(jiff_to_ib_datetime(&end));
                }

                match request.fetch().await {
                    Ok(historical_data) => {
                        // Convert IB bars to Nautilus bars
                        for ib_bar in &historical_data.bars {
                            let ib_bar = apply_bar_price_magnifier(ib_bar, price_magnifier);

                            match ib_bar_to_nautilus_bar(
                                &ib_bar,
                                bar_type,
                                price_precision,
                                size_precision,
                            ) {
                                Ok(bar) => all_bars.push(bar),
                                Err(e) => {
                                    tracing::warn!(
                                        "Failed to convert IB bar to Nautilus bar: {:?}",
                                        e
                                    );
                                }
                            }
                        }
                    }
                    Err(e) => {
                        tracing::warn!(
                            "Historical data request failed for {bar_type} segment ending at {seg_end:?}: {e}, emitting the bars already retrieved"
                        );
                        break;
                    }
                }
            }

            // Return aggregated results
            if all_bars.is_empty() {
                tracing::warn!("No bar data received for {}", bar_type);
            }

            // Sort and deduplicate bars as segments might overlap or be out of order from IB.
            all_bars.sort_by_key(|b| b.ts_event);
            all_bars.dedup();
            if let Some(limit) = limit
                && all_bars.len() > limit
            {
                all_bars = all_bars.split_off(all_bars.len() - limit);
            }

            let bars_count = all_bars.len();

            let response = DataResponse::Bars(BarsResponse::new(
                request_id,
                client_id,
                bar_type,
                all_bars,
                start_nanos,
                end_nanos,
                clock.get_time_ns(),
                params,
            ));

            if let Err(e) = data_sender.send(DataEvent::Response(response)) {
                tracing::error!("Failed to send bars response: {e}");
            } else {
                tracing::debug!(
                    "Successfully sent {} bars for {} (segmented)",
                    bars_count,
                    bar_type
                );
            }
        });

        Ok(())
    }
}

impl InteractiveBrokersDataClient {
    fn has_subscription(
        &self,
        instrument_id: InstrumentId,
        subscription_type: SubscriptionType,
    ) -> anyhow::Result<bool> {
        let subscriptions = lock(&self.subscriptions)?;
        Ok(subscriptions.contains_key(&(instrument_id, subscription_type)))
    }

    fn clear_bar_tracking_state(&self) {
        if let Ok(mut tasks) = self.bar_timeout_tasks.try_lock() {
            for task in tasks.values_mut() {
                task.abort();
            }
        } else {
            tracing::warn!("Failed to lock IB bar timeout tasks for cleanup");
        }

        if let Ok(mut last_bars) = self.last_bars.try_lock() {
            last_bars.clear();
        } else {
            tracing::warn!("Failed to lock IB last bars for cleanup");
        }
    }
}

impl Drop for InteractiveBrokersDataClient {
    fn drop(&mut self) {
        let _ = self.stop();
    }
}

#[cfg(test)]
mod tests {
    use nautilus_common::live::get_runtime;
    use rstest::rstest;

    use super::*;
    use crate::common::consts::IB_CLIENT_ID;

    #[rstest]
    #[case(true, ibapi::market_data::TradingHours::Regular)]
    #[case(false, ibapi::market_data::TradingHours::Extended)]
    fn test_request_trading_hours_uses_config(
        #[case] use_regular_trading_hours: bool,
        #[case] expected: ibapi::market_data::TradingHours,
    ) {
        assert_eq!(request_trading_hours(use_regular_trading_hours), expected);
    }

    #[rstest]
    #[case("true", true)]
    #[case("True", true)]
    #[case("1", true)]
    #[case("false", false)]
    #[case("False", false)]
    #[case("0", false)]
    fn test_parse_bool_param_value(#[case] value: &str, #[case] expected: bool) {
        assert_eq!(parse_bool_param_value(value), expected);
    }

    #[rstest]
    fn test_datetime_to_unix_nanos() {
        let dt = jiff::Timestamp::new(1, 2).unwrap();

        assert_eq!(datetime_to_unix_nanos(dt), UnixNanos::from(1_000_000_002));
    }

    #[rstest]
    fn test_subscription_keys_keep_quote_and_trade_streams_independent() {
        let instrument_id = InstrumentId::from("AAPL.NASDAQ");
        let quote_token = CancellationToken::new();
        let trade_token = CancellationToken::new();
        let mut subscriptions = AHashMap::new();
        subscriptions.insert(
            (instrument_id, SubscriptionType::Quotes),
            SubscriptionInfo {
                cancellation_token: quote_token.clone(),
                id: next_subscription_id(),
            },
        );

        subscriptions.insert(
            (instrument_id, SubscriptionType::Trades),
            SubscriptionInfo {
                cancellation_token: trade_token.clone(),
                id: next_subscription_id(),
            },
        );

        let quote = subscriptions
            .remove(&(instrument_id, SubscriptionType::Quotes))
            .unwrap();
        quote.cancellation_token.cancel();

        assert!(quote_token.is_cancelled());
        assert!(!trade_token.is_cancelled());
        assert_eq!(subscriptions.len(), 1);
        assert!(subscriptions.contains_key(&(instrument_id, SubscriptionType::Trades)));
    }

    #[rstest]
    fn test_subscription_keys_keep_bar_types_of_one_instrument_independent() {
        let minute = BarType::from("AAPL.NASDAQ-1-MINUTE-LAST-EXTERNAL");
        let five_minute = BarType::from("AAPL.NASDAQ-5-MINUTE-LAST-EXTERNAL");
        let instrument_id = minute.instrument_id();
        let minute_token = CancellationToken::new();
        let five_minute_token = CancellationToken::new();
        let mut subscriptions = AHashMap::new();
        subscriptions.insert(
            (instrument_id, SubscriptionType::Bars(minute)),
            SubscriptionInfo {
                cancellation_token: minute_token.clone(),
                id: next_subscription_id(),
            },
        );

        subscriptions.insert(
            (instrument_id, SubscriptionType::Bars(five_minute)),
            SubscriptionInfo {
                cancellation_token: five_minute_token.clone(),
                id: next_subscription_id(),
            },
        );

        let five = subscriptions
            .remove(&(instrument_id, SubscriptionType::Bars(five_minute)))
            .unwrap();
        five.cancellation_token.cancel();

        assert!(five_minute_token.is_cancelled());
        assert!(!minute_token.is_cancelled());
        assert_eq!(subscriptions.len(), 1);
        assert!(subscriptions.contains_key(&(instrument_id, SubscriptionType::Bars(minute))));
    }

    #[rstest]
    fn test_subscription_guard_release_removes_only_matching_registration() {
        let subscriptions = Arc::new(Mutex::new(AHashMap::new()));
        let instrument_id = InstrumentId::from("AAPL.NASDAQ");
        let key = (instrument_id, SubscriptionType::Quotes);

        let first_id = next_subscription_id();
        subscriptions.lock().unwrap().insert(
            key,
            SubscriptionInfo {
                cancellation_token: CancellationToken::new(),
                id: first_id,
            },
        );

        let stale_guard = SubscriptionGuard {
            subscriptions: Arc::clone(&subscriptions),
            key,
            id: first_id,
        };

        // A newer registration for the same key must survive the stale guard.
        let second_id = next_subscription_id();
        subscriptions.lock().unwrap().insert(
            key,
            SubscriptionInfo {
                cancellation_token: CancellationToken::new(),
                id: second_id,
            },
        );

        stale_guard.release();
        assert!(subscriptions.lock().unwrap().contains_key(&key));

        let current_guard = SubscriptionGuard {
            subscriptions: Arc::clone(&subscriptions),
            key,
            id: second_id,
        };

        current_guard.release();
        assert!(!subscriptions.lock().unwrap().contains_key(&key));
    }

    #[rstest]
    fn test_subscription_guard_drop_removes_registration_without_release() {
        let subscriptions = Arc::new(Mutex::new(AHashMap::new()));
        let instrument_id = InstrumentId::from("AAPL.NASDAQ");
        let key = (instrument_id, SubscriptionType::Quotes);

        let id = next_subscription_id();
        subscriptions.lock().unwrap().insert(
            key,
            SubscriptionInfo {
                cancellation_token: CancellationToken::new(),
                id,
            },
        );

        {
            let _guard = SubscriptionGuard {
                subscriptions: Arc::clone(&subscriptions),
                key,
                id,
            };
            // Dropped without release(), as when a stream task panics.
        }

        assert!(!subscriptions.lock().unwrap().contains_key(&key));
    }

    #[rstest]
    fn test_venue_is_none_for_default_routing() {
        // IB is a multi-venue adapter: `venue()` must be `None` so the data engine registers
        // the client for default routing. A `Some(venue)` here means venue-routed subscribe
        // commands (e.g. `BTC/USD.ZEROHASH` bars, `AAPL.NASDAQ` quotes) never reach the client.
        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
        nautilus_common::live::runner::replace_data_event_sender(sender);

        let config = InteractiveBrokersDataClientConfig::default();

        let provider = Arc::new(InteractiveBrokersInstrumentProvider::new(
            config.instrument_provider.clone(),
        ));
        let client = InteractiveBrokersDataClient::new(*IB_CLIENT_ID, config, provider).unwrap();

        assert_eq!(client.venue(), None);
    }

    #[rstest]
    fn test_stop_closes_tasks_until_async_generation_preparation() {
        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
        nautilus_common::live::runner::replace_data_event_sender(sender);

        let config = InteractiveBrokersDataClientConfig::default();

        let provider = Arc::new(InteractiveBrokersInstrumentProvider::new(
            config.instrument_provider.clone(),
        ));
        let mut client =
            InteractiveBrokersDataClient::new(*IB_CLIENT_ID, config, provider).unwrap();

        client
            .session_tasks
            .spawn(async { std::future::pending::<()>().await })
            .unwrap();

        client.stop().unwrap();

        assert!(client.cancellation_token.is_cancelled());
        assert!(client.cancellation_token.child_token().is_cancelled());
        assert!(!client.session_tasks.is_open());
        assert!(!client.command_tasks.is_open());

        get_runtime()
            .block_on(client.prepare_task_groups())
            .expect("prepare replacement task generation");

        assert!(!client.cancellation_token.is_cancelled());
        assert!(!client.cancellation_token.child_token().is_cancelled());
        assert!(client.session_tasks.is_open());
        assert!(client.command_tasks.is_open());
    }

    #[rstest]
    fn test_stop_retains_bar_timeout_handle_until_async_finish() {
        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
        nautilus_common::live::runner::replace_data_event_sender(sender);

        let config = InteractiveBrokersDataClientConfig::default();

        let provider = Arc::new(InteractiveBrokersInstrumentProvider::new(
            config.instrument_provider.clone(),
        ));
        let mut client =
            InteractiveBrokersDataClient::new(*IB_CLIENT_ID, config, provider).unwrap();
        let bar_type = "AAPL.NASDAQ-1-MINUTE-LAST-EXTERNAL".to_string();

        let task = get_runtime().spawn(async {
            std::future::pending::<()>().await;
        });

        get_runtime().block_on(async {
            client
                .bar_timeout_tasks
                .lock()
                .await
                .insert(bar_type.clone(), TaskSlot::from_handle(task));
            client.last_bars.lock().await.insert(
                bar_type.clone(),
                ibapi::market_data::realtime::Bar {
                    date: time::OffsetDateTime::from_unix_timestamp(1).unwrap(),
                    open: 1.0,
                    high: 1.0,
                    low: 1.0,
                    close: 1.0,
                    volume: 1.0,
                    wap: 1.0,
                    count: 1,
                },
            );
        });

        client.stop().unwrap();

        get_runtime().block_on(async {
            assert_eq!(client.bar_timeout_tasks.lock().await.len(), 1);
            assert!(client.last_bars.lock().await.is_empty());

            client
                .prepare_task_groups()
                .await
                .expect("finish stopped task generation");

            assert!(client.bar_timeout_tasks.lock().await.is_empty());
        });
    }
}