nautilus_interactive_brokers/providers/

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

//! Interactive Brokers instrument provider implementation.

use std::{collections::HashMap, fs, path::Path, str::FromStr, sync::Arc, time::Duration};

use anyhow::Context;
use dashmap::DashMap;
use ibapi::{
    contracts::{ComboLegOpenClose, Contract, Exchange, LegAction, SecurityType, Symbol},
    prelude::StreamExt,
    subscriptions::SubscriptionItem,
};
use jiff::{Span, Timestamp, tz::Offset};
use nautilus_common::{cache::Cache, config::ConfigResult};
use nautilus_model::{
    identifiers::{InstrumentId, Venue},
    instruments::{Instrument, InstrumentAny},
};
use serde::{Deserialize, Serialize};

use crate::{
    common::{
        Symbology,
        contracts::{
            ConfiguredContract, KEY_BUILD_FUTURES_CHAIN, KEY_BUILD_OPTIONS_CHAIN,
            KEY_MAX_EXPIRY_DAYS, KEY_MIN_EXPIRY_DAYS, KEY_OPTIONS_CHAIN_EXCHANGE,
            KEY_OPTIONS_CHAIN_EXCHANGE_ALT, parse_contract_from_json,
        },
        enums::{IbAction, IbSecurityType},
        spreads::{
            create_spread_instrument_id, is_spread_instrument_id,
            parse_spread_instrument_id_to_legs,
        },
        symbology::{
            determine_venue_from_contract, exchange_to_mic_venue, possible_exchanges_for_venue,
            symbol_mic_venue,
        },
    },
    config::{InteractiveBrokersInstrumentProviderConfig, SymbologyMethod},
    providers::parse::{
        expiry_timestring_to_unix_nanos, parse_ib_contract_to_instrument,
        parse_spread_instrument_any, split_last_trade_date, trim_last_trade_date,
    },
};

/// Cache structure for persistent instrument caching.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct InstrumentCache {
    /// Timestamp when cache was created.
    cache_timestamp: Timestamp,
    /// Contract ID to Instrument ID mappings.
    contract_id_to_instrument_id: Vec<(i32, String)>,
    /// Instrument ID to Price Magnifier mappings.
    price_magnifiers: Vec<(String, i32)>,
    /// Instrument ID to IB contracts.
    #[serde(default)]
    contracts: Vec<(String, Contract)>,
    /// Instrument ID to IB contract details.
    #[serde(default)]
    contract_details: Vec<(String, ibapi::contracts::ContractDetails)>,
    /// Instruments serialized as JSON strings (since InstrumentAny is serializable).
    instruments: Vec<(String, String)>, // (instrument_id, json)
}

/// Interactive Brokers instrument provider.
///
/// This provider fetches contract details from Interactive Brokers using the `rust-ibapi` library
/// and converts them to NautilusTrader instruments.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(
        module = "nautilus_trader.adapters.interactive_brokers",
        unsendable,
        from_py_object
    )
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(
        module = "nautilus_trader.adapters.interactive_brokers"
    )
)]
#[derive(Debug, Clone)]
pub struct InteractiveBrokersInstrumentProvider {
    /// Configuration for the provider.
    config: InteractiveBrokersInstrumentProviderConfig,
    /// Contract and instrument ID mapping policy.
    symbology: Symbology,
    /// Cache mapping contract IDs to instrument IDs.
    contract_id_to_instrument_id: Arc<DashMap<i32, InstrumentId>>,
    /// Cache mapping instrument IDs to instruments.
    instruments: Arc<DashMap<InstrumentId, InstrumentAny>>,
    /// Cache mapping instrument IDs to contract details.
    contract_details: Arc<DashMap<InstrumentId, ibapi::contracts::ContractDetails>>,
    /// Cache mapping instrument IDs to IB contracts.
    contracts: Arc<DashMap<InstrumentId, Contract>>,
    /// Dedicated cache for price magnifiers for fast lookups.
    price_magnifiers: Arc<DashMap<InstrumentId, i32>>,
    /// Guards startup loading and records whether every configured input resolved.
    startup_initialized: Arc<tokio::sync::Mutex<bool>>,
}

trait StartupInstrumentLoader {
    async fn load_instrument_id(
        &self,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<Option<InstrumentId>>;

    async fn load_contract(
        &self,
        configured: &ConfiguredContract,
    ) -> anyhow::Result<Vec<InstrumentId>>;
}

struct IbStartupInstrumentLoader<'a> {
    provider: &'a InteractiveBrokersInstrumentProvider,
    client: &'a ibapi::Client,
}

impl StartupInstrumentLoader for IbStartupInstrumentLoader<'_> {
    async fn load_instrument_id(
        &self,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<Option<InstrumentId>> {
        self.provider
            .load_with_return_async(self.client, instrument_id, None)
            .await
    }

    async fn load_contract(
        &self,
        configured: &ConfiguredContract,
    ) -> anyhow::Result<Vec<InstrumentId>> {
        let spec = configured.chain_spec_json();
        self.provider
            .load_contract_spec(self.client, &configured.contract, spec.as_ref())
            .await
    }
}

impl InteractiveBrokersInstrumentProvider {
    /// Create a new `InteractiveBrokersInstrumentProvider`.
    ///
    /// # Errors
    ///
    /// Returns an error if configured contracts contain duplicates.
    pub fn new(config: InteractiveBrokersInstrumentProviderConfig) -> ConfigResult<Self> {
        config.validate()?;

        let symbology = Symbology::new(config.symbology_method);

        Ok(Self {
            config,
            symbology,
            contract_id_to_instrument_id: Arc::new(DashMap::new()),
            instruments: Arc::new(DashMap::new()),
            contract_details: Arc::new(DashMap::new()),
            contracts: Arc::new(DashMap::new()),
            price_magnifiers: Arc::new(DashMap::new()),
            startup_initialized: Arc::new(tokio::sync::Mutex::new(false)),
        })
    }

    #[cfg(test)]
    pub(crate) fn insert_test_instrument(
        &self,
        instrument: InstrumentAny,
        contract_id: i32,
        price_magnifier: i32,
    ) {
        let instrument_id = instrument.id();
        self.instruments.insert(instrument_id, instrument);
        self.contract_id_to_instrument_id
            .insert(contract_id, instrument_id);
        self.contracts.insert(
            instrument_id,
            Contract {
                contract_id,
                ..Default::default()
            },
        );

        self.price_magnifiers.insert(instrument_id, price_magnifier);
    }

    #[cfg(test)]
    pub(crate) fn insert_test_contract_id_mapping(
        &self,
        contract_id: i32,
        instrument_id: InstrumentId,
    ) {
        self.contract_id_to_instrument_id
            .insert(contract_id, instrument_id);
    }

    /// Initialize the provider by loading cache if configured.
    ///
    /// This is equivalent to Python's `provider.initialize()` method.
    /// It loads instruments from cache if `cache_path` is configured and cache is valid.
    ///
    /// # Errors
    ///
    /// Returns an error if configured contracts repeat.
    pub async fn initialize(&self) -> anyhow::Result<()> {
        self.validate()?;

        if let Some(ref cache_path) = self.config.cache_path {
            match self.load_cache(cache_path).await {
                Ok(cache_loaded) => {
                    if cache_loaded {
                        tracing::debug!(
                            "Initialized provider with {} instruments from cache",
                            self.count()
                        );
                    } else {
                        tracing::debug!(
                            "Cache file not found or expired, starting with empty cache"
                        );
                    }
                }
                Err(e) => {
                    tracing::warn!("Failed to load cache during initialization: {}", e);
                }
            }
        }

        Ok(())
    }

    /// Initializes the provider and resolves every configured startup input.
    ///
    /// Successful initialization is idempotent. A failed attempt remains retryable.
    ///
    /// # Errors
    ///
    /// Returns an error if initialization fails or any configured input cannot be loaded.
    pub async fn initialize_with_client(
        &self,
        client: &ibapi::Client,
    ) -> anyhow::Result<Vec<InstrumentId>> {
        let loader = IbStartupInstrumentLoader {
            provider: self,
            client,
        };

        self.initialize_with_loader(&loader).await
    }

    async fn initialize_with_loader<L>(&self, loader: &L) -> anyhow::Result<Vec<InstrumentId>>
    where
        L: StartupInstrumentLoader + Sync,
    {
        let mut initialized = self.startup_initialized.lock().await;
        if *initialized {
            return Ok(Vec::new());
        }

        self.initialize().await?;
        let loaded_ids = self.load_configured_instruments(loader).await?;
        *initialized = true;
        Ok(loaded_ids)
    }

    async fn load_configured_instruments<L>(&self, loader: &L) -> anyhow::Result<Vec<InstrumentId>>
    where
        L: StartupInstrumentLoader + Sync,
    {
        let mut loaded_ids = Vec::new();
        let mut unresolved = Vec::new();
        let mut configured_ids: Vec<_> = self.config.load_ids.iter().copied().collect();
        configured_ids.sort_unstable();

        for instrument_id in configured_ids {
            match loader
                .load_instrument_id(instrument_id)
                .await
                .with_context(|| {
                    format!("Failed to load configured IB instrument ID {instrument_id}")
                })? {
                Some(loaded_id) => loaded_ids.push(loaded_id),
                None => unresolved.push(format!("instrument ID {instrument_id}")),
            }
        }

        for (index, configured) in self.config.load_contracts.iter().enumerate() {
            let contract = &configured.contract;

            let mut contract_ids = loader.load_contract(configured).await.with_context(|| {
                format!("Failed to load configured IB contract at index {index}: {contract:?}")
            })?;

            if contract_ids.is_empty() {
                unresolved.push(format!("contract at index {index}: {contract:?}"));
            } else {
                loaded_ids.append(&mut contract_ids);
            }
        }

        if !unresolved.is_empty() {
            anyhow::bail!(
                "Unable to resolve configured Interactive Brokers instruments: {}",
                unresolved.join(", ")
            );
        }

        loaded_ids.sort_unstable();
        loaded_ids.dedup();
        Ok(loaded_ids)
    }

    pub(crate) fn seed_from_cache(&self, cache: &Cache) -> usize {
        let instruments = cache
            .instrument_ids(None)
            .into_iter()
            .filter_map(|instrument_id| cache.instrument(instrument_id).cloned());
        let count = self.add_cached_instruments(instruments);
        if count > 0 {
            tracing::debug!("Seeded IB instrument provider with {count} cached instruments");
        }

        count
    }

    /// Adds instruments already held by the Nautilus cache into the provider cache.
    ///
    /// This mirrors the Python provider's use of `client._cache` for venue resolution and for
    /// recovering stored IB contract metadata from `instrument.info["contract"]`.
    pub fn add_cached_instruments<I>(&self, instruments: I) -> usize
    where
        I: IntoIterator<Item = InstrumentAny>,
    {
        let mut added = 0;

        for instrument in instruments {
            let instrument_id = instrument.id();

            let Some(contract) = contract_from_instrument_info(&instrument) else {
                continue;
            };

            let Some(price_magnifier) =
                price_magnifier_from_instrument_info(&instrument).filter(|value| *value >= 0)
            else {
                continue;
            };

            if self.cache_instrument(
                instrument_id,
                instrument,
                None,
                Some(contract),
                Some(price_magnifier),
                false,
            ) {
                added += 1;
            }
        }

        added
    }

    /// Determine venue from contract using provider configuration.
    ///
    /// This is equivalent to Python's `determine_venue_from_contract` method.
    /// It uses the config's symbol-to-venue mapping and exchange-to-venue conversion settings.
    ///
    /// # Returns
    ///
    /// The determined venue.
    pub fn determine_venue(
        &self,
        contract: &Contract,
        contract_details: Option<&ibapi::contracts::ContractDetails>,
    ) -> Venue {
        if matches!(contract.security_type, SecurityType::Stock) {
            if let Some(venue) = symbol_mic_venue(contract, &self.config.symbol_to_mic_venue) {
                return Venue::from(venue.as_str());
            }
            return Venue::from(self.resolve_stock_exchange_from_contract(contract).as_str());
        }

        let valid_exchanges = contract_details.map(|details| details.valid_exchanges.join(","));
        let venue_str = determine_venue_from_contract(
            contract,
            &self.config.symbol_to_mic_venue,
            self.config.convert_exchange_to_mic_venue,
            valid_exchanges.as_deref(),
        );
        Venue::from(venue_str.as_str())
    }

    fn resolve_stock_exchange_from_contract(&self, contract: &Contract) -> String {
        let cached_venue = self.resolve_cached_symbol_venue(contract);
        if let Some(venue) = cached_venue.as_deref()
            && Self::is_compatible_cached_stock_venue(venue, contract.primary_exchange.as_str())
        {
            return venue.to_string();
        }

        if !contract.primary_exchange.as_str().is_empty()
            && contract.primary_exchange.as_str() != "SMART"
        {
            return if self.config.convert_exchange_to_mic_venue {
                exchange_to_mic_venue(contract.primary_exchange.as_str())
                    .unwrap_or_else(|| contract.primary_exchange.as_str().to_string())
            } else {
                contract.primary_exchange.as_str().to_string()
            };
        }

        if contract.exchange.as_str() == "SMART"
            && let Some(venue) = cached_venue
        {
            return venue;
        }

        let exchange = contract.exchange.as_str();
        if self.config.convert_exchange_to_mic_venue {
            exchange_to_mic_venue(exchange).unwrap_or_else(|| exchange.to_string())
        } else {
            exchange.to_string()
        }
    }

    fn is_compatible_cached_stock_venue(venue: &str, primary_exchange: &str) -> bool {
        if primary_exchange.is_empty() || primary_exchange == "SMART" {
            return true;
        }

        venue == primary_exchange
            || exchange_to_mic_venue(primary_exchange).is_some_and(|mic| mic == venue)
    }

