nautilus_common/cache/

mod.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! In-memory cache for market and execution data, with optional persistent backing.
//!
//! Provides methods to load, query, and update cached data such as instruments, orders, and prices.

pub mod config;
pub mod database;
pub mod fifo;
pub mod quote;
pub mod refs;

mod api;
mod bounded;
mod error;
mod filter;
mod index;
mod position;
mod view;

#[cfg(test)]
mod tests;

use std::{
    borrow::Cow,
    cell::RefCell,
    cmp::{Ordering, Reverse},
    fmt::{Debug, Display},
    rc::Rc,
    time::{SystemTime, UNIX_EPOCH},
};

use ahash::{AHashMap, AHashSet};
pub use api::CacheApi; // Re-export
use bounded::BoundedVecDeque;
use bytes::Bytes;
pub use config::CacheConfig; // Re-export
use database::{CacheDatabaseAdapter, CacheMap, register_loaded_currencies};
pub use error::{
    ACCOUNT_NOT_FOUND, AccountLookupError, CURRENCY_NOT_FOUND, CurrencyLookupError,
    INSTRUMENT_NOT_FOUND, InstrumentLookupError, ORDER_BOOK_NOT_FOUND, ORDER_LIST_NOT_FOUND,
    ORDER_NOT_FOUND, OWN_ORDER_BOOK_NOT_FOUND, OrderBookLookupError, OrderListLookupError,
    OrderLookupError, OwnOrderBookLookupError, POSITION_NOT_FOUND, PositionLookupError,
    SYNTHETIC_INSTRUMENT_NOT_FOUND, SyntheticInstrumentLookupError, VenueOrderIdOwnershipError,
};
use filter::{FilterSources, intersect_pair_or_many};
use index::CacheIndex;
use indexmap::IndexMap;
use nautilus_core::{
    DurationNanos, SharedCell, UnixNanos,
    correctness::{
        check_key_not_in_map, check_predicate_false, check_slice_not_empty, check_valid_string_utf8,
    },
};
use nautilus_model::{
    accounts::{Account, AccountAny},
    data::{
        Bar, BarType, FundingRateUpdate, GreeksData, IndexPriceUpdate, InstrumentClose,
        InstrumentStatus, MarkPriceUpdate, QuoteTick, TradeTick, YieldCurve,
        option_chain::OptionGreeks,
    },
    enums::{
        AggregationSource, ContingencyType, InstrumentClass, OmsType, OrderSide, PositionSide,
        PriceType,
    },
    events::{AccountState, OrderEventAny, OrderFilled, PositionAdjusted},
    identifiers::{
        AccountId, ActorId, ClientId, ClientOrderId, ExecAlgorithmId, InstrumentId, OrderListId,
        PositionId, StrategyId, Venue, VenueOrderId,
    },
    instruments::{Instrument, InstrumentAny, SyntheticInstrument},
    orderbook::{
        OrderBook,
        own::{OwnOrderBook, should_handle_own_book_order},
    },
    orders::{Order, OrderAny, OrderError, OrderList},
    position::{Position, PositionReplayEvent},
    types::{
        Currency, Money, Price, Quantity,
        fixed::{FIXED_PRECISION, MAX_FLOAT_PRECISION, check_fixed_precision},
        price::PriceRaw,
    },
};
pub use position::CacheSnapshotRef;
use position::PositionSnapshotFrame;
pub use refs::{AccountRef, AccountRefMut, OrderRef, OrderRefMut, PositionRef, PositionRefMut};
use rust_decimal::Decimal;
use ustr::Ustr;
pub use view::CacheView; // Re-export

use crate::xrate::get_exchange_rate;

/// A common in-memory `Cache` for market and execution related data.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.common", unsendable)
)]
pub struct Cache {
    config: CacheConfig,
    index: CacheIndex,
    database: Option<Box<dyn CacheDatabaseAdapter>>,
    general: AHashMap<String, Bytes>,
    currencies: AHashMap<Ustr, Currency>,
    instruments: AHashMap<InstrumentId, InstrumentAny>,
    instrument_closes: AHashMap<InstrumentId, InstrumentClose>,
    synthetics: AHashMap<InstrumentId, SyntheticInstrument>,
    books: AHashMap<InstrumentId, OrderBook>,
    own_books: AHashMap<InstrumentId, OwnOrderBook>,
    quotes: AHashMap<InstrumentId, BoundedVecDeque<QuoteTick>>,
    trades: AHashMap<InstrumentId, BoundedVecDeque<TradeTick>>,
    mark_xrates: AHashMap<(Currency, Currency), f64>,
    mark_prices: AHashMap<InstrumentId, BoundedVecDeque<MarkPriceUpdate>>,
    index_prices: AHashMap<InstrumentId, BoundedVecDeque<IndexPriceUpdate>>,
    funding_rates: AHashMap<InstrumentId, BoundedVecDeque<FundingRateUpdate>>,
    instrument_statuses: AHashMap<InstrumentId, BoundedVecDeque<InstrumentStatus>>,
    bars: AHashMap<BarType, BoundedVecDeque<Bar>>,
    greeks: AHashMap<InstrumentId, GreeksData>,
    option_greeks: AHashMap<InstrumentId, OptionGreeks>,
    yield_curves: AHashMap<String, YieldCurve>,
    external_order_claims: AHashMap<InstrumentId, StrategyId>,
    client_accounts: AHashMap<ClientId, AccountId>,
    client_routes: AHashMap<Venue, ClientId>,
    default_client_id: Option<ClientId>,
    external_data_clients: AHashSet<ClientId>,
    external_exec_clients: AHashSet<ClientId>,
    accounts: AHashMap<AccountId, SharedCell<AccountAny>>,
    orders: AHashMap<ClientOrderId, SharedCell<OrderAny>>,
    orders_dispatched: AHashSet<ClientOrderId>,
    order_lists: AHashMap<OrderListId, OrderList>,
    positions: AHashMap<PositionId, SharedCell<Position>>,
    position_snapshots: AHashMap<PositionId, Vec<PositionSnapshotFrame>>,
    position_snapshot_revisions: AHashMap<PositionId, u64>,
    #[cfg(feature = "defi")]
    pub(crate) defi: crate::defi::cache::DefiCache,
}

impl Debug for Cache {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct(stringify!(Cache))
            .field("config", &self.config)
            .field("index", &self.index)
            .field("general", &self.general)
            .field("currencies", &self.currencies)
            .field("instruments", &self.instruments)
            .field("synthetics", &self.synthetics)
            .field("books", &self.books)
            .field("own_books", &self.own_books)
            .field("quotes", &self.quotes)
            .field("trades", &self.trades)
            .field("mark_xrates", &self.mark_xrates)
            .field("mark_prices", &self.mark_prices)
            .field("index_prices", &self.index_prices)
            .field("funding_rates", &self.funding_rates)
            .field("instrument_statuses", &self.instrument_statuses)
            .field("instrument_closes", &self.instrument_closes)
            .field("bars", &self.bars)
            .field("greeks", &self.greeks)
            .field("option_greeks", &self.option_greeks)
            .field("yield_curves", &self.yield_curves)
            .field("external_order_claims", &self.external_order_claims)
            .field("client_accounts", &self.client_accounts)
            .field("client_routes", &self.client_routes)
            .field("default_client_id", &self.default_client_id)
            .field("external_data_clients", &self.external_data_clients)
            .field("external_exec_clients", &self.external_exec_clients)
            .field("accounts", &self.accounts)
            .field("orders", &self.orders)
            .field("orders_dispatched", &self.orders_dispatched)
            .field("order_lists", &self.order_lists)
            .field("positions", &self.positions)
            .field("position_snapshots", &self.position_snapshots)
            .finish()
    }
}

impl Default for Cache {
    /// Creates a new default [`Cache`] instance.
    fn default() -> Self {
        Self::new(Some(CacheConfig::default()), None)
    }
}

impl Cache {
    /// Creates a new [`Cache`] instance with optional configuration and database adapter.
    #[must_use]
    /// # Note
    ///
    /// Uses provided `CacheConfig` or defaults, and optional `CacheDatabaseAdapter` for persistence.
    ///
    /// # Panics
    ///
    /// Panics if the cache config has a tick or bar capacity outside `[1, 1_000_000]`.
    pub fn new(
        config: Option<CacheConfig>,
        database: Option<Box<dyn CacheDatabaseAdapter>>,
    ) -> Self {
        Self::try_new(config, database).expect("invalid `CacheConfig`")
    }

    /// Creates a new [`Cache`] instance with optional configuration and database adapter.
    ///
    /// # Errors
    ///
    /// Returns an error if the cache configuration is invalid.
    pub fn try_new(
        config: Option<CacheConfig>,
        database: Option<Box<dyn CacheDatabaseAdapter>>,
    ) -> crate::config::ConfigResult<Self> {
        let config = config.unwrap_or_default();
        config.validate()?;

        Ok(Self {
            config,
            index: CacheIndex::default(),
            database,
            general: AHashMap::new(),
            currencies: AHashMap::new(),
            instruments: AHashMap::new(),
            instrument_closes: AHashMap::new(),
            synthetics: AHashMap::new(),
            books: AHashMap::new(),
            own_books: AHashMap::new(),
            quotes: AHashMap::new(),
            trades: AHashMap::new(),
            mark_xrates: AHashMap::new(),
            mark_prices: AHashMap::new(),
            index_prices: AHashMap::new(),
            funding_rates: AHashMap::new(),
            instrument_statuses: AHashMap::new(),
            bars: AHashMap::new(),
            greeks: AHashMap::new(),
            option_greeks: AHashMap::new(),
            yield_curves: AHashMap::new(),
            external_order_claims: AHashMap::new(),
            client_accounts: AHashMap::new(),
            client_routes: AHashMap::new(),
            default_client_id: None,
            external_data_clients: AHashSet::new(),
            external_exec_clients: AHashSet::new(),
            accounts: AHashMap::new(),
            orders: AHashMap::new(),
            orders_dispatched: AHashSet::new(),
            order_lists: AHashMap::new(),
            positions: AHashMap::new(),
            position_snapshots: AHashMap::new(),
            position_snapshot_revisions: AHashMap::new(),
            #[cfg(feature = "defi")]
            defi: crate::defi::cache::DefiCache::default(),
        })
    }

    /// Returns the cache instances memory address.
    #[must_use]
    pub fn memory_address(&self) -> String {
        format!("{:?}", std::ptr::from_ref(self))
    }

    /// Returns the strategy claiming external orders for `instrument_id`.
    #[must_use]
    pub fn external_order_claim(&self, instrument_id: &InstrumentId) -> Option<StrategyId> {
        self.external_order_claims.get(instrument_id).copied()
    }

    /// Returns instrument IDs with an external order claim.
    ///
    /// Passing `Some(strategy_id)` filters the result to claims owned by that strategy. Passing
    /// `None` returns every claimed instrument ID.
    #[must_use]
    pub fn external_order_claim_instrument_ids(
        &self,
        strategy_id: Option<StrategyId>,
    ) -> AHashSet<InstrumentId> {
        self.external_order_claims
            .iter()
            .filter_map(|(instrument_id, owner)| {
                strategy_id
                    .is_none_or(|strategy_id| *owner == strategy_id)
                    .then_some(*instrument_id)
            })
            .collect()
    }

    /// Replaces the external order claims owned by `strategy_id`.
    ///
    /// External orders, fills, and materialized reconciliation activity for matching instrument
    /// IDs are assigned to the strategy. Existing claims owned by other strategies are preserved.
    ///
    /// The operation is atomic: either every requested instrument is claimed or the cache is
    /// unchanged. Passing an empty slice clears all claims owned by the strategy.
    ///
    /// # Errors
    ///
    /// Returns an error if an instrument is repeated or claimed by another strategy.
    pub fn set_external_order_claims(
        &mut self,
        strategy_id: StrategyId,
        instrument_ids: &[InstrumentId],
    ) -> anyhow::Result<()> {
        let mut requested = AHashSet::with_capacity(instrument_ids.len());

        for instrument_id in instrument_ids {
            if !requested.insert(*instrument_id) {
                anyhow::bail!(
                    "External order claim for {instrument_id} appears more than once for {strategy_id}"
                );
            }

            if let Some(existing) = self.external_order_claims.get(instrument_id)
                && *existing != strategy_id
            {
                anyhow::bail!(
                    "External order claim for {instrument_id} already exists for {existing}"
                );
            }
        }

        self.external_order_claims
            .retain(|_, owner| *owner != strategy_id);
        self.external_order_claims.extend(
            requested
                .into_iter()
                .map(|instrument_id| (instrument_id, strategy_id)),
        );

        Ok(())
    }

    /// Adds external order claims for `strategy_id` without replacing its existing claims.
    ///
    /// # Errors
    ///
    /// Returns an error if an instrument is repeated or already has a claim.
    pub fn register_external_order_claims(
        &mut self,
        strategy_id: StrategyId,
        instrument_ids: &[InstrumentId],
    ) -> anyhow::Result<()> {
        let mut requested = AHashSet::with_capacity(instrument_ids.len());

        for instrument_id in instrument_ids {
            if !requested.insert(*instrument_id) {
                anyhow::bail!(
                    "External order claim for {instrument_id} appears more than once for {strategy_id}"
                );
            }

            if let Some(existing) = self.external_order_claims.get(instrument_id) {
                anyhow::bail!(
                    "External order claim for {instrument_id} already exists for {existing}"
                );
            }
        }

        self.external_order_claims.extend(
            requested
                .into_iter()
                .map(|instrument_id| (instrument_id, strategy_id)),
        );

        Ok(())
    }

    /// Returns the account ID registered for the execution client `client_id`.
    ///
    /// The registration does not depend on account issuers or client names. The account itself
    /// may not be cached yet, for example before the client reports its first account state.
    #[must_use]
    pub fn account_id_for_client(&self, client_id: &ClientId) -> Option<&AccountId> {
        self.client_accounts.get(client_id)
    }

    /// Registers `account_id` as the account of the execution client `client_id`, replacing any
    /// previous registration for the client.
    ///
    /// Registrations survive [`Self::clear_index`] and [`Self::reset`].
    pub fn add_client_account(&mut self, client_id: ClientId, account_id: AccountId) {
        self.client_accounts.insert(client_id, account_id);
    }

    /// Removes the account registration of the execution client `client_id`.
    pub fn remove_client_account(&mut self, client_id: &ClientId) {
        self.client_accounts.remove(client_id);
    }

    /// Returns the execution client registered as the route for `venue`, else the default client.
    #[must_use]
    pub fn client_id_for_venue(&self, venue: &Venue) -> Option<&ClientId> {
        self.client_routes
            .get(venue)
            .or(self.default_client_id.as_ref())
    }

    /// Registers the execution client `client_id` as the route for commands on `venue`, replacing
    /// any previous route for the venue.
    ///
    /// Routes and the default client survive [`Self::clear_index`] and [`Self::reset`].
    pub fn add_client_route(&mut self, client_id: ClientId, venue: Venue) {
        self.client_routes.insert(venue, client_id);
    }

    /// Registers the execution client `client_id` as the default route for venues without a
    /// route, replacing any previous default.
    pub fn set_default_client(&mut self, client_id: ClientId) {
        self.default_client_id = Some(client_id);
    }

    /// Removes the venue routes and default route of the execution client `client_id`.
    pub fn remove_client_routes(&mut self, client_id: &ClientId) {
        self.client_routes.retain(|_, routed| routed != client_id);

        if self.default_client_id.as_ref() == Some(client_id) {
            self.default_client_id = None;
        }
    }

    /// Returns whether `client_id` is registered as an external data client, whose commands
    /// are published for another process to execute.
    #[must_use]
    pub fn is_external_data_client(&self, client_id: &ClientId) -> bool {
        self.external_data_clients.contains(client_id)
    }

    /// Registers `client_id` as an external data client.
    ///
    /// Registrations survive [`Self::clear_index`] and [`Self::reset`].
    pub fn add_external_data_client(&mut self, client_id: ClientId) {
        self.external_data_clients.insert(client_id);
    }

    /// Returns whether `client_id` is registered as an external execution client, whose commands
    /// are published for another process to execute.
    #[must_use]
    pub fn is_external_exec_client(&self, client_id: &ClientId) -> bool {
        self.external_exec_clients.contains(client_id)
    }

    /// Registers `client_id` as an external execution client.
    ///
    /// Registrations survive [`Self::clear_index`] and [`Self::reset`].
    pub fn add_external_exec_client(&mut self, client_id: ClientId) {
        self.external_exec_clients.insert(client_id);
    }

    /// Sets the cache database adapter for persistence.
    ///
    /// This allows setting or replacing the database adapter after cache construction.
    pub fn set_database(&mut self, database: Box<dyn CacheDatabaseAdapter>) {
        let type_name = std::any::type_name_of_val(&*database);
        log::info!("Cache database adapter set: {type_name}");
        self.database = Some(database);
    }

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

    /// Clears and reloads general entries from the database into the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if loading general cache data fails.
    pub fn cache_general(&mut self) -> anyhow::Result<()> {
        self.general = match &mut self.database {
            Some(db) => db.load()?,
            None => AHashMap::new(),
        };

        log::info!(
            "Cached {} general object(s) from database",
            self.general.len()
        );
        Ok(())
    }

    /// Loads all core caches from the database.
    ///
    /// The loaded instrument closes replace all closes already held in memory. This includes values
    /// added directly before the persistent cache is loaded.
    ///
    /// # Errors
    ///
    /// Returns an error if loading cache data fails.
    pub async fn cache_all(&mut self) -> anyhow::Result<()> {
        let mut cache_map = match &self.database {
            Some(db) => db.load_all().await?,
            None => CacheMap::default(),
        };

        // An adapter registers the loaded currencies before decoding their dependents; repeating
        // it here keeps the cached map consistent with the registry whatever the adapter did.
        register_loaded_currencies(&mut cache_map.currencies)?;
        self.currencies = cache_map.currencies;
        self.instruments = cache_map.instruments;
        self.instrument_closes = cache_map.instrument_closes;
        self.synthetics = cache_map.synthetics;
        self.accounts = cache_map
            .accounts
            .into_iter()
            .map(|(id, account)| (id, SharedCell::new(account)))
            .collect();
        self.orders = cache_map
            .orders
            .into_iter()
            .map(|(id, order)| (id, SharedCell::new(order)))
            .collect();
        self.positions = cache_map
            .positions
            .into_iter()
            .map(|(id, position)| (id, SharedCell::new(position)))
            .collect();

        if let Some(db) = &self.database {
            let order_position = db.load_index_order_position()?;
            self.index.order_position = self.sanitize_order_position_index(order_position);
            self.index.order_client = db.load_index_order_client()?;
        }

        self.cache_position_oms()?;
        self.assign_position_ids_to_contingencies();
        Ok(())
    }

    /// Clears and reloads the currency cache from the database.
    ///
    /// # Errors
    ///
    /// Returns an error if loading currencies cache fails.
    pub async fn cache_currencies(&mut self) -> anyhow::Result<()> {
        self.currencies = match &mut self.database {
            Some(db) => db.load_currencies().await?,
            None => AHashMap::new(),
        };

        register_loaded_currencies(&mut self.currencies)?;

        log::info!("Cached {} currencies from database", self.currencies.len());
        Ok(())
    }

    /// Clears and reloads the instrument cache from the database.
    ///
    /// # Errors
    ///
    /// Returns an error if loading instruments cache fails.
    pub async fn cache_instruments(&mut self) -> anyhow::Result<()> {
        self.instruments = match &mut self.database {
            Some(db) => db.load_instruments().await?,
            None => AHashMap::new(),
        };

        log::info!("Cached {} instruments from database", self.general.len());
        Ok(())
    }

    /// Clears and reloads the synthetic instrument cache from the database.
    ///
    /// # Errors
    ///
    /// Returns an error if loading synthetic instruments cache fails.
    pub async fn cache_synthetics(&mut self) -> anyhow::Result<()> {
        self.synthetics = match &mut self.database {
            Some(db) => db.load_synthetics().await?,
            None => AHashMap::new(),
        };

        log::info!(
            "Cached {} synthetic instruments from database",
            self.general.len()
        );
        Ok(())
    }

    /// Clears and reloads the account cache from the database.
    ///
    /// # Errors
    ///
    /// Returns an error if loading accounts cache fails.
    pub async fn cache_accounts(&mut self) -> anyhow::Result<()> {
        self.accounts = match &mut self.database {
            Some(db) => db
                .load_accounts()
                .await?
                .into_iter()
                .map(|(id, account)| (id, SharedCell::new(account)))
                .collect(),
            None => AHashMap::new(),
        };

        log::info!(
            "Cached {} synthetic instruments from database",
            self.general.len()
        );
        Ok(())
    }

    /// Clears and reloads the order cache from the database.
    ///
    /// # Errors
    ///
    /// Returns an error if loading orders cache fails.
    pub async fn cache_orders(&mut self) -> anyhow::Result<()> {
        self.orders = match &mut self.database {
            Some(db) => db
                .load_orders()
                .await?
                .into_iter()
                .map(|(id, order)| (id, SharedCell::new(order)))
                .collect(),
            None => AHashMap::new(),
        };

        if let Some(db) = &self.database {
            let order_position = db.load_index_order_position()?;
            self.index.order_position = self.sanitize_order_position_index(order_position);
            self.index.order_client = db.load_index_order_client()?;
        }

        log::info!("Cached {} orders from database", self.general.len());

        self.assign_position_ids_to_contingencies();
        Ok(())
    }

    fn sanitize_order_position_index(
        &self,
        mut order_position: AHashMap<ClientOrderId, PositionId>,
    ) -> AHashMap<ClientOrderId, PositionId> {
        let original_len = order_position.len();
        order_position.retain(|client_order_id, _| self.orders.contains_key(client_order_id));
        let removed = original_len - order_position.len();

        if removed > 0 {
            log::warn!(
                "Filtered {removed} stale order-position index entries without backing orders during cache load"
            );
        }

        order_position
    }

    /// Clears and reloads the position cache from the database.
    ///
    /// # Errors
    ///
    /// Returns an error if loading positions cache fails.
    pub async fn cache_positions(&mut self) -> anyhow::Result<()> {
        self.positions = match &mut self.database {
            Some(db) => db
                .load_positions()
                .await?
                .into_iter()
                .map(|(id, position)| (id, SharedCell::new(position)))
                .collect(),
            None => AHashMap::new(),
        };

        self.cache_position_oms()?;
        log::info!("Cached {} positions from database", self.general.len());
        Ok(())
    }

    fn cache_position_oms(&mut self) -> anyhow::Result<()> {
        let persisted = match &self.database {
            Some(database) => database.load()?,
            None => self.general.clone(),
        };

        self.general
            .retain(|key, _| !key.starts_with(POSITION_OMS_KEY_PREFIX));

        for (key, value) in persisted {
            if !key.starts_with(POSITION_OMS_KEY_PREFIX) {
                continue;
            }
            self.general.insert(key, value);
        }

        self.index_position_oms();
        Ok(())
    }

