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, execute_bar_subscriptions, execute_instrument_subscriptions},
    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::{
        IbPriceConverter, 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,
        spreads::{is_spread_instrument_id, parse_spread_instrument_id_to_legs},
        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,
            self.config.max_reconnect_attempts,
        )
        .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();

        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,
                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::Data(instrument.into())) {
                    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<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            self.subscribe_quotes_member(instrument_id, &cmd)
        })
    }

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

            if self.has_subscription(instrument_id, SubscriptionType::IndexPrices)? {
                tracing::debug!(
                    "Index price subscription already active for {}",
                    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(&instrument_id)
                .context(format!(
                    "Instrument {} not found in provider",
                    instrument_id
                ))?;

            let contract = self
                .instrument_provider
                .resolve_contract_for_instrument(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,
                    instrument_id
                );
                return Ok(());
            }

            let prices = IbPriceConverter::for_instrument(
                &instrument,
                self.instrument_provider.get_price_magnifier(&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(
                (instrument_id, SubscriptionType::IndexPrices),
                SubscriptionInfo {
                    cancellation_token: subscription_token.clone(),
                    id: subscription_id,
                },
            );

            let guard = SubscriptionGuard {
                subscriptions: Arc::clone(&self.subscriptions),
                key: (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,
                    prices,
                    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 {}", instrument_id);
            Ok(())
        })
    }

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

            if lock(&self.option_greeks_subscriptions)?.contains_key(&instrument_id) {
                tracing::debug!(
                    "Option greeks subscription already active for {}",
                    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(&instrument_id)
                .context(format!(
                    "Instrument {} not found in provider",
                    instrument_id
                ))?;

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

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

            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(instrument_id, subscription_token);

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

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

            let mut subscriptions = lock(&self.subscriptions)?;
            if let Some(sub_info) = subscriptions.remove(&(instrument_id, SubscriptionType::Quotes))
            {
                sub_info.cancellation_token.cancel();
                tracing::debug!("Unsubscribed from quotes for {instrument_id}");
            } else {
                tracing::warn!("No active quote subscription found for {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<()> {
        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
            tracing::debug!("Unsubscribing from index prices for {}", instrument_id);

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

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

            Ok(())
        })
    }

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

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

            Ok(())
        })
    }

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

            if self.has_subscription(instrument_id, SubscriptionType::Trades)? {
                tracing::debug!("Trade subscription already active for {}", 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(&instrument_id)
                .context(format!(
                    "Instrument {} not found in provider",
                    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: {}",
                    instrument_id
                );
                return Ok(());
            }

            let prices = IbPriceConverter::for_instrument(
                &instrument,
                self.instrument_provider.get_price_magnifier(&instrument_id),
            );
            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 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(
                (instrument_id, SubscriptionType::Trades),
                SubscriptionInfo {
                    cancellation_token: subscription_token.clone(),
                    id: subscription_id,
                },
            );

            let guard = SubscriptionGuard {
                subscriptions: Arc::clone(&self.subscriptions),
                key: (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,
                    prices,
                    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 {}", instrument_id);
            Ok(())
        })
    }

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

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

            Ok(())
        })
    }

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

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

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

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

            let prices = IbPriceConverter::for_instrument(
                &instrument,
                self.instrument_provider.get_price_magnifier(&instrument_id),
            );
            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_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,
                        ),
                        prices,
                        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,
                        ),
                        prices,
                        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<()> {
        execute_bar_subscriptions(&cmd.bar_types, |bar_type| {
            tracing::debug!("Unsubscribing from bars for {}", bar_type);

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

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

            Ok(())
        })
    }

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

            if self.has_subscription(instrument_id, SubscriptionType::BookDeltas)? {
                tracing::debug!(
                    "Book delta subscription already active for {}",
                    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(&instrument_id)
                .context(format!(
                    "Instrument {} not found in provider",
                    instrument_id
                ))?;

            // Book levels are not rounded to the increment: two IB levels could round to one price
            let prices = IbPriceConverter::new(instrument.price_precision())
                .with_price_magnifier(self.instrument_provider.get_price_magnifier(&instrument_id));
            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 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(
                (instrument_id, SubscriptionType::BookDeltas),
                SubscriptionInfo {
                    cancellation_token: subscription_token.clone(),
                    id: subscription_id,
                },
            );

            let guard = SubscriptionGuard {
                subscriptions: Arc::clone(&self.subscriptions),
                key: (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,
                    prices,
                    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 {}", instrument_id);
            Ok(())
        })
    }

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

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

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

            Ok(())
        })
    }

    // Request handlers
    fn request_instrument(&self, cmd: RequestInstrument) -> anyhow::Result<()> {
        let instrument_id = cmd.single_instrument_id()?;

        tracing::debug!("Requesting instrument: {instrument_id}");

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

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

        // 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(&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 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) = load_requested_instrument(
                    &instrument_provider,
                    &client_clone,
                    instrument_id,
                    force_update,
                    filters,
                    &data_sender,
                )
                .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(&instrument_id)
                .context(format!("Instrument {instrument_id} not found in provider"))?
        };

        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),
            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<()> {
        let instrument_id = cmd.single_instrument_id()?;

        tracing::debug!("Requesting quotes for {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(&instrument_id)
            .context(format!("Instrument {instrument_id} not found in provider"))?;

        let prices = IbPriceConverter::for_instrument(
            &instrument,
            self.instrument_provider.get_price_magnifier(&instrument_id),
        );
        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 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 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);

        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(tick.price_bid),
                                Some(tick.price_ask),
                                tick.size_bid,
                                tick.size_ask,
                                prices,
                                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<()> {
        let instrument_id = cmd.single_instrument_id()?;

        tracing::debug!("Requesting trades for {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(&instrument_id)
            .context(format!("Instrument {instrument_id} not found in provider"))?;

        // 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: {instrument_id}"
            );
            return Ok(());
        }

        let prices = IbPriceConverter::for_instrument(
            &instrument,
            self.instrument_provider.get_price_magnifier(&instrument_id),
        );
        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 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 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);

        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,
                                tick.price,
                                size,
                                prices,
                                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<()> {
        let bar_type = cmd.single_bar_type()?;

        tracing::debug!("Requesting bars for {bar_type}");

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

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

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

        let prices = IbPriceConverter::for_instrument(
            &instrument,
            self.instrument_provider.get_price_magnifier(&instrument_id),
        );
        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(&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(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 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());

        // 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 {
                            match ib_bar_to_nautilus_bar(
                                ib_bar,
                                bar_type,
                                prices,
                                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();
    }
}

impl InteractiveBrokersDataClient {
    fn subscribe_quotes_member(
        &self,
        instrument_id: InstrumentId,
        cmd: &SubscribeQuotes,
    ) -> anyhow::Result<()> {
        tracing::debug!("Subscribing to quotes for {instrument_id}");

        if self.has_subscription(instrument_id, SubscriptionType::Quotes)? {
            tracing::debug!("Quote subscription already active for {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(&instrument_id)
            .context(format!("Instrument {instrument_id} not found in provider"))?;

        let prices = IbPriceConverter::for_instrument(
            &instrument,
            self.instrument_provider.get_price_magnifier(&instrument_id),
        );
        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")?;

        // 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 data_sender = self.data_sender.clone();
        let quote_cache = Arc::clone(&self.quote_cache);
        let clock = self.clock;

        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(
            (instrument_id, SubscriptionType::Quotes),
            SubscriptionInfo {
                cancellation_token: subscription_token.clone(),
                id: subscription_id,
            },
        );

        let guard = SubscriptionGuard {
            subscriptions: Arc::clone(&self.subscriptions),
            key: (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,
                    prices,
                    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,
                    prices,
                    size_precision,
                    data_sender.clone(),
                    clock,
                    subscription_token.clone(),
                    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,
                            prices,
                            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 {instrument_id} (method: {})",
            if use_market_data {
                "market_data"
            } else {
                "tick_by_tick_bid_ask"
            }
        );
        Ok(())
    }
}

// Loads a requested instrument into the provider. A generic spread is built as an IB BAG from its
// qualified legs, as `load_ids` does, and its legs are published first because spread fills update
// the leg positions.
async fn load_requested_instrument(
    instrument_provider: &InteractiveBrokersInstrumentProvider,
    client: &ibapi::Client,
    instrument_id: InstrumentId,
    force_update: bool,
    filters: Option<HashMap<String, String>>,
    data_sender: &EventSender<DataEvent>,
) -> anyhow::Result<()> {
    if !is_spread_instrument_id(&instrument_id) {
        return instrument_provider
            .fetch_contract_details(client, instrument_id, force_update, filters)
            .await;
    }

    instrument_provider
        .fetch_spread_instrument(client, instrument_id, force_update, filters)
        .await?;
    let leg_ids = parse_spread_instrument_id_to_legs(&instrument_id)?
        .into_iter()
        .map(|(leg_id, _)| leg_id)
        .collect::<Vec<_>>();

    for leg in instrument_provider.find_all(&leg_ids) {
        if let Err(e) = data_sender.send(DataEvent::Data(leg.into())) {
            tracing::error!("Failed to send spread leg instrument: {e}");
        }
    }

    Ok(())
}

#[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()).unwrap(),
        );
        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()).unwrap(),
        );
        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()).unwrap(),
        );
        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());
        });
    }
}