    fn resolve_cached_symbol_venue(&self, contract: &Contract) -> Option<String> {
        self.instruments.iter().find_map(|entry| {
            let instrument = entry.value();
            let instrument_id = instrument.id();
            (instrument_id.symbol.as_str() == contract.symbol.as_str())
                .then(|| instrument_id.venue.to_string())
        })
    }

    /// Get the symbology method from the provider configuration.
    pub fn symbology_method(&self) -> SymbologyMethod {
        self.symbology.method()
    }

    /// Get an instrument by its ID.
    ///
    /// # Returns
    ///
    /// Returns the instrument if found, `None` otherwise.
    #[must_use]
    pub fn find(&self, instrument_id: &InstrumentId) -> Option<InstrumentAny> {
        self.instruments
            .get(instrument_id)
            .map(|entry| entry.value().clone())
    }

    #[must_use]
    pub(crate) fn find_all(&self, instrument_ids: &[InstrumentId]) -> Vec<InstrumentAny> {
        instrument_ids
            .iter()
            .filter_map(|instrument_id| self.find(instrument_id))
            .collect()
    }

    /// Get an instrument by contract ID.
    ///
    /// # Returns
    ///
    /// Returns the instrument if found, `None` otherwise.
    #[must_use]
    pub fn find_by_contract_id(&self, contract_id: i32) -> Option<InstrumentAny> {
        self.get_instrument_id_by_contract_id(contract_id)
            .and_then(|instrument_id| self.find(&instrument_id))
    }

    /// Get an instrument ID by contract ID.
    ///
    /// # Returns
    ///
    /// Returns the instrument ID if found, `None` otherwise.
    #[must_use]
    pub fn get_instrument_id_by_contract_id(&self, contract_id: i32) -> Option<InstrumentId> {
        self.contract_id_to_instrument_id
            .get(&contract_id)
            .map(|entry| *entry.value())
    }

    /// Resolve an instrument ID from an IB contract using provider symbology and venue rules.
    ///
    /// This first checks the provider's contract ID cache, then derives the instrument ID using the
    /// configured symbology method and `determine_venue`.
    ///
    /// # Errors
    ///
    /// Returns an error if the contract cannot be converted to an instrument ID.
    pub fn resolve_instrument_id_for_contract(
        &self,
        contract: &Contract,
    ) -> anyhow::Result<InstrumentId> {
        if contract.contract_id != 0
            && let Some(instrument_id) = self.get_instrument_id_by_contract_id(contract.contract_id)
        {
            return Ok(instrument_id);
        }

        if contract.security_type == SecurityType::Spread {
            return self.resolve_spread_instrument_id_for_contract(contract);
        }

        let venue = self.determine_venue(contract, None);

        self.symbology.instrument_id(contract, Some(venue))
    }

    fn resolve_spread_instrument_id_for_contract(
        &self,
        contract: &Contract,
    ) -> anyhow::Result<InstrumentId> {
        let spread_instrument_id = self.spread_instrument_id_for_contract(contract)?;

        if self.find(&spread_instrument_id).is_none() {
            anyhow::bail!("Resolved BAG spread {spread_instrument_id} is not cached");
        }

        Ok(spread_instrument_id)
    }

    /// Returns the spread instrument ID of a BAG contract from its cached leg instruments.
    ///
    /// # Errors
    ///
    /// Returns an error if the contract has no combo legs, a leg is not cached, or a leg action
    /// is invalid.
    pub fn spread_instrument_id_for_contract(
        &self,
        contract: &Contract,
    ) -> anyhow::Result<InstrumentId> {
        if contract.combo_legs.is_empty() {
            anyhow::bail!("Cannot resolve BAG contract without combo legs or cached contract ID");
        }

        let mut leg_tuples = Vec::with_capacity(contract.combo_legs.len());

        for combo_leg in &contract.combo_legs {
            let leg_instrument_id = self
                .get_instrument_id_by_contract_id(combo_leg.contract_id)
                .with_context(|| {
                    format!(
                        "Cannot resolve BAG leg con_id {} to cached instrument ID",
                        combo_leg.contract_id
                    )
                })?;

            let ratio = IbAction::from_str(combo_leg.action.as_str())
                .context("Invalid BAG combo leg action")?
                .signed_multiplier()
                * combo_leg.ratio;

            leg_tuples.push((leg_instrument_id, ratio));
        }

        create_spread_instrument_id(&leg_tuples)
            .context("Failed to create spread instrument ID from BAG combo legs")
    }

    /// Check if a security type should be filtered.
    ///
    /// # Returns
    ///
    /// Returns `true` if the security type should be filtered.
    #[must_use]
    pub fn is_filtered_sec_type(&self, sec_type: &str) -> bool {
        IbSecurityType::from_str(sec_type)
            .is_ok_and(|sec_type| self.config.filter_sec_types.contains(&sec_type))
    }

    /// Get all cached instruments.
    ///
    /// # Returns
    ///
    /// Returns a vector of all cached instruments.
    #[must_use]
    pub fn get_all(&self) -> Vec<InstrumentAny> {
        self.instruments
            .iter()
            .map(|entry| entry.value().clone())
            .collect()
    }

    /// Get the number of cached instruments.
    ///
    /// # Returns
    ///
    /// Returns the number of cached instruments.
    #[must_use]
    pub fn count(&self) -> usize {
        self.instruments.len()
    }

    /// Get price magnifier for an instrument ID.
    ///
    /// Price magnifier allows execution and strike prices to be reported consistently
    /// with market data and historical data.
    ///
    /// This method first checks the dedicated price magnifier cache for fast lookup.
    /// If not found, it falls back to checking contract details. If still not found,
    /// it returns the default value of 1 and logs a warning if the instrument exists.
    ///
    /// # Returns
    ///
    /// Returns the price magnifier if found, otherwise 1.
    #[must_use]
    pub fn get_price_magnifier(&self, instrument_id: &InstrumentId) -> i32 {
        if let Some(magnifier) = self.price_magnifier(instrument_id) {
            return magnifier;
        }

        // Not found - check if instrument exists (might not have contract details loaded yet)
        if self.instruments.contains_key(instrument_id) {
            tracing::debug!(
                "Price magnifier not found for instrument {} (has instrument but no contract details), using default 1",
                instrument_id
            );
        } else {
            tracing::trace!(
                "Price magnifier not found for instrument {} (instrument not loaded), using default 1",
                instrument_id
            );
        }

        // Default to 1 if not found
        1
    }

    /// Returns the price magnifier the provider holds for an instrument, from its IB contract
    /// details or cached instrument metadata, or `None` when it holds none.
    #[must_use]
    pub fn price_magnifier(&self, instrument_id: &InstrumentId) -> Option<i32> {
        if let Some(magnifier) = self.price_magnifiers.get(instrument_id) {
            return Some(normalize_price_magnifier(*magnifier.value()));
        }

        let magnifier = normalize_price_magnifier(
            self.contract_details
                .get(instrument_id)?
                .value()
                .price_magnifier,
        );
        self.price_magnifiers.insert(*instrument_id, magnifier);
        Some(magnifier)
    }

    /// Seeds an instrument's IB contract and price magnifier from the Nautilus cache when the
    /// provider holds no price magnifier for it, and returns the magnifier the provider then
    /// holds.
    ///
    /// An instrument another client loads after this provider was seeded reaches the cache, with
    /// its contract and magnifier in `info`, but not this provider.
    pub(crate) fn seed_price_magnifier_from_cache(
        &self,
        cache: &Cache,
        instrument_id: &InstrumentId,
    ) -> Option<i32> {
        if let Some(magnifier) = self.price_magnifier(instrument_id) {
            return Some(magnifier);
        }

        if let Some(instrument) = cache.instrument(instrument_id) {
            self.add_cached_instruments([instrument.clone()]);
        }

        self.price_magnifier(instrument_id)
    }

    /// Returns an instrument's price magnifier, requesting its contract details from IB when the
    /// provider holds none and waiting at most `timeout` for them.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails, does not complete within `timeout`, or IB returns
    /// no contract details for the instrument, so its price magnifier is unknown.
    pub async fn ensure_price_magnifier_async(
        &self,
        client: &ibapi::Client,
        instrument_id: InstrumentId,
        timeout: Duration,
    ) -> anyhow::Result<i32> {
        self.ensure_price_magnifier_with(instrument_id, timeout, || async {
            if is_spread_instrument_id(&instrument_id) {
                self.fetch_spread_instrument(client, instrument_id, true, None)
                    .await
                    .map(|_| ())
            } else {
                self.fetch_contract_details(client, instrument_id, true, None)
                    .await
            }
        })
        .await
    }

    async fn ensure_price_magnifier_with<F, Fut>(
        &self,
        instrument_id: InstrumentId,
        timeout: Duration,
        load: F,
    ) -> anyhow::Result<i32>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = anyhow::Result<()>>,
    {
        if let Some(magnifier) = self.price_magnifier(&instrument_id) {
            return Ok(magnifier);
        }

        // IB can keep the connection open without completing the response, so the wait is
        // bounded and the command is denied rather than left pending
        tokio::time::timeout(timeout, load())
            .await
            .map_err(|_| {
                anyhow::anyhow!(
                    "Timed out after {timeout:?} loading IB contract details for {instrument_id}"
                )
            })?
            .with_context(|| format!("Failed to load IB contract details for {instrument_id}"))?;

        self.price_magnifier(&instrument_id).with_context(|| {
            format!("IB price magnifier for {instrument_id} is unknown: no contract details loaded")
        })
    }

    /// Get an instrument by IB Contract.
    ///
    /// This is equivalent to Python's `get_instrument` method.
    /// Supports BAG contracts by auto-loading legs.
    ///
    /// # Returns
    ///
    /// Returns the instrument if found, `None` otherwise.
    ///
    /// # Errors
    ///
    /// Returns an error if fetching fails.
    pub async fn get_instrument(
        &self,
        client: &ibapi::Client,
        contract: &Contract,
    ) -> anyhow::Result<Option<InstrumentAny>> {
        log::debug!(
            "IB get_instrument request sec_type={:?} con_id={} symbol={} local_symbol={} exchange={} expiry={}",
            contract.security_type,
            contract.contract_id,
            contract.symbol.as_str(),
            contract.local_symbol.as_str(),
            contract.exchange.as_str(),
            contract.last_trade_date_or_contract_month.as_str()
        );
        // Check if security type is filtered
        let sec_type_str = contract.security_type.to_string();
        if self.is_filtered_sec_type(&sec_type_str) {
            tracing::debug!(
                "Skipping filtered security type {} for contract",
                sec_type_str
            );
            return Ok(None);
        }

        let contract_id = contract.contract_id;

        // Check if we already have this instrument by contract ID
        if let Some(cached_instrument_id) = self.contract_id_to_instrument_id.get(&contract_id) {
            log::debug!(
                "IB get_instrument cache hit for contract_id={} -> {}",
                contract_id,
                cached_instrument_id.value()
            );

            if let Some(instrument) = self.find(cached_instrument_id.value()) {
                return Ok(Some(instrument));
            }
        }

        // Special handling for BAG contracts
        if contract.security_type == SecurityType::Spread && !contract.combo_legs.is_empty() {
            // Load BAG contract (which auto-loads legs and creates spread instrument)
            self.fetch_bag_contract(client, contract).await?;

            // Get the spread instrument ID that was created
            if let Some(spread_instrument_id) = self.contract_id_to_instrument_id.get(&contract_id)
            {
                return Ok(self.find(spread_instrument_id.value()));
            }

            if let Ok(spread_instrument_id) =
                self.resolve_spread_instrument_id_for_contract(contract)
            {
                return Ok(self.find(&spread_instrument_id));
            }
        }

        // For non-BAG contracts, fetch contract details and load
        let details_vec = request_contract_details(client, contract)
            .await
            .context("Failed to fetch contract details from IB")?;

        log::debug!(
            "IB get_instrument received {} contract details for sec_type={:?} symbol={} local_symbol={}",
            details_vec.len(),
            contract.security_type,
            contract.symbol.as_str(),
            contract.local_symbol.as_str()
        );

        if details_vec.is_empty() {
            tracing::warn!("No contract details returned for contract {}", contract_id);
            return Ok(None);
        }

        let loaded_ids = self.process_contract_details(details_vec, None, false);

        if contract_id != 0
            && let Some(instrument) = self.find_by_contract_id(contract_id)
        {
            return Ok(Some(instrument));
        }

        Ok(loaded_ids
            .first()
            .and_then(|instrument_id| self.find(instrument_id)))
    }