    /// Rebuilds entity-derived indexes from cached accounts, orders, and positions.
    ///
    /// Retains venue-order-ID aliases, order-position and order-client mappings, and local
    /// pending-cancel state only for cached orders. Rebuilding can update mappings from order
    /// state. Position OMS entries are rebuilt from the general cache.
    /// Reverse venue-order-ID lookups include historical IDs; forward lookups use the current ID.
    /// To discard independently maintained index state, call [`Self::clear_index`] before
    /// reloading entities.
    pub fn build_index(&mut self) {
        log::debug!("Building index");

        self.index.clear_entities();
        self.index
            .venue_order_ids
            .retain(|_, client_order_id| self.orders.contains_key(client_order_id));
        self.index
            .client_order_ids
            .retain(|client_order_id, _| self.orders.contains_key(client_order_id));
        self.index
            .order_position
            .retain(|client_order_id, _| self.orders.contains_key(client_order_id));
        self.index
            .order_client
            .retain(|client_order_id, _| self.orders.contains_key(client_order_id));
        self.index
            .orders_pending_cancel
            .retain(|client_order_id| self.orders.contains_key(client_order_id));

        // Index accounts
        for account_id in self.accounts.keys() {
            self.index.add_venue_account(*account_id);
        }

        // Index orders
        for (client_order_id, order_cell) in &self.orders {
            let order = order_cell.borrow();
            let instrument_id = order.instrument_id();
            let venue = instrument_id.venue;
            let strategy_id = order.strategy_id();

            // 1: Build index.venue_orders -> {Venue, {ClientOrderId}}
            self.index
                .venue_orders
                .entry(venue)
                .or_default()
                .insert(*client_order_id);

            // 2: Build index.venue_order_ids -> {VenueOrderId, ClientOrderId}
            //    and index.client_order_ids -> {ClientOrderId, VenueOrderId}
            for venue_order_id in order.venue_order_ids() {
                self.index
                    .venue_order_ids
                    .insert(*venue_order_id, *client_order_id);
            }

            if let Some(venue_order_id) = order.venue_order_id() {
                self.index
                    .client_order_ids
                    .insert(*client_order_id, venue_order_id);
            }

            // 3: Build index.order_position -> {ClientOrderId, PositionId}
            self.index.order_position.extend(
                order
                    .position_id()
                    .map(|position_id| (*client_order_id, position_id)),
            );

            // 4: Build index.order_strategy -> {ClientOrderId, StrategyId}
            self.index
                .order_strategy
                .insert(*client_order_id, strategy_id);

            // 5: Build index.instrument_orders -> {InstrumentId, {ClientOrderId}}
            self.index
                .instrument_orders
                .entry(instrument_id)
                .or_default()
                .insert(*client_order_id);

            // 6: Build index.strategy_orders -> {StrategyId, {ClientOrderId}}
            self.index
                .strategy_orders
                .entry(strategy_id)
                .or_default()
                .insert(*client_order_id);

            // 7: Build index.account_orders -> {AccountId, {ClientOrderId}}
            if let Some(account_id) = order.account_id() {
                self.index
                    .account_orders
                    .entry(account_id)
                    .or_default()
                    .insert(*client_order_id);
            }

            // 8: Build index.exec_algorithm_orders -> {ExecAlgorithmId, {ClientOrderId}}
            if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
                self.index
                    .exec_algorithm_orders
                    .entry(exec_algorithm_id)
                    .or_default()
                    .insert(*client_order_id);
                self.index.exec_algorithms.insert(exec_algorithm_id);
            }

            // 9: Build index.exec_spawn_orders -> {ClientOrderId, {ClientOrderId}}
            if let Some(exec_spawn_id) = order.exec_spawn_id() {
                self.index
                    .exec_spawn_orders
                    .entry(exec_spawn_id)
                    .or_default()
                    .insert(*client_order_id);
            }

            // 10: Build index.orders -> {ClientOrderId}
            self.index.orders.insert(*client_order_id);

            // 11: Build index.orders_active_local -> {ClientOrderId}
            if order.is_active_local() {
                self.index.orders_active_local.insert(*client_order_id);
            }

            // 12: Build index.orders_open -> {ClientOrderId}
            if order.is_open() {
                self.index.orders_open.insert(*client_order_id);
            }

            // 13: Build index.orders_closed -> {ClientOrderId}
            if order.is_closed() {
                self.index.orders_closed.insert(*client_order_id);
            }

            // 14: Build index.orders_emulated -> {ClientOrderId}
            if order.emulation_trigger().is_some() && !order.is_closed() {
                self.index.orders_emulated.insert(*client_order_id);
            }

            // 15: Build index.orders_inflight -> {ClientOrderId}
            if order.is_inflight() {
                self.index.orders_inflight.insert(*client_order_id);
            }

            // 16: Build index.strategies -> {StrategyId}
            self.index.strategies.insert(strategy_id);
        }

        // Index positions
        for (position_id, position_cell) in &self.positions {
            let position = position_cell.borrow();
            let instrument_id = position.instrument_id;
            let venue = instrument_id.venue;
            let strategy_id = position.strategy_id;

            // 1: Build index.venue_positions -> {Venue, {PositionId}}
            self.index
                .venue_positions
                .entry(venue)
                .or_default()
                .insert(*position_id);

            // 2: Build index.position_strategy -> {PositionId, StrategyId}
            self.index
                .position_strategy
                .insert(*position_id, strategy_id);

            // 3: Build index.position_orders -> {PositionId, {ClientOrderId}}
            let position_orders = self.index.position_orders.entry(*position_id).or_default();
            position_orders.extend(
                position
                    .client_order_ids()
                    .into_iter()
                    .filter(|client_order_id| self.orders.contains_key(client_order_id)),
            );

            // 4: Build index.instrument_positions -> {InstrumentId, {PositionId}}
            self.index
                .instrument_positions
                .entry(instrument_id)
                .or_default()
                .insert(*position_id);
            self.index
                .instrument_orders
                .entry(instrument_id)
                .or_default();

            // 5: Build index.strategy_positions -> {StrategyId, {PositionId}}
            self.index
                .strategy_positions
                .entry(strategy_id)
                .or_default()
                .insert(*position_id);
            self.index.strategy_orders.entry(strategy_id).or_default();

            // 6: Build index.account_positions -> {AccountId, {PositionId}}
            self.index
                .account_positions
                .entry(position.account_id)
                .or_default()
                .insert(*position_id);

            // 7: Build index.positions -> {PositionId}
            self.index.positions.insert(*position_id);

            // 8: Build index.positions_open -> {PositionId}
            if position.is_open() {
                self.index.positions_open.insert(*position_id);
            }

            // 9: Build index.positions_closed -> {PositionId}
            if position.is_closed() {
                self.index.positions_closed.insert(*position_id);
            }

            // 10: Build index.strategies -> {StrategyId}
            self.index.strategies.insert(strategy_id);
        }

