Skip to main content

nautilus_common/cache/
mod.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! In-memory cache for market and execution data, with optional persistent backing.
17//!
18//! Provides methods to load, query, and update cached data such as instruments, orders, and prices.
19
20pub mod config;
21pub mod database;
22pub mod fifo;
23pub mod quote;
24pub mod refs;
25
26mod api;
27mod bounded;
28mod error;
29mod filter;
30mod index;
31mod position;
32mod view;
33
34#[cfg(test)]
35mod tests;
36
37use std::{
38    borrow::Cow,
39    cell::RefCell,
40    cmp::{Ordering, Reverse},
41    fmt::{Debug, Display},
42    rc::Rc,
43    time::{SystemTime, UNIX_EPOCH},
44};
45
46use ahash::{AHashMap, AHashSet};
47pub use api::CacheApi; // Re-export
48use bounded::BoundedVecDeque;
49use bytes::Bytes;
50pub use config::CacheConfig; // Re-export
51use database::{CacheDatabaseAdapter, CacheMap, register_loaded_currencies};
52pub use error::{
53    ACCOUNT_NOT_FOUND, AccountLookupError, CURRENCY_NOT_FOUND, CurrencyLookupError,
54    INSTRUMENT_NOT_FOUND, InstrumentLookupError, ORDER_BOOK_NOT_FOUND, ORDER_LIST_NOT_FOUND,
55    ORDER_NOT_FOUND, OWN_ORDER_BOOK_NOT_FOUND, OrderBookLookupError, OrderListLookupError,
56    OrderLookupError, OwnOrderBookLookupError, POSITION_NOT_FOUND, PositionLookupError,
57    SYNTHETIC_INSTRUMENT_NOT_FOUND, SyntheticInstrumentLookupError, VenueOrderIdOwnershipError,
58};
59use filter::{FilterSources, intersect_pair_or_many};
60use index::CacheIndex;
61use indexmap::IndexMap;
62use nautilus_core::{
63    DurationNanos, SharedCell, UnixNanos,
64    correctness::{
65        check_key_not_in_map, check_predicate_false, check_slice_not_empty, check_valid_string_utf8,
66    },
67};
68use nautilus_model::{
69    accounts::{Account, AccountAny},
70    data::{
71        Bar, BarType, FundingRateUpdate, GreeksData, IndexPriceUpdate, InstrumentClose,
72        InstrumentStatus, MarkPriceUpdate, QuoteTick, TradeTick, YieldCurve,
73        option_chain::OptionGreeks,
74    },
75    enums::{
76        AggregationSource, ContingencyType, InstrumentClass, OmsType, OrderSide, PositionSide,
77        PriceType,
78    },
79    events::{AccountState, OrderEventAny, OrderFilled, PositionAdjusted},
80    identifiers::{
81        AccountId, ActorId, ClientId, ClientOrderId, ExecAlgorithmId, InstrumentId, OrderListId,
82        PositionId, StrategyId, Venue, VenueOrderId,
83    },
84    instruments::{Instrument, InstrumentAny, SyntheticInstrument},
85    orderbook::{
86        OrderBook,
87        own::{OwnOrderBook, should_handle_own_book_order},
88    },
89    orders::{Order, OrderAny, OrderError, OrderList},
90    position::{Position, PositionReplayEvent},
91    types::{
92        Currency, Money, Price, Quantity,
93        fixed::{FIXED_PRECISION, MAX_FLOAT_PRECISION, check_fixed_precision},
94        price::PriceRaw,
95    },
96};
97pub use position::CacheSnapshotRef;
98use position::PositionSnapshotFrame;
99pub use refs::{AccountRef, AccountRefMut, OrderRef, OrderRefMut, PositionRef, PositionRefMut};
100use rust_decimal::Decimal;
101use ustr::Ustr;
102pub use view::CacheView; // Re-export
103
104use crate::xrate::get_exchange_rate;
105
106/// A common in-memory `Cache` for market and execution related data.
107#[cfg_attr(
108    feature = "python",
109    pyo3::pyclass(module = "nautilus_trader.common", unsendable)
110)]
111pub struct Cache {
112    config: CacheConfig,
113    index: CacheIndex,
114    database: Option<Box<dyn CacheDatabaseAdapter>>,
115    general: AHashMap<String, Bytes>,
116    currencies: AHashMap<Ustr, Currency>,
117    instruments: AHashMap<InstrumentId, InstrumentAny>,
118    instrument_closes: AHashMap<InstrumentId, InstrumentClose>,
119    synthetics: AHashMap<InstrumentId, SyntheticInstrument>,
120    books: AHashMap<InstrumentId, OrderBook>,
121    own_books: AHashMap<InstrumentId, OwnOrderBook>,
122    quotes: AHashMap<InstrumentId, BoundedVecDeque<QuoteTick>>,
123    trades: AHashMap<InstrumentId, BoundedVecDeque<TradeTick>>,
124    mark_xrates: AHashMap<(Currency, Currency), f64>,
125    mark_prices: AHashMap<InstrumentId, BoundedVecDeque<MarkPriceUpdate>>,
126    index_prices: AHashMap<InstrumentId, BoundedVecDeque<IndexPriceUpdate>>,
127    funding_rates: AHashMap<InstrumentId, BoundedVecDeque<FundingRateUpdate>>,
128    instrument_statuses: AHashMap<InstrumentId, BoundedVecDeque<InstrumentStatus>>,
129    bars: AHashMap<BarType, BoundedVecDeque<Bar>>,
130    greeks: AHashMap<InstrumentId, GreeksData>,
131    option_greeks: AHashMap<InstrumentId, OptionGreeks>,
132    yield_curves: AHashMap<String, YieldCurve>,
133    external_order_claims: AHashMap<InstrumentId, StrategyId>,
134    client_accounts: AHashMap<ClientId, AccountId>,
135    client_routes: AHashMap<Venue, ClientId>,
136    default_client_id: Option<ClientId>,
137    external_clients: AHashSet<ClientId>,
138    accounts: AHashMap<AccountId, SharedCell<AccountAny>>,
139    orders: AHashMap<ClientOrderId, SharedCell<OrderAny>>,
140    order_lists: AHashMap<OrderListId, OrderList>,
141    positions: AHashMap<PositionId, SharedCell<Position>>,
142    position_snapshots: AHashMap<PositionId, Vec<PositionSnapshotFrame>>,
143    position_snapshot_revisions: AHashMap<PositionId, u64>,
144    #[cfg(feature = "defi")]
145    pub(crate) defi: crate::defi::cache::DefiCache,
146}
147
148impl Debug for Cache {
149    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
150        f.debug_struct(stringify!(Cache))
151            .field("config", &self.config)
152            .field("index", &self.index)
153            .field("general", &self.general)
154            .field("currencies", &self.currencies)
155            .field("instruments", &self.instruments)
156            .field("synthetics", &self.synthetics)
157            .field("books", &self.books)
158            .field("own_books", &self.own_books)
159            .field("quotes", &self.quotes)
160            .field("trades", &self.trades)
161            .field("mark_xrates", &self.mark_xrates)
162            .field("mark_prices", &self.mark_prices)
163            .field("index_prices", &self.index_prices)
164            .field("funding_rates", &self.funding_rates)
165            .field("instrument_statuses", &self.instrument_statuses)
166            .field("instrument_closes", &self.instrument_closes)
167            .field("bars", &self.bars)
168            .field("greeks", &self.greeks)
169            .field("option_greeks", &self.option_greeks)
170            .field("yield_curves", &self.yield_curves)
171            .field("external_order_claims", &self.external_order_claims)
172            .field("client_accounts", &self.client_accounts)
173            .field("client_routes", &self.client_routes)
174            .field("default_client_id", &self.default_client_id)
175            .field("external_clients", &self.external_clients)
176            .field("accounts", &self.accounts)
177            .field("orders", &self.orders)
178            .field("order_lists", &self.order_lists)
179            .field("positions", &self.positions)
180            .field("position_snapshots", &self.position_snapshots)
181            .finish()
182    }
183}
184
185impl Default for Cache {
186    /// Creates a new default [`Cache`] instance.
187    fn default() -> Self {
188        Self::new(Some(CacheConfig::default()), None)
189    }
190}
191
192impl Cache {
193    /// Creates a new [`Cache`] instance with optional configuration and database adapter.
194    #[must_use]
195    /// # Note
196    ///
197    /// Uses provided `CacheConfig` or defaults, and optional `CacheDatabaseAdapter` for persistence.
198    ///
199    /// # Panics
200    ///
201    /// Panics if the cache config has a tick or bar capacity outside `[1, 1_000_000]`.
202    pub fn new(
203        config: Option<CacheConfig>,
204        database: Option<Box<dyn CacheDatabaseAdapter>>,
205    ) -> Self {
206        Self::try_new(config, database).expect("invalid `CacheConfig`")
207    }
208
209    /// Creates a new [`Cache`] instance with optional configuration and database adapter.
210    ///
211    /// # Errors
212    ///
213    /// Returns an error if the cache configuration is invalid.
214    pub fn try_new(
215        config: Option<CacheConfig>,
216        database: Option<Box<dyn CacheDatabaseAdapter>>,
217    ) -> crate::config::ConfigResult<Self> {
218        let config = config.unwrap_or_default();
219        config.validate()?;
220
221        Ok(Self {
222            config,
223            index: CacheIndex::default(),
224            database,
225            general: AHashMap::new(),
226            currencies: AHashMap::new(),
227            instruments: AHashMap::new(),
228            instrument_closes: AHashMap::new(),
229            synthetics: AHashMap::new(),
230            books: AHashMap::new(),
231            own_books: AHashMap::new(),
232            quotes: AHashMap::new(),
233            trades: AHashMap::new(),
234            mark_xrates: AHashMap::new(),
235            mark_prices: AHashMap::new(),
236            index_prices: AHashMap::new(),
237            funding_rates: AHashMap::new(),
238            instrument_statuses: AHashMap::new(),
239            bars: AHashMap::new(),
240            greeks: AHashMap::new(),
241            option_greeks: AHashMap::new(),
242            yield_curves: AHashMap::new(),
243            external_order_claims: AHashMap::new(),
244            client_accounts: AHashMap::new(),
245            client_routes: AHashMap::new(),
246            default_client_id: None,
247            external_clients: AHashSet::new(),
248            accounts: AHashMap::new(),
249            orders: AHashMap::new(),
250            order_lists: AHashMap::new(),
251            positions: AHashMap::new(),
252            position_snapshots: AHashMap::new(),
253            position_snapshot_revisions: AHashMap::new(),
254            #[cfg(feature = "defi")]
255            defi: crate::defi::cache::DefiCache::default(),
256        })
257    }
258
259    /// Returns the cache instances memory address.
260    #[must_use]
261    pub fn memory_address(&self) -> String {
262        format!("{:?}", std::ptr::from_ref(self))
263    }
264
265    /// Returns the strategy claiming external orders for `instrument_id`.
266    #[must_use]
267    pub fn external_order_claim(&self, instrument_id: &InstrumentId) -> Option<StrategyId> {
268        self.external_order_claims.get(instrument_id).copied()
269    }
270
271    /// Returns instrument IDs with an external order claim.
272    ///
273    /// Passing `Some(strategy_id)` filters the result to claims owned by that strategy. Passing
274    /// `None` returns every claimed instrument ID.
275    #[must_use]
276    pub fn external_order_claim_instrument_ids(
277        &self,
278        strategy_id: Option<StrategyId>,
279    ) -> AHashSet<InstrumentId> {
280        self.external_order_claims
281            .iter()
282            .filter_map(|(instrument_id, owner)| {
283                strategy_id
284                    .is_none_or(|strategy_id| *owner == strategy_id)
285                    .then_some(*instrument_id)
286            })
287            .collect()
288    }
289
290    /// Replaces the external order claims owned by `strategy_id`.
291    ///
292    /// External orders, fills, and materialized reconciliation activity for matching instrument
293    /// IDs are assigned to the strategy. Existing claims owned by other strategies are preserved.
294    ///
295    /// The operation is atomic: either every requested instrument is claimed or the cache is
296    /// unchanged. Passing an empty slice clears all claims owned by the strategy.
297    ///
298    /// # Errors
299    ///
300    /// Returns an error if an instrument is repeated or claimed by another strategy.
301    pub fn set_external_order_claims(
302        &mut self,
303        strategy_id: StrategyId,
304        instrument_ids: &[InstrumentId],
305    ) -> anyhow::Result<()> {
306        let mut requested = AHashSet::with_capacity(instrument_ids.len());
307
308        for instrument_id in instrument_ids {
309            if !requested.insert(*instrument_id) {
310                anyhow::bail!(
311                    "External order claim for {instrument_id} appears more than once for {strategy_id}"
312                );
313            }
314
315            if let Some(existing) = self.external_order_claims.get(instrument_id)
316                && *existing != strategy_id
317            {
318                anyhow::bail!(
319                    "External order claim for {instrument_id} already exists for {existing}"
320                );
321            }
322        }
323
324        self.external_order_claims
325            .retain(|_, owner| *owner != strategy_id);
326        self.external_order_claims.extend(
327            requested
328                .into_iter()
329                .map(|instrument_id| (instrument_id, strategy_id)),
330        );
331
332        Ok(())
333    }
334
335    /// Adds external order claims for `strategy_id` without replacing its existing claims.
336    ///
337    /// # Errors
338    ///
339    /// Returns an error if an instrument is repeated or already has a claim.
340    pub fn register_external_order_claims(
341        &mut self,
342        strategy_id: StrategyId,
343        instrument_ids: &[InstrumentId],
344    ) -> anyhow::Result<()> {
345        let mut requested = AHashSet::with_capacity(instrument_ids.len());
346
347        for instrument_id in instrument_ids {
348            if !requested.insert(*instrument_id) {
349                anyhow::bail!(
350                    "External order claim for {instrument_id} appears more than once for {strategy_id}"
351                );
352            }
353
354            if let Some(existing) = self.external_order_claims.get(instrument_id) {
355                anyhow::bail!(
356                    "External order claim for {instrument_id} already exists for {existing}"
357                );
358            }
359        }
360
361        self.external_order_claims.extend(
362            requested
363                .into_iter()
364                .map(|instrument_id| (instrument_id, strategy_id)),
365        );
366
367        Ok(())
368    }
369
370    /// Returns the account ID registered for the execution client `client_id`.
371    ///
372    /// The registration does not depend on account issuers or client names. The account itself
373    /// may not be cached yet, for example before the client reports its first account state.
374    #[must_use]
375    pub fn account_id_for_client(&self, client_id: &ClientId) -> Option<&AccountId> {
376        self.client_accounts.get(client_id)
377    }
378
379    /// Registers `account_id` as the account of the execution client `client_id`, replacing any
380    /// previous registration for the client.
381    ///
382    /// Registrations survive [`Self::clear_index`] and [`Self::reset`].
383    pub fn add_client_account(&mut self, client_id: ClientId, account_id: AccountId) {
384        self.client_accounts.insert(client_id, account_id);
385    }
386
387    /// Removes the account registration of the execution client `client_id`.
388    pub fn remove_client_account(&mut self, client_id: &ClientId) {
389        self.client_accounts.remove(client_id);
390    }
391
392    /// Returns the execution client registered as the route for `venue`, else the default client.
393    #[must_use]
394    pub fn client_id_for_venue(&self, venue: &Venue) -> Option<&ClientId> {
395        self.client_routes
396            .get(venue)
397            .or(self.default_client_id.as_ref())
398    }
399
400    /// Registers the execution client `client_id` as the route for commands on `venue`, replacing
401    /// any previous route for the venue.
402    ///
403    /// Routes and the default client survive [`Self::clear_index`] and [`Self::reset`].
404    pub fn add_client_route(&mut self, client_id: ClientId, venue: Venue) {
405        self.client_routes.insert(venue, client_id);
406    }
407
408    /// Registers the execution client `client_id` as the default route for venues without a
409    /// route, replacing any previous default.
410    pub fn set_default_client(&mut self, client_id: ClientId) {
411        self.default_client_id = Some(client_id);
412    }
413
414    /// Removes the venue routes and default route of the execution client `client_id`.
415    pub fn remove_client_routes(&mut self, client_id: &ClientId) {
416        self.client_routes.retain(|_, routed| routed != client_id);
417
418        if self.default_client_id.as_ref() == Some(client_id) {
419            self.default_client_id = None;
420        }
421    }
422
423    /// Returns whether `client_id` is registered as an external execution client, whose commands
424    /// are published for another process to execute.
425    #[must_use]
426    pub fn is_external_client(&self, client_id: &ClientId) -> bool {
427        self.external_clients.contains(client_id)
428    }
429
430    /// Registers `client_id` as an external execution client.
431    ///
432    /// Registrations survive [`Self::clear_index`] and [`Self::reset`].
433    pub fn add_external_client(&mut self, client_id: ClientId) {
434        self.external_clients.insert(client_id);
435    }
436
437    /// Sets the cache database adapter for persistence.
438    ///
439    /// This allows setting or replacing the database adapter after cache construction.
440    pub fn set_database(&mut self, database: Box<dyn CacheDatabaseAdapter>) {
441        let type_name = std::any::type_name_of_val(&*database);
442        log::info!("Cache database adapter set: {type_name}");
443        self.database = Some(database);
444    }
445
446    // -- COMMANDS --------------------------------------------------------------------------------
447
448    /// Clears and reloads general entries from the database into the cache.
449    ///
450    /// # Errors
451    ///
452    /// Returns an error if loading general cache data fails.
453    pub fn cache_general(&mut self) -> anyhow::Result<()> {
454        self.general = match &mut self.database {
455            Some(db) => db.load()?,
456            None => AHashMap::new(),
457        };
458
459        log::info!(
460            "Cached {} general object(s) from database",
461            self.general.len()
462        );
463        Ok(())
464    }
465
466    /// Loads all core caches from the database.
467    ///
468    /// The loaded instrument closes replace all closes already held in memory. This includes values
469    /// added directly before the persistent cache is loaded.
470    ///
471    /// # Errors
472    ///
473    /// Returns an error if loading cache data fails.
474    pub async fn cache_all(&mut self) -> anyhow::Result<()> {
475        let mut cache_map = match &self.database {
476            Some(db) => db.load_all().await?,
477            None => CacheMap::default(),
478        };
479
480        // An adapter registers the loaded currencies before decoding their dependents; repeating
481        // it here keeps the cached map consistent with the registry whatever the adapter did.
482        register_loaded_currencies(&mut cache_map.currencies)?;
483        self.currencies = cache_map.currencies;
484        self.instruments = cache_map.instruments;
485        self.instrument_closes = cache_map.instrument_closes;
486        self.synthetics = cache_map.synthetics;
487        self.accounts = cache_map
488            .accounts
489            .into_iter()
490            .map(|(id, account)| (id, SharedCell::new(account)))
491            .collect();
492        self.orders = cache_map
493            .orders
494            .into_iter()
495            .map(|(id, order)| (id, SharedCell::new(order)))
496            .collect();
497        self.positions = cache_map
498            .positions
499            .into_iter()
500            .map(|(id, position)| (id, SharedCell::new(position)))
501            .collect();
502
503        if let Some(db) = &self.database {
504            let order_position = db.load_index_order_position()?;
505            self.index.order_position = self.sanitize_order_position_index(order_position);
506            self.index.order_client = db.load_index_order_client()?;
507        }
508
509        self.cache_position_oms()?;
510        self.assign_position_ids_to_contingencies();
511        Ok(())
512    }
513
514    /// Clears and reloads the currency cache from the database.
515    ///
516    /// # Errors
517    ///
518    /// Returns an error if loading currencies cache fails.
519    pub async fn cache_currencies(&mut self) -> anyhow::Result<()> {
520        self.currencies = match &mut self.database {
521            Some(db) => db.load_currencies().await?,
522            None => AHashMap::new(),
523        };
524
525        register_loaded_currencies(&mut self.currencies)?;
526
527        log::info!("Cached {} currencies from database", self.currencies.len());
528        Ok(())
529    }
530
531    /// Clears and reloads the instrument cache from the database.
532    ///
533    /// # Errors
534    ///
535    /// Returns an error if loading instruments cache fails.
536    pub async fn cache_instruments(&mut self) -> anyhow::Result<()> {
537        self.instruments = match &mut self.database {
538            Some(db) => db.load_instruments().await?,
539            None => AHashMap::new(),
540        };
541
542        log::info!("Cached {} instruments from database", self.general.len());
543        Ok(())
544    }
545
546    /// Clears and reloads the synthetic instrument cache from the database.
547    ///
548    /// # Errors
549    ///
550    /// Returns an error if loading synthetic instruments cache fails.
551    pub async fn cache_synthetics(&mut self) -> anyhow::Result<()> {
552        self.synthetics = match &mut self.database {
553            Some(db) => db.load_synthetics().await?,
554            None => AHashMap::new(),
555        };
556
557        log::info!(
558            "Cached {} synthetic instruments from database",
559            self.general.len()
560        );
561        Ok(())
562    }
563
564    /// Clears and reloads the account cache from the database.
565    ///
566    /// # Errors
567    ///
568    /// Returns an error if loading accounts cache fails.
569    pub async fn cache_accounts(&mut self) -> anyhow::Result<()> {
570        self.accounts = match &mut self.database {
571            Some(db) => db
572                .load_accounts()
573                .await?
574                .into_iter()
575                .map(|(id, account)| (id, SharedCell::new(account)))
576                .collect(),
577            None => AHashMap::new(),
578        };
579
580        log::info!(
581            "Cached {} synthetic instruments from database",
582            self.general.len()
583        );
584        Ok(())
585    }
586
587    /// Clears and reloads the order cache from the database.
588    ///
589    /// # Errors
590    ///
591    /// Returns an error if loading orders cache fails.
592    pub async fn cache_orders(&mut self) -> anyhow::Result<()> {
593        self.orders = match &mut self.database {
594            Some(db) => db
595                .load_orders()
596                .await?
597                .into_iter()
598                .map(|(id, order)| (id, SharedCell::new(order)))
599                .collect(),
600            None => AHashMap::new(),
601        };
602
603        if let Some(db) = &self.database {
604            let order_position = db.load_index_order_position()?;
605            self.index.order_position = self.sanitize_order_position_index(order_position);
606            self.index.order_client = db.load_index_order_client()?;
607        }
608
609        log::info!("Cached {} orders from database", self.general.len());
610
611        self.assign_position_ids_to_contingencies();
612        Ok(())
613    }
614
615    fn sanitize_order_position_index(
616        &self,
617        mut order_position: AHashMap<ClientOrderId, PositionId>,
618    ) -> AHashMap<ClientOrderId, PositionId> {
619        let original_len = order_position.len();
620        order_position.retain(|client_order_id, _| self.orders.contains_key(client_order_id));
621        let removed = original_len - order_position.len();
622
623        if removed > 0 {
624            log::warn!(
625                "Filtered {removed} stale order-position index entries without backing orders during cache load"
626            );
627        }
628
629        order_position
630    }
631
632    /// Clears and reloads the position cache from the database.
633    ///
634    /// # Errors
635    ///
636    /// Returns an error if loading positions cache fails.
637    pub async fn cache_positions(&mut self) -> anyhow::Result<()> {
638        self.positions = match &mut self.database {
639            Some(db) => db
640                .load_positions()
641                .await?
642                .into_iter()
643                .map(|(id, position)| (id, SharedCell::new(position)))
644                .collect(),
645            None => AHashMap::new(),
646        };
647
648        self.cache_position_oms()?;
649        log::info!("Cached {} positions from database", self.general.len());
650        Ok(())
651    }
652
653    fn cache_position_oms(&mut self) -> anyhow::Result<()> {
654        let persisted = match &self.database {
655            Some(database) => database.load()?,
656            None => self.general.clone(),
657        };
658
659        self.general
660            .retain(|key, _| !key.starts_with(POSITION_OMS_KEY_PREFIX));
661
662        for (key, value) in persisted {
663            if !key.starts_with(POSITION_OMS_KEY_PREFIX) {
664                continue;
665            }
666            self.general.insert(key, value);
667        }
668
669        self.index_position_oms();
670        Ok(())
671    }
672
673    /// Clears the current cache index and re-build.
674    pub fn build_index(&mut self) {
675        log::debug!("Building index");
676
677        // Index accounts
678        for account_id in self.accounts.keys() {
679            self.index.add_venue_account(*account_id);
680        }
681
682        // Index orders
683        for (client_order_id, order_cell) in &self.orders {
684            let order = order_cell.borrow();
685            let instrument_id = order.instrument_id();
686            let venue = instrument_id.venue;
687            let strategy_id = order.strategy_id();
688
689            // 1: Build index.venue_orders -> {Venue, {ClientOrderId}}
690            self.index
691                .venue_orders
692                .entry(venue)
693                .or_default()
694                .insert(*client_order_id);
695
696            // 2: Build index.venue_order_ids -> {VenueOrderId, ClientOrderId}
697            //    and index.client_order_ids -> {ClientOrderId, VenueOrderId}
698            if let Some(venue_order_id) = order.venue_order_id() {
699                self.index
700                    .venue_order_ids
701                    .insert(venue_order_id, *client_order_id);
702                self.index
703                    .client_order_ids
704                    .insert(*client_order_id, venue_order_id);
705            }
706
707            // 3: Build index.order_position -> {ClientOrderId, PositionId}
708            self.index.order_position.extend(
709                order
710                    .position_id()
711                    .map(|position_id| (*client_order_id, position_id)),
712            );
713
714            // 4: Build index.order_strategy -> {ClientOrderId, StrategyId}
715            self.index
716                .order_strategy
717                .insert(*client_order_id, strategy_id);
718
719            // 5: Build index.instrument_orders -> {InstrumentId, {ClientOrderId}}
720            self.index
721                .instrument_orders
722                .entry(instrument_id)
723                .or_default()
724                .insert(*client_order_id);
725
726            // 6: Build index.strategy_orders -> {StrategyId, {ClientOrderId}}
727            self.index
728                .strategy_orders
729                .entry(strategy_id)
730                .or_default()
731                .insert(*client_order_id);
732
733            // 7: Build index.account_orders -> {AccountId, {ClientOrderId}}
734            if let Some(account_id) = order.account_id() {
735                self.index
736                    .account_orders
737                    .entry(account_id)
738                    .or_default()
739                    .insert(*client_order_id);
740            }
741
742            // 8: Build index.exec_algorithm_orders -> {ExecAlgorithmId, {ClientOrderId}}
743            if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
744                self.index
745                    .exec_algorithm_orders
746                    .entry(exec_algorithm_id)
747                    .or_default()
748                    .insert(*client_order_id);
749                self.index.exec_algorithms.insert(exec_algorithm_id);
750            }
751
752            // 9: Build index.exec_spawn_orders -> {ClientOrderId, {ClientOrderId}}
753            if let Some(exec_spawn_id) = order.exec_spawn_id() {
754                self.index
755                    .exec_spawn_orders
756                    .entry(exec_spawn_id)
757                    .or_default()
758                    .insert(*client_order_id);
759            }
760
761            // 10: Build index.orders -> {ClientOrderId}
762            self.index.orders.insert(*client_order_id);
763
764            // 11: Build index.orders_active_local -> {ClientOrderId}
765            if order.is_active_local() {
766                self.index.orders_active_local.insert(*client_order_id);
767            }
768
769            // 12: Build index.orders_open -> {ClientOrderId}
770            if order.is_open() {
771                self.index.orders_open.insert(*client_order_id);
772            }
773
774            // 13: Build index.orders_closed -> {ClientOrderId}
775            if order.is_closed() {
776                self.index.orders_closed.insert(*client_order_id);
777            }
778
779            // 14: Build index.orders_emulated -> {ClientOrderId}
780            if order.emulation_trigger().is_some() && !order.is_closed() {
781                self.index.orders_emulated.insert(*client_order_id);
782            }
783
784            // 15: Build index.orders_inflight -> {ClientOrderId}
785            if order.is_inflight() {
786                self.index.orders_inflight.insert(*client_order_id);
787            }
788
789            // 16: Build index.strategies -> {StrategyId}
790            self.index.strategies.insert(strategy_id);
791        }
792
793        // Index positions
794        for (position_id, position_cell) in &self.positions {
795            let position = position_cell.borrow();
796            let instrument_id = position.instrument_id;
797            let venue = instrument_id.venue;
798            let strategy_id = position.strategy_id;
799
800            // 1: Build index.venue_positions -> {Venue, {PositionId}}
801            self.index
802                .venue_positions
803                .entry(venue)
804                .or_default()
805                .insert(*position_id);
806
807            // 2: Build index.position_strategy -> {PositionId, StrategyId}
808            self.index
809                .position_strategy
810                .insert(*position_id, strategy_id);
811
812            // 3: Build index.position_orders -> {PositionId, {ClientOrderId}}
813            let position_orders = self.index.position_orders.entry(*position_id).or_default();
814            position_orders.extend(
815                position
816                    .client_order_ids()
817                    .into_iter()
818                    .filter(|client_order_id| self.orders.contains_key(client_order_id)),
819            );
820
821            // 4: Build index.instrument_positions -> {InstrumentId, {PositionId}}
822            self.index
823                .instrument_positions
824                .entry(instrument_id)
825                .or_default()
826                .insert(*position_id);
827            self.index
828                .instrument_orders
829                .entry(instrument_id)
830                .or_default();
831
832            // 5: Build index.strategy_positions -> {StrategyId, {PositionId}}
833            self.index
834                .strategy_positions
835                .entry(strategy_id)
836                .or_default()
837                .insert(*position_id);
838            self.index.strategy_orders.entry(strategy_id).or_default();
839
840            // 6: Build index.account_positions -> {AccountId, {PositionId}}
841            self.index
842                .account_positions
843                .entry(position.account_id)
844                .or_default()
845                .insert(*position_id);
846
847            // 7: Build index.positions -> {PositionId}
848            self.index.positions.insert(*position_id);
849
850            // 8: Build index.positions_open -> {PositionId}
851            if position.is_open() {
852                self.index.positions_open.insert(*position_id);
853            }
854
855            // 9: Build index.positions_closed -> {PositionId}
856            if position.is_closed() {
857                self.index.positions_closed.insert(*position_id);
858            }
859
860            // 10: Build index.strategies -> {StrategyId}
861            self.index.strategies.insert(strategy_id);
862        }
863
864        self.index_position_oms();
865        let client_order_ids: Vec<_> = self.orders.keys().copied().collect();
866        for client_order_id in client_order_ids {
867            self.reindex_hedging_position(client_order_id);
868        }
869    }
870
871    fn index_position_oms(&mut self) {
872        self.index.position_oms.clear();
873
874        for (key, value) in &self.general {
875            let Some(position_id) = key.strip_prefix(POSITION_OMS_KEY_PREFIX) else {
876                continue;
877            };
878            let position_id = PositionId::new(position_id);
879            if !self.positions.contains_key(&position_id) {
880                continue;
881            }
882
883            match serde_json::from_slice::<OmsType>(value) {
884                Ok(oms_type) => {
885                    self.index.position_oms.insert(position_id, oms_type);
886                }
887                Err(e) => {
888                    log::error!("Failed to decode position OMS for {position_id}: {e}");
889                }
890            }
891        }
892
893        for position in self.positions.values().map(|cell| cell.borrow()) {
894            if !self.index.position_oms.contains_key(&position.id)
895                && position.id.as_str()
896                    == format!("{}-{}", position.instrument_id, position.strategy_id)
897            {
898                self.index
899                    .position_oms
900                    .insert(position.id, OmsType::Netting);
901            }
902        }
903    }
904
905    /// Returns whether the cache has a backing database.
906    #[must_use]
907    pub const fn has_backing(&self) -> bool {
908        self.database.is_some()
909    }
910
911    /// Loads persisted actor state.
912    ///
913    /// Returns `None` when the cache has no backing database.
914    ///
915    /// # Errors
916    ///
917    /// Returns an error if loading actor state fails.
918    pub fn load_actor_state(
919        &self,
920        actor_id: &ActorId,
921    ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
922        self.database
923            .as_ref()
924            .map(|database| database.load_actor(actor_id))
925            .transpose()
926            .map(|state| state.map(Self::decode_component_state))
927    }
928
929    /// Loads persisted strategy state.
930    ///
931    /// Returns `None` when the cache has no backing database.
932    ///
933    /// # Errors
934    ///
935    /// Returns an error if loading strategy state fails.
936    pub fn load_strategy_state(
937        &self,
938        strategy_id: &StrategyId,
939    ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
940        self.database
941            .as_ref()
942            .map(|database| database.load_strategy(strategy_id))
943            .transpose()
944            .map(|state| state.map(Self::decode_component_state))
945    }
946
947    /// Persists actor state when the cache has a backing database.
948    ///
949    /// # Errors
950    ///
951    /// Returns an error if updating actor state fails.
952    pub fn update_actor_state(
953        &self,
954        actor_id: &ActorId,
955        state: &IndexMap<String, Vec<u8>>,
956    ) -> anyhow::Result<()> {
957        if let Some(database) = &self.database {
958            database.update_actor(actor_id, &Self::encode_component_state(state))?;
959        }
960        Ok(())
961    }
962
963    /// Persists strategy state when the cache has a backing database.
964    ///
965    /// # Errors
966    ///
967    /// Returns an error if updating strategy state fails.
968    pub fn update_strategy_state(
969        &self,
970        strategy_id: &StrategyId,
971        state: &IndexMap<String, Vec<u8>>,
972    ) -> anyhow::Result<()> {
973        if let Some(database) = &self.database {
974            database.update_strategy(strategy_id, &Self::encode_component_state(state))?;
975        }
976        Ok(())
977    }
978
979    fn decode_component_state(state: AHashMap<String, Bytes>) -> IndexMap<String, Vec<u8>> {
980        state
981            .into_iter()
982            .map(|(key, value)| (key, value.to_vec()))
983            .collect()
984    }
985
986    fn encode_component_state(state: &IndexMap<String, Vec<u8>>) -> AHashMap<String, Bytes> {
987        state
988            .iter()
989            .map(|(key, value)| (key.clone(), Bytes::copy_from_slice(value)))
990            .collect()
991    }
992
993    // Calculate the unrealized profit and loss (PnL) for `position`.
994    #[must_use]
995    pub fn calculate_unrealized_pnl(&self, position: &Position) -> Option<Money> {
996        let Some(quote) = self.quote(&position.instrument_id) else {
997            log::warn!(
998                "Cannot calculate unrealized PnL for {}, no quotes for {}",
999                position.id,
1000                position.instrument_id
1001            );
1002            return None;
1003        };
1004
1005        // Use exit price for mark-to-market: longs exit at bid, shorts exit at ask
1006        let last = match position.side {
1007            PositionSide::Flat => {
1008                return Some(Money::zero(position.settlement_currency));
1009            }
1010            PositionSide::Long => quote.bid_price,
1011            PositionSide::Short => quote.ask_price,
1012        };
1013
1014        position
1015            .try_unrealized_pnl(last)
1016            .inspect_err(|e| {
1017                log::error!("Cannot calculate unrealized PnL for {}: {e}", position.id);
1018            })
1019            .ok()
1020    }
1021
1022    /// Checks integrity of data within the cache.
1023    ///
1024    /// All data should be loaded from the database prior to this call.
1025    /// If an error is found then a log error message will also be produced.
1026    ///
1027    /// # Panics
1028    ///
1029    /// Panics if failure calling system clock.
1030    #[must_use]
1031    pub fn check_integrity(&mut self) -> bool {
1032        let mut error_count = 0;
1033        let failure = "Integrity failure";
1034
1035        // Get current timestamp in microseconds
1036        let timestamp_us = SystemTime::now()
1037            .duration_since(UNIX_EPOCH)
1038            .expect("Time went backwards")
1039            .as_micros();
1040
1041        log::info!("Checking data integrity");
1042
1043        // Check object caches
1044        for account_id in self.accounts.keys() {
1045            if !self
1046                .index
1047                .venue_accounts
1048                .get(&account_id.get_issuer())
1049                .is_some_and(|account_ids| account_ids.contains(account_id))
1050            {
1051                log::error!(
1052                    "{failure} in accounts: {account_id} not found in `self.index.venue_accounts`",
1053                );
1054                error_count += 1;
1055            }
1056        }
1057
1058        for (client_order_id, order_cell) in &self.orders {
1059            let order = order_cell.borrow();
1060
1061            if !self.index.order_strategy.contains_key(client_order_id) {
1062                log::error!(
1063                    "{failure} in orders: {client_order_id} not found in `self.index.order_strategy`"
1064                );
1065                error_count += 1;
1066            }
1067
1068            if !self.index.orders.contains(client_order_id) {
1069                log::error!(
1070                    "{failure} in orders: {client_order_id} not found in `self.index.orders`",
1071                );
1072                error_count += 1;
1073            }
1074
1075            if order.is_inflight() && !self.index.orders_inflight.contains(client_order_id) {
1076                log::error!(
1077                    "{failure} in orders: {client_order_id} not found in `self.index.orders_inflight`",
1078                );
1079                error_count += 1;
1080            }
1081
1082            if order.is_active_local() && !self.index.orders_active_local.contains(client_order_id)
1083            {
1084                log::error!(
1085                    "{failure} in orders: {client_order_id} not found in `self.index.orders_active_local`",
1086                );
1087                error_count += 1;
1088            }
1089
1090            if order.is_open() && !self.index.orders_open.contains(client_order_id) {
1091                log::error!(
1092                    "{failure} in orders: {client_order_id} not found in `self.index.orders_open`",
1093                );
1094                error_count += 1;
1095            }
1096
1097            if order.is_closed() && !self.index.orders_closed.contains(client_order_id) {
1098                log::error!(
1099                    "{failure} in orders: {client_order_id} not found in `self.index.orders_closed`",
1100                );
1101                error_count += 1;
1102            }
1103
1104            if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
1105                if !self
1106                    .index
1107                    .exec_algorithm_orders
1108                    .contains_key(&exec_algorithm_id)
1109                {
1110                    log::error!(
1111                        "{failure} in orders: {client_order_id} not found in `self.index.exec_algorithm_orders`",
1112                    );
1113                    error_count += 1;
1114                }
1115
1116                if order.exec_spawn_id().is_none()
1117                    && !self.index.exec_spawn_orders.contains_key(client_order_id)
1118                {
1119                    log::error!(
1120                        "{failure} in orders: {client_order_id} not found in `self.index.exec_spawn_orders`",
1121                    );
1122                    error_count += 1;
1123                }
1124            }
1125        }
1126
1127        for (position_id, position_cell) in &self.positions {
1128            let position = position_cell.borrow();
1129
1130            if !self.index.position_strategy.contains_key(position_id) {
1131                log::error!(
1132                    "{failure} in positions: {position_id} not found in `self.index.position_strategy`",
1133                );
1134                error_count += 1;
1135            }
1136
1137            if !self.index.position_orders.contains_key(position_id) {
1138                log::error!(
1139                    "{failure} in positions: {position_id} not found in `self.index.position_orders`",
1140                );
1141                error_count += 1;
1142            }
1143
1144            if !self.index.positions.contains(position_id) {
1145                log::error!(
1146                    "{failure} in positions: {position_id} not found in `self.index.positions`",
1147                );
1148                error_count += 1;
1149            }
1150
1151            if position.is_open() && !self.index.positions_open.contains(position_id) {
1152                log::error!(
1153                    "{failure} in positions: {position_id} not found in `self.index.positions_open`",
1154                );
1155                error_count += 1;
1156            }
1157
1158            if position.is_closed() && !self.index.positions_closed.contains(position_id) {
1159                log::error!(
1160                    "{failure} in positions: {position_id} not found in `self.index.positions_closed`",
1161                );
1162                error_count += 1;
1163            }
1164        }
1165
1166        // Check indexes
1167        for account_id in self.index.venue_accounts.values().flatten() {
1168            if !self.accounts.contains_key(account_id) {
1169                log::error!(
1170                    "{failure} in `index.venue_accounts`: {account_id} not found in `self.accounts`",
1171                );
1172                error_count += 1;
1173            }
1174        }
1175
1176        for client_order_id in self.index.venue_order_ids.values() {
1177            if !self.orders.contains_key(client_order_id) {
1178                log::error!(
1179                    "{failure} in `index.venue_order_ids`: {client_order_id} not found in `self.orders`",
1180                );
1181                error_count += 1;
1182            }
1183        }
1184
1185        for client_order_id in self.index.client_order_ids.keys() {
1186            if !self.orders.contains_key(client_order_id) {
1187                log::error!(
1188                    "{failure} in `index.client_order_ids`: {client_order_id} not found in `self.orders`",
1189                );
1190                error_count += 1;
1191            }
1192        }
1193
1194        for client_order_id in self.index.order_position.keys() {
1195            if !self.orders.contains_key(client_order_id) {
1196                log::error!(
1197                    "{failure} in `index.order_position`: {client_order_id} not found in `self.orders`",
1198                );
1199                error_count += 1;
1200            }
1201        }
1202
1203        // Check indexes
1204        for client_order_id in self.index.order_strategy.keys() {
1205            if !self.orders.contains_key(client_order_id) {
1206                log::error!(
1207                    "{failure} in `index.order_strategy`: {client_order_id} not found in `self.orders`",
1208                );
1209                error_count += 1;
1210            }
1211        }
1212
1213        for position_id in self.index.position_strategy.keys() {
1214            if !self.positions.contains_key(position_id) {
1215                log::error!(
1216                    "{failure} in `index.position_strategy`: {position_id} not found in `self.positions`",
1217                );
1218                error_count += 1;
1219            }
1220        }
1221
1222        for position_id in self.index.position_orders.keys() {
1223            if !self.positions.contains_key(position_id) {
1224                log::error!(
1225                    "{failure} in `index.position_orders`: {position_id} not found in `self.positions`",
1226                );
1227                error_count += 1;
1228            }
1229        }
1230
1231        for (instrument_id, client_order_ids) in &self.index.instrument_orders {
1232            for client_order_id in client_order_ids {
1233                if !self.orders.contains_key(client_order_id) {
1234                    log::error!(
1235                        "{failure} in `index.instrument_orders`: {instrument_id} not found in `self.orders`",
1236                    );
1237                    error_count += 1;
1238                }
1239            }
1240        }
1241
1242        for instrument_id in self.index.instrument_positions.keys() {
1243            if !self.index.instrument_orders.contains_key(instrument_id) {
1244                log::error!(
1245                    "{failure} in `index.instrument_positions`: {instrument_id} not found in `index.instrument_orders`",
1246                );
1247                error_count += 1;
1248            }
1249        }
1250
1251        for client_order_ids in self.index.strategy_orders.values() {
1252            for client_order_id in client_order_ids {
1253                if !self.orders.contains_key(client_order_id) {
1254                    log::error!(
1255                        "{failure} in `index.strategy_orders`: {client_order_id} not found in `self.orders`",
1256                    );
1257                    error_count += 1;
1258                }
1259            }
1260        }
1261
1262        for position_ids in self.index.strategy_positions.values() {
1263            for position_id in position_ids {
1264                if !self.positions.contains_key(position_id) {
1265                    log::error!(
1266                        "{failure} in `index.strategy_positions`: {position_id} not found in `self.positions`",
1267                    );
1268                    error_count += 1;
1269                }
1270            }
1271        }
1272
1273        for client_order_id in &self.index.orders {
1274            if !self.orders.contains_key(client_order_id) {
1275                log::error!(
1276                    "{failure} in `index.orders`: {client_order_id} not found in `self.orders`",
1277                );
1278                error_count += 1;
1279            }
1280        }
1281
1282        for client_order_id in &self.index.orders_emulated {
1283            if !self.orders.contains_key(client_order_id) {
1284                log::error!(
1285                    "{failure} in `index.orders_emulated`: {client_order_id} not found in `self.orders`",
1286                );
1287                error_count += 1;
1288            }
1289        }
1290
1291        for client_order_id in &self.index.orders_active_local {
1292            if !self.orders.contains_key(client_order_id) {
1293                log::error!(
1294                    "{failure} in `index.orders_active_local`: {client_order_id} not found in `self.orders`",
1295                );
1296                error_count += 1;
1297            }
1298        }
1299
1300        for client_order_id in &self.index.orders_inflight {
1301            if !self.orders.contains_key(client_order_id) {
1302                log::error!(
1303                    "{failure} in `index.orders_inflight`: {client_order_id} not found in `self.orders`",
1304                );
1305                error_count += 1;
1306            }
1307        }
1308
1309        for client_order_id in &self.index.orders_open {
1310            if !self.orders.contains_key(client_order_id) {
1311                log::error!(
1312                    "{failure} in `index.orders_open`: {client_order_id} not found in `self.orders`",
1313                );
1314                error_count += 1;
1315            }
1316        }
1317
1318        for client_order_id in &self.index.orders_closed {
1319            if !self.orders.contains_key(client_order_id) {
1320                log::error!(
1321                    "{failure} in `index.orders_closed`: {client_order_id} not found in `self.orders`",
1322                );
1323                error_count += 1;
1324            }
1325        }
1326
1327        for position_id in &self.index.positions {
1328            if !self.positions.contains_key(position_id) {
1329                log::error!(
1330                    "{failure} in `index.positions`: {position_id} not found in `self.positions`",
1331                );
1332                error_count += 1;
1333            }
1334        }
1335
1336        for position_id in &self.index.positions_open {
1337            if !self.positions.contains_key(position_id) {
1338                log::error!(
1339                    "{failure} in `index.positions_open`: {position_id} not found in `self.positions`",
1340                );
1341                error_count += 1;
1342            }
1343        }
1344
1345        for position_id in &self.index.positions_closed {
1346            if !self.positions.contains_key(position_id) {
1347                log::error!(
1348                    "{failure} in `index.positions_closed`: {position_id} not found in `self.positions`",
1349                );
1350                error_count += 1;
1351            }
1352        }
1353
1354        for strategy_id in &self.index.strategies {
1355            if !self.index.strategy_orders.contains_key(strategy_id) {
1356                log::error!(
1357                    "{failure} in `index.strategies`: {strategy_id} not found in `index.strategy_orders`",
1358                );
1359                error_count += 1;
1360            }
1361        }
1362
1363        for exec_algorithm_id in &self.index.exec_algorithms {
1364            if !self
1365                .index
1366                .exec_algorithm_orders
1367                .contains_key(exec_algorithm_id)
1368            {
1369                log::error!(
1370                    "{failure} in `index.exec_algorithms`: {exec_algorithm_id} not found in `index.exec_algorithm_orders`",
1371                );
1372                error_count += 1;
1373            }
1374        }
1375
1376        let total_us = SystemTime::now()
1377            .duration_since(UNIX_EPOCH)
1378            .expect("Time went backwards")
1379            .as_micros()
1380            - timestamp_us;
1381
1382        if error_count == 0 {
1383            log::info!("Integrity check passed in {total_us}μs");
1384            true
1385        } else {
1386            log::error!(
1387                "Integrity check failed with {error_count} error{} in {total_us}μs",
1388                if error_count == 1 { "" } else { "s" },
1389            );
1390            false
1391        }
1392    }
1393
1394    /// Checks for any residual open state and log warnings if any are found.
1395    ///
1396    ///'Open state' is considered to be open orders and open positions.
1397    #[must_use]
1398    pub fn check_residuals(&self) -> bool {
1399        log::debug!("Checking residuals");
1400
1401        let mut residuals = false;
1402
1403        // Check for any open orders
1404        for order in self.orders_open(None, None, None, None, None) {
1405            residuals = true;
1406            log::warn!("Residual {order}");
1407        }
1408
1409        // Check for any open positions
1410        for position in self.positions_open(None, None, None, None, None) {
1411            residuals = true;
1412            log::warn!("Residual {position}");
1413        }
1414
1415        residuals
1416    }
1417
1418    /// Purges all closed orders from the cache that are older than `buffer_secs`.
1419    ///
1420    ///
1421    /// Only orders that have been closed for at least this amount of time will be purged.
1422    /// A value of 0 means purge all closed orders regardless of when they were closed.
1423    pub fn purge_closed_orders(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
1424        log::debug!(
1425            "Purging closed orders{}",
1426            if buffer_secs > 0 {
1427                format!(" with buffer_secs={buffer_secs}")
1428            } else {
1429                String::new()
1430            }
1431        );
1432
1433        let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
1434            log::warn!(
1435                "Cannot purge closed orders: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
1436            );
1437            return;
1438        };
1439        let purge_cutoff = ts_now.checked_sub(buffer_ns);
1440
1441        let mut affected_order_list_ids: AHashSet<OrderListId> = AHashSet::new();
1442        let mut purged_client_order_ids: AHashSet<ClientOrderId> = AHashSet::new();
1443
1444        'outer: for client_order_id in self.index.orders_closed.clone() {
1445            let purge_target = self.orders.get(&client_order_id).and_then(|order_cell| {
1446                let order = order_cell.borrow();
1447                if order.is_closed()
1448                    && let Some(ts_closed) = order.ts_closed()
1449                    && purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
1450                {
1451                    let linked = order.linked_order_ids().map(<[_]>::to_vec);
1452                    let order_list_id = order.order_list_id();
1453                    Some((linked, order_list_id))
1454                } else {
1455                    None
1456                }
1457            });
1458
1459            let Some((linked, order_list_id)) = purge_target else {
1460                continue;
1461            };
1462
1463            // Check any linked orders (contingency orders)
1464            if let Some(linked_order_ids) = linked {
1465                for linked_order_id in &linked_order_ids {
1466                    if let Some(linked_order_cell) = self.orders.get(linked_order_id)
1467                        && linked_order_cell.borrow().is_open()
1468                    {
1469                        // Do not purge if linked order still open
1470                        continue 'outer;
1471                    }
1472                }
1473            }
1474
1475            if let Some(order_list_id) = order_list_id {
1476                affected_order_list_ids.insert(order_list_id);
1477            }
1478
1479            if self.purge_order_except_aliases(client_order_id) {
1480                purged_client_order_ids.insert(client_order_id);
1481            }
1482        }
1483
1484        if !purged_client_order_ids.is_empty() {
1485            self.index
1486                .venue_order_ids
1487                .retain(|_, owner| !purged_client_order_ids.contains(owner));
1488        }
1489
1490        for order_list_id in affected_order_list_ids {
1491            if let Some(order_list) = self.order_lists.get(&order_list_id) {
1492                let all_purged = order_list
1493                    .client_order_ids
1494                    .iter()
1495                    .all(|id| !self.orders.contains_key(id));
1496
1497                if all_purged {
1498                    self.order_lists.remove(&order_list_id);
1499                    log::info!("Purged {order_list_id}");
1500                }
1501            }
1502        }
1503    }
1504
1505    /// Purges all closed positions from the cache that are older than `buffer_secs`.
1506    pub fn purge_closed_positions(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
1507        log::debug!(
1508            "Purging closed positions{}",
1509            if buffer_secs > 0 {
1510                format!(" with buffer_secs={buffer_secs}")
1511            } else {
1512                String::new()
1513            }
1514        );
1515
1516        let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
1517            log::warn!(
1518                "Cannot purge closed positions: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
1519            );
1520            return;
1521        };
1522        let purge_cutoff = ts_now.checked_sub(buffer_ns);
1523
1524        for position_id in self.index.positions_closed.clone() {
1525            let should_purge = self.positions.get(&position_id).is_some_and(|cell| {
1526                let position = cell.borrow();
1527                position.is_closed()
1528                    && position.ts_closed.is_some_and(|ts_closed| {
1529                        purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
1530                    })
1531            });
1532
1533            if should_purge {
1534                self.purge_position(position_id);
1535            }
1536        }
1537    }
1538
1539    /// Purges the order with the `client_order_id` from the cache (if found).
1540    ///
1541    /// For safety, an order is prevented from being purged if it's open.
1542    pub fn purge_order(&mut self, client_order_id: ClientOrderId) {
1543        if self.purge_order_except_aliases(client_order_id) {
1544            self.index
1545                .venue_order_ids
1546                .retain(|_, owner| owner != &client_order_id);
1547        }
1548    }
1549
1550    /// Removes the order and its indexes, leaving the reverse venue order ID aliases for the
1551    /// caller to sweep by owner, so a bulk purge pays for one pass rather than one pass per order.
1552    ///
1553    /// Returns whether the order was purged, so a skipped purge leaves its aliases intact.
1554    fn purge_order_except_aliases(&mut self, client_order_id: ClientOrderId) -> bool {
1555        struct OrderDetails {
1556            is_open: bool,
1557            instrument_id: InstrumentId,
1558            strategy_id: StrategyId,
1559            account_id: Option<AccountId>,
1560            exec_algorithm_id: Option<ExecAlgorithmId>,
1561            exec_spawn_id: Option<ClientOrderId>,
1562            position_id: Option<PositionId>,
1563        }
1564
1565        let order_cell = self.orders.get(&client_order_id).cloned();
1566        let order_details = order_cell.as_ref().map(|cell| {
1567            let order = cell.borrow();
1568            OrderDetails {
1569                is_open: order.is_open(),
1570                instrument_id: order.instrument_id(),
1571                strategy_id: order.strategy_id(),
1572                account_id: order.account_id(),
1573                exec_algorithm_id: order.exec_algorithm_id(),
1574                exec_spawn_id: order.exec_spawn_id(),
1575                position_id: order.position_id(),
1576            }
1577        });
1578
1579        if order_details
1580            .as_ref()
1581            .is_some_and(|details| details.is_open)
1582        {
1583            log::warn!("Order {client_order_id} found open when purging, skipping purge");
1584            return false;
1585        }
1586
1587        if order_details.is_some() {
1588            self.orders.remove(&client_order_id);
1589        } else {
1590            log::warn!("Order {client_order_id} not found when purging");
1591        }
1592
1593        let indexed_position_id = self.index.order_position.remove(&client_order_id);
1594        let indexed_strategy_id = self.index.order_strategy.remove(&client_order_id);
1595        self.index.order_client.remove(&client_order_id);
1596        self.index.client_order_ids.remove(&client_order_id);
1597
1598        if let Some(details) = &order_details {
1599            if let Some(venue_orders) = self
1600                .index
1601                .venue_orders
1602                .get_mut(&details.instrument_id.venue)
1603            {
1604                venue_orders.remove(&client_order_id);
1605                if venue_orders.is_empty() {
1606                    self.index.venue_orders.remove(&details.instrument_id.venue);
1607                }
1608            }
1609
1610            // As with the strategy buckets below, an absent bucket is left absent: recreating
1611            // it would suppress the `index.instrument_positions` integrity check.
1612            // As with the strategy buckets below, an absent bucket is left absent: recreating
1613            // it would suppress the `index.instrument_positions` integrity check.
1614            let instrument_orders_became_empty = self
1615                .index
1616                .instrument_orders
1617                .get_mut(&details.instrument_id)
1618                .is_some_and(|instrument_orders| {
1619                    instrument_orders.remove(&client_order_id);
1620                    instrument_orders.is_empty()
1621                });
1622
1623            let has_instrument_positions = self
1624                .index
1625                .instrument_positions
1626                .get(&details.instrument_id)
1627                .is_some_and(|positions| !positions.is_empty());
1628
1629            if instrument_orders_became_empty && !has_instrument_positions {
1630                self.index.instrument_orders.remove(&details.instrument_id);
1631            }
1632
1633            if let Some(exec_algorithm_id) = details.exec_algorithm_id {
1634                let became_empty = self
1635                    .index
1636                    .exec_algorithm_orders
1637                    .get_mut(&exec_algorithm_id)
1638                    .is_some_and(|orders| {
1639                        orders.remove(&client_order_id);
1640                        orders.is_empty()
1641                    });
1642
1643                if became_empty {
1644                    self.index.exec_algorithm_orders.remove(&exec_algorithm_id);
1645                    self.index.exec_algorithms.remove(&exec_algorithm_id);
1646                }
1647            }
1648
1649            if let Some(account_id) = details.account_id
1650                && let Some(account_orders) = self.index.account_orders.get_mut(&account_id)
1651            {
1652                account_orders.remove(&client_order_id);
1653                if account_orders.is_empty() {
1654                    self.index.account_orders.remove(&account_id);
1655                }
1656            }
1657
1658            if let Some(exec_spawn_id) = details.exec_spawn_id
1659                && let Some(spawn_orders) = self.index.exec_spawn_orders.get_mut(&exec_spawn_id)
1660            {
1661                spawn_orders.remove(&client_order_id);
1662                if spawn_orders.is_empty() {
1663                    self.index.exec_spawn_orders.remove(&exec_spawn_id);
1664                }
1665            }
1666        }
1667
1668        let mut position_ids = AHashSet::new();
1669        if let Some(position_id) = indexed_position_id {
1670            position_ids.insert(position_id);
1671        }
1672
1673        if let Some(position_id) = order_details
1674            .as_ref()
1675            .and_then(|details| details.position_id)
1676        {
1677            position_ids.insert(position_id);
1678        }
1679
1680        let mut strategy_ids = AHashSet::new();
1681        if let Some(strategy_id) = indexed_strategy_id {
1682            strategy_ids.insert(strategy_id);
1683        }
1684
1685        if let Some(details) = &order_details {
1686            strategy_ids.insert(details.strategy_id);
1687        }
1688
1689        for position_id in position_ids {
1690            if self.positions.contains_key(&position_id) {
1691                if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
1692                    position_orders.remove(&client_order_id);
1693                }
1694                continue;
1695            }
1696
1697            let has_other_orders =
1698                if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
1699                    position_orders.remove(&client_order_id);
1700                    !position_orders.is_empty()
1701                } else {
1702                    self.index
1703                        .order_position
1704                        .values()
1705                        .any(|candidate| *candidate == position_id)
1706                };
1707
1708            if has_other_orders {
1709                continue;
1710            }
1711
1712            self.index.position_orders.remove(&position_id);
1713            if let Some(strategy_id) = self.index.position_strategy.remove(&position_id) {
1714                strategy_ids.insert(strategy_id);
1715                if let Some(strategy_positions) =
1716                    self.index.strategy_positions.get_mut(&strategy_id)
1717                {
1718                    strategy_positions.remove(&position_id);
1719                    if strategy_positions.is_empty() {
1720                        self.index.strategy_positions.remove(&strategy_id);
1721                    }
1722                }
1723            }
1724
1725            if let Some(details) = &order_details
1726                && let Some(venue_positions) = self
1727                    .index
1728                    .venue_positions
1729                    .get_mut(&details.instrument_id.venue)
1730            {
1731                venue_positions.remove(&position_id);
1732                if venue_positions.is_empty() {
1733                    self.index
1734                        .venue_positions
1735                        .remove(&details.instrument_id.venue);
1736                }
1737            }
1738        }
1739
1740        for strategy_id in strategy_ids {
1741            // An absent reverse bucket is not an empty one: it means the index is already
1742            // inconsistent, possibly while another cached order still uses this strategy.
1743            // Retiring the registry entry here would both drop a live strategy from
1744            // `strategy_ids` and stop `check_integrity` reporting the missing bucket, so the
1745            // absent case is left exactly as found.
1746            let strategy_orders_became_empty = self
1747                .index
1748                .strategy_orders
1749                .get_mut(&strategy_id)
1750                .is_some_and(|strategy_orders| {
1751                    strategy_orders.remove(&client_order_id);
1752                    strategy_orders.is_empty()
1753                });
1754
1755            let has_positions = self
1756                .index
1757                .strategy_positions
1758                .get(&strategy_id)
1759                .is_some_and(|strategy_positions| !strategy_positions.is_empty());
1760
1761            if strategy_orders_became_empty && !has_positions {
1762                self.index.strategy_orders.remove(&strategy_id);
1763                self.index.strategies.remove(&strategy_id);
1764            }
1765        }
1766
1767        self.index.exec_spawn_orders.remove(&client_order_id);
1768
1769        self.index.orders.remove(&client_order_id);
1770        self.index.orders_active_local.remove(&client_order_id);
1771        self.index.orders_open.remove(&client_order_id);
1772        self.index.orders_closed.remove(&client_order_id);
1773        self.index.orders_emulated.remove(&client_order_id);
1774        self.index.orders_inflight.remove(&client_order_id);
1775        self.index.orders_pending_cancel.remove(&client_order_id);
1776
1777        if order_details.is_some() {
1778            log::info!("Purged order {client_order_id}");
1779        }
1780
1781        true
1782    }
1783
1784    /// Purges the position with the `position_id` from the cache (if found).
1785    ///
1786    /// For safety, a position is prevented from being purged if it's open.
1787    pub fn purge_position(&mut self, position_id: PositionId) {
1788        // Snapshot the position so we can release the borrow before mutating indexes.
1789        let position = self
1790            .positions
1791            .get(&position_id)
1792            .map(|cell| cell.borrow().clone());
1793
1794        // Prevent purging open positions
1795        if let Some(ref pos) = position
1796            && pos.is_open()
1797        {
1798            log::warn!("Position {position_id} found open when purging, skipping purge");
1799            return;
1800        }
1801
1802        // If position exists in cache, remove it and clean up position-specific indices
1803        if let Some(ref pos) = position {
1804            self.positions.remove(&position_id);
1805
1806            // Remove from venue positions index
1807            if let Some(venue_positions) =
1808                self.index.venue_positions.get_mut(&pos.instrument_id.venue)
1809            {
1810                venue_positions.remove(&position_id);
1811                if venue_positions.is_empty() {
1812                    self.index.venue_positions.remove(&pos.instrument_id.venue);
1813                }
1814            }
1815
1816            // Remove from instrument positions index
1817            let instrument_positions_became_empty = self
1818                .index
1819                .instrument_positions
1820                .get_mut(&pos.instrument_id)
1821                .is_some_and(|positions| {
1822                    positions.remove(&position_id);
1823                    positions.is_empty()
1824                });
1825
1826            if instrument_positions_became_empty {
1827                self.index.instrument_positions.remove(&pos.instrument_id);
1828                let instrument_orders_empty = self
1829                    .index
1830                    .instrument_orders
1831                    .get(&pos.instrument_id)
1832                    .is_some_and(|orders| orders.is_empty());
1833
1834                if instrument_orders_empty {
1835                    self.index.instrument_orders.remove(&pos.instrument_id);
1836                }
1837            }
1838
1839            // Remove from strategy positions index
1840            let strategy_positions_became_empty = self
1841                .index
1842                .strategy_positions
1843                .get_mut(&pos.strategy_id)
1844                .is_some_and(|positions| {
1845                    positions.remove(&position_id);
1846                    positions.is_empty()
1847                });
1848
1849            if strategy_positions_became_empty {
1850                self.index.strategy_positions.remove(&pos.strategy_id);
1851                let strategy_orders_empty = self
1852                    .index
1853                    .strategy_orders
1854                    .get(&pos.strategy_id)
1855                    .is_some_and(|orders| orders.is_empty());
1856
1857                if strategy_orders_empty {
1858                    self.index.strategy_orders.remove(&pos.strategy_id);
1859                    self.index.strategies.remove(&pos.strategy_id);
1860                }
1861            }
1862
1863            // Remove from account positions index
1864            if let Some(account_positions) = self.index.account_positions.get_mut(&pos.account_id) {
1865                account_positions.remove(&position_id);
1866                if account_positions.is_empty() {
1867                    self.index.account_positions.remove(&pos.account_id);
1868                }
1869            }
1870
1871            // Remove position ID from orders that reference it
1872            for client_order_id in pos.client_order_ids() {
1873                self.index.order_position.remove(&client_order_id);
1874            }
1875
1876            log::info!("Purged position {position_id}");
1877        } else {
1878            log::warn!("Position {position_id} not found when purging");
1879        }
1880
1881        // Always clean up position indices (even if position not in cache)
1882        self.index.position_strategy.remove(&position_id);
1883        self.index.position_oms.remove(&position_id);
1884        self.index.position_orders.remove(&position_id);
1885        self.index.positions.remove(&position_id);
1886        self.index.positions_open.remove(&position_id);
1887        self.index.positions_closed.remove(&position_id);
1888
1889        // Always clean up position snapshots (even if position not in cache)
1890        self.position_snapshots.remove(&position_id);
1891        self.bump_position_snapshot_revision(position_id);
1892    }
1893
1894    /// Purges the instrument with the `instrument_id` from the cache (if found).
1895    ///
1896    /// The instrument record and its market data are removed:
1897    /// any synthetic with the same id, order book and own-order-book state, quote/trade
1898    /// histories, mark/index/funding price histories, instrument status and close, bars
1899    /// for any `BarType` referencing the instrument. Order and position indexes remain
1900    /// while their records are retained, preserving historical queries and realized PnL.
1901    ///
1902    /// For safety, an instrument is prevented from being purged while any associated
1903    /// order is non-terminal (anything not in `orders_closed`, including
1904    /// initialized, submitted, accepted, emulated, released, or inflight states) or
1905    /// any associated position is non-closed.
1906    ///
1907    /// Active subscriptions and other live data-engine state are not touched here;
1908    /// those belong to the data and execution engines.
1909    ///
1910    /// # Warning
1911    ///
1912    /// Intended for actors and strategies that have their own lifecycle logic for
1913    /// deciding when an instrument is no longer needed. Purging an instrument that any
1914    /// other actor, strategy, or engine still relies on may cause incorrect behavior
1915    /// (missing instrument lookups, lost market-data history). The caller is
1916    /// responsible for ensuring the instrument is no longer in use before purging.
1917    fn purge_instrument_inner(&mut self, instrument_id: InstrumentId, skip_order_guard: bool) {
1918        #[cfg(feature = "defi")]
1919        let defi_found = self.defi.pools.contains_key(&instrument_id)
1920            || self.defi.pool_profilers.contains_key(&instrument_id);
1921        #[cfg(not(feature = "defi"))]
1922        let defi_found = false;
1923
1924        let found = self.instruments.contains_key(&instrument_id)
1925            || self.synthetics.contains_key(&instrument_id)
1926            || defi_found;
1927
1928        if !found {
1929            log::warn!("Instrument {instrument_id} not found when purging");
1930            return;
1931        }
1932
1933        if !skip_order_guard && let Some(orders) = self.index.instrument_orders.get(&instrument_id)
1934        {
1935            let has_non_terminal = orders
1936                .iter()
1937                .any(|client_order_id| !self.index.orders_closed.contains(client_order_id));
1938
1939            if has_non_terminal {
1940                log::warn!(
1941                    "Instrument {instrument_id} has non-terminal orders when purging, skipping purge"
1942                );
1943                return;
1944            }
1945        }
1946
1947        if let Some(positions) = self.index.instrument_positions.get(&instrument_id) {
1948            let has_non_closed = positions
1949                .iter()
1950                .any(|position_id| !self.index.positions_closed.contains(position_id));
1951
1952            if has_non_closed {
1953                log::warn!(
1954                    "Instrument {instrument_id} has non-closed positions when purging, skipping purge"
1955                );
1956                return;
1957            }
1958        }
1959
1960        self.instruments.remove(&instrument_id);
1961        self.synthetics.remove(&instrument_id);
1962        self.books.remove(&instrument_id);
1963        self.own_books.remove(&instrument_id);
1964        self.quotes.remove(&instrument_id);
1965        self.trades.remove(&instrument_id);
1966        self.mark_prices.remove(&instrument_id);
1967        self.index_prices.remove(&instrument_id);
1968        self.funding_rates.remove(&instrument_id);
1969        self.instrument_statuses.remove(&instrument_id);
1970        self.instrument_closes.remove(&instrument_id);
1971        self.greeks.remove(&instrument_id);
1972        self.option_greeks.remove(&instrument_id);
1973
1974        self.bars
1975            .retain(|bar_type, _| bar_type.instrument_id() != instrument_id);
1976
1977        #[cfg(feature = "defi")]
1978        {
1979            self.defi.pools.remove(&instrument_id);
1980            self.defi.pool_profilers.remove(&instrument_id);
1981        }
1982
1983        log::info!("Purged instrument {instrument_id}");
1984    }
1985
1986    /// Purges the instrument with the `instrument_id` from the cache.
1987    ///
1988    /// This refuses to purge when associated orders or positions remain in
1989    /// non-terminal state.
1990    pub fn purge_instrument(&mut self, instrument_id: InstrumentId) {
1991        self.purge_instrument_inner(instrument_id, false);
1992    }
1993
1994    /// Purges the instrument with the `instrument_id` from the cache while skipping the
1995    /// non-terminal order guard.
1996    ///
1997    /// This still refuses to purge when any associated position is non-closed. Intended
1998    /// for actors which own an instrument-expiration lifecycle and have already invalidated
1999    /// any remaining order state externally, but may still observe order-terminal events
2000    /// arriving later than the cleanup decision. Order objects and their indexes remain
2001    /// available even after the instrument record is removed.
2002    pub fn purge_instrument_skip_order_guard(&mut self, instrument_id: InstrumentId) {
2003        self.purge_instrument_inner(instrument_id, true);
2004    }
2005
2006    /// Purges all account state events which are outside the lookback window.
2007    ///
2008    /// Only events which are outside the lookback window will be purged.
2009    /// A value of 0 means purge all account state events.
2010    pub fn purge_account_events(&mut self, ts_now: UnixNanos, lookback_secs: u64) {
2011        log::debug!(
2012            "Purging account events{}",
2013            if lookback_secs > 0 {
2014                format!(" with lookback_secs={lookback_secs}")
2015            } else {
2016                String::new()
2017            }
2018        );
2019
2020        for account_cell in self.accounts.values() {
2021            let mut account = account_cell.borrow_mut();
2022            let event_count = account.event_count();
2023            account.purge_account_events(ts_now, lookback_secs);
2024            let count_diff = event_count - account.event_count();
2025            if count_diff > 0 {
2026                log::info!(
2027                    "Purged {} event(s) from account {}",
2028                    count_diff,
2029                    account.id()
2030                );
2031            }
2032        }
2033    }
2034
2035    /// Clears the caches index.
2036    pub fn clear_index(&mut self) {
2037        self.index.clear();
2038        log::debug!("Cleared index");
2039    }
2040
2041    /// Resets the cache.
2042    ///
2043    /// All stateful fields are reset to their initial value. Instruments,
2044    /// currencies, and synthetics are retained when `drop_instruments_on_reset`
2045    /// is `false` so that repeated backtest runs can reuse the same dataset. External order claims
2046    /// and execution client account, route, and external client registrations are retained so
2047    /// registered strategy and client routing remain configured across resets.
2048    pub fn reset(&mut self) {
2049        log::debug!("Resetting cache");
2050
2051        self.general.clear();
2052        self.books.clear();
2053        self.own_books.clear();
2054        self.quotes.clear();
2055        self.trades.clear();
2056        self.mark_xrates.clear();
2057        self.mark_prices.clear();
2058        self.index_prices.clear();
2059        self.funding_rates.clear();
2060        self.instrument_statuses.clear();
2061        self.instrument_closes.clear();
2062        self.bars.clear();
2063        self.accounts.clear();
2064        self.orders.clear();
2065        self.order_lists.clear();
2066        self.positions.clear();
2067        self.position_snapshots.clear();
2068        self.position_snapshot_revisions.clear();
2069        self.greeks.clear();
2070        self.option_greeks.clear();
2071        self.yield_curves.clear();
2072
2073        if self.config.drop_instruments_on_reset {
2074            self.currencies.clear();
2075            self.instruments.clear();
2076            self.synthetics.clear();
2077        }
2078
2079        #[cfg(feature = "defi")]
2080        {
2081            self.defi.pools.clear();
2082            self.defi.pool_profilers.clear();
2083        }
2084
2085        self.clear_index();
2086
2087        log::info!("Reset cache");
2088    }
2089
2090    /// Dispose of the cache which will close any underlying database adapter.
2091    ///
2092    /// If closing the database connection fails, an error is logged.
2093    pub fn dispose(&mut self) {
2094        self.reset();
2095
2096        if let Some(database) = &mut self.database
2097            && let Err(e) = database.close()
2098        {
2099            log::error!("Failed to close database during dispose: {e}");
2100        }
2101    }
2102
2103    /// Flushes the cache database, permanently removing the data the backing owns.
2104    ///
2105    /// Does nothing when no database is configured.
2106    ///
2107    /// # Errors
2108    ///
2109    /// Returns an error if flushing the database fails.
2110    pub fn flush_db(&mut self) -> anyhow::Result<()> {
2111        match &mut self.database {
2112            Some(database) => database.flush(),
2113            None => Ok(()),
2114        }
2115    }
2116
2117    /// Adds a raw bytes `value` to the cache under the `key`.
2118    ///
2119    /// The cache stores only raw bytes; interpretation is the caller's responsibility.
2120    ///
2121    /// # Errors
2122    ///
2123    /// Returns an error if persisting the entry to the backing database fails.
2124    pub fn add(&mut self, key: &str, value: Bytes) -> anyhow::Result<()> {
2125        // Keys are opaque and may embed exchange symbols, which need not be ASCII
2126        check_valid_string_utf8(key, stringify!(key))?;
2127        check_predicate_false(value.is_empty(), stringify!(value))?;
2128
2129        log::debug!("Adding general {key}");
2130        self.general.insert(key.to_string(), value.clone());
2131
2132        if let Some(database) = &mut self.database {
2133            database.add(key.to_string(), value)?;
2134        }
2135        Ok(())
2136    }
2137
2138    /// Adds an `OrderBook` to the cache.
2139    ///
2140    /// # Errors
2141    ///
2142    /// Returns an error if persisting the order book to the backing database fails.
2143    pub fn add_order_book(&mut self, book: OrderBook) -> anyhow::Result<()> {
2144        log::debug!("Adding `OrderBook` {}", book.instrument_id);
2145
2146        if self.config.save_market_data
2147            && let Some(database) = &mut self.database
2148        {
2149            database.add_order_book(&book)?;
2150        }
2151
2152        self.books.insert(book.instrument_id, book);
2153        Ok(())
2154    }
2155
2156    /// Adds an `OwnOrderBook` to the cache.
2157    ///
2158    /// # Errors
2159    ///
2160    /// Returns an error if persisting the own order book fails.
2161    pub fn add_own_order_book(&mut self, own_book: OwnOrderBook) -> anyhow::Result<()> {
2162        log::debug!("Adding `OwnOrderBook` {}", own_book.instrument_id);
2163
2164        self.own_books.insert(own_book.instrument_id, own_book);
2165        Ok(())
2166    }
2167
2168    /// Adds the `mark_price` update to the cache.
2169    ///
2170    /// # Errors
2171    ///
2172    /// Returns an error if persisting the mark price to the backing database fails.
2173    pub fn add_mark_price(&mut self, mark_price: MarkPriceUpdate) -> anyhow::Result<()> {
2174        log::debug!("Adding `MarkPriceUpdate` for {}", mark_price.instrument_id);
2175
2176        if self.config.save_market_data {
2177            // TODO: Placeholder and return Result for consistency
2178        }
2179
2180        let mark_prices_deque = self
2181            .mark_prices
2182            .entry(mark_price.instrument_id)
2183            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2184        mark_prices_deque.push_front(mark_price);
2185        Ok(())
2186    }
2187
2188    /// Adds the `index_price` update to the cache.
2189    ///
2190    /// # Errors
2191    ///
2192    /// Returns an error if persisting the index price to the backing database fails.
2193    pub fn add_index_price(&mut self, index_price: IndexPriceUpdate) -> anyhow::Result<()> {
2194        log::debug!(
2195            "Adding `IndexPriceUpdate` for {}",
2196            index_price.instrument_id
2197        );
2198
2199        if self.config.save_market_data {
2200            // TODO: Placeholder and return Result for consistency
2201        }
2202
2203        let index_prices_deque = self
2204            .index_prices
2205            .entry(index_price.instrument_id)
2206            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2207        index_prices_deque.push_front(index_price);
2208        Ok(())
2209    }
2210
2211    /// Adds the `funding_rate` update to the cache.
2212    ///
2213    /// # Errors
2214    ///
2215    /// Returns an error if persisting the funding rate update to the backing database fails.
2216    pub fn add_funding_rate(&mut self, funding_rate: FundingRateUpdate) -> anyhow::Result<()> {
2217        log::debug!(
2218            "Adding `FundingRateUpdate` for {}",
2219            funding_rate.instrument_id
2220        );
2221
2222        if self.config.save_market_data {
2223            // TODO: Placeholder and return Result for consistency
2224        }
2225
2226        let funding_rates_deque = self
2227            .funding_rates
2228            .entry(funding_rate.instrument_id)
2229            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2230        funding_rates_deque.push_front(funding_rate);
2231        Ok(())
2232    }
2233
2234    /// Adds the given `funding rates` to the cache.
2235    ///
2236    /// # Errors
2237    ///
2238    /// Returns an error if persisting the trade ticks to the backing database fails.
2239    pub fn add_funding_rates(&mut self, funding_rates: &[FundingRateUpdate]) -> anyhow::Result<()> {
2240        check_slice_not_empty(funding_rates, stringify!(funding_rates))?;
2241
2242        let instrument_id = funding_rates[0].instrument_id;
2243        log::debug!(
2244            "Adding `FundingRateUpdate`[{}] {instrument_id}",
2245            funding_rates.len()
2246        );
2247
2248        if self.config.save_market_data
2249            && let Some(database) = &mut self.database
2250        {
2251            for funding_rate in funding_rates {
2252                database.add_funding_rate(funding_rate)?;
2253            }
2254        }
2255
2256        let funding_rate_deque = self
2257            .funding_rates
2258            .entry(instrument_id)
2259            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2260
2261        for funding_rate in funding_rates {
2262            funding_rate_deque.push_front(*funding_rate);
2263        }
2264        Ok(())
2265    }
2266
2267    /// Adds the `instrument_status` update to the cache.
2268    ///
2269    /// # Errors
2270    ///
2271    /// Returns an error if persisting the instrument status to the backing database fails.
2272    pub fn add_instrument_status(&mut self, status: InstrumentStatus) -> anyhow::Result<()> {
2273        log::debug!("Adding `InstrumentStatus` for {}", status.instrument_id);
2274
2275        if self.config.save_market_data {
2276            // TODO: Placeholder and return Result for consistency
2277        }
2278
2279        let statuses_deque = self
2280            .instrument_statuses
2281            .entry(status.instrument_id)
2282            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2283        statuses_deque.push_front(status);
2284        Ok(())
2285    }
2286
2287    /// Adds an instrument close to the cache.
2288    ///
2289    /// A close already cached for the same instrument is overwritten. With a backing database, the
2290    /// replacement is queued for persistence before the in-memory value is updated. A later
2291    /// database error is logged by the adapter and does not roll back the cached value.
2292    ///
2293    /// # Errors
2294    ///
2295    /// Returns an error if the close cannot be queued for persistence.
2296    pub fn add_instrument_close(&mut self, close: InstrumentClose) -> anyhow::Result<()> {
2297        log::debug!("Adding `InstrumentClose` for {}", close.instrument_id);
2298
2299        if let Some(database) = &self.database {
2300            database.add_instrument_close(&close)?;
2301        }
2302
2303        self.instrument_closes.insert(close.instrument_id, close);
2304        Ok(())
2305    }
2306
2307    /// Adds the `quote` tick to the cache.
2308    ///
2309    /// # Errors
2310    ///
2311    /// Returns an error if persisting the quote tick to the backing database fails.
2312    pub fn add_quote(&mut self, quote: QuoteTick) -> anyhow::Result<()> {
2313        log::debug!("Adding `QuoteTick` {}", quote.instrument_id);
2314
2315        if self.config.save_market_data
2316            && let Some(database) = &mut self.database
2317        {
2318            database.add_quote(&quote)?;
2319        }
2320
2321        let quotes_deque = self
2322            .quotes
2323            .entry(quote.instrument_id)
2324            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2325        quotes_deque.push_front(quote);
2326        Ok(())
2327    }
2328
2329    /// Adds the `quotes` to the cache.
2330    ///
2331    /// # Errors
2332    ///
2333    /// Returns an error if persisting the quote ticks to the backing database fails.
2334    pub fn add_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
2335        check_slice_not_empty(quotes, stringify!(quotes))?;
2336
2337        let instrument_id = quotes[0].instrument_id;
2338        log::debug!("Adding `QuoteTick`[{}] {instrument_id}", quotes.len());
2339
2340        if self.config.save_market_data
2341            && let Some(database) = &mut self.database
2342        {
2343            for quote in quotes {
2344                database.add_quote(quote)?;
2345            }
2346        }
2347
2348        let quotes_deque = self
2349            .quotes
2350            .entry(instrument_id)
2351            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2352
2353        for quote in quotes {
2354            quotes_deque.push_front(*quote);
2355        }
2356        Ok(())
2357    }
2358
2359    /// Adds the `trade` tick to the cache.
2360    ///
2361    /// # Errors
2362    ///
2363    /// Returns an error if persisting the trade tick to the backing database fails.
2364    pub fn add_trade(&mut self, trade: TradeTick) -> anyhow::Result<()> {
2365        log::debug!("Adding `TradeTick` {}", trade.instrument_id);
2366
2367        if self.config.save_market_data
2368            && let Some(database) = &mut self.database
2369        {
2370            database.add_trade(&trade)?;
2371        }
2372
2373        let trades_deque = self
2374            .trades
2375            .entry(trade.instrument_id)
2376            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2377        trades_deque.push_front(trade);
2378        Ok(())
2379    }
2380
2381    /// Adds the give `trades` to the cache.
2382    ///
2383    /// # Errors
2384    ///
2385    /// Returns an error if persisting the trade ticks to the backing database fails.
2386    pub fn add_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
2387        check_slice_not_empty(trades, stringify!(trades))?;
2388
2389        let instrument_id = trades[0].instrument_id;
2390        log::debug!("Adding `TradeTick`[{}] {instrument_id}", trades.len());
2391
2392        if self.config.save_market_data
2393            && let Some(database) = &mut self.database
2394        {
2395            for trade in trades {
2396                database.add_trade(trade)?;
2397            }
2398        }
2399
2400        let trades_deque = self
2401            .trades
2402            .entry(instrument_id)
2403            .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2404
2405        for trade in trades {
2406            trades_deque.push_front(*trade);
2407        }
2408        Ok(())
2409    }
2410
2411    /// Adds the `bar` to the cache, keeping the per-`bar_type` series newest-first.
2412    ///
2413    /// A newer bar is pushed, an older `ts_event` is skipped, and an equal
2414    /// `ts_event` replaces the front bar for time bars.
2415    ///
2416    /// # Errors
2417    ///
2418    /// Returns an error if persisting the bar to the backing database fails.
2419    pub fn add_bar(&mut self, bar: Bar) -> anyhow::Result<()> {
2420        log::debug!("Adding `Bar` {}", bar.bar_type);
2421
2422        if self.config.save_market_data
2423            && let Some(database) = &mut self.database
2424        {
2425            database.add_bar(&bar)?;
2426        }
2427
2428        let bars = self
2429            .bars
2430            .entry(bar.bar_type)
2431            .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
2432        insert_bar(bars, bar);
2433        Ok(())
2434    }
2435
2436    /// Adds the `bars` to the cache, each following [`Cache::add_bar`].
2437    ///
2438    /// # Errors
2439    ///
2440    /// Returns an error if persisting the bars to the backing database fails.
2441    pub fn add_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
2442        check_slice_not_empty(bars, stringify!(bars))?;
2443
2444        let bar_type = bars[0].bar_type;
2445        log::debug!("Adding `Bar`[{}] {bar_type}", bars.len());
2446
2447        if self.config.save_market_data
2448            && let Some(database) = &mut self.database
2449        {
2450            for bar in bars {
2451                database.add_bar(bar)?;
2452            }
2453        }
2454
2455        let bars_deque = self
2456            .bars
2457            .entry(bar_type)
2458            .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
2459
2460        for bar in bars {
2461            insert_bar(bars_deque, *bar);
2462        }
2463        Ok(())
2464    }
2465
2466    /// Adds the historical `bar` at its ordered position in the series.
2467    ///
2468    /// Request-generated bars are added this way, since the cache is their only
2469    /// delivery path: bars older than the front are kept rather than skipped.
2470    ///
2471    /// # Errors
2472    ///
2473    /// Returns an error if persisting the bar to the backing database fails.
2474    pub fn add_bar_historical(&mut self, bar: Bar) -> anyhow::Result<()> {
2475        log::debug!("Adding historical `Bar` {}", bar.bar_type);
2476
2477        if self.config.save_market_data
2478            && let Some(database) = &mut self.database
2479        {
2480            database.add_bar(&bar)?;
2481        }
2482
2483        let bars = self
2484            .bars
2485            .entry(bar.bar_type)
2486            .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
2487        insert_bar_historical(bars, bar);
2488        Ok(())
2489    }
2490
2491    /// Adds the `greeks` data to the cache.
2492    ///
2493    /// # Errors
2494    ///
2495    /// Returns an error if persisting the greeks data to the backing database fails.
2496    pub fn add_greeks(&mut self, greeks: GreeksData) -> anyhow::Result<()> {
2497        log::debug!("Adding `GreeksData` {}", greeks.instrument_id);
2498
2499        if self.config.save_market_data
2500            && let Some(_database) = &mut self.database
2501        {
2502            // TODO: Implement database.add_greeks(&greeks) when database adapter is updated
2503        }
2504
2505        self.greeks.insert(greeks.instrument_id, greeks);
2506        Ok(())
2507    }
2508
2509    /// Gets the greeks data for the `instrument_id`.
2510    pub fn greeks(&self, instrument_id: &InstrumentId) -> Option<GreeksData> {
2511        self.greeks.get(instrument_id).cloned()
2512    }
2513
2514    /// Adds exchange-provided option greeks to the cache.
2515    pub fn add_option_greeks(&mut self, greeks: OptionGreeks) {
2516        log::debug!("Adding `OptionGreeks` {}", greeks.instrument_id);
2517        self.option_greeks.insert(greeks.instrument_id, greeks);
2518    }
2519
2520    /// Gets a reference to the exchange-provided option greeks for the `instrument_id`.
2521    #[must_use]
2522    pub fn option_greeks(&self, instrument_id: &InstrumentId) -> Option<&OptionGreeks> {
2523        self.option_greeks.get(instrument_id)
2524    }
2525
2526    /// Adds the `yield_curve` data to the cache.
2527    ///
2528    /// # Errors
2529    ///
2530    /// Returns an error if:
2531    /// - The yield curve fails [`YieldCurve::validate`].
2532    /// - Persisting the yield curve data to the backing database fails.
2533    pub fn add_yield_curve(&mut self, yield_curve: YieldCurve) -> anyhow::Result<()> {
2534        yield_curve.validate()?;
2535
2536        log::debug!("Adding `YieldCurve` {}", yield_curve.curve_name);
2537
2538        if self.config.save_market_data
2539            && let Some(_database) = &mut self.database
2540        {
2541            // TODO: Implement database.add_yield_curve(&yield_curve) when database adapter is updated
2542        }
2543
2544        self.yield_curves
2545            .insert(yield_curve.curve_name.clone(), yield_curve);
2546        Ok(())
2547    }
2548
2549    /// Gets the yield curve for the `key`.
2550    pub fn yield_curve(&self, key: &str) -> Option<Box<dyn Fn(f64) -> f64>> {
2551        self.yield_curves.get(key).map(|curve| {
2552            let curve_clone = curve.clone();
2553            Box::new(move |expiry_in_years: f64| curve_clone.get_rate(expiry_in_years))
2554                as Box<dyn Fn(f64) -> f64>
2555        })
2556    }
2557
2558    /// Adds the `currency` to the cache.
2559    ///
2560    /// # Errors
2561    ///
2562    /// Returns an error if persisting the currency to the backing database fails.
2563    pub fn add_currency(&mut self, currency: Currency) -> anyhow::Result<()> {
2564        if self.currencies.contains_key(&currency.code) {
2565            return Ok(());
2566        }
2567        log::debug!("Adding `Currency` {}", currency.code);
2568
2569        if let Some(database) = &mut self.database {
2570            database.add_currency(&currency)?;
2571        }
2572
2573        self.currencies.insert(currency.code, currency);
2574        Ok(())
2575    }
2576
2577    /// Adds the `instrument` to the cache.
2578    ///
2579    /// # Errors
2580    ///
2581    /// Returns an error if persisting the instrument to the backing database fails.
2582    pub fn add_instrument(&mut self, instrument: InstrumentAny) -> anyhow::Result<()> {
2583        log::debug!("Adding `Instrument` {}", instrument.id());
2584
2585        // Ensure currencies exist in cache - safe to call repeatedly as add_currency is idempotent
2586        if let Some(base_currency) = instrument.base_currency() {
2587            self.add_currency(base_currency)?;
2588        }
2589        self.add_currency(instrument.quote_currency())?;
2590        self.add_currency(instrument.settlement_currency())?;
2591
2592        if let Some(database) = &mut self.database {
2593            database.add_instrument(&instrument)?;
2594        }
2595
2596        self.instruments.insert(instrument.id(), instrument);
2597        Ok(())
2598    }
2599
2600    /// Adds the `synthetic` instrument to the cache.
2601    ///
2602    /// # Errors
2603    ///
2604    /// Returns an error if persisting the synthetic instrument to the backing database fails.
2605    pub fn add_synthetic(&mut self, synthetic: SyntheticInstrument) -> anyhow::Result<()> {
2606        log::debug!("Adding `SyntheticInstrument` {}", synthetic.id);
2607
2608        if let Some(database) = &mut self.database {
2609            database.add_synthetic(&synthetic)?;
2610        }
2611
2612        self.synthetics.insert(synthetic.id, synthetic);
2613        Ok(())
2614    }
2615
2616    /// The currencies an account state references: its balances, its margins and its base currency.
2617    fn state_currencies(state: &AccountState) -> impl Iterator<Item = Currency> + '_ {
2618        state
2619            .balances
2620            .iter()
2621            .map(|balance| balance.currency)
2622            .chain(state.margins.iter().map(|margin| margin.currency))
2623            .chain(state.base_currency)
2624    }
2625
2626    /// Persists the currencies an account references, so a restart can restore them.
2627    ///
2628    /// An account can hold collateral in a currency no instrument carries, and a `Currency`
2629    /// persists as its bare code, so without this the account cannot be decoded in a process
2630    /// where that code was never registered. Mirrors what [`Self::add_instrument`] does for an
2631    /// instrument's base, quote and settlement currencies.
2632    fn add_account_currencies(&mut self, account: &AccountAny) -> anyhow::Result<()> {
2633        let Some(state) = account.last_event() else {
2634            return Ok(());
2635        };
2636
2637        for currency in Self::state_currencies(&state) {
2638            self.add_currency(currency)?;
2639        }
2640        Ok(())
2641    }
2642
2643    /// Adds the `account` to the cache.
2644    ///
2645    /// # Errors
2646    ///
2647    /// Returns an error if persisting the account or its currencies to the backing database fails.
2648    pub fn add_account(&mut self, account: AccountAny) -> anyhow::Result<()> {
2649        log::debug!("Adding `Account` {}", account.id());
2650
2651        self.add_account_currencies(&account)?;
2652
2653        if let Some(database) = &mut self.database {
2654            database.add_account(&account)?;
2655        }
2656
2657        let account_id = account.id();
2658        self.accounts.insert(account_id, SharedCell::new(account));
2659        self.index.add_venue_account(account_id);
2660        Ok(())
2661    }
2662
2663    /// Indexes the `client_order_id` with the `venue_order_id`.
2664    ///
2665    /// The `overwrite` parameter determines whether to overwrite any existing cached identifier.
2666    ///
2667    /// # Errors
2668    ///
2669    /// Returns an error if the client already has a different venue order ID and `overwrite` is
2670    /// false, or if the venue order ID is owned by a different client order.
2671    pub fn add_venue_order_id(
2672        &mut self,
2673        client_order_id: &ClientOrderId,
2674        venue_order_id: &VenueOrderId,
2675        overwrite: bool,
2676    ) -> anyhow::Result<()> {
2677        self.validate_venue_order_id_claim(client_order_id, venue_order_id, overwrite)?;
2678
2679        self.index
2680            .client_order_ids
2681            .insert(*client_order_id, *venue_order_id);
2682        self.index
2683            .venue_order_ids
2684            .insert(*venue_order_id, *client_order_id);
2685
2686        Ok(())
2687    }
2688
2689    /// Indexes the reverse alias `venue_order_id` to `client_order_id` for routing.
2690    ///
2691    /// Unlike [`Cache::add_venue_order_id`], an existing forward mapping for the client
2692    /// order is never moved, so superseded venue order ID generations from mass status
2693    /// reports register idempotently. The forward mapping's authority stays with order
2694    /// event application via [`Cache::update_order`].
2695    ///
2696    /// # Errors
2697    ///
2698    /// Returns an error if the venue order ID is owned by a different client order.
2699    pub fn index_venue_order_id(
2700        &mut self,
2701        client_order_id: &ClientOrderId,
2702        venue_order_id: &VenueOrderId,
2703    ) -> anyhow::Result<()> {
2704        self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
2705
2706        self.index
2707            .venue_order_ids
2708            .insert(*venue_order_id, *client_order_id);
2709        self.index
2710            .client_order_ids
2711            .entry(*client_order_id)
2712            .or_insert(*venue_order_id);
2713
2714        Ok(())
2715    }
2716
2717    fn validate_venue_order_id_claim(
2718        &self,
2719        client_order_id: &ClientOrderId,
2720        venue_order_id: &VenueOrderId,
2721        overwrite: bool,
2722    ) -> anyhow::Result<()> {
2723        self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
2724
2725        if let Some(existing_venue_order_id) = self.index.client_order_ids.get(client_order_id)
2726            && !overwrite
2727            && existing_venue_order_id != venue_order_id
2728        {
2729            anyhow::bail!(
2730                "Existing {existing_venue_order_id} for {client_order_id}
2731                    did not match the given {venue_order_id}.
2732                    If you are writing a test then try a different `venue_order_id`,
2733                    otherwise this is probably a bug."
2734            );
2735        }
2736
2737        Ok(())
2738    }
2739
2740    fn validate_venue_order_id_ownership(
2741        &self,
2742        client_order_id: &ClientOrderId,
2743        venue_order_id: &VenueOrderId,
2744    ) -> anyhow::Result<()> {
2745        if let Some(existing_client_order_id) = self.index.venue_order_ids.get(venue_order_id)
2746            && existing_client_order_id != client_order_id
2747        {
2748            return Err(VenueOrderIdOwnershipError {
2749                venue_order_id: *venue_order_id,
2750                existing_client_order_id: *existing_client_order_id,
2751                claimant_client_order_id: *client_order_id,
2752            }
2753            .into());
2754        }
2755
2756        Ok(())
2757    }
2758
2759    /// Adds the `order` to the cache indexed with any given identifiers.
2760    ///
2761    /// # Parameters
2762    ///
2763    /// `override_existing`: If the added order should 'override' any existing order and replace
2764    /// it in the cache. This is currently used for emulated orders which are
2765    /// being released and transformed into another type.
2766    ///
2767    /// # Errors
2768    ///
2769    /// Returns an error if not `replace_existing` and the `order.client_order_id` is already contained in the cache,
2770    /// or if persisting the order to the backing database fails. The order and every index are
2771    /// committed to memory before persistence is attempted, so a persistence error leaves the
2772    /// cache internally consistent.
2773    pub fn add_order(
2774        &mut self,
2775        order: OrderAny,
2776        position_id: Option<PositionId>,
2777        client_id: Option<ClientId>,
2778        replace_existing: bool,
2779    ) -> anyhow::Result<()> {
2780        let instrument_id = order.instrument_id();
2781        let venue = instrument_id.venue;
2782        let client_order_id = order.client_order_id();
2783        let strategy_id = order.strategy_id();
2784        let exec_algorithm_id = order.exec_algorithm_id();
2785        let exec_spawn_id = order.exec_spawn_id();
2786
2787        if !replace_existing {
2788            check_key_not_in_map(
2789                &client_order_id,
2790                &self.orders,
2791                stringify!(client_order_id),
2792                stringify!(orders),
2793            )?;
2794        }
2795
2796        log::debug!("Adding {order:?}");
2797
2798        self.index.orders.insert(client_order_id);
2799
2800        if order.is_active_local() {
2801            self.index.orders_active_local.insert(client_order_id);
2802        }
2803        self.index
2804            .order_strategy
2805            .insert(client_order_id, strategy_id);
2806        self.index.strategies.insert(strategy_id);
2807
2808        // Update venue -> orders index
2809        self.index
2810            .venue_orders
2811            .entry(venue)
2812            .or_default()
2813            .insert(client_order_id);
2814
2815        // Update instrument -> orders index
2816        self.index
2817            .instrument_orders
2818            .entry(instrument_id)
2819            .or_default()
2820            .insert(client_order_id);
2821
2822        // Update strategy -> orders index
2823        self.index
2824            .strategy_orders
2825            .entry(strategy_id)
2826            .or_default()
2827            .insert(client_order_id);
2828
2829        // Update account -> orders index (if account_id known at creation)
2830        if let Some(account_id) = order.account_id() {
2831            self.index
2832                .account_orders
2833                .entry(account_id)
2834                .or_default()
2835                .insert(client_order_id);
2836        }
2837
2838        // Update exec_algorithm -> orders index
2839        if let Some(exec_algorithm_id) = exec_algorithm_id {
2840            self.index.exec_algorithms.insert(exec_algorithm_id);
2841
2842            self.index
2843                .exec_algorithm_orders
2844                .entry(exec_algorithm_id)
2845                .or_default()
2846                .insert(client_order_id);
2847        }
2848
2849        // Update exec_spawn -> orders index
2850        if let Some(exec_spawn_id) = exec_spawn_id {
2851            self.index
2852                .exec_spawn_orders
2853                .entry(exec_spawn_id)
2854                .or_default()
2855                .insert(client_order_id);
2856        }
2857
2858        // Update emulation index
2859        if order.emulation_trigger().is_some() {
2860            self.index.orders_emulated.insert(client_order_id);
2861        }
2862
2863        // Index position ID if provided
2864        if let Some(position_id) = position_id {
2865            self.index_position_id_in_memory(&position_id, &venue, &client_order_id, &strategy_id);
2866        }
2867
2868        // Index client ID if provided
2869        if let Some(client_id) = client_id {
2870            self.index.order_client.insert(client_order_id, client_id);
2871            log::debug!("Indexed {client_id:?}");
2872        }
2873
2874        // Reuse the existing cell on replace so the canonical entry stays in place
2875        // rather than orphaning a stale cell.
2876        let order_cell = if let Some(order_cell) = self.orders.get(&client_order_id) {
2877            *order_cell.borrow_mut() = order;
2878            order_cell.clone()
2879        } else {
2880            let order_cell = SharedCell::new(order);
2881            self.orders.insert(client_order_id, order_cell.clone());
2882            order_cell
2883        };
2884
2885        if let Some(position_id) = position_id {
2886            self.persist_position_id(&position_id, &client_order_id)?;
2887        }
2888
2889        if let Some(database) = &mut self.database {
2890            database.add_order(&order_cell.borrow(), client_id)?;
2891            // TODO: Implement
2892            // if self.config.snapshot_orders {
2893            //     database.snapshot_order_state(order)?;
2894            // }
2895        }
2896
2897        Ok(())
2898    }
2899
2900    /// Claims the execution-client origin for one or more cached orders.
2901    ///
2902    /// Claims are write-once: an unclaimed order is assigned to `client_id`, a matching existing
2903    /// claim is idempotent, and a conflicting claim is rejected. The complete batch is validated
2904    /// and its persistence command is successfully enqueued before any in-memory index is
2905    /// changed.
2906    ///
2907    /// # Errors
2908    ///
2909    /// Returns an error if an order is not cached, an order is already claimed by another client,
2910    /// the same order has conflicting claims in the batch, or persistence cannot be enqueued.
2911    pub fn claim_order_clients(
2912        &mut self,
2913        claims: &[(ClientOrderId, ClientId)],
2914    ) -> anyhow::Result<()> {
2915        let mut requested = AHashMap::with_capacity(claims.len());
2916        let mut ordered_claims = Vec::with_capacity(claims.len());
2917
2918        for (client_order_id, client_id) in claims {
2919            if let Some(existing_client_id) = requested.get(client_order_id) {
2920                if existing_client_id != client_id {
2921                    anyhow::bail!(
2922                        "Conflicting execution client claims for {client_order_id}: \
2923                         {existing_client_id} and {client_id}"
2924                    );
2925                }
2926                continue;
2927            }
2928
2929            requested.insert(*client_order_id, *client_id);
2930            ordered_claims.push((*client_order_id, *client_id));
2931        }
2932
2933        let mut pending_claims = Vec::with_capacity(ordered_claims.len());
2934        for (client_order_id, client_id) in ordered_claims {
2935            if !self.orders.contains_key(&client_order_id) {
2936                return Err(OrderLookupError::not_found(client_order_id).into());
2937            }
2938
2939            match self.index.order_client.get(&client_order_id) {
2940                Some(existing_client_id) if *existing_client_id == client_id => {}
2941                Some(existing_client_id) => {
2942                    anyhow::bail!(
2943                        "Order {client_order_id} is already claimed by execution client \
2944                         {existing_client_id} and cannot be claimed by {client_id}"
2945                    );
2946                }
2947                None => pending_claims.push((client_order_id, client_id)),
2948            }
2949        }
2950
2951        if pending_claims.is_empty() {
2952            return Ok(());
2953        }
2954
2955        if let Some(database) = &self.database {
2956            database.index_order_clients(&pending_claims)?;
2957        }
2958
2959        for (client_order_id, client_id) in pending_claims {
2960            self.index.order_client.insert(client_order_id, client_id);
2961            log::debug!("Claimed {client_order_id} for execution client {client_id}");
2962        }
2963
2964        Ok(())
2965    }
2966
2967    /// Adds the `order_list` to the cache.
2968    ///
2969    /// # Errors
2970    ///
2971    /// Returns an error if the order list ID is already contained in the cache.
2972    pub fn add_order_list(&mut self, order_list: OrderList) -> anyhow::Result<()> {
2973        let order_list_id = order_list.id;
2974        check_key_not_in_map(
2975            &order_list_id,
2976            &self.order_lists,
2977            stringify!(order_list_id),
2978            stringify!(order_lists),
2979        )?;
2980
2981        log::debug!("Adding {order_list}");
2982        self.order_lists.insert(order_list_id, order_list);
2983        Ok(())
2984    }
2985
2986    /// Indexes the `position_id` with the other given IDs.
2987    ///
2988    /// A cached `EXTERNAL` position retains its ownership when an order from another strategy
2989    /// is linked to it. Otherwise, the supplied `strategy_id` applies.
2990    ///
2991    /// # Errors
2992    ///
2993    /// Returns an error if indexing position ID in the backing database fails. The complete index
2994    /// operation is committed to memory before persistence is attempted, so a persistence error
2995    /// leaves the cache internally consistent.
2996    pub fn add_position_id(
2997        &mut self,
2998        position_id: &PositionId,
2999        venue: &Venue,
3000        client_order_id: &ClientOrderId,
3001        strategy_id: &StrategyId,
3002    ) -> anyhow::Result<()> {
3003        self.index_position_id_in_memory(position_id, venue, client_order_id, strategy_id);
3004        self.persist_position_id(position_id, client_order_id)
3005    }
3006
3007    fn index_position_id_in_memory(
3008        &mut self,
3009        position_id: &PositionId,
3010        venue: &Venue,
3011        client_order_id: &ClientOrderId,
3012        strategy_id: &StrategyId,
3013    ) {
3014        self.index
3015            .order_position
3016            .insert(*client_order_id, *position_id);
3017        self.index_position(position_id, venue, strategy_id);
3018        self.index
3019            .position_orders
3020            .entry(*position_id)
3021            .or_default()
3022            .insert(*client_order_id);
3023    }
3024
3025    fn persist_position_id(
3026        &mut self,
3027        position_id: &PositionId,
3028        client_order_id: &ClientOrderId,
3029    ) -> anyhow::Result<()> {
3030        if let Some(database) = &mut self.database {
3031            database.index_order_position(*client_order_id, *position_id)?;
3032        }
3033
3034        Ok(())
3035    }
3036
3037    fn index_position(
3038        &mut self,
3039        position_id: &PositionId,
3040        venue: &Venue,
3041        strategy_id: &StrategyId,
3042    ) {
3043        let strategy_id = self
3044            .positions
3045            .get(position_id)
3046            .map(|position| position.borrow().strategy_id)
3047            .filter(StrategyId::is_external)
3048            .unwrap_or(*strategy_id);
3049
3050        // Index: PositionId -> StrategyId
3051        self.index
3052            .position_strategy
3053            .insert(*position_id, strategy_id);
3054
3055        // Every position has a reverse-order bucket, including orderless positions.
3056        self.index.position_orders.entry(*position_id).or_default();
3057
3058        // Index: StrategyId -> set[PositionId]
3059        self.index
3060            .strategy_positions
3061            .entry(strategy_id)
3062            .or_default()
3063            .insert(*position_id);
3064
3065        // Index: Venue -> set[PositionId]
3066        self.index
3067            .venue_positions
3068            .entry(*venue)
3069            .or_default()
3070            .insert(*position_id);
3071    }
3072
3073    // Propagates parent OTO `position_id` to contingent children that are missing one.
3074    //
3075    // Recovers from a partial-write window during fill handling: the fill-time path in the
3076    // execution engine assigns `position_id` to each contingent child in a non-atomic loop
3077    // (`set_position_id` then `add_position_id`), so a crash mid-loop can leave the database
3078    // with the parent updated and some children un-updated. This pass re-applies any missing
3079    // assignments after load.
3080    fn assign_position_ids_to_contingencies(&mut self) {
3081        let mut assignments: Vec<(PositionId, ClientOrderId)> = Vec::new();
3082
3083        for parent_order_cell in self.orders.values() {
3084            let parent = parent_order_cell.borrow();
3085            if parent.contingency_type() != Some(ContingencyType::Oto) {
3086                continue;
3087            }
3088            let Some(parent_position_id) = parent.position_id() else {
3089                continue;
3090            };
3091            let Some(linked_order_ids) = parent.linked_order_ids() else {
3092                continue;
3093            };
3094
3095            for client_order_id in linked_order_ids {
3096                match self.orders.get(client_order_id) {
3097                    None => {
3098                        log::error!("Contingency order {client_order_id} not found");
3099                    }
3100                    Some(contingent_order_cell) => {
3101                        if contingent_order_cell.borrow().position_id().is_none() {
3102                            assignments.push((parent_position_id, *client_order_id));
3103                        }
3104                    }
3105                }
3106            }
3107        }
3108
3109        for (position_id, client_order_id) in assignments {
3110            let Some((venue, strategy_id)) = self.orders.get(&client_order_id).map(|order_cell| {
3111                let mut contingent = order_cell.borrow_mut();
3112                contingent.set_position_id(Some(position_id));
3113                (contingent.instrument_id().venue, contingent.strategy_id())
3114            }) else {
3115                continue;
3116            };
3117
3118            // Re-indexing through `add_position_id` also replays the database write, making the
3119            // recovered assignment durable across another restart.
3120            if let Err(e) =
3121                self.add_position_id(&position_id, &venue, &client_order_id, &strategy_id)
3122            {
3123                log::error!("Failed to re-index {client_order_id} -> {position_id}: {e}");
3124            }
3125        }
3126    }
3127
3128    /// Adds the `position` to the cache.
3129    ///
3130    /// # Errors
3131    ///
3132    /// Returns an error if persisting the position to the backing database fails. After
3133    /// serialization succeeds, the complete operation is committed to memory before persistence
3134    /// is attempted, so a persistence error leaves the cache internally consistent.
3135    pub fn add_position(&mut self, position: &Position, oms_type: OmsType) -> anyhow::Result<()> {
3136        self.add_position_inner(position.clone(), oms_type, true, false)
3137    }
3138
3139    /// Adds a position whose opening fill intentionally has no backing order.
3140    ///
3141    /// # Errors
3142    ///
3143    /// Returns an error if persisting the position to the backing database fails. After
3144    /// serialization succeeds, the complete operation is committed to memory before persistence
3145    /// is attempted, so a persistence error leaves the cache internally consistent.
3146    pub fn add_position_without_order(
3147        &mut self,
3148        position: &Position,
3149        oms_type: OmsType,
3150    ) -> anyhow::Result<()> {
3151        self.add_position_inner(position.clone(), oms_type, false, false)
3152    }
3153
3154    /// Replaces the cached position holding `position.id`, optionally moving the prior cycle's
3155    /// durable replay state into it.
3156    ///
3157    /// Pass `index_order` false when the opening fill intentionally has no backing order, matching
3158    /// [`Self::add_position_without_order`]. The carry runs under the same borrow as the swap, so a
3159    /// validation failure cannot leave the prior position stripped of its replay history.
3160    ///
3161    /// # Errors
3162    ///
3163    /// Returns an error if validating or persisting the position fails.
3164    pub fn replace_position(
3165        &mut self,
3166        position: &Position,
3167        oms_type: OmsType,
3168        index_order: bool,
3169        carry_replay_state: bool,
3170    ) -> anyhow::Result<()> {
3171        self.add_position_inner(position.clone(), oms_type, index_order, carry_replay_state)
3172    }
3173
3174    fn add_position_inner(
3175        &mut self,
3176        mut position: Position,
3177        oms_type: OmsType,
3178        index_order: bool,
3179        carry_replay_state: bool,
3180    ) -> anyhow::Result<()> {
3181        // Validate and serialize the OMS entry up front: both are construction failures, and
3182        // committing the position before they run would leave the cache mutated by one.
3183        // The key embeds the position ID, which embeds the instrument ID and venue
3184        // symbol, and exchange symbols may contain non-ASCII characters.
3185        let key = position_oms_key(position.id);
3186        check_valid_string_utf8(&key, stringify!(key))?;
3187        let value = Bytes::from(serde_json::to_vec(&oms_type)?);
3188        check_predicate_false(value.is_empty(), stringify!(value))?;
3189
3190        let position_id = position.id;
3191        let strategy_id = position.strategy_id;
3192        let instrument_id = position.instrument_id;
3193        let account_id = position.account_id;
3194        let opening_order_id = position.opening_order_id;
3195
3196        log::debug!("Adding {position}");
3197
3198        // Reuse the existing cell on a NETTING reopen so the prior value is replaced in place,
3199        // which also lets the carried replay state move out of it instead of being copied. The
3200        // transfer and the swap share one borrow, so a failure above cannot strip the prior.
3201        let position_cell = if let Some(position_cell) = self.positions.get(&position_id).cloned() {
3202            let mut prior = position_cell.borrow_mut();
3203            if carry_replay_state {
3204                position.transfer_replay_state_from(&mut prior);
3205            }
3206            *prior = position;
3207            drop(prior);
3208            position_cell
3209        } else {
3210            let position_cell = SharedCell::new(position);
3211            self.positions.insert(position_id, position_cell.clone());
3212            position_cell
3213        };
3214
3215        self.index.position_oms.insert(position_id, oms_type);
3216        self.index.positions.insert(position_id);
3217        self.index.positions_open.insert(position_id);
3218        self.index.positions_closed.remove(&position_id); // Cleanup for NETTING reopen
3219        self.index.strategies.insert(strategy_id);
3220        self.index.strategy_orders.entry(strategy_id).or_default();
3221
3222        if index_order {
3223            self.index_position_id_in_memory(
3224                &position_id,
3225                &instrument_id.venue,
3226                &opening_order_id,
3227                &strategy_id,
3228            );
3229        } else {
3230            self.index_position(&position_id, &instrument_id.venue, &strategy_id);
3231        }
3232
3233        // Index: InstrumentId -> AHashSet
3234        let instrument_positions = self
3235            .index
3236            .instrument_positions
3237            .entry(instrument_id)
3238            .or_default();
3239        instrument_positions.insert(position_id);
3240        self.index
3241            .instrument_orders
3242            .entry(instrument_id)
3243            .or_default();
3244
3245        // Index: AccountId -> AHashSet<PositionId>
3246        self.index
3247            .account_positions
3248            .entry(account_id)
3249            .or_default()
3250            .insert(position_id);
3251
3252        log::debug!("Adding general {key}");
3253        self.general.insert(key.clone(), value.clone());
3254
3255        if index_order {
3256            self.persist_position_id(&position_id, &opening_order_id)?;
3257        }
3258
3259        if let Some(database) = &mut self.database {
3260            database.add_position(&position_cell.borrow())?;
3261            // TODO: Implement position snapshots
3262            // if self.snapshot_positions {
3263            //     database.snapshot_position_state(
3264            //         position,
3265            //         position.ts_last,
3266            //         self.calculate_unrealized_pnl(&position),
3267            //     )?;
3268            // }
3269            database.add(key, value)?;
3270        }
3271
3272        Ok(())
3273    }
3274
3275    /// Updates the `account` in the cache.
3276    ///
3277    /// Reuses the existing cell when present so any held [`AccountRef`] handles continue to point
3278    /// at the canonical entry; only inserts a new cell when the account is unknown.
3279    ///
3280    /// # Errors
3281    ///
3282    /// Returns an error if updating the account in the database fails.
3283    pub fn update_account(&mut self, account: &AccountAny) -> anyhow::Result<()> {
3284        let account_id = account.id();
3285        match self.accounts.get(&account_id) {
3286            Some(account_cell) => *account_cell.borrow_mut() = account.clone(),
3287            None => {
3288                self.accounts
3289                    .insert(account_id, SharedCell::new(account.clone()));
3290            }
3291        }
3292
3293        self.add_account_currencies(account)?;
3294
3295        if let Some(database) = &mut self.database {
3296            database.update_account(account)?;
3297        }
3298        Ok(())
3299    }
3300
3301    /// Returns an owned `account`, removing its cache entry when ownership is exclusive.
3302    ///
3303    /// This supports hot paths which need owned account mutation without
3304    /// cloning the account event history. The cache is the sole owner of the
3305    /// account cell (the field is private and accessors only hand out
3306    /// lifetime-scoped [`AccountRef`] borrows). When the cache is the sole owner, the value is
3307    /// moved out of its cell rather than cloned.
3308    ///
3309    /// If another strong handle exists, the canonical entry remains cached and this returns
3310    /// `None`.
3311    #[must_use]
3312    pub fn take_account(&mut self, account_id: &AccountId) -> Option<AccountAny> {
3313        let cell = self.accounts.remove(account_id)?;
3314        let rc: Rc<RefCell<AccountAny>> = cell.into();
3315
3316        match Rc::try_unwrap(rc) {
3317            Ok(cell) => Some(cell.into_inner()),
3318            Err(rc) => {
3319                log::error!(
3320                    "Cannot move account {account_id} out of cache: account cell has an outstanding owner"
3321                );
3322                self.accounts.insert(*account_id, rc.into());
3323                None
3324            }
3325        }
3326    }
3327
3328    /// Caches the `account` in memory without updating the database.
3329    pub fn cache_account_owned(&mut self, account: AccountAny) {
3330        let account_id = account.id();
3331        self.index.add_venue_account(account_id);
3332        match self.accounts.get(&account_id) {
3333            Some(account_cell) => *account_cell.borrow_mut() = account,
3334            None => {
3335                self.accounts.insert(account_id, SharedCell::new(account));
3336            }
3337        }
3338    }
3339
3340    /// Updates the `account` in the cache, taking ownership of the updated account.
3341    ///
3342    /// # Errors
3343    ///
3344    /// Returns an error if updating the account in the database fails.
3345    pub fn update_account_owned(&mut self, account: AccountAny) -> anyhow::Result<()> {
3346        let account_id = account.id();
3347        // The in-memory account is updated before anything can fail, so a database failure while
3348        // saving its currencies never drops an account the caller has taken out of the cache.
3349        let currencies: Vec<Currency> = account
3350            .last_event()
3351            .map(|state| Self::state_currencies(&state).collect())
3352            .unwrap_or_default();
3353        self.cache_account_owned(account);
3354
3355        for currency in currencies {
3356            self.add_currency(currency)?;
3357        }
3358
3359        if let Some(database) = &mut self.database {
3360            let Some(account_cell) = self.accounts.get(&account_id) else {
3361                anyhow::bail!("Account {account_id} not found after cache update");
3362            };
3363            database.update_account(&account_cell.borrow())?;
3364        }
3365        Ok(())
3366    }
3367
3368    /// Applies an account state event to the cached account.
3369    ///
3370    /// Mutates the cached account in place to avoid cloning the account event
3371    /// history on the hot path; long-running sessions accumulate many events
3372    /// per account, so a snapshot-clone here would be O(history) per update.
3373    ///
3374    /// # Errors
3375    ///
3376    /// Returns an error if applying or persisting the account state fails.
3377    pub fn update_account_state(&mut self, event: &AccountState) -> anyhow::Result<()> {
3378        let Some(cell) = self.accounts.get(&event.account_id) else {
3379            return self.add_account(AccountAny::from_events(std::slice::from_ref(event))?);
3380        };
3381
3382        // The event is applied before anything can fail, so a database failure while saving its
3383        // currencies leaves the in-memory account current.
3384        cell.borrow_mut().apply(event.clone())?;
3385
3386        // The event carries every currency the account references after it, so the account
3387        // itself is not cloned on this path.
3388        for currency in Self::state_currencies(event) {
3389            self.add_currency(currency)?;
3390        }
3391
3392        let Some(cell) = self.accounts.get(&event.account_id) else {
3393            anyhow::bail!("Account {} not found after apply", event.account_id);
3394        };
3395
3396        if let Some(database) = &mut self.database {
3397            database.update_account(&cell.borrow())?;
3398        }
3399        Ok(())
3400    }
3401
3402    /// Replaces the cached `order` from a non-event snapshot.
3403    ///
3404    /// Prefer [`Self::update_order`] for lifecycle state changes. Use this only for order state
3405    /// that is not represented by [`OrderEventAny`].
3406    ///
3407    /// # Errors
3408    ///
3409    /// Returns an error if validation or persistence fails. After validation succeeds, the
3410    /// canonical order is committed to memory before its indexes and database are refreshed, so a
3411    /// persistence error leaves the cache internally consistent.
3412    pub fn replace_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
3413        let client_order_id = order.client_order_id();
3414        if let Some(venue_order_id) = order.venue_order_id() {
3415            self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
3416        }
3417
3418        match self.orders.get(&client_order_id) {
3419            // Reuse the existing cell so the canonical entry stays in place rather than
3420            // orphaning a stale cell.
3421            Some(order_cell) => *order_cell.borrow_mut() = order.clone(),
3422            None => {
3423                self.orders
3424                    .insert(client_order_id, SharedCell::new(order.clone()));
3425            }
3426        }
3427
3428        self.refresh_order(order)
3429    }
3430
3431    /// Updates the cached order by applying an event and refreshing derived cache state.
3432    ///
3433    /// # Errors
3434    ///
3435    /// Returns an error if the order is not found or rejects the event.
3436    pub fn update_order(&mut self, event: &OrderEventAny) -> anyhow::Result<OrderAny> {
3437        let event_client_order_id = event.client_order_id();
3438        let client_order_id = if self.order_exists(&event_client_order_id) {
3439            event_client_order_id
3440        } else if let Some(venue_order_id) = event.venue_order_id() {
3441            self.index
3442                .venue_order_ids
3443                .get(&venue_order_id)
3444                .copied()
3445                .ok_or(OrderError::NotFound(event_client_order_id))?
3446        } else {
3447            return Err(OrderError::NotFound(event_client_order_id).into());
3448        };
3449
3450        let order_cell = self
3451            .orders
3452            .get(&client_order_id)
3453            .cloned()
3454            .ok_or(OrderError::NotFound(client_order_id))?;
3455
3456        // Apply on a snapshot first so a fallible `apply` (e.g. invalid state
3457        // transition) leaves the canonical cell untouched. On success we swap the
3458        // post-event value back into the cell so subsequent reads see the new state.
3459        let mut snapshot = order_cell.borrow().clone();
3460        snapshot.apply(event.clone())?;
3461
3462        // Preflight only reverse ownership. A same-client forward mismatch remains a logged
3463        // refresh inconsistency, while other refresh failures, such as a backing database error,
3464        // remain logged after the canonical state is committed.
3465        if let Some(venue_order_id) = snapshot.venue_order_id() {
3466            self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
3467        }
3468
3469        *order_cell.borrow_mut() = snapshot.clone();
3470
3471        if matches!(event, OrderEventAny::FillVoided(_)) {
3472            self.reindex_hedging_position(client_order_id);
3473        }
3474
3475        if let Err(e) = self.refresh_order(&snapshot) {
3476            log::error!("Error updating order in cache: {e}");
3477        }
3478
3479        Ok(snapshot)
3480    }
3481
3482    fn refresh_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
3483        let client_order_id = order.client_order_id();
3484
3485        // Claim the venue order ID before mutating any other derived state. An updated event may
3486        // change the current ID for the same client order, while historical reverse aliases remain.
3487        if let Some(venue_order_id) = order.venue_order_id() {
3488            let overwrite = matches!(order.last_event(), OrderEventAny::Updated(_));
3489            if let Err(e) = self.add_venue_order_id(&client_order_id, &venue_order_id, overwrite) {
3490                if e.is::<VenueOrderIdOwnershipError>() {
3491                    return Err(e);
3492                }
3493                log::error!("Error indexing venue order ID in cache: {e}");
3494            }
3495        }
3496
3497        if order.is_active_local() {
3498            self.index.orders_active_local.insert(client_order_id);
3499        } else {
3500            self.index.orders_active_local.remove(&client_order_id);
3501        }
3502
3503        // Update in-flight state
3504        if order.is_inflight() {
3505            self.index.orders_inflight.insert(client_order_id);
3506        } else {
3507            self.index.orders_inflight.remove(&client_order_id);
3508        }
3509
3510        // Update open/closed state
3511        if order.is_open() {
3512            self.index.orders_closed.remove(&client_order_id);
3513            self.index.orders_open.insert(client_order_id);
3514        } else {
3515            self.index.orders_open.remove(&client_order_id);
3516        }
3517
3518        if order.is_closed() {
3519            self.index.orders_pending_cancel.remove(&client_order_id);
3520            self.index.orders_closed.insert(client_order_id);
3521        }
3522
3523        // A cancel rejection resolves the outstanding cancel request
3524        if matches!(order.last_event(), OrderEventAny::CancelRejected(_)) {
3525            self.index.orders_pending_cancel.remove(&client_order_id);
3526        }
3527
3528        // Update emulation index
3529        if order.emulation_trigger().is_some() && !order.is_closed() {
3530            self.index.orders_emulated.insert(client_order_id);
3531        } else {
3532            self.index.orders_emulated.remove(&client_order_id);
3533        }
3534
3535        // Update account orders index when account_id becomes available
3536        if let Some(account_id) = order.account_id() {
3537            self.index
3538                .account_orders
3539                .entry(account_id)
3540                .or_default()
3541                .insert(client_order_id);
3542        }
3543
3544        // Update own book
3545        if !self.own_books.is_empty() {
3546            let own_book = self.own_order_book(&order.instrument_id());
3547            if (own_book.is_some() && order.is_closed()) || should_handle_own_book_order(order) {
3548                self.update_own_order_book(order);
3549            }
3550        }
3551
3552        if let Some(database) = &mut self.database {
3553            database.update_order(order.last_event())?;
3554            // TODO: Implement order snapshots
3555            // if self.snapshot_orders {
3556            //     database.snapshot_order_state(order)?;
3557            // }
3558        }
3559
3560        Ok(())
3561    }
3562
3563    /// Updates the `order` as pending cancel locally.
3564    pub fn update_order_pending_cancel_local(&mut self, order: &OrderAny) {
3565        self.index
3566            .orders_pending_cancel
3567            .insert(order.client_order_id());
3568    }
3569
3570    /// Updates a `position` already held in the cache.
3571    ///
3572    /// Reuses the existing cell so any held [`PositionRef`] handles continue to point at the
3573    /// canonical entry.
3574    ///
3575    /// # Errors
3576    ///
3577    /// Returns an error if the position is not already held in the cache, or if updating the
3578    /// position in the database fails.
3579    pub fn update_position(&mut self, position: &Position) -> anyhow::Result<()> {
3580        let Some(position_cell) = self.positions.get(&position.id).cloned() else {
3581            anyhow::bail!("Cannot update position {}: not found in cache", position.id);
3582        };
3583
3584        self.refresh_position_indexes(position);
3585
3586        *position_cell.borrow_mut() = position.clone();
3587
3588        if let Some(database) = &mut self.database {
3589            database.update_position(position)?;
3590            // TODO: Implement order snapshots
3591            // if self.snapshot_orders {
3592            //     database.snapshot_order_state(order)?;
3593            // }
3594        }
3595
3596        Ok(())
3597    }
3598
3599    /// Updates a cached position by applying an order fill in place.
3600    ///
3601    /// Returns a transient copy of the updated state without stored history. The canonical cached
3602    /// position retains its complete history.
3603    ///
3604    /// # Errors
3605    ///
3606    /// Returns an error if the position is not already held in the cache, or if updating the
3607    /// position in the database fails.
3608    pub fn update_position_from_fill(
3609        &mut self,
3610        position_id: PositionId,
3611        fill: &OrderFilled,
3612    ) -> anyhow::Result<Position> {
3613        let Some(position_cell) = self.positions.get(&position_id).cloned() else {
3614            anyhow::bail!("Cannot update position {position_id}: not found in cache");
3615        };
3616
3617        let position = {
3618            let mut position = position_cell.borrow_mut();
3619            position.apply(fill);
3620            position.clone_without_events()
3621        };
3622
3623        self.refresh_position_indexes(&position);
3624
3625        if let Some(database) = &mut self.database {
3626            database.update_position(&position_cell.borrow())?;
3627        }
3628
3629        Ok(position)
3630    }
3631
3632    /// Updates a cached position by applying an adjustment in place.
3633    ///
3634    /// The canonical cached position retains its complete history. As with
3635    /// [`Self::update_position_from_fill`], a failed database update leaves the adjustment
3636    /// applied; callers that need to roll back use [`Self::revert_position_adjustment`].
3637    ///
3638    /// # Errors
3639    ///
3640    /// Returns an error if the position is not already held in the cache, or if updating the
3641    /// position in the database fails.
3642    pub fn update_position_from_adjustment(
3643        &mut self,
3644        position_id: PositionId,
3645        adjustment: PositionAdjusted,
3646    ) -> anyhow::Result<()> {
3647        let Some(position_cell) = self.positions.get(&position_id).cloned() else {
3648            anyhow::bail!("Cannot update position {position_id}: not found in cache");
3649        };
3650
3651        position_cell.borrow_mut().apply_adjustment(adjustment);
3652        self.refresh_position_indexes(&position_cell.borrow());
3653
3654        if let Some(database) = &mut self.database {
3655            database.update_position(&position_cell.borrow())?;
3656        }
3657
3658        Ok(())
3659    }
3660
3661    /// Reverts the last adjustment applied to a cached position in place.
3662    ///
3663    /// Restores the position state from `prior`, the snapshot taken before the adjustment was
3664    /// applied, then refreshes the position indexes and persists the restored position.
3665    ///
3666    /// # Errors
3667    ///
3668    /// Returns an error if the position is not held in the cache, if `prior` cannot revert its
3669    /// last adjustment (see [`Position::revert_last_adjustment`]), or if updating the position in
3670    /// the database fails.
3671    pub fn revert_position_adjustment(&mut self, prior: &Position) -> anyhow::Result<()> {
3672        let Some(position_cell) = self.positions.get(&prior.id).cloned() else {
3673            anyhow::bail!("Cannot revert position {}: not found in cache", prior.id);
3674        };
3675
3676        position_cell.borrow_mut().revert_last_adjustment(prior)?;
3677        self.refresh_position_indexes(&position_cell.borrow());
3678
3679        if let Some(database) = &mut self.database {
3680            database.update_position(&position_cell.borrow())?;
3681        }
3682
3683        Ok(())
3684    }
3685
3686    /// Updates a cached position by applying an authoritative instrument close in place.
3687    ///
3688    /// Returns a transient copy of the settled state without stored history. The canonical cached
3689    /// position retains its complete history, including the close. A failed database update is
3690    /// logged rather than returned, because the cached position has already settled and callers
3691    /// still need its settled state.
3692    ///
3693    /// # Errors
3694    ///
3695    /// Returns an error if the position is not already held in the cache or the close cannot be
3696    /// applied. An error leaves the position unchanged.
3697    pub fn update_position_from_instrument_close(
3698        &mut self,
3699        position_id: PositionId,
3700        close: InstrumentClose,
3701    ) -> anyhow::Result<Position> {
3702        let Some(position_cell) = self.positions.get(&position_id).cloned() else {
3703            anyhow::bail!("Cannot update position {position_id}: not found in cache");
3704        };
3705
3706        let position = {
3707            let mut position = position_cell.borrow_mut();
3708            position.apply_instrument_close(close)?;
3709            position.clone_without_events()
3710        };
3711
3712        self.refresh_position_indexes(&position);
3713
3714        if let Some(database) = &mut self.database
3715            && let Err(e) = database.update_position(&position_cell.borrow())
3716        {
3717            log::error!("Failed to persist settled position {position_id}: {e}");
3718        }
3719
3720        Ok(position)
3721    }
3722
3723    fn refresh_position_indexes(&mut self, position: &Position) {
3724        if position.is_open() {
3725            self.index.positions_open.insert(position.id);
3726            self.index.positions_closed.remove(&position.id);
3727        } else {
3728            self.index.positions_closed.insert(position.id);
3729            self.index.positions_open.remove(&position.id);
3730        }
3731    }
3732
3733    /// Gets the OMS type for the `position_id`.
3734    #[must_use]
3735    pub fn oms_type(&self, position_id: &PositionId) -> Option<OmsType> {
3736        self.index.position_oms.get(position_id).copied()
3737    }
3738
3739    /// Snapshots the `order` state in the database.
3740    ///
3741    /// # Errors
3742    ///
3743    /// Returns an error if snapshotting the order state fails.
3744    pub fn snapshot_order_state(&self, order: &OrderAny) -> anyhow::Result<()> {
3745        let Some(database) = &self.database else {
3746            log::warn!(
3747                "Cannot snapshot order state for {} (no database configured)",
3748                order.client_order_id()
3749            );
3750            return Ok(());
3751        };
3752
3753        database.snapshot_order_state(order)
3754    }
3755
3756    // -- IDENTIFIER QUERIES ----------------------------------------------------------------------
3757
3758    // Collects references to the index sets that constrain an order query.
3759    //
3760    // Returns:
3761    // - `FilterSources::Unfiltered` when no filter is provided (the caller should iterate
3762    //   the full bucket).
3763    // - `FilterSources::Empty` when a filter is provided but the index has no entry for it
3764    //   (the resolved set is unconditionally empty, no further work needed).
3765    // - `FilterSources::Sets` with borrowed references to each filter source set.
3766    fn collect_order_filter_sources<'a>(
3767        &'a self,
3768        venue: Option<&Venue>,
3769        instrument_id: Option<&InstrumentId>,
3770        strategy_id: Option<&StrategyId>,
3771        account_id: Option<&AccountId>,
3772    ) -> FilterSources<'a, ClientOrderId> {
3773        let mut sources: Vec<&AHashSet<ClientOrderId>> = Vec::with_capacity(4);
3774
3775        if let Some(venue) = venue {
3776            match self.index.venue_orders.get(venue) {
3777                Some(set) => sources.push(set),
3778                None => return FilterSources::Empty,
3779            }
3780        }
3781
3782        if let Some(instrument_id) = instrument_id {
3783            match self.index.instrument_orders.get(instrument_id) {
3784                Some(set) => sources.push(set),
3785                None => return FilterSources::Empty,
3786            }
3787        }
3788
3789        if let Some(strategy_id) = strategy_id {
3790            match self.index.strategy_orders.get(strategy_id) {
3791                Some(set) => sources.push(set),
3792                None => return FilterSources::Empty,
3793            }
3794        }
3795
3796        if let Some(account_id) = account_id {
3797            match self.index.account_orders.get(account_id) {
3798                Some(set) => sources.push(set),
3799                None => return FilterSources::Empty,
3800            }
3801        }
3802
3803        if sources.is_empty() {
3804            FilterSources::Unfiltered
3805        } else {
3806            FilterSources::Sets(sources)
3807        }
3808    }
3809
3810    fn collect_position_filter_sources<'a>(
3811        &'a self,
3812        venue: Option<&Venue>,
3813        instrument_id: Option<&InstrumentId>,
3814        strategy_id: Option<&StrategyId>,
3815        account_id: Option<&AccountId>,
3816    ) -> FilterSources<'a, PositionId> {
3817        let mut sources: Vec<&AHashSet<PositionId>> = Vec::with_capacity(4);
3818
3819        if let Some(venue) = venue {
3820            match self.index.venue_positions.get(venue) {
3821                Some(set) => sources.push(set),
3822                None => return FilterSources::Empty,
3823            }
3824        }
3825
3826        if let Some(instrument_id) = instrument_id {
3827            match self.index.instrument_positions.get(instrument_id) {
3828                Some(set) => sources.push(set),
3829                None => return FilterSources::Empty,
3830            }
3831        }
3832
3833        if let Some(strategy_id) = strategy_id {
3834            match self.index.strategy_positions.get(strategy_id) {
3835                Some(set) => sources.push(set),
3836                None => return FilterSources::Empty,
3837            }
3838        }
3839
3840        if let Some(account_id) = account_id {
3841            match self.index.account_positions.get(account_id) {
3842                Some(set) => sources.push(set),
3843                None => return FilterSources::Empty,
3844            }
3845        }
3846
3847        if sources.is_empty() {
3848            FilterSources::Unfiltered
3849        } else {
3850            FilterSources::Sets(sources)
3851        }
3852    }
3853
3854    // Materializes the `ClientOrderId`s in `bucket` matching the optional filter parameters.
3855    //
3856    // Folds the bucket into the filter sources and runs a single size-ordered intersection,
3857    // avoiding the legacy two-step build-filter-set + bucket-intersection that allocated and
3858    // rehashed twice.
3859    fn query_orders_in_bucket(
3860        &self,
3861        bucket: &AHashSet<ClientOrderId>,
3862        venue: Option<&Venue>,
3863        instrument_id: Option<&InstrumentId>,
3864        strategy_id: Option<&StrategyId>,
3865        account_id: Option<&AccountId>,
3866    ) -> AHashSet<ClientOrderId> {
3867        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3868            FilterSources::Empty => AHashSet::new(),
3869            FilterSources::Unfiltered => bucket.clone(),
3870            FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
3871        }
3872    }
3873
3874    fn query_positions_in_bucket(
3875        &self,
3876        bucket: &AHashSet<PositionId>,
3877        venue: Option<&Venue>,
3878        instrument_id: Option<&InstrumentId>,
3879        strategy_id: Option<&StrategyId>,
3880        account_id: Option<&AccountId>,
3881    ) -> AHashSet<PositionId> {
3882        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3883            FilterSources::Empty => AHashSet::new(),
3884            FilterSources::Unfiltered => bucket.clone(),
3885            FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
3886        }
3887    }
3888
3889    // Returns a borrowed or owned view of the orders in `bucket` matching the optional filter
3890    // parameters. Avoids cloning the bucket when no filter narrows it.
3891    fn view_orders_in_bucket<'a>(
3892        &'a self,
3893        bucket: &'a AHashSet<ClientOrderId>,
3894        venue: Option<&Venue>,
3895        instrument_id: Option<&InstrumentId>,
3896        strategy_id: Option<&StrategyId>,
3897        account_id: Option<&AccountId>,
3898    ) -> Cow<'a, AHashSet<ClientOrderId>> {
3899        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3900            FilterSources::Empty => Cow::Owned(AHashSet::new()),
3901            FilterSources::Unfiltered => Cow::Borrowed(bucket),
3902            FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
3903        }
3904    }
3905
3906    fn view_positions_in_bucket<'a>(
3907        &'a self,
3908        bucket: &'a AHashSet<PositionId>,
3909        venue: Option<&Venue>,
3910        instrument_id: Option<&InstrumentId>,
3911        strategy_id: Option<&StrategyId>,
3912        account_id: Option<&AccountId>,
3913    ) -> Cow<'a, AHashSet<PositionId>> {
3914        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3915            FilterSources::Empty => Cow::Owned(AHashSet::new()),
3916            FilterSources::Unfiltered => Cow::Borrowed(bucket),
3917            FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
3918        }
3919    }
3920
3921    // Returns a lazy iterator yielding the [`ClientOrderId`]s in `bucket` matching the optional
3922    // filter parameters. Avoids any [`Vec`] or [`AHashSet`] materialization in the result path,
3923    // and (for multi-filter calls) drives intersection from the smallest source while looking
3924    // up membership in the rest.
3925    fn iter_orders_in_bucket<'a>(
3926        &'a self,
3927        bucket: &'a AHashSet<ClientOrderId>,
3928        venue: Option<&Venue>,
3929        instrument_id: Option<&InstrumentId>,
3930        strategy_id: Option<&StrategyId>,
3931        account_id: Option<&AccountId>,
3932    ) -> Box<dyn Iterator<Item = ClientOrderId> + 'a> {
3933        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3934            FilterSources::Empty => Box::new(std::iter::empty()),
3935            FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
3936            FilterSources::Sets(mut sources) => {
3937                sources.push(bucket);
3938                sources.sort_unstable_by_key(|s| s.len());
3939                let driver = sources[0];
3940                let rest: Vec<&'a AHashSet<ClientOrderId>> = sources[1..].to_vec();
3941                Box::new(
3942                    driver
3943                        .iter()
3944                        .copied()
3945                        .filter(move |id| rest.iter().all(|s| s.contains(id))),
3946                )
3947            }
3948        }
3949    }
3950
3951    fn iter_positions_in_bucket<'a>(
3952        &'a self,
3953        bucket: &'a AHashSet<PositionId>,
3954        venue: Option<&Venue>,
3955        instrument_id: Option<&InstrumentId>,
3956        strategy_id: Option<&StrategyId>,
3957        account_id: Option<&AccountId>,
3958    ) -> Box<dyn Iterator<Item = PositionId> + 'a> {
3959        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3960            FilterSources::Empty => Box::new(std::iter::empty()),
3961            FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
3962            FilterSources::Sets(mut sources) => {
3963                sources.push(bucket);
3964                sources.sort_unstable_by_key(|s| s.len());
3965                let driver = sources[0];
3966                let rest: Vec<&'a AHashSet<PositionId>> = sources[1..].to_vec();
3967                Box::new(
3968                    driver
3969                        .iter()
3970                        .copied()
3971                        .filter(move |id| rest.iter().all(|s| s.contains(id))),
3972                )
3973            }
3974        }
3975    }
3976
3977    // Counts orders in `bucket` matching the optional filter parameters.
3978    //
3979    // Drives intersection from the smallest filter source (or the bucket itself when no filter
3980    // is provided) and short-circuits by counting rather than collecting. With a side filter,
3981    // each candidate order is borrowed via its cell only long enough to inspect the side.
3982    fn count_orders_in_bucket(
3983        &self,
3984        bucket: &AHashSet<ClientOrderId>,
3985        venue: Option<&Venue>,
3986        instrument_id: Option<&InstrumentId>,
3987        strategy_id: Option<&StrategyId>,
3988        account_id: Option<&AccountId>,
3989        side: Option<OrderSide>,
3990    ) -> usize {
3991        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3992            FilterSources::Empty => 0,
3993            FilterSources::Unfiltered => side.map_or_else(
3994                || bucket.len(),
3995                |side| {
3996                    bucket
3997                        .iter()
3998                        .filter(|id| self.order_side_matches(id, side))
3999                        .count()
4000                },
4001            ),
4002            FilterSources::Sets(mut sources) => {
4003                sources.push(bucket);
4004                sources.sort_unstable_by_key(|s| s.len());
4005                let driver = sources[0];
4006                let rest = &sources[1..];
4007
4008                driver
4009                    .iter()
4010                    .filter(|id| rest.iter().all(|s| s.contains(id)))
4011                    .filter(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
4012                    .count()
4013            }
4014        }
4015    }
4016
4017    fn count_positions_in_bucket(
4018        &self,
4019        bucket: &AHashSet<PositionId>,
4020        venue: Option<&Venue>,
4021        instrument_id: Option<&InstrumentId>,
4022        strategy_id: Option<&StrategyId>,
4023        account_id: Option<&AccountId>,
4024        side: Option<PositionSide>,
4025    ) -> usize {
4026        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
4027            FilterSources::Empty => 0,
4028            FilterSources::Unfiltered => side.map_or_else(
4029                || bucket.len(),
4030                |side| {
4031                    bucket
4032                        .iter()
4033                        .filter(|id| self.position_side_matches(id, side))
4034                        .count()
4035                },
4036            ),
4037            FilterSources::Sets(mut sources) => {
4038                sources.push(bucket);
4039                sources.sort_unstable_by_key(|s| s.len());
4040                let driver = sources[0];
4041                let rest = &sources[1..];
4042
4043                driver
4044                    .iter()
4045                    .filter(|id| rest.iter().all(|s| s.contains(id)))
4046                    .filter(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
4047                    .count()
4048            }
4049        }
4050    }
4051
4052    // Returns whether any order in `bucket` matches the optional filter parameters.
4053    //
4054    // Mirrors `count_orders_in_bucket` but short-circuits on the first match. Useful for
4055    // `is_empty`-style gating in hot paths where the caller only needs to know whether at
4056    // least one matching order exists.
4057    fn any_orders_in_bucket(
4058        &self,
4059        bucket: &AHashSet<ClientOrderId>,
4060        venue: Option<&Venue>,
4061        instrument_id: Option<&InstrumentId>,
4062        strategy_id: Option<&StrategyId>,
4063        account_id: Option<&AccountId>,
4064        side: Option<OrderSide>,
4065    ) -> bool {
4066        match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
4067            FilterSources::Empty => false,
4068            FilterSources::Unfiltered => side.map_or_else(
4069                || !bucket.is_empty(),
4070                |side| bucket.iter().any(|id| self.order_side_matches(id, side)),
4071            ),
4072            FilterSources::Sets(mut sources) => {
4073                sources.push(bucket);
4074                sources.sort_unstable_by_key(|s| s.len());
4075                let driver = sources[0];
4076                let rest = &sources[1..];
4077
4078                driver
4079                    .iter()
4080                    .filter(|id| rest.iter().all(|s| s.contains(id)))
4081                    .any(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
4082            }
4083        }
4084    }
4085
4086    fn any_positions_in_bucket(
4087        &self,
4088        bucket: &AHashSet<PositionId>,
4089        venue: Option<&Venue>,
4090        instrument_id: Option<&InstrumentId>,
4091        strategy_id: Option<&StrategyId>,
4092        account_id: Option<&AccountId>,
4093        side: Option<PositionSide>,
4094    ) -> bool {
4095        match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
4096            FilterSources::Empty => false,
4097            FilterSources::Unfiltered => side.map_or_else(
4098                || !bucket.is_empty(),
4099                |side| bucket.iter().any(|id| self.position_side_matches(id, side)),
4100            ),
4101            FilterSources::Sets(mut sources) => {
4102                sources.push(bucket);
4103                sources.sort_unstable_by_key(|s| s.len());
4104                let driver = sources[0];
4105                let rest = &sources[1..];
4106
4107                driver
4108                    .iter()
4109                    .filter(|id| rest.iter().all(|s| s.contains(id)))
4110                    .any(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
4111            }
4112        }
4113    }
4114
4115    fn order_side_matches(&self, client_order_id: &ClientOrderId, side: OrderSide) -> bool {
4116        self.orders
4117            .get(client_order_id)
4118            .is_some_and(|cell| cell.borrow().order_side() == side)
4119    }
4120
4121    fn position_side_matches(&self, position_id: &PositionId, side: PositionSide) -> bool {
4122        self.positions
4123            .get(position_id)
4124            .is_some_and(|cell| cell.borrow().side == side)
4125    }
4126
4127    /// Retrieves orders corresponding to the `client_order_ids`, optionally filtering by `side`.
4128    ///
4129    /// # Panics
4130    ///
4131    /// Panics if any `client_order_id` in the input is not found in the cache.
4132    fn get_orders_for_ids(
4133        &self,
4134        client_order_ids: impl IntoIterator<Item = ClientOrderId>,
4135        side: Option<OrderSide>,
4136    ) -> Vec<OrderRef<'_>> {
4137        const UNCACHED_SORT_MAX_LEN: usize = 32;
4138
4139        let mut orders = Vec::new();
4140
4141        for client_order_id in client_order_ids {
4142            let order_cell = self
4143                .orders
4144                .get(&client_order_id)
4145                .unwrap_or_else(|| panic!("Order {client_order_id} not found"));
4146            let order = OrderRef::new(order_cell.borrow());
4147
4148            if side.is_none_or(|side| side == order.order_side()) {
4149                orders.push(order);
4150            }
4151        }
4152
4153        // Sort so callers receive a deterministic Vec across runs; the
4154        // underlying ID sources are AHash-backed.
4155        let key = |order: &OrderRef<'_>| order.client_order_id();
4156
4157        if orders.len() <= UNCACHED_SORT_MAX_LEN {
4158            orders.sort_by_key(key);
4159        } else {
4160            orders.sort_by_cached_key(key);
4161        }
4162
4163        orders
4164    }
4165
4166    /// Retrieves positions corresponding to the `position_ids`, optionally filtering by `side`.
4167    ///
4168    /// Each [`PositionRef`] in the returned vector borrows its underlying cell; mutating any of
4169    /// those positions while the vector is alive will panic at runtime. Drop the vector before
4170    /// issuing writes.
4171    ///
4172    /// # Panics
4173    ///
4174    /// Panics if any `position_id` in the set is not found in the cache.
4175    fn get_positions_for_ids(
4176        &self,
4177        position_ids: &AHashSet<PositionId>,
4178        side: Option<PositionSide>,
4179    ) -> Vec<PositionRef<'_>> {
4180        let mut positions = Vec::new();
4181
4182        for position_id in position_ids {
4183            let position_cell = self
4184                .positions
4185                .get(position_id)
4186                .unwrap_or_else(|| panic!("Position {position_id} not found"));
4187            let position = PositionRef::new(position_cell.borrow());
4188
4189            if side.is_none_or(|side| side == position.side) {
4190                positions.push(position);
4191            }
4192        }
4193
4194        // Sort so callers receive a deterministic Vec across runs; the
4195        // underlying position_ids set is AHash-backed.
4196        positions.sort_by_key(|p| p.id);
4197        positions
4198    }
4199
4200    /// Returns the `ClientOrderId`s of all orders.
4201    #[must_use]
4202    pub fn client_order_ids(
4203        &self,
4204        venue: Option<&Venue>,
4205        instrument_id: Option<&InstrumentId>,
4206        strategy_id: Option<&StrategyId>,
4207        account_id: Option<&AccountId>,
4208    ) -> AHashSet<ClientOrderId> {
4209        self.query_orders_in_bucket(
4210            &self.index.orders,
4211            venue,
4212            instrument_id,
4213            strategy_id,
4214            account_id,
4215        )
4216    }
4217
4218    /// Returns the `ClientOrderId`s of all open orders.
4219    #[must_use]
4220    pub fn client_order_ids_open(
4221        &self,
4222        venue: Option<&Venue>,
4223        instrument_id: Option<&InstrumentId>,
4224        strategy_id: Option<&StrategyId>,
4225        account_id: Option<&AccountId>,
4226    ) -> AHashSet<ClientOrderId> {
4227        self.query_orders_in_bucket(
4228            &self.index.orders_open,
4229            venue,
4230            instrument_id,
4231            strategy_id,
4232            account_id,
4233        )
4234    }
4235
4236    /// Returns the `ClientOrderId`s of all closed orders.
4237    #[must_use]
4238    pub fn client_order_ids_closed(
4239        &self,
4240        venue: Option<&Venue>,
4241        instrument_id: Option<&InstrumentId>,
4242        strategy_id: Option<&StrategyId>,
4243        account_id: Option<&AccountId>,
4244    ) -> AHashSet<ClientOrderId> {
4245        self.query_orders_in_bucket(
4246            &self.index.orders_closed,
4247            venue,
4248            instrument_id,
4249            strategy_id,
4250            account_id,
4251        )
4252    }
4253
4254    /// Returns the `ClientOrderId`s of all locally active orders.
4255    ///
4256    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state
4257    /// (a superset of emulated orders).
4258    #[must_use]
4259    pub fn client_order_ids_active_local(
4260        &self,
4261        venue: Option<&Venue>,
4262        instrument_id: Option<&InstrumentId>,
4263        strategy_id: Option<&StrategyId>,
4264        account_id: Option<&AccountId>,
4265    ) -> AHashSet<ClientOrderId> {
4266        self.query_orders_in_bucket(
4267            &self.index.orders_active_local,
4268            venue,
4269            instrument_id,
4270            strategy_id,
4271            account_id,
4272        )
4273    }
4274
4275    /// Returns the `ClientOrderId`s of all emulated orders.
4276    #[must_use]
4277    pub fn client_order_ids_emulated(
4278        &self,
4279        venue: Option<&Venue>,
4280        instrument_id: Option<&InstrumentId>,
4281        strategy_id: Option<&StrategyId>,
4282        account_id: Option<&AccountId>,
4283    ) -> AHashSet<ClientOrderId> {
4284        self.query_orders_in_bucket(
4285            &self.index.orders_emulated,
4286            venue,
4287            instrument_id,
4288            strategy_id,
4289            account_id,
4290        )
4291    }
4292
4293    /// Returns the `ClientOrderId`s of all in-flight orders.
4294    #[must_use]
4295    pub fn client_order_ids_inflight(
4296        &self,
4297        venue: Option<&Venue>,
4298        instrument_id: Option<&InstrumentId>,
4299        strategy_id: Option<&StrategyId>,
4300        account_id: Option<&AccountId>,
4301    ) -> AHashSet<ClientOrderId> {
4302        self.query_orders_in_bucket(
4303            &self.index.orders_inflight,
4304            venue,
4305            instrument_id,
4306            strategy_id,
4307            account_id,
4308        )
4309    }
4310
4311    /// Returns `PositionId`s of all positions.
4312    #[must_use]
4313    pub fn position_ids(
4314        &self,
4315        venue: Option<&Venue>,
4316        instrument_id: Option<&InstrumentId>,
4317        strategy_id: Option<&StrategyId>,
4318        account_id: Option<&AccountId>,
4319    ) -> AHashSet<PositionId> {
4320        self.query_positions_in_bucket(
4321            &self.index.positions,
4322            venue,
4323            instrument_id,
4324            strategy_id,
4325            account_id,
4326        )
4327    }
4328
4329    /// Returns the `PositionId`s of all open positions.
4330    #[must_use]
4331    pub fn position_open_ids(
4332        &self,
4333        venue: Option<&Venue>,
4334        instrument_id: Option<&InstrumentId>,
4335        strategy_id: Option<&StrategyId>,
4336        account_id: Option<&AccountId>,
4337    ) -> AHashSet<PositionId> {
4338        self.query_positions_in_bucket(
4339            &self.index.positions_open,
4340            venue,
4341            instrument_id,
4342            strategy_id,
4343            account_id,
4344        )
4345    }
4346
4347    /// Returns the `PositionId`s of all closed positions.
4348    #[must_use]
4349    pub fn position_closed_ids(
4350        &self,
4351        venue: Option<&Venue>,
4352        instrument_id: Option<&InstrumentId>,
4353        strategy_id: Option<&StrategyId>,
4354        account_id: Option<&AccountId>,
4355    ) -> AHashSet<PositionId> {
4356        self.query_positions_in_bucket(
4357            &self.index.positions_closed,
4358            venue,
4359            instrument_id,
4360            strategy_id,
4361            account_id,
4362        )
4363    }
4364
4365    /// Returns a borrowed view over the [`ClientOrderId`]s of all orders matching the optional
4366    /// filter parameters.
4367    ///
4368    /// The returned [`Cow`] borrows the underlying index when no filter is provided and only
4369    /// allocates an owned [`AHashSet`] when an intersection is required. Prefer this over
4370    /// [`Self::client_order_ids`] when the caller only needs to iterate or read membership.
4371    #[must_use]
4372    pub fn client_order_ids_view(
4373        &self,
4374        venue: Option<&Venue>,
4375        instrument_id: Option<&InstrumentId>,
4376        strategy_id: Option<&StrategyId>,
4377        account_id: Option<&AccountId>,
4378    ) -> Cow<'_, AHashSet<ClientOrderId>> {
4379        self.view_orders_in_bucket(
4380            &self.index.orders,
4381            venue,
4382            instrument_id,
4383            strategy_id,
4384            account_id,
4385        )
4386    }
4387
4388    /// Returns a borrowed view over the [`ClientOrderId`]s of all open orders.
4389    #[must_use]
4390    pub fn client_order_ids_open_view(
4391        &self,
4392        venue: Option<&Venue>,
4393        instrument_id: Option<&InstrumentId>,
4394        strategy_id: Option<&StrategyId>,
4395        account_id: Option<&AccountId>,
4396    ) -> Cow<'_, AHashSet<ClientOrderId>> {
4397        self.view_orders_in_bucket(
4398            &self.index.orders_open,
4399            venue,
4400            instrument_id,
4401            strategy_id,
4402            account_id,
4403        )
4404    }
4405
4406    /// Returns a borrowed view over the [`ClientOrderId`]s of all closed orders.
4407    #[must_use]
4408    pub fn client_order_ids_closed_view(
4409        &self,
4410        venue: Option<&Venue>,
4411        instrument_id: Option<&InstrumentId>,
4412        strategy_id: Option<&StrategyId>,
4413        account_id: Option<&AccountId>,
4414    ) -> Cow<'_, AHashSet<ClientOrderId>> {
4415        self.view_orders_in_bucket(
4416            &self.index.orders_closed,
4417            venue,
4418            instrument_id,
4419            strategy_id,
4420            account_id,
4421        )
4422    }
4423
4424    /// Returns a borrowed view over the [`ClientOrderId`]s of all locally active orders.
4425    #[must_use]
4426    pub fn client_order_ids_active_local_view(
4427        &self,
4428        venue: Option<&Venue>,
4429        instrument_id: Option<&InstrumentId>,
4430        strategy_id: Option<&StrategyId>,
4431        account_id: Option<&AccountId>,
4432    ) -> Cow<'_, AHashSet<ClientOrderId>> {
4433        self.view_orders_in_bucket(
4434            &self.index.orders_active_local,
4435            venue,
4436            instrument_id,
4437            strategy_id,
4438            account_id,
4439        )
4440    }
4441
4442    /// Returns a borrowed view over the [`ClientOrderId`]s of all emulated orders.
4443    #[must_use]
4444    pub fn client_order_ids_emulated_view(
4445        &self,
4446        venue: Option<&Venue>,
4447        instrument_id: Option<&InstrumentId>,
4448        strategy_id: Option<&StrategyId>,
4449        account_id: Option<&AccountId>,
4450    ) -> Cow<'_, AHashSet<ClientOrderId>> {
4451        self.view_orders_in_bucket(
4452            &self.index.orders_emulated,
4453            venue,
4454            instrument_id,
4455            strategy_id,
4456            account_id,
4457        )
4458    }
4459
4460    /// Returns a borrowed view over the [`ClientOrderId`]s of all in-flight orders.
4461    #[must_use]
4462    pub fn client_order_ids_inflight_view(
4463        &self,
4464        venue: Option<&Venue>,
4465        instrument_id: Option<&InstrumentId>,
4466        strategy_id: Option<&StrategyId>,
4467        account_id: Option<&AccountId>,
4468    ) -> Cow<'_, AHashSet<ClientOrderId>> {
4469        self.view_orders_in_bucket(
4470            &self.index.orders_inflight,
4471            venue,
4472            instrument_id,
4473            strategy_id,
4474            account_id,
4475        )
4476    }
4477
4478    /// Returns a borrowed view over the [`PositionId`]s of all positions.
4479    #[must_use]
4480    pub fn position_ids_view(
4481        &self,
4482        venue: Option<&Venue>,
4483        instrument_id: Option<&InstrumentId>,
4484        strategy_id: Option<&StrategyId>,
4485        account_id: Option<&AccountId>,
4486    ) -> Cow<'_, AHashSet<PositionId>> {
4487        self.view_positions_in_bucket(
4488            &self.index.positions,
4489            venue,
4490            instrument_id,
4491            strategy_id,
4492            account_id,
4493        )
4494    }
4495
4496    /// Returns a borrowed view over the [`PositionId`]s of all open positions.
4497    #[must_use]
4498    pub fn position_open_ids_view(
4499        &self,
4500        venue: Option<&Venue>,
4501        instrument_id: Option<&InstrumentId>,
4502        strategy_id: Option<&StrategyId>,
4503        account_id: Option<&AccountId>,
4504    ) -> Cow<'_, AHashSet<PositionId>> {
4505        self.view_positions_in_bucket(
4506            &self.index.positions_open,
4507            venue,
4508            instrument_id,
4509            strategy_id,
4510            account_id,
4511        )
4512    }
4513
4514    /// Returns a borrowed view over the [`PositionId`]s of all closed positions.
4515    #[must_use]
4516    pub fn position_closed_ids_view(
4517        &self,
4518        venue: Option<&Venue>,
4519        instrument_id: Option<&InstrumentId>,
4520        strategy_id: Option<&StrategyId>,
4521        account_id: Option<&AccountId>,
4522    ) -> Cow<'_, AHashSet<PositionId>> {
4523        self.view_positions_in_bucket(
4524            &self.index.positions_closed,
4525            venue,
4526            instrument_id,
4527            strategy_id,
4528            account_id,
4529        )
4530    }
4531
4532    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all orders matching the optional
4533    /// filter parameters.
4534    ///
4535    /// Avoids the [`AHashSet`] allocation performed by [`Self::client_order_ids`]. Useful when
4536    /// the caller iterates the result once and discards it.
4537    pub fn iter_client_order_ids(
4538        &self,
4539        venue: Option<&Venue>,
4540        instrument_id: Option<&InstrumentId>,
4541        strategy_id: Option<&StrategyId>,
4542        account_id: Option<&AccountId>,
4543    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4544        self.iter_orders_in_bucket(
4545            &self.index.orders,
4546            venue,
4547            instrument_id,
4548            strategy_id,
4549            account_id,
4550        )
4551    }
4552
4553    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all open orders.
4554    pub fn iter_client_order_ids_open(
4555        &self,
4556        venue: Option<&Venue>,
4557        instrument_id: Option<&InstrumentId>,
4558        strategy_id: Option<&StrategyId>,
4559        account_id: Option<&AccountId>,
4560    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4561        self.iter_orders_in_bucket(
4562            &self.index.orders_open,
4563            venue,
4564            instrument_id,
4565            strategy_id,
4566            account_id,
4567        )
4568    }
4569
4570    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all closed orders.
4571    pub fn iter_client_order_ids_closed(
4572        &self,
4573        venue: Option<&Venue>,
4574        instrument_id: Option<&InstrumentId>,
4575        strategy_id: Option<&StrategyId>,
4576        account_id: Option<&AccountId>,
4577    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4578        self.iter_orders_in_bucket(
4579            &self.index.orders_closed,
4580            venue,
4581            instrument_id,
4582            strategy_id,
4583            account_id,
4584        )
4585    }
4586
4587    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all locally active orders.
4588    pub fn iter_client_order_ids_active_local(
4589        &self,
4590        venue: Option<&Venue>,
4591        instrument_id: Option<&InstrumentId>,
4592        strategy_id: Option<&StrategyId>,
4593        account_id: Option<&AccountId>,
4594    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4595        self.iter_orders_in_bucket(
4596            &self.index.orders_active_local,
4597            venue,
4598            instrument_id,
4599            strategy_id,
4600            account_id,
4601        )
4602    }
4603
4604    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all emulated orders.
4605    pub fn iter_client_order_ids_emulated(
4606        &self,
4607        venue: Option<&Venue>,
4608        instrument_id: Option<&InstrumentId>,
4609        strategy_id: Option<&StrategyId>,
4610        account_id: Option<&AccountId>,
4611    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4612        self.iter_orders_in_bucket(
4613            &self.index.orders_emulated,
4614            venue,
4615            instrument_id,
4616            strategy_id,
4617            account_id,
4618        )
4619    }
4620
4621    /// Returns a lazy iterator yielding [`ClientOrderId`]s of all in-flight orders.
4622    pub fn iter_client_order_ids_inflight(
4623        &self,
4624        venue: Option<&Venue>,
4625        instrument_id: Option<&InstrumentId>,
4626        strategy_id: Option<&StrategyId>,
4627        account_id: Option<&AccountId>,
4628    ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4629        self.iter_orders_in_bucket(
4630            &self.index.orders_inflight,
4631            venue,
4632            instrument_id,
4633            strategy_id,
4634            account_id,
4635        )
4636    }
4637
4638    /// Returns a lazy iterator yielding [`PositionId`]s of all positions matching the filters.
4639    pub fn iter_position_ids(
4640        &self,
4641        venue: Option<&Venue>,
4642        instrument_id: Option<&InstrumentId>,
4643        strategy_id: Option<&StrategyId>,
4644        account_id: Option<&AccountId>,
4645    ) -> Box<dyn Iterator<Item = PositionId> + '_> {
4646        self.iter_positions_in_bucket(
4647            &self.index.positions,
4648            venue,
4649            instrument_id,
4650            strategy_id,
4651            account_id,
4652        )
4653    }
4654
4655    /// Returns a lazy iterator yielding [`PositionId`]s of all open positions.
4656    pub fn iter_position_open_ids(
4657        &self,
4658        venue: Option<&Venue>,
4659        instrument_id: Option<&InstrumentId>,
4660        strategy_id: Option<&StrategyId>,
4661        account_id: Option<&AccountId>,
4662    ) -> Box<dyn Iterator<Item = PositionId> + '_> {
4663        self.iter_positions_in_bucket(
4664            &self.index.positions_open,
4665            venue,
4666            instrument_id,
4667            strategy_id,
4668            account_id,
4669        )
4670    }
4671
4672    /// Returns a lazy iterator yielding [`PositionId`]s of all closed positions.
4673    pub fn iter_position_closed_ids(
4674        &self,
4675        venue: Option<&Venue>,
4676        instrument_id: Option<&InstrumentId>,
4677        strategy_id: Option<&StrategyId>,
4678        account_id: Option<&AccountId>,
4679    ) -> Box<dyn Iterator<Item = PositionId> + '_> {
4680        self.iter_positions_in_bucket(
4681            &self.index.positions_closed,
4682            venue,
4683            instrument_id,
4684            strategy_id,
4685            account_id,
4686        )
4687    }
4688
4689    /// Returns the `StrategyId`s of all strategies.
4690    #[must_use]
4691    pub fn strategy_ids(&self) -> AHashSet<StrategyId> {
4692        self.index.strategies.clone()
4693    }
4694
4695    /// Returns the `ExecAlgorithmId`s of all execution algorithms.
4696    #[must_use]
4697    pub fn exec_algorithm_ids(&self) -> AHashSet<ExecAlgorithmId> {
4698        self.index.exec_algorithms.clone()
4699    }
4700
4701    // -- ORDER QUERIES ---------------------------------------------------------------------------
4702
4703    /// Gets a borrow of the order with the `client_order_id` (if found).
4704    ///
4705    /// The returned [`OrderRef`] is tied to the cache borrow's scope and panics at runtime if
4706    /// held across a mutation of the same order. Drop the borrow before dispatching events; if
4707    /// post-event state is required, perform a fresh lookup. Use [`Self::order_owned`] when an
4708    /// owned snapshot is needed for a boundary handover.
4709    #[must_use]
4710    pub fn order_ref(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
4711        self.orders
4712            .get(client_order_id)
4713            .map(|order_cell| OrderRef::new(order_cell.borrow()))
4714    }
4715
4716    /// Gets a borrow of the order with the `client_order_id` (if found).
4717    ///
4718    /// Prefer [`Self::order_ref`] in new native code.
4719    #[must_use]
4720    pub fn order(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
4721        self.order_ref(client_order_id)
4722    }
4723
4724    /// Gets a borrow of the order with the `client_order_id`.
4725    ///
4726    /// # Errors
4727    ///
4728    /// Returns an error if the order is not present in the cache.
4729    pub fn try_order_ref(
4730        &self,
4731        client_order_id: &ClientOrderId,
4732    ) -> Result<OrderRef<'_>, OrderLookupError> {
4733        self.orders
4734            .get(client_order_id)
4735            .map(|order_cell| OrderRef::new(order_cell.borrow()))
4736            .ok_or_else(|| OrderLookupError::not_found(*client_order_id))
4737    }
4738
4739    /// Gets a borrow of the order with the `client_order_id`.
4740    ///
4741    /// Prefer [`Self::try_order_ref`] in new native code.
4742    ///
4743    /// # Errors
4744    ///
4745    /// Returns an error if the order is not present in the cache.
4746    pub fn try_order(
4747        &self,
4748        client_order_id: &ClientOrderId,
4749    ) -> Result<OrderRef<'_>, OrderLookupError> {
4750        self.try_order_ref(client_order_id)
4751    }
4752
4753    /// Gets an exclusive write borrow of the order with the `client_order_id` (if found).
4754    ///
4755    /// Requires `&mut Cache` so cache writes are reachable only by privileged crates that hold
4756    /// `Rc<RefCell<Cache>>` directly. Adapter-facing code receives [`CacheView`], which only
4757    /// exposes immutable cache borrows and therefore cannot reach this method.
4758    ///
4759    /// While the returned [`OrderRefMut`] is alive, no other read or write of the same order is
4760    /// permitted. Drop the borrow before dispatching events or taking any other cache borrow that
4761    /// may re-enter the same order.
4762    #[must_use]
4763    pub fn order_mut(&mut self, client_order_id: &ClientOrderId) -> Option<OrderRefMut<'_>> {
4764        self.orders
4765            .get(client_order_id)
4766            .map(|order_cell| OrderRefMut::new(order_cell.borrow_mut()))
4767    }
4768
4769    /// Gets an owned copy of the order with the `client_order_id` (if found).
4770    ///
4771    /// Use when downstream needs an owned [`OrderAny`] that crosses a boundary (for example, an
4772    /// adapter `get_order` API). The copy will not reflect later cache mutations.
4773    #[must_use]
4774    pub fn order_owned(&self, client_order_id: &ClientOrderId) -> Option<OrderAny> {
4775        self.orders
4776            .get(client_order_id)
4777            .map(|order_cell| order_cell.borrow().clone())
4778    }
4779
4780    /// Gets an owned snapshot of the order with the `client_order_id`.
4781    ///
4782    /// # Errors
4783    ///
4784    /// Returns an error if the order is not present in the cache.
4785    pub fn try_order_owned(
4786        &self,
4787        client_order_id: &ClientOrderId,
4788    ) -> Result<OrderAny, OrderLookupError> {
4789        self.try_order_ref(client_order_id)
4790            .map(|order| order.cloned())
4791    }
4792
4793    /// Gets cloned orders for the given `client_order_ids`, logging an error for any missing.
4794    #[must_use]
4795    pub fn orders_for_ids(
4796        &self,
4797        client_order_ids: &[ClientOrderId],
4798        context: &dyn Display,
4799    ) -> Vec<OrderAny> {
4800        let mut orders = Vec::with_capacity(client_order_ids.len());
4801        for id in client_order_ids {
4802            match self.orders.get(id) {
4803                Some(order_cell) => orders.push(order_cell.borrow().clone()),
4804                None => log::error!("Order {id} not found in cache for {context}"),
4805            }
4806        }
4807        orders
4808    }
4809
4810    /// Gets a reference to the client order ID for the `venue_order_id` (if found).
4811    #[must_use]
4812    pub fn client_order_id(&self, venue_order_id: &VenueOrderId) -> Option<&ClientOrderId> {
4813        self.index.venue_order_ids.get(venue_order_id)
4814    }
4815
4816    /// Gets a reference to the venue order ID for the `client_order_id` (if found).
4817    #[must_use]
4818    pub fn venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<&VenueOrderId> {
4819        self.index.client_order_ids.get(client_order_id)
4820    }
4821
4822    /// Gets a reference to the client ID indexed for then `client_order_id` (if found).
4823    #[must_use]
4824    pub fn client_id(&self, client_order_id: &ClientOrderId) -> Option<&ClientId> {
4825        self.index.order_client.get(client_order_id)
4826    }
4827
4828    /// Returns borrows of all orders matching the optional filter parameters.
4829    ///
4830    /// Each [`OrderRef`] in the returned vector borrows its underlying cell; mutating any of
4831    /// those orders while the vector is alive will panic at runtime. Drop the vector
4832    /// before issuing writes.
4833    #[must_use]
4834    pub fn orders_refs(
4835        &self,
4836        venue: Option<&Venue>,
4837        instrument_id: Option<&InstrumentId>,
4838        strategy_id: Option<&StrategyId>,
4839        account_id: Option<&AccountId>,
4840        side: Option<OrderSide>,
4841    ) -> Vec<OrderRef<'_>> {
4842        if venue.is_none()
4843            && instrument_id.is_none()
4844            && strategy_id.is_none()
4845            && account_id.is_none()
4846        {
4847            return self.get_orders_for_ids(self.index.orders.iter().copied(), side);
4848        }
4849
4850        let client_order_ids =
4851            self.iter_client_order_ids(venue, instrument_id, strategy_id, account_id);
4852        self.get_orders_for_ids(client_order_ids, side)
4853    }
4854
4855    /// Returns borrows of all orders matching the optional filter parameters.
4856    ///
4857    /// Prefer [`Self::orders_refs`] in new native code.
4858    #[must_use]
4859    pub fn orders(
4860        &self,
4861        venue: Option<&Venue>,
4862        instrument_id: Option<&InstrumentId>,
4863        strategy_id: Option<&StrategyId>,
4864        account_id: Option<&AccountId>,
4865        side: Option<OrderSide>,
4866    ) -> Vec<OrderRef<'_>> {
4867        self.orders_refs(venue, instrument_id, strategy_id, account_id, side)
4868    }
4869
4870    /// Returns borrows of all open orders matching the optional filter parameters.
4871    #[must_use]
4872    pub fn orders_open_refs(
4873        &self,
4874        venue: Option<&Venue>,
4875        instrument_id: Option<&InstrumentId>,
4876        strategy_id: Option<&StrategyId>,
4877        account_id: Option<&AccountId>,
4878        side: Option<OrderSide>,
4879    ) -> Vec<OrderRef<'_>> {
4880        let client_order_ids =
4881            self.client_order_ids_open(venue, instrument_id, strategy_id, account_id);
4882        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4883    }
4884
4885    /// Returns borrows of all open orders matching the optional filter parameters.
4886    ///
4887    /// Prefer [`Self::orders_open_refs`] in new native code.
4888    #[must_use]
4889    pub fn orders_open(
4890        &self,
4891        venue: Option<&Venue>,
4892        instrument_id: Option<&InstrumentId>,
4893        strategy_id: Option<&StrategyId>,
4894        account_id: Option<&AccountId>,
4895        side: Option<OrderSide>,
4896    ) -> Vec<OrderRef<'_>> {
4897        self.orders_open_refs(venue, instrument_id, strategy_id, account_id, side)
4898    }
4899
4900    /// Returns borrows of all closed orders matching the optional filter parameters.
4901    #[must_use]
4902    pub fn orders_closed_refs(
4903        &self,
4904        venue: Option<&Venue>,
4905        instrument_id: Option<&InstrumentId>,
4906        strategy_id: Option<&StrategyId>,
4907        account_id: Option<&AccountId>,
4908        side: Option<OrderSide>,
4909    ) -> Vec<OrderRef<'_>> {
4910        let client_order_ids =
4911            self.client_order_ids_closed(venue, instrument_id, strategy_id, account_id);
4912        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4913    }
4914
4915    /// Returns borrows of all closed orders matching the optional filter parameters.
4916    ///
4917    /// Prefer [`Self::orders_closed_refs`] in new native code.
4918    #[must_use]
4919    pub fn orders_closed(
4920        &self,
4921        venue: Option<&Venue>,
4922        instrument_id: Option<&InstrumentId>,
4923        strategy_id: Option<&StrategyId>,
4924        account_id: Option<&AccountId>,
4925        side: Option<OrderSide>,
4926    ) -> Vec<OrderRef<'_>> {
4927        self.orders_closed_refs(venue, instrument_id, strategy_id, account_id, side)
4928    }
4929
4930    /// Returns borrows of all locally active orders matching the optional filter parameters.
4931    ///
4932    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state
4933    /// (a superset of emulated orders).
4934    #[must_use]
4935    pub fn orders_active_local_refs(
4936        &self,
4937        venue: Option<&Venue>,
4938        instrument_id: Option<&InstrumentId>,
4939        strategy_id: Option<&StrategyId>,
4940        account_id: Option<&AccountId>,
4941        side: Option<OrderSide>,
4942    ) -> Vec<OrderRef<'_>> {
4943        let client_order_ids =
4944            self.client_order_ids_active_local(venue, instrument_id, strategy_id, account_id);
4945        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4946    }
4947
4948    /// Returns borrows of all locally active orders matching the optional filter parameters.
4949    ///
4950    /// Prefer [`Self::orders_active_local_refs`] in new native code.
4951    #[must_use]
4952    pub fn orders_active_local(
4953        &self,
4954        venue: Option<&Venue>,
4955        instrument_id: Option<&InstrumentId>,
4956        strategy_id: Option<&StrategyId>,
4957        account_id: Option<&AccountId>,
4958        side: Option<OrderSide>,
4959    ) -> Vec<OrderRef<'_>> {
4960        self.orders_active_local_refs(venue, instrument_id, strategy_id, account_id, side)
4961    }
4962
4963    /// Returns borrows of all emulated orders matching the optional filter parameters.
4964    #[must_use]
4965    pub fn orders_emulated_refs(
4966        &self,
4967        venue: Option<&Venue>,
4968        instrument_id: Option<&InstrumentId>,
4969        strategy_id: Option<&StrategyId>,
4970        account_id: Option<&AccountId>,
4971        side: Option<OrderSide>,
4972    ) -> Vec<OrderRef<'_>> {
4973        let client_order_ids =
4974            self.client_order_ids_emulated(venue, instrument_id, strategy_id, account_id);
4975        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4976    }
4977
4978    /// Returns borrows of all emulated orders matching the optional filter parameters.
4979    ///
4980    /// Prefer [`Self::orders_emulated_refs`] in new native code.
4981    #[must_use]
4982    pub fn orders_emulated(
4983        &self,
4984        venue: Option<&Venue>,
4985        instrument_id: Option<&InstrumentId>,
4986        strategy_id: Option<&StrategyId>,
4987        account_id: Option<&AccountId>,
4988        side: Option<OrderSide>,
4989    ) -> Vec<OrderRef<'_>> {
4990        self.orders_emulated_refs(venue, instrument_id, strategy_id, account_id, side)
4991    }
4992
4993    /// Returns borrows of all in-flight orders matching the optional filter parameters.
4994    #[must_use]
4995    pub fn orders_inflight_refs(
4996        &self,
4997        venue: Option<&Venue>,
4998        instrument_id: Option<&InstrumentId>,
4999        strategy_id: Option<&StrategyId>,
5000        account_id: Option<&AccountId>,
5001        side: Option<OrderSide>,
5002    ) -> Vec<OrderRef<'_>> {
5003        let client_order_ids =
5004            self.client_order_ids_inflight(venue, instrument_id, strategy_id, account_id);
5005        self.get_orders_for_ids(client_order_ids.iter().copied(), side)
5006    }
5007
5008    /// Returns borrows of all in-flight orders matching the optional filter parameters.
5009    ///
5010    /// Prefer [`Self::orders_inflight_refs`] in new native code.
5011    #[must_use]
5012    pub fn orders_inflight(
5013        &self,
5014        venue: Option<&Venue>,
5015        instrument_id: Option<&InstrumentId>,
5016        strategy_id: Option<&StrategyId>,
5017        account_id: Option<&AccountId>,
5018        side: Option<OrderSide>,
5019    ) -> Vec<OrderRef<'_>> {
5020        self.orders_inflight_refs(venue, instrument_id, strategy_id, account_id, side)
5021    }
5022
5023    /// Returns borrows of all orders for the `position_id`.
5024    #[must_use]
5025    pub fn orders_for_position(&self, position_id: &PositionId) -> Vec<OrderRef<'_>> {
5026        match self.index.position_orders.get(position_id) {
5027            Some(client_order_ids) => {
5028                self.get_orders_for_ids(client_order_ids.iter().copied(), None)
5029            }
5030            None => Vec::new(),
5031        }
5032    }
5033
5034    /// Returns whether an order with the `client_order_id` exists.
5035    #[must_use]
5036    pub fn order_exists(&self, client_order_id: &ClientOrderId) -> bool {
5037        self.index.orders.contains(client_order_id)
5038    }
5039
5040    /// Returns whether an order with the `client_order_id` is open.
5041    #[must_use]
5042    pub fn is_order_open(&self, client_order_id: &ClientOrderId) -> bool {
5043        self.index.orders_open.contains(client_order_id)
5044    }
5045
5046    /// Returns whether an order with the `client_order_id` is closed.
5047    #[must_use]
5048    pub fn is_order_closed(&self, client_order_id: &ClientOrderId) -> bool {
5049        self.index.orders_closed.contains(client_order_id)
5050    }
5051
5052    /// Returns whether an order with the `client_order_id` is locally active.
5053    ///
5054    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state
5055    /// (a superset of emulated orders).
5056    #[must_use]
5057    pub fn is_order_active_local(&self, client_order_id: &ClientOrderId) -> bool {
5058        self.index.orders_active_local.contains(client_order_id)
5059    }
5060
5061    /// Returns whether an order with the `client_order_id` is emulated.
5062    #[must_use]
5063    pub fn is_order_emulated(&self, client_order_id: &ClientOrderId) -> bool {
5064        self.index.orders_emulated.contains(client_order_id)
5065    }
5066
5067    /// Returns whether an order with the `client_order_id` is in-flight.
5068    #[must_use]
5069    pub fn is_order_inflight(&self, client_order_id: &ClientOrderId) -> bool {
5070        self.index.orders_inflight.contains(client_order_id)
5071    }
5072
5073    /// Returns whether an order with the `client_order_id` is `PENDING_CANCEL` locally.
5074    #[must_use]
5075    pub fn is_order_pending_cancel_local(&self, client_order_id: &ClientOrderId) -> bool {
5076        self.index.orders_pending_cancel.contains(client_order_id)
5077    }
5078
5079    /// Returns the count of all open orders.
5080    #[must_use]
5081    pub fn orders_open_count(
5082        &self,
5083        venue: Option<&Venue>,
5084        instrument_id: Option<&InstrumentId>,
5085        strategy_id: Option<&StrategyId>,
5086        account_id: Option<&AccountId>,
5087        side: Option<OrderSide>,
5088    ) -> usize {
5089        self.count_orders_in_bucket(
5090            &self.index.orders_open,
5091            venue,
5092            instrument_id,
5093            strategy_id,
5094            account_id,
5095            side,
5096        )
5097    }
5098
5099    /// Returns the count of all closed orders.
5100    #[must_use]
5101    pub fn orders_closed_count(
5102        &self,
5103        venue: Option<&Venue>,
5104        instrument_id: Option<&InstrumentId>,
5105        strategy_id: Option<&StrategyId>,
5106        account_id: Option<&AccountId>,
5107        side: Option<OrderSide>,
5108    ) -> usize {
5109        self.count_orders_in_bucket(
5110            &self.index.orders_closed,
5111            venue,
5112            instrument_id,
5113            strategy_id,
5114            account_id,
5115            side,
5116        )
5117    }
5118
5119    /// Returns the count of all locally active orders.
5120    ///
5121    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state
5122    /// (a superset of emulated orders).
5123    #[must_use]
5124    pub fn orders_active_local_count(
5125        &self,
5126        venue: Option<&Venue>,
5127        instrument_id: Option<&InstrumentId>,
5128        strategy_id: Option<&StrategyId>,
5129        account_id: Option<&AccountId>,
5130        side: Option<OrderSide>,
5131    ) -> usize {
5132        self.count_orders_in_bucket(
5133            &self.index.orders_active_local,
5134            venue,
5135            instrument_id,
5136            strategy_id,
5137            account_id,
5138            side,
5139        )
5140    }
5141
5142    /// Returns the count of all emulated orders.
5143    #[must_use]
5144    pub fn orders_emulated_count(
5145        &self,
5146        venue: Option<&Venue>,
5147        instrument_id: Option<&InstrumentId>,
5148        strategy_id: Option<&StrategyId>,
5149        account_id: Option<&AccountId>,
5150        side: Option<OrderSide>,
5151    ) -> usize {
5152        self.count_orders_in_bucket(
5153            &self.index.orders_emulated,
5154            venue,
5155            instrument_id,
5156            strategy_id,
5157            account_id,
5158            side,
5159        )
5160    }
5161
5162    /// Returns the count of all in-flight orders.
5163    #[must_use]
5164    pub fn orders_inflight_count(
5165        &self,
5166        venue: Option<&Venue>,
5167        instrument_id: Option<&InstrumentId>,
5168        strategy_id: Option<&StrategyId>,
5169        account_id: Option<&AccountId>,
5170        side: Option<OrderSide>,
5171    ) -> usize {
5172        self.count_orders_in_bucket(
5173            &self.index.orders_inflight,
5174            venue,
5175            instrument_id,
5176            strategy_id,
5177            account_id,
5178            side,
5179        )
5180    }
5181
5182    /// Returns the count of all orders.
5183    #[must_use]
5184    pub fn orders_total_count(
5185        &self,
5186        venue: Option<&Venue>,
5187        instrument_id: Option<&InstrumentId>,
5188        strategy_id: Option<&StrategyId>,
5189        account_id: Option<&AccountId>,
5190        side: Option<OrderSide>,
5191    ) -> usize {
5192        self.count_orders_in_bucket(
5193            &self.index.orders,
5194            venue,
5195            instrument_id,
5196            strategy_id,
5197            account_id,
5198            side,
5199        )
5200    }
5201
5202    /// Returns whether any open order matches the optional filter parameters.
5203    ///
5204    /// Short-circuits on the first match, avoiding the full intersection walk performed by
5205    /// [`Self::orders_open_count`]. Prefer this over `orders_open_count(...) > 0` when only
5206    /// existence matters.
5207    #[must_use]
5208    pub fn has_orders_open(
5209        &self,
5210        venue: Option<&Venue>,
5211        instrument_id: Option<&InstrumentId>,
5212        strategy_id: Option<&StrategyId>,
5213        account_id: Option<&AccountId>,
5214        side: Option<OrderSide>,
5215    ) -> bool {
5216        self.any_orders_in_bucket(
5217            &self.index.orders_open,
5218            venue,
5219            instrument_id,
5220            strategy_id,
5221            account_id,
5222            side,
5223        )
5224    }
5225
5226    /// Returns whether any closed order matches the optional filter parameters.
5227    #[must_use]
5228    pub fn has_orders_closed(
5229        &self,
5230        venue: Option<&Venue>,
5231        instrument_id: Option<&InstrumentId>,
5232        strategy_id: Option<&StrategyId>,
5233        account_id: Option<&AccountId>,
5234        side: Option<OrderSide>,
5235    ) -> bool {
5236        self.any_orders_in_bucket(
5237            &self.index.orders_closed,
5238            venue,
5239            instrument_id,
5240            strategy_id,
5241            account_id,
5242            side,
5243        )
5244    }
5245
5246    /// Returns whether any locally active order matches the optional filter parameters.
5247    ///
5248    /// Locally active orders are in the `INITIALIZED`, `EMULATED`, or `RELEASED` state.
5249    #[must_use]
5250    pub fn has_orders_active_local(
5251        &self,
5252        venue: Option<&Venue>,
5253        instrument_id: Option<&InstrumentId>,
5254        strategy_id: Option<&StrategyId>,
5255        account_id: Option<&AccountId>,
5256        side: Option<OrderSide>,
5257    ) -> bool {
5258        self.any_orders_in_bucket(
5259            &self.index.orders_active_local,
5260            venue,
5261            instrument_id,
5262            strategy_id,
5263            account_id,
5264            side,
5265        )
5266    }
5267
5268    /// Returns whether any emulated order matches the optional filter parameters.
5269    #[must_use]
5270    pub fn has_orders_emulated(
5271        &self,
5272        venue: Option<&Venue>,
5273        instrument_id: Option<&InstrumentId>,
5274        strategy_id: Option<&StrategyId>,
5275        account_id: Option<&AccountId>,
5276        side: Option<OrderSide>,
5277    ) -> bool {
5278        self.any_orders_in_bucket(
5279            &self.index.orders_emulated,
5280            venue,
5281            instrument_id,
5282            strategy_id,
5283            account_id,
5284            side,
5285        )
5286    }
5287
5288    /// Returns whether any in-flight order matches the optional filter parameters.
5289    #[must_use]
5290    pub fn has_orders_inflight(
5291        &self,
5292        venue: Option<&Venue>,
5293        instrument_id: Option<&InstrumentId>,
5294        strategy_id: Option<&StrategyId>,
5295        account_id: Option<&AccountId>,
5296        side: Option<OrderSide>,
5297    ) -> bool {
5298        self.any_orders_in_bucket(
5299            &self.index.orders_inflight,
5300            venue,
5301            instrument_id,
5302            strategy_id,
5303            account_id,
5304            side,
5305        )
5306    }
5307
5308    /// Returns whether any order (in any state) matches the optional filter parameters.
5309    #[must_use]
5310    pub fn has_orders(
5311        &self,
5312        venue: Option<&Venue>,
5313        instrument_id: Option<&InstrumentId>,
5314        strategy_id: Option<&StrategyId>,
5315        account_id: Option<&AccountId>,
5316        side: Option<OrderSide>,
5317    ) -> bool {
5318        self.any_orders_in_bucket(
5319            &self.index.orders,
5320            venue,
5321            instrument_id,
5322            strategy_id,
5323            account_id,
5324            side,
5325        )
5326    }
5327
5328    /// Returns the order list for the `order_list_id`.
5329    #[must_use]
5330    pub fn order_list(&self, order_list_id: &OrderListId) -> Option<&OrderList> {
5331        self.order_lists.get(order_list_id)
5332    }
5333
5334    /// Returns the order list for the `order_list_id`.
5335    ///
5336    /// # Errors
5337    ///
5338    /// Returns an error if the order list is not present in the cache.
5339    pub fn try_order_list(
5340        &self,
5341        order_list_id: &OrderListId,
5342    ) -> Result<&OrderList, OrderListLookupError> {
5343        self.order_lists
5344            .get(order_list_id)
5345            .ok_or_else(|| OrderListLookupError::not_found(*order_list_id))
5346    }
5347
5348    /// Returns all order lists matching the optional filter parameters.
5349    #[must_use]
5350    pub fn order_lists(
5351        &self,
5352        venue: Option<&Venue>,
5353        instrument_id: Option<&InstrumentId>,
5354        strategy_id: Option<&StrategyId>,
5355        account_id: Option<&AccountId>,
5356    ) -> Vec<&OrderList> {
5357        let mut order_lists = self.order_lists.values().collect::<Vec<&OrderList>>();
5358
5359        if let Some(venue) = venue {
5360            order_lists.retain(|ol| &ol.instrument_id.venue == venue);
5361        }
5362
5363        if let Some(instrument_id) = instrument_id {
5364            order_lists.retain(|ol| &ol.instrument_id == instrument_id);
5365        }
5366
5367        if let Some(strategy_id) = strategy_id {
5368            order_lists.retain(|ol| &ol.strategy_id == strategy_id);
5369        }
5370
5371        if let Some(account_id) = account_id {
5372            order_lists.retain(|ol| {
5373                ol.client_order_ids.iter().any(|client_order_id| {
5374                    self.orders.get(client_order_id).is_some_and(|order_cell| {
5375                        order_cell.borrow().account_id().as_ref() == Some(account_id)
5376                    })
5377                })
5378            });
5379        }
5380
5381        order_lists
5382    }
5383
5384    /// Returns whether an order list with the `order_list_id` exists.
5385    #[must_use]
5386    pub fn order_list_exists(&self, order_list_id: &OrderListId) -> bool {
5387        self.order_lists.contains_key(order_list_id)
5388    }
5389
5390    // -- EXEC ALGORITHM QUERIES ------------------------------------------------------------------
5391
5392    /// Returns references to all orders associated with the `exec_algorithm_id` matching the
5393    /// optional filter parameters.
5394    #[must_use]
5395    pub fn orders_for_exec_algorithm(
5396        &self,
5397        exec_algorithm_id: &ExecAlgorithmId,
5398        venue: Option<&Venue>,
5399        instrument_id: Option<&InstrumentId>,
5400        strategy_id: Option<&StrategyId>,
5401        account_id: Option<&AccountId>,
5402        side: Option<OrderSide>,
5403    ) -> Vec<OrderRef<'_>> {
5404        let Some(exec_algorithm_order_ids) =
5405            self.index.exec_algorithm_orders.get(exec_algorithm_id)
5406        else {
5407            return Vec::new();
5408        };
5409
5410        let filtered = self.query_orders_in_bucket(
5411            exec_algorithm_order_ids,
5412            venue,
5413            instrument_id,
5414            strategy_id,
5415            account_id,
5416        );
5417        self.get_orders_for_ids(filtered.iter().copied(), side)
5418    }
5419
5420    /// Returns references to all orders with the `exec_spawn_id`.
5421    #[must_use]
5422    pub fn orders_for_exec_spawn(&self, exec_spawn_id: &ClientOrderId) -> Vec<OrderRef<'_>> {
5423        match self.index.exec_spawn_orders.get(exec_spawn_id) {
5424            Some(ids) => self.get_orders_for_ids(ids.iter().copied(), None),
5425            None => Vec::new(),
5426        }
5427    }
5428
5429    /// Returns the total order quantity for the `exec_spawn_id`.
5430    #[must_use]
5431    pub fn exec_spawn_total_quantity(
5432        &self,
5433        exec_spawn_id: &ClientOrderId,
5434        active_only: bool,
5435    ) -> Option<Quantity> {
5436        self.exec_spawn_total(exec_spawn_id, active_only, Order::quantity)
5437    }
5438
5439    /// Returns the total filled quantity for all orders with the `exec_spawn_id`.
5440    #[must_use]
5441    pub fn exec_spawn_total_filled_qty(
5442        &self,
5443        exec_spawn_id: &ClientOrderId,
5444        active_only: bool,
5445    ) -> Option<Quantity> {
5446        self.exec_spawn_total(exec_spawn_id, active_only, Order::filled_qty)
5447    }
5448
5449    /// Returns the total leaves quantity for all orders with the `exec_spawn_id`.
5450    #[must_use]
5451    pub fn exec_spawn_total_leaves_qty(
5452        &self,
5453        exec_spawn_id: &ClientOrderId,
5454        active_only: bool,
5455    ) -> Option<Quantity> {
5456        self.exec_spawn_total(exec_spawn_id, active_only, Order::leaves_qty)
5457    }
5458
5459    fn exec_spawn_total(
5460        &self,
5461        exec_spawn_id: &ClientOrderId,
5462        active_only: bool,
5463        quantity: impl Fn(&OrderAny) -> Quantity,
5464    ) -> Option<Quantity> {
5465        self.orders_for_exec_spawn(exec_spawn_id)
5466            .into_iter()
5467            .filter(|order| !active_only || !order.is_closed())
5468            .map(|order| quantity(&order))
5469            .reduce(|total, quantity| total + quantity)
5470    }
5471
5472    // -- POSITION QUERIES ------------------------------------------------------------------------
5473
5474    /// Returns a borrow of the position with the `position_id` (if found).
5475    #[must_use]
5476    pub fn position_ref(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
5477        self.positions
5478            .get(position_id)
5479            .map(|position_cell| PositionRef::new(position_cell.borrow()))
5480    }
5481
5482    /// Returns a borrow of the position with the `position_id` (if found).
5483    ///
5484    /// Prefer [`Self::position_ref`] in new native code.
5485    #[must_use]
5486    pub fn position(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
5487        self.position_ref(position_id)
5488    }
5489
5490    /// Returns a borrow of the position with the `position_id`.
5491    ///
5492    /// # Errors
5493    ///
5494    /// Returns an error if the position is not present in the cache.
5495    pub fn try_position_ref(
5496        &self,
5497        position_id: &PositionId,
5498    ) -> Result<PositionRef<'_>, PositionLookupError> {
5499        self.positions
5500            .get(position_id)
5501            .map(|position_cell| PositionRef::new(position_cell.borrow()))
5502            .ok_or_else(|| PositionLookupError::not_found(*position_id))
5503    }
5504
5505    /// Returns a borrow of the position with the `position_id`.
5506    ///
5507    /// Prefer [`Self::try_position_ref`] in new native code.
5508    ///
5509    /// # Errors
5510    ///
5511    /// Returns an error if the position is not present in the cache.
5512    pub fn try_position(
5513        &self,
5514        position_id: &PositionId,
5515    ) -> Result<PositionRef<'_>, PositionLookupError> {
5516        self.try_position_ref(position_id)
5517    }
5518
5519    /// Gets an exclusive write borrow of the position with the `position_id` (if found).
5520    ///
5521    /// Requires `&mut Cache` so cache writes are reachable only by privileged crates that hold
5522    /// `Rc<RefCell<Cache>>` directly. Adapter-facing code receives [`CacheView`], which only
5523    /// exposes immutable cache borrows and therefore cannot reach this method.
5524    ///
5525    /// While the returned [`PositionRefMut`] is alive, no other read or write of the same position
5526    /// is permitted. Drop the borrow before dispatching events or taking any other cache borrow
5527    /// that may re-enter the same position.
5528    #[must_use]
5529    pub fn position_mut(&mut self, position_id: &PositionId) -> Option<PositionRefMut<'_>> {
5530        self.positions
5531            .get(position_id)
5532            .map(|position_cell| PositionRefMut::new(position_cell.borrow_mut()))
5533    }
5534
5535    /// Gets an owned copy of the position with the `position_id` (if found).
5536    ///
5537    /// Use when downstream needs an owned [`Position`] that crosses a boundary. The copy will not
5538    /// reflect later cache mutations.
5539    #[must_use]
5540    pub fn position_owned(&self, position_id: &PositionId) -> Option<Position> {
5541        self.positions
5542            .get(position_id)
5543            .map(|position_cell| position_cell.borrow().clone())
5544    }
5545
5546    /// Returns a borrow of the position for the `client_order_id` (if found).
5547    #[must_use]
5548    pub fn position_for_order_ref(
5549        &self,
5550        client_order_id: &ClientOrderId,
5551    ) -> Option<PositionRef<'_>> {
5552        self.index
5553            .order_position
5554            .get(client_order_id)
5555            .and_then(|position_id| self.positions.get(position_id))
5556            .map(|position_cell| PositionRef::new(position_cell.borrow()))
5557    }
5558
5559    /// Returns a borrow of the position for the `client_order_id` (if found).
5560    ///
5561    /// Prefer [`Self::position_for_order_ref`] in new native code.
5562    #[must_use]
5563    pub fn position_for_order(&self, client_order_id: &ClientOrderId) -> Option<PositionRef<'_>> {
5564        self.position_for_order_ref(client_order_id)
5565    }
5566
5567    /// Returns a reference to the position ID for the `client_order_id` (if found).
5568    #[must_use]
5569    pub fn position_id(&self, client_order_id: &ClientOrderId) -> Option<&PositionId> {
5570        self.index.order_position.get(client_order_id)
5571    }
5572
5573    /// Returns the proven virtual HEDGING flip IDs, with the current destination second.
5574    ///
5575    /// The route comes from the original fill and its causal split fragments, independently of
5576    /// the order's mutable position ID. Returns `None` when no virtual split is recorded.
5577    ///
5578    /// # Errors
5579    ///
5580    /// Returns an error if split history is incomplete, conflicts, or a correction leaves both
5581    /// sides of a split open.
5582    pub fn hedging_flip_position_ids(
5583        &self,
5584        client_order_id: ClientOrderId,
5585    ) -> anyhow::Result<Option<[PositionId; 2]>> {
5586        let Some(order) = self.order_ref(&client_order_id) else {
5587            return Ok(None);
5588        };
5589
5590        let mut flip = None;
5591        let mut corrected_source = None;
5592
5593        for event in order.events() {
5594            let OrderEventAny::Filled(fill) = event else {
5595                continue;
5596            };
5597
5598            let Some(source_id) = fill.position_id.filter(PositionId::is_virtual) else {
5599                continue;
5600            };
5601
5602            let source = self.position_ref(&source_id);
5603            if source.is_some() && self.oms_type(&source_id) != Some(OmsType::Hedging) {
5604                continue;
5605            }
5606
5607            if source.as_ref().is_some_and(|source| {
5608                source.replay_events.iter().any(|event| {
5609                    matches!(event, PositionReplayEvent::Filled(fragment) if fragment == fill)
5610                })
5611            }) {
5612                continue;
5613            }
5614
5615            let closing = source.as_ref().and_then(|source| {
5616                source
5617                    .fill_fragments(client_order_id, fill.trade_id)
5618                    .into_iter()
5619                    .find(|fragment| {
5620                        fragment.event_id == fill.event_id && fragment.last_qty < fill.last_qty
5621                    })
5622                    .map(|closing| (source, closing))
5623            });
5624
5625            let Some((source, closing)) = closing else {
5626                let has_opening = self
5627                    .positions_refs(
5628                        None,
5629                        Some(&fill.instrument_id),
5630                        Some(&fill.strategy_id),
5631                        Some(&fill.account_id),
5632                        None,
5633                    )
5634                    .iter()
5635                    .any(|position| {
5636                        position.id != source_id
5637                            && position.id.is_virtual()
5638                            && self.oms_type(&position.id) == Some(OmsType::Hedging)
5639                            && position.replay_events.iter().any(|event| {
5640                                matches!(event, PositionReplayEvent::Filled(opening)
5641                                if opening.causation_id == Some(fill.event_id)
5642                                    && opening.client_order_id == client_order_id
5643                                    && opening.trade_id == fill.trade_id)
5644                            })
5645                    });
5646
5647                anyhow::ensure!(
5648                    !has_opening,
5649                    "missing virtual flip source fragment for order {client_order_id}"
5650                );
5651                continue;
5652            };
5653
5654            let (positions, corrected_closing) =
5655                self.hedging_flip_destination(fill, source, closing)?;
5656            anyhow::ensure!(
5657                flip.is_none() || corrected_source == Some(source.id),
5658                "ambiguous virtual flip history for order {client_order_id}"
5659            );
5660            flip = Some(positions);
5661            corrected_source = corrected_closing.then_some(source.id);
5662        }
5663
5664        Ok(flip)
5665    }
5666
5667    fn hedging_flip_destination(
5668        &self,
5669        fill: &OrderFilled,
5670        source: &Position,
5671        closing: &OrderFilled,
5672    ) -> anyhow::Result<([PositionId; 2], bool)> {
5673        let positions = self.positions_refs(
5674            None,
5675            Some(&fill.instrument_id),
5676            Some(&fill.strategy_id),
5677            Some(&fill.account_id),
5678            None,
5679        );
5680
5681        let mut openings = positions.iter().flat_map(|position| {
5682            position
5683                .replay_events
5684                .iter()
5685                .filter_map(move |event| match event {
5686                    PositionReplayEvent::Filled(opening)
5687                        if opening.causation_id == Some(fill.event_id)
5688                            && position.id != source.id =>
5689                    {
5690                        Some((position, opening))
5691                    }
5692                    _ => None,
5693                })
5694        });
5695
5696        let (destination, opening) = openings.next().ok_or_else(|| {
5697            anyhow::anyhow!(
5698                "missing virtual flip fragment for order {}",
5699                fill.client_order_id
5700            )
5701        })?;
5702
5703        anyhow::ensure!(
5704            openings.next().is_none(),
5705            "ambiguous virtual flip fragments for order {}",
5706            fill.client_order_id
5707        );
5708        anyhow::ensure!(
5709            source.instrument_id == fill.instrument_id
5710                && source.account_id == fill.account_id
5711                && source.strategy_id == fill.strategy_id
5712                && destination.id.is_virtual()
5713                && self.oms_type(&destination.id) == Some(OmsType::Hedging),
5714            "conflicting virtual flip positions for order {}",
5715            fill.client_order_id
5716        );
5717        let (expected_closing, expected_opening) =
5718            fill.split_for_position_flip(closing.last_qty, Some(destination.id), opening.event_id)?;
5719        anyhow::ensure!(
5720            *closing == expected_closing && *opening == expected_opening,
5721            "conflicting virtual flip provenance for order {}",
5722            fill.client_order_id
5723        );
5724
5725        let corrected_closing = source.fill_voids.iter().any(|correction| {
5726            correction.event.client_order_id == fill.client_order_id
5727                && correction.event.trade_id == fill.trade_id
5728                && correction.voided_qty.non_zero()
5729        });
5730
5731        if corrected_closing && source.is_open() {
5732            anyhow::ensure!(
5733                !destination.is_open(),
5734                "ambiguous corrected virtual flip for order {}",
5735                fill.client_order_id
5736            );
5737            return Ok(([destination.id, source.id], corrected_closing));
5738        }
5739
5740        Ok(([source.id, destination.id], corrected_closing))
5741    }
5742
5743    fn reindex_hedging_position(&mut self, client_order_id: ClientOrderId) {
5744        match self.hedging_flip_position_ids(client_order_id) {
5745            Ok(Some([_, position_id])) => {
5746                let position = self
5747                    .position_ref(&position_id)
5748                    .expect("proven position is cached");
5749                let venue = position.instrument_id.venue;
5750                let strategy_id = position.strategy_id;
5751                drop(position);
5752
5753                if let Err(e) =
5754                    self.add_position_id(&position_id, &venue, &client_order_id, &strategy_id)
5755                {
5756                    log::error!("Failed to index virtual flip for order {client_order_id}: {e}");
5757                }
5758            }
5759            Ok(None) => {}
5760            Err(e) => {
5761                self.index.order_position.remove(&client_order_id);
5762
5763                log::error!("Cannot index virtual flip for order {client_order_id}: {e}");
5764            }
5765        }
5766    }
5767
5768    /// Returns borrows of all positions matching the optional filter parameters.
5769    ///
5770    /// Each [`PositionRef`] in the returned vector borrows its underlying cell; mutating any of
5771    /// those positions while the vector is alive will panic at runtime. Drop the vector before
5772    /// issuing writes.
5773    #[must_use]
5774    pub fn positions_refs(
5775        &self,
5776        venue: Option<&Venue>,
5777        instrument_id: Option<&InstrumentId>,
5778        strategy_id: Option<&StrategyId>,
5779        account_id: Option<&AccountId>,
5780        side: Option<PositionSide>,
5781    ) -> Vec<PositionRef<'_>> {
5782        let position_ids = self.position_ids(venue, instrument_id, strategy_id, account_id);
5783        self.get_positions_for_ids(&position_ids, side)
5784    }
5785
5786    /// Returns borrows of all positions matching the optional filter parameters.
5787    ///
5788    /// Prefer [`Self::positions_refs`] in new native code.
5789    #[must_use]
5790    pub fn positions(
5791        &self,
5792        venue: Option<&Venue>,
5793        instrument_id: Option<&InstrumentId>,
5794        strategy_id: Option<&StrategyId>,
5795        account_id: Option<&AccountId>,
5796        side: Option<PositionSide>,
5797    ) -> Vec<PositionRef<'_>> {
5798        self.positions_refs(venue, instrument_id, strategy_id, account_id, side)
5799    }
5800
5801    /// Returns borrows of all open positions matching the optional filter parameters.
5802    #[must_use]
5803    pub fn positions_open_refs(
5804        &self,
5805        venue: Option<&Venue>,
5806        instrument_id: Option<&InstrumentId>,
5807        strategy_id: Option<&StrategyId>,
5808        account_id: Option<&AccountId>,
5809        side: Option<PositionSide>,
5810    ) -> Vec<PositionRef<'_>> {
5811        let position_ids = self.position_open_ids(venue, instrument_id, strategy_id, account_id);
5812        self.get_positions_for_ids(&position_ids, side)
5813    }
5814
5815    /// Returns borrows of all open positions matching the optional filter parameters.
5816    ///
5817    /// Prefer [`Self::positions_open_refs`] in new native code.
5818    #[must_use]
5819    pub fn positions_open(
5820        &self,
5821        venue: Option<&Venue>,
5822        instrument_id: Option<&InstrumentId>,
5823        strategy_id: Option<&StrategyId>,
5824        account_id: Option<&AccountId>,
5825        side: Option<PositionSide>,
5826    ) -> Vec<PositionRef<'_>> {
5827        self.positions_open_refs(venue, instrument_id, strategy_id, account_id, side)
5828    }
5829
5830    /// Returns borrows of all closed positions matching the optional filter parameters.
5831    #[must_use]
5832    pub fn positions_closed_refs(
5833        &self,
5834        venue: Option<&Venue>,
5835        instrument_id: Option<&InstrumentId>,
5836        strategy_id: Option<&StrategyId>,
5837        account_id: Option<&AccountId>,
5838        side: Option<PositionSide>,
5839    ) -> Vec<PositionRef<'_>> {
5840        let position_ids = self.position_closed_ids(venue, instrument_id, strategy_id, account_id);
5841        self.get_positions_for_ids(&position_ids, side)
5842    }
5843
5844    /// Returns borrows of all closed positions matching the optional filter parameters.
5845    ///
5846    /// Prefer [`Self::positions_closed_refs`] in new native code.
5847    #[must_use]
5848    pub fn positions_closed(
5849        &self,
5850        venue: Option<&Venue>,
5851        instrument_id: Option<&InstrumentId>,
5852        strategy_id: Option<&StrategyId>,
5853        account_id: Option<&AccountId>,
5854        side: Option<PositionSide>,
5855    ) -> Vec<PositionRef<'_>> {
5856        self.positions_closed_refs(venue, instrument_id, strategy_id, account_id, side)
5857    }
5858
5859    /// Returns whether a position with the `position_id` exists.
5860    #[must_use]
5861    pub fn position_exists(&self, position_id: &PositionId) -> bool {
5862        self.index.positions.contains(position_id)
5863    }
5864
5865    /// Returns whether a position with the `position_id` is open.
5866    #[must_use]
5867    pub fn is_position_open(&self, position_id: &PositionId) -> bool {
5868        self.index.positions_open.contains(position_id)
5869    }
5870
5871    /// Returns whether a position with the `position_id` is closed.
5872    #[must_use]
5873    pub fn is_position_closed(&self, position_id: &PositionId) -> bool {
5874        self.index.positions_closed.contains(position_id)
5875    }
5876
5877    /// Returns the count of all open positions.
5878    #[must_use]
5879    pub fn positions_open_count(
5880        &self,
5881        venue: Option<&Venue>,
5882        instrument_id: Option<&InstrumentId>,
5883        strategy_id: Option<&StrategyId>,
5884        account_id: Option<&AccountId>,
5885        side: Option<PositionSide>,
5886    ) -> usize {
5887        self.count_positions_in_bucket(
5888            &self.index.positions_open,
5889            venue,
5890            instrument_id,
5891            strategy_id,
5892            account_id,
5893            side,
5894        )
5895    }
5896
5897    /// Returns the count of all closed positions.
5898    #[must_use]
5899    pub fn positions_closed_count(
5900        &self,
5901        venue: Option<&Venue>,
5902        instrument_id: Option<&InstrumentId>,
5903        strategy_id: Option<&StrategyId>,
5904        account_id: Option<&AccountId>,
5905        side: Option<PositionSide>,
5906    ) -> usize {
5907        self.count_positions_in_bucket(
5908            &self.index.positions_closed,
5909            venue,
5910            instrument_id,
5911            strategy_id,
5912            account_id,
5913            side,
5914        )
5915    }
5916
5917    /// Returns the count of all positions.
5918    #[must_use]
5919    pub fn positions_total_count(
5920        &self,
5921        venue: Option<&Venue>,
5922        instrument_id: Option<&InstrumentId>,
5923        strategy_id: Option<&StrategyId>,
5924        account_id: Option<&AccountId>,
5925        side: Option<PositionSide>,
5926    ) -> usize {
5927        self.count_positions_in_bucket(
5928            &self.index.positions,
5929            venue,
5930            instrument_id,
5931            strategy_id,
5932            account_id,
5933            side,
5934        )
5935    }
5936
5937    /// Returns whether any open position matches the optional filter parameters.
5938    ///
5939    /// Short-circuits on the first match, avoiding the full intersection walk performed by
5940    /// [`Self::positions_open_count`]. Prefer this over `positions_open_count(...) > 0` when
5941    /// only existence matters.
5942    #[must_use]
5943    pub fn has_positions_open(
5944        &self,
5945        venue: Option<&Venue>,
5946        instrument_id: Option<&InstrumentId>,
5947        strategy_id: Option<&StrategyId>,
5948        account_id: Option<&AccountId>,
5949        side: Option<PositionSide>,
5950    ) -> bool {
5951        self.any_positions_in_bucket(
5952            &self.index.positions_open,
5953            venue,
5954            instrument_id,
5955            strategy_id,
5956            account_id,
5957            side,
5958        )
5959    }
5960
5961    /// Returns whether any closed position matches the optional filter parameters.
5962    #[must_use]
5963    pub fn has_positions_closed(
5964        &self,
5965        venue: Option<&Venue>,
5966        instrument_id: Option<&InstrumentId>,
5967        strategy_id: Option<&StrategyId>,
5968        account_id: Option<&AccountId>,
5969        side: Option<PositionSide>,
5970    ) -> bool {
5971        self.any_positions_in_bucket(
5972            &self.index.positions_closed,
5973            venue,
5974            instrument_id,
5975            strategy_id,
5976            account_id,
5977            side,
5978        )
5979    }
5980
5981    /// Returns whether any position (open or closed) matches the optional filter parameters.
5982    #[must_use]
5983    pub fn has_positions(
5984        &self,
5985        venue: Option<&Venue>,
5986        instrument_id: Option<&InstrumentId>,
5987        strategy_id: Option<&StrategyId>,
5988        account_id: Option<&AccountId>,
5989        side: Option<PositionSide>,
5990    ) -> bool {
5991        self.any_positions_in_bucket(
5992            &self.index.positions,
5993            venue,
5994            instrument_id,
5995            strategy_id,
5996            account_id,
5997            side,
5998        )
5999    }
6000
6001    // -- STRATEGY QUERIES ------------------------------------------------------------------------
6002
6003    /// Gets a reference to the strategy ID for the `client_order_id` (if found).
6004    #[must_use]
6005    pub fn strategy_id_for_order(&self, client_order_id: &ClientOrderId) -> Option<&StrategyId> {
6006        self.index.order_strategy.get(client_order_id)
6007    }
6008
6009    /// Gets a reference to the strategy ID for the `position_id` (if found).
6010    #[must_use]
6011    pub fn strategy_id_for_position(&self, position_id: &PositionId) -> Option<&StrategyId> {
6012        self.index.position_strategy.get(position_id)
6013    }
6014
6015    // -- GENERAL ---------------------------------------------------------------------------------
6016
6017    /// Gets a reference to the general value for the `key` (if found).
6018    ///
6019    /// # Errors
6020    ///
6021    /// Returns an error if the `key` is invalid.
6022    pub fn get(&self, key: &str) -> anyhow::Result<Option<&Bytes>> {
6023        check_valid_string_utf8(key, stringify!(key))?;
6024
6025        Ok(self.general.get(key))
6026    }
6027
6028    // -- DATA QUERIES ----------------------------------------------------------------------------
6029
6030    /// Returns the price for the `instrument_id` and `price_type` (if found).
6031    ///
6032    /// For `Mid`, returns `None` if no quote is cached or either price is a sentinel.
6033    /// For quote precision `p`, the midpoint has precision `p + 1` when exactly representable,
6034    /// otherwise `p`, rounded half-even if necessary. The fallback applies when the precision
6035    /// limit or raw range rules out `p + 1`, and when `p` is the maximum float precision (16),
6036    /// so a midpoint of a float-convertible quote stays float-convertible.
6037    #[must_use]
6038    pub fn price(&self, instrument_id: &InstrumentId, price_type: PriceType) -> Option<Price> {
6039        match price_type {
6040            PriceType::Bid => self
6041                .quotes
6042                .get(instrument_id)
6043                .and_then(|quotes| quotes.front().map(|quote| quote.bid_price)),
6044            PriceType::Ask => self
6045                .quotes
6046                .get(instrument_id)
6047                .and_then(|quotes| quotes.front().map(|quote| quote.ask_price)),
6048            PriceType::Mid => self.quotes.get(instrument_id).and_then(|quotes| {
6049                let quote = quotes.front()?;
6050                let bid = quote.bid_price;
6051                let ask = quote.ask_price;
6052                if bid.is_undefined() || bid.is_error() || ask.is_undefined() || ask.is_error() {
6053                    return None;
6054                }
6055
6056                // Also rejects `ERROR_PRICE`, whose precision is 255
6057                let precision = bid.precision;
6058                check_fixed_precision(precision).ok()?;
6059
6060                #[allow(
6061                    clippy::useless_conversion,
6062                    reason = "i128::from is a widening conversion when PriceRaw is i64"
6063                )]
6064                let (mut bid_raw, mut ask_raw) = (i128::from(bid.raw()), i128::from(ask.raw()));
6065
6066                if precision < FIXED_PRECISION {
6067                    let factor = 10_i128.pow(u32::from(FIXED_PRECISION - precision));
6068                    bid_raw = (bid_raw / factor) * factor;
6069                    ask_raw = (ask_raw / factor) * factor;
6070                }
6071                let sum = bid_raw.checked_add(ask_raw)?;
6072
6073                if precision < FIXED_PRECISION {
6074                    let raw = PriceRaw::try_from(sum / 2).ok()?;
6075                    return Price::from_raw_checked(raw, precision + 1).ok();
6076                }
6077
6078                if precision != MAX_FLOAT_PRECISION
6079                    && let Some(price) = sum
6080                        .checked_mul(5)
6081                        .and_then(|raw| PriceRaw::try_from(raw).ok())
6082                        .and_then(|raw| Price::from_raw_checked(raw, precision + 1).ok())
6083                {
6084                    return Some(price);
6085                }
6086
6087                let mut raw = sum / 2;
6088                if sum % 2 != 0 && raw % 2 != 0 {
6089                    raw += sum.signum();
6090                }
6091                let raw = PriceRaw::try_from(raw).ok()?;
6092                Price::from_raw_checked(raw, precision).ok()
6093            }),
6094            PriceType::Last => self
6095                .trades
6096                .get(instrument_id)
6097                .and_then(|trades| trades.front().map(|trade| trade.price)),
6098            PriceType::Mark => self
6099                .mark_prices
6100                .get(instrument_id)
6101                .and_then(|marks| marks.front().map(|mark| mark.value)),
6102        }
6103    }
6104
6105    /// Gets all quotes for the `instrument_id`.
6106    #[must_use]
6107    pub fn quotes(&self, instrument_id: &InstrumentId) -> Option<Vec<QuoteTick>> {
6108        self.quotes
6109            .get(instrument_id)
6110            .map(|quotes| quotes.iter().copied().collect())
6111    }
6112
6113    /// Gets all trades for the `instrument_id`.
6114    #[must_use]
6115    pub fn trades(&self, instrument_id: &InstrumentId) -> Option<Vec<TradeTick>> {
6116        self.trades
6117            .get(instrument_id)
6118            .map(|trades| trades.iter().copied().collect())
6119    }
6120
6121    /// Gets all mark price updates for the `instrument_id`.
6122    #[must_use]
6123    pub fn mark_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<MarkPriceUpdate>> {
6124        self.mark_prices
6125            .get(instrument_id)
6126            .map(|mark_prices| mark_prices.iter().copied().collect())
6127    }
6128
6129    /// Gets all index price updates for the `instrument_id`.
6130    #[must_use]
6131    pub fn index_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<IndexPriceUpdate>> {
6132        self.index_prices
6133            .get(instrument_id)
6134            .map(|index_prices| index_prices.iter().copied().collect())
6135    }
6136
6137    /// Gets all funding rate updates for the `instrument_id`.
6138    #[must_use]
6139    pub fn funding_rates(&self, instrument_id: &InstrumentId) -> Option<Vec<FundingRateUpdate>> {
6140        self.funding_rates
6141            .get(instrument_id)
6142            .map(|funding_rates| funding_rates.iter().copied().collect())
6143    }
6144
6145    /// Gets all instrument status updates for the `instrument_id`.
6146    #[must_use]
6147    pub fn instrument_statuses(
6148        &self,
6149        instrument_id: &InstrumentId,
6150    ) -> Option<Vec<InstrumentStatus>> {
6151        self.instrument_statuses
6152            .get(instrument_id)
6153            .map(|statuses| statuses.iter().copied().collect())
6154    }
6155
6156    /// Gets all bars for the `bar_type`.
6157    #[must_use]
6158    pub fn bars(&self, bar_type: &BarType) -> Option<Vec<Bar>> {
6159        self.bars
6160            .get(bar_type)
6161            .map(|bars| bars.iter().copied().collect())
6162    }
6163
6164    /// Gets a reference to the order book for the `instrument_id`.
6165    #[must_use]
6166    pub fn order_book(&self, instrument_id: &InstrumentId) -> Option<&OrderBook> {
6167        self.books.get(instrument_id)
6168    }
6169
6170    /// Gets a reference to the order book for the `instrument_id`.
6171    ///
6172    /// # Errors
6173    ///
6174    /// Returns an error if the order book is not present in the cache.
6175    pub fn try_order_book(
6176        &self,
6177        instrument_id: &InstrumentId,
6178    ) -> Result<&OrderBook, OrderBookLookupError> {
6179        self.books
6180            .get(instrument_id)
6181            .ok_or_else(|| OrderBookLookupError::not_found(*instrument_id))
6182    }
6183
6184    /// Gets a reference to the order book for the `instrument_id`.
6185    #[must_use]
6186    pub fn order_book_mut(&mut self, instrument_id: &InstrumentId) -> Option<&mut OrderBook> {
6187        self.books.get_mut(instrument_id)
6188    }
6189
6190    /// Gets a reference to the own order book for the `instrument_id`.
6191    #[must_use]
6192    pub fn own_order_book(&self, instrument_id: &InstrumentId) -> Option<&OwnOrderBook> {
6193        self.own_books.get(instrument_id)
6194    }
6195
6196    /// Gets a reference to the own order book for the `instrument_id`.
6197    ///
6198    /// # Errors
6199    ///
6200    /// Returns an error if the own order book is not present in the cache.
6201    pub fn try_own_order_book(
6202        &self,
6203        instrument_id: &InstrumentId,
6204    ) -> Result<&OwnOrderBook, OwnOrderBookLookupError> {
6205        self.own_books
6206            .get(instrument_id)
6207            .ok_or_else(|| OwnOrderBookLookupError::not_found(*instrument_id))
6208    }
6209
6210    /// Gets a reference to the own order book for the `instrument_id`.
6211    #[must_use]
6212    pub fn own_order_book_mut(
6213        &mut self,
6214        instrument_id: &InstrumentId,
6215    ) -> Option<&mut OwnOrderBook> {
6216        self.own_books.get_mut(instrument_id)
6217    }
6218
6219    /// Gets a reference to the latest quote for the `instrument_id`.
6220    #[must_use]
6221    pub fn quote(&self, instrument_id: &InstrumentId) -> Option<&QuoteTick> {
6222        self.quotes
6223            .get(instrument_id)
6224            .and_then(|quotes| quotes.front())
6225    }
6226
6227    /// Gets a reference to the quote at `index` for the `instrument_id`.
6228    ///
6229    /// Index 0 is the most recent.
6230    #[must_use]
6231    pub fn quote_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&QuoteTick> {
6232        self.quotes
6233            .get(instrument_id)
6234            .and_then(|quotes| quotes.get(index))
6235    }
6236
6237    /// Gets a reference to the latest trade for the `instrument_id`.
6238    #[must_use]
6239    pub fn trade(&self, instrument_id: &InstrumentId) -> Option<&TradeTick> {
6240        self.trades
6241            .get(instrument_id)
6242            .and_then(|trades| trades.front())
6243    }
6244
6245    /// Gets a reference to the trade at `index` for the `instrument_id`.
6246    ///
6247    /// Index 0 is the most recent.
6248    #[must_use]
6249    pub fn trade_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&TradeTick> {
6250        self.trades
6251            .get(instrument_id)
6252            .and_then(|trades| trades.get(index))
6253    }
6254
6255    /// Gets a reference to the latest mark price update for the `instrument_id`.
6256    #[must_use]
6257    pub fn mark_price(&self, instrument_id: &InstrumentId) -> Option<&MarkPriceUpdate> {
6258        self.mark_prices
6259            .get(instrument_id)
6260            .and_then(|mark_prices| mark_prices.front())
6261    }
6262
6263    /// Gets a reference to the latest index price update for the `instrument_id`.
6264    #[must_use]
6265    pub fn index_price(&self, instrument_id: &InstrumentId) -> Option<&IndexPriceUpdate> {
6266        self.index_prices
6267            .get(instrument_id)
6268            .and_then(|index_prices| index_prices.front())
6269    }
6270
6271    /// Gets a reference to the latest funding rate update for the `instrument_id`.
6272    #[must_use]
6273    pub fn funding_rate(&self, instrument_id: &InstrumentId) -> Option<&FundingRateUpdate> {
6274        self.funding_rates
6275            .get(instrument_id)
6276            .and_then(|funding_rates| funding_rates.front())
6277    }
6278
6279    /// Gets a reference to the latest instrument status update for the `instrument_id`.
6280    #[must_use]
6281    pub fn instrument_status(&self, instrument_id: &InstrumentId) -> Option<&InstrumentStatus> {
6282        self.instrument_statuses
6283            .get(instrument_id)
6284            .and_then(|statuses| statuses.front())
6285    }
6286
6287    /// Returns the close cached for `instrument_id`, if present.
6288    #[must_use]
6289    pub fn instrument_close(&self, instrument_id: &InstrumentId) -> Option<&InstrumentClose> {
6290        self.instrument_closes.get(instrument_id)
6291    }
6292
6293    /// Returns references to all instrument IDs with a cached close.
6294    #[must_use]
6295    pub fn instrument_close_ids(&self) -> Vec<&InstrumentId> {
6296        self.instrument_closes.keys().collect()
6297    }
6298
6299    /// Gets a reference to the latest bar for the `bar_type`.
6300    #[must_use]
6301    pub fn bar(&self, bar_type: &BarType) -> Option<&Bar> {
6302        self.bars.get(bar_type).and_then(|bars| bars.front())
6303    }
6304
6305    /// Gets a reference to the bar at `index` for the `bar_type`.
6306    ///
6307    /// Index 0 is the most recent.
6308    #[must_use]
6309    pub fn bar_at_index(&self, bar_type: &BarType, index: usize) -> Option<&Bar> {
6310        self.bars.get(bar_type).and_then(|bars| bars.get(index))
6311    }
6312
6313    /// Gets the order book update count for the `instrument_id`.
6314    #[must_use]
6315    pub fn book_update_count(&self, instrument_id: &InstrumentId) -> usize {
6316        self.books
6317            .get(instrument_id)
6318            .map_or(0, |book| book.update_count) as usize
6319    }
6320
6321    /// Gets the quote tick count for the `instrument_id`.
6322    #[must_use]
6323    pub fn quote_count(&self, instrument_id: &InstrumentId) -> usize {
6324        self.quotes
6325            .get(instrument_id)
6326            .map_or(0, BoundedVecDeque::len)
6327    }
6328
6329    /// Gets the trade tick count for the `instrument_id`.
6330    #[must_use]
6331    pub fn trade_count(&self, instrument_id: &InstrumentId) -> usize {
6332        self.trades
6333            .get(instrument_id)
6334            .map_or(0, BoundedVecDeque::len)
6335    }
6336
6337    /// Gets the mark price update count for the `instrument_id`.
6338    #[must_use]
6339    pub fn mark_price_count(&self, instrument_id: &InstrumentId) -> usize {
6340        self.mark_prices
6341            .get(instrument_id)
6342            .map_or(0, BoundedVecDeque::len)
6343    }
6344
6345    /// Gets the index price update count for the `instrument_id`.
6346    #[must_use]
6347    pub fn index_price_count(&self, instrument_id: &InstrumentId) -> usize {
6348        self.index_prices
6349            .get(instrument_id)
6350            .map_or(0, BoundedVecDeque::len)
6351    }
6352
6353    /// Gets the funding rate update count for the `instrument_id`.
6354    #[must_use]
6355    pub fn funding_rate_count(&self, instrument_id: &InstrumentId) -> usize {
6356        self.funding_rates
6357            .get(instrument_id)
6358            .map_or(0, BoundedVecDeque::len)
6359    }
6360
6361    /// Gets the instrument status update count for the `instrument_id`.
6362    #[must_use]
6363    pub fn instrument_status_count(&self, instrument_id: &InstrumentId) -> usize {
6364        self.instrument_statuses
6365            .get(instrument_id)
6366            .map_or(0, BoundedVecDeque::len)
6367    }
6368
6369    /// Gets the bar count for the `instrument_id`.
6370    #[must_use]
6371    pub fn bar_count(&self, bar_type: &BarType) -> usize {
6372        self.bars.get(bar_type).map_or(0, BoundedVecDeque::len)
6373    }
6374
6375    /// Returns whether the cache contains an order book for the `instrument_id`.
6376    #[must_use]
6377    pub fn has_order_book(&self, instrument_id: &InstrumentId) -> bool {
6378        self.books.contains_key(instrument_id)
6379    }
6380
6381    /// Returns whether the cache contains quotes for the `instrument_id`.
6382    #[must_use]
6383    pub fn has_quote_ticks(&self, instrument_id: &InstrumentId) -> bool {
6384        self.quote_count(instrument_id) > 0
6385    }
6386
6387    /// Returns whether the cache contains trades for the `instrument_id`.
6388    #[must_use]
6389    pub fn has_trade_ticks(&self, instrument_id: &InstrumentId) -> bool {
6390        self.trade_count(instrument_id) > 0
6391    }
6392
6393    /// Returns whether the cache contains mark price updates for the `instrument_id`.
6394    #[must_use]
6395    pub fn has_mark_prices(&self, instrument_id: &InstrumentId) -> bool {
6396        self.mark_price_count(instrument_id) > 0
6397    }
6398
6399    /// Returns whether the cache contains index price updates for the `instrument_id`.
6400    #[must_use]
6401    pub fn has_index_prices(&self, instrument_id: &InstrumentId) -> bool {
6402        self.index_price_count(instrument_id) > 0
6403    }
6404
6405    /// Returns whether the cache contains funding rate updates for the `instrument_id`.
6406    #[must_use]
6407    pub fn has_funding_rates(&self, instrument_id: &InstrumentId) -> bool {
6408        self.funding_rate_count(instrument_id) > 0
6409    }
6410
6411    /// Returns whether the cache contains instrument status updates for the `instrument_id`.
6412    #[must_use]
6413    pub fn has_instrument_statuses(&self, instrument_id: &InstrumentId) -> bool {
6414        self.instrument_status_count(instrument_id) > 0
6415    }
6416
6417    /// Returns whether the cache contains a close for the `instrument_id`.
6418    #[must_use]
6419    pub fn has_instrument_close(&self, instrument_id: &InstrumentId) -> bool {
6420        self.instrument_closes.contains_key(instrument_id)
6421    }
6422
6423    /// Returns whether the cache contains bars for the `bar_type`.
6424    #[must_use]
6425    pub fn has_bars(&self, bar_type: &BarType) -> bool {
6426        self.bar_count(bar_type) > 0
6427    }
6428
6429    #[must_use]
6430    pub fn get_xrate(
6431        &self,
6432        venue: Venue,
6433        from_currency: Currency,
6434        to_currency: Currency,
6435        price_type: PriceType,
6436    ) -> Option<Decimal> {
6437        match self.try_get_xrate(venue, from_currency, to_currency, price_type) {
6438            Ok(rate) => rate,
6439            Err(e) => {
6440                log::error!("Failed to calculate xrate: {e}");
6441                None
6442            }
6443        }
6444    }
6445
6446    /// Tries to calculate the exchange rate without logging calculation errors.
6447    ///
6448    /// # Errors
6449    ///
6450    /// Returns an error if the cached quotes cannot form a valid exchange rate calculation.
6451    pub fn try_get_xrate(
6452        &self,
6453        venue: Venue,
6454        from_currency: Currency,
6455        to_currency: Currency,
6456        price_type: PriceType,
6457    ) -> anyhow::Result<Option<Decimal>> {
6458        if from_currency == to_currency {
6459            // When the source and target currencies are identical,
6460            // no conversion is needed; return an exchange rate of one.
6461            return Ok(Some(Decimal::ONE));
6462        }
6463
6464        let (bid_quote, ask_quote) = self.build_quote_table(&venue);
6465
6466        get_exchange_rate(
6467            from_currency.code,
6468            to_currency.code,
6469            price_type,
6470            bid_quote,
6471            ask_quote,
6472        )
6473    }
6474
6475    fn build_quote_table(
6476        &self,
6477        venue: &Venue,
6478    ) -> (AHashMap<Ustr, Decimal>, AHashMap<Ustr, Decimal>) {
6479        let mut bid_quotes = AHashMap::new();
6480        let mut ask_quotes = AHashMap::new();
6481        let mut quote_sources = AHashMap::new();
6482        let mut bar_quotes = None;
6483        let mut pair_buffer = String::new();
6484
6485        for (instrument_id, instrument) in &self.instruments {
6486            if instrument_id.venue != *venue {
6487                continue;
6488            }
6489
6490            let Some(base_currency) = instrument.base_currency() else {
6491                continue;
6492            };
6493
6494            let (bid_price, ask_price) = if let Some(ticks) = self.quotes.get(instrument_id) {
6495                if let Some(tick) = ticks.front() {
6496                    (tick.bid_price, tick.ask_price)
6497                } else {
6498                    continue; // Empty ticks vector
6499                }
6500            } else {
6501                let quotes = bar_quotes.get_or_insert_with(|| self.build_bar_quote_table(venue));
6502                match (
6503                    quotes.get(&(*instrument_id, PriceType::Bid)),
6504                    quotes.get(&(*instrument_id, PriceType::Ask)),
6505                ) {
6506                    (Some((_, bid_bar)), Some((_, ask_bar))) => (bid_bar.close, ask_bar.close),
6507                    _ => continue,
6508                }
6509            };
6510
6511            let base = base_currency.code.as_str();
6512            let quote = instrument.quote_currency().code.as_str();
6513            pair_buffer.clear();
6514            pair_buffer.reserve(base.len() + 1 + quote.len());
6515            pair_buffer.push_str(base);
6516            pair_buffer.push('/');
6517            pair_buffer.push_str(quote);
6518            let pair = Ustr::from(pair_buffer.as_str());
6519            let preference = (
6520                bid_price.is_positive() && ask_price.is_positive(),
6521                instrument.instrument_class() == InstrumentClass::Spot,
6522                Reverse(*instrument_id),
6523            );
6524
6525            if quote_sources
6526                .get(&pair)
6527                .is_some_and(|current| current >= &preference)
6528            {
6529                continue;
6530            }
6531
6532            bid_quotes.insert(pair, bid_price.as_decimal());
6533            ask_quotes.insert(pair, ask_price.as_decimal());
6534            quote_sources.insert(pair, preference);
6535        }
6536
6537        (bid_quotes, ask_quotes)
6538    }
6539
6540    fn build_bar_quote_table(
6541        &self,
6542        venue: &Venue,
6543    ) -> AHashMap<(InstrumentId, PriceType), (&BarType, &Bar)> {
6544        let mut quotes: AHashMap<_, (&BarType, &Bar)> = AHashMap::new();
6545
6546        for (bar_type, bars) in &self.bars {
6547            let instrument_id = bar_type.instrument_id();
6548            let price_type = bar_type.spec().price_type;
6549
6550            if instrument_id.venue != *venue
6551                || !matches!(price_type, PriceType::Bid | PriceType::Ask)
6552            {
6553                continue;
6554            }
6555
6556            let Some(bar) = bars.front() else {
6557                continue;
6558            };
6559
6560            // Select the newest front bar per side, breaking timestamp ties by bar type
6561            quotes
6562                .entry((instrument_id, price_type))
6563                .and_modify(|current| {
6564                    if (current.1.ts_init, current.0) < (bar.ts_init, bar_type) {
6565                        *current = (bar_type, bar);
6566                    }
6567                })
6568                .or_insert((bar_type, bar));
6569        }
6570
6571        quotes
6572    }
6573
6574    /// Returns the mark exchange rate for the given currency pair, or `None` if not set.
6575    #[must_use]
6576    pub fn get_mark_xrate(&self, from_currency: Currency, to_currency: Currency) -> Option<f64> {
6577        self.mark_xrates.get(&(from_currency, to_currency)).copied()
6578    }
6579
6580    /// Sets the mark exchange rate for the given currency pair and automatically sets the inverse rate.
6581    ///
6582    /// # Panics
6583    ///
6584    /// Panics if `xrate` is not positive.
6585    pub fn set_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency, xrate: f64) {
6586        assert!(xrate > 0.0, "xrate was zero");
6587        self.mark_xrates.insert((from_currency, to_currency), xrate);
6588        self.mark_xrates
6589            .insert((to_currency, from_currency), 1.0 / xrate);
6590    }
6591
6592    /// Clears the mark exchange rate for the given currency pair direction.
6593    ///
6594    /// Removes only the `(from_currency, to_currency)` entry; the inverse rate written
6595    /// by [`Self::set_mark_xrate`] is retained until cleared separately or
6596    /// [`Self::clear_mark_xrates`] is called.
6597    pub fn clear_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency) {
6598        let _ = self.mark_xrates.remove(&(from_currency, to_currency));
6599    }
6600
6601    /// Clears all mark exchange rates.
6602    pub fn clear_mark_xrates(&mut self) {
6603        self.mark_xrates.clear();
6604    }
6605
6606    /// Returns a reference to the currency for the `code` (if found).
6607    #[must_use]
6608    pub fn currency(&self, code: &Ustr) -> Option<&Currency> {
6609        self.currencies.get(code)
6610    }
6611
6612    /// Returns a reference to the currency for the `code`.
6613    ///
6614    /// # Errors
6615    ///
6616    /// Returns an error if the currency is not present in the cache.
6617    pub fn try_currency(&self, code: &Ustr) -> Result<&Currency, CurrencyLookupError> {
6618        self.currencies
6619            .get(code)
6620            .ok_or_else(|| CurrencyLookupError::not_found(*code))
6621    }
6622
6623    // -- INSTRUMENT QUERIES ----------------------------------------------------------------------
6624
6625    /// Returns a reference to the instrument for the `instrument_id` (if found).
6626    #[must_use]
6627    pub fn instrument(&self, instrument_id: &InstrumentId) -> Option<&InstrumentAny> {
6628        self.instruments.get(instrument_id)
6629    }
6630
6631    /// Returns a reference to the instrument for the `instrument_id`.
6632    ///
6633    /// # Errors
6634    ///
6635    /// Returns an error if the instrument is not present in the cache.
6636    pub fn try_instrument(
6637        &self,
6638        instrument_id: &InstrumentId,
6639    ) -> Result<&InstrumentAny, InstrumentLookupError> {
6640        self.instruments
6641            .get(instrument_id)
6642            .ok_or_else(|| InstrumentLookupError::not_found(*instrument_id))
6643    }
6644
6645    /// Returns references to all instrument IDs for the `venue`.
6646    #[must_use]
6647    pub fn instrument_ids(&self, venue: Option<&Venue>) -> Vec<&InstrumentId> {
6648        match venue {
6649            Some(v) => self.instruments.keys().filter(|i| &i.venue == v).collect(),
6650            None => self.instruments.keys().collect(),
6651        }
6652    }
6653
6654    /// Returns references to all instruments for the `venue`.
6655    #[must_use]
6656    pub fn instruments(&self, venue: &Venue, underlying: Option<&Ustr>) -> Vec<&InstrumentAny> {
6657        self.instruments
6658            .values()
6659            .filter(|i| &i.id().venue == venue)
6660            .filter(|i| underlying.is_none_or(|u| i.underlying() == Some(*u)))
6661            .collect()
6662    }
6663
6664    /// Returns references to all instruments for the `venue` whose underlying
6665    /// equals `root` and whose [`InstrumentClass`] equals `class`.
6666    ///
6667    /// Use when expanding a parent-symbol subscription: filtering by class as
6668    /// well as root prevents leaves of a different class (e.g. options when
6669    /// the user asked for futures, or vice versa) from being pulled in.
6670    #[must_use]
6671    pub fn instruments_by_parent(
6672        &self,
6673        venue: &Venue,
6674        root: &Ustr,
6675        class: InstrumentClass,
6676    ) -> Vec<&InstrumentAny> {
6677        self.instruments
6678            .values()
6679            .filter(|i| &i.id().venue == venue)
6680            .filter(|i| i.underlying() == Some(*root))
6681            .filter(|i| i.instrument_class() == class)
6682            .collect()
6683    }
6684
6685    /// Returns references to all bar types contained in the cache.
6686    #[must_use]
6687    pub fn bar_types(
6688        &self,
6689        instrument_id: Option<&InstrumentId>,
6690        price_type: Option<&PriceType>,
6691        aggregation_source: AggregationSource,
6692    ) -> Vec<&BarType> {
6693        let mut bar_types = self
6694            .bars
6695            .keys()
6696            .filter(|bar_type| bar_type.aggregation_source() == aggregation_source)
6697            .collect::<Vec<&BarType>>();
6698
6699        if let Some(instrument_id) = instrument_id {
6700            bar_types.retain(|bar_type| bar_type.instrument_id() == *instrument_id);
6701        }
6702
6703        if let Some(price_type) = price_type {
6704            bar_types.retain(|bar_type| &bar_type.spec().price_type == price_type);
6705        }
6706
6707        bar_types
6708    }
6709
6710    // -- SYNTHETIC QUERIES -----------------------------------------------------------------------
6711
6712    /// Returns a reference to the synthetic instrument for the `instrument_id` (if found).
6713    #[must_use]
6714    pub fn synthetic(&self, instrument_id: &InstrumentId) -> Option<&SyntheticInstrument> {
6715        self.synthetics.get(instrument_id)
6716    }
6717
6718    /// Returns a reference to the synthetic instrument for the `instrument_id`.
6719    ///
6720    /// # Errors
6721    ///
6722    /// Returns an error if the synthetic instrument is not present in the cache.
6723    pub fn try_synthetic(
6724        &self,
6725        instrument_id: &InstrumentId,
6726    ) -> Result<&SyntheticInstrument, SyntheticInstrumentLookupError> {
6727        self.synthetics
6728            .get(instrument_id)
6729            .ok_or_else(|| SyntheticInstrumentLookupError::not_found(*instrument_id))
6730    }
6731
6732    /// Returns references to instrument IDs for all synthetic instruments contained in the cache.
6733    #[must_use]
6734    pub fn synthetic_ids(&self) -> Vec<&InstrumentId> {
6735        self.synthetics.keys().collect()
6736    }
6737
6738    /// Returns references to all synthetic instruments contained in the cache.
6739    #[must_use]
6740    pub fn synthetics(&self) -> Vec<&SyntheticInstrument> {
6741        self.synthetics.values().collect()
6742    }
6743
6744    // -- ACCOUNT QUERIES -----------------------------------------------------------------------
6745
6746    /// Returns a borrow of the account for the `account_id` (if found).
6747    #[must_use]
6748    pub fn account_ref(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
6749        self.accounts
6750            .get(account_id)
6751            .map(|account_cell| AccountRef::new(account_cell.borrow()))
6752    }
6753
6754    /// Returns a borrow of the account for the `account_id` (if found).
6755    ///
6756    /// Prefer [`Self::account_ref`] in new native code.
6757    #[must_use]
6758    pub fn account(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
6759        self.account_ref(account_id)
6760    }
6761
6762    /// Returns a borrow of the account for the `account_id`.
6763    ///
6764    /// # Errors
6765    ///
6766    /// Returns an error if the account is not present in the cache.
6767    pub fn try_account_ref(
6768        &self,
6769        account_id: &AccountId,
6770    ) -> Result<AccountRef<'_>, AccountLookupError> {
6771        self.accounts
6772            .get(account_id)
6773            .map(|account_cell| AccountRef::new(account_cell.borrow()))
6774            .ok_or_else(|| AccountLookupError::not_found(*account_id))
6775    }
6776
6777    /// Returns a borrow of the account for the `account_id`.
6778    ///
6779    /// Prefer [`Self::try_account_ref`] in new native code.
6780    ///
6781    /// # Errors
6782    ///
6783    /// Returns an error if the account is not present in the cache.
6784    pub fn try_account(
6785        &self,
6786        account_id: &AccountId,
6787    ) -> Result<AccountRef<'_>, AccountLookupError> {
6788        self.try_account_ref(account_id)
6789    }
6790
6791    /// Gets an exclusive write borrow of the account with the `account_id` (if found).
6792    ///
6793    /// Requires `&mut Cache` so cache writes are reachable only by privileged crates that hold
6794    /// `Rc<RefCell<Cache>>` directly. Adapter-facing code receives [`CacheView`], which only
6795    /// exposes immutable cache borrows and therefore cannot reach this method.
6796    ///
6797    /// While the returned [`AccountRefMut`] is alive, no other read or write of the same account
6798    /// is permitted. Drop the borrow before dispatching events or taking any other cache borrow
6799    /// that may re-enter the same account.
6800    #[must_use]
6801    pub fn account_mut(&mut self, account_id: &AccountId) -> Option<AccountRefMut<'_>> {
6802        self.accounts
6803            .get(account_id)
6804            .map(|account_cell| AccountRefMut::new(account_cell.borrow_mut()))
6805    }
6806
6807    /// Gets an owned snapshot of the account with the `account_id` when present and not mutably
6808    /// borrowed.
6809    ///
6810    /// Use when downstream needs an owned [`AccountAny`] that crosses a boundary. The snapshot
6811    /// will not reflect later cache mutations.
6812    #[must_use]
6813    pub fn account_owned(&self, account_id: &AccountId) -> Option<AccountAny> {
6814        self.accounts.get(account_id).and_then(|account_cell| {
6815            account_cell
6816                .try_borrow()
6817                .ok()
6818                .map(|account| account.clone())
6819        })
6820    }
6821
6822    /// Returns a borrow of the account for the `venue` (if found).
6823    ///
6824    /// Returns `None` when more than one account is issued under the `venue`; look those
6825    /// accounts up by account ID instead.
6826    #[must_use]
6827    pub fn account_for_venue(&self, venue: &Venue) -> Option<AccountRef<'_>> {
6828        self.account_id(venue)
6829            .and_then(|account_id| self.accounts.get(account_id))
6830            .map(|account_cell| AccountRef::new(account_cell.borrow()))
6831    }
6832
6833    /// Returns an owned snapshot of the account for the `venue` (if found).
6834    ///
6835    /// Use when downstream needs an owned [`AccountAny`] that crosses a boundary. The snapshot
6836    /// will not reflect later cache mutations. Returns `None` when more than one account is
6837    /// issued under the `venue`.
6838    #[must_use]
6839    pub fn account_for_venue_owned(&self, venue: &Venue) -> Option<AccountAny> {
6840        self.account_id(venue)
6841            .and_then(|account_id| self.accounts.get(account_id))
6842            .map(|account_cell| account_cell.borrow().clone())
6843    }
6844
6845    /// Returns a reference to the account ID for the `venue` (if found).
6846    ///
6847    /// Returns `None` when more than one account is issued under the `venue`.
6848    #[must_use]
6849    pub fn account_id(&self, venue: &Venue) -> Option<&AccountId> {
6850        let mut account_ids = self.index.venue_accounts.get(venue)?.iter();
6851        let account_id = account_ids.next()?;
6852        account_ids.next().is_none().then_some(account_id)
6853    }
6854
6855    /// Returns borrows of all accounts for the `account_id`.
6856    ///
6857    /// Each [`AccountRef`] in the returned vector borrows its underlying cell; mutating any of
6858    /// those accounts while the vector is alive will panic at runtime. Drop the vector before
6859    /// issuing writes.
6860    #[must_use]
6861    pub fn accounts(&self, account_id: &AccountId) -> Vec<AccountRef<'_>> {
6862        self.accounts
6863            .values()
6864            .filter(|account_cell| &account_cell.borrow().id() == account_id)
6865            .map(|account_cell| AccountRef::new(account_cell.borrow()))
6866            .collect()
6867    }
6868
6869    /// Returns owned copies of every account in the cache.
6870    #[must_use]
6871    pub fn accounts_all_owned(&self) -> Vec<AccountAny> {
6872        self.accounts
6873            .values()
6874            .map(|account_cell| account_cell.borrow().clone())
6875            .collect()
6876    }
6877
6878    /// Updates the own order book with an order.
6879    ///
6880    /// This method adds, updates, or removes an order from the own order book
6881    /// based on the order's current state.
6882    ///
6883    /// Orders without prices (MARKET, etc.) are skipped as they cannot be
6884    /// represented in own books.
6885    pub fn update_own_order_book(&mut self, order: &OrderAny) {
6886        if !order.has_price() {
6887            return;
6888        }
6889
6890        let instrument_id = order.instrument_id();
6891
6892        if !self.own_books.contains_key(&instrument_id) {
6893            if order.is_closed() {
6894                return;
6895            }
6896
6897            self.own_books
6898                .insert(instrument_id, OwnOrderBook::new(instrument_id));
6899        }
6900
6901        let Some(own_book) = self.own_books.get_mut(&instrument_id) else {
6902            return;
6903        };
6904
6905        let own_book_order = order.to_own_book_order();
6906
6907        if order.is_closed() {
6908            if let Err(e) = own_book.delete(own_book_order) {
6909                log::debug!(
6910                    "Failed to delete order {} from own book: {e}",
6911                    order.client_order_id(),
6912                );
6913            } else {
6914                log::debug!("Deleted order {} from own book", order.client_order_id());
6915            }
6916        } else {
6917            // Add or update the order in the own book
6918            if let Err(e) = own_book.update(own_book_order) {
6919                log::debug!(
6920                    "Failed to update order {} in own book: {e}; inserting instead",
6921                    order.client_order_id(),
6922                );
6923                own_book.add(own_book_order);
6924            }
6925            log::debug!("Updated order {} in own book", order.client_order_id());
6926        }
6927    }
6928
6929    /// Force removal of an order from own order books and clean up all indexes.
6930    ///
6931    /// This method is used when order event application fails and we need to ensure
6932    /// terminal orders are properly cleaned up from own books and all relevant indexes.
6933    /// Replicates the index cleanup that `update_order` performs for closed orders.
6934    pub fn force_remove_from_own_order_book(&mut self, client_order_id: &ClientOrderId) {
6935        let Some(order_cell) = self.orders.get(client_order_id) else {
6936            return;
6937        };
6938        let order = order_cell.borrow();
6939        let instrument_id = order.instrument_id();
6940        let own_book_order = if order.has_price() {
6941            Some(order.to_own_book_order())
6942        } else {
6943            None
6944        };
6945        drop(order);
6946
6947        self.index.orders_open.remove(client_order_id);
6948        self.index.orders_pending_cancel.remove(client_order_id);
6949        self.index.orders_inflight.remove(client_order_id);
6950        self.index.orders_emulated.remove(client_order_id);
6951        self.index.orders_active_local.remove(client_order_id);
6952
6953        if let Some(own_book) = self.own_books.get_mut(&instrument_id)
6954            && let Some(own_book_order) = own_book_order
6955        {
6956            if let Err(e) = own_book.delete(own_book_order) {
6957                log::debug!("Could not force delete {client_order_id} from own book: {e}");
6958            } else {
6959                log::debug!("Force deleted {client_order_id} from own book");
6960            }
6961        }
6962
6963        self.index.orders_closed.insert(*client_order_id);
6964    }
6965
6966    /// Audit all own order books against active order indexes.
6967    ///
6968    /// Ensures orders absent from the open, inflight, and active-local indexes are removed from
6969    /// own order books.
6970    pub fn audit_own_order_books(&mut self) {
6971        log::debug!("Starting own books audit");
6972        let start = std::time::Instant::now();
6973
6974        let valid_order_ids: AHashSet<ClientOrderId> = self
6975            .index
6976            .orders_open
6977            .iter()
6978            .chain(&self.index.orders_inflight)
6979            .chain(&self.index.orders_active_local)
6980            .copied()
6981            .collect();
6982
6983        for own_book in self.own_books.values_mut() {
6984            own_book.audit_open_orders(&valid_order_ids);
6985        }
6986
6987        log::debug!("Completed own books audit in {:?}", start.elapsed());
6988    }
6989}
6990
6991const POSITION_OMS_KEY_PREFIX: &str = "position_oms:";
6992
6993fn position_oms_key(position_id: PositionId) -> String {
6994    format!("{POSITION_OMS_KEY_PREFIX}{position_id}")
6995}
6996
6997/// Inserts `bar` into the newest-first bars deque for a `bar_type`.
6998///
6999/// A newer bar is pushed, an older `ts_event` is skipped, and an equal
7000/// `ts_event` replaces the front bar for time bars.
7001fn insert_bar(bars: &mut BoundedVecDeque<Bar>, bar: Bar) {
7002    match bars.front() {
7003        None => bars.push_front(bar),
7004        Some(front) => match bar.ts_event.cmp(&front.ts_event) {
7005            Ordering::Greater => bars.push_front(bar),
7006            Ordering::Equal => {
7007                if bar.bar_type.spec().is_time_aggregated() {
7008                    bars.replace_front(bar);
7009                } else {
7010                    bars.push_front(bar);
7011                }
7012            }
7013            Ordering::Less => log::debug!(
7014                "Skipping bar {bar} with `ts_event` older than last bar `ts_event` {}",
7015                front.ts_event,
7016            ),
7017        },
7018    }
7019}
7020
7021/// Inserts `bar` at its ordered position in the newest-first bars deque.
7022///
7023/// Bars older than the front are kept; a time bar replaces the cached bar
7024/// with an equal `ts_event`.
7025fn insert_bar_historical(bars: &mut BoundedVecDeque<Bar>, bar: Bar) {
7026    let index = bars.partition_point(|cached| cached.ts_event > bar.ts_event);
7027
7028    if bar.bar_type.spec().is_time_aggregated()
7029        && bars
7030            .get(index)
7031            .is_some_and(|cached| cached.ts_event == bar.ts_event)
7032    {
7033        bars.replace(index, bar);
7034    } else {
7035        bars.insert(index, bar);
7036    }
7037}