    pub(crate) async fn load_contract_spec(
        &self,
        client: &ibapi::Client,
        contract: &Contract,
        spec: Option<&serde_json::Value>,
    ) -> anyhow::Result<Vec<InstrumentId>> {
        let mut loaded_ids = Vec::new();
        let build_futures_chain = json_opt_bool(spec, KEY_BUILD_FUTURES_CHAIN)
            .or(self.config.build_futures_chain)
            .unwrap_or(false);
        let build_options_chain = json_opt_bool(spec, KEY_BUILD_OPTIONS_CHAIN)
            .or(self.config.build_options_chain)
            .unwrap_or(false);
        let min_expiry_days = json_u32(spec, KEY_MIN_EXPIRY_DAYS).or(self.config.min_expiry_days);
        let max_expiry_days = json_u32(spec, KEY_MAX_EXPIRY_DAYS).or(self.config.max_expiry_days);
        let options_chain_exchange = json_string(spec, KEY_OPTIONS_CHAIN_EXCHANGE)
            .or_else(|| json_string(spec, KEY_OPTIONS_CHAIN_EXCHANGE_ALT));

        let chain_contract = if contract.security_type == SecurityType::ContinuousFuture
            && (build_futures_chain || build_options_chain)
        {
            match request_contract_details(client, contract).await {
                Ok(details_vec) => details_vec
                    .into_iter()
                    .next().map_or_else(|| contract.clone(), |details| {
                        tracing::debug!(
                            "Qualified continuous future contract {}.{} as local_symbol={} trading_class={} con_id={}",
                            contract.symbol.as_str(),
                            contract.exchange.as_str(),
                            details.contract.local_symbol.as_str(),
                            details.contract.trading_class.as_str(),
                            details.contract.contract_id,
                        );
                        details.contract
                    }),
                Err(e) if e.is_connection_lost() => {
                    return Err(e).context("Failed to qualify continuous future contract");
                }
                Err(e) => {
                    tracing::warn!(
                        "Failed to qualify continuous future contract {:?}: {}",
                        contract,
                        e
                    );
                    contract.clone()
                }
            }
        } else {
            contract.clone()
        };

        let chain_trading_class = (!chain_contract.trading_class.is_empty())
            .then_some(chain_contract.trading_class.as_str());

        if build_futures_chain {
            let loaded = self
                .fetch_futures_chain(
                    client,
                    chain_contract.symbol.as_str(),
                    chain_contract.exchange.as_str(),
                    chain_contract.currency.as_str(),
                    chain_trading_class,
                    contract.security_type == SecurityType::ContinuousFuture,
                    min_expiry_days,
                    max_expiry_days,
                )
                .await?;
            tracing::debug!(
                "Loaded {} futures instruments for chain request {}.{}",
                loaded,
                chain_contract.symbol.as_str(),
                chain_contract.exchange.as_str(),
            );
            loaded_ids.extend(self.cached_contract_ids_for(
                chain_contract.symbol.as_str(),
                chain_contract.exchange.as_str(),
                &[SecurityType::Future],
            ));
        }

        if build_options_chain {
            let expiry_min = expiry_bound_from_days(min_expiry_days);
            let expiry_max = expiry_bound_from_days(max_expiry_days);
            // Options list on the requested exchange, which can differ from the exchange the
            // underlying resolves to (SPY requested on SMART resolves to ARCA)
            let options_exchange = options_chain_exchange
                .as_deref()
                .unwrap_or_else(|| contract.exchange.as_str());
            let mut underlyings = Vec::new();

            if contract.security_type == SecurityType::ContinuousFuture {
                // The expiry window bounds the options, not their underlying futures: a weekly
                // ES option expiring this week belongs to a future that expires months later
                self.fetch_futures_chain(
                    client,
                    chain_contract.symbol.as_str(),
                    chain_contract.exchange.as_str(),
                    chain_contract.currency.as_str(),
                    chain_trading_class,
                    true,
                    Some(0),
                    Some(u32::MAX),
                )
                .await?;

                underlyings.extend(
                    self.cached_contracts_for(
                        contract.symbol.as_str(),
                        chain_contract.exchange.as_str(),
                        &[SecurityType::Future],
                    )
                    .into_iter()
                    .map(|(_, contract)| contract),
                );
            } else if let Some(instrument) = self.get_instrument(client, contract).await? {
                let instrument_id = instrument.id();
                loaded_ids.push(instrument_id);
                if let Some(underlying) = self.instrument_id_to_ib_contract(&instrument_id) {
                    underlyings.push(underlying);
                }
            }

            for underlying in underlyings {
                let loaded = self
                    .fetch_option_chain_by_range(
                        client,
                        &underlying,
                        expiry_min.as_deref(),
                        expiry_max.as_deref(),
                        Some(options_exchange),
                    )
                    .await?;
                tracing::debug!(
                    "Loaded {} option instruments for chain request {}.{}",
                    loaded,
                    underlying.symbol.as_str(),
                    underlying.exchange.as_str(),
                );
            }

            loaded_ids.extend(self.cached_contract_ids_for(
                contract.symbol.as_str(),
                options_exchange,
                &[SecurityType::Option, SecurityType::FuturesOption],
            ));
        }

        if !build_futures_chain
            && !build_options_chain
            && let Some(instrument) = self.get_instrument(client, contract).await?
        {
            loaded_ids.push(instrument.id());
        }

        loaded_ids.sort_unstable();
        loaded_ids.dedup();
        Ok(loaded_ids)
    }

    fn cached_contract_ids_for(
        &self,
        symbol: &str,
        exchange: &str,
        security_types: &[SecurityType],
    ) -> Vec<InstrumentId> {
        self.cached_contracts_for(symbol, exchange, security_types)
            .into_iter()
            .map(|(instrument_id, _)| instrument_id)
            .collect()
    }

    fn cached_contracts_for(
        &self,
        symbol: &str,
        exchange: &str,
        security_types: &[SecurityType],
    ) -> Vec<(InstrumentId, Contract)> {
        self.contracts
            .iter()
            .filter_map(|entry| {
                let instrument_id = *entry.key();
                let contract = entry.value();
                let exchange_matches =
                    exchange.is_empty() || contract.exchange.as_str() == exchange;
                if contract.symbol.as_str() == symbol
                    && exchange_matches
                    && security_types.contains(&contract.security_type)
                {
                    Some((instrument_id, contract.clone()))
                } else {
                    None
                }
            })
            .collect()
    }

    /// Convert an instrument ID to IB contract details.
    ///
    /// This is equivalent to Python's `instrument_id_to_ib_contract_details` method.
    ///
    /// # Returns
    ///
    /// Returns the contract details if found, `None` otherwise.
    #[must_use]
    pub fn instrument_id_to_ib_contract_details(
        &self,
        instrument_id: &InstrumentId,
    ) -> Option<ibapi::contracts::ContractDetails> {
        self.contract_details
            .get(instrument_id)
            .map(|entry| entry.value().clone())
    }

    #[must_use]
    pub fn instrument_id_to_ib_contract(&self, instrument_id: &InstrumentId) -> Option<Contract> {
        self.contracts
            .get(instrument_id)
            .map(|entry| entry.value().clone())
    }

    pub fn resolve_contract_for_instrument(
        &self,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<Contract> {
        let cached_contract = self.instrument_id_to_ib_contract(&instrument_id);
        if let Some(contract) = cached_contract.as_ref()
            && (contract.contract_id != 0 || is_spread_instrument_id(&instrument_id))
        {
            return Ok(contract.clone());
        }

        if let Some(details) = self.instrument_id_to_ib_contract_details(&instrument_id) {
            return Ok(details.contract);
        }

        if let Some(contract) = cached_contract {
            return Ok(contract);
        }

        let currency = self
            .instruments
            .get(&instrument_id)
            .map(|instrument| instrument.quote_currency().code.to_string());
        self.symbology
            .contract_with_currency(instrument_id, None, currency.as_deref())
    }

    pub async fn resolve_contract_for_instrument_async(
        &self,
        client: &ibapi::Client,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<Contract> {
        if let Ok(contract) = self.resolve_contract_for_instrument(instrument_id)
            && (contract.contract_id != 0 || self.contract_details.contains_key(&instrument_id))
        {
            return Ok(contract);
        }

        if is_spread_instrument_id(&instrument_id) {
            self.fetch_spread_instrument(client, instrument_id, false, None)
                .await?;
        } else {
            self.fetch_contract_details(client, instrument_id, false, None)
                .await?;
        }

        self.resolve_contract_for_instrument(instrument_id)
    }

    /// Load a single instrument and return the loaded instrument ID.
    ///
    /// This is equivalent to Python's `load_with_return_async` method.
    ///
    /// # Returns
    ///
    /// Returns the loaded instrument ID if successful, `None` otherwise.
    ///
    /// # Errors
    ///
    /// Returns an error if loading fails.
    pub async fn load_with_return_async(
        &self,
        client: &ibapi::Client,
        instrument_id: InstrumentId,
        filters: Option<HashMap<String, String>>,
    ) -> anyhow::Result<Option<InstrumentId>> {
        let filters: Option<HashMap<String, String>> = filters;
        let force_instrument_update = filters
            .as_ref()
            .and_then(|f| f.get("force_instrument_update"))
            .is_some_and(|v| v == "true");

        if is_spread_instrument_id(&instrument_id) {
            self.fetch_spread_instrument(client, instrument_id, force_instrument_update, filters)
                .await?;
        } else {
            self.fetch_contract_details(client, instrument_id, force_instrument_update, filters)
                .await?;
        }

        if self.instruments.contains_key(&instrument_id) {
            Ok(Some(instrument_id))
        } else {
            Ok(None)
        }
    }

    /// Load multiple instruments and return the loaded instrument IDs.
    ///
    /// This is equivalent to Python's `load_ids_with_return_async` method.
    ///
    /// # Returns
    ///
    /// Returns a vector of successfully loaded instrument IDs.
    ///
    /// # Errors
    ///
    /// Returns an error if loading fails.
    pub async fn load_ids_with_return_async(
        &self,
        client: &ibapi::Client,
        instrument_ids: Vec<InstrumentId>,
        filters: Option<HashMap<String, String>>,
    ) -> anyhow::Result<Vec<InstrumentId>> {
        let mut loaded_ids = Vec::new();

        for instrument_id in instrument_ids {
            match self
                .load_with_return_async(client, instrument_id, filters.clone())
                .await
            {
                Ok(Some(loaded_id)) => loaded_ids.push(loaded_id),
                Ok(None) => {}
                Err(e) => {
                    tracing::warn!("Failed to load instrument {}: {}", instrument_id, e);
                }
            }
        }

        Ok(loaded_ids)
    }

    fn create_bag_contract_from_legs(
        &self,
        leg_contract_details: &[(ibapi::contracts::ContractDetails, i32)],
        instrument_id: Option<InstrumentId>,
        bag_contract: Option<&Contract>,
    ) -> anyhow::Result<Contract> {
        if let Some(bag_contract) = bag_contract {
            return Ok(bag_contract.clone());
        }

        let (first_details, _) = leg_contract_details
            .first()
            .ok_or_else(|| anyhow::anyhow!("Cannot create BAG contract without leg details"))?;

        let combo_legs = leg_contract_details
            .iter()
            .map(|(details, ratio)| ibapi::contracts::ComboLeg {
                contract_id: details.contract.contract_id,
                ratio: ratio.abs(),
                action: if *ratio > 0 {
                    LegAction::Buy
                } else {
                    LegAction::Sell
                },
                exchange: details.contract.exchange.to_string(),
                open_close: ComboLegOpenClose::Same,
                short_sale_slot: 0,
                designated_location: String::new(),
                exempt_code: -1,
            })
            .collect();

        Ok(Contract {
            symbol: first_details.contract.symbol.clone(),
            security_type: SecurityType::Spread,
            exchange: Exchange::from("SMART"),
            currency: first_details.contract.currency.clone(),
            local_symbol: instrument_id.map_or_else(String::new, |id| id.symbol.to_string()),
            combo_legs_description: instrument_id.map_or_else(
                || "Spread".to_string(),
                |id| format!("Spread: {}", id.symbol),
            ),
            combo_legs,
            ..Default::default()
        })
    }

    /// Fetch a spread instrument by loading its individual legs.
    ///
    /// This is equivalent to Python's `_fetch_spread_instrument` method.
    /// It parses the spread instrument ID to extract leg tuples, loads each leg,
    /// and then creates the spread instrument.
    ///
    /// # Returns
    ///
    /// Returns `true` if the spread instrument was successfully loaded, `false` otherwise.
    ///
    /// # Errors
    ///
    /// Returns an error if parsing or loading fails.
    pub async fn fetch_spread_instrument(
        &self,
        client: &ibapi::Client,
        spread_instrument_id: InstrumentId,
        force_instrument_update: bool,
        filters: Option<HashMap<String, String>>,
    ) -> anyhow::Result<bool> {
        // Check if already cached (unless forcing update)
        if !force_instrument_update && self.instruments.contains_key(&spread_instrument_id) {
            tracing::debug!("Spread instrument {} already cached", spread_instrument_id);
            return Ok(true);
        }

        // Parse the spread ID to get individual legs
        let leg_tuples = parse_spread_instrument_id_to_legs(&spread_instrument_id)
            .context("Failed to parse spread instrument ID to leg tuples")?;

        if leg_tuples.is_empty() {
            tracing::error!("Spread instrument {} has no legs", spread_instrument_id);
            return Ok(false);
        }

        tracing::debug!(
            "Loading spread instrument {} with {} legs",
            spread_instrument_id,
            leg_tuples.len()
        );

        // First, load all individual leg instruments to get their contract details
        let mut leg_contract_details = Vec::new();

        for (leg_instrument_id, ratio) in &leg_tuples {
            tracing::debug!(
                "Loading leg instrument: {} (ratio: {})",
                leg_instrument_id,
                ratio
            );

            // Load the individual leg instrument
            self.fetch_contract_details(
                client,
                *leg_instrument_id,
                force_instrument_update,
                filters.clone(),
            )
            .await
            .with_context(|| format!("Failed to load leg instrument: {leg_instrument_id}"))?;

            // Get the contract details for this leg
            let leg_details = self
                .contract_details
                .get(leg_instrument_id)
                .map(|entry| entry.value().clone())
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "Leg instrument {leg_instrument_id} not found in contract details after loading"
                    )
                })?;

            leg_contract_details.push((leg_details, *ratio));
        }

        // Create the spread instrument
        let timestamp = nautilus_core::time::get_atomic_clock_realtime().get_time_ns();
        let leg_details_refs: Vec<(&ibapi::contracts::ContractDetails, i32)> =
            leg_contract_details.iter().map(|(d, r)| (d, *r)).collect();

        let bag_contract = self.create_bag_contract_from_legs(
            &leg_contract_details,
            Some(spread_instrument_id),
            None,
        )?;
        let spread_instrument = parse_spread_instrument_any(
            spread_instrument_id,
            &leg_details_refs,
            Some(&bag_contract),
            Some(timestamp),
        )
        .context("Failed to parse spread instrument")?;

        // Cache the spread instrument
        self.instruments
            .insert(spread_instrument_id, spread_instrument);
        self.contracts.insert(spread_instrument_id, bag_contract);

        if let Some((first_details, _)) = leg_contract_details.first() {
            self.price_magnifiers
                .insert(spread_instrument_id, first_details.price_magnifier);
        }