        self.index_position_oms();
        let client_order_ids: Vec<_> = self.orders.keys().copied().collect();
        for client_order_id in client_order_ids {
            self.reindex_hedging_position(client_order_id);
        }
    }

    fn index_position_oms(&mut self) {
        self.index.position_oms.clear();

        for (key, value) in &self.general {
            let Some(position_id) = key.strip_prefix(POSITION_OMS_KEY_PREFIX) else {
                continue;
            };
            let position_id = PositionId::new(position_id);
            if !self.positions.contains_key(&position_id) {
                continue;
            }

            match serde_json::from_slice::<OmsType>(value) {
                Ok(oms_type) => {
                    self.index.position_oms.insert(position_id, oms_type);
                }
                Err(e) => {
                    log::error!("Failed to decode position OMS for {position_id}: {e}");
                }
            }
        }

        for position in self.positions.values().map(|cell| cell.borrow()) {
            if !self.index.position_oms.contains_key(&position.id)
                && resolve_position_oms(&position, OmsType::Unspecified) == OmsType::Netting
            {
                self.index
                    .position_oms
                    .insert(position.id, OmsType::Netting);
            }
        }
    }

    /// Returns whether the cache has a backing database.
    #[must_use]
    pub const fn has_backing(&self) -> bool {
        self.database.is_some()
    }

    /// Loads persisted actor state.
    ///
    /// Returns `None` when the cache has no backing database.
    ///
    /// # Errors
    ///
    /// Returns an error if loading actor state fails.
    pub fn load_actor_state(
        &self,
        actor_id: &ActorId,
    ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
        self.database
            .as_ref()
            .map(|database| database.load_actor(actor_id))
            .transpose()
            .map(|state| state.map(Self::decode_component_state))
    }

    /// Loads persisted strategy state.
    ///
    /// Returns `None` when the cache has no backing database.
    ///
    /// # Errors
    ///
    /// Returns an error if loading strategy state fails.
    pub fn load_strategy_state(
        &self,
        strategy_id: &StrategyId,
    ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
        self.database
            .as_ref()
            .map(|database| database.load_strategy(strategy_id))
            .transpose()
            .map(|state| state.map(Self::decode_component_state))
    }

    /// Persists actor state when the cache has a backing database.
    ///
    /// # Errors
    ///
    /// Returns an error if updating actor state fails.
    pub fn update_actor_state(
        &self,
        actor_id: &ActorId,
        state: &IndexMap<String, Vec<u8>>,
    ) -> anyhow::Result<()> {
        if let Some(database) = &self.database {
            database.update_actor(actor_id, &Self::encode_component_state(state))?;
        }
        Ok(())
    }

    /// Persists strategy state when the cache has a backing database.
    ///
    /// # Errors
    ///
    /// Returns an error if updating strategy state fails.
    pub fn update_strategy_state(
        &self,
        strategy_id: &StrategyId,
        state: &IndexMap<String, Vec<u8>>,
    ) -> anyhow::Result<()> {
        if let Some(database) = &self.database {
            database.update_strategy(strategy_id, &Self::encode_component_state(state))?;
        }
        Ok(())
    }

    fn decode_component_state(state: AHashMap<String, Bytes>) -> IndexMap<String, Vec<u8>> {
        state
            .into_iter()
            .map(|(key, value)| (key, value.to_vec()))
            .collect()
    }

    fn encode_component_state(state: &IndexMap<String, Vec<u8>>) -> AHashMap<String, Bytes> {
        state
            .iter()
            .map(|(key, value)| (key.clone(), Bytes::copy_from_slice(value)))
            .collect()
    }

    // Calculate the unrealized profit and loss (PnL) for `position`.
    #[must_use]
    pub fn calculate_unrealized_pnl(&self, position: &Position) -> Option<Money> {
        let Some(quote) = self.quote(&position.instrument_id) else {
            log::warn!(
                "Cannot calculate unrealized PnL for {}, no quotes for {}",
                position.id,
                position.instrument_id
            );
            return None;
        };

        // Use exit price for mark-to-market: longs exit at bid, shorts exit at ask
        let last = match position.side {
            PositionSide::Flat => {
                return Some(Money::zero(position.settlement_currency));
            }
            PositionSide::Long => quote.bid_price,
            PositionSide::Short => quote.ask_price,
        };

        position
            .try_unrealized_pnl(last)
            .inspect_err(|e| {
                log::error!("Cannot calculate unrealized PnL for {}: {e}", position.id);
            })
            .ok()
    }

    /// Checks integrity of data within the cache.
    ///
    /// All data should be loaded from the database prior to this call.
    /// If an error is found then a log error message will also be produced.
    ///
    /// # Panics
    ///
    /// Panics if failure calling system clock.
    #[must_use]
    pub fn check_integrity(&mut self) -> bool {
        let mut error_count = 0;
        let failure = "Integrity failure";

        // Get current timestamp in microseconds
        let timestamp_us = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("Time went backwards")
            .as_micros();

        log::info!("Checking data integrity");

        // Check object caches
        for account_id in self.accounts.keys() {
            if !self
                .index
                .venue_accounts
                .get(&account_id.get_issuer())
                .is_some_and(|account_ids| account_ids.contains(account_id))
            {
                log::error!(
                    "{failure} in accounts: {account_id} not found in `self.index.venue_accounts`",
                );
                error_count += 1;
            }
        }

        for (client_order_id, order_cell) in &self.orders {
            let order = order_cell.borrow();

            if !self.index.order_strategy.contains_key(client_order_id) {
                log::error!(
                    "{failure} in orders: {client_order_id} not found in `self.index.order_strategy`"
                );
                error_count += 1;
            }

            if !self.index.orders.contains(client_order_id) {
                log::error!(
                    "{failure} in orders: {client_order_id} not found in `self.index.orders`",
                );
                error_count += 1;
            }

            if order.is_inflight() && !self.index.orders_inflight.contains(client_order_id) {
                log::error!(
                    "{failure} in orders: {client_order_id} not found in `self.index.orders_inflight`",
                );
                error_count += 1;
            }

            if order.is_active_local() && !self.index.orders_active_local.contains(client_order_id)
            {
                log::error!(
                    "{failure} in orders: {client_order_id} not found in `self.index.orders_active_local`",
                );
                error_count += 1;
            }

            if order.is_open() && !self.index.orders_open.contains(client_order_id) {
                log::error!(
                    "{failure} in orders: {client_order_id} not found in `self.index.orders_open`",
                );
                error_count += 1;
            }

            if order.is_closed() && !self.index.orders_closed.contains(client_order_id) {
                log::error!(
                    "{failure} in orders: {client_order_id} not found in `self.index.orders_closed`",
                );
                error_count += 1;
            }

            if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
                if !self
                    .index
                    .exec_algorithm_orders
                    .contains_key(&exec_algorithm_id)
                {
                    log::error!(
                        "{failure} in orders: {client_order_id} not found in `self.index.exec_algorithm_orders`",
                    );
                    error_count += 1;
                }

                if order.exec_spawn_id().is_none()
                    && !self.index.exec_spawn_orders.contains_key(client_order_id)
                {
                    log::error!(
                        "{failure} in orders: {client_order_id} not found in `self.index.exec_spawn_orders`",
                    );
                    error_count += 1;
                }
            }
        }

        for (position_id, position_cell) in &self.positions {
            let position = position_cell.borrow();

            if !self.index.position_strategy.contains_key(position_id) {
                log::error!(
                    "{failure} in positions: {position_id} not found in `self.index.position_strategy`",
                );
                error_count += 1;
            }

            if !self.index.position_orders.contains_key(position_id) {
                log::error!(
                    "{failure} in positions: {position_id} not found in `self.index.position_orders`",
                );
                error_count += 1;
            }

            if !self.index.positions.contains(position_id) {
                log::error!(
                    "{failure} in positions: {position_id} not found in `self.index.positions`",
                );
                error_count += 1;
            }

            if position.is_open() && !self.index.positions_open.contains(position_id) {
                log::error!(
                    "{failure} in positions: {position_id} not found in `self.index.positions_open`",
                );
                error_count += 1;
            }

            if position.is_closed() && !self.index.positions_closed.contains(position_id) {
                log::error!(
                    "{failure} in positions: {position_id} not found in `self.index.positions_closed`",
                );
                error_count += 1;
            }
        }

        // Check indexes
        for account_id in self.index.venue_accounts.values().flatten() {
            if !self.accounts.contains_key(account_id) {
                log::error!(
                    "{failure} in `index.venue_accounts`: {account_id} not found in `self.accounts`",
                );
                error_count += 1;
            }
        }

        for client_order_id in self.index.venue_order_ids.values() {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.venue_order_ids`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        for client_order_id in self.index.client_order_ids.keys() {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.client_order_ids`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        for client_order_id in self.index.order_position.keys() {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.order_position`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        // Check indexes
        for client_order_id in self.index.order_strategy.keys() {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.order_strategy`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        for position_id in self.index.position_strategy.keys() {
            if !self.positions.contains_key(position_id) {
                log::error!(
                    "{failure} in `index.position_strategy`: {position_id} not found in `self.positions`",
                );
                error_count += 1;
            }
        }

        for position_id in self.index.position_orders.keys() {
            if !self.positions.contains_key(position_id) {
                log::error!(
                    "{failure} in `index.position_orders`: {position_id} not found in `self.positions`",
                );
                error_count += 1;
            }
        }

        for (instrument_id, client_order_ids) in &self.index.instrument_orders {
            for client_order_id in client_order_ids {
                if !self.orders.contains_key(client_order_id) {
                    log::error!(
                        "{failure} in `index.instrument_orders`: {instrument_id} not found in `self.orders`",
                    );
                    error_count += 1;
                }
            }
        }

        for instrument_id in self.index.instrument_positions.keys() {
            if !self.index.instrument_orders.contains_key(instrument_id) {
                log::error!(
                    "{failure} in `index.instrument_positions`: {instrument_id} not found in `index.instrument_orders`",
                );
                error_count += 1;
            }
        }

        for client_order_ids in self.index.strategy_orders.values() {
            for client_order_id in client_order_ids {
                if !self.orders.contains_key(client_order_id) {
                    log::error!(
                        "{failure} in `index.strategy_orders`: {client_order_id} not found in `self.orders`",
                    );
                    error_count += 1;
                }
            }
        }

        for position_ids in self.index.strategy_positions.values() {
            for position_id in position_ids {
                if !self.positions.contains_key(position_id) {
                    log::error!(
                        "{failure} in `index.strategy_positions`: {position_id} not found in `self.positions`",
                    );
                    error_count += 1;
                }
            }
        }

        for client_order_id in &self.index.orders {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.orders`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        for client_order_id in &self.index.orders_emulated {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.orders_emulated`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        for client_order_id in &self.index.orders_active_local {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.orders_active_local`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        for client_order_id in &self.index.orders_inflight {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.orders_inflight`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        for client_order_id in &self.index.orders_open {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.orders_open`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        for client_order_id in &self.index.orders_closed {
            if !self.orders.contains_key(client_order_id) {
                log::error!(
                    "{failure} in `index.orders_closed`: {client_order_id} not found in `self.orders`",
                );
                error_count += 1;
            }
        }

        for position_id in &self.index.positions {
            if !self.positions.contains_key(position_id) {
                log::error!(
                    "{failure} in `index.positions`: {position_id} not found in `self.positions`",
                );
                error_count += 1;
            }
        }

        for position_id in &self.index.positions_open {
            if !self.positions.contains_key(position_id) {
                log::error!(
                    "{failure} in `index.positions_open`: {position_id} not found in `self.positions`",
                );
                error_count += 1;
            }
        }

        for position_id in &self.index.positions_closed {
            if !self.positions.contains_key(position_id) {
                log::error!(
                    "{failure} in `index.positions_closed`: {position_id} not found in `self.positions`",
                );
                error_count += 1;
            }
        }

        for strategy_id in &self.index.strategies {
            if !self.index.strategy_orders.contains_key(strategy_id) {
                log::error!(
                    "{failure} in `index.strategies`: {strategy_id} not found in `index.strategy_orders`",
                );
                error_count += 1;
            }
        }

        for exec_algorithm_id in &self.index.exec_algorithms {
            if !self
                .index
                .exec_algorithm_orders
                .contains_key(exec_algorithm_id)
            {
                log::error!(
                    "{failure} in `index.exec_algorithms`: {exec_algorithm_id} not found in `index.exec_algorithm_orders`",
                );
                error_count += 1;
            }
        }

        let total_us = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("Time went backwards")
            .as_micros()
            - timestamp_us;

        if error_count == 0 {
            log::info!("Integrity check passed in {total_us}μs");
            true
        } else {
            log::error!(
                "Integrity check failed with {error_count} error{} in {total_us}μs",
                if error_count == 1 { "" } else { "s" },
            );
            false
        }
    }

    /// Checks for any residual open state and log warnings if any are found.
    ///
    ///'Open state' is considered to be open orders and open positions.
    #[must_use]
    pub fn check_residuals(&self) -> bool {
        log::debug!("Checking residuals");

        let mut residuals = false;

        // Check for any open orders
        for order in self.orders_open(None, None, None, None, None) {
            residuals = true;
            log::warn!("Residual {order}");
        }

        // Check for any open positions
        for position in self.positions_open(None, None, None, None, None) {
            residuals = true;
            log::warn!("Residual {position}");
        }

        residuals
    }

    /// Purges all closed orders from the cache that are older than `buffer_secs`.
    ///
    ///
    /// Only orders that have been closed for at least this amount of time will be purged.
    /// A value of 0 means purge all closed orders regardless of when they were closed.
    pub fn purge_closed_orders(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
        log::debug!(
            "Purging closed orders{}",
            if buffer_secs > 0 {
                format!(" with buffer_secs={buffer_secs}")
            } else {
                String::new()
            }
        );

        let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
            log::warn!(
                "Cannot purge closed orders: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
            );
            return;
        };
        let purge_cutoff = ts_now.checked_sub(buffer_ns);

        let mut affected_order_list_ids: AHashSet<OrderListId> = AHashSet::new();
        let mut purged_client_order_ids: AHashSet<ClientOrderId> = AHashSet::new();

        'outer: for client_order_id in self.index.orders_closed.clone() {
            let purge_target = self.orders.get(&client_order_id).and_then(|order_cell| {
                let order = order_cell.borrow();
                if order.is_closed()
                    && let Some(ts_closed) = order.ts_closed()
                    && purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
                {
                    let linked = order.linked_order_ids().map(<[_]>::to_vec);
                    let order_list_id = order.order_list_id();
                    Some((linked, order_list_id))
                } else {
                    None
                }
            });

            let Some((linked, order_list_id)) = purge_target else {
                continue;
            };

            // Check any linked orders (contingency orders)
            if let Some(linked_order_ids) = linked {
                for linked_order_id in &linked_order_ids {
                    if let Some(linked_order_cell) = self.orders.get(linked_order_id)
                        && linked_order_cell.borrow().is_open()
                    {
                        // Do not purge if linked order still open
                        continue 'outer;
                    }
                }
            }

            if let Some(order_list_id) = order_list_id {
                affected_order_list_ids.insert(order_list_id);
            }

            if self.purge_order_except_aliases(client_order_id) {
                purged_client_order_ids.insert(client_order_id);
            }
        }

        if !purged_client_order_ids.is_empty() {
            self.index
                .venue_order_ids
                .retain(|_, owner| !purged_client_order_ids.contains(owner));
        }

        for order_list_id in affected_order_list_ids {
            if let Some(order_list) = self.order_lists.get(&order_list_id) {
                let all_purged = order_list
                    .client_order_ids
                    .iter()
                    .all(|id| !self.orders.contains_key(id));

                if all_purged {
                    self.order_lists.remove(&order_list_id);
                    log::info!("Purged {order_list_id}");
                }
            }
        }
    }

    /// Purges all closed positions from the cache that are older than `buffer_secs`.
    pub fn purge_closed_positions(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
        log::debug!(
            "Purging closed positions{}",
            if buffer_secs > 0 {
                format!(" with buffer_secs={buffer_secs}")
            } else {
                String::new()
            }
        );

        let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
            log::warn!(
                "Cannot purge closed positions: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
            );
            return;
        };
        let purge_cutoff = ts_now.checked_sub(buffer_ns);

        for position_id in self.index.positions_closed.clone() {
            let should_purge = self.positions.get(&position_id).is_some_and(|cell| {
                let position = cell.borrow();
                position.is_closed()
                    && position.ts_closed.is_some_and(|ts_closed| {
                        purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
                    })
            });

            if should_purge {
                self.purge_position(position_id);
            }
        }
    }

    /// Purges the order with the `client_order_id` from the cache (if found).
    ///
    /// For safety, an order is prevented from being purged if it's open.
    pub fn purge_order(&mut self, client_order_id: ClientOrderId) {
        if self.purge_order_except_aliases(client_order_id) {
            self.index
                .venue_order_ids
                .retain(|_, owner| owner != &client_order_id);
        }
    }

    /// Removes the order and its indexes, leaving the reverse venue order ID aliases for the
    /// caller to sweep by owner, so a bulk purge pays for one pass rather than one pass per order.
    ///
    /// Returns whether the order was purged, so a skipped purge leaves its aliases intact.
    fn purge_order_except_aliases(&mut self, client_order_id: ClientOrderId) -> bool {
        struct OrderDetails {
            is_open: bool,
            instrument_id: InstrumentId,
            strategy_id: StrategyId,
            account_id: Option<AccountId>,
            exec_algorithm_id: Option<ExecAlgorithmId>,
            exec_spawn_id: Option<ClientOrderId>,
            position_id: Option<PositionId>,
        }

        let order_cell = self.orders.get(&client_order_id).cloned();
        let order_details = order_cell.as_ref().map(|cell| {
            let order = cell.borrow();
            OrderDetails {
                is_open: order.is_open(),
                instrument_id: order.instrument_id(),
                strategy_id: order.strategy_id(),
                account_id: order.account_id(),
                exec_algorithm_id: order.exec_algorithm_id(),
                exec_spawn_id: order.exec_spawn_id(),
                position_id: order.position_id(),
            }
        });

        if order_details
            .as_ref()
            .is_some_and(|details| details.is_open)
        {
            log::warn!("Order {client_order_id} found open when purging, skipping purge");
            return false;
        }

        if order_details.is_some() {
            self.orders.remove(&client_order_id);
        } else {
            log::warn!("Order {client_order_id} not found when purging");
        }

        let indexed_position_id = self.index.order_position.remove(&client_order_id);
        let indexed_strategy_id = self.index.order_strategy.remove(&client_order_id);
        self.index.order_client.remove(&client_order_id);
        self.index.client_order_ids.remove(&client_order_id);

        if let Some(details) = &order_details {
            if let Some(venue_orders) = self
                .index
                .venue_orders
                .get_mut(&details.instrument_id.venue)
            {
                venue_orders.remove(&client_order_id);
                if venue_orders.is_empty() {
                    self.index.venue_orders.remove(&details.instrument_id.venue);
                }
            }

            // As with the strategy buckets below, an absent bucket is left absent: recreating
            // it would suppress the `index.instrument_positions` integrity check.
            // As with the strategy buckets below, an absent bucket is left absent: recreating
            // it would suppress the `index.instrument_positions` integrity check.
            let instrument_orders_became_empty = self
                .index
                .instrument_orders
                .get_mut(&details.instrument_id)
                .is_some_and(|instrument_orders| {
                    instrument_orders.remove(&client_order_id);
                    instrument_orders.is_empty()
                });

            let has_instrument_positions = self
                .index
                .instrument_positions
                .get(&details.instrument_id)
                .is_some_and(|positions| !positions.is_empty());

            if instrument_orders_became_empty && !has_instrument_positions {
                self.index.instrument_orders.remove(&details.instrument_id);
            }

            if let Some(exec_algorithm_id) = details.exec_algorithm_id {
                let became_empty = self
                    .index
                    .exec_algorithm_orders
                    .get_mut(&exec_algorithm_id)
                    .is_some_and(|orders| {
                        orders.remove(&client_order_id);
                        orders.is_empty()
                    });

                if became_empty {
                    self.index.exec_algorithm_orders.remove(&exec_algorithm_id);
                    self.index.exec_algorithms.remove(&exec_algorithm_id);
                }
            }

            if let Some(account_id) = details.account_id
                && let Some(account_orders) = self.index.account_orders.get_mut(&account_id)
            {
                account_orders.remove(&client_order_id);
                if account_orders.is_empty() {
                    self.index.account_orders.remove(&account_id);
                }
            }

            if let Some(exec_spawn_id) = details.exec_spawn_id
                && let Some(spawn_orders) = self.index.exec_spawn_orders.get_mut(&exec_spawn_id)
            {
                spawn_orders.remove(&client_order_id);
                if spawn_orders.is_empty() {
                    self.index.exec_spawn_orders.remove(&exec_spawn_id);
                }
            }
        }

        let mut position_ids = AHashSet::new();
        if let Some(position_id) = indexed_position_id {
            position_ids.insert(position_id);
        }

        if let Some(position_id) = order_details
            .as_ref()
            .and_then(|details| details.position_id)
        {
            position_ids.insert(position_id);
        }

        let mut strategy_ids = AHashSet::new();
        if let Some(strategy_id) = indexed_strategy_id {
            strategy_ids.insert(strategy_id);
        }

        if let Some(details) = &order_details {
            strategy_ids.insert(details.strategy_id);
        }

        for position_id in position_ids {
            if self.positions.contains_key(&position_id) {
                if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
                    position_orders.remove(&client_order_id);
                }
                continue;
            }

            let has_other_orders =
                if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
                    position_orders.remove(&client_order_id);
                    !position_orders.is_empty()
                } else {
                    self.index
                        .order_position
                        .values()
                        .any(|candidate| *candidate == position_id)
                };

            if has_other_orders {
                continue;
            }

            self.index.position_orders.remove(&position_id);
            if let Some(strategy_id) = self.index.position_strategy.remove(&position_id) {
                strategy_ids.insert(strategy_id);
                if let Some(strategy_positions) =
                    self.index.strategy_positions.get_mut(&strategy_id)
                {
                    strategy_positions.remove(&position_id);
                    if strategy_positions.is_empty() {
                        self.index.strategy_positions.remove(&strategy_id);
                    }
                }
            }

            if let Some(details) = &order_details
                && let Some(venue_positions) = self
                    .index
                    .venue_positions
                    .get_mut(&details.instrument_id.venue)
            {
                venue_positions.remove(&position_id);
                if venue_positions.is_empty() {
                    self.index
                        .venue_positions
                        .remove(&details.instrument_id.venue);
                }
            }
        }

        for strategy_id in strategy_ids {
            // An absent reverse bucket is not an empty one: it means the index is already
            // inconsistent, possibly while another cached order still uses this strategy.
            // Retiring the registry entry here would both drop a live strategy from
            // `strategy_ids` and stop `check_integrity` reporting the missing bucket, so the
            // absent case is left exactly as found.
            let strategy_orders_became_empty = self
                .index
                .strategy_orders
                .get_mut(&strategy_id)
                .is_some_and(|strategy_orders| {
                    strategy_orders.remove(&client_order_id);
                    strategy_orders.is_empty()
                });

            let has_positions = self
                .index
                .strategy_positions
                .get(&strategy_id)
                .is_some_and(|strategy_positions| !strategy_positions.is_empty());

            if strategy_orders_became_empty && !has_positions {
                self.index.strategy_orders.remove(&strategy_id);
                self.index.strategies.remove(&strategy_id);
            }
        }

        self.index.exec_spawn_orders.remove(&client_order_id);

        self.index.orders.remove(&client_order_id);
        self.index.orders_active_local.remove(&client_order_id);
        self.index.orders_open.remove(&client_order_id);
        self.index.orders_closed.remove(&client_order_id);
        self.index.orders_emulated.remove(&client_order_id);
        self.index.orders_inflight.remove(&client_order_id);
        self.index.orders_pending_cancel.remove(&client_order_id);

        if order_details.is_some() {
            log::info!("Purged order {client_order_id}");
        }

        true
    }

    /// Purges the position with the `position_id` from the cache (if found).
    ///
    /// For safety, a position is prevented from being purged if it's open.
    pub fn purge_position(&mut self, position_id: PositionId) {
        // Keep the cell alive during index cleanup without cloning position history
        let position_cell = self.positions.get(&position_id).cloned();
        let position = position_cell.as_ref().map(|cell| cell.borrow());

        // Prevent purging open positions
        if let Some(ref pos) = position
            && pos.is_open()
        {
            log::warn!("Position {position_id} found open when purging, skipping purge");
            return;
        }

        // If position exists in cache, remove it and clean up position-specific indices
        if let Some(ref pos) = position {
            self.positions.remove(&position_id);

            // Remove from venue positions index
            if let Some(venue_positions) =
                self.index.venue_positions.get_mut(&pos.instrument_id.venue)
            {
                venue_positions.remove(&position_id);
                if venue_positions.is_empty() {
                    self.index.venue_positions.remove(&pos.instrument_id.venue);
                }
            }

            // Remove from instrument positions index
            let instrument_positions_became_empty = self
                .index
                .instrument_positions
                .get_mut(&pos.instrument_id)
                .is_some_and(|positions| {
                    positions.remove(&position_id);
                    positions.is_empty()
                });

            if instrument_positions_became_empty {
                self.index.instrument_positions.remove(&pos.instrument_id);
                let instrument_orders_empty = self
                    .index
                    .instrument_orders
                    .get(&pos.instrument_id)
                    .is_some_and(|orders| orders.is_empty());

                if instrument_orders_empty {
                    self.index.instrument_orders.remove(&pos.instrument_id);
                }
            }

            // Remove from strategy positions index
            let strategy_positions_became_empty = self
                .index
                .strategy_positions
                .get_mut(&pos.strategy_id)
                .is_some_and(|positions| {
                    positions.remove(&position_id);
                    positions.is_empty()
                });

            if strategy_positions_became_empty {
                self.index.strategy_positions.remove(&pos.strategy_id);
                let strategy_orders_empty = self
                    .index
                    .strategy_orders
                    .get(&pos.strategy_id)
                    .is_some_and(|orders| orders.is_empty());

                if strategy_orders_empty {
                    self.index.strategy_orders.remove(&pos.strategy_id);
                    self.index.strategies.remove(&pos.strategy_id);
                }
            }

            // Remove from account positions index
            if let Some(account_positions) = self.index.account_positions.get_mut(&pos.account_id) {
                account_positions.remove(&position_id);
                if account_positions.is_empty() {
                    self.index.account_positions.remove(&pos.account_id);
                }
            }

            // Remove position ID from orders that reference it
            for client_order_id in pos.client_order_ids() {
                self.index.order_position.remove(&client_order_id);
            }

            log::info!("Purged position {position_id}");
        } else {
            log::warn!("Position {position_id} not found when purging");
        }

        // Always clean up position indices (even if position not in cache)
        self.index.position_strategy.remove(&position_id);
        self.index.position_oms.remove(&position_id);
        self.index.position_orders.remove(&position_id);
        self.index.positions.remove(&position_id);
        self.index.positions_open.remove(&position_id);
        self.index.positions_closed.remove(&position_id);

        // Always clean up position snapshots (even if position not in cache)
        self.position_snapshots.remove(&position_id);
        self.bump_position_snapshot_revision(position_id);
    }

    /// Purges the instrument with the `instrument_id` from the cache (if found).
    ///
    /// The instrument record and its market data are removed:
    /// any synthetic with the same id, order book and own-order-book state, quote/trade
    /// histories, mark/index/funding price histories, instrument status and close, bars
    /// for any `BarType` referencing the instrument. Order and position indexes remain
    /// while their records are retained, preserving historical queries and realized PnL.
    ///
    /// For safety, an instrument is prevented from being purged while any associated
    /// order is non-terminal (anything not in `orders_closed`, including
    /// initialized, submitted, accepted, emulated, released, or inflight states) or
    /// any associated position is non-closed.
    ///
    /// Active subscriptions and other live data-engine state are not touched here;
    /// those belong to the data and execution engines.
    ///
    /// # Warning
    ///
    /// Intended for actors and strategies that have their own lifecycle logic for
    /// deciding when an instrument is no longer needed. Purging an instrument that any
    /// other actor, strategy, or engine still relies on may cause incorrect behavior
    /// (missing instrument lookups, lost market-data history). The caller is
    /// responsible for ensuring the instrument is no longer in use before purging.
    fn purge_instrument_inner(&mut self, instrument_id: InstrumentId, skip_order_guard: bool) {
        #[cfg(feature = "defi")]
        let defi_found = self.defi.pools.contains_key(&instrument_id)
            || self.defi.pool_profilers.contains_key(&instrument_id);
        #[cfg(not(feature = "defi"))]
        let defi_found = false;

        let found = self.instruments.contains_key(&instrument_id)
            || self.synthetics.contains_key(&instrument_id)
            || defi_found;

        if !found {
            log::warn!("Instrument {instrument_id} not found when purging");
            return;
        }

        if !skip_order_guard && let Some(orders) = self.index.instrument_orders.get(&instrument_id)
        {
            let has_non_terminal = orders
                .iter()
                .any(|client_order_id| !self.index.orders_closed.contains(client_order_id));

            if has_non_terminal {
                log::warn!(
                    "Instrument {instrument_id} has non-terminal orders when purging, skipping purge"
                );
                return;
            }
        }

        if let Some(positions) = self.index.instrument_positions.get(&instrument_id) {
            let has_non_closed = positions
                .iter()
                .any(|position_id| !self.index.positions_closed.contains(position_id));

            if has_non_closed {
                log::warn!(
                    "Instrument {instrument_id} has non-closed positions when purging, skipping purge"
                );
                return;
            }
        }

        self.instruments.remove(&instrument_id);
        self.synthetics.remove(&instrument_id);
        self.books.remove(&instrument_id);
        self.own_books.remove(&instrument_id);
        self.quotes.remove(&instrument_id);
        self.trades.remove(&instrument_id);
        self.mark_prices.remove(&instrument_id);
        self.index_prices.remove(&instrument_id);
        self.funding_rates.remove(&instrument_id);
        self.instrument_statuses.remove(&instrument_id);
        self.instrument_closes.remove(&instrument_id);
        self.greeks.remove(&instrument_id);
        self.option_greeks.remove(&instrument_id);

        self.bars
            .retain(|bar_type, _| bar_type.instrument_id() != instrument_id);

        #[cfg(feature = "defi")]
        {
            self.defi.pools.remove(&instrument_id);
            self.defi.pool_profilers.remove(&instrument_id);
        }

        log::info!("Purged instrument {instrument_id}");
    }

    /// Purges the instrument with the `instrument_id` from the cache.
    ///
    /// This refuses to purge when associated orders or positions remain in
    /// non-terminal state.
    pub fn purge_instrument(&mut self, instrument_id: InstrumentId) {
        self.purge_instrument_inner(instrument_id, false);
    }

    /// Purges the instrument with the `instrument_id` from the cache while skipping the
    /// non-terminal order guard.
    ///
    /// This still refuses to purge when any associated position is non-closed. Intended
    /// for actors which own an instrument-expiration lifecycle and have already invalidated
    /// any remaining order state externally, but may still observe order-terminal events
    /// arriving later than the cleanup decision. Order objects and their indexes remain
    /// available even after the instrument record is removed.
    pub fn purge_instrument_skip_order_guard(&mut self, instrument_id: InstrumentId) {
        self.purge_instrument_inner(instrument_id, true);
    }

    /// Purges all account state events which are outside the lookback window.
    ///
    /// Only events which are outside the lookback window will be purged.
    /// A value of 0 means purge all account state events.
    pub fn purge_account_events(&mut self, ts_now: UnixNanos, lookback_secs: u64) {
        log::debug!(
            "Purging account events{}",
            if lookback_secs > 0 {
                format!(" with lookback_secs={lookback_secs}")
            } else {
                String::new()
            }
        );

        for account_cell in self.accounts.values() {
            let mut account = account_cell.borrow_mut();
            let event_count = account.event_count();
            account.purge_account_events(ts_now, lookback_secs);
            let count_diff = event_count - account.event_count();
            if count_diff > 0 {
                log::info!(
                    "Purged {} event(s) from account {}",
                    count_diff,
                    account.id()
                );
            }
        }
    }

    /// Clears the caches index.
    pub fn clear_index(&mut self) {
        self.index.clear();
        log::debug!("Cleared index");
    }

    /// Resets the cache.
    ///
    /// All stateful fields are reset to their initial value. Instruments,
    /// currencies, and synthetics are retained when `drop_instruments_on_reset`
    /// is `false` so that repeated backtest runs can reuse the same dataset. External order claims
    /// and execution client account, route, and external client registrations are retained so
    /// registered strategy and client routing remain configured across resets.
    pub fn reset(&mut self) {
        log::debug!("Resetting cache");

        self.general.clear();
        self.books.clear();
        self.own_books.clear();
        self.quotes.clear();
        self.trades.clear();
        self.mark_xrates.clear();
        self.mark_prices.clear();
        self.index_prices.clear();
        self.funding_rates.clear();
        self.instrument_statuses.clear();
        self.instrument_closes.clear();
        self.bars.clear();
        self.accounts.clear();
        self.orders.clear();
        self.orders_dispatched.clear();
        self.order_lists.clear();
        self.positions.clear();
        self.position_snapshots.clear();
        self.position_snapshot_revisions.clear();
        self.greeks.clear();
        self.option_greeks.clear();
        self.yield_curves.clear();

        if self.config.drop_instruments_on_reset {
            self.currencies.clear();
            self.instruments.clear();
            self.synthetics.clear();
        }

        #[cfg(feature = "defi")]
        {
            self.defi.pools.clear();
            self.defi.pool_profilers.clear();
        }

        self.clear_index();

        log::info!("Reset cache");
    }

    /// Dispose of the cache which will close any underlying database adapter.
    ///
    /// If closing the database connection fails, an error is logged.
    pub fn dispose(&mut self) {
        self.reset();

        if let Some(database) = &mut self.database
            && let Err(e) = database.close()
        {
            log::error!("Failed to close database during dispose: {e}");
        }
    }

    /// Flushes the cache database, permanently removing the data the backing owns.
    ///
    /// Does nothing when no database is configured.
    ///
    /// # Errors
    ///
    /// Returns an error if flushing the database fails.
    pub fn flush_db(&mut self) -> anyhow::Result<()> {
        match &mut self.database {
            Some(database) => database.flush(),
            None => Ok(()),
        }
    }

    /// Adds a raw bytes `value` to the cache under the `key`.
    ///
    /// The cache stores only raw bytes; interpretation is the caller's responsibility.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the entry to the backing database fails.
    pub fn add(&mut self, key: &str, value: Bytes) -> anyhow::Result<()> {
        // Keys are opaque and may embed exchange symbols, which need not be ASCII
        check_valid_string_utf8(key, stringify!(key))?;
        check_predicate_false(value.is_empty(), stringify!(value))?;

        log::debug!("Adding general {key}");
        self.general.insert(key.to_string(), value.clone());

        if let Some(database) = &mut self.database {
            database.add(key.to_string(), value)?;
        }
        Ok(())
    }

    /// Adds an `OrderBook` to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the order book to the backing database fails.
    pub fn add_order_book(&mut self, book: OrderBook) -> anyhow::Result<()> {
        log::debug!("Adding `OrderBook` {}", book.instrument_id);

        if self.config.save_market_data
            && let Some(database) = &mut self.database
        {
            database.add_order_book(&book)?;
        }

        self.books.insert(book.instrument_id, book);
        Ok(())
    }

    /// Adds an `OwnOrderBook` to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the own order book fails.
    pub fn add_own_order_book(&mut self, own_book: OwnOrderBook) -> anyhow::Result<()> {
        log::debug!("Adding `OwnOrderBook` {}", own_book.instrument_id);

        self.own_books.insert(own_book.instrument_id, own_book);
        Ok(())
    }

    /// Adds the `mark_price` update to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the mark price to the backing database fails.
    pub fn add_mark_price(&mut self, mark_price: MarkPriceUpdate) -> anyhow::Result<()> {
        log::debug!("Adding `MarkPriceUpdate` for {}", mark_price.instrument_id);

        if self.config.save_market_data {
            // TODO: Placeholder and return Result for consistency
        }

        let mark_prices_deque = self
            .mark_prices
            .entry(mark_price.instrument_id)
            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
        mark_prices_deque.push_front(mark_price);
        Ok(())
    }

    /// Adds the `index_price` update to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the index price to the backing database fails.
    pub fn add_index_price(&mut self, index_price: IndexPriceUpdate) -> anyhow::Result<()> {
        log::debug!(
            "Adding `IndexPriceUpdate` for {}",
            index_price.instrument_id
        );

        if self.config.save_market_data {
            // TODO: Placeholder and return Result for consistency
        }

        let index_prices_deque = self
            .index_prices
            .entry(index_price.instrument_id)
            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
        index_prices_deque.push_front(index_price);
        Ok(())
    }

    /// Adds the `funding_rate` update to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the funding rate update to the backing database fails.
    pub fn add_funding_rate(&mut self, funding_rate: FundingRateUpdate) -> anyhow::Result<()> {
        log::debug!(
            "Adding `FundingRateUpdate` for {}",
            funding_rate.instrument_id
        );

        if self.config.save_market_data {
            // TODO: Placeholder and return Result for consistency
        }

        let funding_rates_deque = self
            .funding_rates
            .entry(funding_rate.instrument_id)
            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
        funding_rates_deque.push_front(funding_rate);
        Ok(())
    }

    /// Adds the given `funding rates` to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the trade ticks to the backing database fails.
    pub fn add_funding_rates(&mut self, funding_rates: &[FundingRateUpdate]) -> anyhow::Result<()> {
        check_slice_not_empty(funding_rates, stringify!(funding_rates))?;

        let instrument_id = funding_rates[0].instrument_id;
        log::debug!(
            "Adding `FundingRateUpdate`[{}] {instrument_id}",
            funding_rates.len()
        );

        if self.config.save_market_data
            && let Some(database) = &mut self.database
        {
            for funding_rate in funding_rates {
                database.add_funding_rate(funding_rate)?;
            }
        }

        let funding_rate_deque = self
            .funding_rates
            .entry(instrument_id)
            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));

        for funding_rate in funding_rates {
            funding_rate_deque.push_front(*funding_rate);
        }
        Ok(())
    }

    /// Adds the `instrument_status` update to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the instrument status to the backing database fails.
    pub fn add_instrument_status(&mut self, status: InstrumentStatus) -> anyhow::Result<()> {
        log::debug!("Adding `InstrumentStatus` for {}", status.instrument_id);

        if self.config.save_market_data {
            // TODO: Placeholder and return Result for consistency
        }

        let statuses_deque = self
            .instrument_statuses
            .entry(status.instrument_id)
            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
        statuses_deque.push_front(status);
        Ok(())
    }

    /// Adds an instrument close to the cache.
    ///
    /// A close already cached for the same instrument is overwritten. With a backing database, the
    /// replacement is queued for persistence before the in-memory value is updated. A later
    /// database error is logged by the adapter and does not roll back the cached value.
    ///
    /// # Errors
    ///
    /// Returns an error if the close cannot be queued for persistence.
    pub fn add_instrument_close(&mut self, close: InstrumentClose) -> anyhow::Result<()> {
        log::debug!("Adding `InstrumentClose` for {}", close.instrument_id);

        if let Some(database) = &self.database {
            database.add_instrument_close(&close)?;
        }

        self.instrument_closes.insert(close.instrument_id, close);
        Ok(())
    }

    /// Adds the `quote` tick to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the quote tick to the backing database fails.
    pub fn add_quote(&mut self, quote: QuoteTick) -> anyhow::Result<()> {
        log::debug!("Adding `QuoteTick` {}", quote.instrument_id);

        if self.config.save_market_data
            && let Some(database) = &mut self.database
        {
            database.add_quote(&quote)?;
        }

        let quotes_deque = self
            .quotes
            .entry(quote.instrument_id)
            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
        quotes_deque.push_front(quote);
        Ok(())
    }

    /// Adds the `quotes` to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the quote ticks to the backing database fails.
    pub fn add_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
        check_slice_not_empty(quotes, stringify!(quotes))?;

        let instrument_id = quotes[0].instrument_id;
        log::debug!("Adding `QuoteTick`[{}] {instrument_id}", quotes.len());

        if self.config.save_market_data
            && let Some(database) = &mut self.database
        {
            for quote in quotes {
                database.add_quote(quote)?;
            }
        }

        let quotes_deque = self
            .quotes
            .entry(instrument_id)
            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));

        for quote in quotes {
            quotes_deque.push_front(*quote);
        }
        Ok(())
    }

    /// Adds the `trade` tick to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the trade tick to the backing database fails.
    pub fn add_trade(&mut self, trade: TradeTick) -> anyhow::Result<()> {
        log::debug!("Adding `TradeTick` {}", trade.instrument_id);

        if self.config.save_market_data
            && let Some(database) = &mut self.database
        {
            database.add_trade(&trade)?;
        }

        let trades_deque = self
            .trades
            .entry(trade.instrument_id)
            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
        trades_deque.push_front(trade);
        Ok(())
    }

    /// Adds the give `trades` to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the trade ticks to the backing database fails.
    pub fn add_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
        check_slice_not_empty(trades, stringify!(trades))?;

        let instrument_id = trades[0].instrument_id;
        log::debug!("Adding `TradeTick`[{}] {instrument_id}", trades.len());

        if self.config.save_market_data
            && let Some(database) = &mut self.database
        {
            for trade in trades {
                database.add_trade(trade)?;
            }
        }

        let trades_deque = self
            .trades
            .entry(instrument_id)
            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));

        for trade in trades {
            trades_deque.push_front(*trade);
        }
        Ok(())
    }

    /// Adds the `bar` to the cache, keeping the per-`bar_type` series newest-first.
    ///
    /// A newer bar is pushed, an older `ts_event` is skipped, and an equal
    /// `ts_event` replaces the front bar for time bars.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the bar to the backing database fails.
    pub fn add_bar(&mut self, bar: Bar) -> anyhow::Result<()> {
        log::debug!("Adding `Bar` {}", bar.bar_type);

        if self.config.save_market_data
            && let Some(database) = &mut self.database
        {
            database.add_bar(&bar)?;
        }

        let bars = self
            .bars
            .entry(bar.bar_type)
            .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
        insert_bar(bars, bar);
        Ok(())
    }

    /// Adds the `bars` to the cache, each following [`Cache::add_bar`].
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the bars to the backing database fails.
    pub fn add_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
        check_slice_not_empty(bars, stringify!(bars))?;

        let bar_type = bars[0].bar_type;
        log::debug!("Adding `Bar`[{}] {bar_type}", bars.len());

        if self.config.save_market_data
            && let Some(database) = &mut self.database
        {
            for bar in bars {
                database.add_bar(bar)?;
            }
        }

        let bars_deque = self
            .bars
            .entry(bar_type)
            .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));

        for bar in bars {
            insert_bar(bars_deque, *bar);
        }
        Ok(())
    }

    /// Adds the historical `bar` at its ordered position in the series.
    ///
    /// Request-generated bars are added this way, since the cache is their only
    /// delivery path: bars older than the front are kept rather than skipped.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the bar to the backing database fails.
    pub fn add_bar_historical(&mut self, bar: Bar) -> anyhow::Result<()> {
        log::debug!("Adding historical `Bar` {}", bar.bar_type);

        if self.config.save_market_data
            && let Some(database) = &mut self.database
        {
            database.add_bar(&bar)?;
        }

        let bars = self
            .bars
            .entry(bar.bar_type)
            .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
        insert_bar_historical(bars, bar);
        Ok(())
    }

    /// Adds the `greeks` data to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the greeks data to the backing database fails.
    pub fn add_greeks(&mut self, greeks: GreeksData) -> anyhow::Result<()> {
        log::debug!("Adding `GreeksData` {}", greeks.instrument_id);

        if self.config.save_market_data
            && let Some(_database) = &mut self.database
        {
            // TODO: Implement database.add_greeks(&greeks) when database adapter is updated
        }

        self.greeks.insert(greeks.instrument_id, greeks);
        Ok(())
    }

    /// Gets the greeks data for the `instrument_id`.
    pub fn greeks(&self, instrument_id: &InstrumentId) -> Option<GreeksData> {
        self.greeks.get(instrument_id).cloned()
    }

    /// Adds exchange-provided option greeks to the cache.
    pub fn add_option_greeks(&mut self, greeks: OptionGreeks) {
        log::debug!("Adding `OptionGreeks` {}", greeks.instrument_id);
        self.option_greeks.insert(greeks.instrument_id, greeks);
    }

    /// Gets a reference to the exchange-provided option greeks for the `instrument_id`.
    #[must_use]
    pub fn option_greeks(&self, instrument_id: &InstrumentId) -> Option<&OptionGreeks> {
        self.option_greeks.get(instrument_id)
    }

    /// Adds the `yield_curve` data to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The yield curve fails [`YieldCurve::validate`].
    /// - Persisting the yield curve data to the backing database fails.
    pub fn add_yield_curve(&mut self, yield_curve: YieldCurve) -> anyhow::Result<()> {
        yield_curve.validate()?;

        log::debug!("Adding `YieldCurve` {}", yield_curve.curve_name);

        if self.config.save_market_data
            && let Some(_database) = &mut self.database
        {
            // TODO: Implement database.add_yield_curve(&yield_curve) when database adapter is updated
        }

        self.yield_curves
            .insert(yield_curve.curve_name.clone(), yield_curve);
        Ok(())
    }

    /// Gets the yield curve for the `key`.
    pub fn yield_curve(&self, key: &str) -> Option<Box<dyn Fn(f64) -> f64>> {
        self.yield_curves.get(key).map(|curve| {
            let curve_clone = curve.clone();
            Box::new(move |expiry_in_years: f64| curve_clone.get_rate(expiry_in_years))
                as Box<dyn Fn(f64) -> f64>
        })
    }

    /// Adds the `currency` to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the currency to the backing database fails.
    pub fn add_currency(&mut self, currency: Currency) -> anyhow::Result<()> {
        if self.currencies.contains_key(&currency.code) {
            return Ok(());
        }
        log::debug!("Adding `Currency` {}", currency.code);

        if let Some(database) = &mut self.database {
            database.add_currency(&currency)?;
        }

        self.currencies.insert(currency.code, currency);
        Ok(())
    }

    /// Adds the `instrument` to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the instrument to the backing database fails.
    pub fn add_instrument(&mut self, instrument: InstrumentAny) -> anyhow::Result<()> {
        log::debug!("Adding `Instrument` {}", instrument.id());

        // Ensure currencies exist in cache - safe to call repeatedly as add_currency is idempotent
        if let Some(base_currency) = instrument.base_currency() {
            self.add_currency(base_currency)?;
        }
        self.add_currency(instrument.quote_currency())?;
        self.add_currency(instrument.settlement_currency())?;

        if let Some(database) = &mut self.database {
            database.add_instrument(&instrument)?;
        }

        self.instruments.insert(instrument.id(), instrument);
        Ok(())
    }

    /// Adds the `synthetic` instrument to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the synthetic instrument to the backing database fails.
    pub fn add_synthetic(&mut self, synthetic: SyntheticInstrument) -> anyhow::Result<()> {
        log::debug!("Adding `SyntheticInstrument` {}", synthetic.id);

        if let Some(database) = &mut self.database {
            database.add_synthetic(&synthetic)?;
        }

        self.synthetics.insert(synthetic.id, synthetic);
        Ok(())
    }

    /// The currencies an account state references: its balances, its margins and its base currency.
    fn state_currencies(state: &AccountState) -> impl Iterator<Item = Currency> + '_ {
        state
            .balances
            .iter()
            .map(|balance| balance.currency)
            .chain(state.margins.iter().map(|margin| margin.currency))
            .chain(state.base_currency)
    }

    /// Persists the currencies an account references, so a restart can restore them.
    ///
    /// An account can hold collateral in a currency no instrument carries, and a `Currency`
    /// persists as its bare code, so without this the account cannot be decoded in a process
    /// where that code was never registered. Mirrors what [`Self::add_instrument`] does for an
    /// instrument's base, quote and settlement currencies.
    fn add_account_currencies(&mut self, account: &AccountAny) -> anyhow::Result<()> {
        let Some(state) = account.last_event() else {
            return Ok(());
        };

        for currency in Self::state_currencies(&state) {
            self.add_currency(currency)?;
        }
        Ok(())
    }

    /// Adds the `account` to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the account or its currencies to the backing database fails.
    pub fn add_account(&mut self, account: AccountAny) -> anyhow::Result<()> {
        log::debug!("Adding `Account` {}", account.id());

        self.add_account_currencies(&account)?;

        if let Some(database) = &mut self.database {
            database.add_account(&account)?;
        }

        let account_id = account.id();
        self.accounts.insert(account_id, SharedCell::new(account));
        self.index.add_venue_account(account_id);
        Ok(())
    }

    /// Indexes the `client_order_id` with the `venue_order_id`.
    ///
    /// The `overwrite` parameter determines whether to overwrite any existing cached identifier.
    ///
    /// # Errors
    ///
    /// Returns an error if the client already has a different venue order ID and `overwrite` is
    /// false, or if the venue order ID is owned by a different client order.
    pub fn add_venue_order_id(
        &mut self,
        client_order_id: &ClientOrderId,
        venue_order_id: &VenueOrderId,
        overwrite: bool,
    ) -> anyhow::Result<()> {
        self.validate_venue_order_id_claim(client_order_id, venue_order_id, overwrite)?;

        self.index
            .client_order_ids
            .insert(*client_order_id, *venue_order_id);
        self.index
            .venue_order_ids
            .insert(*venue_order_id, *client_order_id);

        Ok(())
    }

    /// Indexes the reverse alias `venue_order_id` to `client_order_id` for routing.
    ///
    /// Unlike [`Cache::add_venue_order_id`], an existing forward mapping for the client
    /// order is never moved, so superseded venue order ID generations from mass status
    /// reports register idempotently. The forward mapping's authority stays with order
    /// event application via [`Cache::update_order`].
    ///
    /// # Errors
    ///
    /// Returns an error if the venue order ID is owned by a different client order.
    pub fn index_venue_order_id(
        &mut self,
        client_order_id: &ClientOrderId,
        venue_order_id: &VenueOrderId,
    ) -> anyhow::Result<()> {
        self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;

        self.index
            .venue_order_ids
            .insert(*venue_order_id, *client_order_id);
        self.index
            .client_order_ids
            .entry(*client_order_id)
            .or_insert(*venue_order_id);

        Ok(())
    }

    fn validate_venue_order_id_claim(
        &self,
        client_order_id: &ClientOrderId,
        venue_order_id: &VenueOrderId,
        overwrite: bool,
    ) -> anyhow::Result<()> {
        self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;

        if let Some(existing_venue_order_id) = self.index.client_order_ids.get(client_order_id)
            && !overwrite
            && existing_venue_order_id != venue_order_id
        {
            anyhow::bail!(
                "Existing {existing_venue_order_id} for {client_order_id}
                    did not match the given {venue_order_id}.
                    If you are writing a test then try a different `venue_order_id`,
                    otherwise this is probably a bug."
            );
        }

        Ok(())
    }

    fn validate_venue_order_id_ownership(
        &self,
        client_order_id: &ClientOrderId,
        venue_order_id: &VenueOrderId,
    ) -> anyhow::Result<()> {
        if let Some(existing_client_order_id) = self.index.venue_order_ids.get(venue_order_id)
            && existing_client_order_id != client_order_id
        {
            return Err(VenueOrderIdOwnershipError {
                venue_order_id: *venue_order_id,
                existing_client_order_id: *existing_client_order_id,
                claimant_client_order_id: *client_order_id,
            }
            .into());
        }

        Ok(())
    }

    /// Adds the `order` to the cache indexed with any given identifiers.
    ///
    /// # Parameters
    ///
    /// `override_existing`: If the added order should 'override' any existing order and replace
    /// it in the cache. This is currently used for emulated orders which are
    /// being released and transformed into another type.
    ///
    /// # Errors
    ///
    /// Returns an error if not `replace_existing` and the `order.client_order_id` is already contained in the cache,
    /// or if persisting the order to the backing database fails. The order and every index are
    /// committed to memory before persistence is attempted, so a persistence error leaves the
    /// cache internally consistent.
    pub fn add_order(
        &mut self,
        order: OrderAny,
        position_id: Option<PositionId>,
        client_id: Option<ClientId>,
        replace_existing: bool,
    ) -> anyhow::Result<()> {
        let instrument_id = order.instrument_id();
        let venue = instrument_id.venue;
        let client_order_id = order.client_order_id();
        let strategy_id = order.strategy_id();
        let exec_algorithm_id = order.exec_algorithm_id();
        let exec_spawn_id = order.exec_spawn_id();

        if !replace_existing {
            check_key_not_in_map(
                &client_order_id,
                &self.orders,
                stringify!(client_order_id),
                stringify!(orders),
            )?;
        }

        log::debug!("Adding {order:?}");

        self.index.orders.insert(client_order_id);

        if order.is_active_local() {
            self.index.orders_active_local.insert(client_order_id);
        }
        self.index
            .order_strategy
            .insert(client_order_id, strategy_id);
        self.index.strategies.insert(strategy_id);

        // Update venue -> orders index
        self.index
            .venue_orders
            .entry(venue)
            .or_default()
            .insert(client_order_id);

        // Update instrument -> orders index
        self.index
            .instrument_orders
            .entry(instrument_id)
            .or_default()
            .insert(client_order_id);

        // Update strategy -> orders index
        self.index
            .strategy_orders
            .entry(strategy_id)
            .or_default()
            .insert(client_order_id);

        // Update account -> orders index (if account_id known at creation)
        if let Some(account_id) = order.account_id() {
            self.index
                .account_orders
                .entry(account_id)
                .or_default()
                .insert(client_order_id);
        }

        // Update exec_algorithm -> orders index
        if let Some(exec_algorithm_id) = exec_algorithm_id {
            self.index.exec_algorithms.insert(exec_algorithm_id);

            self.index
                .exec_algorithm_orders
                .entry(exec_algorithm_id)
                .or_default()
                .insert(client_order_id);
        }

        // Update exec_spawn -> orders index
        if let Some(exec_spawn_id) = exec_spawn_id {
            self.index
                .exec_spawn_orders
                .entry(exec_spawn_id)
                .or_default()
                .insert(client_order_id);
        }

        // Update emulation index
        if order.emulation_trigger().is_some() {
            self.index.orders_emulated.insert(client_order_id);
        }

        // Index position ID if provided
        if let Some(position_id) = position_id {
            self.index_position_id_in_memory(&position_id, &venue, &client_order_id, &strategy_id);
        }

        // Index client ID if provided
        if let Some(client_id) = client_id {
            self.index.order_client.insert(client_order_id, client_id);
            log::debug!("Indexed {client_id:?}");
        }

        // Reuse the existing cell on replace so the canonical entry stays in place
        // rather than orphaning a stale cell.
        let order_cell = if let Some(order_cell) = self.orders.get(&client_order_id) {
            *order_cell.borrow_mut() = order;
            order_cell.clone()
        } else {
            let order_cell = SharedCell::new(order);
            self.orders.insert(client_order_id, order_cell.clone());
            order_cell
        };

        if let Some(position_id) = position_id {
            self.persist_position_id(&position_id, &client_order_id)?;
        }

        if let Some(database) = &mut self.database {
            database.add_order(&order_cell.borrow(), client_id)?;
            // TODO: Implement
            // if self.config.snapshot_orders {
            //     database.snapshot_order_state(order)?;
            // }
        }

        Ok(())
    }

    /// Claims the execution-client origin for one or more cached orders.
    ///
    /// Claims are write-once: an unclaimed order is assigned to `client_id`, a matching existing
    /// claim is idempotent, and a conflicting claim is rejected. The complete batch is validated
    /// and its persistence command is successfully enqueued before any in-memory index is
    /// changed.
    ///
    /// # Errors
    ///
    /// Returns an error if an order is not cached, an order is already claimed by another client,
    /// the same order has conflicting claims in the batch, or persistence cannot be enqueued.
    pub fn claim_order_clients(
        &mut self,
        claims: &[(ClientOrderId, ClientId)],
    ) -> anyhow::Result<()> {
        let mut requested = AHashMap::with_capacity(claims.len());
        let mut ordered_claims = Vec::with_capacity(claims.len());

        for (client_order_id, client_id) in claims {
            if let Some(existing_client_id) = requested.get(client_order_id) {
                if existing_client_id != client_id {
                    anyhow::bail!(
                        "Conflicting execution client claims for {client_order_id}: \
                         {existing_client_id} and {client_id}"
                    );
                }
                continue;
            }

            requested.insert(*client_order_id, *client_id);
            ordered_claims.push((*client_order_id, *client_id));
        }

        let mut pending_claims = Vec::with_capacity(ordered_claims.len());
        for (client_order_id, client_id) in ordered_claims {
            if !self.orders.contains_key(&client_order_id) {
                return Err(OrderLookupError::not_found(client_order_id).into());
            }

            match self.index.order_client.get(&client_order_id) {
                Some(existing_client_id) if *existing_client_id == client_id => {}
                Some(existing_client_id) => {
                    anyhow::bail!(
                        "Order {client_order_id} is already claimed by execution client \
                         {existing_client_id} and cannot be claimed by {client_id}"
                    );
                }
                None => pending_claims.push((client_order_id, client_id)),
            }
        }

        if pending_claims.is_empty() {
            return Ok(());
        }

        if let Some(database) = &self.database {
            database.index_order_clients(&pending_claims)?;
        }

        for (client_order_id, client_id) in pending_claims {
            self.index.order_client.insert(client_order_id, client_id);
            log::debug!("Claimed {client_order_id} for execution client {client_id}");
        }

        Ok(())
    }

    /// Adds the `order_list` to the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if the order list ID is already contained in the cache.
    pub fn add_order_list(&mut self, order_list: OrderList) -> anyhow::Result<()> {
        let order_list_id = order_list.id;
        check_key_not_in_map(
            &order_list_id,
            &self.order_lists,
            stringify!(order_list_id),
            stringify!(order_lists),
        )?;

        log::debug!("Adding {order_list}");
        self.order_lists.insert(order_list_id, order_list);
        Ok(())
    }

    /// Indexes the `position_id` with the other given IDs.
    ///
    /// A cached `EXTERNAL` position retains its ownership when an order from another strategy
    /// is linked to it. Otherwise, the supplied `strategy_id` applies.
    ///
    /// # Errors
    ///
    /// Returns an error if indexing position ID in the backing database fails. The complete index
    /// operation is committed to memory before persistence is attempted, so a persistence error
    /// leaves the cache internally consistent.
    pub fn add_position_id(
        &mut self,
        position_id: &PositionId,
        venue: &Venue,
        client_order_id: &ClientOrderId,
        strategy_id: &StrategyId,
    ) -> anyhow::Result<()> {
        self.index_position_id_in_memory(position_id, venue, client_order_id, strategy_id);
        self.persist_position_id(position_id, client_order_id)
    }

    fn index_position_id_in_memory(
        &mut self,
        position_id: &PositionId,
        venue: &Venue,
        client_order_id: &ClientOrderId,
        strategy_id: &StrategyId,
    ) {
        self.index
            .order_position
            .insert(*client_order_id, *position_id);
        self.index_position(position_id, venue, strategy_id);
        self.index
            .position_orders
            .entry(*position_id)
            .or_default()
            .insert(*client_order_id);
    }

    fn persist_position_id(
        &mut self,
        position_id: &PositionId,
        client_order_id: &ClientOrderId,
    ) -> anyhow::Result<()> {
        if let Some(database) = &mut self.database {
            database.index_order_position(*client_order_id, *position_id)?;
        }

        Ok(())
    }

    fn index_position(
        &mut self,
        position_id: &PositionId,
        venue: &Venue,
        strategy_id: &StrategyId,
    ) {
        let strategy_id = self
            .positions
            .get(position_id)
            .map(|position| position.borrow().strategy_id)
            .filter(StrategyId::is_external)
            .unwrap_or(*strategy_id);

        // Index: PositionId -> StrategyId
        self.index
            .position_strategy
            .insert(*position_id, strategy_id);

        // Every position has a reverse-order bucket, including orderless positions.
        self.index.position_orders.entry(*position_id).or_default();

        // Index: StrategyId -> set[PositionId]
        self.index
            .strategy_positions
            .entry(strategy_id)
            .or_default()
            .insert(*position_id);

        // Index: Venue -> set[PositionId]
        self.index
            .venue_positions
            .entry(*venue)
            .or_default()
            .insert(*position_id);
    }

    // Propagates parent OTO `position_id` to contingent children that are missing one.
    //
    // Recovers from a partial-write window during fill handling: the fill-time path in the
    // execution engine assigns `position_id` to each contingent child in a non-atomic loop
    // (`set_position_id` then `add_position_id`), so a crash mid-loop can leave the database
    // with the parent updated and some children un-updated. This pass re-applies any missing
    // assignments after load.
    fn assign_position_ids_to_contingencies(&mut self) {
        let mut assignments: Vec<(PositionId, ClientOrderId)> = Vec::new();

        for parent_order_cell in self.orders.values() {
            let parent = parent_order_cell.borrow();
            if parent.contingency_type() != Some(ContingencyType::Oto) {
                continue;
            }
            let Some(parent_position_id) = parent.position_id() else {
                continue;
            };
            let Some(linked_order_ids) = parent.linked_order_ids() else {
                continue;
            };

            for client_order_id in linked_order_ids {
                match self.orders.get(client_order_id) {
                    None => {
                        log::error!("Contingency order {client_order_id} not found");
                    }
                    Some(contingent_order_cell) => {
                        if contingent_order_cell.borrow().position_id().is_none() {
                            assignments.push((parent_position_id, *client_order_id));
                        }
                    }
                }
            }
        }

        for (position_id, client_order_id) in assignments {
            let Some((venue, strategy_id)) = self.orders.get(&client_order_id).map(|order_cell| {
                let mut contingent = order_cell.borrow_mut();
                contingent.set_position_id(Some(position_id));
                (contingent.instrument_id().venue, contingent.strategy_id())
            }) else {
                continue;
            };

            // Re-indexing through `add_position_id` also replays the database write, making the
            // recovered assignment durable across another restart.
            if let Err(e) =
                self.add_position_id(&position_id, &venue, &client_order_id, &strategy_id)
            {
                log::error!("Failed to re-index {client_order_id} -> {position_id}: {e}");
            }
        }
    }

    /// Adds the `position` to the cache.
    ///
    /// For [`OmsType::Unspecified`], IDs matching `{instrument_id}-{strategy_id}` resolve to
    /// [`OmsType::Netting`]. Other IDs retain the supplied OMS type.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the position to the backing database fails. After
    /// serialization succeeds, the complete operation is committed to memory before persistence
    /// is attempted, so a persistence error leaves the cache internally consistent.
    pub fn add_position(&mut self, position: &Position, oms_type: OmsType) -> anyhow::Result<()> {
        self.add_position_inner(position.clone(), oms_type, true, false)
    }

    /// Adds a position whose opening fill intentionally has no backing order.
    ///
    /// Resolves unspecified OMS types as [`Self::add_position`] does.
    ///
    /// # Errors
    ///
    /// Returns an error if persisting the position to the backing database fails. After
    /// serialization succeeds, the complete operation is committed to memory before persistence
    /// is attempted, so a persistence error leaves the cache internally consistent.
    pub fn add_position_without_order(
        &mut self,
        position: &Position,
        oms_type: OmsType,
    ) -> anyhow::Result<()> {
        self.add_position_inner(position.clone(), oms_type, false, false)
    }

    /// Replaces the cached position holding `position.id`, optionally moving the prior cycle's
    /// durable replay state into it.
    ///
    /// Pass `index_order` false when the opening fill intentionally has no backing order, matching
    /// [`Self::add_position_without_order`]. The carry runs under the same borrow as the swap, so a
    /// validation failure cannot leave the prior position stripped of its replay history.
    ///
    /// Resolves unspecified OMS types as [`Self::add_position`] does.
    ///
    /// # Errors
    ///
    /// Returns an error if validating or persisting the position fails.
    pub fn replace_position(
        &mut self,
        position: &Position,
        oms_type: OmsType,
        index_order: bool,
        carry_replay_state: bool,
    ) -> anyhow::Result<()> {
        self.add_position_inner(position.clone(), oms_type, index_order, carry_replay_state)
    }

    fn add_position_inner(
        &mut self,
        mut position: Position,
        oms_type: OmsType,
        index_order: bool,
        carry_replay_state: bool,
    ) -> anyhow::Result<()> {
        let oms_type = resolve_position_oms(&position, oms_type);

        // Validate and serialize the OMS entry up front: both are construction failures, and
        // committing the position before they run would leave the cache mutated by one.
        // The key embeds the position ID, which embeds the instrument ID and venue
        // symbol, and exchange symbols may contain non-ASCII characters.
        let key = position_oms_key(position.id);
        check_valid_string_utf8(&key, stringify!(key))?;
        let value = Bytes::from(serde_json::to_vec(&oms_type)?);
        check_predicate_false(value.is_empty(), stringify!(value))?;

        let position_id = position.id;
        let strategy_id = position.strategy_id;
        let instrument_id = position.instrument_id;
        let account_id = position.account_id;
        let opening_order_id = position.opening_order_id;

        log::debug!("Adding {position}");

        // Reuse the existing cell on a NETTING reopen so the prior value is replaced in place,
        // which also lets the carried replay state move out of it instead of being copied. The
        // transfer and the swap share one borrow, so a failure above cannot strip the prior.
        let position_cell = if let Some(position_cell) = self.positions.get(&position_id).cloned() {
            let mut prior = position_cell.borrow_mut();
            if carry_replay_state {
                position.transfer_replay_state_from(&mut prior);
            }
            *prior = position;
            drop(prior);
            position_cell
        } else {
            let position_cell = SharedCell::new(position);
            self.positions.insert(position_id, position_cell.clone());
            position_cell
        };

        self.index.position_oms.insert(position_id, oms_type);
        self.index.positions.insert(position_id);
        self.index.positions_open.insert(position_id);
        self.index.positions_closed.remove(&position_id); // Cleanup for NETTING reopen
        self.index.strategies.insert(strategy_id);
        self.index.strategy_orders.entry(strategy_id).or_default();

        if index_order {
            self.index_position_id_in_memory(
                &position_id,
                &instrument_id.venue,
                &opening_order_id,
                &strategy_id,
            );
        } else {
            self.index_position(&position_id, &instrument_id.venue, &strategy_id);
        }

        // Index: InstrumentId -> AHashSet
        let instrument_positions = self
            .index
            .instrument_positions
            .entry(instrument_id)
            .or_default();
        instrument_positions.insert(position_id);
        self.index
            .instrument_orders
            .entry(instrument_id)
            .or_default();

        // Index: AccountId -> AHashSet<PositionId>
        self.index
            .account_positions
            .entry(account_id)
            .or_default()
            .insert(position_id);

        log::debug!("Adding general {key}");
        self.general.insert(key.clone(), value.clone());

        if index_order {
            self.persist_position_id(&position_id, &opening_order_id)?;
        }

        if let Some(database) = &mut self.database {
            database.add_position(&position_cell.borrow())?;
            // TODO: Implement position snapshots
            // if self.snapshot_positions {
            //     database.snapshot_position_state(
            //         position,
            //         position.ts_last,
            //         self.calculate_unrealized_pnl(&position),
            //     )?;
            // }
            database.add(key, value)?;
        }

        Ok(())
    }

    /// Updates the `account` in the cache.
    ///
    /// Reuses the existing cell when present so any held [`AccountRef`] handles continue to point
    /// at the canonical entry; only inserts a new cell when the account is unknown.
    ///
    /// # Errors
    ///
    /// Returns an error if updating the account in the database fails.
    pub fn update_account(&mut self, account: &AccountAny) -> anyhow::Result<()> {
        let account_id = account.id();
        match self.accounts.get(&account_id) {
            Some(account_cell) => *account_cell.borrow_mut() = account.clone(),
            None => {
                self.accounts
                    .insert(account_id, SharedCell::new(account.clone()));
            }
        }

        self.add_account_currencies(account)?;

        if let Some(database) = &mut self.database {
            database.update_account(account)?;
        }
        Ok(())
    }

    /// Returns an owned `account`, removing its cache entry when ownership is exclusive.
    ///
    /// This supports hot paths which need owned account mutation without
    /// cloning the account event history. The cache is the sole owner of the
    /// account cell (the field is private and accessors only hand out
    /// lifetime-scoped [`AccountRef`] borrows). When the cache is the sole owner, the value is
    /// moved out of its cell rather than cloned.
    ///
    /// If another strong handle exists, the canonical entry remains cached and this returns
    /// `None`.
    #[must_use]
    pub fn take_account(&mut self, account_id: &AccountId) -> Option<AccountAny> {
        let cell = self.accounts.remove(account_id)?;
        let rc: Rc<RefCell<AccountAny>> = cell.into();

        match Rc::try_unwrap(rc) {
            Ok(cell) => Some(cell.into_inner()),
            Err(rc) => {
                log::error!(
                    "Cannot move account {account_id} out of cache: account cell has an outstanding owner"
                );
                self.accounts.insert(*account_id, rc.into());
                None
            }
        }
    }

    /// Caches the `account` in memory without updating the database.
    pub fn cache_account_owned(&mut self, account: AccountAny) {
        let account_id = account.id();
        self.index.add_venue_account(account_id);
        match self.accounts.get(&account_id) {
            Some(account_cell) => *account_cell.borrow_mut() = account,
            None => {
                self.accounts.insert(account_id, SharedCell::new(account));
            }
        }
    }

    /// Updates the `account` in the cache, taking ownership of the updated account.
    ///
    /// # Errors
    ///
    /// Returns an error if updating the account in the database fails.
    pub fn update_account_owned(&mut self, account: AccountAny) -> anyhow::Result<()> {
        let account_id = account.id();
        // The in-memory account is updated before anything can fail, so a database failure while
        // saving its currencies never drops an account the caller has taken out of the cache.
        let currencies: Vec<Currency> = account
            .last_event()
            .map(|state| Self::state_currencies(&state).collect())
            .unwrap_or_default();
        self.cache_account_owned(account);

        for currency in currencies {
            self.add_currency(currency)?;
        }

        if let Some(database) = &mut self.database {
            let Some(account_cell) = self.accounts.get(&account_id) else {
                anyhow::bail!("Account {account_id} not found after cache update");
            };
            database.update_account(&account_cell.borrow())?;
        }
        Ok(())
    }

    /// Applies an account state event to the cached account.
    ///
    /// Mutates the cached account in place to avoid cloning the account event
    /// history on the hot path; long-running sessions accumulate many events
    /// per account, so a snapshot-clone here would be O(history) per update.
    ///
    /// # Errors
    ///
    /// Returns an error if applying or persisting the account state fails.
    pub fn update_account_state(&mut self, event: &AccountState) -> anyhow::Result<()> {
        let Some(cell) = self.accounts.get(&event.account_id) else {
            return self.add_account(AccountAny::from_events(std::slice::from_ref(event))?);
        };

        // The event is applied before anything can fail, so a database failure while saving its
        // currencies leaves the in-memory account current.
        cell.borrow_mut().apply(event.clone())?;

        // The event carries every currency the account references after it, so the account
        // itself is not cloned on this path.
        for currency in Self::state_currencies(event) {
            self.add_currency(currency)?;
        }

        let Some(cell) = self.accounts.get(&event.account_id) else {
            anyhow::bail!("Account {} not found after apply", event.account_id);
        };

        if let Some(database) = &mut self.database {
            database.update_account(&cell.borrow())?;
        }
        Ok(())
    }

    /// Replaces the cached `order` from a non-event snapshot.
    ///
    /// Prefer [`Self::update_order`] for lifecycle state changes. Use this only for order state
    /// that is not represented by [`OrderEventAny`].
    ///
    /// # Errors
    ///
    /// Returns an error if validation or persistence fails. After validation succeeds, the
    /// canonical order is committed to memory before its indexes and database are refreshed, so a
    /// persistence error leaves the cache internally consistent.
    pub fn replace_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
        let client_order_id = order.client_order_id();
        if let Some(venue_order_id) = order.venue_order_id() {
            self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
        }

        match self.orders.get(&client_order_id) {
            // Reuse the existing cell so the canonical entry stays in place rather than
            // orphaning a stale cell.
            Some(order_cell) => *order_cell.borrow_mut() = order.clone(),
            None => {
                self.orders
                    .insert(client_order_id, SharedCell::new(order.clone()));
            }
        }

        self.refresh_order(order)
    }

    /// Updates the cached order by applying an event and refreshing derived cache state.
    ///
    /// # Errors
    ///
    /// Returns an error if the order is not found or rejects the event.
    pub fn update_order(&mut self, event: &OrderEventAny) -> anyhow::Result<OrderAny> {
        let event_client_order_id = event.client_order_id();
        let client_order_id = if self.order_exists(&event_client_order_id) {
            event_client_order_id
        } else if let Some(venue_order_id) = event.venue_order_id() {
            self.index
                .venue_order_ids
                .get(&venue_order_id)
                .copied()
                .ok_or(OrderError::NotFound(event_client_order_id))?
        } else {
            return Err(OrderError::NotFound(event_client_order_id).into());
        };

        let order_cell = self
            .orders
            .get(&client_order_id)
            .cloned()
            .ok_or(OrderError::NotFound(client_order_id))?;

        // Apply on a snapshot first so a fallible `apply` (e.g. invalid state
        // transition) leaves the canonical cell untouched. On success we swap the
        // post-event value back into the cell so subsequent reads see the new state.
        let mut snapshot = order_cell.borrow().clone();
        snapshot.apply(event.clone())?;

        // Preflight only reverse ownership. A same-client forward mismatch remains a logged
        // refresh inconsistency, while other refresh failures, such as a backing database error,
        // remain logged after the canonical state is committed.
        if let Some(venue_order_id) = snapshot.venue_order_id() {
            self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
        }

        *order_cell.borrow_mut() = snapshot.clone();

        if matches!(event, OrderEventAny::FillVoided(_)) {
            self.reindex_hedging_position(client_order_id);
        }

        if let Err(e) = self.refresh_order(&snapshot) {
            log::error!("Error updating order in cache: {e}");
        }

        Ok(snapshot)
    }

    fn refresh_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
        let client_order_id = order.client_order_id();

        // Claim the venue order ID before mutating any other derived state. An updated event may
        // change the current ID for the same client order, while historical reverse aliases remain.
        if let Some(venue_order_id) = order.venue_order_id() {
            let overwrite = matches!(order.last_event(), OrderEventAny::Updated(_));
            if let Err(e) = self.add_venue_order_id(&client_order_id, &venue_order_id, overwrite) {
                if e.is::<VenueOrderIdOwnershipError>() {
                    return Err(e);
                }
                log::error!("Error indexing venue order ID in cache: {e}");
            }
        }

        if order.is_active_local() {
            self.index.orders_active_local.insert(client_order_id);
        } else {
            self.index.orders_active_local.remove(&client_order_id);
        }

        // Update in-flight state
        if order.is_inflight() {
            self.index.orders_inflight.insert(client_order_id);
        } else {
            self.index.orders_inflight.remove(&client_order_id);
        }

        // Update open/closed state
        if order.is_open() {
            self.index.orders_closed.remove(&client_order_id);
            self.index.orders_open.insert(client_order_id);
        } else {
            self.index.orders_open.remove(&client_order_id);
        }

        if order.is_closed() {
            self.index.orders_pending_cancel.remove(&client_order_id);
            self.index.orders_closed.insert(client_order_id);
        }

        // A cancel rejection resolves the outstanding cancel request
        if matches!(order.last_event(), OrderEventAny::CancelRejected(_)) {
            self.index.orders_pending_cancel.remove(&client_order_id);
        }

        // Update emulation index
        if order.emulation_trigger().is_some() && !order.is_closed() {
            self.index.orders_emulated.insert(client_order_id);
        } else {
            self.index.orders_emulated.remove(&client_order_id);
        }

        // Update account orders index when account_id becomes available
        if let Some(account_id) = order.account_id() {
            self.index
                .account_orders
                .entry(account_id)
                .or_default()
                .insert(client_order_id);
        }

        // Update own book
        if !self.own_books.is_empty() {
            let own_book = self.own_order_book(&order.instrument_id());
            if (own_book.is_some() && order.is_closed()) || should_handle_own_book_order(order) {
                self.update_own_order_book(order);
            }
        }

        if let Some(database) = &mut self.database {
            database.update_order(order.last_event())?;
            // TODO: Implement order snapshots
            // if self.snapshot_orders {
            //     database.snapshot_order_state(order)?;
            // }
        }

        Ok(())
    }

    /// Updates the `order` as pending cancel locally.
    pub fn update_order_pending_cancel_local(&mut self, order: &OrderAny) {
        self.index
            .orders_pending_cancel
            .insert(order.client_order_id());
    }

    /// Updates a `position` already held in the cache.
    ///
    /// Reuses the existing cell so any held [`PositionRef`] handles continue to point at the
    /// canonical entry.
    ///
    /// # Errors
    ///
    /// Returns an error if the position is not already held in the cache, or if updating the
    /// position in the database fails.
    pub fn update_position(&mut self, position: &Position) -> anyhow::Result<()> {
        let Some(position_cell) = self.positions.get(&position.id).cloned() else {
            anyhow::bail!("Cannot update position {}: not found in cache", position.id);
        };

        self.refresh_position_indexes(position);

        *position_cell.borrow_mut() = position.clone();

        if let Some(database) = &mut self.database {
            database.update_position(position)?;
            // TODO: Implement order snapshots
            // if self.snapshot_orders {
            //     database.snapshot_order_state(order)?;
            // }
        }

        Ok(())
    }

    /// Updates a cached position by applying an order fill in place.
    ///
    /// Returns a transient copy of the updated state without stored history. The canonical cached
    /// position retains its complete history. A fill without a position ID uses `position_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the position is not cached, the fill fails accounting validation,
    /// or updating the position in the database fails.
    pub fn update_position_from_fill(
        &mut self,
        position_id: PositionId,
        fill: &OrderFilled,
    ) -> anyhow::Result<Position> {
        let Some(position_cell) = self.positions.get(&position_id).cloned() else {
            anyhow::bail!("Cannot update position {position_id}: not found in cache");
        };

        let mut resolved_fill;
        let fill = if fill.position_id.is_none() {
            resolved_fill = fill.clone();
            resolved_fill.position_id = Some(position_id);
            &resolved_fill
        } else {
            fill
        };

        let position = {
            let mut position = position_cell.borrow_mut();
            position.try_apply(fill)?;
            position.clone_without_events()
        };

        self.refresh_position_indexes(&position);

        if let Some(database) = &mut self.database {
            database.update_position(&position_cell.borrow())?;
        }

        Ok(position)
    }

    /// Updates a cached position by applying an adjustment in place.
    ///
    /// The canonical cached position retains its complete history. As with
    /// [`Self::update_position_from_fill`], a failed database update leaves the adjustment
    /// applied; callers that need to roll back use [`Self::revert_position_adjustment`].
    ///
    /// # Errors
    ///
    /// Returns an error if the position is not already held in the cache, or if updating the
    /// position in the database fails.
    pub fn update_position_from_adjustment(
        &mut self,
        position_id: PositionId,
        adjustment: PositionAdjusted,
    ) -> anyhow::Result<()> {
        let Some(position_cell) = self.positions.get(&position_id).cloned() else {
            anyhow::bail!("Cannot update position {position_id}: not found in cache");
        };

        position_cell.borrow_mut().apply_adjustment(adjustment)?;
        self.refresh_position_indexes(&position_cell.borrow());

        if let Some(database) = &mut self.database {
            database.update_position(&position_cell.borrow())?;
        }

        Ok(())
    }

    /// Reverts the last adjustment applied to a cached position in place.
    ///
    /// Restores the position state from `prior`, the snapshot taken before the adjustment was
    /// applied, then refreshes the position indexes and persists the restored position.
    ///
    /// # Errors
    ///
    /// Returns an error if the position is not held in the cache, if `prior` cannot revert its
    /// last adjustment (see [`Position::revert_last_adjustment`]), or if updating the position in
    /// the database fails.
    pub fn revert_position_adjustment(&mut self, prior: &Position) -> anyhow::Result<()> {
        let Some(position_cell) = self.positions.get(&prior.id).cloned() else {
            anyhow::bail!("Cannot revert position {}: not found in cache", prior.id);
        };

        position_cell.borrow_mut().revert_last_adjustment(prior)?;
        self.refresh_position_indexes(&position_cell.borrow());

        if let Some(database) = &mut self.database {
            database.update_position(&position_cell.borrow())?;
        }

        Ok(())
    }

    /// Updates a cached position by applying an authoritative instrument close in place.
    ///
    /// Returns a transient copy of the settled state without stored history. The canonical cached
    /// position retains its complete history, including the close. A failed database update is
    /// logged rather than returned, because the cached position has already settled and callers
    /// still need its settled state.
    ///
    /// # Errors
    ///
    /// Returns an error if the position is not already held in the cache or the close cannot be
    /// applied. An error leaves the position unchanged.
    pub fn update_position_from_instrument_close(
        &mut self,
        position_id: PositionId,
        close: InstrumentClose,
    ) -> anyhow::Result<Position> {
        let Some(position_cell) = self.positions.get(&position_id).cloned() else {
            anyhow::bail!("Cannot update position {position_id}: not found in cache");
        };

        let position = {
            let mut position = position_cell.borrow_mut();
            position.apply_instrument_close(close)?;
            position.clone_without_events()
        };

        self.refresh_position_indexes(&position);

        if let Some(database) = &mut self.database
            && let Err(e) = database.update_position(&position_cell.borrow())
        {
            log::error!("Failed to persist settled position {position_id}: {e}");
        }

        Ok(position)
    }

    fn refresh_position_indexes(&mut self, position: &Position) {
        if position.is_open() {
            self.index.positions_open.insert(position.id);
            self.index.positions_closed.remove(&position.id);
        } else {
            self.index.positions_closed.insert(position.id);
            self.index.positions_open.remove(&position.id);
        }
    }

    /// Gets the OMS type for the `position_id`.
    #[must_use]
    pub fn oms_type(&self, position_id: &PositionId) -> Option<OmsType> {
        self.index.position_oms.get(position_id).copied()
    }

    /// Snapshots the `order` state in the database.
    ///
    /// # Errors
    ///
    /// Returns an error if snapshotting the order state fails.
    pub fn snapshot_order_state(&self, order: &OrderAny) -> anyhow::Result<()> {
        let Some(database) = &self.database else {
            log::warn!(
                "Cannot snapshot order state for {} (no database configured)",
                order.client_order_id()
            );
            return Ok(());
        };

        database.snapshot_order_state(order)
    }

    // -- IDENTIFIER QUERIES ----------------------------------------------------------------------

    // Collects references to the index sets that constrain an order query.
    //
    // Returns:
    // - `FilterSources::Unfiltered` when no filter is provided (the caller should iterate
    //   the full bucket).
    // - `FilterSources::Empty` when a filter is provided but the index has no entry for it
    //   (the resolved set is unconditionally empty, no further work needed).
    // - `FilterSources::Sets` with borrowed references to each filter source set.
    fn collect_order_filter_sources<'a>(
        &'a self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> FilterSources<'a, ClientOrderId> {
        let mut sources: Vec<&AHashSet<ClientOrderId>> = Vec::with_capacity(4);

        if let Some(venue) = venue {
            match self.index.venue_orders.get(venue) {
                Some(set) => sources.push(set),
                None => return FilterSources::Empty,
            }
        }

        if let Some(instrument_id) = instrument_id {
            match self.index.instrument_orders.get(instrument_id) {
                Some(set) => sources.push(set),
                None => return FilterSources::Empty,
            }
        }

        if let Some(strategy_id) = strategy_id {
            match self.index.strategy_orders.get(strategy_id) {
                Some(set) => sources.push(set),
                None => return FilterSources::Empty,
            }
        }

        if let Some(account_id) = account_id {
            match self.index.account_orders.get(account_id) {
                Some(set) => sources.push(set),
                None => return FilterSources::Empty,
            }
        }

        if sources.is_empty() {
            FilterSources::Unfiltered
        } else {
            FilterSources::Sets(sources)
        }
    }

    fn collect_position_filter_sources<'a>(
        &'a self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> FilterSources<'a, PositionId> {
        let mut sources: Vec<&AHashSet<PositionId>> = Vec::with_capacity(4);

        if let Some(venue) = venue {
            match self.index.venue_positions.get(venue) {
                Some(set) => sources.push(set),
                None => return FilterSources::Empty,
            }
        }

        if let Some(instrument_id) = instrument_id {
            match self.index.instrument_positions.get(instrument_id) {
                Some(set) => sources.push(set),
                None => return FilterSources::Empty,
            }
        }

        if let Some(strategy_id) = strategy_id {
            match self.index.strategy_positions.get(strategy_id) {
                Some(set) => sources.push(set),
                None => return FilterSources::Empty,
            }
        }

        if let Some(account_id) = account_id {
            match self.index.account_positions.get(account_id) {
                Some(set) => sources.push(set),
                None => return FilterSources::Empty,
            }
        }

        if sources.is_empty() {
            FilterSources::Unfiltered
        } else {
            FilterSources::Sets(sources)
        }
    }

    // Materializes the `ClientOrderId`s in `bucket` matching the optional filter parameters.
    //
    // Folds the bucket into the filter sources and runs a single size-ordered intersection,
    // avoiding the legacy two-step build-filter-set + bucket-intersection that allocated and
    // rehashed twice.
    fn query_orders_in_bucket(
        &self,
        bucket: &AHashSet<ClientOrderId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<ClientOrderId> {
        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => AHashSet::new(),
            FilterSources::Unfiltered => bucket.clone(),
            FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
        }
    }

    fn query_positions_in_bucket(
        &self,
        bucket: &AHashSet<PositionId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<PositionId> {
        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => AHashSet::new(),
            FilterSources::Unfiltered => bucket.clone(),
            FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
        }
    }

    // Returns a borrowed or owned view of the orders in `bucket` matching the optional filter
    // parameters. Avoids cloning the bucket when no filter narrows it.
    fn view_orders_in_bucket<'a>(
        &'a self,
        bucket: &'a AHashSet<ClientOrderId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'a, AHashSet<ClientOrderId>> {
        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => Cow::Owned(AHashSet::new()),
            FilterSources::Unfiltered => Cow::Borrowed(bucket),
            FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
        }
    }

    fn view_positions_in_bucket<'a>(
        &'a self,
        bucket: &'a AHashSet<PositionId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'a, AHashSet<PositionId>> {
        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => Cow::Owned(AHashSet::new()),
            FilterSources::Unfiltered => Cow::Borrowed(bucket),
            FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
        }
    }

    // Returns a lazy iterator yielding the [`ClientOrderId`]s in `bucket` matching the optional
    // filter parameters. Avoids any [`Vec`] or [`AHashSet`] materialization in the result path,
    // and (for multi-filter calls) drives intersection from the smallest source while looking
    // up membership in the rest.
    fn iter_orders_in_bucket<'a>(
        &'a self,
        bucket: &'a AHashSet<ClientOrderId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = ClientOrderId> + 'a> {
        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => Box::new(std::iter::empty()),
            FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
            FilterSources::Sets(mut sources) => {
                sources.push(bucket);
                sources.sort_unstable_by_key(|s| s.len());
                let driver = sources[0];
                let rest: Vec<&'a AHashSet<ClientOrderId>> = sources[1..].to_vec();
                Box::new(
                    driver
                        .iter()
                        .copied()
                        .filter(move |id| rest.iter().all(|s| s.contains(id))),
                )
            }
        }
    }

    fn iter_positions_in_bucket<'a>(
        &'a self,
        bucket: &'a AHashSet<PositionId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = PositionId> + 'a> {
        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => Box::new(std::iter::empty()),
            FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
            FilterSources::Sets(mut sources) => {
                sources.push(bucket);
                sources.sort_unstable_by_key(|s| s.len());
                let driver = sources[0];
                let rest: Vec<&'a AHashSet<PositionId>> = sources[1..].to_vec();
                Box::new(
                    driver
                        .iter()
                        .copied()
                        .filter(move |id| rest.iter().all(|s| s.contains(id))),
                )
            }
        }
    }

    // Counts orders in `bucket` matching the optional filter parameters.
    //
    // Drives intersection from the smallest filter source (or the bucket itself when no filter
    // is provided) and short-circuits by counting rather than collecting. With a side filter,
    // each candidate order is borrowed via its cell only long enough to inspect the side.
    fn count_orders_in_bucket(
        &self,
        bucket: &AHashSet<ClientOrderId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> usize {
        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => 0,
            FilterSources::Unfiltered => side.map_or_else(
                || bucket.len(),
                |side| {
                    bucket
                        .iter()
                        .filter(|id| self.order_side_matches(id, side))
                        .count()
                },
            ),
            FilterSources::Sets(mut sources) => {
                sources.push(bucket);
                sources.sort_unstable_by_key(|s| s.len());
                let driver = sources[0];
                let rest = &sources[1..];

                driver
                    .iter()
                    .filter(|id| rest.iter().all(|s| s.contains(id)))
                    .filter(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
                    .count()
            }
        }
    }

    fn count_positions_in_bucket(
        &self,
        bucket: &AHashSet<PositionId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> usize {
        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => 0,
            FilterSources::Unfiltered => side.map_or_else(
                || bucket.len(),
                |side| {
                    bucket
                        .iter()
                        .filter(|id| self.position_side_matches(id, side))
                        .count()
                },
            ),
            FilterSources::Sets(mut sources) => {
                sources.push(bucket);
                sources.sort_unstable_by_key(|s| s.len());
                let driver = sources[0];
                let rest = &sources[1..];

                driver
                    .iter()
                    .filter(|id| rest.iter().all(|s| s.contains(id)))
                    .filter(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
                    .count()
            }
        }
    }

    // Returns whether any order in `bucket` matches the optional filter parameters.
    //
    // Mirrors `count_orders_in_bucket` but short-circuits on the first match. Useful for
    // `is_empty`-style gating in hot paths where the caller only needs to know whether at
    // least one matching order exists.
    fn any_orders_in_bucket(
        &self,
        bucket: &AHashSet<ClientOrderId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> bool {
        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => false,
            FilterSources::Unfiltered => side.map_or_else(
                || !bucket.is_empty(),
                |side| bucket.iter().any(|id| self.order_side_matches(id, side)),
            ),
            FilterSources::Sets(mut sources) => {
                sources.push(bucket);
                sources.sort_unstable_by_key(|s| s.len());
                let driver = sources[0];
                let rest = &sources[1..];

                driver
                    .iter()
                    .filter(|id| rest.iter().all(|s| s.contains(id)))
                    .any(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
            }
        }
    }

    fn any_positions_in_bucket(
        &self,
        bucket: &AHashSet<PositionId>,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> bool {
        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
            FilterSources::Empty => false,
            FilterSources::Unfiltered => side.map_or_else(
                || !bucket.is_empty(),
                |side| bucket.iter().any(|id| self.position_side_matches(id, side)),
            ),
            FilterSources::Sets(mut sources) => {
                sources.push(bucket);
                sources.sort_unstable_by_key(|s| s.len());
                let driver = sources[0];
                let rest = &sources[1..];

                driver
                    .iter()
                    .filter(|id| rest.iter().all(|s| s.contains(id)))
                    .any(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
            }
        }
    }

    fn order_side_matches(&self, client_order_id: &ClientOrderId, side: OrderSide) -> bool {
        self.orders
            .get(client_order_id)
            .is_some_and(|cell| cell.borrow().order_side() == side)
    }

    fn position_side_matches(&self, position_id: &PositionId, side: PositionSide) -> bool {
        self.positions
            .get(position_id)
            .is_some_and(|cell| cell.borrow().side == side)
    }

    /// Retrieves orders corresponding to the `client_order_ids`, optionally filtering by `side`.
    ///
    /// # Panics
    ///
    /// Panics if any `client_order_id` in the input is not found in the cache.
    fn get_orders_for_ids(
        &self,
        client_order_ids: impl IntoIterator<Item = ClientOrderId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        const UNCACHED_SORT_MAX_LEN: usize = 32;

        let mut orders = Vec::new();

        for client_order_id in client_order_ids {
            let order_cell = self
                .orders
                .get(&client_order_id)
                .unwrap_or_else(|| panic!("Order {client_order_id} not found"));
            let order = OrderRef::new(order_cell.borrow());

            if side.is_none_or(|side| side == order.order_side()) {
                orders.push(order);
            }
        }

        // Sort so callers receive a deterministic Vec across runs; the
        // underlying ID sources are AHash-backed.
        let key = |order: &OrderRef<'_>| order.client_order_id();

        if orders.len() <= UNCACHED_SORT_MAX_LEN {
            orders.sort_by_key(key);
        } else {
            orders.sort_by_cached_key(key);
        }

        orders
    }

    /// Retrieves positions corresponding to the `position_ids`, optionally filtering by `side`.
    ///
    /// Each [`PositionRef`] in the returned vector borrows its underlying cell; mutating any of
    /// those positions while the vector is alive will panic at runtime. Drop the vector before
    /// issuing writes.
    ///
    /// # Panics
    ///
    /// Panics if any `position_id` in the set is not found in the cache.
    fn get_positions_for_ids(
        &self,
        position_ids: &AHashSet<PositionId>,
        side: Option<PositionSide>,
    ) -> Vec<PositionRef<'_>> {
        let mut positions = Vec::new();

        for position_id in position_ids {
            let position_cell = self
                .positions
                .get(position_id)
                .unwrap_or_else(|| panic!("Position {position_id} not found"));
            let position = PositionRef::new(position_cell.borrow());

            if side.is_none_or(|side| side == position.side) {
                positions.push(position);
            }
        }

        // Sort so callers receive a deterministic Vec across runs; the
        // underlying position_ids set is AHash-backed.
        positions.sort_by_key(|p| p.id);
        positions
    }

    /// Returns the `ClientOrderId`s of all orders.
    #[must_use]
    pub fn client_order_ids(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<ClientOrderId> {
        self.query_orders_in_bucket(
            &self.index.orders,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns the `ClientOrderId`s of all open orders.
    #[must_use]
    pub fn client_order_ids_open(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<ClientOrderId> {
        self.query_orders_in_bucket(
            &self.index.orders_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns the `ClientOrderId`s of all closed orders.
    #[must_use]
    pub fn client_order_ids_closed(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<ClientOrderId> {
        self.query_orders_in_bucket(
            &self.index.orders_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns the `ClientOrderId`s of all locally active orders.
    ///
    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state
    /// (a superset of emulated orders).
    #[must_use]
    pub fn client_order_ids_active_local(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<ClientOrderId> {
        self.query_orders_in_bucket(
            &self.index.orders_active_local,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns the `ClientOrderId`s of all emulated orders.
    #[must_use]
    pub fn client_order_ids_emulated(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<ClientOrderId> {
        self.query_orders_in_bucket(
            &self.index.orders_emulated,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns the `ClientOrderId`s of all in-flight orders.
    #[must_use]
    pub fn client_order_ids_inflight(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<ClientOrderId> {
        self.query_orders_in_bucket(
            &self.index.orders_inflight,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns `PositionId`s of all positions.
    #[must_use]
    pub fn position_ids(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<PositionId> {
        self.query_positions_in_bucket(
            &self.index.positions,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns the `PositionId`s of all open positions.
    #[must_use]
    pub fn position_open_ids(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<PositionId> {
        self.query_positions_in_bucket(
            &self.index.positions_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns the `PositionId`s of all closed positions.
    #[must_use]
    pub fn position_closed_ids(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> AHashSet<PositionId> {
        self.query_positions_in_bucket(
            &self.index.positions_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a borrowed view over the [`ClientOrderId`]s of all orders matching the optional
    /// filter parameters.
    ///
    /// The returned [`Cow`] borrows the underlying index when no filter is provided and only
    /// allocates an owned [`AHashSet`] when an intersection is required. Prefer this over
    /// [`Self::client_order_ids`] when the caller only needs to iterate or read membership.
    #[must_use]
    pub fn client_order_ids_view(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'_, AHashSet<ClientOrderId>> {
        self.view_orders_in_bucket(
            &self.index.orders,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a borrowed view over the [`ClientOrderId`]s of all open orders.
    #[must_use]
    pub fn client_order_ids_open_view(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'_, AHashSet<ClientOrderId>> {
        self.view_orders_in_bucket(
            &self.index.orders_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a borrowed view over the [`ClientOrderId`]s of all closed orders.
    #[must_use]
    pub fn client_order_ids_closed_view(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'_, AHashSet<ClientOrderId>> {
        self.view_orders_in_bucket(
            &self.index.orders_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a borrowed view over the [`ClientOrderId`]s of all locally active orders.
    #[must_use]
    pub fn client_order_ids_active_local_view(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'_, AHashSet<ClientOrderId>> {
        self.view_orders_in_bucket(
            &self.index.orders_active_local,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a borrowed view over the [`ClientOrderId`]s of all emulated orders.
    #[must_use]
    pub fn client_order_ids_emulated_view(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'_, AHashSet<ClientOrderId>> {
        self.view_orders_in_bucket(
            &self.index.orders_emulated,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a borrowed view over the [`ClientOrderId`]s of all in-flight orders.
    #[must_use]
    pub fn client_order_ids_inflight_view(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'_, AHashSet<ClientOrderId>> {
        self.view_orders_in_bucket(
            &self.index.orders_inflight,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a borrowed view over the [`PositionId`]s of all positions.
    #[must_use]
    pub fn position_ids_view(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'_, AHashSet<PositionId>> {
        self.view_positions_in_bucket(
            &self.index.positions,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a borrowed view over the [`PositionId`]s of all open positions.
    #[must_use]
    pub fn position_open_ids_view(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'_, AHashSet<PositionId>> {
        self.view_positions_in_bucket(
            &self.index.positions_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a borrowed view over the [`PositionId`]s of all closed positions.
    #[must_use]
    pub fn position_closed_ids_view(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Cow<'_, AHashSet<PositionId>> {
        self.view_positions_in_bucket(
            &self.index.positions_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all orders matching the optional
    /// filter parameters.
    ///
    /// Avoids the [`AHashSet`] allocation performed by [`Self::client_order_ids`]. Useful when
    /// the caller iterates the result once and discards it.
    pub fn iter_client_order_ids(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
        self.iter_orders_in_bucket(
            &self.index.orders,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all open orders.
    pub fn iter_client_order_ids_open(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
        self.iter_orders_in_bucket(
            &self.index.orders_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all closed orders.
    pub fn iter_client_order_ids_closed(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
        self.iter_orders_in_bucket(
            &self.index.orders_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all locally active orders.
    pub fn iter_client_order_ids_active_local(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
        self.iter_orders_in_bucket(
            &self.index.orders_active_local,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all emulated orders.
    pub fn iter_client_order_ids_emulated(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
        self.iter_orders_in_bucket(
            &self.index.orders_emulated,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all in-flight orders.
    pub fn iter_client_order_ids_inflight(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
        self.iter_orders_in_bucket(
            &self.index.orders_inflight,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a lazy iterator yielding [`PositionId`]s of all positions matching the filters.
    pub fn iter_position_ids(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = PositionId> + '_> {
        self.iter_positions_in_bucket(
            &self.index.positions,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a lazy iterator yielding [`PositionId`]s of all open positions.
    pub fn iter_position_open_ids(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = PositionId> + '_> {
        self.iter_positions_in_bucket(
            &self.index.positions_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns a lazy iterator yielding [`PositionId`]s of all closed positions.
    pub fn iter_position_closed_ids(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Box<dyn Iterator<Item = PositionId> + '_> {
        self.iter_positions_in_bucket(
            &self.index.positions_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        )
    }

    /// Returns the `StrategyId`s of all strategies.
    #[must_use]
    pub fn strategy_ids(&self) -> AHashSet<StrategyId> {
        self.index.strategies.clone()
    }

    /// Returns the `ExecAlgorithmId`s of all execution algorithms.
    #[must_use]
    pub fn exec_algorithm_ids(&self) -> AHashSet<ExecAlgorithmId> {
        self.index.exec_algorithms.clone()
    }

    // -- ORDER QUERIES ---------------------------------------------------------------------------

    /// Gets a borrow of the order with the `client_order_id` (if found).
    ///
    /// The returned [`OrderRef`] is tied to the cache borrow's scope and panics at runtime if
    /// held across a mutation of the same order. Drop the borrow before dispatching events; if
    /// post-event state is required, perform a fresh lookup. Use [`Self::order_owned`] when an
    /// owned snapshot is needed for a boundary handover.
    #[must_use]
    pub fn order_ref(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
        self.orders
            .get(client_order_id)
            .map(|order_cell| OrderRef::new(order_cell.borrow()))
    }

    /// Gets a borrow of the order with the `client_order_id` (if found).
    ///
    /// Prefer [`Self::order_ref`] in new native code.
    #[must_use]
    pub fn order(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
        self.order_ref(client_order_id)
    }

    /// Gets a borrow of the order with the `client_order_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the order is not present in the cache.
    pub fn try_order_ref(
        &self,
        client_order_id: &ClientOrderId,
    ) -> Result<OrderRef<'_>, OrderLookupError> {
        self.orders
            .get(client_order_id)
            .map(|order_cell| OrderRef::new(order_cell.borrow()))
            .ok_or_else(|| OrderLookupError::not_found(*client_order_id))
    }

    /// Gets a borrow of the order with the `client_order_id`.
    ///
    /// Prefer [`Self::try_order_ref`] in new native code.
    ///
    /// # Errors
    ///
    /// Returns an error if the order is not present in the cache.
    pub fn try_order(
        &self,
        client_order_id: &ClientOrderId,
    ) -> Result<OrderRef<'_>, OrderLookupError> {
        self.try_order_ref(client_order_id)
    }

    /// Gets an exclusive write borrow of the order with the `client_order_id` (if found).
    ///
    /// Requires `&mut Cache` so cache writes are reachable only by privileged crates that hold
    /// `Rc<RefCell<Cache>>` directly. Adapter-facing code receives [`CacheView`], which only
    /// exposes immutable cache borrows and therefore cannot reach this method.
    ///
    /// While the returned [`OrderRefMut`] is alive, no other read or write of the same order is
    /// permitted. Drop the borrow before dispatching events or taking any other cache borrow that
    /// may re-enter the same order.
    #[must_use]
    pub fn order_mut(&mut self, client_order_id: &ClientOrderId) -> Option<OrderRefMut<'_>> {
        self.orders
            .get(client_order_id)
            .map(|order_cell| OrderRefMut::new(order_cell.borrow_mut()))
    }

    /// Gets an owned copy of the order with the `client_order_id` (if found).
    ///
    /// Use when downstream needs an owned [`OrderAny`] that crosses a boundary (for example, an
    /// adapter `get_order` API). The copy will not reflect later cache mutations.
    #[must_use]
    pub fn order_owned(&self, client_order_id: &ClientOrderId) -> Option<OrderAny> {
        self.orders
            .get(client_order_id)
            .map(|order_cell| order_cell.borrow().clone())
    }

    /// Gets an owned snapshot of the order with the `client_order_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the order is not present in the cache.
    pub fn try_order_owned(
        &self,
        client_order_id: &ClientOrderId,
    ) -> Result<OrderAny, OrderLookupError> {
        self.try_order_ref(client_order_id)
            .map(|order| order.cloned())
    }

    /// Gets cloned orders for the given `client_order_ids`, logging an error for any missing.
    #[must_use]
    pub fn orders_for_ids(
        &self,
        client_order_ids: &[ClientOrderId],
        context: &dyn Display,
    ) -> Vec<OrderAny> {
        let mut orders = Vec::with_capacity(client_order_ids.len());
        for id in client_order_ids {
            match self.orders.get(id) {
                Some(order_cell) => orders.push(order_cell.borrow().clone()),
                None => log::error!("Order {id} not found in cache for {context}"),
            }
        }
        orders
    }

    /// Gets a reference to the client order ID for the `venue_order_id` (if found).
    #[must_use]
    pub fn client_order_id(&self, venue_order_id: &VenueOrderId) -> Option<&ClientOrderId> {
        self.index.venue_order_ids.get(venue_order_id)
    }

    /// Gets a reference to the venue order ID for the `client_order_id` (if found).
    #[must_use]
    pub fn venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<&VenueOrderId> {
        self.index.client_order_ids.get(client_order_id)
    }

    /// Gets a reference to the client ID indexed for then `client_order_id` (if found).
    #[must_use]
    pub fn client_id(&self, client_order_id: &ClientOrderId) -> Option<&ClientId> {
        self.index.order_client.get(client_order_id)
    }

    /// Returns borrows of all orders matching the optional filter parameters.
    ///
    /// Each [`OrderRef`] in the returned vector borrows its underlying cell; mutating any of
    /// those orders while the vector is alive will panic at runtime. Drop the vector
    /// before issuing writes.
    #[must_use]
    pub fn orders_refs(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        if venue.is_none()
            && instrument_id.is_none()
            && strategy_id.is_none()
            && account_id.is_none()
        {
            return self.get_orders_for_ids(self.index.orders.iter().copied(), side);
        }

        let client_order_ids =
            self.iter_client_order_ids(venue, instrument_id, strategy_id, account_id);
        self.get_orders_for_ids(client_order_ids, side)
    }

    /// Returns borrows of all orders matching the optional filter parameters.
    ///
    /// Prefer [`Self::orders_refs`] in new native code.
    #[must_use]
    pub fn orders(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        self.orders_refs(venue, instrument_id, strategy_id, account_id, side)
    }

    /// Returns borrows of all open orders matching the optional filter parameters.
    #[must_use]
    pub fn orders_open_refs(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        let client_order_ids =
            self.client_order_ids_open(venue, instrument_id, strategy_id, account_id);
        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
    }

    /// Returns borrows of all open orders matching the optional filter parameters.
    ///
    /// Prefer [`Self::orders_open_refs`] in new native code.
    #[must_use]
    pub fn orders_open(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        self.orders_open_refs(venue, instrument_id, strategy_id, account_id, side)
    }

    /// Returns borrows of all closed orders matching the optional filter parameters.
    #[must_use]
    pub fn orders_closed_refs(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        let client_order_ids =
            self.client_order_ids_closed(venue, instrument_id, strategy_id, account_id);
        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
    }

    /// Returns borrows of all closed orders matching the optional filter parameters.
    ///
    /// Prefer [`Self::orders_closed_refs`] in new native code.
    #[must_use]
    pub fn orders_closed(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        self.orders_closed_refs(venue, instrument_id, strategy_id, account_id, side)
    }

    /// Returns borrows of all locally active orders matching the optional filter parameters.
    ///
    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state
    /// (a superset of emulated orders).
    #[must_use]
    pub fn orders_active_local_refs(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        let client_order_ids =
            self.client_order_ids_active_local(venue, instrument_id, strategy_id, account_id);
        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
    }

    /// Returns borrows of all locally active orders matching the optional filter parameters.
    ///
    /// Prefer [`Self::orders_active_local_refs`] in new native code.
    #[must_use]
    pub fn orders_active_local(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        self.orders_active_local_refs(venue, instrument_id, strategy_id, account_id, side)
    }

    /// Returns borrows of all emulated orders matching the optional filter parameters.
    #[must_use]
    pub fn orders_emulated_refs(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        let client_order_ids =
            self.client_order_ids_emulated(venue, instrument_id, strategy_id, account_id);
        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
    }

    /// Returns borrows of all emulated orders matching the optional filter parameters.
    ///
    /// Prefer [`Self::orders_emulated_refs`] in new native code.
    #[must_use]
    pub fn orders_emulated(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        self.orders_emulated_refs(venue, instrument_id, strategy_id, account_id, side)
    }

    /// Returns borrows of all in-flight orders matching the optional filter parameters.
    #[must_use]
    pub fn orders_inflight_refs(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        let client_order_ids =
            self.client_order_ids_inflight(venue, instrument_id, strategy_id, account_id);
        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
    }

    /// Returns borrows of all in-flight orders matching the optional filter parameters.
    ///
    /// Prefer [`Self::orders_inflight_refs`] in new native code.
    #[must_use]
    pub fn orders_inflight(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        self.orders_inflight_refs(venue, instrument_id, strategy_id, account_id, side)
    }

    /// Returns borrows of all orders for the `position_id`.
    #[must_use]
    pub fn orders_for_position(&self, position_id: &PositionId) -> Vec<OrderRef<'_>> {
        match self.index.position_orders.get(position_id) {
            Some(client_order_ids) => {
                self.get_orders_for_ids(client_order_ids.iter().copied(), None)
            }
            None => Vec::new(),
        }
    }

    /// Records execution client handoffs until their first status transitions.
    ///
    /// Records are process-local and may refer to uncached external orders.
    pub fn mark_orders_dispatched(
        &mut self,
        client_order_ids: impl IntoIterator<Item = ClientOrderId>,
    ) {
        self.orders_dispatched.extend(client_order_ids);
    }

    /// Removes the execution dispatch record for an order.
    pub fn clear_order_dispatch(&mut self, client_order_id: &ClientOrderId) {
        self.orders_dispatched.remove(client_order_id);
    }

    /// Returns whether an order was dispatched and awaits its first status transition.
    #[must_use]
    pub fn is_order_dispatched(&self, client_order_id: &ClientOrderId) -> bool {
        self.orders_dispatched.contains(client_order_id)
    }

    /// Returns whether an order with the `client_order_id` exists.
    #[must_use]
    pub fn order_exists(&self, client_order_id: &ClientOrderId) -> bool {
        self.index.orders.contains(client_order_id)
    }

    /// Returns whether an order with the `client_order_id` is open.
    #[must_use]
    pub fn is_order_open(&self, client_order_id: &ClientOrderId) -> bool {
        self.index.orders_open.contains(client_order_id)
    }

    /// Returns whether an order with the `client_order_id` is closed.
    #[must_use]
    pub fn is_order_closed(&self, client_order_id: &ClientOrderId) -> bool {
        self.index.orders_closed.contains(client_order_id)
    }

    /// Returns whether an order with the `client_order_id` is locally active.
    ///
    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state
    /// (a superset of emulated orders).
    #[must_use]
    pub fn is_order_active_local(&self, client_order_id: &ClientOrderId) -> bool {
        self.index.orders_active_local.contains(client_order_id)
    }

    /// Returns whether an order with the `client_order_id` is emulated.
    #[must_use]
    pub fn is_order_emulated(&self, client_order_id: &ClientOrderId) -> bool {
        self.index.orders_emulated.contains(client_order_id)
    }

    /// Returns whether an order with the `client_order_id` is in-flight.
    #[must_use]
    pub fn is_order_inflight(&self, client_order_id: &ClientOrderId) -> bool {
        self.index.orders_inflight.contains(client_order_id)
    }

    /// Returns whether an order with the `client_order_id` is `PENDING_CANCEL` locally.
    #[must_use]
    pub fn is_order_pending_cancel_local(&self, client_order_id: &ClientOrderId) -> bool {
        self.index.orders_pending_cancel.contains(client_order_id)
    }

    /// Returns the count of all open orders.
    #[must_use]
    pub fn orders_open_count(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> usize {
        self.count_orders_in_bucket(
            &self.index.orders_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns the count of all closed orders.
    #[must_use]
    pub fn orders_closed_count(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> usize {
        self.count_orders_in_bucket(
            &self.index.orders_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns the count of all locally active orders.
    ///
    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state
    /// (a superset of emulated orders).
    #[must_use]
    pub fn orders_active_local_count(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> usize {
        self.count_orders_in_bucket(
            &self.index.orders_active_local,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns the count of all emulated orders.
    #[must_use]
    pub fn orders_emulated_count(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> usize {
        self.count_orders_in_bucket(
            &self.index.orders_emulated,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns the count of all in-flight orders.
    #[must_use]
    pub fn orders_inflight_count(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> usize {
        self.count_orders_in_bucket(
            &self.index.orders_inflight,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns the count of all orders.
    #[must_use]
    pub fn orders_total_count(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> usize {
        self.count_orders_in_bucket(
            &self.index.orders,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns whether any open order matches the optional filter parameters.
    ///
    /// Short-circuits on the first match, avoiding the full intersection walk performed by
    /// [`Self::orders_open_count`]. Prefer this over `orders_open_count(...) > 0` when only
    /// existence matters.
    #[must_use]
    pub fn has_orders_open(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> bool {
        self.any_orders_in_bucket(
            &self.index.orders_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns whether any closed order matches the optional filter parameters.
    #[must_use]
    pub fn has_orders_closed(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> bool {
        self.any_orders_in_bucket(
            &self.index.orders_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns whether any locally active order matches the optional filter parameters.
    ///
    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state.
    #[must_use]
    pub fn has_orders_active_local(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> bool {
        self.any_orders_in_bucket(
            &self.index.orders_active_local,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns whether any emulated order matches the optional filter parameters.
    #[must_use]
    pub fn has_orders_emulated(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> bool {
        self.any_orders_in_bucket(
            &self.index.orders_emulated,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns whether any in-flight order matches the optional filter parameters.
    #[must_use]
    pub fn has_orders_inflight(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> bool {
        self.any_orders_in_bucket(
            &self.index.orders_inflight,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns whether any order (in any state) matches the optional filter parameters.
    #[must_use]
    pub fn has_orders(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> bool {
        self.any_orders_in_bucket(
            &self.index.orders,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns the order list for the `order_list_id`.
    #[must_use]
    pub fn order_list(&self, order_list_id: &OrderListId) -> Option<&OrderList> {
        self.order_lists.get(order_list_id)
    }

    /// Returns the order list for the `order_list_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the order list is not present in the cache.
    pub fn try_order_list(
        &self,
        order_list_id: &OrderListId,
    ) -> Result<&OrderList, OrderListLookupError> {
        self.order_lists
            .get(order_list_id)
            .ok_or_else(|| OrderListLookupError::not_found(*order_list_id))
    }

    /// Returns all order lists matching the optional filter parameters.
    #[must_use]
    pub fn order_lists(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
    ) -> Vec<&OrderList> {
        let mut order_lists = self.order_lists.values().collect::<Vec<&OrderList>>();

        if let Some(venue) = venue {
            order_lists.retain(|ol| &ol.instrument_id.venue == venue);
        }

        if let Some(instrument_id) = instrument_id {
            order_lists.retain(|ol| &ol.instrument_id == instrument_id);
        }

        if let Some(strategy_id) = strategy_id {
            order_lists.retain(|ol| &ol.strategy_id == strategy_id);
        }

        if let Some(account_id) = account_id {
            order_lists.retain(|ol| {
                ol.client_order_ids.iter().any(|client_order_id| {
                    self.orders.get(client_order_id).is_some_and(|order_cell| {
                        order_cell.borrow().account_id().as_ref() == Some(account_id)
                    })
                })
            });
        }

        order_lists
    }

    /// Returns whether an order list with the `order_list_id` exists.
    #[must_use]
    pub fn order_list_exists(&self, order_list_id: &OrderListId) -> bool {
        self.order_lists.contains_key(order_list_id)
    }

    // -- EXEC ALGORITHM QUERIES ------------------------------------------------------------------

    /// Returns references to all orders associated with the `exec_algorithm_id` matching the
    /// optional filter parameters.
    #[must_use]
    pub fn orders_for_exec_algorithm(
        &self,
        exec_algorithm_id: &ExecAlgorithmId,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<OrderSide>,
    ) -> Vec<OrderRef<'_>> {
        let Some(exec_algorithm_order_ids) =
            self.index.exec_algorithm_orders.get(exec_algorithm_id)
        else {
            return Vec::new();
        };

        let filtered = self.query_orders_in_bucket(
            exec_algorithm_order_ids,
            venue,
            instrument_id,
            strategy_id,
            account_id,
        );
        self.get_orders_for_ids(filtered.iter().copied(), side)
    }

    /// Returns references to all orders with the `exec_spawn_id`.
    #[must_use]
    pub fn orders_for_exec_spawn(&self, exec_spawn_id: &ClientOrderId) -> Vec<OrderRef<'_>> {
        match self.index.exec_spawn_orders.get(exec_spawn_id) {
            Some(ids) => self.get_orders_for_ids(ids.iter().copied(), None),
            None => Vec::new(),
        }
    }

    /// Returns the total order quantity for the `exec_spawn_id`.
    #[must_use]
    pub fn exec_spawn_total_quantity(
        &self,
        exec_spawn_id: &ClientOrderId,
        active_only: bool,
    ) -> Option<Quantity> {
        self.exec_spawn_total(exec_spawn_id, active_only, Order::quantity)
    }

    /// Returns the total filled quantity for all orders with the `exec_spawn_id`.
    #[must_use]
    pub fn exec_spawn_total_filled_qty(
        &self,
        exec_spawn_id: &ClientOrderId,
        active_only: bool,
    ) -> Option<Quantity> {
        self.exec_spawn_total(exec_spawn_id, active_only, Order::filled_qty)
    }

    /// Returns the total leaves quantity for all orders with the `exec_spawn_id`.
    #[must_use]
    pub fn exec_spawn_total_leaves_qty(
        &self,
        exec_spawn_id: &ClientOrderId,
        active_only: bool,
    ) -> Option<Quantity> {
        self.exec_spawn_total(exec_spawn_id, active_only, Order::leaves_qty)
    }

    fn exec_spawn_total(
        &self,
        exec_spawn_id: &ClientOrderId,
        active_only: bool,
        quantity: impl Fn(&OrderAny) -> Quantity,
    ) -> Option<Quantity> {
        self.orders_for_exec_spawn(exec_spawn_id)
            .into_iter()
            .filter(|order| !active_only || !order.is_closed())
            .map(|order| quantity(&order))
            .reduce(|total, quantity| total + quantity)
    }

    // -- POSITION QUERIES ------------------------------------------------------------------------

    /// Returns a borrow of the position with the `position_id` (if found).
    #[must_use]
    pub fn position_ref(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
        self.positions
            .get(position_id)
            .map(|position_cell| PositionRef::new(position_cell.borrow()))
    }

    /// Returns a borrow of the position with the `position_id` (if found).
    ///
    /// Prefer [`Self::position_ref`] in new native code.
    #[must_use]
    pub fn position(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
        self.position_ref(position_id)
    }

    /// Returns a borrow of the position with the `position_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the position is not present in the cache.
    pub fn try_position_ref(
        &self,
        position_id: &PositionId,
    ) -> Result<PositionRef<'_>, PositionLookupError> {
        self.positions
            .get(position_id)
            .map(|position_cell| PositionRef::new(position_cell.borrow()))
            .ok_or_else(|| PositionLookupError::not_found(*position_id))
    }

    /// Returns a borrow of the position with the `position_id`.
    ///
    /// Prefer [`Self::try_position_ref`] in new native code.
    ///
    /// # Errors
    ///
    /// Returns an error if the position is not present in the cache.
    pub fn try_position(
        &self,
        position_id: &PositionId,
    ) -> Result<PositionRef<'_>, PositionLookupError> {
        self.try_position_ref(position_id)
    }

    /// Gets an exclusive write borrow of the position with the `position_id` (if found).
    ///
    /// Requires `&mut Cache` so cache writes are reachable only by privileged crates that hold
    /// `Rc<RefCell<Cache>>` directly. Adapter-facing code receives [`CacheView`], which only
    /// exposes immutable cache borrows and therefore cannot reach this method.
    ///
    /// While the returned [`PositionRefMut`] is alive, no other read or write of the same position
    /// is permitted. Drop the borrow before dispatching events or taking any other cache borrow
    /// that may re-enter the same position.
    #[must_use]
    pub fn position_mut(&mut self, position_id: &PositionId) -> Option<PositionRefMut<'_>> {
        self.positions
            .get(position_id)
            .map(|position_cell| PositionRefMut::new(position_cell.borrow_mut()))
    }

    /// Gets an owned copy of the position with the `position_id` (if found).
    ///
    /// Use when downstream needs an owned [`Position`] that crosses a boundary. The copy will not
    /// reflect later cache mutations.
    #[must_use]
    pub fn position_owned(&self, position_id: &PositionId) -> Option<Position> {
        self.positions
            .get(position_id)
            .map(|position_cell| position_cell.borrow().clone())
    }

    /// Returns a borrow of the position for the `client_order_id` (if found).
    #[must_use]
    pub fn position_for_order_ref(
        &self,
        client_order_id: &ClientOrderId,
    ) -> Option<PositionRef<'_>> {
        self.index
            .order_position
            .get(client_order_id)
            .and_then(|position_id| self.positions.get(position_id))
            .map(|position_cell| PositionRef::new(position_cell.borrow()))
    }

    /// Returns a borrow of the position for the `client_order_id` (if found).
    ///
    /// Prefer [`Self::position_for_order_ref`] in new native code.
    #[must_use]
    pub fn position_for_order(&self, client_order_id: &ClientOrderId) -> Option<PositionRef<'_>> {
        self.position_for_order_ref(client_order_id)
    }

    /// Returns a reference to the position ID for the `client_order_id` (if found).
    #[must_use]
    pub fn position_id(&self, client_order_id: &ClientOrderId) -> Option<&PositionId> {
        self.index.order_position.get(client_order_id)
    }

    /// Returns the proven virtual HEDGING flip IDs, with the current destination second.
    ///
    /// The route comes from the original fill and its causal split fragments, independently of
    /// the order's mutable position ID. Returns `None` when no virtual split is recorded.
    ///
    /// # Errors
    ///
    /// Returns an error if split history is incomplete, conflicts, or a correction leaves both
    /// sides of a split open.
    pub fn hedging_flip_position_ids(
        &self,
        client_order_id: ClientOrderId,
    ) -> anyhow::Result<Option<[PositionId; 2]>> {
        let Some(order) = self.order_ref(&client_order_id) else {
            return Ok(None);
        };

        let mut flip = None;
        let mut corrected_source = None;

        // Keep ordered fill checks linear in each position's replay history
        let mut replay_offsets = AHashMap::new();

        for event in order.events() {
            let OrderEventAny::Filled(fill) = event else {
                continue;
            };

            let Some(source_id) = fill.position_id.filter(PositionId::is_virtual) else {
                continue;
            };

            let source = self.position_ref(&source_id);
            if source.is_some() && self.oms_type(&source_id) != Some(OmsType::Hedging) {
                continue;
            }

            if source.as_ref().is_some_and(|source| {
                let offset = replay_offsets.entry(source_id).or_default();
                Self::replay_contains_fill(&source.replay_events, fill, offset)
            }) {
                continue;
            }

            let closing = source.as_ref().and_then(|source| {
                source
                    .fill_fragments(client_order_id, fill.trade_id)
                    .into_iter()
                    .find(|fragment| {
                        fragment.event_id == fill.event_id && fragment.last_qty < fill.last_qty
                    })
                    .map(|closing| (source, closing))
            });

            let Some((source, closing)) = closing else {
                let has_opening = self
                    .positions_refs(
                        None,
                        Some(&fill.instrument_id),
                        Some(&fill.strategy_id),
                        Some(&fill.account_id),
                        None,
                    )
                    .iter()
                    .any(|position| {
                        position.id != source_id
                            && position.id.is_virtual()
                            && self.oms_type(&position.id) == Some(OmsType::Hedging)
                            && position.replay_events.iter().any(|event| {
                                matches!(event, PositionReplayEvent::Filled(opening)
                                if opening.causation_id == Some(fill.event_id)
                                    && opening.client_order_id == client_order_id
                                    && opening.trade_id == fill.trade_id)
                            })
                    });

                anyhow::ensure!(
                    !has_opening,
                    "missing virtual flip source fragment for order {client_order_id}"
                );
                continue;
            };

            let (positions, corrected_closing) =
                self.hedging_flip_destination(fill, source, closing)?;
            anyhow::ensure!(
                flip.is_none() || corrected_source == Some(source.id),
                "ambiguous virtual flip history for order {client_order_id}"
            );
            flip = Some(positions);
            corrected_source = corrected_closing.then_some(source.id);
        }

        Ok(flip)
    }

    fn replay_contains_fill(
        events: &[PositionReplayEvent],
        fill: &OrderFilled,
        offset: &mut usize,
    ) -> bool {
        let matches_fill = |event: &PositionReplayEvent| matches!(event, PositionReplayEvent::Filled(fragment) if fragment == fill);

        if let Some(index) = events[*offset..].iter().position(matches_fill) {
            *offset += index + 1;
            return true;
        }

        // Repeated or reordered fills can still match before the cursor
        events[..*offset].iter().any(matches_fill)
    }

    fn hedging_flip_destination(
        &self,
        fill: &OrderFilled,
        source: &Position,
        closing: &OrderFilled,
    ) -> anyhow::Result<([PositionId; 2], bool)> {
        let positions = self.positions_refs(
            None,
            Some(&fill.instrument_id),
            Some(&fill.strategy_id),
            Some(&fill.account_id),
            None,
        );

        let mut openings = positions.iter().flat_map(|position| {
            position
                .replay_events
                .iter()
                .filter_map(move |event| match event {
                    PositionReplayEvent::Filled(opening)
                        if opening.causation_id == Some(fill.event_id)
                            && position.id != source.id =>
                    {
                        Some((position, opening))
                    }
                    _ => None,
                })
        });

        let (destination, opening) = openings.next().ok_or_else(|| {
            anyhow::anyhow!(
                "missing virtual flip fragment for order {}",
                fill.client_order_id
            )
        })?;

        anyhow::ensure!(
            openings.next().is_none(),
            "ambiguous virtual flip fragments for order {}",
            fill.client_order_id
        );
        anyhow::ensure!(
            source.instrument_id == fill.instrument_id
                && source.account_id == fill.account_id
                && source.strategy_id == fill.strategy_id
                && destination.id.is_virtual()
                && self.oms_type(&destination.id) == Some(OmsType::Hedging),
            "conflicting virtual flip positions for order {}",
            fill.client_order_id
        );
        let (expected_closing, expected_opening) =
            fill.split_for_position_flip(closing.last_qty, Some(destination.id), opening.event_id)?;
        anyhow::ensure!(
            *closing == expected_closing && *opening == expected_opening,
            "conflicting virtual flip provenance for order {}",
            fill.client_order_id
        );

        let corrected_closing = source.fill_voids.iter().any(|correction| {
            correction.event.client_order_id == fill.client_order_id
                && correction.event.trade_id == fill.trade_id
                && correction.voided_qty.non_zero()
        });

        if corrected_closing && source.is_open() {
            anyhow::ensure!(
                !destination.is_open(),
                "ambiguous corrected virtual flip for order {}",
                fill.client_order_id
            );
            return Ok(([destination.id, source.id], corrected_closing));
        }

        Ok(([source.id, destination.id], corrected_closing))
    }

    fn reindex_hedging_position(&mut self, client_order_id: ClientOrderId) {
        match self.hedging_flip_position_ids(client_order_id) {
            Ok(Some([_, position_id])) => {
                let position = self
                    .position_ref(&position_id)
                    .expect("proven position is cached");
                let venue = position.instrument_id.venue;
                let strategy_id = position.strategy_id;
                drop(position);

                if let Err(e) =
                    self.add_position_id(&position_id, &venue, &client_order_id, &strategy_id)
                {
                    log::error!("Failed to index virtual flip for order {client_order_id}: {e}");
                }
            }
            Ok(None) => {}
            Err(e) => {
                self.index.order_position.remove(&client_order_id);

                log::error!("Cannot index virtual flip for order {client_order_id}: {e}");
            }
        }
    }

    /// Returns borrows of all positions matching the optional filter parameters.
    ///
    /// Each [`PositionRef`] in the returned vector borrows its underlying cell; mutating any of
    /// those positions while the vector is alive will panic at runtime. Drop the vector before
    /// issuing writes.
    #[must_use]
    pub fn positions_refs(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> Vec<PositionRef<'_>> {
        let position_ids = self.position_ids(venue, instrument_id, strategy_id, account_id);
        self.get_positions_for_ids(&position_ids, side)
    }

    /// Returns borrows of all positions matching the optional filter parameters.
    ///
    /// Prefer [`Self::positions_refs`] in new native code.
    #[must_use]
    pub fn positions(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> Vec<PositionRef<'_>> {
        self.positions_refs(venue, instrument_id, strategy_id, account_id, side)
    }

    /// Returns borrows of all open positions matching the optional filter parameters.
    #[must_use]
    pub fn positions_open_refs(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> Vec<PositionRef<'_>> {
        let position_ids = self.position_open_ids(venue, instrument_id, strategy_id, account_id);
        self.get_positions_for_ids(&position_ids, side)
    }

    /// Returns borrows of all open positions matching the optional filter parameters.
    ///
    /// Prefer [`Self::positions_open_refs`] in new native code.
    #[must_use]
    pub fn positions_open(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> Vec<PositionRef<'_>> {
        self.positions_open_refs(venue, instrument_id, strategy_id, account_id, side)
    }

    /// Returns borrows of all closed positions matching the optional filter parameters.
    #[must_use]
    pub fn positions_closed_refs(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> Vec<PositionRef<'_>> {
        let position_ids = self.position_closed_ids(venue, instrument_id, strategy_id, account_id);
        self.get_positions_for_ids(&position_ids, side)
    }

    /// Returns borrows of all closed positions matching the optional filter parameters.
    ///
    /// Prefer [`Self::positions_closed_refs`] in new native code.
    #[must_use]
    pub fn positions_closed(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> Vec<PositionRef<'_>> {
        self.positions_closed_refs(venue, instrument_id, strategy_id, account_id, side)
    }

    /// Returns whether a position with the `position_id` exists.
    #[must_use]
    pub fn position_exists(&self, position_id: &PositionId) -> bool {
        self.index.positions.contains(position_id)
    }

    /// Returns whether a position with the `position_id` is open.
    #[must_use]
    pub fn is_position_open(&self, position_id: &PositionId) -> bool {
        self.index.positions_open.contains(position_id)
    }

    /// Returns whether a position with the `position_id` is closed.
    #[must_use]
    pub fn is_position_closed(&self, position_id: &PositionId) -> bool {
        self.index.positions_closed.contains(position_id)
    }

    /// Returns the count of all open positions.
    #[must_use]
    pub fn positions_open_count(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> usize {
        self.count_positions_in_bucket(
            &self.index.positions_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns the count of all closed positions.
    #[must_use]
    pub fn positions_closed_count(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> usize {
        self.count_positions_in_bucket(
            &self.index.positions_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns the count of all positions.
    #[must_use]
    pub fn positions_total_count(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> usize {
        self.count_positions_in_bucket(
            &self.index.positions,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns whether any open position matches the optional filter parameters.
    ///
    /// Short-circuits on the first match, avoiding the full intersection walk performed by
    /// [`Self::positions_open_count`]. Prefer this over `positions_open_count(...) > 0` when
    /// only existence matters.
    #[must_use]
    pub fn has_positions_open(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> bool {
        self.any_positions_in_bucket(
            &self.index.positions_open,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns whether any closed position matches the optional filter parameters.
    #[must_use]
    pub fn has_positions_closed(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> bool {
        self.any_positions_in_bucket(
            &self.index.positions_closed,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    /// Returns whether any position (open or closed) matches the optional filter parameters.
    #[must_use]
    pub fn has_positions(
        &self,
        venue: Option<&Venue>,
        instrument_id: Option<&InstrumentId>,
        strategy_id: Option<&StrategyId>,
        account_id: Option<&AccountId>,
        side: Option<PositionSide>,
    ) -> bool {
        self.any_positions_in_bucket(
            &self.index.positions,
            venue,
            instrument_id,
            strategy_id,
            account_id,
            side,
        )
    }

    // -- STRATEGY QUERIES ------------------------------------------------------------------------

    /// Gets a reference to the strategy ID for the `client_order_id` (if found).
    #[must_use]
    pub fn strategy_id_for_order(&self, client_order_id: &ClientOrderId) -> Option<&StrategyId> {
        self.index.order_strategy.get(client_order_id)
    }

    /// Gets a reference to the strategy ID for the `position_id` (if found).
    #[must_use]
    pub fn strategy_id_for_position(&self, position_id: &PositionId) -> Option<&StrategyId> {
        self.index.position_strategy.get(position_id)
    }

    // -- GENERAL ---------------------------------------------------------------------------------

    /// Gets a reference to the general value for the `key` (if found).
    ///
    /// # Errors
    ///
    /// Returns an error if the `key` is invalid.
    pub fn get(&self, key: &str) -> anyhow::Result<Option<&Bytes>> {
        check_valid_string_utf8(key, stringify!(key))?;

        Ok(self.general.get(key))
    }

    // -- DATA QUERIES ----------------------------------------------------------------------------

    /// Returns the price for the `instrument_id` and `price_type` (if found).
    ///
    /// For `Mid`, returns `None` if no quote is cached or either price is a sentinel.
    /// For quote precision `p`, the midpoint has precision `p + 1` when exactly representable,
    /// otherwise `p`, rounded half-even if necessary. The fallback applies when the precision
    /// limit or raw range rules out `p + 1`, and when `p` is the maximum float precision (16),
    /// so a midpoint of a float-convertible quote stays float-convertible.
    #[must_use]
    pub fn price(&self, instrument_id: &InstrumentId, price_type: PriceType) -> Option<Price> {
        match price_type {
            PriceType::Bid => self
                .quotes
                .get(instrument_id)
                .and_then(|quotes| quotes.front().map(|quote| quote.bid_price)),
            PriceType::Ask => self
                .quotes
                .get(instrument_id)
                .and_then(|quotes| quotes.front().map(|quote| quote.ask_price)),
            PriceType::Mid => self.quotes.get(instrument_id).and_then(|quotes| {
                let quote = quotes.front()?;
                let bid = quote.bid_price;
                let ask = quote.ask_price;
                if bid.is_undefined() || bid.is_error() || ask.is_undefined() || ask.is_error() {
                    return None;
                }

                // Also rejects `ERROR_PRICE`, whose precision is 255
                let precision = bid.precision;
                check_fixed_precision(precision).ok()?;

                #[allow(
                    clippy::useless_conversion,
                    reason = "i128::from is a widening conversion when PriceRaw is i64"
                )]
                let (mut bid_raw, mut ask_raw) = (i128::from(bid.raw()), i128::from(ask.raw()));

                if precision < FIXED_PRECISION {
                    let factor = 10_i128.pow(u32::from(FIXED_PRECISION - precision));
                    bid_raw = (bid_raw / factor) * factor;
                    ask_raw = (ask_raw / factor) * factor;
                }
                let sum = bid_raw.checked_add(ask_raw)?;

                if precision < FIXED_PRECISION {
                    let raw = PriceRaw::try_from(sum / 2).ok()?;
                    return Price::from_raw_checked(raw, precision + 1).ok();
                }

                if precision != MAX_FLOAT_PRECISION
                    && let Some(price) = sum
                        .checked_mul(5)
                        .and_then(|raw| PriceRaw::try_from(raw).ok())
                        .and_then(|raw| Price::from_raw_checked(raw, precision + 1).ok())
                {
                    return Some(price);
                }

                let mut raw = sum / 2;
                if sum % 2 != 0 && raw % 2 != 0 {
                    raw += sum.signum();
                }
                let raw = PriceRaw::try_from(raw).ok()?;
                Price::from_raw_checked(raw, precision).ok()
            }),
            PriceType::Last => self
                .trades
                .get(instrument_id)
                .and_then(|trades| trades.front().map(|trade| trade.price)),
            PriceType::Mark => self
                .mark_prices
                .get(instrument_id)
                .and_then(|marks| marks.front().map(|mark| mark.value)),
        }
    }

    /// Returns cached quotes for the `instrument_id` in reverse insertion order.
    ///
    /// Index 0 is the most recently added quote. Entries are not sorted by timestamp.
    /// Returns `None` when no quotes are cached for the instrument.
    #[must_use]
    pub fn quotes(&self, instrument_id: &InstrumentId) -> Option<Vec<QuoteTick>> {
        self.quotes
            .get(instrument_id)
            .map(|quotes| quotes.iter().copied().collect())
    }

    /// Returns cached trades for the `instrument_id` in reverse insertion order.
    ///
    /// Index 0 is the most recently added trade. Entries are not sorted by timestamp.
    /// Returns `None` when no trades are cached for the instrument.
    #[must_use]
    pub fn trades(&self, instrument_id: &InstrumentId) -> Option<Vec<TradeTick>> {
        self.trades
            .get(instrument_id)
            .map(|trades| trades.iter().copied().collect())
    }

    /// Returns cached mark price updates for the `instrument_id` in reverse insertion order.
    ///
    /// Index 0 is the most recently added update. Entries are not sorted by timestamp.
    /// Returns `None` when no mark price updates are cached for the instrument.
    #[must_use]
    pub fn mark_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<MarkPriceUpdate>> {
        self.mark_prices
            .get(instrument_id)
            .map(|mark_prices| mark_prices.iter().copied().collect())
    }

    /// Returns cached index price updates for the `instrument_id` in reverse insertion order.
    ///
    /// Index 0 is the most recently added update. Entries are not sorted by timestamp.
    /// Returns `None` when no index price updates are cached for the instrument.
    #[must_use]
    pub fn index_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<IndexPriceUpdate>> {
        self.index_prices
            .get(instrument_id)
            .map(|index_prices| index_prices.iter().copied().collect())
    }

    /// Returns cached funding rate updates for the `instrument_id` in reverse insertion order.
    ///
    /// Index 0 is the most recently added update. Entries are not sorted by timestamp.
    /// Returns `None` when no funding rate updates are cached for the instrument.
    #[must_use]
    pub fn funding_rates(&self, instrument_id: &InstrumentId) -> Option<Vec<FundingRateUpdate>> {
        self.funding_rates
            .get(instrument_id)
            .map(|funding_rates| funding_rates.iter().copied().collect())
    }

    /// Returns cached instrument status updates for the `instrument_id` in reverse insertion order.
    ///
    /// Index 0 is the most recently added update. Entries are not sorted by timestamp.
    /// Returns `None` when no instrument status updates are cached for the instrument.
    #[must_use]
    pub fn instrument_statuses(
        &self,
        instrument_id: &InstrumentId,
    ) -> Option<Vec<InstrumentStatus>> {
        self.instrument_statuses
            .get(instrument_id)
            .map(|statuses| statuses.iter().copied().collect())
    }

    /// Returns cached bars for the `bar_type` in reverse chronological order by `ts_event`.
    ///
    /// Index 0 is the most recent bar. Returns `None` when no bars are cached for the bar type.
    #[must_use]
    pub fn bars(&self, bar_type: &BarType) -> Option<Vec<Bar>> {
        self.bars
            .get(bar_type)
            .map(|bars| bars.iter().copied().collect())
    }

    /// Returns the current order book for the `instrument_id`, or `None` if absent.
    ///
    /// The cache holds the current book state, not a history of book updates.
    #[must_use]
    pub fn order_book(&self, instrument_id: &InstrumentId) -> Option<&OrderBook> {
        self.books.get(instrument_id)
    }

    /// Gets a reference to the order book for the `instrument_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the order book is not present in the cache.
    pub fn try_order_book(
        &self,
        instrument_id: &InstrumentId,
    ) -> Result<&OrderBook, OrderBookLookupError> {
        self.books
            .get(instrument_id)
            .ok_or_else(|| OrderBookLookupError::not_found(*instrument_id))
    }

    /// Gets a reference to the order book for the `instrument_id`.
    #[must_use]
    pub fn order_book_mut(&mut self, instrument_id: &InstrumentId) -> Option<&mut OrderBook> {
        self.books.get_mut(instrument_id)
    }

    /// Gets a reference to the own order book for the `instrument_id`.
    #[must_use]
    pub fn own_order_book(&self, instrument_id: &InstrumentId) -> Option<&OwnOrderBook> {
        self.own_books.get(instrument_id)
    }

    /// Gets a reference to the own order book for the `instrument_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the own order book is not present in the cache.
    pub fn try_own_order_book(
        &self,
        instrument_id: &InstrumentId,
    ) -> Result<&OwnOrderBook, OwnOrderBookLookupError> {
        self.own_books
            .get(instrument_id)
            .ok_or_else(|| OwnOrderBookLookupError::not_found(*instrument_id))
    }

    /// Gets a reference to the own order book for the `instrument_id`.
    #[must_use]
    pub fn own_order_book_mut(
        &mut self,
        instrument_id: &InstrumentId,
    ) -> Option<&mut OwnOrderBook> {
        self.own_books.get_mut(instrument_id)
    }

    /// Returns the most recently added quote for the `instrument_id`, or `None` if absent.
    #[must_use]
    pub fn quote(&self, instrument_id: &InstrumentId) -> Option<&QuoteTick> {
        self.quotes
            .get(instrument_id)
            .and_then(|quotes| quotes.front())
    }

    /// Returns the quote at `index` for the `instrument_id`.
    ///
    /// Index 0 is the most recently added quote; index 1 is the previous insertion.
    /// Entries are not sorted by timestamp. Returns `None` when no quote exists at the index.
    #[must_use]
    pub fn quote_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&QuoteTick> {
        self.quotes
            .get(instrument_id)
            .and_then(|quotes| quotes.get(index))
    }

    /// Returns the most recently added trade for the `instrument_id`, or `None` if absent.
    #[must_use]
    pub fn trade(&self, instrument_id: &InstrumentId) -> Option<&TradeTick> {
        self.trades
            .get(instrument_id)
            .and_then(|trades| trades.front())
    }

    /// Returns the trade at `index` for the `instrument_id`.
    ///
    /// Index 0 is the most recently added trade; index 1 is the previous insertion.
    /// Entries are not sorted by timestamp. Returns `None` when no trade exists at the index.
    #[must_use]
    pub fn trade_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&TradeTick> {
        self.trades
            .get(instrument_id)
            .and_then(|trades| trades.get(index))
    }

    /// Returns the most recently added mark price update for the `instrument_id`, or `None` if absent.
    #[must_use]
    pub fn mark_price(&self, instrument_id: &InstrumentId) -> Option<&MarkPriceUpdate> {
        self.mark_prices
            .get(instrument_id)
            .and_then(|mark_prices| mark_prices.front())
    }

    /// Returns the most recently added index price update for the `instrument_id`, or `None` if absent.
    #[must_use]
    pub fn index_price(&self, instrument_id: &InstrumentId) -> Option<&IndexPriceUpdate> {
        self.index_prices
            .get(instrument_id)
            .and_then(|index_prices| index_prices.front())
    }

    /// Returns the most recently added funding rate update for the `instrument_id`, or `None` if absent.
    #[must_use]
    pub fn funding_rate(&self, instrument_id: &InstrumentId) -> Option<&FundingRateUpdate> {
        self.funding_rates
            .get(instrument_id)
            .and_then(|funding_rates| funding_rates.front())
    }

    /// Returns the most recently added instrument status for the `instrument_id`, or `None` if absent.
    #[must_use]
    pub fn instrument_status(&self, instrument_id: &InstrumentId) -> Option<&InstrumentStatus> {
        self.instrument_statuses
            .get(instrument_id)
            .and_then(|statuses| statuses.front())
    }

    /// Returns the close cached for `instrument_id`, if present.
    ///
    /// A later insertion replaces the cached close for the instrument; no history is kept.
    #[must_use]
    pub fn instrument_close(&self, instrument_id: &InstrumentId) -> Option<&InstrumentClose> {
        self.instrument_closes.get(instrument_id)
    }

    /// Returns references to all instrument IDs with a cached close.
    #[must_use]
    pub fn instrument_close_ids(&self) -> Vec<&InstrumentId> {
        self.instrument_closes.keys().collect()
    }

    /// Returns the most recent bar by `ts_event` for the `bar_type`, or `None` if absent.
    #[must_use]
    pub fn bar(&self, bar_type: &BarType) -> Option<&Bar> {
        self.bars.get(bar_type).and_then(|bars| bars.front())
    }

    /// Returns the bar at `index` for the `bar_type`.
    ///
    /// Bars are ordered by `ts_event`, newest first. Index 0 is the most recent bar;
    /// index 1 is the previous bar. Returns `None` when no bar exists at the index.
    #[must_use]
    pub fn bar_at_index(&self, bar_type: &BarType, index: usize) -> Option<&Bar> {
        self.bars.get(bar_type).and_then(|bars| bars.get(index))
    }

    /// Gets the order book update count for the `instrument_id`.
    #[must_use]
    pub fn book_update_count(&self, instrument_id: &InstrumentId) -> usize {
        self.books
            .get(instrument_id)
            .map_or(0, |book| book.update_count) as usize
    }

    /// Gets the quote tick count for the `instrument_id`.
    #[must_use]
    pub fn quote_count(&self, instrument_id: &InstrumentId) -> usize {
        self.quotes
            .get(instrument_id)
            .map_or(0, BoundedVecDeque::len)
    }

    /// Gets the trade tick count for the `instrument_id`.
    #[must_use]
    pub fn trade_count(&self, instrument_id: &InstrumentId) -> usize {
        self.trades
            .get(instrument_id)
            .map_or(0, BoundedVecDeque::len)
    }

    /// Gets the mark price update count for the `instrument_id`.
    #[must_use]
    pub fn mark_price_count(&self, instrument_id: &InstrumentId) -> usize {
        self.mark_prices
            .get(instrument_id)
            .map_or(0, BoundedVecDeque::len)
    }

    /// Gets the index price update count for the `instrument_id`.
    #[must_use]
    pub fn index_price_count(&self, instrument_id: &InstrumentId) -> usize {
        self.index_prices
            .get(instrument_id)
            .map_or(0, BoundedVecDeque::len)
    }

    /// Gets the funding rate update count for the `instrument_id`.
    #[must_use]
    pub fn funding_rate_count(&self, instrument_id: &InstrumentId) -> usize {
        self.funding_rates
            .get(instrument_id)
            .map_or(0, BoundedVecDeque::len)
    }

    /// Gets the instrument status update count for the `instrument_id`.
    #[must_use]
    pub fn instrument_status_count(&self, instrument_id: &InstrumentId) -> usize {
        self.instrument_statuses
            .get(instrument_id)
            .map_or(0, BoundedVecDeque::len)
    }

    /// Gets the bar count for the `instrument_id`.
    #[must_use]
    pub fn bar_count(&self, bar_type: &BarType) -> usize {
        self.bars.get(bar_type).map_or(0, BoundedVecDeque::len)
    }

    /// Returns whether the cache contains an order book for the `instrument_id`.
    #[must_use]
    pub fn has_order_book(&self, instrument_id: &InstrumentId) -> bool {
        self.books.contains_key(instrument_id)
    }

    /// Returns whether the cache contains quotes for the `instrument_id`.
    #[must_use]
    pub fn has_quote_ticks(&self, instrument_id: &InstrumentId) -> bool {
        self.quote_count(instrument_id) > 0
    }

    /// Returns whether the cache contains trades for the `instrument_id`.
    #[must_use]
    pub fn has_trade_ticks(&self, instrument_id: &InstrumentId) -> bool {
        self.trade_count(instrument_id) > 0
    }

    /// Returns whether the cache contains mark price updates for the `instrument_id`.
    #[must_use]
    pub fn has_mark_prices(&self, instrument_id: &InstrumentId) -> bool {
        self.mark_price_count(instrument_id) > 0
    }

    /// Returns whether the cache contains index price updates for the `instrument_id`.
    #[must_use]
    pub fn has_index_prices(&self, instrument_id: &InstrumentId) -> bool {
        self.index_price_count(instrument_id) > 0
    }

    /// Returns whether the cache contains funding rate updates for the `instrument_id`.
    #[must_use]
    pub fn has_funding_rates(&self, instrument_id: &InstrumentId) -> bool {
        self.funding_rate_count(instrument_id) > 0
    }

    /// Returns whether the cache contains instrument status updates for the `instrument_id`.
    #[must_use]
    pub fn has_instrument_statuses(&self, instrument_id: &InstrumentId) -> bool {
        self.instrument_status_count(instrument_id) > 0
    }

    /// Returns whether the cache contains a close for the `instrument_id`.
    #[must_use]
    pub fn has_instrument_close(&self, instrument_id: &InstrumentId) -> bool {
        self.instrument_closes.contains_key(instrument_id)
    }

    /// Returns whether the cache contains bars for the `bar_type`.
    #[must_use]
    pub fn has_bars(&self, bar_type: &BarType) -> bool {
        self.bar_count(bar_type) > 0
    }

    #[must_use]
    pub fn get_xrate(
        &self,
        venue: Venue,
        from_currency: Currency,
        to_currency: Currency,
        price_type: PriceType,
    ) -> Option<Decimal> {
        match self.try_get_xrate(venue, from_currency, to_currency, price_type) {
            Ok(rate) => rate,
            Err(e) => {
                log::error!("Failed to calculate xrate: {e}");
                None
            }
        }
    }

    /// Tries to calculate the exchange rate without logging calculation errors.
    ///
    /// # Errors
    ///
    /// Returns an error if the cached quotes cannot form a valid exchange rate calculation.
    pub fn try_get_xrate(
        &self,
        venue: Venue,
        from_currency: Currency,
        to_currency: Currency,
        price_type: PriceType,
    ) -> anyhow::Result<Option<Decimal>> {
        if from_currency == to_currency {
            // When the source and target currencies are identical,
            // no conversion is needed; return an exchange rate of one.
            return Ok(Some(Decimal::ONE));
        }

        let (bid_quote, ask_quote) = self.build_quote_table(&venue);

        get_exchange_rate(
            from_currency.code,
            to_currency.code,
            price_type,
            bid_quote,
            ask_quote,
        )
    }

    fn build_quote_table(
        &self,
        venue: &Venue,
    ) -> (AHashMap<Ustr, Decimal>, AHashMap<Ustr, Decimal>) {
        let mut bid_quotes = AHashMap::new();
        let mut ask_quotes = AHashMap::new();
        let mut quote_sources = AHashMap::new();
        let mut bar_quotes = None;
        let mut pair_buffer = String::new();

        for (instrument_id, instrument) in &self.instruments {
            if instrument_id.venue != *venue {
                continue;
            }

            let Some(base_currency) = instrument.base_currency() else {
                continue;
            };

            let (bid_price, ask_price) = if let Some(ticks) = self.quotes.get(instrument_id) {
                if let Some(tick) = ticks.front() {
                    (tick.bid_price, tick.ask_price)
                } else {
                    continue; // Empty ticks vector
                }
            } else {
                let quotes = bar_quotes.get_or_insert_with(|| self.build_bar_quote_table(venue));
                match (
                    quotes.get(&(*instrument_id, PriceType::Bid)),
                    quotes.get(&(*instrument_id, PriceType::Ask)),
                ) {
                    (Some((_, bid_bar)), Some((_, ask_bar))) => (bid_bar.close, ask_bar.close),
                    _ => continue,
                }
            };

            let base = base_currency.code.as_str();
            let quote = instrument.quote_currency().code.as_str();
            pair_buffer.clear();
            pair_buffer.reserve(base.len() + 1 + quote.len());
            pair_buffer.push_str(base);
            pair_buffer.push('/');
            pair_buffer.push_str(quote);
            let pair = Ustr::from(pair_buffer.as_str());
            let preference = (
                bid_price.is_positive() && ask_price.is_positive(),
                instrument.instrument_class() == InstrumentClass::Spot,
                Reverse(*instrument_id),
            );

            if quote_sources
                .get(&pair)
                .is_some_and(|current| current >= &preference)
            {
                continue;
            }

            bid_quotes.insert(pair, bid_price.as_decimal());
            ask_quotes.insert(pair, ask_price.as_decimal());
            quote_sources.insert(pair, preference);
        }

        (bid_quotes, ask_quotes)
    }

    fn build_bar_quote_table(
        &self,
        venue: &Venue,
    ) -> AHashMap<(InstrumentId, PriceType), (&BarType, &Bar)> {
        let mut quotes: AHashMap<_, (&BarType, &Bar)> = AHashMap::new();

        for (bar_type, bars) in &self.bars {
            let instrument_id = bar_type.instrument_id();
            let price_type = bar_type.spec().price_type;

            if instrument_id.venue != *venue
                || !matches!(price_type, PriceType::Bid | PriceType::Ask)
            {
                continue;
            }

            let Some(bar) = bars.front() else {
                continue;
            };

            // Select the newest front bar per side, breaking timestamp ties by bar type
            quotes
                .entry((instrument_id, price_type))
                .and_modify(|current| {
                    if (current.1.ts_init, current.0) < (bar.ts_init, bar_type) {
                        *current = (bar_type, bar);
                    }
                })
                .or_insert((bar_type, bar));
        }

        quotes
    }

    /// Returns the mark exchange rate for the given currency pair, or `None` if not set.
    #[must_use]
    pub fn get_mark_xrate(&self, from_currency: Currency, to_currency: Currency) -> Option<f64> {
        self.mark_xrates.get(&(from_currency, to_currency)).copied()
    }

    /// Sets the mark exchange rate for the given currency pair and automatically sets the inverse rate.
    ///
    /// # Panics
    ///
    /// Panics if `xrate` is not positive.
    pub fn set_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency, xrate: f64) {
        assert!(xrate > 0.0, "xrate was zero");
        self.mark_xrates.insert((from_currency, to_currency), xrate);
        self.mark_xrates
            .insert((to_currency, from_currency), 1.0 / xrate);
    }

    /// Clears the mark exchange rate for the given currency pair direction.
    ///
    /// Removes only the `(from_currency, to_currency)` entry; the inverse rate written
    /// by [`Self::set_mark_xrate`] is retained until cleared separately or
    /// [`Self::clear_mark_xrates`] is called.
    pub fn clear_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency) {
        let _ = self.mark_xrates.remove(&(from_currency, to_currency));
    }

    /// Clears all mark exchange rates.
    pub fn clear_mark_xrates(&mut self) {
        self.mark_xrates.clear();
    }

    /// Returns a reference to the currency for the `code` (if found).
    #[must_use]
    pub fn currency(&self, code: &Ustr) -> Option<&Currency> {
        self.currencies.get(code)
    }

    /// Returns a reference to the currency for the `code`.
    ///
    /// # Errors
    ///
    /// Returns an error if the currency is not present in the cache.
    pub fn try_currency(&self, code: &Ustr) -> Result<&Currency, CurrencyLookupError> {
        self.currencies
            .get(code)
            .ok_or_else(|| CurrencyLookupError::not_found(*code))
    }

    // -- INSTRUMENT QUERIES ----------------------------------------------------------------------

    /// Returns a reference to the instrument for the `instrument_id` (if found).
    #[must_use]
    pub fn instrument(&self, instrument_id: &InstrumentId) -> Option<&InstrumentAny> {
        self.instruments.get(instrument_id)
    }

    /// Returns a reference to the instrument for the `instrument_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the instrument is not present in the cache.
    pub fn try_instrument(
        &self,
        instrument_id: &InstrumentId,
    ) -> Result<&InstrumentAny, InstrumentLookupError> {
        self.instruments
            .get(instrument_id)
            .ok_or_else(|| InstrumentLookupError::not_found(*instrument_id))
    }

    /// Returns references to all instrument IDs for the `venue`.
    #[must_use]
    pub fn instrument_ids(&self, venue: Option<&Venue>) -> Vec<&InstrumentId> {
        match venue {
            Some(v) => self.instruments.keys().filter(|i| &i.venue == v).collect(),
            None => self.instruments.keys().collect(),
        }
    }

    /// Returns references to all instruments for the `venue`.
    #[must_use]
    pub fn instruments(&self, venue: &Venue, underlying: Option<&Ustr>) -> Vec<&InstrumentAny> {
        self.instruments
            .values()
            .filter(|i| &i.id().venue == venue)
            .filter(|i| underlying.is_none_or(|u| i.underlying() == Some(*u)))
            .collect()
    }

    /// Returns references to all instruments for the `venue` whose underlying
    /// equals `root` and whose [`InstrumentClass`] equals `class`.
    ///
    /// Use when expanding a parent-symbol subscription: filtering by class as
    /// well as root prevents leaves of a different class (e.g. options when
    /// the user asked for futures, or vice versa) from being pulled in.
    #[must_use]
    pub fn instruments_by_parent(
        &self,
        venue: &Venue,
        root: &Ustr,
        class: InstrumentClass,
    ) -> Vec<&InstrumentAny> {
        self.instruments
            .values()
            .filter(|i| &i.id().venue == venue)
            .filter(|i| i.underlying() == Some(*root))
            .filter(|i| i.instrument_class() == class)
            .collect()
    }

    /// Returns references to all bar types contained in the cache.
    #[must_use]
    pub fn bar_types(
        &self,
        instrument_id: Option<&InstrumentId>,
        price_type: Option<&PriceType>,
        aggregation_source: AggregationSource,
    ) -> Vec<&BarType> {
        let mut bar_types = self
            .bars
            .keys()
            .filter(|bar_type| bar_type.aggregation_source() == aggregation_source)
            .collect::<Vec<&BarType>>();

        if let Some(instrument_id) = instrument_id {
            bar_types.retain(|bar_type| bar_type.instrument_id() == *instrument_id);
        }

        if let Some(price_type) = price_type {
            bar_types.retain(|bar_type| &bar_type.spec().price_type == price_type);
        }

        bar_types
    }

    // -- SYNTHETIC QUERIES -----------------------------------------------------------------------

    /// Returns a reference to the synthetic instrument for the `instrument_id` (if found).
    #[must_use]
    pub fn synthetic(&self, instrument_id: &InstrumentId) -> Option<&SyntheticInstrument> {
        self.synthetics.get(instrument_id)
    }

    /// Returns a reference to the synthetic instrument for the `instrument_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the synthetic instrument is not present in the cache.
    pub fn try_synthetic(
        &self,
        instrument_id: &InstrumentId,
    ) -> Result<&SyntheticInstrument, SyntheticInstrumentLookupError> {
        self.synthetics
            .get(instrument_id)
            .ok_or_else(|| SyntheticInstrumentLookupError::not_found(*instrument_id))
    }

    /// Returns references to instrument IDs for all synthetic instruments contained in the cache.
    #[must_use]
    pub fn synthetic_ids(&self) -> Vec<&InstrumentId> {
        self.synthetics.keys().collect()
    }

    /// Returns references to all synthetic instruments contained in the cache.
    #[must_use]
    pub fn synthetics(&self) -> Vec<&SyntheticInstrument> {
        self.synthetics.values().collect()
    }

    // -- ACCOUNT QUERIES -----------------------------------------------------------------------

    /// Returns a borrow of the account for the `account_id` (if found).
    #[must_use]
    pub fn account_ref(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
        self.accounts
            .get(account_id)
            .map(|account_cell| AccountRef::new(account_cell.borrow()))
    }

    /// Returns a borrow of the account for the `account_id` (if found).
    ///
    /// Prefer [`Self::account_ref`] in new native code.
    #[must_use]
    pub fn account(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
        self.account_ref(account_id)
    }

    /// Returns a borrow of the account for the `account_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the account is not present in the cache.
    pub fn try_account_ref(
        &self,
        account_id: &AccountId,
    ) -> Result<AccountRef<'_>, AccountLookupError> {
        self.accounts
            .get(account_id)
            .map(|account_cell| AccountRef::new(account_cell.borrow()))
            .ok_or_else(|| AccountLookupError::not_found(*account_id))
    }

    /// Returns a borrow of the account for the `account_id`.
    ///
    /// Prefer [`Self::try_account_ref`] in new native code.
    ///
    /// # Errors
    ///
    /// Returns an error if the account is not present in the cache.
    pub fn try_account(
        &self,
        account_id: &AccountId,
    ) -> Result<AccountRef<'_>, AccountLookupError> {
        self.try_account_ref(account_id)
    }

    /// Gets an exclusive write borrow of the account with the `account_id` (if found).
    ///
    /// Requires `&mut Cache` so cache writes are reachable only by privileged crates that hold
    /// `Rc<RefCell<Cache>>` directly. Adapter-facing code receives [`CacheView`], which only
    /// exposes immutable cache borrows and therefore cannot reach this method.
    ///
    /// While the returned [`AccountRefMut`] is alive, no other read or write of the same account
    /// is permitted. Drop the borrow before dispatching events or taking any other cache borrow
    /// that may re-enter the same account.
    #[must_use]
    pub fn account_mut(&mut self, account_id: &AccountId) -> Option<AccountRefMut<'_>> {
        self.accounts
            .get(account_id)
            .map(|account_cell| AccountRefMut::new(account_cell.borrow_mut()))
    }

    /// Gets an owned snapshot of the account with the `account_id` when present and not mutably
    /// borrowed.
    ///
    /// Use when downstream needs an owned [`AccountAny`] that crosses a boundary. The snapshot
    /// will not reflect later cache mutations.
    #[must_use]
    pub fn account_owned(&self, account_id: &AccountId) -> Option<AccountAny> {
        self.accounts.get(account_id).and_then(|account_cell| {
            account_cell
                .try_borrow()
                .ok()
                .map(|account| account.clone())
        })
    }

    /// Returns a borrow of the account for the `venue` (if found).
    ///
    /// Returns `None` when more than one account is issued under the `venue`; look those
    /// accounts up by account ID instead.
    #[must_use]
    pub fn account_for_venue(&self, venue: &Venue) -> Option<AccountRef<'_>> {
        self.account_id(venue)
            .and_then(|account_id| self.accounts.get(account_id))
            .map(|account_cell| AccountRef::new(account_cell.borrow()))
    }

    /// Returns an owned snapshot of the account for the `venue` (if found).
    ///
    /// Use when downstream needs an owned [`AccountAny`] that crosses a boundary. The snapshot
    /// will not reflect later cache mutations. Returns `None` when more than one account is
    /// issued under the `venue`.
    #[must_use]
    pub fn account_for_venue_owned(&self, venue: &Venue) -> Option<AccountAny> {
        self.account_id(venue)
            .and_then(|account_id| self.accounts.get(account_id))
            .map(|account_cell| account_cell.borrow().clone())
    }

    /// Returns a reference to the account ID for the `venue` (if found).
    ///
    /// Returns `None` when more than one account is issued under the `venue`.
    #[must_use]
    pub fn account_id(&self, venue: &Venue) -> Option<&AccountId> {
        let mut account_ids = self.index.venue_accounts.get(venue)?.iter();
        let account_id = account_ids.next()?;
        account_ids.next().is_none().then_some(account_id)
    }

    /// Returns borrows of all accounts for the `account_id`.
    ///
    /// Each [`AccountRef`] in the returned vector borrows its underlying cell; mutating any of
    /// those accounts while the vector is alive will panic at runtime. Drop the vector before
    /// issuing writes.
    #[must_use]
    pub fn accounts(&self, account_id: &AccountId) -> Vec<AccountRef<'_>> {
        self.accounts
            .values()
            .filter(|account_cell| &account_cell.borrow().id() == account_id)
            .map(|account_cell| AccountRef::new(account_cell.borrow()))
            .collect()
    }

    /// Returns owned copies of every account in the cache.
    #[must_use]
    pub fn accounts_all_owned(&self) -> Vec<AccountAny> {
        self.accounts
            .values()
            .map(|account_cell| account_cell.borrow().clone())
            .collect()
    }

    /// Updates the own order book with an order.
    ///
    /// This method adds, updates, or removes an order from the own order book
    /// based on the order's current state.
    ///
    /// Orders without prices (MARKET, etc.) are skipped as they cannot be
    /// represented in own books.
    pub fn update_own_order_book(&mut self, order: &OrderAny) {
        if !order.has_price() {
            return;
        }

        let instrument_id = order.instrument_id();

        if !self.own_books.contains_key(&instrument_id) {
            if order.is_closed() {
                return;
            }

            self.own_books
                .insert(instrument_id, OwnOrderBook::new(instrument_id));
        }

        let Some(own_book) = self.own_books.get_mut(&instrument_id) else {
            return;
        };

        let own_book_order = order.to_own_book_order();

        if order.is_closed() {
            if let Err(e) = own_book.delete(own_book_order) {
                log::debug!(
                    "Failed to delete order {} from own book: {e}",
                    order.client_order_id(),
                );
            } else {
                log::debug!("Deleted order {} from own book", order.client_order_id());
            }
        } else {
            // Add or update the order in the own book
            if let Err(e) = own_book.update(own_book_order) {
                log::debug!(
                    "Failed to update order {} in own book: {e}; inserting instead",
                    order.client_order_id(),
                );
                own_book.add(own_book_order);
            }
            log::debug!("Updated order {} in own book", order.client_order_id());
        }
    }

    /// Force removal of an order from own order books and clean up all indexes.
    ///
    /// This method is used when order event application fails and we need to ensure
    /// terminal orders are properly cleaned up from own books and all relevant indexes.
    /// Replicates the index cleanup that `update_order` performs for closed orders.
    pub fn force_remove_from_own_order_book(&mut self, client_order_id: &ClientOrderId) {
        let Some(order_cell) = self.orders.get(client_order_id) else {
            return;
        };
        let order = order_cell.borrow();
        let instrument_id = order.instrument_id();
        let own_book_order = if order.has_price() {
            Some(order.to_own_book_order())
        } else {
            None
        };
        drop(order);

        self.index.orders_open.remove(client_order_id);
        self.index.orders_pending_cancel.remove(client_order_id);
        self.index.orders_inflight.remove(client_order_id);
        self.index.orders_emulated.remove(client_order_id);
        self.index.orders_active_local.remove(client_order_id);

        if let Some(own_book) = self.own_books.get_mut(&instrument_id)
            && let Some(own_book_order) = own_book_order
        {
            if let Err(e) = own_book.delete(own_book_order) {
                log::debug!("Could not force delete {client_order_id} from own book: {e}");
            } else {
                log::debug!("Force deleted {client_order_id} from own book");
            }
        }

        self.index.orders_closed.insert(*client_order_id);
    }

    /// Audit all own order books against active order indexes.
    ///
    /// Ensures orders absent from the open, inflight, and active-local indexes are removed from
    /// own order books.
    pub fn audit_own_order_books(&mut self) {
        log::debug!("Starting own books audit");
        let start = std::time::Instant::now();

        let valid_order_ids: AHashSet<ClientOrderId> = self
            .index
            .orders_open
            .iter()
            .chain(&self.index.orders_inflight)
            .chain(&self.index.orders_active_local)
            .copied()
            .collect();

        for own_book in self.own_books.values_mut() {
            own_book.audit_open_orders(&valid_order_ids);
        }

        log::debug!("Completed own books audit in {:?}", start.elapsed());
    }
}

const POSITION_OMS_KEY_PREFIX: &str = "position_oms:";

fn position_oms_key(position_id: PositionId) -> String {
    format!("{POSITION_OMS_KEY_PREFIX}{position_id}")
}

fn resolve_position_oms(position: &Position, oms_type: OmsType) -> OmsType {
    if oms_type == OmsType::Unspecified
        && position.id.as_str() == format!("{}-{}", position.instrument_id, position.strategy_id)
    {
        OmsType::Netting
    } else {
        oms_type
    }
}

/// Inserts `bar` into the newest-first bars deque for a `bar_type`.
///
/// A newer bar is pushed, an older `ts_event` is skipped, and an equal
/// `ts_event` replaces the front bar for time bars.
fn insert_bar(bars: &mut BoundedVecDeque<Bar>, bar: Bar) {
    match bars.front() {
        None => bars.push_front(bar),
        Some(front) => match bar.ts_event.cmp(&front.ts_event) {
            Ordering::Greater => bars.push_front(bar),
            Ordering::Equal => {
                if bar.bar_type.spec().is_time_aggregated() {
                    bars.replace_front(bar);
                } else {
                    bars.push_front(bar);
                }
            }
            Ordering::Less => log::debug!(
                "Skipping bar {bar} with `ts_event` older than last bar `ts_event` {}",
                front.ts_event,
            ),
        },
    }
}

/// Inserts `bar` at its ordered position in the newest-first bars deque.
///
/// Bars older than the front are kept; a time bar replaces the cached bar
/// with an equal `ts_event`.
fn insert_bar_historical(bars: &mut BoundedVecDeque<Bar>, bar: Bar) {
    let index = bars.partition_point(|cached| cached.ts_event > bar.ts_event);

    if bar.bar_type.spec().is_time_aggregated()
        && bars
            .get(index)
            .is_some_and(|cached| cached.ts_event == bar.ts_event)
    {
        bars.replace(index, bar);
    } else {
        bars.insert(index, bar);
    }
}