        tracing::debug!(
            "Successfully loaded spread instrument {}",
            spread_instrument_id
        );
        Ok(true)
    }

    /// Load all instruments from provided IDs and contracts.
    ///
    /// This is equivalent to Python's `load_all_async` method.
    /// Python version loads from config's `_load_ids_on_start` and `_load_contracts_on_start`.
    /// Rust version accepts these as parameters for flexibility.
    ///
    /// # Errors
    ///
    /// Returns an error if loading fails.
    pub async fn load_all_async(
        &self,
        client: &ibapi::Client,
        instrument_ids: Option<Vec<InstrumentId>>,
        contracts: Option<Vec<Contract>>,
        force_instrument_update: bool,
    ) -> anyhow::Result<Vec<InstrumentId>> {
        self.validate()?;
        let mut loaded_ids = Vec::new();

        // Load from instrument IDs
        let ids_to_load =
            instrument_ids.unwrap_or_else(|| self.config.load_ids.iter().copied().collect());

        if !ids_to_load.is_empty() {
            let mut filters = std::collections::HashMap::new();
            if force_instrument_update {
                filters.insert("force_instrument_update".to_string(), "true".to_string());
            }

            let filters = if filters.is_empty() {
                None
            } else {
                Some(filters)
            };

            let ids_result = self
                .load_ids_with_return_async(client, ids_to_load, filters)
                .await
                .context("Failed to load instruments from IDs")?;
            loaded_ids.extend(ids_result);
        }

        // Load from contracts
        if let Some(contracts_to_load) = contracts {
            for contract in contracts_to_load {
                match self.load_contract_spec(client, &contract, None).await {
                    Ok(mut instrument_ids) => {
                        loaded_ids.append(&mut instrument_ids);
                    }
                    Err(e) => {
                        tracing::warn!(
                            "Error loading instrument from contract {:?}: {}",
                            contract,
                            e
                        );
                    }
                }
            }
        } else {
            for configured in &self.config.load_contracts {
                let spec = configured.chain_spec_json();

                match self
                    .load_contract_spec(client, &configured.contract, spec.as_ref())
                    .await
                {
                    Ok(mut instrument_ids) => {
                        loaded_ids.append(&mut instrument_ids);
                    }
                    Err(e) => {
                        tracing::warn!(
                            "Error loading instrument from contract {:?}: {}",
                            configured.contract,
                            e
                        );
                    }
                }
            }
        }

        if loaded_ids.is_empty() {
            tracing::debug!("load_all_async called but no instruments were loaded");
        } else {
            tracing::debug!("load_all_async loaded {} instruments", loaded_ids.len());
        }

        Ok(loaded_ids)
    }

    fn validate(&self) -> ConfigResult<()> {
        self.config.validate()
    }
}

fn normalize_price_magnifier(price_magnifier: i32) -> i32 {
    if price_magnifier > 0 {
        price_magnifier
    } else {
        1
    }
}

// Requests contract details with each expiry's time split from its date, so the contract can
// be sent back to IB (see `split_last_trade_date`)
async fn request_contract_details(
    client: &ibapi::Client,
    contract: &Contract,
) -> Result<Vec<ibapi::contracts::ContractDetails>, ibapi::Error> {
    let mut details_vec = client.contract_details(contract).await?;
    details_vec.iter_mut().for_each(split_last_trade_date);
    Ok(details_vec)
}

fn json_opt_bool(spec: Option<&serde_json::Value>, key: &str) -> Option<bool> {
    spec.and_then(|value| value.get(key))
        .and_then(serde_json::Value::as_bool)
}

fn json_u32(spec: Option<&serde_json::Value>, key: &str) -> Option<u32> {
    spec.and_then(|value| value.get(key))
        .and_then(serde_json::Value::as_u64)
        .and_then(|value| u32::try_from(value).ok())
}

fn json_string(spec: Option<&serde_json::Value>, key: &str) -> Option<String> {
    spec.and_then(|value| value.get(key))
        .and_then(serde_json::Value::as_str)
        .filter(|value| !value.is_empty())
        .map(ToString::to_string)
}

fn contract_from_instrument_info(instrument: &InstrumentAny) -> Option<Contract> {
    let value = serde_json::to_value(instrument).ok()?;
    let contract_json = find_contract_json(&value)?;
    let mut contract = parse_contract_from_json(contract_json).ok()?;
    // An instrument stored before expiries were split may still hold one with its time
    trim_last_trade_date(&mut contract);
    Some(contract)
}

fn price_magnifier_from_instrument_info(instrument: &InstrumentAny) -> Option<i32> {
    let value = serde_json::to_value(instrument).ok()?;
    let price_magnifier = find_price_magnifier_json(&value)?;
    parse_i32_json(price_magnifier)
}

fn find_contract_json(value: &serde_json::Value) -> Option<&serde_json::Value> {
    if let Some(contract_json) = value.get("info").and_then(|info| info.get("contract")) {
        return Some(contract_json);
    }

    value.as_object()?.values().find_map(find_contract_json)
}

fn find_price_magnifier_json(value: &serde_json::Value) -> Option<&serde_json::Value> {
    if let Some(info) = value.get("info")
        && let Some(price_magnifier) = info
            .get("priceMagnifier")
            .or_else(|| info.get("price_magnifier"))
    {
        return Some(price_magnifier);
    }

    value
        .as_object()?
        .values()
        .find_map(find_price_magnifier_json)
}

fn parse_i32_json(value: &serde_json::Value) -> Option<i32> {
    if let Some(value) = value.as_i64() {
        return i32::try_from(value).ok();
    }

    if let Some(value) = value.as_u64() {
        return i32::try_from(value).ok();
    }

    value.as_str()?.parse::<i32>().ok()
}

fn expiry_bound_from_days(days: Option<u32>) -> Option<String> {
    days.map(|days| {
        Offset::UTC
            .to_datetime(Timestamp::now())
            .date()
            .checked_add(Span::new().days(i64::from(days)))
            .expect("expiry bound date in range")
            .strftime("%Y%m%d")
            .to_string()
    })
}

impl InteractiveBrokersInstrumentProvider {
    /// Fetch and cache contract details for an instrument ID using the provided IB client.
    ///
    /// # Errors
    ///
    /// Returns an error if fetching fails.
    pub async fn fetch_contract_details(
        &self,
        client: &ibapi::Client,
        instrument_id: InstrumentId,
        force_instrument_update: bool,
        filters: Option<HashMap<String, String>>,
    ) -> anyhow::Result<()> {
        if !force_instrument_update
            && self.instruments.contains_key(&instrument_id)
            && (self.contract_details.contains_key(&instrument_id)
                || self.contracts.contains_key(&instrument_id))
        {
            tracing::debug!(
                "Instrument {} already cached, skipping fetch",
                instrument_id
            );
            return Ok(());
        }

        // Convert instrument ID to IB contract
        let exchange = filters
            .as_ref()
            .and_then(|f| f.get("exchange"))
            .map(|s| s.as_str());

        let exchanges_to_try: Vec<String> = if let Some(exchange) = exchange {
            vec![exchange.to_string()]
        } else {
            possible_exchanges_for_venue(instrument_id.venue.as_str())
        };

        let mut details_vec = Vec::new();
        let mut last_error = None;

        for candidate_exchange in exchanges_to_try {
            let contract = self
                .symbology
                .contract(instrument_id, Some(candidate_exchange.as_str()))
                .with_context(|| format!("Failed to convert instrument_id {instrument_id} to IB contract. Check that the instrument ID format is correct and the venue/symbol are valid."))?;

            match request_contract_details(client, &contract).await {
                Ok(result) if !result.is_empty() => {
                    details_vec = result;
                    break;
                }
                Ok(_) => {}
                Err(e) => {
                    last_error = Some((candidate_exchange.clone(), e));
                }
            }
        }

        if details_vec.is_empty() {
            if let Some((candidate_exchange, e)) = last_error {
                return Err(e).with_context(|| {
                    format!(
                        "Failed to fetch contract details for {instrument_id} on {candidate_exchange}"
                    )
                });
            } else {
                tracing::warn!(
                    "No contract details returned for {} - instrument may not exist in IB or contract specification is incomplete",
                    instrument_id
                );
            }

            return Ok(());
        }

        let loaded_ids = self.process_contract_details(
            details_vec,
            Some(instrument_id.venue),
            force_instrument_update,
        );

        if loaded_ids.is_empty() {
            tracing::warn!("No contract details were processed for {}", instrument_id);
        } else {
            tracing::debug!(
                "Successfully loaded {} instrument(s) for {}",
                loaded_ids.len(),
                instrument_id
            );
        }

        Ok(())
    }

    fn process_contract_details(
        &self,
        details_vec: Vec<ibapi::contracts::ContractDetails>,
        venue: Option<Venue>,
        force_instrument_update: bool,
    ) -> Vec<InstrumentId> {
        let mut processed_ids = Vec::new();

        for details in details_vec {
            match self.process_contract_detail(&details, venue, force_instrument_update) {
                Ok(Some(instrument_id)) => processed_ids.push(instrument_id),
                Ok(None) => {}
                Err(e) => {
                    tracing::warn!(
                        "Failed to process IB contract details con_id={} sec_type={}: {}",
                        details.contract.contract_id,
                        details.contract.security_type,
                        e
                    );
                }
            }
        }

        processed_ids
    }

    fn process_contract_detail(
        &self,
        details: &ibapi::contracts::ContractDetails,
        venue: Option<Venue>,
        force_instrument_update: bool,
    ) -> anyhow::Result<Option<InstrumentId>> {
        let sec_type = details.contract.security_type.to_string();
        if self.is_filtered_sec_type(&sec_type) {
            tracing::debug!(
                "Skipping filtered security type {} for contract {:?}",
                sec_type,
                details.contract
            );
            return Ok(None);
        }

        let resolved_venue =
            venue.unwrap_or_else(|| self.determine_venue(&details.contract, Some(details)));
        let instrument_id = self
            .instrument_id_from_contract(&details.contract, resolved_venue)
            .context("Failed to convert IB contract to instrument ID")?;

        let instrument = match parse_ib_contract_to_instrument(details, instrument_id) {
            Ok(instrument) => instrument,
            Err(e) => {
                tracing::warn!(
                    "Failed to parse IB contract details for {}: {}",
                    instrument_id,
                    e
                );
                return Ok(None);
            }
        };

        if !self.passes_filter_callable(&instrument)? {
            return Ok(None);
        }

        self.cache_instrument(
            instrument_id,
            instrument,
            Some(details.clone()),
            None,
            None,
            force_instrument_update,
        );

        Ok(Some(instrument_id))
    }

    pub(crate) fn instrument_id_from_contract(
        &self,
        contract: &Contract,
        venue: Venue,
    ) -> anyhow::Result<InstrumentId> {
        self.symbology.instrument_id(contract, Some(venue))
    }

    fn cache_instrument(
        &self,
        instrument_id: InstrumentId,
        instrument: InstrumentAny,
        details: Option<ibapi::contracts::ContractDetails>,
        contract: Option<Contract>,
        price_magnifier: Option<i32>,
        force_instrument_update: bool,
    ) -> bool {
        let should_update =
            force_instrument_update || !self.instruments.contains_key(&instrument_id);
        if should_update {
            self.instruments.insert(instrument_id, instrument);
        }

        if let Some(details) = details {
            let contract_id = details.contract.contract_id;
            self.contracts
                .insert(instrument_id, details.contract.clone());
            self.contract_details.insert(instrument_id, details.clone());

            if contract_id != 0 {
                self.contract_id_to_instrument_id
                    .insert(contract_id, instrument_id);
            }

            self.price_magnifiers.insert(
                instrument_id,
                normalize_price_magnifier(details.price_magnifier),
            );
        } else if let Some(contract) = contract {
            if contract.contract_id != 0 {
                self.contract_id_to_instrument_id
                    .insert(contract.contract_id, instrument_id);
            }

            self.contracts.insert(instrument_id, contract);
        }

        if let Some(price_magnifier) = price_magnifier {
            self.price_magnifiers
                .insert(instrument_id, normalize_price_magnifier(price_magnifier));
        }

        should_update
    }

    fn passes_filter_callable(&self, instrument: &InstrumentAny) -> anyhow::Result<bool> {
        let Some(filter_callable) = self.config.filter_callable.as_deref() else {
            return Ok(true);
        };

        #[cfg(feature = "python")]
        {
            use nautilus_model::python::instruments::instrument_any_to_pyobject;
            use pyo3::{prelude::*, types::PyModule};

            Python::attach(|py| {
                let (module_name, callable_name) =
                    filter_callable.rsplit_once('.').ok_or_else(|| {
                        anyhow::anyhow!(
                            "Invalid filter_callable path {filter_callable:?}; expected module.callable"
                        )
                    })?;

                let callable = PyModule::import(py, module_name)
                    .map_err(|e| anyhow::anyhow!("Failed to import {module_name}: {e}"))?
                    .getattr(callable_name)
                    .map_err(|e| anyhow::anyhow!("Failed to resolve {filter_callable}: {e}"))?;
                let py_instrument = instrument_any_to_pyobject(py, instrument.clone())
                    .map_err(|e| anyhow::anyhow!("Failed to convert instrument to Python: {e}"))?;
                callable
                    .call1((py_instrument,))
                    .and_then(|result| result.extract::<bool>())
                    .map_err(|e| anyhow::anyhow!("filter_callable {filter_callable} failed: {e}"))
            })
        }

        #[cfg(not(feature = "python"))]
        {
            let _ = instrument;
            anyhow::bail!(
                "filter_callable {filter_callable:?} requires the Interactive Brokers adapter to be built with the python feature"
            );
        }
    }

    /// Batch load multiple instrument IDs.
    ///
    /// This method fetches and caches contract details for multiple instrument IDs in parallel.
    ///
    /// # Returns
    ///
    /// Returns a vector of successfully loaded instrument IDs.
    ///
    /// # Errors
    ///
    /// Returns an error if fetching fails.
    pub async fn batch_load(
        &self,
        client: &ibapi::Client,
        instrument_ids: Vec<InstrumentId>,
        filters: Option<&[String]>,
    ) -> anyhow::Result<Vec<InstrumentId>> {
        let mut loaded_ids = Vec::new();

        // Apply filters if provided
        let filtered_ids: Vec<InstrumentId> = if let Some(filter_list) = filters {
            // Filter instrument IDs by matching against filter patterns
            // Filters can be:
            // - Security type filters (e.g., "STK", "OPT", "FUT")
            // - Venue filters (e.g., "SMART", "NASDAQ")
            // - Symbol patterns (partial matching)
            instrument_ids
                .into_iter()
                .filter(|instrument_id| {
                    // Check if instrument matches any filter
                    for filter in filter_list {
                        // Check symbol match (case-insensitive partial match)
                        if instrument_id
                            .symbol
                            .as_str()
                            .to_lowercase()
                            .contains(&filter.to_lowercase())
                        {
                            return true;
                        }

                        // Check venue match
                        if instrument_id.venue.as_str() == filter {
                            return true;
                        }

                        // Check security type (try to infer from instrument)
                        if let Some(contract_details) = self.contract_details.get(instrument_id) {
                            let sec_type_str = contract_details.contract.security_type.to_string();

                            if sec_type_str.to_uppercase().contains(&filter.to_uppercase()) {
                                return true;
                            }
                        }
                    }

                    false
                })
                .collect()
        } else {
            instrument_ids
        };

        // Load instruments sequentially (can be parallelized in future if needed)
        let filtered_count = filtered_ids.len();
        for instrument_id in filtered_ids {
            match self
                .fetch_contract_details(client, instrument_id, false, None)
                .await
            {
                Ok(()) => loaded_ids.push(instrument_id),
                Err(e) => {
                    tracing::warn!("Failed to load instrument {}: {}", instrument_id, e);
                }
            }
        }

        tracing::debug!(
            "Batch loaded {} instruments ({} after filtering)",
            loaded_ids.len(),
            filtered_count
        );

        // Save cache if cache_path is configured
        if !loaded_ids.is_empty()
            && let Some(ref cache_path) = self.config.cache_path
            && let Err(e) = self.save_cache(cache_path).await
        {
            tracing::warn!("Failed to save instrument cache to {}: {}", cache_path, e);
        }

        Ok(loaded_ids)
    }

    /// Fetch option chain for a given underlying contract with expiry filtering.
    ///
    /// This is equivalent to Python's `get_option_chain_details_by_range`.
    /// It uses `contract_details` to fetch options with precise expiry filtering,
    /// which is more flexible than the basic `option_chain` API.
    ///
    /// # Returns
    ///
    /// Returns the number of option instruments loaded.
    ///
    /// # Errors
    ///
    /// Returns an error if fetching fails.
    pub async fn fetch_option_chain_by_range(
        &self,
        client: &ibapi::Client,
        underlying: &Contract,
        expiry_min: Option<&str>,
        expiry_max: Option<&str>,
        option_chain_exchange: Option<&str>,
    ) -> anyhow::Result<usize> {
        let exchange = option_chain_exchange.unwrap_or_else(|| underlying.exchange.as_str());
        tracing::debug!(
            "Building option chain for {}.{} (sec_type={:?}, contract_id={}, expiry_min={:?}, expiry_max={:?}, config_min_days={:?}, config_max_days={:?})",
            underlying.symbol.as_str(),
            exchange,
            underlying.security_type,
            underlying.contract_id,
            expiry_min,
            expiry_max,
            self.config.min_expiry_days,
            self.config.max_expiry_days,
        );

        // First, get option chain metadata to determine expirations
        let symbol = underlying.symbol.as_str();
        let mut option_chain_request = client.option_chain(
            symbol,
            underlying.security_type.clone(),
            underlying.contract_id,
        );

        // IB names the exchange only for futures options; for any other underlying it returns
        // one chain per listing exchange, and naming an exchange returns none
        if underlying.security_type == SecurityType::Future {
            option_chain_request = option_chain_request.exchange(exchange);
        }

        let mut option_chain_stream = option_chain_request
            .subscribe()
            .await
            .context("Failed to request option chain from IB")?;

        let mut total_loaded = 0;

        // Get current time for expiry day calculation
        let now = nautilus_core::time::get_atomic_clock_realtime().get_time_ns();

        // Collect all expirations from the metadata
        let mut all_expirations = Vec::new();

        while let Some(result) = option_chain_stream.next().await {
            match result {
                Ok(SubscriptionItem::Data(chain)) => {
                    tracing::debug!(
                        "Received option chain metadata exchange={} trading_class={} expirations={} strikes={}",
                        chain.exchange,
                        chain.trading_class,
                        chain.expirations.len(),
                        chain.strikes.len(),
                    );

                    if chain.exchange != exchange {
                        continue;
                    }

                    for expiration in &chain.expirations {
                        // Filter by expiry date string if specified
                        let date_filter_pass = match (expiry_min, expiry_max) {
                            (Some(min), Some(max)) => {
                                expiration.as_str() >= min && expiration.as_str() <= max
                            }
                            (Some(min), None) => expiration.as_str() >= min,
                            (None, Some(max)) => expiration.as_str() <= max,
                            (None, None) => true,
                        };

                        // Filter by expiry days from config if specified
                        let days_filter_pass = {
                            let expiry_ns =
                                expiry_timestring_to_unix_nanos(expiration.as_str(), None)
                                    .unwrap_or(now);
                            let days_until_expiry =
                                (expiry_ns.as_u64().saturating_sub(now.as_u64()))
                                    / (24 * 60 * 60 * 1_000_000_000);

                            let min_days_ok = self
                                .config
                                .min_expiry_days
                                .is_none_or(|min| days_until_expiry >= min as u64);
                            let max_days_ok = self
                                .config
                                .max_expiry_days
                                .is_none_or(|max| days_until_expiry <= max as u64);

                            min_days_ok && max_days_ok
                        };

                        if date_filter_pass
                            && days_filter_pass
                            && !all_expirations.contains(expiration)
                        {
                            all_expirations.push(expiration.clone());
                        }
                    }
                }
                Ok(SubscriptionItem::Notice(notice)) => {
                    tracing::debug!("Received option chain notice: {notice:?}");
                }
                Err(e) => {
                    tracing::warn!("Error receiving option chain metadata: {e}");
                }
            }
        }

        all_expirations.sort_unstable();

        tracing::debug!(
            "Filtered {} option expirations for {}.{}",
            all_expirations.len(),
            underlying.symbol.as_str(),
            exchange,
        );

        // Now fetch contract details for each expiry using contract_details
        for expiration in all_expirations {
            tracing::debug!(
                "Requesting option contract details for {}.{} expiry {}",
                underlying.symbol.as_str(),
                exchange,
                expiration,
            );

            let option_contract = Contract {
                contract_id: 0,
                symbol: underlying.symbol.clone(),
                security_type: if underlying.security_type == SecurityType::Future {
                    SecurityType::FuturesOption
                } else {
                    SecurityType::Option
                },
                last_trade_date_or_contract_month: expiration.clone(),
                strike: f64::MAX,
                right: None,
                multiplier: String::new(),
                exchange: Exchange::from(exchange),
                currency: underlying.currency.clone(),
                local_symbol: String::new(),
                primary_exchange: Exchange::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,
            };

            match request_contract_details(client, &option_contract).await {
                Ok(details_vec) => {
                    tracing::debug!(
                        "Received {} raw option contract details for {}.{} expiry {}",
                        details_vec.len(),
                        underlying.symbol.as_str(),
                        exchange,
                        expiration,
                    );

                    for details in details_vec {
                        // Filter by underlying contract ID
                        if details.under_contract_id != underlying.contract_id {
                            continue;
                        }

                        let contract_id = details.contract.contract_id;

                        if self.contract_id_to_instrument_id.contains_key(&contract_id) {
                            continue;
                        }

                        match self.process_contract_detail(&details, None, false) {
                            Ok(Some(_instrument_id)) => {
                                total_loaded += 1;
                            }
                            Ok(None) => {}
                            Err(e) => {
                                tracing::warn!("Failed to parse option instrument: {}", e);
                            }
                        }
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        "Failed to fetch contract details for expiration {}: {}",
                        expiration,
                        e
                    );
                }
            }
        }

        tracing::debug!(
            "Successfully loaded {} option instruments from chain for {}.{}",
            total_loaded,
            underlying.symbol.as_str(),
            exchange,
        );

        // Save cache if cache_path is configured
        if total_loaded > 0
            && let Some(ref cache_path) = self.config.cache_path
            && let Err(e) = self.save_cache(cache_path).await
        {
            tracing::warn!("Failed to save instrument cache to {}: {}", cache_path, e);
        }

        Ok(total_loaded)
    }

    /// Fetch and cache futures chain (all futures contracts for a symbol).
    ///
    /// This method fetches all futures contracts for a given underlying symbol
    /// and populates the cache with all individual futures instruments.
    ///
    /// # Returns
    ///
    /// Returns the number of futures instruments loaded.
    ///
    /// # Errors
    ///
    /// Returns an error if fetching fails.
    #[allow(clippy::too_many_arguments)] // Public query options map directly to IB filters.
    pub async fn fetch_futures_chain(
        &self,
        client: &ibapi::Client,
        symbol: &str,
        exchange: &str,
        currency: &str,
        trading_class: Option<&str>,
        include_expired: bool,
        min_expiry_days: Option<u32>,
        max_expiry_days: Option<u32>,
    ) -> anyhow::Result<usize> {
        tracing::debug!(
            "Building futures chain for {}.{} (currency={}, trading_class={:?}, include_expired={}, min_days={:?}, max_days={:?}, config_min_days={:?}, config_max_days={:?})",
            symbol,
            exchange,
            currency,
            trading_class,
            include_expired,
            min_expiry_days,
            max_expiry_days,
            self.config.min_expiry_days,
            self.config.max_expiry_days,
        );

        // Build futures contract for lookup
        let futures_contract = Contract {
            contract_id: 0,
            symbol: Symbol::from(symbol.to_string()),
            security_type: SecurityType::Future,
            last_trade_date_or_contract_month: String::new(),
            strike: f64::MAX,
            right: None,
            multiplier: String::new(),
            exchange: Exchange::from(exchange.to_string()),
            currency: ibapi::contracts::Currency::from(currency.to_string()),
            local_symbol: String::new(),
            primary_exchange: Exchange::from(""),
            trading_class: trading_class.unwrap_or_default().to_string(),
            include_expired,
            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,
        };

        // Fetch contract details for all matching futures
        let details_vec = request_contract_details(client, &futures_contract)
            .await
            .context("Failed to fetch futures chain from IB")?;

        tracing::debug!(
            "Received {} raw futures contract details for {}.{}",
            details_vec.len(),
            symbol,
            exchange,
        );

        let mut total_loaded = 0;
        let now = nautilus_core::time::get_atomic_clock_realtime().get_time_ns();

        for details in details_vec {
            let contract_id = details.contract.contract_id;

            // Check if already cached
            if self.contract_id_to_instrument_id.contains_key(&contract_id) {
                continue;
            }

            // Check if security type is filtered
            let sec_type_str = details.contract.security_type.to_string();
            if self.is_filtered_sec_type(&sec_type_str) {
                continue;
            }

            // Filter by expiry days for futures
            if !details
                .contract
                .last_trade_date_or_contract_month
                .is_empty()
                && let Ok(expiry_ns) = expiry_timestring_to_unix_nanos(
                    &details.contract.last_trade_date_or_contract_month,
                    Some(&details),
                )
            {
                let days_until_expiry = (expiry_ns.as_u64().saturating_sub(now.as_u64()))
                    / (24 * 60 * 60 * 1_000_000_000);

                let min_days_ok = min_expiry_days
                    .or(self.config.min_expiry_days)
                    .is_none_or(|min| days_until_expiry >= min as u64);
                let max_days_ok = max_expiry_days
                    .or(self.config.max_expiry_days)
                    .is_none_or(|max| days_until_expiry <= max as u64);

                if !min_days_ok || !max_days_ok {
                    continue;
                }
            }

            match self.process_contract_detail(&details, None, false) {
                Ok(Some(_instrument_id)) => {
                    total_loaded += 1;
                }
                Ok(None) => {}
                Err(e) => {
                    tracing::warn!("Failed to parse futures instrument: {}", e);
                }
            }
        }

        tracing::debug!(
            "Successfully loaded {} futures instruments from chain",
            total_loaded
        );

        // Save cache if cache_path is configured
        if total_loaded > 0
            && let Some(ref cache_path) = self.config.cache_path
            && let Err(e) = self.save_cache(cache_path).await
        {
            tracing::warn!("Failed to save instrument cache to {}: {}", cache_path, e);
        }

        Ok(total_loaded)
    }

    /// Fetch and cache a BAG (spread) contract.
    ///
    /// This method fetches contract details for a spread contract by requesting
    /// contract details with a BAG contract. The BAG contract should have its
    /// combo_legs populated with the individual leg contract IDs.
    ///
    /// # Returns
    ///
    /// Returns the number of spread instruments loaded (0 or 1).
    ///
    /// # Errors
    ///
    /// Returns an error if fetching fails.
    ///
    /// # Notes
    ///
    /// This method now auto-loads all leg instruments from combo_legs and creates
    /// a proper spread instrument, matching Python's `_load_bag_contract` behavior.
    pub async fn fetch_bag_contract(
        &self,
        client: &ibapi::Client,
        bag_contract: &Contract,
    ) -> anyhow::Result<usize> {
        // Validate BAG contract
        if bag_contract.security_type != SecurityType::Spread || bag_contract.combo_legs.is_empty()
        {
            anyhow::bail!(
                "Invalid BAG contract: must have security_type=Spread and non-empty combo_legs"
            );
        }

        tracing::debug!(
            "Loading BAG contract with {} legs",
            bag_contract.combo_legs.len()
        );

        // First, load all individual leg instruments and collect their details
        let mut leg_contract_details = Vec::new();
        let mut leg_tuples = Vec::new();

        for combo_leg in &bag_contract.combo_legs {
            // Create a leg contract using information from the combo leg
            let leg_contract = Contract {
                contract_id: combo_leg.contract_id,  // Use conId from combo_leg
                symbol: bag_contract.symbol.clone(), // Use underlying symbol from BAG
                security_type: SecurityType::Option, // Default to Option, will be determined from contract details
                last_trade_date_or_contract_month: String::new(),
                strike: 0.0,
                right: None,
                multiplier: String::new(),
                exchange: Exchange::from(combo_leg.exchange.as_str()),
                currency: bag_contract.currency.clone(), // Use currency from BAG
                local_symbol: String::new(),
                primary_exchange: Exchange::default(),
                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,
            };

            // Fetch contract details for this leg
            let leg_details_vec = request_contract_details(client, &leg_contract)
                .await
                .with_context(|| {
                    format!(
                        "Failed to fetch contract details for leg conId {}",
                        combo_leg.contract_id
                    )
                })?;

            if leg_details_vec.is_empty() {
                tracing::warn!(
                    "No contract details returned for leg conId {}",
                    combo_leg.contract_id
                );
                continue;
            }

            let leg_details = &leg_details_vec[0];
            let leg_contract_id = leg_details.contract.contract_id;

            // Check if leg is already cached
            let leg_instrument_id =
                if let Some(cached_id) = self.contract_id_to_instrument_id.get(&leg_contract_id) {
                    *cached_id.value()
                } else {
                    // Load the leg instrument
                    let leg_venue = self.determine_venue(&leg_details.contract, Some(leg_details));
                    let leg_instrument_id = self
                        .symbology
                        .instrument_id(&leg_details.contract, Some(leg_venue))
                        .context("Failed to convert leg contract to instrument ID")?;

                    // Parse and cache the leg instrument
                    let leg_instrument =
                        parse_ib_contract_to_instrument(leg_details, leg_instrument_id)
                            .context("Failed to parse leg instrument")?;

                    self.instruments.insert(leg_instrument_id, leg_instrument);
                    self.contract_details
                        .insert(leg_instrument_id, leg_details.clone());
                    self.contracts
                        .insert(leg_instrument_id, leg_details.contract.clone());
                    self.contract_id_to_instrument_id
                        .insert(leg_contract_id, leg_instrument_id);
                    self.price_magnifiers
                        .insert(leg_instrument_id, leg_details.price_magnifier);

                    leg_instrument_id
                };

            // Determine ratio (positive for BUY, negative for SELL)
            let ratio = IbAction::from_str(combo_leg.action.as_str())
                .context("Invalid combo leg action")?
                .signed_multiplier()
                * combo_leg.ratio;

            // Get the contract details for this leg (should be cached now)
            let leg_details_clone = self
                .contract_details
                .get(&leg_instrument_id)
                .map(|entry| entry.value().clone())
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "Contract details not found for leg {leg_instrument_id} after loading"
                    )
                })?;

            leg_contract_details.push((leg_details_clone, ratio));
            leg_tuples.push((leg_instrument_id, ratio));
        }

        if leg_tuples.is_empty() {
            anyhow::bail!("No valid legs loaded for BAG contract");
        }

        // Create spread instrument ID from leg tuples
        let spread_instrument_id = create_spread_instrument_id(&leg_tuples)
            .context("Failed to create spread instrument ID from leg tuples")?;

        // Fetch BAG contract details (for storing the mapping)
        let bag_details_vec = request_contract_details(client, bag_contract)
            .await
            .context("Failed to fetch BAG contract details from IB")?;

        if bag_details_vec.is_empty() {
            tracing::warn!("No contract details returned for BAG contract");

            if bag_contract.contract_id != 0 && self.instruments.contains_key(&spread_instrument_id)
            {
                self.contract_id_to_instrument_id
                    .insert(bag_contract.contract_id, spread_instrument_id);
            }

            return Ok(0);
        }

        let bag_details = &bag_details_vec[0];
        let bag_contract_id = bag_details.contract.contract_id;

        if bag_contract_id != 0 {
            self.contract_id_to_instrument_id
                .insert(bag_contract_id, spread_instrument_id);
        }

        // Check if spread is already cached after ensuring the BAG contract ID is mapped.
        if self.instruments.contains_key(&spread_instrument_id) {
            tracing::debug!("Spread instrument {} already cached", spread_instrument_id);
            self.contract_details
                .insert(spread_instrument_id, bag_details.clone());
            self.contracts
                .insert(spread_instrument_id, bag_details.contract.clone());
            self.price_magnifiers
                .insert(spread_instrument_id, bag_details.price_magnifier);
            return Ok(0);
        }

        // Create the spread instrument
        let timestamp = nautilus_core::time::get_atomic_clock_realtime().get_time_ns();

        // Convert leg_contract_details to the format needed by parse_spread_instrument_id
        let leg_details_refs: Vec<(&ibapi::contracts::ContractDetails, i32)> =
            leg_contract_details.iter().map(|(d, r)| (d, *r)).collect();

        let spread_instrument = parse_spread_instrument_any(
            spread_instrument_id,
            &leg_details_refs,
            Some(&bag_details.contract),
            Some(timestamp),
        )
        .context("Failed to parse spread instrument")?;

        // Cache the spread instrument and mappings
        self.instruments
            .insert(spread_instrument_id, spread_instrument);
        self.contract_details
            .insert(spread_instrument_id, bag_details.clone());
        self.contracts
            .insert(spread_instrument_id, bag_details.contract.clone());
        self.price_magnifiers
            .insert(spread_instrument_id, bag_details.price_magnifier);

        tracing::debug!(
            "Successfully loaded spread instrument {} with {} legs",
            spread_instrument_id,
            leg_tuples.len()
        );

        // Save cache if cache_path is configured
        if let Some(ref cache_path) = self.config.cache_path
            && let Err(e) = self.save_cache(cache_path).await
        {
            tracing::warn!("Failed to save instrument cache to {}: {}", cache_path, e);
        }

        Ok(1)
    }

    /// Save the current instrument cache to disk.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or file I/O fails.
    pub async fn save_cache(&self, cache_path: &str) -> anyhow::Result<()> {
        let cache = InstrumentCache {
            cache_timestamp: Timestamp::now(),
            contract_id_to_instrument_id: self
                .contract_id_to_instrument_id
                .iter()
                .map(|entry| (*entry.key(), entry.value().to_string()))
                .collect(),
            price_magnifiers: self
                .price_magnifiers
                .iter()
                .map(|entry| (entry.key().to_string(), *entry.value()))
                .collect(),
            contracts: self
                .contracts
                .iter()
                .map(|entry| (entry.key().to_string(), entry.value().clone()))
                .collect(),
            contract_details: self
                .contract_details
                .iter()
                .map(|entry| (entry.key().to_string(), entry.value().clone()))
                .collect(),
            instruments: self
                .instruments
                .iter()
                .map(|entry| {
                    let instrument_id = entry.key().to_string();
                    let json =
                        serde_json::to_string(entry.value()).unwrap_or_else(|_| String::new());
                    (instrument_id, json)
                })
                .collect(),
        };

        // Ensure parent directory exists
        if let Some(parent) = Path::new(cache_path).parent() {
            fs::create_dir_all(parent)?;
        }

        // Write cache to file
        let json = serde_json::to_string_pretty(&cache)?;
        fs::write(cache_path, json)?;
        tracing::debug!(
            "Saved instrument cache to {} ({} instruments)",
            cache_path,
            cache.instruments.len()
        );
        Ok(())
    }

    /// Load instrument cache from disk if valid.
    ///
    /// # Returns
    ///
    /// Returns `true` if cache was loaded successfully and is valid, `false` otherwise.
    ///
    /// # Errors
    ///
    /// Returns an error if deserialization or file I/O fails (but treats missing file as non-error).
    pub async fn load_cache(&self, cache_path: &str) -> anyhow::Result<bool> {
        // Check if cache file exists
        if !Path::new(cache_path).exists() {
            tracing::debug!("Cache file does not exist: {}", cache_path);
            return Ok(false);
        }

        // Load cache from file
        let json = fs::read_to_string(cache_path)?;
        let cache: InstrumentCache = serde_json::from_str(&json)?;

        // Check cache validity
        if let Some(validity_days) = self.config.cache_validity_days {
            let cache_age = cache.cache_timestamp.duration_until(Timestamp::now());
            let max_age = jiff::SignedDuration::from_hours(24 * (validity_days as i64));
            if cache_age > max_age {
                tracing::debug!(
                    "Cache is expired (age: {} days, max: {} days). Ignoring cache",
                    cache_age.as_secs() / (24 * 60 * 60),
                    validity_days
                );
                return Ok(false);
            }
        }

        // Deserialize and restore instruments
        let mut loaded_count = 0;

        for (instrument_id_str, instrument_json) in &cache.instruments {
            match InstrumentId::from_str(instrument_id_str) {
                Ok(instrument_id) => match serde_json::from_str::<InstrumentAny>(instrument_json) {
                    Ok(instrument) => {
                        self.instruments.insert(instrument_id, instrument);

                        if let Ok(value) =
                            serde_json::from_str::<serde_json::Value>(instrument_json)
                            && let Some(contract_json) = find_contract_json(&value)
                            && let Ok(mut contract) = parse_contract_from_json(contract_json)
                        {
                            trim_last_trade_date(&mut contract);

                            if contract.contract_id != 0 {
                                self.contract_id_to_instrument_id
                                    .insert(contract.contract_id, instrument_id);
                            }

                            self.contracts.insert(instrument_id, contract);
                        }

                        loaded_count += 1;
                    }
                    Err(e) => {
                        tracing::warn!(
                            "Failed to deserialize instrument {}: {}",
                            instrument_id_str,
                            e
                        );
                    }
                },
                Err(e) => {
                    tracing::warn!("Failed to parse instrument ID {}: {}", instrument_id_str, e);
                }
            }
        }

        // Restore contracts and contract details; a cache written before expiries were split
        // may still hold one with its time (see `split_last_trade_date`)
        for (instrument_id_str, contract) in &cache.contracts {
            if let Ok(instrument_id) = InstrumentId::from_str(instrument_id_str) {
                if contract.contract_id != 0 {
                    self.contract_id_to_instrument_id
                        .insert(contract.contract_id, instrument_id);
                }

                let mut contract = contract.clone();
                trim_last_trade_date(&mut contract);
                self.contracts.insert(instrument_id, contract);
            }
        }

        for (instrument_id_str, details) in &cache.contract_details {
            if let Ok(instrument_id) = InstrumentId::from_str(instrument_id_str) {
                if details.contract.contract_id != 0 {
                    self.contract_id_to_instrument_id
                        .insert(details.contract.contract_id, instrument_id);
                }

                let mut details = details.clone();
                split_last_trade_date(&mut details);
                self.contracts
                    .insert(instrument_id, details.contract.clone());
                self.contract_details.insert(instrument_id, details);
            }
        }

        // Restore contract ID mappings
        for (contract_id, instrument_id_str) in &cache.contract_id_to_instrument_id {
            if let Ok(instrument_id) = InstrumentId::from_str(instrument_id_str) {
                self.contract_id_to_instrument_id
                    .insert(*contract_id, instrument_id);
            }
        }

        // Restore price magnifiers
        for (instrument_id_str, magnifier) in &cache.price_magnifiers {
            if let Ok(instrument_id) = InstrumentId::from_str(instrument_id_str) {
                self.price_magnifiers.insert(instrument_id, *magnifier);
            }
        }

        tracing::debug!(
            "Loaded instrument cache from {} ({} instruments, created at {})",
            cache_path,
            loaded_count,
            cache.cache_timestamp
        );
        Ok(true)
    }
}

#[cfg(test)]
mod tests {
    use std::{
        fs,
        sync::atomic::{AtomicBool, AtomicUsize, Ordering},
    };

    use indexmap::IndexMap;
    use nautilus_common::config::ConfigError;
    use nautilus_core::{Params, UnixNanos};
    use nautilus_model::{
        identifiers::{Symbol, Venue},
        instruments::CurrencyPair,
        types::{Price, Quantity, currency::Currency},
    };
    use rstest::rstest;
    use tempfile::TempDir;

    use super::*;
    use crate::common::contract_to_json_value;

    struct TestStartupLoader {
        id_calls: AtomicUsize,
        contract_calls: AtomicUsize,
        fail_next_id: AtomicBool,
        resolve_ids: AtomicBool,
        resolve_contracts: AtomicBool,
        yield_on_load: bool,
    }

    impl TestStartupLoader {
        fn new(resolve_ids: bool, resolve_contracts: bool) -> Self {
            Self {
                id_calls: AtomicUsize::new(0),
                contract_calls: AtomicUsize::new(0),
                fail_next_id: AtomicBool::new(false),
                resolve_ids: AtomicBool::new(resolve_ids),
                resolve_contracts: AtomicBool::new(resolve_contracts),
                yield_on_load: false,
            }
        }
    }

    impl StartupInstrumentLoader for TestStartupLoader {
        async fn load_instrument_id(
            &self,
            instrument_id: InstrumentId,
        ) -> anyhow::Result<Option<InstrumentId>> {
            self.id_calls.fetch_add(1, Ordering::SeqCst);

            if self.yield_on_load {
                tokio::task::yield_now().await;
            }

            if self.fail_next_id.swap(false, Ordering::SeqCst) {
                anyhow::bail!("Socket disconnected");
            }

            Ok(self
                .resolve_ids
                .load(Ordering::SeqCst)
                .then_some(instrument_id))
        }

        async fn load_contract(
            &self,
            _configured: &ConfiguredContract,
        ) -> anyhow::Result<Vec<InstrumentId>> {
            self.contract_calls.fetch_add(1, Ordering::SeqCst);

            if self.yield_on_load {
                tokio::task::yield_now().await;
            }

            Ok(if self.resolve_contracts.load(Ordering::SeqCst) {
                vec![InstrumentId::new(
                    Symbol::from("MSFT"),
                    Venue::from("NASDAQ"),
                )]
            } else {
                Vec::new()
            })
        }
    }

    fn create_test_provider_with_cache() -> (InteractiveBrokersInstrumentProvider, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let cache_path = temp_dir
            .path()
            .join("test_cache.json")
            .to_str()
            .unwrap()
            .to_string();

        let config = InteractiveBrokersInstrumentProviderConfig::builder()
            .cache_path(cache_path)
            .cache_validity_days(7u32)
            .build();

        let provider = InteractiveBrokersInstrumentProvider::new(config).unwrap();
        (provider, temp_dir)
    }

    fn opra_option_contract_details(mut contract: Contract) -> ibapi::contracts::ContractDetails {
        contract.contract_id = 12_345;
        contract.symbol = ibapi::contracts::Symbol::from("AAPL");
        contract.security_type = SecurityType::Option;
        contract.exchange = Exchange::from("SMART");
        contract.currency = ibapi::contracts::Currency::from("USD");
        contract.local_symbol = "AAPL  270115P00155000".to_string();
        contract.last_trade_date_or_contract_month = "20270115".to_string();
        contract.strike = 155.0;
        contract.right = Some(ibapi::contracts::OptionRight::Put);
        contract.multiplier = "100".to_string();

        ibapi::contracts::ContractDetails {
            contract,
            min_tick: 0.01,
            under_symbol: "AAPL".to_string(),
            under_security_type: "STK".to_string(),
            valid_exchanges: vec!["SMART".to_string(), "CBOE".to_string()],
            ..Default::default()
        }
    }

    fn create_test_instrument(instrument_id: InstrumentId) -> InstrumentAny {
        create_test_instrument_with_info(instrument_id, None)
    }

    #[rstest]
    fn test_qualified_opra_details_preserve_canonical_instrument_identity() {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let requested_id = InstrumentId::from("AAPL  270115P00155000.OPRA");
        let request = provider
            .resolve_contract_for_instrument(requested_id)
            .unwrap();

        assert_eq!(request.security_type, SecurityType::Option);
        assert_eq!(request.exchange.as_str(), "SMART");
        assert!(request.symbol.as_str().is_empty());
        assert_eq!(request.currency.as_str(), "USD");
        assert_eq!(request.local_symbol, "AAPL  270115P00155000");
        assert!(request.last_trade_date_or_contract_month.is_empty());
        assert!(request.right.is_none());
        assert_eq!(request.strike, 0.0);

        let details = opra_option_contract_details(request);
        let loaded_id = provider
            .process_contract_detail(&details, Some(requested_id.venue), false)
            .unwrap()
            .unwrap();

        assert_eq!(loaded_id, requested_id);
        assert_eq!(provider.count(), 1);
        assert_eq!(
            provider.get_instrument_id_by_contract_id(12_345),
            Some(requested_id)
        );
        assert_eq!(
            provider
                .resolve_instrument_id_for_contract(&details.contract)
                .unwrap(),
            requested_id
        );

        let cached = provider.find(&requested_id).unwrap();

        let InstrumentAny::OptionContract(option) = cached else {
            panic!("expected option contract");
        };

        assert_eq!(option.id, requested_id);
        assert_eq!(option.id.venue.as_str(), "OPRA");
        assert!(
            provider
                .find(&InstrumentId::from("AAPL  270115P00155000.SMART"))
                .is_none()
        );

        let cached_contract = provider
            .instrument_id_to_ib_contract(&requested_id)
            .unwrap();
        assert_eq!(cached_contract.contract_id, 12_345);
        assert_eq!(cached_contract.security_type, SecurityType::Option);
        assert_eq!(cached_contract.exchange.as_str(), "SMART");

        let resolved_contract = provider
            .resolve_contract_for_instrument(requested_id)
            .unwrap();
        assert_eq!(resolved_contract, cached_contract);

        let cached_details = provider
            .instrument_id_to_ib_contract_details(&requested_id)
            .unwrap();
        assert_eq!(cached_details.contract.contract_id, 12_345);
        assert_eq!(
            cached_details.valid_exchanges,
            vec!["SMART".to_string(), "CBOE".to_string()]
        );
    }

    fn create_test_instrument_with_info(
        instrument_id: InstrumentId,
        info: Option<Params>,
    ) -> InstrumentAny {
        CurrencyPair::builder()
            .instrument_id(instrument_id)
            .raw_symbol(Symbol::from("EUR/USD"))
            .base_currency(Currency::from("EUR"))
            .quote_currency(Currency::from("USD"))
            .price_precision(4)
            .size_precision(0)
            .price_increment(Price::from("0.0001"))
            .size_increment(Quantity::from(1))
            .maybe_info(info)
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap()
            .into()
    }

    fn create_contract_info(contract: &Contract, price_magnifier: Option<i32>) -> Params {
        let mut info =
            IndexMap::from([(String::from("contract"), contract_to_json_value(contract))]);

        if let Some(price_magnifier) = price_magnifier {
            info.insert(
                String::from("priceMagnifier"),
                serde_json::Value::from(price_magnifier),
            );
        }

        Params::new(info)
    }

    #[tokio::test]
    async fn test_initialize_loads_all_configured_inputs_once() {
        let instrument_id = InstrumentId::new(Symbol::from("AAPL"), Venue::from("NASDAQ"));

        let contract_spec = Contract {
            security_type: SecurityType::Stock,
            symbol: ibapi::contracts::Symbol::from("MSFT"),
            exchange: Exchange::from("NASDAQ"),
            ..Default::default()
        };

        let config = InteractiveBrokersInstrumentProviderConfig {
            load_ids: [instrument_id].into_iter().collect(),
            load_contracts: vec![ConfiguredContract::from(contract_spec)],
            ..Default::default()
        };

        let provider = InteractiveBrokersInstrumentProvider::new(config).unwrap();
        let loader = TestStartupLoader::new(true, true);

        let loaded_ids = provider.initialize_with_loader(&loader).await.unwrap();
        let second_result = provider.initialize_with_loader(&loader).await.unwrap();

        assert_eq!(
            loaded_ids,
            vec![
                instrument_id,
                InstrumentId::new(Symbol::from("MSFT"), Venue::from("NASDAQ")),
            ]
        );
        assert!(second_result.is_empty());
        assert_eq!(loader.id_calls.load(Ordering::SeqCst), 1);
        assert_eq!(loader.contract_calls.load(Ordering::SeqCst), 1);
        assert!(*provider.startup_initialized.lock().await);
    }

    #[tokio::test]
    async fn test_initialize_fails_closed_and_retries_unresolved_input() {
        let instrument_id = InstrumentId::new(Symbol::from("AAPL"), Venue::from("NASDAQ"));

        let contract_spec = Contract {
            security_type: SecurityType::Stock,
            symbol: ibapi::contracts::Symbol::from("MSFT"),
            exchange: Exchange::from("NASDAQ"),
            ..Default::default()
        };

        let config = InteractiveBrokersInstrumentProviderConfig {
            load_ids: [instrument_id].into_iter().collect(),
            load_contracts: vec![ConfiguredContract::from(contract_spec)],
            ..Default::default()
        };

        let provider = InteractiveBrokersInstrumentProvider::new(config).unwrap();
        let loader = TestStartupLoader::new(true, false);

        let error = provider.initialize_with_loader(&loader).await.unwrap_err();

        assert!(error.to_string().contains("contract at index 0"));
        assert!(!*provider.startup_initialized.lock().await);

        loader.resolve_contracts.store(true, Ordering::SeqCst);
        provider.initialize_with_loader(&loader).await.unwrap();

        assert_eq!(loader.id_calls.load(Ordering::SeqCst), 2);
        assert_eq!(loader.contract_calls.load(Ordering::SeqCst), 2);
        assert!(*provider.startup_initialized.lock().await);
    }

    #[tokio::test]
    async fn test_initialize_preserves_load_error_and_allows_retry() {
        let instrument_id = InstrumentId::new(Symbol::from("AAPL"), Venue::from("NASDAQ"));

        let config = InteractiveBrokersInstrumentProviderConfig {
            load_ids: [instrument_id].into_iter().collect(),
            ..Default::default()
        };

        let provider = InteractiveBrokersInstrumentProvider::new(config).unwrap();
        let loader = TestStartupLoader::new(true, true);
        loader.fail_next_id.store(true, Ordering::SeqCst);

        let error = provider.initialize_with_loader(&loader).await.unwrap_err();

        let error_chain = format!("{error:#}");
        assert!(error_chain.contains("Failed to load configured IB instrument ID AAPL.NASDAQ"));
        assert!(error_chain.contains("Socket disconnected"));
        assert!(!*provider.startup_initialized.lock().await);

        provider.initialize_with_loader(&loader).await.unwrap();

        assert_eq!(loader.id_calls.load(Ordering::SeqCst), 2);
        assert!(*provider.startup_initialized.lock().await);
    }

    #[tokio::test]
    async fn test_initialize_serializes_concurrent_calls() {
        let instrument_id = InstrumentId::new(Symbol::from("AAPL"), Venue::from("NASDAQ"));

        let config = InteractiveBrokersInstrumentProviderConfig {
            load_ids: [instrument_id].into_iter().collect(),
            ..Default::default()
        };

        let provider = InteractiveBrokersInstrumentProvider::new(config).unwrap();
        let mut loader = TestStartupLoader::new(true, true);
        loader.yield_on_load = true;

        let (first, second) = tokio::join!(
            provider.initialize_with_loader(&loader),
            provider.initialize_with_loader(&loader),
        );

        assert_eq!(first.unwrap().len() + second.unwrap().len(), 1);
        assert_eq!(loader.id_calls.load(Ordering::SeqCst), 1);
        assert!(*provider.startup_initialized.lock().await);
    }

    #[tokio::test]
    async fn test_save_cache() {
        let (provider, _temp_dir) = create_test_provider_with_cache();
        let cache_path = provider.config.cache_path.as_ref().unwrap().clone();

        // Add some test instruments
        let instrument_id1 = InstrumentId::new(Symbol::from("EUR/USD"), Venue::from("IDEALPRO"));
        let instrument_id2 = InstrumentId::new(Symbol::from("GBP/USD"), Venue::from("IDEALPRO"));

        let instrument1 = create_test_instrument(instrument_id1);
        let instrument2 = create_test_instrument(instrument_id2);

        provider.instruments.insert(instrument_id1, instrument1);
        provider.instruments.insert(instrument_id2, instrument2);
        provider
            .contract_id_to_instrument_id
            .insert(100, instrument_id1);
        provider
            .contract_id_to_instrument_id
            .insert(200, instrument_id2);
        provider.price_magnifiers.insert(instrument_id1, 1);
        provider.price_magnifiers.insert(instrument_id2, 1);

        // Save cache
        let result = provider.save_cache(&cache_path).await;
        assert!(result.is_ok(), "save_cache should succeed");

        // Verify file exists
        assert!(Path::new(&cache_path).exists(), "Cache file should exist");

        // Verify file contains JSON
        let contents = fs::read_to_string(&cache_path).unwrap();
        assert!(
            contents.contains("EUR/USD"),
            "Cache should contain instrument data"
        );
        assert!(
            contents.contains("cache_timestamp"),
            "Cache should contain timestamp"
        );
    }

    #[tokio::test]
    async fn test_load_cache_valid() {
        let (provider, _temp_dir) = create_test_provider_with_cache();
        let cache_path = provider.config.cache_path.as_ref().unwrap().clone();

        // First save a cache
        let instrument_id = InstrumentId::new(Symbol::from("EUR/USD"), Venue::from("IDEALPRO"));
        let instrument = create_test_instrument(instrument_id);

        provider
            .instruments
            .insert(instrument_id, instrument.clone());
        provider
            .contract_id_to_instrument_id
            .insert(100, instrument_id);
        provider.price_magnifiers.insert(instrument_id, 1);

        provider.save_cache(&cache_path).await.unwrap();

        // Create a new provider and load the cache
        let new_config = InteractiveBrokersInstrumentProviderConfig::builder()
            .cache_path(cache_path.clone())
            .cache_validity_days(7u32)
            .build();

        let new_provider = InteractiveBrokersInstrumentProvider::new(new_config).unwrap();

        let result = new_provider.load_cache(&cache_path).await;
        assert!(result.is_ok(), "load_cache should succeed");
        assert!(
            result.unwrap(),
            "load_cache should return true for valid cache"
        );

        // Verify instrument was loaded
        assert!(
            new_provider.find(&instrument_id).is_some(),
            "Instrument should be loaded from cache"
        );
        assert_eq!(new_provider.count(), 1, "Provider should have 1 instrument");
    }

    #[tokio::test]
    async fn test_load_cache_reads_chrono_timestamp() {
        let provider = InteractiveBrokersInstrumentProvider::new(
            InteractiveBrokersInstrumentProviderConfig::builder()
                .cache_validity_days(7u32)
                .build(),
        )
        .unwrap();
        let cache_path = concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/test_data/instrument_cache_chrono.json"
        );

        let loaded = provider.load_cache(cache_path).await.unwrap();

        assert!(loaded);
        assert_eq!(provider.count(), 0);
    }

    #[tokio::test]
    async fn test_load_cache_restores_contract_details() {
        let (provider, _temp_dir) = create_test_provider_with_cache();
        let cache_path = provider.config.cache_path.as_ref().unwrap().clone();
        let instrument_id = InstrumentId::new(Symbol::from("AAPL"), Venue::from("XNAS"));
        let instrument = create_test_instrument(instrument_id);

        let contract = Contract {
            contract_id: 265598,
            symbol: ibapi::contracts::Symbol::from("AAPL"),
            security_type: SecurityType::Stock,
            exchange: Exchange::from("SMART"),
            primary_exchange: Exchange::from("NASDAQ"),
            currency: ibapi::contracts::Currency::from("USD"),
            ..Default::default()
        };

        let details = ibapi::contracts::ContractDetails {
            contract: contract.clone(),
            price_magnifier: 1,
            ..Default::default()
        };

        provider.cache_instrument(
            instrument_id,
            instrument,
            Some(details),
            Some(contract),
            Some(1),
            false,
        );
        provider.save_cache(&cache_path).await.unwrap();

        let new_provider =
            InteractiveBrokersInstrumentProvider::new(provider.config.clone()).unwrap();

        assert!(new_provider.load_cache(&cache_path).await.unwrap());
        assert_eq!(
            new_provider
                .resolve_contract_for_instrument(instrument_id)
                .unwrap()
                .contract_id,
            265598
        );
        assert_eq!(
            new_provider
                .instrument_id_to_ib_contract_details(&instrument_id)
                .unwrap()
                .contract
                .contract_id,
            265598
        );
    }

    #[rstest]
    fn test_filter_sec_types_uses_typed_ib_codes() {
        let config = InteractiveBrokersInstrumentProviderConfig {
            filter_sec_types: [IbSecurityType::Option].into_iter().collect(),
            ..Default::default()
        };

        let provider = InteractiveBrokersInstrumentProvider::new(config).unwrap();

        assert!(provider.is_filtered_sec_type(&SecurityType::Option.to_string()));
        assert!(!provider.is_filtered_sec_type(&SecurityType::Stock.to_string()));
    }

    #[rstest]
    fn test_add_cached_instruments_only_seeds_ib_contracts() {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let ib_instrument_id = InstrumentId::new(Symbol::from("AAPL"), Venue::from("XNAS"));
        let non_ib_instrument_id =
            InstrumentId::new(Symbol::from("BTCUSDT"), Venue::from("BINANCE"));

        let contract = Contract {
            contract_id: 265598,
            symbol: ibapi::contracts::Symbol::from("AAPL"),
            security_type: SecurityType::Stock,
            exchange: Exchange::from("SMART"),
            primary_exchange: Exchange::from("NASDAQ"),
            currency: ibapi::contracts::Currency::from("USD"),
            ..Default::default()
        };

        let ib_instrument = create_test_instrument_with_info(
            ib_instrument_id,
            Some(create_contract_info(&contract, Some(100))),
        );
        let non_ib_instrument = create_test_instrument(non_ib_instrument_id);

        let count = provider.add_cached_instruments([ib_instrument, non_ib_instrument]);

        assert_eq!(count, 1);
        assert_eq!(provider.count(), 1);
        assert!(provider.find(&ib_instrument_id).is_some());
        assert!(provider.find(&non_ib_instrument_id).is_none());
        assert_eq!(
            provider
                .resolve_contract_for_instrument(ib_instrument_id)
                .unwrap()
                .contract_id,
            265598
        );
        assert_eq!(provider.get_price_magnifier(&ib_instrument_id), 100);
    }

    fn create_future_contract_with_expiry_time() -> Contract {
        Contract {
            contract_id: 515416632,
            symbol: ibapi::contracts::Symbol::from("ES"),
            security_type: SecurityType::Future,
            exchange: Exchange::from("CME"),
            currency: ibapi::contracts::Currency::from("USD"),
            local_symbol: String::from("ESZ6"),
            last_trade_date_or_contract_month: String::from("20261218 08:30:00 US/Central"),
            ..Default::default()
        }
    }

    #[rstest]
    fn test_add_cached_instruments_cuts_stored_expiry_to_date() {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let instrument_id = InstrumentId::new(Symbol::from("ESZ6"), Venue::from("CME"));
        let contract = create_future_contract_with_expiry_time();
        let instrument = create_test_instrument_with_info(
            instrument_id,
            Some(create_contract_info(&contract, Some(1))),
        );

        assert_eq!(provider.add_cached_instruments([instrument]), 1);
        assert_eq!(
            provider
                .resolve_contract_for_instrument(instrument_id)
                .unwrap()
                .last_trade_date_or_contract_month,
            "20261218"
        );
    }

    #[rstest]
    #[case::instrument_info(false, false)]
    #[case::contract(true, false)]
    #[case::contract_details(true, true)]
    #[tokio::test]
    async fn test_load_cache_cuts_cached_expiry_to_date(
        #[case] with_contract: bool,
        #[case] with_details: bool,
    ) {
        let (provider, _temp_dir) = create_test_provider_with_cache();
        let cache_path = provider.config.cache_path.as_ref().unwrap().clone();
        let instrument_id = InstrumentId::new(Symbol::from("ESZ6"), Venue::from("CME"));
        let contract = create_future_contract_with_expiry_time();
        let instrument = create_test_instrument_with_info(
            instrument_id,
            Some(create_contract_info(&contract, Some(1))),
        );
        let details = ibapi::contracts::ContractDetails {
            contract: contract.clone(),
            price_magnifier: 1,
            time_zone_id: String::from("US/Central"),
            ..Default::default()
        };

        provider.cache_instrument(
            instrument_id,
            instrument,
            with_details.then_some(details),
            with_contract.then_some(contract),
            Some(1),
            false,
        );
        provider.save_cache(&cache_path).await.unwrap();

        let new_provider =
            InteractiveBrokersInstrumentProvider::new(provider.config.clone()).unwrap();

        assert!(new_provider.load_cache(&cache_path).await.unwrap());
        assert_eq!(
            new_provider
                .resolve_contract_for_instrument(instrument_id)
                .unwrap()
                .last_trade_date_or_contract_month,
            "20261218"
        );

        if with_details {
            let details = new_provider
                .instrument_id_to_ib_contract_details(&instrument_id)
                .unwrap();
            assert_eq!(
                details.contract.last_trade_date_or_contract_month,
                "20261218"
            );
            assert_eq!(details.last_trade_time, "08:30:00 US/Central");
        }
    }

    #[tokio::test]
    async fn test_load_cache_missing_file() {
        let (provider, _temp_dir) = create_test_provider_with_cache();
        let cache_path = "/nonexistent/path/cache.json";

        let result = provider.load_cache(cache_path).await;
        assert!(
            result.is_ok(),
            "load_cache should not error on missing file"
        );
        assert!(
            !result.unwrap(),
            "load_cache should return false for missing file"
        );
    }

    #[tokio::test]
    async fn test_load_cache_expired() {
        let (provider, _temp_dir) = create_test_provider_with_cache();
        let cache_path = provider.config.cache_path.as_ref().unwrap().clone();

        // Create an expired cache manually
        let old_timestamp = Timestamp::now() - jiff::SignedDuration::from_hours(24 * (10));

        let expired_cache = InstrumentCache {
            cache_timestamp: old_timestamp,
            contract_id_to_instrument_id: vec![],
            price_magnifiers: vec![],
            contracts: vec![],
            contract_details: vec![],
            instruments: vec![],
        };

        let json = serde_json::to_string_pretty(&expired_cache).unwrap();
        fs::write(&cache_path, json).unwrap();

        // Try to load with validity_days = 7
        let result = provider.load_cache(&cache_path).await;
        assert!(
            result.is_ok(),
            "load_cache should not error on expired cache"
        );
        assert!(
            !result.unwrap(),
            "load_cache should return false for expired cache"
        );
    }
    #[rstest]
    fn restored_cache_seeds_provider_with_price_magnifier() {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let mut cache = Cache::default();
        assert_eq!(provider.seed_from_cache(&cache), 0);
        let instrument_id = InstrumentId::from("AAPL.NASDAQ");

        let contract = Contract {
            contract_id: 265598,
            symbol: ibapi::contracts::Symbol::from("AAPL"),
            security_type: SecurityType::Stock,
            exchange: Exchange::from("SMART"),
            currency: ibapi::contracts::Currency::from("USD"),
            ..Default::default()
        };

        let instrument = create_test_instrument_with_info(
            instrument_id,
            Some(create_contract_info(&contract, Some(100))),
        );
        cache.add_instrument(instrument.clone()).unwrap();
        let seeded = provider.seed_from_cache(&cache);
        assert_eq!(seeded, 1);
        assert_eq!(provider.find(&instrument_id), Some(instrument));
        assert_eq!(provider.get_price_magnifier(&instrument_id), 100);
        assert_eq!(
            provider
                .resolve_contract_for_instrument(instrument_id)
                .unwrap(),
            contract
        );
    }

    #[rstest]
    #[case(None)]
    #[case(Some(-1))]
    fn cached_instrument_without_valid_magnifier_requires_qualification(
        #[case] magnifier: Option<i32>,
    ) {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let instrument_id = InstrumentId::from("AAPL.NASDAQ");

        let contract = Contract {
            contract_id: 265598,
            ..Default::default()
        };

        let instrument = create_test_instrument_with_info(
            instrument_id,
            Some(create_contract_info(&contract, magnifier)),
        );
        assert_eq!(provider.add_cached_instruments([instrument]), 0);
        assert_eq!(provider.find(&instrument_id), None);
    }

    #[rstest]
    fn price_magnifier_is_none_until_contract_metadata_is_loaded() {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let instrument_id = InstrumentId::from("ZCZ6.CBOT");
        provider
            .instruments
            .insert(instrument_id, create_test_instrument(instrument_id));

        assert_eq!(provider.price_magnifier(&instrument_id), None);
        assert_eq!(provider.get_price_magnifier(&instrument_id), 1);

        let details = ibapi::contracts::ContractDetails {
            price_magnifier: 100,
            ..Default::default()
        };
        provider.contract_details.insert(instrument_id, details);

        assert_eq!(provider.price_magnifier(&instrument_id), Some(100));
        assert_eq!(provider.get_price_magnifier(&instrument_id), 100);
    }

    #[rstest]
    fn instrument_cached_after_seeding_supplies_its_price_magnifier() {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let mut cache = Cache::default();
        let instrument_id = InstrumentId::from("ZCZ6.CBOT");
        assert_eq!(provider.seed_from_cache(&cache), 0);
        assert_eq!(
            provider.seed_price_magnifier_from_cache(&cache, &instrument_id),
            None
        );

        let contract = Contract {
            contract_id: 815824252,
            symbol: ibapi::contracts::Symbol::from("ZC"),
            security_type: SecurityType::Future,
            exchange: Exchange::from("CBOT"),
            currency: ibapi::contracts::Currency::from("USD"),
            ..Default::default()
        };
        let instrument = create_test_instrument_with_info(
            instrument_id,
            Some(create_contract_info(&contract, Some(100))),
        );
        cache.add_instrument(instrument).unwrap();

        assert_eq!(
            provider.seed_price_magnifier_from_cache(&cache, &instrument_id),
            Some(100)
        );
        assert_eq!(
            provider
                .resolve_contract_for_instrument(instrument_id)
                .unwrap(),
            contract
        );
    }

    #[tokio::test]
    async fn ensure_price_magnifier_skips_the_request_when_the_magnifier_is_known() {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let instrument_id = InstrumentId::from("ZCZ6.CBOT");
        provider.price_magnifiers.insert(instrument_id, 100);
        let requests = AtomicUsize::new(0);

        let magnifier = provider
            .ensure_price_magnifier_with(instrument_id, Duration::from_secs(5), || async {
                requests.fetch_add(1, Ordering::SeqCst);
                Ok(())
            })
            .await
            .unwrap();

        assert_eq!(magnifier, 100);
        assert_eq!(requests.load(Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn ensure_price_magnifier_takes_it_from_requested_contract_details() {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let instrument_id = InstrumentId::from("ZCZ6.CBOT");

        let magnifier = provider
            .ensure_price_magnifier_with(instrument_id, Duration::from_secs(5), || async {
                let details = ibapi::contracts::ContractDetails {
                    price_magnifier: 100,
                    ..Default::default()
                };
                provider.contract_details.insert(instrument_id, details);
                Ok(())
            })
            .await
            .unwrap();

        assert_eq!(magnifier, 100);
    }

    #[rstest]
    #[case::no_details(false, "IB price magnifier for ZCZ6.CBOT is unknown")]
    #[case::request_failed(true, "Failed to load IB contract details for ZCZ6.CBOT")]
    #[tokio::test]
    async fn ensure_price_magnifier_fails_when_ib_supplies_none(
        #[case] request_fails: bool,
        #[case] expected: &str,
    ) {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let instrument_id = InstrumentId::from("ZCZ6.CBOT");

        let error = provider
            .ensure_price_magnifier_with(instrument_id, Duration::from_secs(5), || async {
                if request_fails {
                    anyhow::bail!("Socket disconnected");
                }
                Ok(())
            })
            .await
            .unwrap_err();

        assert!(format!("{error:#}").starts_with(expected), "{error:#}");
        assert_eq!(provider.price_magnifier(&instrument_id), None);
    }

    #[rstest]
    #[tokio::test(start_paused = true)]
    async fn ensure_price_magnifier_times_out_when_the_request_never_completes() {
        let provider = InteractiveBrokersInstrumentProvider::new(Default::default()).unwrap();
        let instrument_id = InstrumentId::from("ZCZ6.CBOT");
        let timeout = Duration::from_secs(5);
        let start = tokio::time::Instant::now();

        let error = provider
            .ensure_price_magnifier_with(instrument_id, timeout, std::future::pending)
            .await
            .unwrap_err();

        assert_eq!(
            format!("{error:#}"),
            "Timed out after 5s loading IB contract details for ZCZ6.CBOT"
        );
        assert_eq!(tokio::time::Instant::now() - start, timeout);
        assert_eq!(provider.price_magnifier(&instrument_id), None);
    }

    #[rstest]
    fn test_provider_rejects_duplicate_contracts_before_loading() {
        let config = serde_json::from_value(serde_json::json!({
            "load_contracts": [{"conId": 265598}, {"conId": 265598}]
        }))
        .unwrap();
        let error = InteractiveBrokersInstrumentProvider::new(config).unwrap_err();

        assert_eq!(error, ConfigError::duplicate("load_contracts", None));
    }
}