1pub 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::Reverse,
41 fmt::{Debug, Display},
42 rc::Rc,
43 time::{SystemTime, UNIX_EPOCH},
44};
45
46use ahash::{AHashMap, AHashSet};
47pub use api::CacheApi; use bounded::BoundedVecDeque;
49use bytes::Bytes;
50pub use config::CacheConfig; use database::{CacheDatabaseAdapter, CacheMap};
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,
66 check_valid_string_ascii,
67 },
68};
69use nautilus_model::{
70 accounts::{Account, AccountAny},
71 data::{
72 Bar, BarType, FundingRateUpdate, GreeksData, IndexPriceUpdate, InstrumentClose,
73 InstrumentStatus, MarkPriceUpdate, QuoteTick, TradeTick, YieldCurveData,
74 option_chain::OptionGreeks,
75 },
76 enums::{
77 AggregationSource, ContingencyType, InstrumentClass, OmsType, OrderSide, PositionSide,
78 PriceType,
79 },
80 events::{AccountState, OrderEventAny, OrderFilled},
81 identifiers::{
82 AccountId, ActorId, ClientId, ClientOrderId, ExecAlgorithmId, InstrumentId, OrderListId,
83 PositionId, StrategyId, Venue, VenueOrderId,
84 },
85 instruments::{Instrument, InstrumentAny, SyntheticInstrument},
86 orderbook::{
87 OrderBook,
88 own::{OwnOrderBook, should_handle_own_book_order},
89 },
90 orders::{Order, OrderAny, OrderError, OrderList},
91 position::Position,
92 types::{Currency, Money, Price, Quantity},
93};
94pub use position::CacheSnapshotRef;
95use position::PositionSnapshotFrame;
96pub use refs::{AccountRef, AccountRefMut, OrderRef, OrderRefMut, PositionRef, PositionRefMut};
97use rust_decimal::Decimal;
98use ustr::Ustr;
99pub use view::CacheView; use crate::xrate::get_exchange_rate;
102
103#[cfg_attr(
105 feature = "python",
106 pyo3::pyclass(module = "nautilus_trader.common", unsendable)
107)]
108pub struct Cache {
109 config: CacheConfig,
110 index: CacheIndex,
111 database: Option<Box<dyn CacheDatabaseAdapter>>,
112 general: AHashMap<String, Bytes>,
113 currencies: AHashMap<Ustr, Currency>,
114 instruments: AHashMap<InstrumentId, InstrumentAny>,
115 instrument_closes: AHashMap<InstrumentId, InstrumentClose>,
116 synthetics: AHashMap<InstrumentId, SyntheticInstrument>,
117 books: AHashMap<InstrumentId, OrderBook>,
118 own_books: AHashMap<InstrumentId, OwnOrderBook>,
119 quotes: AHashMap<InstrumentId, BoundedVecDeque<QuoteTick>>,
120 trades: AHashMap<InstrumentId, BoundedVecDeque<TradeTick>>,
121 mark_xrates: AHashMap<(Currency, Currency), f64>,
122 mark_prices: AHashMap<InstrumentId, BoundedVecDeque<MarkPriceUpdate>>,
123 index_prices: AHashMap<InstrumentId, BoundedVecDeque<IndexPriceUpdate>>,
124 funding_rates: AHashMap<InstrumentId, BoundedVecDeque<FundingRateUpdate>>,
125 instrument_statuses: AHashMap<InstrumentId, BoundedVecDeque<InstrumentStatus>>,
126 bars: AHashMap<BarType, BoundedVecDeque<Bar>>,
127 greeks: AHashMap<InstrumentId, GreeksData>,
128 option_greeks: AHashMap<InstrumentId, OptionGreeks>,
129 yield_curves: AHashMap<String, YieldCurveData>,
130 external_order_claims: AHashMap<InstrumentId, StrategyId>,
131 accounts: AHashMap<AccountId, SharedCell<AccountAny>>,
132 orders: AHashMap<ClientOrderId, SharedCell<OrderAny>>,
133 order_lists: AHashMap<OrderListId, OrderList>,
134 positions: AHashMap<PositionId, SharedCell<Position>>,
135 position_snapshots: AHashMap<PositionId, Vec<PositionSnapshotFrame>>,
136 position_snapshot_revisions: AHashMap<PositionId, u64>,
137 #[cfg(feature = "defi")]
138 pub(crate) defi: crate::defi::cache::DefiCache,
139}
140
141impl Debug for Cache {
142 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
143 f.debug_struct(stringify!(Cache))
144 .field("config", &self.config)
145 .field("index", &self.index)
146 .field("general", &self.general)
147 .field("currencies", &self.currencies)
148 .field("instruments", &self.instruments)
149 .field("synthetics", &self.synthetics)
150 .field("books", &self.books)
151 .field("own_books", &self.own_books)
152 .field("quotes", &self.quotes)
153 .field("trades", &self.trades)
154 .field("mark_xrates", &self.mark_xrates)
155 .field("mark_prices", &self.mark_prices)
156 .field("index_prices", &self.index_prices)
157 .field("funding_rates", &self.funding_rates)
158 .field("instrument_statuses", &self.instrument_statuses)
159 .field("instrument_closes", &self.instrument_closes)
160 .field("bars", &self.bars)
161 .field("greeks", &self.greeks)
162 .field("option_greeks", &self.option_greeks)
163 .field("yield_curves", &self.yield_curves)
164 .field("external_order_claims", &self.external_order_claims)
165 .field("accounts", &self.accounts)
166 .field("orders", &self.orders)
167 .field("order_lists", &self.order_lists)
168 .field("positions", &self.positions)
169 .field("position_snapshots", &self.position_snapshots)
170 .finish()
171 }
172}
173
174impl Default for Cache {
175 fn default() -> Self {
177 Self::new(Some(CacheConfig::default()), None)
178 }
179}
180
181impl Cache {
182 #[must_use]
184 pub fn new(
192 config: Option<CacheConfig>,
193 database: Option<Box<dyn CacheDatabaseAdapter>>,
194 ) -> Self {
195 Self::try_new(config, database).expect("invalid `CacheConfig`")
196 }
197
198 pub fn try_new(
204 config: Option<CacheConfig>,
205 database: Option<Box<dyn CacheDatabaseAdapter>>,
206 ) -> crate::config::ConfigResult<Self> {
207 let config = config.unwrap_or_default();
208 config.validate()?;
209
210 Ok(Self {
211 config,
212 index: CacheIndex::default(),
213 database,
214 general: AHashMap::new(),
215 currencies: AHashMap::new(),
216 instruments: AHashMap::new(),
217 instrument_closes: AHashMap::new(),
218 synthetics: AHashMap::new(),
219 books: AHashMap::new(),
220 own_books: AHashMap::new(),
221 quotes: AHashMap::new(),
222 trades: AHashMap::new(),
223 mark_xrates: AHashMap::new(),
224 mark_prices: AHashMap::new(),
225 index_prices: AHashMap::new(),
226 funding_rates: AHashMap::new(),
227 instrument_statuses: AHashMap::new(),
228 bars: AHashMap::new(),
229 greeks: AHashMap::new(),
230 option_greeks: AHashMap::new(),
231 yield_curves: AHashMap::new(),
232 external_order_claims: AHashMap::new(),
233 accounts: AHashMap::new(),
234 orders: AHashMap::new(),
235 order_lists: AHashMap::new(),
236 positions: AHashMap::new(),
237 position_snapshots: AHashMap::new(),
238 position_snapshot_revisions: AHashMap::new(),
239 #[cfg(feature = "defi")]
240 defi: crate::defi::cache::DefiCache::default(),
241 })
242 }
243
244 #[must_use]
246 pub fn memory_address(&self) -> String {
247 format!("{:?}", std::ptr::from_ref(self))
248 }
249
250 #[must_use]
252 pub fn external_order_claim(&self, instrument_id: &InstrumentId) -> Option<StrategyId> {
253 self.external_order_claims.get(instrument_id).copied()
254 }
255
256 #[must_use]
261 pub fn external_order_claim_instrument_ids(
262 &self,
263 strategy_id: Option<StrategyId>,
264 ) -> AHashSet<InstrumentId> {
265 self.external_order_claims
266 .iter()
267 .filter_map(|(instrument_id, owner)| {
268 strategy_id
269 .is_none_or(|strategy_id| *owner == strategy_id)
270 .then_some(*instrument_id)
271 })
272 .collect()
273 }
274
275 pub fn set_external_order_claims(
287 &mut self,
288 strategy_id: StrategyId,
289 instrument_ids: &[InstrumentId],
290 ) -> anyhow::Result<()> {
291 let mut requested = AHashSet::with_capacity(instrument_ids.len());
292
293 for instrument_id in instrument_ids {
294 if !requested.insert(*instrument_id) {
295 anyhow::bail!(
296 "External order claim for {instrument_id} appears more than once for {strategy_id}"
297 );
298 }
299
300 if let Some(existing) = self.external_order_claims.get(instrument_id)
301 && *existing != strategy_id
302 {
303 anyhow::bail!(
304 "External order claim for {instrument_id} already exists for {existing}"
305 );
306 }
307 }
308
309 self.external_order_claims
310 .retain(|_, owner| *owner != strategy_id);
311 self.external_order_claims.extend(
312 requested
313 .into_iter()
314 .map(|instrument_id| (instrument_id, strategy_id)),
315 );
316
317 Ok(())
318 }
319
320 pub fn register_external_order_claims(
326 &mut self,
327 strategy_id: StrategyId,
328 instrument_ids: &[InstrumentId],
329 ) -> anyhow::Result<()> {
330 let mut requested = AHashSet::with_capacity(instrument_ids.len());
331
332 for instrument_id in instrument_ids {
333 if !requested.insert(*instrument_id) {
334 anyhow::bail!(
335 "External order claim for {instrument_id} appears more than once for {strategy_id}"
336 );
337 }
338
339 if let Some(existing) = self.external_order_claims.get(instrument_id) {
340 anyhow::bail!(
341 "External order claim for {instrument_id} already exists for {existing}"
342 );
343 }
344 }
345
346 self.external_order_claims.extend(
347 requested
348 .into_iter()
349 .map(|instrument_id| (instrument_id, strategy_id)),
350 );
351
352 Ok(())
353 }
354
355 pub fn set_database(&mut self, database: Box<dyn CacheDatabaseAdapter>) {
359 let type_name = std::any::type_name_of_val(&*database);
360 log::info!("Cache database adapter set: {type_name}");
361 self.database = Some(database);
362 }
363
364 pub fn cache_general(&mut self) -> anyhow::Result<()> {
372 self.general = match &mut self.database {
373 Some(db) => db.load()?,
374 None => AHashMap::new(),
375 };
376
377 log::info!(
378 "Cached {} general object(s) from database",
379 self.general.len()
380 );
381 Ok(())
382 }
383
384 pub async fn cache_all(&mut self) -> anyhow::Result<()> {
393 let cache_map = match &self.database {
394 Some(db) => db.load_all().await?,
395 None => CacheMap::default(),
396 };
397
398 self.currencies = cache_map.currencies;
399 self.instruments = cache_map.instruments;
400 self.instrument_closes = cache_map.instrument_closes;
401 self.synthetics = cache_map.synthetics;
402 self.accounts = cache_map
403 .accounts
404 .into_iter()
405 .map(|(id, account)| (id, SharedCell::new(account)))
406 .collect();
407 self.orders = cache_map
408 .orders
409 .into_iter()
410 .map(|(id, order)| (id, SharedCell::new(order)))
411 .collect();
412 self.positions = cache_map
413 .positions
414 .into_iter()
415 .map(|(id, position)| (id, SharedCell::new(position)))
416 .collect();
417
418 if let Some(db) = &self.database {
419 let order_position = db.load_index_order_position()?;
420 self.index.order_position = self.sanitize_order_position_index(order_position);
421 self.index.order_client = db.load_index_order_client()?;
422 }
423
424 self.cache_position_oms()?;
425 self.assign_position_ids_to_contingencies();
426 Ok(())
427 }
428
429 pub async fn cache_currencies(&mut self) -> anyhow::Result<()> {
435 self.currencies = match &mut self.database {
436 Some(db) => db.load_currencies().await?,
437 None => AHashMap::new(),
438 };
439
440 log::info!("Cached {} currencies from database", self.general.len());
441 Ok(())
442 }
443
444 pub async fn cache_instruments(&mut self) -> anyhow::Result<()> {
450 self.instruments = match &mut self.database {
451 Some(db) => db.load_instruments().await?,
452 None => AHashMap::new(),
453 };
454
455 log::info!("Cached {} instruments from database", self.general.len());
456 Ok(())
457 }
458
459 pub async fn cache_synthetics(&mut self) -> anyhow::Result<()> {
465 self.synthetics = match &mut self.database {
466 Some(db) => db.load_synthetics().await?,
467 None => AHashMap::new(),
468 };
469
470 log::info!(
471 "Cached {} synthetic instruments from database",
472 self.general.len()
473 );
474 Ok(())
475 }
476
477 pub async fn cache_accounts(&mut self) -> anyhow::Result<()> {
483 self.accounts = match &mut self.database {
484 Some(db) => db
485 .load_accounts()
486 .await?
487 .into_iter()
488 .map(|(id, account)| (id, SharedCell::new(account)))
489 .collect(),
490 None => AHashMap::new(),
491 };
492
493 log::info!(
494 "Cached {} synthetic instruments from database",
495 self.general.len()
496 );
497 Ok(())
498 }
499
500 pub async fn cache_orders(&mut self) -> anyhow::Result<()> {
506 self.orders = match &mut self.database {
507 Some(db) => db
508 .load_orders()
509 .await?
510 .into_iter()
511 .map(|(id, order)| (id, SharedCell::new(order)))
512 .collect(),
513 None => AHashMap::new(),
514 };
515
516 if let Some(db) = &self.database {
517 let order_position = db.load_index_order_position()?;
518 self.index.order_position = self.sanitize_order_position_index(order_position);
519 self.index.order_client = db.load_index_order_client()?;
520 }
521
522 log::info!("Cached {} orders from database", self.general.len());
523
524 self.assign_position_ids_to_contingencies();
525 Ok(())
526 }
527
528 fn sanitize_order_position_index(
529 &self,
530 mut order_position: AHashMap<ClientOrderId, PositionId>,
531 ) -> AHashMap<ClientOrderId, PositionId> {
532 let original_len = order_position.len();
533 order_position.retain(|client_order_id, _| self.orders.contains_key(client_order_id));
534 let removed = original_len - order_position.len();
535
536 if removed > 0 {
537 log::warn!(
538 "Filtered {removed} stale order-position index entries without backing orders during cache load"
539 );
540 }
541
542 order_position
543 }
544
545 pub async fn cache_positions(&mut self) -> anyhow::Result<()> {
551 self.positions = match &mut self.database {
552 Some(db) => db
553 .load_positions()
554 .await?
555 .into_iter()
556 .map(|(id, position)| (id, SharedCell::new(position)))
557 .collect(),
558 None => AHashMap::new(),
559 };
560
561 self.cache_position_oms()?;
562 log::info!("Cached {} positions from database", self.general.len());
563 Ok(())
564 }
565
566 fn cache_position_oms(&mut self) -> anyhow::Result<()> {
567 let persisted = match &self.database {
568 Some(database) => database.load()?,
569 None => self.general.clone(),
570 };
571
572 self.general
573 .retain(|key, _| !key.starts_with(POSITION_OMS_KEY_PREFIX));
574
575 for (key, value) in persisted {
576 if !key.starts_with(POSITION_OMS_KEY_PREFIX) {
577 continue;
578 }
579 self.general.insert(key, value);
580 }
581
582 self.index_position_oms();
583 Ok(())
584 }
585
586 pub fn build_index(&mut self) {
588 log::debug!("Building index");
589
590 for account_id in self.accounts.keys() {
592 self.index
593 .venue_account
594 .insert(account_id.get_issuer(), *account_id);
595 }
596
597 for (client_order_id, order_cell) in &self.orders {
599 let order = order_cell.borrow();
600 let instrument_id = order.instrument_id();
601 let venue = instrument_id.venue;
602 let strategy_id = order.strategy_id();
603
604 self.index
606 .venue_orders
607 .entry(venue)
608 .or_default()
609 .insert(*client_order_id);
610
611 if let Some(venue_order_id) = order.venue_order_id() {
614 self.index
615 .venue_order_ids
616 .insert(venue_order_id, *client_order_id);
617 self.index
618 .client_order_ids
619 .insert(*client_order_id, venue_order_id);
620 }
621
622 if let Some(position_id) = order.position_id() {
624 self.index
625 .order_position
626 .insert(*client_order_id, position_id);
627 }
628
629 self.index
631 .order_strategy
632 .insert(*client_order_id, strategy_id);
633
634 self.index
636 .instrument_orders
637 .entry(instrument_id)
638 .or_default()
639 .insert(*client_order_id);
640
641 self.index
643 .strategy_orders
644 .entry(strategy_id)
645 .or_default()
646 .insert(*client_order_id);
647
648 if let Some(account_id) = order.account_id() {
650 self.index
651 .account_orders
652 .entry(account_id)
653 .or_default()
654 .insert(*client_order_id);
655 }
656
657 if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
659 self.index
660 .exec_algorithm_orders
661 .entry(exec_algorithm_id)
662 .or_default()
663 .insert(*client_order_id);
664 self.index.exec_algorithms.insert(exec_algorithm_id);
665 }
666
667 if let Some(exec_spawn_id) = order.exec_spawn_id() {
669 self.index
670 .exec_spawn_orders
671 .entry(exec_spawn_id)
672 .or_default()
673 .insert(*client_order_id);
674 }
675
676 self.index.orders.insert(*client_order_id);
678
679 if order.is_active_local() {
681 self.index.orders_active_local.insert(*client_order_id);
682 }
683
684 if order.is_open() {
686 self.index.orders_open.insert(*client_order_id);
687 }
688
689 if order.is_closed() {
691 self.index.orders_closed.insert(*client_order_id);
692 }
693
694 if order.emulation_trigger().is_some() && !order.is_closed() {
696 self.index.orders_emulated.insert(*client_order_id);
697 }
698
699 if order.is_inflight() {
701 self.index.orders_inflight.insert(*client_order_id);
702 }
703
704 self.index.strategies.insert(strategy_id);
706 }
707
708 for (position_id, position_cell) in &self.positions {
710 let position = position_cell.borrow();
711 let instrument_id = position.instrument_id;
712 let venue = instrument_id.venue;
713 let strategy_id = position.strategy_id;
714
715 self.index
717 .venue_positions
718 .entry(venue)
719 .or_default()
720 .insert(*position_id);
721
722 self.index
724 .position_strategy
725 .insert(*position_id, strategy_id);
726
727 let position_orders = self.index.position_orders.entry(*position_id).or_default();
729 position_orders.extend(
730 position
731 .client_order_ids()
732 .into_iter()
733 .filter(|client_order_id| self.orders.contains_key(client_order_id)),
734 );
735
736 self.index
738 .instrument_positions
739 .entry(instrument_id)
740 .or_default()
741 .insert(*position_id);
742 self.index
743 .instrument_orders
744 .entry(instrument_id)
745 .or_default();
746
747 self.index
749 .strategy_positions
750 .entry(strategy_id)
751 .or_default()
752 .insert(*position_id);
753 self.index.strategy_orders.entry(strategy_id).or_default();
754
755 self.index
757 .account_positions
758 .entry(position.account_id)
759 .or_default()
760 .insert(*position_id);
761
762 self.index.positions.insert(*position_id);
764
765 if position.is_open() {
767 self.index.positions_open.insert(*position_id);
768 }
769
770 if position.is_closed() {
772 self.index.positions_closed.insert(*position_id);
773 }
774
775 self.index.strategies.insert(strategy_id);
777 }
778
779 self.index_position_oms();
780 }
781
782 fn index_position_oms(&mut self) {
783 self.index.position_oms.clear();
784
785 for (key, value) in &self.general {
786 let Some(position_id) = key.strip_prefix(POSITION_OMS_KEY_PREFIX) else {
787 continue;
788 };
789 let position_id = PositionId::new(position_id);
790 if !self.positions.contains_key(&position_id) {
791 continue;
792 }
793
794 match serde_json::from_slice::<OmsType>(value) {
795 Ok(oms_type) => {
796 self.index.position_oms.insert(position_id, oms_type);
797 }
798 Err(e) => {
799 log::error!("Failed to decode position OMS for {position_id}: {e}");
800 }
801 }
802 }
803
804 for position in self.positions.values().map(|cell| cell.borrow()) {
805 if !self.index.position_oms.contains_key(&position.id)
806 && position.id.as_str()
807 == format!("{}-{}", position.instrument_id, position.strategy_id)
808 {
809 self.index
810 .position_oms
811 .insert(position.id, OmsType::Netting);
812 }
813 }
814 }
815
816 #[must_use]
818 pub const fn has_backing(&self) -> bool {
819 self.database.is_some()
820 }
821
822 pub fn load_actor_state(
830 &self,
831 actor_id: &ActorId,
832 ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
833 self.database
834 .as_ref()
835 .map(|database| database.load_actor(actor_id))
836 .transpose()
837 .map(|state| state.map(Self::decode_component_state))
838 }
839
840 pub fn load_strategy_state(
848 &self,
849 strategy_id: &StrategyId,
850 ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
851 self.database
852 .as_ref()
853 .map(|database| database.load_strategy(strategy_id))
854 .transpose()
855 .map(|state| state.map(Self::decode_component_state))
856 }
857
858 pub fn update_actor_state(
864 &self,
865 actor_id: &ActorId,
866 state: &IndexMap<String, Vec<u8>>,
867 ) -> anyhow::Result<()> {
868 if let Some(database) = &self.database {
869 database.update_actor(actor_id, &Self::encode_component_state(state))?;
870 }
871 Ok(())
872 }
873
874 pub fn update_strategy_state(
880 &self,
881 strategy_id: &StrategyId,
882 state: &IndexMap<String, Vec<u8>>,
883 ) -> anyhow::Result<()> {
884 if let Some(database) = &self.database {
885 database.update_strategy(strategy_id, &Self::encode_component_state(state))?;
886 }
887 Ok(())
888 }
889
890 fn decode_component_state(state: AHashMap<String, Bytes>) -> IndexMap<String, Vec<u8>> {
891 state
892 .into_iter()
893 .map(|(key, value)| (key, value.to_vec()))
894 .collect()
895 }
896
897 fn encode_component_state(state: &IndexMap<String, Vec<u8>>) -> AHashMap<String, Bytes> {
898 state
899 .iter()
900 .map(|(key, value)| (key.clone(), Bytes::copy_from_slice(value)))
901 .collect()
902 }
903
904 #[must_use]
906 pub fn calculate_unrealized_pnl(&self, position: &Position) -> Option<Money> {
907 let Some(quote) = self.quote(&position.instrument_id) else {
908 log::warn!(
909 "Cannot calculate unrealized PnL for {}, no quotes for {}",
910 position.id,
911 position.instrument_id
912 );
913 return None;
914 };
915
916 let last = match position.side {
918 PositionSide::Flat => {
919 return Some(Money::zero(position.settlement_currency));
920 }
921 PositionSide::Long => quote.bid_price,
922 PositionSide::Short => quote.ask_price,
923 };
924
925 position
926 .try_unrealized_pnl(last)
927 .inspect_err(|e| {
928 log::error!("Cannot calculate unrealized PnL for {}: {e}", position.id);
929 })
930 .ok()
931 }
932
933 #[must_use]
942 pub fn check_integrity(&mut self) -> bool {
943 let mut error_count = 0;
944 let failure = "Integrity failure";
945
946 let timestamp_us = SystemTime::now()
948 .duration_since(UNIX_EPOCH)
949 .expect("Time went backwards")
950 .as_micros();
951
952 log::info!("Checking data integrity");
953
954 for account_id in self.accounts.keys() {
956 if !self
957 .index
958 .venue_account
959 .contains_key(&account_id.get_issuer())
960 {
961 log::error!(
962 "{failure} in accounts: {account_id} not found in `self.index.venue_account`",
963 );
964 error_count += 1;
965 }
966 }
967
968 for (client_order_id, order_cell) in &self.orders {
969 let order = order_cell.borrow();
970
971 if !self.index.order_strategy.contains_key(client_order_id) {
972 log::error!(
973 "{failure} in orders: {client_order_id} not found in `self.index.order_strategy`"
974 );
975 error_count += 1;
976 }
977
978 if !self.index.orders.contains(client_order_id) {
979 log::error!(
980 "{failure} in orders: {client_order_id} not found in `self.index.orders`",
981 );
982 error_count += 1;
983 }
984
985 if order.is_inflight() && !self.index.orders_inflight.contains(client_order_id) {
986 log::error!(
987 "{failure} in orders: {client_order_id} not found in `self.index.orders_inflight`",
988 );
989 error_count += 1;
990 }
991
992 if order.is_active_local() && !self.index.orders_active_local.contains(client_order_id)
993 {
994 log::error!(
995 "{failure} in orders: {client_order_id} not found in `self.index.orders_active_local`",
996 );
997 error_count += 1;
998 }
999
1000 if order.is_open() && !self.index.orders_open.contains(client_order_id) {
1001 log::error!(
1002 "{failure} in orders: {client_order_id} not found in `self.index.orders_open`",
1003 );
1004 error_count += 1;
1005 }
1006
1007 if order.is_closed() && !self.index.orders_closed.contains(client_order_id) {
1008 log::error!(
1009 "{failure} in orders: {client_order_id} not found in `self.index.orders_closed`",
1010 );
1011 error_count += 1;
1012 }
1013
1014 if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
1015 if !self
1016 .index
1017 .exec_algorithm_orders
1018 .contains_key(&exec_algorithm_id)
1019 {
1020 log::error!(
1021 "{failure} in orders: {client_order_id} not found in `self.index.exec_algorithm_orders`",
1022 );
1023 error_count += 1;
1024 }
1025
1026 if order.exec_spawn_id().is_none()
1027 && !self.index.exec_spawn_orders.contains_key(client_order_id)
1028 {
1029 log::error!(
1030 "{failure} in orders: {client_order_id} not found in `self.index.exec_spawn_orders`",
1031 );
1032 error_count += 1;
1033 }
1034 }
1035 }
1036
1037 for (position_id, position_cell) in &self.positions {
1038 let position = position_cell.borrow();
1039
1040 if !self.index.position_strategy.contains_key(position_id) {
1041 log::error!(
1042 "{failure} in positions: {position_id} not found in `self.index.position_strategy`",
1043 );
1044 error_count += 1;
1045 }
1046
1047 if !self.index.position_orders.contains_key(position_id) {
1048 log::error!(
1049 "{failure} in positions: {position_id} not found in `self.index.position_orders`",
1050 );
1051 error_count += 1;
1052 }
1053
1054 if !self.index.positions.contains(position_id) {
1055 log::error!(
1056 "{failure} in positions: {position_id} not found in `self.index.positions`",
1057 );
1058 error_count += 1;
1059 }
1060
1061 if position.is_open() && !self.index.positions_open.contains(position_id) {
1062 log::error!(
1063 "{failure} in positions: {position_id} not found in `self.index.positions_open`",
1064 );
1065 error_count += 1;
1066 }
1067
1068 if position.is_closed() && !self.index.positions_closed.contains(position_id) {
1069 log::error!(
1070 "{failure} in positions: {position_id} not found in `self.index.positions_closed`",
1071 );
1072 error_count += 1;
1073 }
1074 }
1075
1076 for account_id in self.index.venue_account.values() {
1078 if !self.accounts.contains_key(account_id) {
1079 log::error!(
1080 "{failure} in `index.venue_account`: {account_id} not found in `self.accounts`",
1081 );
1082 error_count += 1;
1083 }
1084 }
1085
1086 for client_order_id in self.index.venue_order_ids.values() {
1087 if !self.orders.contains_key(client_order_id) {
1088 log::error!(
1089 "{failure} in `index.venue_order_ids`: {client_order_id} not found in `self.orders`",
1090 );
1091 error_count += 1;
1092 }
1093 }
1094
1095 for client_order_id in self.index.client_order_ids.keys() {
1096 if !self.orders.contains_key(client_order_id) {
1097 log::error!(
1098 "{failure} in `index.client_order_ids`: {client_order_id} not found in `self.orders`",
1099 );
1100 error_count += 1;
1101 }
1102 }
1103
1104 for client_order_id in self.index.order_position.keys() {
1105 if !self.orders.contains_key(client_order_id) {
1106 log::error!(
1107 "{failure} in `index.order_position`: {client_order_id} not found in `self.orders`",
1108 );
1109 error_count += 1;
1110 }
1111 }
1112
1113 for client_order_id in self.index.order_strategy.keys() {
1115 if !self.orders.contains_key(client_order_id) {
1116 log::error!(
1117 "{failure} in `index.order_strategy`: {client_order_id} not found in `self.orders`",
1118 );
1119 error_count += 1;
1120 }
1121 }
1122
1123 for position_id in self.index.position_strategy.keys() {
1124 if !self.positions.contains_key(position_id) {
1125 log::error!(
1126 "{failure} in `index.position_strategy`: {position_id} not found in `self.positions`",
1127 );
1128 error_count += 1;
1129 }
1130 }
1131
1132 for position_id in self.index.position_orders.keys() {
1133 if !self.positions.contains_key(position_id) {
1134 log::error!(
1135 "{failure} in `index.position_orders`: {position_id} not found in `self.positions`",
1136 );
1137 error_count += 1;
1138 }
1139 }
1140
1141 for (instrument_id, client_order_ids) in &self.index.instrument_orders {
1142 for client_order_id in client_order_ids {
1143 if !self.orders.contains_key(client_order_id) {
1144 log::error!(
1145 "{failure} in `index.instrument_orders`: {instrument_id} not found in `self.orders`",
1146 );
1147 error_count += 1;
1148 }
1149 }
1150 }
1151
1152 for instrument_id in self.index.instrument_positions.keys() {
1153 if !self.index.instrument_orders.contains_key(instrument_id) {
1154 log::error!(
1155 "{failure} in `index.instrument_positions`: {instrument_id} not found in `index.instrument_orders`",
1156 );
1157 error_count += 1;
1158 }
1159 }
1160
1161 for client_order_ids in self.index.strategy_orders.values() {
1162 for client_order_id in client_order_ids {
1163 if !self.orders.contains_key(client_order_id) {
1164 log::error!(
1165 "{failure} in `index.strategy_orders`: {client_order_id} not found in `self.orders`",
1166 );
1167 error_count += 1;
1168 }
1169 }
1170 }
1171
1172 for position_ids in self.index.strategy_positions.values() {
1173 for position_id in position_ids {
1174 if !self.positions.contains_key(position_id) {
1175 log::error!(
1176 "{failure} in `index.strategy_positions`: {position_id} not found in `self.positions`",
1177 );
1178 error_count += 1;
1179 }
1180 }
1181 }
1182
1183 for client_order_id in &self.index.orders {
1184 if !self.orders.contains_key(client_order_id) {
1185 log::error!(
1186 "{failure} in `index.orders`: {client_order_id} not found in `self.orders`",
1187 );
1188 error_count += 1;
1189 }
1190 }
1191
1192 for client_order_id in &self.index.orders_emulated {
1193 if !self.orders.contains_key(client_order_id) {
1194 log::error!(
1195 "{failure} in `index.orders_emulated`: {client_order_id} not found in `self.orders`",
1196 );
1197 error_count += 1;
1198 }
1199 }
1200
1201 for client_order_id in &self.index.orders_active_local {
1202 if !self.orders.contains_key(client_order_id) {
1203 log::error!(
1204 "{failure} in `index.orders_active_local`: {client_order_id} not found in `self.orders`",
1205 );
1206 error_count += 1;
1207 }
1208 }
1209
1210 for client_order_id in &self.index.orders_inflight {
1211 if !self.orders.contains_key(client_order_id) {
1212 log::error!(
1213 "{failure} in `index.orders_inflight`: {client_order_id} not found in `self.orders`",
1214 );
1215 error_count += 1;
1216 }
1217 }
1218
1219 for client_order_id in &self.index.orders_open {
1220 if !self.orders.contains_key(client_order_id) {
1221 log::error!(
1222 "{failure} in `index.orders_open`: {client_order_id} not found in `self.orders`",
1223 );
1224 error_count += 1;
1225 }
1226 }
1227
1228 for client_order_id in &self.index.orders_closed {
1229 if !self.orders.contains_key(client_order_id) {
1230 log::error!(
1231 "{failure} in `index.orders_closed`: {client_order_id} not found in `self.orders`",
1232 );
1233 error_count += 1;
1234 }
1235 }
1236
1237 for position_id in &self.index.positions {
1238 if !self.positions.contains_key(position_id) {
1239 log::error!(
1240 "{failure} in `index.positions`: {position_id} not found in `self.positions`",
1241 );
1242 error_count += 1;
1243 }
1244 }
1245
1246 for position_id in &self.index.positions_open {
1247 if !self.positions.contains_key(position_id) {
1248 log::error!(
1249 "{failure} in `index.positions_open`: {position_id} not found in `self.positions`",
1250 );
1251 error_count += 1;
1252 }
1253 }
1254
1255 for position_id in &self.index.positions_closed {
1256 if !self.positions.contains_key(position_id) {
1257 log::error!(
1258 "{failure} in `index.positions_closed`: {position_id} not found in `self.positions`",
1259 );
1260 error_count += 1;
1261 }
1262 }
1263
1264 for strategy_id in &self.index.strategies {
1265 if !self.index.strategy_orders.contains_key(strategy_id) {
1266 log::error!(
1267 "{failure} in `index.strategies`: {strategy_id} not found in `index.strategy_orders`",
1268 );
1269 error_count += 1;
1270 }
1271 }
1272
1273 for exec_algorithm_id in &self.index.exec_algorithms {
1274 if !self
1275 .index
1276 .exec_algorithm_orders
1277 .contains_key(exec_algorithm_id)
1278 {
1279 log::error!(
1280 "{failure} in `index.exec_algorithms`: {exec_algorithm_id} not found in `index.exec_algorithm_orders`",
1281 );
1282 error_count += 1;
1283 }
1284 }
1285
1286 let total_us = SystemTime::now()
1287 .duration_since(UNIX_EPOCH)
1288 .expect("Time went backwards")
1289 .as_micros()
1290 - timestamp_us;
1291
1292 if error_count == 0 {
1293 log::info!("Integrity check passed in {total_us}μs");
1294 true
1295 } else {
1296 log::error!(
1297 "Integrity check failed with {error_count} error{} in {total_us}μs",
1298 if error_count == 1 { "" } else { "s" },
1299 );
1300 false
1301 }
1302 }
1303
1304 #[must_use]
1308 pub fn check_residuals(&self) -> bool {
1309 log::debug!("Checking residuals");
1310
1311 let mut residuals = false;
1312
1313 for order in self.orders_open(None, None, None, None, None) {
1315 residuals = true;
1316 log::warn!("Residual {order}");
1317 }
1318
1319 for position in self.positions_open(None, None, None, None, None) {
1321 residuals = true;
1322 log::warn!("Residual {position}");
1323 }
1324
1325 residuals
1326 }
1327
1328 pub fn purge_closed_orders(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
1334 log::debug!(
1335 "Purging closed orders{}",
1336 if buffer_secs > 0 {
1337 format!(" with buffer_secs={buffer_secs}")
1338 } else {
1339 String::new()
1340 }
1341 );
1342
1343 let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
1344 log::warn!(
1345 "Cannot purge closed orders: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
1346 );
1347 return;
1348 };
1349 let purge_cutoff = ts_now.checked_sub(buffer_ns);
1350
1351 let mut affected_order_list_ids: AHashSet<OrderListId> = AHashSet::new();
1352 let mut purged_client_order_ids: AHashSet<ClientOrderId> = AHashSet::new();
1353
1354 'outer: for client_order_id in self.index.orders_closed.clone() {
1355 let purge_target = self.orders.get(&client_order_id).and_then(|order_cell| {
1356 let order = order_cell.borrow();
1357 if order.is_closed()
1358 && let Some(ts_closed) = order.ts_closed()
1359 && purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
1360 {
1361 let linked = order.linked_order_ids().map(<[_]>::to_vec);
1362 let order_list_id = order.order_list_id();
1363 Some((linked, order_list_id))
1364 } else {
1365 None
1366 }
1367 });
1368
1369 let Some((linked, order_list_id)) = purge_target else {
1370 continue;
1371 };
1372
1373 if let Some(linked_order_ids) = linked {
1375 for linked_order_id in &linked_order_ids {
1376 if let Some(linked_order_cell) = self.orders.get(linked_order_id)
1377 && linked_order_cell.borrow().is_open()
1378 {
1379 continue 'outer;
1381 }
1382 }
1383 }
1384
1385 if let Some(order_list_id) = order_list_id {
1386 affected_order_list_ids.insert(order_list_id);
1387 }
1388
1389 if self.purge_order_except_aliases(client_order_id) {
1390 purged_client_order_ids.insert(client_order_id);
1391 }
1392 }
1393
1394 if !purged_client_order_ids.is_empty() {
1395 self.index
1396 .venue_order_ids
1397 .retain(|_, owner| !purged_client_order_ids.contains(owner));
1398 }
1399
1400 for order_list_id in affected_order_list_ids {
1401 if let Some(order_list) = self.order_lists.get(&order_list_id) {
1402 let all_purged = order_list
1403 .client_order_ids
1404 .iter()
1405 .all(|id| !self.orders.contains_key(id));
1406
1407 if all_purged {
1408 self.order_lists.remove(&order_list_id);
1409 log::info!("Purged {order_list_id}");
1410 }
1411 }
1412 }
1413 }
1414
1415 pub fn purge_closed_positions(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
1417 log::debug!(
1418 "Purging closed positions{}",
1419 if buffer_secs > 0 {
1420 format!(" with buffer_secs={buffer_secs}")
1421 } else {
1422 String::new()
1423 }
1424 );
1425
1426 let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
1427 log::warn!(
1428 "Cannot purge closed positions: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
1429 );
1430 return;
1431 };
1432 let purge_cutoff = ts_now.checked_sub(buffer_ns);
1433
1434 for position_id in self.index.positions_closed.clone() {
1435 let should_purge = self.positions.get(&position_id).is_some_and(|cell| {
1436 let position = cell.borrow();
1437 position.is_closed()
1438 && position.ts_closed.is_some_and(|ts_closed| {
1439 purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
1440 })
1441 });
1442
1443 if should_purge {
1444 self.purge_position(position_id);
1445 }
1446 }
1447 }
1448
1449 pub fn purge_order(&mut self, client_order_id: ClientOrderId) {
1453 if self.purge_order_except_aliases(client_order_id) {
1454 self.index
1455 .venue_order_ids
1456 .retain(|_, owner| owner != &client_order_id);
1457 }
1458 }
1459
1460 fn purge_order_except_aliases(&mut self, client_order_id: ClientOrderId) -> bool {
1465 struct OrderDetails {
1466 is_open: bool,
1467 instrument_id: InstrumentId,
1468 strategy_id: StrategyId,
1469 account_id: Option<AccountId>,
1470 exec_algorithm_id: Option<ExecAlgorithmId>,
1471 exec_spawn_id: Option<ClientOrderId>,
1472 position_id: Option<PositionId>,
1473 }
1474
1475 let order_cell = self.orders.get(&client_order_id).cloned();
1476 let order_details = order_cell.as_ref().map(|cell| {
1477 let order = cell.borrow();
1478 OrderDetails {
1479 is_open: order.is_open(),
1480 instrument_id: order.instrument_id(),
1481 strategy_id: order.strategy_id(),
1482 account_id: order.account_id(),
1483 exec_algorithm_id: order.exec_algorithm_id(),
1484 exec_spawn_id: order.exec_spawn_id(),
1485 position_id: order.position_id(),
1486 }
1487 });
1488
1489 if order_details
1490 .as_ref()
1491 .is_some_and(|details| details.is_open)
1492 {
1493 log::warn!("Order {client_order_id} found open when purging, skipping purge");
1494 return false;
1495 }
1496
1497 if order_details.is_some() {
1498 self.orders.remove(&client_order_id);
1499 } else {
1500 log::warn!("Order {client_order_id} not found when purging");
1501 }
1502
1503 let indexed_position_id = self.index.order_position.remove(&client_order_id);
1504 let indexed_strategy_id = self.index.order_strategy.remove(&client_order_id);
1505 self.index.order_client.remove(&client_order_id);
1506 self.index.client_order_ids.remove(&client_order_id);
1507
1508 if let Some(details) = &order_details {
1509 if let Some(venue_orders) = self
1510 .index
1511 .venue_orders
1512 .get_mut(&details.instrument_id.venue)
1513 {
1514 venue_orders.remove(&client_order_id);
1515 if venue_orders.is_empty() {
1516 self.index.venue_orders.remove(&details.instrument_id.venue);
1517 }
1518 }
1519
1520 let instrument_orders_became_empty = self
1525 .index
1526 .instrument_orders
1527 .get_mut(&details.instrument_id)
1528 .is_some_and(|instrument_orders| {
1529 instrument_orders.remove(&client_order_id);
1530 instrument_orders.is_empty()
1531 });
1532
1533 let has_instrument_positions = self
1534 .index
1535 .instrument_positions
1536 .get(&details.instrument_id)
1537 .is_some_and(|positions| !positions.is_empty());
1538
1539 if instrument_orders_became_empty && !has_instrument_positions {
1540 self.index.instrument_orders.remove(&details.instrument_id);
1541 }
1542
1543 if let Some(exec_algorithm_id) = details.exec_algorithm_id {
1544 let became_empty = self
1545 .index
1546 .exec_algorithm_orders
1547 .get_mut(&exec_algorithm_id)
1548 .is_some_and(|orders| {
1549 orders.remove(&client_order_id);
1550 orders.is_empty()
1551 });
1552
1553 if became_empty {
1554 self.index.exec_algorithm_orders.remove(&exec_algorithm_id);
1555 self.index.exec_algorithms.remove(&exec_algorithm_id);
1556 }
1557 }
1558
1559 if let Some(account_id) = details.account_id
1560 && let Some(account_orders) = self.index.account_orders.get_mut(&account_id)
1561 {
1562 account_orders.remove(&client_order_id);
1563 if account_orders.is_empty() {
1564 self.index.account_orders.remove(&account_id);
1565 }
1566 }
1567
1568 if let Some(exec_spawn_id) = details.exec_spawn_id
1569 && let Some(spawn_orders) = self.index.exec_spawn_orders.get_mut(&exec_spawn_id)
1570 {
1571 spawn_orders.remove(&client_order_id);
1572 if spawn_orders.is_empty() {
1573 self.index.exec_spawn_orders.remove(&exec_spawn_id);
1574 }
1575 }
1576 }
1577
1578 let mut position_ids = AHashSet::new();
1579 if let Some(position_id) = indexed_position_id {
1580 position_ids.insert(position_id);
1581 }
1582
1583 if let Some(position_id) = order_details
1584 .as_ref()
1585 .and_then(|details| details.position_id)
1586 {
1587 position_ids.insert(position_id);
1588 }
1589
1590 let mut strategy_ids = AHashSet::new();
1591 if let Some(strategy_id) = indexed_strategy_id {
1592 strategy_ids.insert(strategy_id);
1593 }
1594
1595 if let Some(details) = &order_details {
1596 strategy_ids.insert(details.strategy_id);
1597 }
1598
1599 for position_id in position_ids {
1600 if self.positions.contains_key(&position_id) {
1601 if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
1602 position_orders.remove(&client_order_id);
1603 }
1604 continue;
1605 }
1606
1607 let has_other_orders =
1608 if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
1609 position_orders.remove(&client_order_id);
1610 !position_orders.is_empty()
1611 } else {
1612 self.index
1613 .order_position
1614 .values()
1615 .any(|candidate| *candidate == position_id)
1616 };
1617
1618 if has_other_orders {
1619 continue;
1620 }
1621
1622 self.index.position_orders.remove(&position_id);
1623 if let Some(strategy_id) = self.index.position_strategy.remove(&position_id) {
1624 strategy_ids.insert(strategy_id);
1625 if let Some(strategy_positions) =
1626 self.index.strategy_positions.get_mut(&strategy_id)
1627 {
1628 strategy_positions.remove(&position_id);
1629 if strategy_positions.is_empty() {
1630 self.index.strategy_positions.remove(&strategy_id);
1631 }
1632 }
1633 }
1634
1635 if let Some(details) = &order_details
1636 && let Some(venue_positions) = self
1637 .index
1638 .venue_positions
1639 .get_mut(&details.instrument_id.venue)
1640 {
1641 venue_positions.remove(&position_id);
1642 if venue_positions.is_empty() {
1643 self.index
1644 .venue_positions
1645 .remove(&details.instrument_id.venue);
1646 }
1647 }
1648 }
1649
1650 for strategy_id in strategy_ids {
1651 let strategy_orders_became_empty = self
1657 .index
1658 .strategy_orders
1659 .get_mut(&strategy_id)
1660 .is_some_and(|strategy_orders| {
1661 strategy_orders.remove(&client_order_id);
1662 strategy_orders.is_empty()
1663 });
1664
1665 let has_positions = self
1666 .index
1667 .strategy_positions
1668 .get(&strategy_id)
1669 .is_some_and(|strategy_positions| !strategy_positions.is_empty());
1670
1671 if strategy_orders_became_empty && !has_positions {
1672 self.index.strategy_orders.remove(&strategy_id);
1673 self.index.strategies.remove(&strategy_id);
1674 }
1675 }
1676
1677 self.index.exec_spawn_orders.remove(&client_order_id);
1678
1679 self.index.orders.remove(&client_order_id);
1680 self.index.orders_active_local.remove(&client_order_id);
1681 self.index.orders_open.remove(&client_order_id);
1682 self.index.orders_closed.remove(&client_order_id);
1683 self.index.orders_emulated.remove(&client_order_id);
1684 self.index.orders_inflight.remove(&client_order_id);
1685 self.index.orders_pending_cancel.remove(&client_order_id);
1686
1687 if order_details.is_some() {
1688 log::info!("Purged order {client_order_id}");
1689 }
1690
1691 true
1692 }
1693
1694 pub fn purge_position(&mut self, position_id: PositionId) {
1698 let position = self
1700 .positions
1701 .get(&position_id)
1702 .map(|cell| cell.borrow().clone());
1703
1704 if let Some(ref pos) = position
1706 && pos.is_open()
1707 {
1708 log::warn!("Position {position_id} found open when purging, skipping purge");
1709 return;
1710 }
1711
1712 if let Some(ref pos) = position {
1714 self.positions.remove(&position_id);
1715
1716 if let Some(venue_positions) =
1718 self.index.venue_positions.get_mut(&pos.instrument_id.venue)
1719 {
1720 venue_positions.remove(&position_id);
1721 if venue_positions.is_empty() {
1722 self.index.venue_positions.remove(&pos.instrument_id.venue);
1723 }
1724 }
1725
1726 let instrument_positions_became_empty = self
1728 .index
1729 .instrument_positions
1730 .get_mut(&pos.instrument_id)
1731 .is_some_and(|positions| {
1732 positions.remove(&position_id);
1733 positions.is_empty()
1734 });
1735
1736 if instrument_positions_became_empty {
1737 self.index.instrument_positions.remove(&pos.instrument_id);
1738 let instrument_orders_empty = self
1739 .index
1740 .instrument_orders
1741 .get(&pos.instrument_id)
1742 .is_some_and(|orders| orders.is_empty());
1743
1744 if instrument_orders_empty {
1745 self.index.instrument_orders.remove(&pos.instrument_id);
1746 }
1747 }
1748
1749 let strategy_positions_became_empty = self
1751 .index
1752 .strategy_positions
1753 .get_mut(&pos.strategy_id)
1754 .is_some_and(|positions| {
1755 positions.remove(&position_id);
1756 positions.is_empty()
1757 });
1758
1759 if strategy_positions_became_empty {
1760 self.index.strategy_positions.remove(&pos.strategy_id);
1761 let strategy_orders_empty = self
1762 .index
1763 .strategy_orders
1764 .get(&pos.strategy_id)
1765 .is_some_and(|orders| orders.is_empty());
1766
1767 if strategy_orders_empty {
1768 self.index.strategy_orders.remove(&pos.strategy_id);
1769 self.index.strategies.remove(&pos.strategy_id);
1770 }
1771 }
1772
1773 if let Some(account_positions) = self.index.account_positions.get_mut(&pos.account_id) {
1775 account_positions.remove(&position_id);
1776 if account_positions.is_empty() {
1777 self.index.account_positions.remove(&pos.account_id);
1778 }
1779 }
1780
1781 for client_order_id in pos.client_order_ids() {
1783 self.index.order_position.remove(&client_order_id);
1784 }
1785
1786 log::info!("Purged position {position_id}");
1787 } else {
1788 log::warn!("Position {position_id} not found when purging");
1789 }
1790
1791 self.index.position_strategy.remove(&position_id);
1793 self.index.position_oms.remove(&position_id);
1794 self.index.position_orders.remove(&position_id);
1795 self.index.positions.remove(&position_id);
1796 self.index.positions_open.remove(&position_id);
1797 self.index.positions_closed.remove(&position_id);
1798
1799 self.position_snapshots.remove(&position_id);
1801 self.bump_position_snapshot_revision(position_id);
1802 }
1803
1804 fn purge_instrument_inner(&mut self, instrument_id: InstrumentId, skip_order_guard: bool) {
1828 #[cfg(feature = "defi")]
1829 let defi_found = self.defi.pools.contains_key(&instrument_id)
1830 || self.defi.pool_profilers.contains_key(&instrument_id);
1831 #[cfg(not(feature = "defi"))]
1832 let defi_found = false;
1833
1834 let found = self.instruments.contains_key(&instrument_id)
1835 || self.synthetics.contains_key(&instrument_id)
1836 || defi_found;
1837
1838 if !found {
1839 log::warn!("Instrument {instrument_id} not found when purging");
1840 return;
1841 }
1842
1843 if !skip_order_guard && let Some(orders) = self.index.instrument_orders.get(&instrument_id)
1844 {
1845 let has_non_terminal = orders
1846 .iter()
1847 .any(|client_order_id| !self.index.orders_closed.contains(client_order_id));
1848
1849 if has_non_terminal {
1850 log::warn!(
1851 "Instrument {instrument_id} has non-terminal orders when purging, skipping purge"
1852 );
1853 return;
1854 }
1855 }
1856
1857 if let Some(positions) = self.index.instrument_positions.get(&instrument_id) {
1858 let has_non_closed = positions
1859 .iter()
1860 .any(|position_id| !self.index.positions_closed.contains(position_id));
1861
1862 if has_non_closed {
1863 log::warn!(
1864 "Instrument {instrument_id} has non-closed positions when purging, skipping purge"
1865 );
1866 return;
1867 }
1868 }
1869
1870 self.instruments.remove(&instrument_id);
1871 self.synthetics.remove(&instrument_id);
1872 self.books.remove(&instrument_id);
1873 self.own_books.remove(&instrument_id);
1874 self.quotes.remove(&instrument_id);
1875 self.trades.remove(&instrument_id);
1876 self.mark_prices.remove(&instrument_id);
1877 self.index_prices.remove(&instrument_id);
1878 self.funding_rates.remove(&instrument_id);
1879 self.instrument_statuses.remove(&instrument_id);
1880 self.instrument_closes.remove(&instrument_id);
1881 self.greeks.remove(&instrument_id);
1882 self.option_greeks.remove(&instrument_id);
1883
1884 self.bars
1885 .retain(|bar_type, _| bar_type.instrument_id() != instrument_id);
1886
1887 #[cfg(feature = "defi")]
1888 {
1889 self.defi.pools.remove(&instrument_id);
1890 self.defi.pool_profilers.remove(&instrument_id);
1891 }
1892
1893 self.index.instrument_orders.remove(&instrument_id);
1894 self.index.instrument_positions.remove(&instrument_id);
1895
1896 log::info!("Purged instrument {instrument_id}");
1897 }
1898
1899 pub fn purge_instrument(&mut self, instrument_id: InstrumentId) {
1904 self.purge_instrument_inner(instrument_id, false);
1905 }
1906
1907 pub fn purge_instrument_skip_order_guard(&mut self, instrument_id: InstrumentId) {
1916 self.purge_instrument_inner(instrument_id, true);
1917 }
1918
1919 pub fn purge_account_events(&mut self, ts_now: UnixNanos, lookback_secs: u64) {
1924 log::debug!(
1925 "Purging account events{}",
1926 if lookback_secs > 0 {
1927 format!(" with lookback_secs={lookback_secs}")
1928 } else {
1929 String::new()
1930 }
1931 );
1932
1933 for account_cell in self.accounts.values() {
1934 let mut account = account_cell.borrow_mut();
1935 let event_count = account.event_count();
1936 account.purge_account_events(ts_now, lookback_secs);
1937 let count_diff = event_count - account.event_count();
1938 if count_diff > 0 {
1939 log::info!(
1940 "Purged {} event(s) from account {}",
1941 count_diff,
1942 account.id()
1943 );
1944 }
1945 }
1946 }
1947
1948 pub fn clear_index(&mut self) {
1950 self.index.clear();
1951 log::debug!("Cleared index");
1952 }
1953
1954 pub fn reset(&mut self) {
1961 log::debug!("Resetting cache");
1962
1963 self.general.clear();
1964 self.books.clear();
1965 self.own_books.clear();
1966 self.quotes.clear();
1967 self.trades.clear();
1968 self.mark_xrates.clear();
1969 self.mark_prices.clear();
1970 self.index_prices.clear();
1971 self.funding_rates.clear();
1972 self.instrument_statuses.clear();
1973 self.instrument_closes.clear();
1974 self.bars.clear();
1975 self.accounts.clear();
1976 self.orders.clear();
1977 self.order_lists.clear();
1978 self.positions.clear();
1979 self.position_snapshots.clear();
1980 self.position_snapshot_revisions.clear();
1981 self.greeks.clear();
1982 self.option_greeks.clear();
1983 self.yield_curves.clear();
1984
1985 if self.config.drop_instruments_on_reset {
1986 self.currencies.clear();
1987 self.instruments.clear();
1988 self.synthetics.clear();
1989 }
1990
1991 #[cfg(feature = "defi")]
1992 {
1993 self.defi.pools.clear();
1994 self.defi.pool_profilers.clear();
1995 }
1996
1997 self.clear_index();
1998
1999 log::info!("Reset cache");
2000 }
2001
2002 pub fn dispose(&mut self) {
2006 self.reset();
2007
2008 if let Some(database) = &mut self.database
2009 && let Err(e) = database.close()
2010 {
2011 log::error!("Failed to close database during dispose: {e}");
2012 }
2013 }
2014
2015 pub fn flush_db(&mut self) {
2019 if let Some(database) = &mut self.database
2020 && let Err(e) = database.flush()
2021 {
2022 log::error!("Failed to flush database: {e}");
2023 }
2024 }
2025
2026 pub fn add(&mut self, key: &str, value: Bytes) -> anyhow::Result<()> {
2034 check_valid_string_ascii(key, stringify!(key))?;
2035 check_predicate_false(value.is_empty(), stringify!(value))?;
2036
2037 log::debug!("Adding general {key}");
2038 self.general.insert(key.to_string(), value.clone());
2039
2040 if let Some(database) = &mut self.database {
2041 database.add(key.to_string(), value)?;
2042 }
2043 Ok(())
2044 }
2045
2046 pub fn add_order_book(&mut self, book: OrderBook) -> anyhow::Result<()> {
2052 log::debug!("Adding `OrderBook` {}", book.instrument_id);
2053
2054 if self.config.save_market_data
2055 && let Some(database) = &mut self.database
2056 {
2057 database.add_order_book(&book)?;
2058 }
2059
2060 self.books.insert(book.instrument_id, book);
2061 Ok(())
2062 }
2063
2064 pub fn add_own_order_book(&mut self, own_book: OwnOrderBook) -> anyhow::Result<()> {
2070 log::debug!("Adding `OwnOrderBook` {}", own_book.instrument_id);
2071
2072 self.own_books.insert(own_book.instrument_id, own_book);
2073 Ok(())
2074 }
2075
2076 pub fn add_mark_price(&mut self, mark_price: MarkPriceUpdate) -> anyhow::Result<()> {
2082 log::debug!("Adding `MarkPriceUpdate` for {}", mark_price.instrument_id);
2083
2084 if self.config.save_market_data {
2085 }
2087
2088 let mark_prices_deque = self
2089 .mark_prices
2090 .entry(mark_price.instrument_id)
2091 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2092 mark_prices_deque.push_front(mark_price);
2093 Ok(())
2094 }
2095
2096 pub fn add_index_price(&mut self, index_price: IndexPriceUpdate) -> anyhow::Result<()> {
2102 log::debug!(
2103 "Adding `IndexPriceUpdate` for {}",
2104 index_price.instrument_id
2105 );
2106
2107 if self.config.save_market_data {
2108 }
2110
2111 let index_prices_deque = self
2112 .index_prices
2113 .entry(index_price.instrument_id)
2114 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2115 index_prices_deque.push_front(index_price);
2116 Ok(())
2117 }
2118
2119 pub fn add_funding_rate(&mut self, funding_rate: FundingRateUpdate) -> anyhow::Result<()> {
2125 log::debug!(
2126 "Adding `FundingRateUpdate` for {}",
2127 funding_rate.instrument_id
2128 );
2129
2130 if self.config.save_market_data {
2131 }
2133
2134 let funding_rates_deque = self
2135 .funding_rates
2136 .entry(funding_rate.instrument_id)
2137 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2138 funding_rates_deque.push_front(funding_rate);
2139 Ok(())
2140 }
2141
2142 pub fn add_funding_rates(&mut self, funding_rates: &[FundingRateUpdate]) -> anyhow::Result<()> {
2148 check_slice_not_empty(funding_rates, stringify!(funding_rates))?;
2149
2150 let instrument_id = funding_rates[0].instrument_id;
2151 log::debug!(
2152 "Adding `FundingRateUpdate`[{}] {instrument_id}",
2153 funding_rates.len()
2154 );
2155
2156 if self.config.save_market_data
2157 && let Some(database) = &mut self.database
2158 {
2159 for funding_rate in funding_rates {
2160 database.add_funding_rate(funding_rate)?;
2161 }
2162 }
2163
2164 let funding_rate_deque = self
2165 .funding_rates
2166 .entry(instrument_id)
2167 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2168
2169 for funding_rate in funding_rates {
2170 funding_rate_deque.push_front(*funding_rate);
2171 }
2172 Ok(())
2173 }
2174
2175 pub fn add_instrument_status(&mut self, status: InstrumentStatus) -> anyhow::Result<()> {
2181 log::debug!("Adding `InstrumentStatus` for {}", status.instrument_id);
2182
2183 if self.config.save_market_data {
2184 }
2186
2187 let statuses_deque = self
2188 .instrument_statuses
2189 .entry(status.instrument_id)
2190 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2191 statuses_deque.push_front(status);
2192 Ok(())
2193 }
2194
2195 pub fn add_instrument_close(&mut self, close: InstrumentClose) -> anyhow::Result<()> {
2205 log::debug!("Adding `InstrumentClose` for {}", close.instrument_id);
2206
2207 if let Some(database) = &self.database {
2208 database.add_instrument_close(&close)?;
2209 }
2210
2211 self.instrument_closes.insert(close.instrument_id, close);
2212 Ok(())
2213 }
2214
2215 pub fn add_quote(&mut self, quote: QuoteTick) -> anyhow::Result<()> {
2221 log::debug!("Adding `QuoteTick` {}", quote.instrument_id);
2222
2223 if self.config.save_market_data
2224 && let Some(database) = &mut self.database
2225 {
2226 database.add_quote("e)?;
2227 }
2228
2229 let quotes_deque = self
2230 .quotes
2231 .entry(quote.instrument_id)
2232 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2233 quotes_deque.push_front(quote);
2234 Ok(())
2235 }
2236
2237 pub fn add_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
2243 check_slice_not_empty(quotes, stringify!(quotes))?;
2244
2245 let instrument_id = quotes[0].instrument_id;
2246 log::debug!("Adding `QuoteTick`[{}] {instrument_id}", quotes.len());
2247
2248 if self.config.save_market_data
2249 && let Some(database) = &mut self.database
2250 {
2251 for quote in quotes {
2252 database.add_quote(quote)?;
2253 }
2254 }
2255
2256 let quotes_deque = self
2257 .quotes
2258 .entry(instrument_id)
2259 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2260
2261 for quote in quotes {
2262 quotes_deque.push_front(*quote);
2263 }
2264 Ok(())
2265 }
2266
2267 pub fn add_trade(&mut self, trade: TradeTick) -> anyhow::Result<()> {
2273 log::debug!("Adding `TradeTick` {}", trade.instrument_id);
2274
2275 if self.config.save_market_data
2276 && let Some(database) = &mut self.database
2277 {
2278 database.add_trade(&trade)?;
2279 }
2280
2281 let trades_deque = self
2282 .trades
2283 .entry(trade.instrument_id)
2284 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2285 trades_deque.push_front(trade);
2286 Ok(())
2287 }
2288
2289 pub fn add_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
2295 check_slice_not_empty(trades, stringify!(trades))?;
2296
2297 let instrument_id = trades[0].instrument_id;
2298 log::debug!("Adding `TradeTick`[{}] {instrument_id}", trades.len());
2299
2300 if self.config.save_market_data
2301 && let Some(database) = &mut self.database
2302 {
2303 for trade in trades {
2304 database.add_trade(trade)?;
2305 }
2306 }
2307
2308 let trades_deque = self
2309 .trades
2310 .entry(instrument_id)
2311 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2312
2313 for trade in trades {
2314 trades_deque.push_front(*trade);
2315 }
2316 Ok(())
2317 }
2318
2319 pub fn add_bar(&mut self, bar: Bar) -> anyhow::Result<()> {
2325 log::debug!("Adding `Bar` {}", bar.bar_type);
2326
2327 if self.config.save_market_data
2328 && let Some(database) = &mut self.database
2329 {
2330 database.add_bar(&bar)?;
2331 }
2332
2333 let bars = self
2334 .bars
2335 .entry(bar.bar_type)
2336 .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
2337 bars.push_front(bar);
2338 Ok(())
2339 }
2340
2341 pub fn add_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
2347 check_slice_not_empty(bars, stringify!(bars))?;
2348
2349 let bar_type = bars[0].bar_type;
2350 log::debug!("Adding `Bar`[{}] {bar_type}", bars.len());
2351
2352 if self.config.save_market_data
2353 && let Some(database) = &mut self.database
2354 {
2355 for bar in bars {
2356 database.add_bar(bar)?;
2357 }
2358 }
2359
2360 let bars_deque = self
2361 .bars
2362 .entry(bar_type)
2363 .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
2364
2365 for bar in bars {
2366 bars_deque.push_front(*bar);
2367 }
2368 Ok(())
2369 }
2370
2371 pub fn add_greeks(&mut self, greeks: GreeksData) -> anyhow::Result<()> {
2377 log::debug!("Adding `GreeksData` {}", greeks.instrument_id);
2378
2379 if self.config.save_market_data
2380 && let Some(_database) = &mut self.database
2381 {
2382 }
2384
2385 self.greeks.insert(greeks.instrument_id, greeks);
2386 Ok(())
2387 }
2388
2389 pub fn greeks(&self, instrument_id: &InstrumentId) -> Option<GreeksData> {
2391 self.greeks.get(instrument_id).cloned()
2392 }
2393
2394 pub fn add_option_greeks(&mut self, greeks: OptionGreeks) {
2396 log::debug!("Adding `OptionGreeks` {}", greeks.instrument_id);
2397 self.option_greeks.insert(greeks.instrument_id, greeks);
2398 }
2399
2400 #[must_use]
2402 pub fn option_greeks(&self, instrument_id: &InstrumentId) -> Option<&OptionGreeks> {
2403 self.option_greeks.get(instrument_id)
2404 }
2405
2406 pub fn add_yield_curve(&mut self, yield_curve: YieldCurveData) -> anyhow::Result<()> {
2412 log::debug!("Adding `YieldCurveData` {}", yield_curve.curve_name);
2413
2414 if self.config.save_market_data
2415 && let Some(_database) = &mut self.database
2416 {
2417 }
2419
2420 self.yield_curves
2421 .insert(yield_curve.curve_name.clone(), yield_curve);
2422 Ok(())
2423 }
2424
2425 pub fn yield_curve(&self, key: &str) -> Option<Box<dyn Fn(f64) -> f64>> {
2427 self.yield_curves.get(key).map(|curve| {
2428 let curve_clone = curve.clone();
2429 Box::new(move |expiry_in_years: f64| curve_clone.get_rate(expiry_in_years))
2430 as Box<dyn Fn(f64) -> f64>
2431 })
2432 }
2433
2434 pub fn add_currency(&mut self, currency: Currency) -> anyhow::Result<()> {
2440 if self.currencies.contains_key(¤cy.code) {
2441 return Ok(());
2442 }
2443 log::debug!("Adding `Currency` {}", currency.code);
2444
2445 if let Some(database) = &mut self.database {
2446 database.add_currency(¤cy)?;
2447 }
2448
2449 self.currencies.insert(currency.code, currency);
2450 Ok(())
2451 }
2452
2453 pub fn add_instrument(&mut self, instrument: InstrumentAny) -> anyhow::Result<()> {
2459 log::debug!("Adding `Instrument` {}", instrument.id());
2460
2461 if let Some(base_currency) = instrument.base_currency() {
2463 self.add_currency(base_currency)?;
2464 }
2465 self.add_currency(instrument.quote_currency())?;
2466 self.add_currency(instrument.settlement_currency())?;
2467
2468 if let Some(database) = &mut self.database {
2469 database.add_instrument(&instrument)?;
2470 }
2471
2472 self.instruments.insert(instrument.id(), instrument);
2473 Ok(())
2474 }
2475
2476 pub fn add_synthetic(&mut self, synthetic: SyntheticInstrument) -> anyhow::Result<()> {
2482 log::debug!("Adding `SyntheticInstrument` {}", synthetic.id);
2483
2484 if let Some(database) = &mut self.database {
2485 database.add_synthetic(&synthetic)?;
2486 }
2487
2488 self.synthetics.insert(synthetic.id, synthetic);
2489 Ok(())
2490 }
2491
2492 pub fn add_account(&mut self, account: AccountAny) -> anyhow::Result<()> {
2498 log::debug!("Adding `Account` {}", account.id());
2499
2500 if let Some(database) = &mut self.database {
2501 database.add_account(&account)?;
2502 }
2503
2504 let account_id = account.id();
2505 self.accounts.insert(account_id, SharedCell::new(account));
2506 self.index
2507 .venue_account
2508 .insert(account_id.get_issuer(), account_id);
2509 Ok(())
2510 }
2511
2512 pub fn add_venue_order_id(
2521 &mut self,
2522 client_order_id: &ClientOrderId,
2523 venue_order_id: &VenueOrderId,
2524 overwrite: bool,
2525 ) -> anyhow::Result<()> {
2526 self.validate_venue_order_id_claim(client_order_id, venue_order_id, overwrite)?;
2527
2528 self.index
2529 .client_order_ids
2530 .insert(*client_order_id, *venue_order_id);
2531 self.index
2532 .venue_order_ids
2533 .insert(*venue_order_id, *client_order_id);
2534
2535 Ok(())
2536 }
2537
2538 pub fn index_venue_order_id(
2549 &mut self,
2550 client_order_id: &ClientOrderId,
2551 venue_order_id: &VenueOrderId,
2552 ) -> anyhow::Result<()> {
2553 self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
2554
2555 self.index
2556 .venue_order_ids
2557 .insert(*venue_order_id, *client_order_id);
2558 self.index
2559 .client_order_ids
2560 .entry(*client_order_id)
2561 .or_insert(*venue_order_id);
2562
2563 Ok(())
2564 }
2565
2566 fn validate_venue_order_id_claim(
2567 &self,
2568 client_order_id: &ClientOrderId,
2569 venue_order_id: &VenueOrderId,
2570 overwrite: bool,
2571 ) -> anyhow::Result<()> {
2572 self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
2573
2574 if let Some(existing_venue_order_id) = self.index.client_order_ids.get(client_order_id)
2575 && !overwrite
2576 && existing_venue_order_id != venue_order_id
2577 {
2578 anyhow::bail!(
2579 "Existing {existing_venue_order_id} for {client_order_id}
2580 did not match the given {venue_order_id}.
2581 If you are writing a test then try a different `venue_order_id`,
2582 otherwise this is probably a bug."
2583 );
2584 }
2585
2586 Ok(())
2587 }
2588
2589 fn validate_venue_order_id_ownership(
2590 &self,
2591 client_order_id: &ClientOrderId,
2592 venue_order_id: &VenueOrderId,
2593 ) -> anyhow::Result<()> {
2594 if let Some(existing_client_order_id) = self.index.venue_order_ids.get(venue_order_id)
2595 && existing_client_order_id != client_order_id
2596 {
2597 return Err(VenueOrderIdOwnershipError {
2598 venue_order_id: *venue_order_id,
2599 existing_client_order_id: *existing_client_order_id,
2600 claimant_client_order_id: *client_order_id,
2601 }
2602 .into());
2603 }
2604
2605 Ok(())
2606 }
2607
2608 pub fn add_order(
2623 &mut self,
2624 order: OrderAny,
2625 position_id: Option<PositionId>,
2626 client_id: Option<ClientId>,
2627 replace_existing: bool,
2628 ) -> anyhow::Result<()> {
2629 let instrument_id = order.instrument_id();
2630 let venue = instrument_id.venue;
2631 let client_order_id = order.client_order_id();
2632 let strategy_id = order.strategy_id();
2633 let exec_algorithm_id = order.exec_algorithm_id();
2634 let exec_spawn_id = order.exec_spawn_id();
2635
2636 if !replace_existing {
2637 check_key_not_in_map(
2638 &client_order_id,
2639 &self.orders,
2640 stringify!(client_order_id),
2641 stringify!(orders),
2642 )?;
2643 }
2644
2645 log::debug!("Adding {order:?}");
2646
2647 self.index.orders.insert(client_order_id);
2648
2649 if order.is_active_local() {
2650 self.index.orders_active_local.insert(client_order_id);
2651 }
2652 self.index
2653 .order_strategy
2654 .insert(client_order_id, strategy_id);
2655 self.index.strategies.insert(strategy_id);
2656
2657 self.index
2659 .venue_orders
2660 .entry(venue)
2661 .or_default()
2662 .insert(client_order_id);
2663
2664 self.index
2666 .instrument_orders
2667 .entry(instrument_id)
2668 .or_default()
2669 .insert(client_order_id);
2670
2671 self.index
2673 .strategy_orders
2674 .entry(strategy_id)
2675 .or_default()
2676 .insert(client_order_id);
2677
2678 if let Some(account_id) = order.account_id() {
2680 self.index
2681 .account_orders
2682 .entry(account_id)
2683 .or_default()
2684 .insert(client_order_id);
2685 }
2686
2687 if let Some(exec_algorithm_id) = exec_algorithm_id {
2689 self.index.exec_algorithms.insert(exec_algorithm_id);
2690
2691 self.index
2692 .exec_algorithm_orders
2693 .entry(exec_algorithm_id)
2694 .or_default()
2695 .insert(client_order_id);
2696 }
2697
2698 if let Some(exec_spawn_id) = exec_spawn_id {
2700 self.index
2701 .exec_spawn_orders
2702 .entry(exec_spawn_id)
2703 .or_default()
2704 .insert(client_order_id);
2705 }
2706
2707 if order.emulation_trigger().is_some() {
2709 self.index.orders_emulated.insert(client_order_id);
2710 }
2711
2712 if let Some(position_id) = position_id {
2714 self.index_position_id_in_memory(&position_id, &venue, &client_order_id, &strategy_id);
2715 }
2716
2717 if let Some(client_id) = client_id {
2719 self.index.order_client.insert(client_order_id, client_id);
2720 log::debug!("Indexed {client_id:?}");
2721 }
2722
2723 let order_cell = if let Some(order_cell) = self.orders.get(&client_order_id) {
2726 *order_cell.borrow_mut() = order;
2727 order_cell.clone()
2728 } else {
2729 let order_cell = SharedCell::new(order);
2730 self.orders.insert(client_order_id, order_cell.clone());
2731 order_cell
2732 };
2733
2734 if let Some(position_id) = position_id {
2735 self.persist_position_id(&position_id, &client_order_id)?;
2736 }
2737
2738 if let Some(database) = &mut self.database {
2739 database.add_order(&order_cell.borrow(), client_id)?;
2740 }
2745
2746 Ok(())
2747 }
2748
2749 pub fn claim_order_clients(
2761 &mut self,
2762 claims: &[(ClientOrderId, ClientId)],
2763 ) -> anyhow::Result<()> {
2764 let mut requested = AHashMap::with_capacity(claims.len());
2765 let mut ordered_claims = Vec::with_capacity(claims.len());
2766
2767 for (client_order_id, client_id) in claims {
2768 if let Some(existing_client_id) = requested.get(client_order_id) {
2769 if existing_client_id != client_id {
2770 anyhow::bail!(
2771 "Conflicting execution client claims for {client_order_id}: \
2772 {existing_client_id} and {client_id}"
2773 );
2774 }
2775 continue;
2776 }
2777
2778 requested.insert(*client_order_id, *client_id);
2779 ordered_claims.push((*client_order_id, *client_id));
2780 }
2781
2782 let mut pending_claims = Vec::with_capacity(ordered_claims.len());
2783 for (client_order_id, client_id) in ordered_claims {
2784 if !self.orders.contains_key(&client_order_id) {
2785 return Err(OrderLookupError::not_found(client_order_id).into());
2786 }
2787
2788 match self.index.order_client.get(&client_order_id) {
2789 Some(existing_client_id) if *existing_client_id == client_id => {}
2790 Some(existing_client_id) => {
2791 anyhow::bail!(
2792 "Order {client_order_id} is already claimed by execution client \
2793 {existing_client_id} and cannot be claimed by {client_id}"
2794 );
2795 }
2796 None => pending_claims.push((client_order_id, client_id)),
2797 }
2798 }
2799
2800 if pending_claims.is_empty() {
2801 return Ok(());
2802 }
2803
2804 if let Some(database) = &self.database {
2805 database.index_order_clients(&pending_claims)?;
2806 }
2807
2808 for (client_order_id, client_id) in pending_claims {
2809 self.index.order_client.insert(client_order_id, client_id);
2810 log::debug!("Claimed {client_order_id} for execution client {client_id}");
2811 }
2812
2813 Ok(())
2814 }
2815
2816 pub fn add_order_list(&mut self, order_list: OrderList) -> anyhow::Result<()> {
2822 let order_list_id = order_list.id;
2823 check_key_not_in_map(
2824 &order_list_id,
2825 &self.order_lists,
2826 stringify!(order_list_id),
2827 stringify!(order_lists),
2828 )?;
2829
2830 log::debug!("Adding {order_list}");
2831 self.order_lists.insert(order_list_id, order_list);
2832 Ok(())
2833 }
2834
2835 pub fn add_position_id(
2846 &mut self,
2847 position_id: &PositionId,
2848 venue: &Venue,
2849 client_order_id: &ClientOrderId,
2850 strategy_id: &StrategyId,
2851 ) -> anyhow::Result<()> {
2852 self.index_position_id_in_memory(position_id, venue, client_order_id, strategy_id);
2853 self.persist_position_id(position_id, client_order_id)
2854 }
2855
2856 fn index_position_id_in_memory(
2857 &mut self,
2858 position_id: &PositionId,
2859 venue: &Venue,
2860 client_order_id: &ClientOrderId,
2861 strategy_id: &StrategyId,
2862 ) {
2863 self.index
2864 .order_position
2865 .insert(*client_order_id, *position_id);
2866 self.index_position(position_id, venue, strategy_id);
2867 self.index
2868 .position_orders
2869 .entry(*position_id)
2870 .or_default()
2871 .insert(*client_order_id);
2872 }
2873
2874 fn persist_position_id(
2875 &mut self,
2876 position_id: &PositionId,
2877 client_order_id: &ClientOrderId,
2878 ) -> anyhow::Result<()> {
2879 if let Some(database) = &mut self.database {
2880 database.index_order_position(*client_order_id, *position_id)?;
2881 }
2882
2883 Ok(())
2884 }
2885
2886 fn index_position(
2887 &mut self,
2888 position_id: &PositionId,
2889 venue: &Venue,
2890 strategy_id: &StrategyId,
2891 ) {
2892 let strategy_id = self
2893 .positions
2894 .get(position_id)
2895 .map(|position| position.borrow().strategy_id)
2896 .filter(StrategyId::is_external)
2897 .unwrap_or(*strategy_id);
2898
2899 self.index
2901 .position_strategy
2902 .insert(*position_id, strategy_id);
2903
2904 self.index.position_orders.entry(*position_id).or_default();
2906
2907 self.index
2909 .strategy_positions
2910 .entry(strategy_id)
2911 .or_default()
2912 .insert(*position_id);
2913
2914 self.index
2916 .venue_positions
2917 .entry(*venue)
2918 .or_default()
2919 .insert(*position_id);
2920 }
2921
2922 fn assign_position_ids_to_contingencies(&mut self) {
2930 let mut assignments: Vec<(PositionId, ClientOrderId)> = Vec::new();
2931
2932 for parent_order_cell in self.orders.values() {
2933 let parent = parent_order_cell.borrow();
2934 if parent.contingency_type() != Some(ContingencyType::Oto) {
2935 continue;
2936 }
2937 let Some(parent_position_id) = parent.position_id() else {
2938 continue;
2939 };
2940 let Some(linked_order_ids) = parent.linked_order_ids() else {
2941 continue;
2942 };
2943
2944 for client_order_id in linked_order_ids {
2945 match self.orders.get(client_order_id) {
2946 None => {
2947 log::error!("Contingency order {client_order_id} not found");
2948 }
2949 Some(contingent_order_cell) => {
2950 if contingent_order_cell.borrow().position_id().is_none() {
2951 assignments.push((parent_position_id, *client_order_id));
2952 }
2953 }
2954 }
2955 }
2956 }
2957
2958 for (position_id, client_order_id) in assignments {
2959 let Some((venue, strategy_id)) = self.orders.get(&client_order_id).map(|order_cell| {
2960 let mut contingent = order_cell.borrow_mut();
2961 contingent.set_position_id(Some(position_id));
2962 (contingent.instrument_id().venue, contingent.strategy_id())
2963 }) else {
2964 continue;
2965 };
2966
2967 if let Err(e) =
2970 self.add_position_id(&position_id, &venue, &client_order_id, &strategy_id)
2971 {
2972 log::error!("Failed to re-index {client_order_id} -> {position_id}: {e}");
2973 }
2974 }
2975 }
2976
2977 pub fn add_position(&mut self, position: &Position, oms_type: OmsType) -> anyhow::Result<()> {
2985 self.add_position_inner(position.clone(), oms_type, true, false)
2986 }
2987
2988 pub fn add_position_without_order(
2996 &mut self,
2997 position: &Position,
2998 oms_type: OmsType,
2999 ) -> anyhow::Result<()> {
3000 self.add_position_inner(position.clone(), oms_type, false, false)
3001 }
3002
3003 pub fn replace_position(
3014 &mut self,
3015 position: &Position,
3016 oms_type: OmsType,
3017 index_order: bool,
3018 carry_replay_state: bool,
3019 ) -> anyhow::Result<()> {
3020 self.add_position_inner(position.clone(), oms_type, index_order, carry_replay_state)
3021 }
3022
3023 fn add_position_inner(
3024 &mut self,
3025 mut position: Position,
3026 oms_type: OmsType,
3027 index_order: bool,
3028 carry_replay_state: bool,
3029 ) -> anyhow::Result<()> {
3030 let key = position_oms_key(position.id);
3033 check_valid_string_ascii(&key, stringify!(key))?;
3034 let value = Bytes::from(serde_json::to_vec(&oms_type)?);
3035 check_predicate_false(value.is_empty(), stringify!(value))?;
3036
3037 let position_id = position.id;
3038 let strategy_id = position.strategy_id;
3039 let instrument_id = position.instrument_id;
3040 let account_id = position.account_id;
3041 let opening_order_id = position.opening_order_id;
3042
3043 log::debug!("Adding {position}");
3044
3045 let position_cell = if let Some(position_cell) = self.positions.get(&position_id).cloned() {
3049 let mut prior = position_cell.borrow_mut();
3050 if carry_replay_state {
3051 position.transfer_replay_state_from(&mut prior);
3052 }
3053 *prior = position;
3054 drop(prior);
3055 position_cell
3056 } else {
3057 let position_cell = SharedCell::new(position);
3058 self.positions.insert(position_id, position_cell.clone());
3059 position_cell
3060 };
3061
3062 self.index.position_oms.insert(position_id, oms_type);
3063 self.index.positions.insert(position_id);
3064 self.index.positions_open.insert(position_id);
3065 self.index.positions_closed.remove(&position_id); self.index.strategies.insert(strategy_id);
3067 self.index.strategy_orders.entry(strategy_id).or_default();
3068
3069 if index_order {
3070 self.index_position_id_in_memory(
3071 &position_id,
3072 &instrument_id.venue,
3073 &opening_order_id,
3074 &strategy_id,
3075 );
3076 } else {
3077 self.index_position(&position_id, &instrument_id.venue, &strategy_id);
3078 }
3079
3080 let instrument_positions = self
3082 .index
3083 .instrument_positions
3084 .entry(instrument_id)
3085 .or_default();
3086 instrument_positions.insert(position_id);
3087 self.index
3088 .instrument_orders
3089 .entry(instrument_id)
3090 .or_default();
3091
3092 self.index
3094 .account_positions
3095 .entry(account_id)
3096 .or_default()
3097 .insert(position_id);
3098
3099 log::debug!("Adding general {key}");
3100 self.general.insert(key.clone(), value.clone());
3101
3102 if index_order {
3103 self.persist_position_id(&position_id, &opening_order_id)?;
3104 }
3105
3106 if let Some(database) = &mut self.database {
3107 database.add_position(&position_cell.borrow())?;
3108 database.add(key, value)?;
3117 }
3118
3119 Ok(())
3120 }
3121
3122 pub fn update_account(&mut self, account: &AccountAny) -> anyhow::Result<()> {
3131 let account_id = account.id();
3132 match self.accounts.get(&account_id) {
3133 Some(account_cell) => *account_cell.borrow_mut() = account.clone(),
3134 None => {
3135 self.accounts
3136 .insert(account_id, SharedCell::new(account.clone()));
3137 }
3138 }
3139
3140 if let Some(database) = &mut self.database {
3141 database.update_account(account)?;
3142 }
3143 Ok(())
3144 }
3145
3146 #[must_use]
3157 pub fn take_account(&mut self, account_id: &AccountId) -> Option<AccountAny> {
3158 let cell = self.accounts.remove(account_id)?;
3159 let rc: Rc<RefCell<AccountAny>> = cell.into();
3160
3161 match Rc::try_unwrap(rc) {
3162 Ok(cell) => Some(cell.into_inner()),
3163 Err(rc) => {
3164 log::error!(
3165 "Cannot move account {account_id} out of cache: account cell has an outstanding owner"
3166 );
3167 self.accounts.insert(*account_id, rc.into());
3168 None
3169 }
3170 }
3171 }
3172
3173 pub fn cache_account_owned(&mut self, account: AccountAny) {
3175 let account_id = account.id();
3176 self.index
3177 .venue_account
3178 .insert(account_id.get_issuer(), account_id);
3179 match self.accounts.get(&account_id) {
3180 Some(account_cell) => *account_cell.borrow_mut() = account,
3181 None => {
3182 self.accounts.insert(account_id, SharedCell::new(account));
3183 }
3184 }
3185 }
3186
3187 pub fn update_account_owned(&mut self, account: AccountAny) -> anyhow::Result<()> {
3193 let account_id = account.id();
3194 self.cache_account_owned(account);
3195
3196 if let Some(database) = &mut self.database {
3197 let Some(account_cell) = self.accounts.get(&account_id) else {
3198 anyhow::bail!("Account {account_id} not found after cache update");
3199 };
3200 database.update_account(&account_cell.borrow())?;
3201 }
3202 Ok(())
3203 }
3204
3205 pub fn update_account_state(&mut self, event: &AccountState) -> anyhow::Result<()> {
3215 let Some(cell) = self.accounts.get(&event.account_id) else {
3216 return self.add_account(AccountAny::from_events(std::slice::from_ref(event))?);
3217 };
3218
3219 cell.borrow_mut().apply(event.clone())?;
3220
3221 if let Some(database) = &mut self.database {
3222 database.update_account(&cell.borrow())?;
3223 }
3224 Ok(())
3225 }
3226
3227 pub fn replace_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
3238 let client_order_id = order.client_order_id();
3239 if let Some(venue_order_id) = order.venue_order_id() {
3240 self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
3241 }
3242
3243 match self.orders.get(&client_order_id) {
3244 Some(order_cell) => *order_cell.borrow_mut() = order.clone(),
3247 None => {
3248 self.orders
3249 .insert(client_order_id, SharedCell::new(order.clone()));
3250 }
3251 }
3252
3253 self.refresh_order(order)
3254 }
3255
3256 pub fn update_order(&mut self, event: &OrderEventAny) -> anyhow::Result<OrderAny> {
3262 let event_client_order_id = event.client_order_id();
3263 let client_order_id = if self.order_exists(&event_client_order_id) {
3264 event_client_order_id
3265 } else if let Some(venue_order_id) = event.venue_order_id() {
3266 self.index
3267 .venue_order_ids
3268 .get(&venue_order_id)
3269 .copied()
3270 .ok_or(OrderError::NotFound(event_client_order_id))?
3271 } else {
3272 return Err(OrderError::NotFound(event_client_order_id).into());
3273 };
3274
3275 let order_cell = self
3276 .orders
3277 .get(&client_order_id)
3278 .cloned()
3279 .ok_or(OrderError::NotFound(client_order_id))?;
3280
3281 let mut snapshot = order_cell.borrow().clone();
3285 snapshot.apply(event.clone())?;
3286
3287 if let Some(venue_order_id) = snapshot.venue_order_id() {
3291 self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
3292 }
3293
3294 *order_cell.borrow_mut() = snapshot.clone();
3295
3296 if let Err(e) = self.refresh_order(&snapshot) {
3297 log::error!("Error updating order in cache: {e}");
3298 }
3299
3300 Ok(snapshot)
3301 }
3302
3303 fn refresh_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
3304 let client_order_id = order.client_order_id();
3305
3306 if let Some(venue_order_id) = order.venue_order_id() {
3309 let overwrite = matches!(order.last_event(), OrderEventAny::Updated(_));
3310 if let Err(e) = self.add_venue_order_id(&client_order_id, &venue_order_id, overwrite) {
3311 if e.is::<VenueOrderIdOwnershipError>() {
3312 return Err(e);
3313 }
3314 log::error!("Error indexing venue order ID in cache: {e}");
3315 }
3316 }
3317
3318 if order.is_active_local() {
3319 self.index.orders_active_local.insert(client_order_id);
3320 } else {
3321 self.index.orders_active_local.remove(&client_order_id);
3322 }
3323
3324 if order.is_inflight() {
3326 self.index.orders_inflight.insert(client_order_id);
3327 } else {
3328 self.index.orders_inflight.remove(&client_order_id);
3329 }
3330
3331 if order.is_open() {
3333 self.index.orders_closed.remove(&client_order_id);
3334 self.index.orders_open.insert(client_order_id);
3335 } else if order.is_closed() {
3336 self.index.orders_open.remove(&client_order_id);
3337 self.index.orders_pending_cancel.remove(&client_order_id);
3338 self.index.orders_closed.insert(client_order_id);
3339 }
3340
3341 if matches!(order.last_event(), OrderEventAny::CancelRejected(_)) {
3343 self.index.orders_pending_cancel.remove(&client_order_id);
3344 }
3345
3346 if order.emulation_trigger().is_some() && !order.is_closed() {
3348 self.index.orders_emulated.insert(client_order_id);
3349 } else {
3350 self.index.orders_emulated.remove(&client_order_id);
3351 }
3352
3353 if let Some(account_id) = order.account_id() {
3355 self.index
3356 .account_orders
3357 .entry(account_id)
3358 .or_default()
3359 .insert(client_order_id);
3360 }
3361
3362 if !self.own_books.is_empty() {
3364 let own_book = self.own_order_book(&order.instrument_id());
3365 if (own_book.is_some() && order.is_closed()) || should_handle_own_book_order(order) {
3366 self.update_own_order_book(order);
3367 }
3368 }
3369
3370 if let Some(database) = &mut self.database {
3371 database.update_order(order.last_event())?;
3372 }
3377
3378 Ok(())
3379 }
3380
3381 pub fn update_order_pending_cancel_local(&mut self, order: &OrderAny) {
3383 self.index
3384 .orders_pending_cancel
3385 .insert(order.client_order_id());
3386 }
3387
3388 pub fn update_position(&mut self, position: &Position) -> anyhow::Result<()> {
3398 let Some(position_cell) = self.positions.get(&position.id).cloned() else {
3399 anyhow::bail!("Cannot update position {}: not found in cache", position.id);
3400 };
3401
3402 self.refresh_position_indexes(position);
3403
3404 *position_cell.borrow_mut() = position.clone();
3405
3406 if let Some(database) = &mut self.database {
3407 database.update_position(position)?;
3408 }
3413
3414 Ok(())
3415 }
3416
3417 pub fn update_position_from_fill(
3427 &mut self,
3428 position_id: PositionId,
3429 fill: &OrderFilled,
3430 ) -> anyhow::Result<Position> {
3431 let Some(position_cell) = self.positions.get(&position_id).cloned() else {
3432 anyhow::bail!("Cannot update position {position_id}: not found in cache");
3433 };
3434
3435 let position = {
3436 let mut position = position_cell.borrow_mut();
3437 position.apply(fill);
3438 position.clone_without_events()
3439 };
3440
3441 self.refresh_position_indexes(&position);
3442
3443 if let Some(database) = &mut self.database {
3444 database.update_position(&position_cell.borrow())?;
3445 }
3446
3447 Ok(position)
3448 }
3449
3450 fn refresh_position_indexes(&mut self, position: &Position) {
3451 if position.is_open() {
3452 self.index.positions_open.insert(position.id);
3453 self.index.positions_closed.remove(&position.id);
3454 } else {
3455 self.index.positions_closed.insert(position.id);
3456 self.index.positions_open.remove(&position.id);
3457 }
3458 }
3459
3460 #[must_use]
3462 pub fn oms_type(&self, position_id: &PositionId) -> Option<OmsType> {
3463 self.index.position_oms.get(position_id).copied()
3464 }
3465
3466 pub fn snapshot_order_state(&self, order: &OrderAny) -> anyhow::Result<()> {
3472 let Some(database) = &self.database else {
3473 log::warn!(
3474 "Cannot snapshot order state for {} (no database configured)",
3475 order.client_order_id()
3476 );
3477 return Ok(());
3478 };
3479
3480 database.snapshot_order_state(order)
3481 }
3482
3483 fn collect_order_filter_sources<'a>(
3494 &'a self,
3495 venue: Option<&Venue>,
3496 instrument_id: Option<&InstrumentId>,
3497 strategy_id: Option<&StrategyId>,
3498 account_id: Option<&AccountId>,
3499 ) -> FilterSources<'a, ClientOrderId> {
3500 let mut sources: Vec<&AHashSet<ClientOrderId>> = Vec::with_capacity(4);
3501
3502 if let Some(venue) = venue {
3503 match self.index.venue_orders.get(venue) {
3504 Some(set) => sources.push(set),
3505 None => return FilterSources::Empty,
3506 }
3507 }
3508
3509 if let Some(instrument_id) = instrument_id {
3510 match self.index.instrument_orders.get(instrument_id) {
3511 Some(set) => sources.push(set),
3512 None => return FilterSources::Empty,
3513 }
3514 }
3515
3516 if let Some(strategy_id) = strategy_id {
3517 match self.index.strategy_orders.get(strategy_id) {
3518 Some(set) => sources.push(set),
3519 None => return FilterSources::Empty,
3520 }
3521 }
3522
3523 if let Some(account_id) = account_id {
3524 match self.index.account_orders.get(account_id) {
3525 Some(set) => sources.push(set),
3526 None => return FilterSources::Empty,
3527 }
3528 }
3529
3530 if sources.is_empty() {
3531 FilterSources::Unfiltered
3532 } else {
3533 FilterSources::Sets(sources)
3534 }
3535 }
3536
3537 fn collect_position_filter_sources<'a>(
3538 &'a self,
3539 venue: Option<&Venue>,
3540 instrument_id: Option<&InstrumentId>,
3541 strategy_id: Option<&StrategyId>,
3542 account_id: Option<&AccountId>,
3543 ) -> FilterSources<'a, PositionId> {
3544 let mut sources: Vec<&AHashSet<PositionId>> = Vec::with_capacity(4);
3545
3546 if let Some(venue) = venue {
3547 match self.index.venue_positions.get(venue) {
3548 Some(set) => sources.push(set),
3549 None => return FilterSources::Empty,
3550 }
3551 }
3552
3553 if let Some(instrument_id) = instrument_id {
3554 match self.index.instrument_positions.get(instrument_id) {
3555 Some(set) => sources.push(set),
3556 None => return FilterSources::Empty,
3557 }
3558 }
3559
3560 if let Some(strategy_id) = strategy_id {
3561 match self.index.strategy_positions.get(strategy_id) {
3562 Some(set) => sources.push(set),
3563 None => return FilterSources::Empty,
3564 }
3565 }
3566
3567 if let Some(account_id) = account_id {
3568 match self.index.account_positions.get(account_id) {
3569 Some(set) => sources.push(set),
3570 None => return FilterSources::Empty,
3571 }
3572 }
3573
3574 if sources.is_empty() {
3575 FilterSources::Unfiltered
3576 } else {
3577 FilterSources::Sets(sources)
3578 }
3579 }
3580
3581 fn query_orders_in_bucket(
3587 &self,
3588 bucket: &AHashSet<ClientOrderId>,
3589 venue: Option<&Venue>,
3590 instrument_id: Option<&InstrumentId>,
3591 strategy_id: Option<&StrategyId>,
3592 account_id: Option<&AccountId>,
3593 ) -> AHashSet<ClientOrderId> {
3594 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3595 FilterSources::Empty => AHashSet::new(),
3596 FilterSources::Unfiltered => bucket.clone(),
3597 FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
3598 }
3599 }
3600
3601 fn query_positions_in_bucket(
3602 &self,
3603 bucket: &AHashSet<PositionId>,
3604 venue: Option<&Venue>,
3605 instrument_id: Option<&InstrumentId>,
3606 strategy_id: Option<&StrategyId>,
3607 account_id: Option<&AccountId>,
3608 ) -> AHashSet<PositionId> {
3609 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3610 FilterSources::Empty => AHashSet::new(),
3611 FilterSources::Unfiltered => bucket.clone(),
3612 FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
3613 }
3614 }
3615
3616 fn view_orders_in_bucket<'a>(
3619 &'a self,
3620 bucket: &'a AHashSet<ClientOrderId>,
3621 venue: Option<&Venue>,
3622 instrument_id: Option<&InstrumentId>,
3623 strategy_id: Option<&StrategyId>,
3624 account_id: Option<&AccountId>,
3625 ) -> Cow<'a, AHashSet<ClientOrderId>> {
3626 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3627 FilterSources::Empty => Cow::Owned(AHashSet::new()),
3628 FilterSources::Unfiltered => Cow::Borrowed(bucket),
3629 FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
3630 }
3631 }
3632
3633 fn view_positions_in_bucket<'a>(
3634 &'a self,
3635 bucket: &'a AHashSet<PositionId>,
3636 venue: Option<&Venue>,
3637 instrument_id: Option<&InstrumentId>,
3638 strategy_id: Option<&StrategyId>,
3639 account_id: Option<&AccountId>,
3640 ) -> Cow<'a, AHashSet<PositionId>> {
3641 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3642 FilterSources::Empty => Cow::Owned(AHashSet::new()),
3643 FilterSources::Unfiltered => Cow::Borrowed(bucket),
3644 FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
3645 }
3646 }
3647
3648 fn iter_orders_in_bucket<'a>(
3653 &'a self,
3654 bucket: &'a AHashSet<ClientOrderId>,
3655 venue: Option<&Venue>,
3656 instrument_id: Option<&InstrumentId>,
3657 strategy_id: Option<&StrategyId>,
3658 account_id: Option<&AccountId>,
3659 ) -> Box<dyn Iterator<Item = ClientOrderId> + 'a> {
3660 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3661 FilterSources::Empty => Box::new(std::iter::empty()),
3662 FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
3663 FilterSources::Sets(mut sources) => {
3664 sources.push(bucket);
3665 sources.sort_unstable_by_key(|s| s.len());
3666 let driver = sources[0];
3667 let rest: Vec<&'a AHashSet<ClientOrderId>> = sources[1..].to_vec();
3668 Box::new(
3669 driver
3670 .iter()
3671 .copied()
3672 .filter(move |id| rest.iter().all(|s| s.contains(id))),
3673 )
3674 }
3675 }
3676 }
3677
3678 fn iter_positions_in_bucket<'a>(
3679 &'a self,
3680 bucket: &'a AHashSet<PositionId>,
3681 venue: Option<&Venue>,
3682 instrument_id: Option<&InstrumentId>,
3683 strategy_id: Option<&StrategyId>,
3684 account_id: Option<&AccountId>,
3685 ) -> Box<dyn Iterator<Item = PositionId> + 'a> {
3686 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3687 FilterSources::Empty => Box::new(std::iter::empty()),
3688 FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
3689 FilterSources::Sets(mut sources) => {
3690 sources.push(bucket);
3691 sources.sort_unstable_by_key(|s| s.len());
3692 let driver = sources[0];
3693 let rest: Vec<&'a AHashSet<PositionId>> = sources[1..].to_vec();
3694 Box::new(
3695 driver
3696 .iter()
3697 .copied()
3698 .filter(move |id| rest.iter().all(|s| s.contains(id))),
3699 )
3700 }
3701 }
3702 }
3703
3704 fn count_orders_in_bucket(
3710 &self,
3711 bucket: &AHashSet<ClientOrderId>,
3712 venue: Option<&Venue>,
3713 instrument_id: Option<&InstrumentId>,
3714 strategy_id: Option<&StrategyId>,
3715 account_id: Option<&AccountId>,
3716 side: Option<OrderSide>,
3717 ) -> usize {
3718 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3719 FilterSources::Empty => 0,
3720 FilterSources::Unfiltered => side.map_or_else(
3721 || bucket.len(),
3722 |side| {
3723 bucket
3724 .iter()
3725 .filter(|id| self.order_side_matches(id, side))
3726 .count()
3727 },
3728 ),
3729 FilterSources::Sets(mut sources) => {
3730 sources.push(bucket);
3731 sources.sort_unstable_by_key(|s| s.len());
3732 let driver = sources[0];
3733 let rest = &sources[1..];
3734
3735 driver
3736 .iter()
3737 .filter(|id| rest.iter().all(|s| s.contains(id)))
3738 .filter(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
3739 .count()
3740 }
3741 }
3742 }
3743
3744 fn count_positions_in_bucket(
3745 &self,
3746 bucket: &AHashSet<PositionId>,
3747 venue: Option<&Venue>,
3748 instrument_id: Option<&InstrumentId>,
3749 strategy_id: Option<&StrategyId>,
3750 account_id: Option<&AccountId>,
3751 side: Option<PositionSide>,
3752 ) -> usize {
3753 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3754 FilterSources::Empty => 0,
3755 FilterSources::Unfiltered => side.map_or_else(
3756 || bucket.len(),
3757 |side| {
3758 bucket
3759 .iter()
3760 .filter(|id| self.position_side_matches(id, side))
3761 .count()
3762 },
3763 ),
3764 FilterSources::Sets(mut sources) => {
3765 sources.push(bucket);
3766 sources.sort_unstable_by_key(|s| s.len());
3767 let driver = sources[0];
3768 let rest = &sources[1..];
3769
3770 driver
3771 .iter()
3772 .filter(|id| rest.iter().all(|s| s.contains(id)))
3773 .filter(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
3774 .count()
3775 }
3776 }
3777 }
3778
3779 fn any_orders_in_bucket(
3785 &self,
3786 bucket: &AHashSet<ClientOrderId>,
3787 venue: Option<&Venue>,
3788 instrument_id: Option<&InstrumentId>,
3789 strategy_id: Option<&StrategyId>,
3790 account_id: Option<&AccountId>,
3791 side: Option<OrderSide>,
3792 ) -> bool {
3793 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3794 FilterSources::Empty => false,
3795 FilterSources::Unfiltered => side.map_or_else(
3796 || !bucket.is_empty(),
3797 |side| bucket.iter().any(|id| self.order_side_matches(id, side)),
3798 ),
3799 FilterSources::Sets(mut sources) => {
3800 sources.push(bucket);
3801 sources.sort_unstable_by_key(|s| s.len());
3802 let driver = sources[0];
3803 let rest = &sources[1..];
3804
3805 driver
3806 .iter()
3807 .filter(|id| rest.iter().all(|s| s.contains(id)))
3808 .any(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
3809 }
3810 }
3811 }
3812
3813 fn any_positions_in_bucket(
3814 &self,
3815 bucket: &AHashSet<PositionId>,
3816 venue: Option<&Venue>,
3817 instrument_id: Option<&InstrumentId>,
3818 strategy_id: Option<&StrategyId>,
3819 account_id: Option<&AccountId>,
3820 side: Option<PositionSide>,
3821 ) -> bool {
3822 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3823 FilterSources::Empty => false,
3824 FilterSources::Unfiltered => side.map_or_else(
3825 || !bucket.is_empty(),
3826 |side| bucket.iter().any(|id| self.position_side_matches(id, side)),
3827 ),
3828 FilterSources::Sets(mut sources) => {
3829 sources.push(bucket);
3830 sources.sort_unstable_by_key(|s| s.len());
3831 let driver = sources[0];
3832 let rest = &sources[1..];
3833
3834 driver
3835 .iter()
3836 .filter(|id| rest.iter().all(|s| s.contains(id)))
3837 .any(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
3838 }
3839 }
3840 }
3841
3842 fn order_side_matches(&self, client_order_id: &ClientOrderId, side: OrderSide) -> bool {
3843 self.orders
3844 .get(client_order_id)
3845 .is_some_and(|cell| cell.borrow().order_side() == side)
3846 }
3847
3848 fn position_side_matches(&self, position_id: &PositionId, side: PositionSide) -> bool {
3849 self.positions
3850 .get(position_id)
3851 .is_some_and(|cell| cell.borrow().side == side)
3852 }
3853
3854 fn get_orders_for_ids(
3860 &self,
3861 client_order_ids: impl IntoIterator<Item = ClientOrderId>,
3862 side: Option<OrderSide>,
3863 ) -> Vec<OrderRef<'_>> {
3864 const UNCACHED_SORT_MAX_LEN: usize = 32;
3865
3866 let mut orders = Vec::new();
3867
3868 for client_order_id in client_order_ids {
3869 let order_cell = self
3870 .orders
3871 .get(&client_order_id)
3872 .unwrap_or_else(|| panic!("Order {client_order_id} not found"));
3873 let order = OrderRef::new(order_cell.borrow());
3874
3875 if side.is_none_or(|side| side == order.order_side()) {
3876 orders.push(order);
3877 }
3878 }
3879
3880 let key = |order: &OrderRef<'_>| order.client_order_id();
3883
3884 if orders.len() <= UNCACHED_SORT_MAX_LEN {
3885 orders.sort_by_key(key);
3886 } else {
3887 orders.sort_by_cached_key(key);
3888 }
3889
3890 orders
3891 }
3892
3893 fn get_positions_for_ids(
3903 &self,
3904 position_ids: &AHashSet<PositionId>,
3905 side: Option<PositionSide>,
3906 ) -> Vec<PositionRef<'_>> {
3907 let mut positions = Vec::new();
3908
3909 for position_id in position_ids {
3910 let position_cell = self
3911 .positions
3912 .get(position_id)
3913 .unwrap_or_else(|| panic!("Position {position_id} not found"));
3914 let position = PositionRef::new(position_cell.borrow());
3915
3916 if side.is_none_or(|side| side == position.side) {
3917 positions.push(position);
3918 }
3919 }
3920
3921 positions.sort_by_key(|p| p.id);
3924 positions
3925 }
3926
3927 #[must_use]
3929 pub fn client_order_ids(
3930 &self,
3931 venue: Option<&Venue>,
3932 instrument_id: Option<&InstrumentId>,
3933 strategy_id: Option<&StrategyId>,
3934 account_id: Option<&AccountId>,
3935 ) -> AHashSet<ClientOrderId> {
3936 self.query_orders_in_bucket(
3937 &self.index.orders,
3938 venue,
3939 instrument_id,
3940 strategy_id,
3941 account_id,
3942 )
3943 }
3944
3945 #[must_use]
3947 pub fn client_order_ids_open(
3948 &self,
3949 venue: Option<&Venue>,
3950 instrument_id: Option<&InstrumentId>,
3951 strategy_id: Option<&StrategyId>,
3952 account_id: Option<&AccountId>,
3953 ) -> AHashSet<ClientOrderId> {
3954 self.query_orders_in_bucket(
3955 &self.index.orders_open,
3956 venue,
3957 instrument_id,
3958 strategy_id,
3959 account_id,
3960 )
3961 }
3962
3963 #[must_use]
3965 pub fn client_order_ids_closed(
3966 &self,
3967 venue: Option<&Venue>,
3968 instrument_id: Option<&InstrumentId>,
3969 strategy_id: Option<&StrategyId>,
3970 account_id: Option<&AccountId>,
3971 ) -> AHashSet<ClientOrderId> {
3972 self.query_orders_in_bucket(
3973 &self.index.orders_closed,
3974 venue,
3975 instrument_id,
3976 strategy_id,
3977 account_id,
3978 )
3979 }
3980
3981 #[must_use]
3986 pub fn client_order_ids_active_local(
3987 &self,
3988 venue: Option<&Venue>,
3989 instrument_id: Option<&InstrumentId>,
3990 strategy_id: Option<&StrategyId>,
3991 account_id: Option<&AccountId>,
3992 ) -> AHashSet<ClientOrderId> {
3993 self.query_orders_in_bucket(
3994 &self.index.orders_active_local,
3995 venue,
3996 instrument_id,
3997 strategy_id,
3998 account_id,
3999 )
4000 }
4001
4002 #[must_use]
4004 pub fn client_order_ids_emulated(
4005 &self,
4006 venue: Option<&Venue>,
4007 instrument_id: Option<&InstrumentId>,
4008 strategy_id: Option<&StrategyId>,
4009 account_id: Option<&AccountId>,
4010 ) -> AHashSet<ClientOrderId> {
4011 self.query_orders_in_bucket(
4012 &self.index.orders_emulated,
4013 venue,
4014 instrument_id,
4015 strategy_id,
4016 account_id,
4017 )
4018 }
4019
4020 #[must_use]
4022 pub fn client_order_ids_inflight(
4023 &self,
4024 venue: Option<&Venue>,
4025 instrument_id: Option<&InstrumentId>,
4026 strategy_id: Option<&StrategyId>,
4027 account_id: Option<&AccountId>,
4028 ) -> AHashSet<ClientOrderId> {
4029 self.query_orders_in_bucket(
4030 &self.index.orders_inflight,
4031 venue,
4032 instrument_id,
4033 strategy_id,
4034 account_id,
4035 )
4036 }
4037
4038 #[must_use]
4040 pub fn position_ids(
4041 &self,
4042 venue: Option<&Venue>,
4043 instrument_id: Option<&InstrumentId>,
4044 strategy_id: Option<&StrategyId>,
4045 account_id: Option<&AccountId>,
4046 ) -> AHashSet<PositionId> {
4047 self.query_positions_in_bucket(
4048 &self.index.positions,
4049 venue,
4050 instrument_id,
4051 strategy_id,
4052 account_id,
4053 )
4054 }
4055
4056 #[must_use]
4058 pub fn position_open_ids(
4059 &self,
4060 venue: Option<&Venue>,
4061 instrument_id: Option<&InstrumentId>,
4062 strategy_id: Option<&StrategyId>,
4063 account_id: Option<&AccountId>,
4064 ) -> AHashSet<PositionId> {
4065 self.query_positions_in_bucket(
4066 &self.index.positions_open,
4067 venue,
4068 instrument_id,
4069 strategy_id,
4070 account_id,
4071 )
4072 }
4073
4074 #[must_use]
4076 pub fn position_closed_ids(
4077 &self,
4078 venue: Option<&Venue>,
4079 instrument_id: Option<&InstrumentId>,
4080 strategy_id: Option<&StrategyId>,
4081 account_id: Option<&AccountId>,
4082 ) -> AHashSet<PositionId> {
4083 self.query_positions_in_bucket(
4084 &self.index.positions_closed,
4085 venue,
4086 instrument_id,
4087 strategy_id,
4088 account_id,
4089 )
4090 }
4091
4092 #[must_use]
4099 pub fn client_order_ids_view(
4100 &self,
4101 venue: Option<&Venue>,
4102 instrument_id: Option<&InstrumentId>,
4103 strategy_id: Option<&StrategyId>,
4104 account_id: Option<&AccountId>,
4105 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4106 self.view_orders_in_bucket(
4107 &self.index.orders,
4108 venue,
4109 instrument_id,
4110 strategy_id,
4111 account_id,
4112 )
4113 }
4114
4115 #[must_use]
4117 pub fn client_order_ids_open_view(
4118 &self,
4119 venue: Option<&Venue>,
4120 instrument_id: Option<&InstrumentId>,
4121 strategy_id: Option<&StrategyId>,
4122 account_id: Option<&AccountId>,
4123 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4124 self.view_orders_in_bucket(
4125 &self.index.orders_open,
4126 venue,
4127 instrument_id,
4128 strategy_id,
4129 account_id,
4130 )
4131 }
4132
4133 #[must_use]
4135 pub fn client_order_ids_closed_view(
4136 &self,
4137 venue: Option<&Venue>,
4138 instrument_id: Option<&InstrumentId>,
4139 strategy_id: Option<&StrategyId>,
4140 account_id: Option<&AccountId>,
4141 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4142 self.view_orders_in_bucket(
4143 &self.index.orders_closed,
4144 venue,
4145 instrument_id,
4146 strategy_id,
4147 account_id,
4148 )
4149 }
4150
4151 #[must_use]
4153 pub fn client_order_ids_active_local_view(
4154 &self,
4155 venue: Option<&Venue>,
4156 instrument_id: Option<&InstrumentId>,
4157 strategy_id: Option<&StrategyId>,
4158 account_id: Option<&AccountId>,
4159 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4160 self.view_orders_in_bucket(
4161 &self.index.orders_active_local,
4162 venue,
4163 instrument_id,
4164 strategy_id,
4165 account_id,
4166 )
4167 }
4168
4169 #[must_use]
4171 pub fn client_order_ids_emulated_view(
4172 &self,
4173 venue: Option<&Venue>,
4174 instrument_id: Option<&InstrumentId>,
4175 strategy_id: Option<&StrategyId>,
4176 account_id: Option<&AccountId>,
4177 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4178 self.view_orders_in_bucket(
4179 &self.index.orders_emulated,
4180 venue,
4181 instrument_id,
4182 strategy_id,
4183 account_id,
4184 )
4185 }
4186
4187 #[must_use]
4189 pub fn client_order_ids_inflight_view(
4190 &self,
4191 venue: Option<&Venue>,
4192 instrument_id: Option<&InstrumentId>,
4193 strategy_id: Option<&StrategyId>,
4194 account_id: Option<&AccountId>,
4195 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4196 self.view_orders_in_bucket(
4197 &self.index.orders_inflight,
4198 venue,
4199 instrument_id,
4200 strategy_id,
4201 account_id,
4202 )
4203 }
4204
4205 #[must_use]
4207 pub fn position_ids_view(
4208 &self,
4209 venue: Option<&Venue>,
4210 instrument_id: Option<&InstrumentId>,
4211 strategy_id: Option<&StrategyId>,
4212 account_id: Option<&AccountId>,
4213 ) -> Cow<'_, AHashSet<PositionId>> {
4214 self.view_positions_in_bucket(
4215 &self.index.positions,
4216 venue,
4217 instrument_id,
4218 strategy_id,
4219 account_id,
4220 )
4221 }
4222
4223 #[must_use]
4225 pub fn position_open_ids_view(
4226 &self,
4227 venue: Option<&Venue>,
4228 instrument_id: Option<&InstrumentId>,
4229 strategy_id: Option<&StrategyId>,
4230 account_id: Option<&AccountId>,
4231 ) -> Cow<'_, AHashSet<PositionId>> {
4232 self.view_positions_in_bucket(
4233 &self.index.positions_open,
4234 venue,
4235 instrument_id,
4236 strategy_id,
4237 account_id,
4238 )
4239 }
4240
4241 #[must_use]
4243 pub fn position_closed_ids_view(
4244 &self,
4245 venue: Option<&Venue>,
4246 instrument_id: Option<&InstrumentId>,
4247 strategy_id: Option<&StrategyId>,
4248 account_id: Option<&AccountId>,
4249 ) -> Cow<'_, AHashSet<PositionId>> {
4250 self.view_positions_in_bucket(
4251 &self.index.positions_closed,
4252 venue,
4253 instrument_id,
4254 strategy_id,
4255 account_id,
4256 )
4257 }
4258
4259 pub fn iter_client_order_ids(
4265 &self,
4266 venue: Option<&Venue>,
4267 instrument_id: Option<&InstrumentId>,
4268 strategy_id: Option<&StrategyId>,
4269 account_id: Option<&AccountId>,
4270 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4271 self.iter_orders_in_bucket(
4272 &self.index.orders,
4273 venue,
4274 instrument_id,
4275 strategy_id,
4276 account_id,
4277 )
4278 }
4279
4280 pub fn iter_client_order_ids_open(
4282 &self,
4283 venue: Option<&Venue>,
4284 instrument_id: Option<&InstrumentId>,
4285 strategy_id: Option<&StrategyId>,
4286 account_id: Option<&AccountId>,
4287 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4288 self.iter_orders_in_bucket(
4289 &self.index.orders_open,
4290 venue,
4291 instrument_id,
4292 strategy_id,
4293 account_id,
4294 )
4295 }
4296
4297 pub fn iter_client_order_ids_closed(
4299 &self,
4300 venue: Option<&Venue>,
4301 instrument_id: Option<&InstrumentId>,
4302 strategy_id: Option<&StrategyId>,
4303 account_id: Option<&AccountId>,
4304 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4305 self.iter_orders_in_bucket(
4306 &self.index.orders_closed,
4307 venue,
4308 instrument_id,
4309 strategy_id,
4310 account_id,
4311 )
4312 }
4313
4314 pub fn iter_client_order_ids_active_local(
4316 &self,
4317 venue: Option<&Venue>,
4318 instrument_id: Option<&InstrumentId>,
4319 strategy_id: Option<&StrategyId>,
4320 account_id: Option<&AccountId>,
4321 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4322 self.iter_orders_in_bucket(
4323 &self.index.orders_active_local,
4324 venue,
4325 instrument_id,
4326 strategy_id,
4327 account_id,
4328 )
4329 }
4330
4331 pub fn iter_client_order_ids_emulated(
4333 &self,
4334 venue: Option<&Venue>,
4335 instrument_id: Option<&InstrumentId>,
4336 strategy_id: Option<&StrategyId>,
4337 account_id: Option<&AccountId>,
4338 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4339 self.iter_orders_in_bucket(
4340 &self.index.orders_emulated,
4341 venue,
4342 instrument_id,
4343 strategy_id,
4344 account_id,
4345 )
4346 }
4347
4348 pub fn iter_client_order_ids_inflight(
4350 &self,
4351 venue: Option<&Venue>,
4352 instrument_id: Option<&InstrumentId>,
4353 strategy_id: Option<&StrategyId>,
4354 account_id: Option<&AccountId>,
4355 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4356 self.iter_orders_in_bucket(
4357 &self.index.orders_inflight,
4358 venue,
4359 instrument_id,
4360 strategy_id,
4361 account_id,
4362 )
4363 }
4364
4365 pub fn iter_position_ids(
4367 &self,
4368 venue: Option<&Venue>,
4369 instrument_id: Option<&InstrumentId>,
4370 strategy_id: Option<&StrategyId>,
4371 account_id: Option<&AccountId>,
4372 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
4373 self.iter_positions_in_bucket(
4374 &self.index.positions,
4375 venue,
4376 instrument_id,
4377 strategy_id,
4378 account_id,
4379 )
4380 }
4381
4382 pub fn iter_position_open_ids(
4384 &self,
4385 venue: Option<&Venue>,
4386 instrument_id: Option<&InstrumentId>,
4387 strategy_id: Option<&StrategyId>,
4388 account_id: Option<&AccountId>,
4389 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
4390 self.iter_positions_in_bucket(
4391 &self.index.positions_open,
4392 venue,
4393 instrument_id,
4394 strategy_id,
4395 account_id,
4396 )
4397 }
4398
4399 pub fn iter_position_closed_ids(
4401 &self,
4402 venue: Option<&Venue>,
4403 instrument_id: Option<&InstrumentId>,
4404 strategy_id: Option<&StrategyId>,
4405 account_id: Option<&AccountId>,
4406 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
4407 self.iter_positions_in_bucket(
4408 &self.index.positions_closed,
4409 venue,
4410 instrument_id,
4411 strategy_id,
4412 account_id,
4413 )
4414 }
4415
4416 #[must_use]
4418 pub fn strategy_ids(&self) -> AHashSet<StrategyId> {
4419 self.index.strategies.clone()
4420 }
4421
4422 #[must_use]
4424 pub fn exec_algorithm_ids(&self) -> AHashSet<ExecAlgorithmId> {
4425 self.index.exec_algorithms.clone()
4426 }
4427
4428 #[must_use]
4437 pub fn order_ref(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
4438 self.orders
4439 .get(client_order_id)
4440 .map(|order_cell| OrderRef::new(order_cell.borrow()))
4441 }
4442
4443 #[must_use]
4447 pub fn order(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
4448 self.order_ref(client_order_id)
4449 }
4450
4451 pub fn try_order_ref(
4457 &self,
4458 client_order_id: &ClientOrderId,
4459 ) -> Result<OrderRef<'_>, OrderLookupError> {
4460 self.orders
4461 .get(client_order_id)
4462 .map(|order_cell| OrderRef::new(order_cell.borrow()))
4463 .ok_or_else(|| OrderLookupError::not_found(*client_order_id))
4464 }
4465
4466 pub fn try_order(
4474 &self,
4475 client_order_id: &ClientOrderId,
4476 ) -> Result<OrderRef<'_>, OrderLookupError> {
4477 self.try_order_ref(client_order_id)
4478 }
4479
4480 #[must_use]
4490 pub fn order_mut(&mut self, client_order_id: &ClientOrderId) -> Option<OrderRefMut<'_>> {
4491 self.orders
4492 .get(client_order_id)
4493 .map(|order_cell| OrderRefMut::new(order_cell.borrow_mut()))
4494 }
4495
4496 #[must_use]
4501 pub fn order_owned(&self, client_order_id: &ClientOrderId) -> Option<OrderAny> {
4502 self.orders
4503 .get(client_order_id)
4504 .map(|order_cell| order_cell.borrow().clone())
4505 }
4506
4507 pub fn try_order_owned(
4513 &self,
4514 client_order_id: &ClientOrderId,
4515 ) -> Result<OrderAny, OrderLookupError> {
4516 self.try_order_ref(client_order_id)
4517 .map(|order| order.cloned())
4518 }
4519
4520 #[must_use]
4522 pub fn orders_for_ids(
4523 &self,
4524 client_order_ids: &[ClientOrderId],
4525 context: &dyn Display,
4526 ) -> Vec<OrderAny> {
4527 let mut orders = Vec::with_capacity(client_order_ids.len());
4528 for id in client_order_ids {
4529 match self.orders.get(id) {
4530 Some(order_cell) => orders.push(order_cell.borrow().clone()),
4531 None => log::error!("Order {id} not found in cache for {context}"),
4532 }
4533 }
4534 orders
4535 }
4536
4537 #[must_use]
4539 pub fn client_order_id(&self, venue_order_id: &VenueOrderId) -> Option<&ClientOrderId> {
4540 self.index.venue_order_ids.get(venue_order_id)
4541 }
4542
4543 #[must_use]
4545 pub fn venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<&VenueOrderId> {
4546 self.index.client_order_ids.get(client_order_id)
4547 }
4548
4549 #[must_use]
4551 pub fn client_id(&self, client_order_id: &ClientOrderId) -> Option<&ClientId> {
4552 self.index.order_client.get(client_order_id)
4553 }
4554
4555 #[must_use]
4561 pub fn orders_refs(
4562 &self,
4563 venue: Option<&Venue>,
4564 instrument_id: Option<&InstrumentId>,
4565 strategy_id: Option<&StrategyId>,
4566 account_id: Option<&AccountId>,
4567 side: Option<OrderSide>,
4568 ) -> Vec<OrderRef<'_>> {
4569 if venue.is_none()
4570 && instrument_id.is_none()
4571 && strategy_id.is_none()
4572 && account_id.is_none()
4573 {
4574 return self.get_orders_for_ids(self.index.orders.iter().copied(), side);
4575 }
4576
4577 let client_order_ids =
4578 self.iter_client_order_ids(venue, instrument_id, strategy_id, account_id);
4579 self.get_orders_for_ids(client_order_ids, side)
4580 }
4581
4582 #[must_use]
4586 pub fn orders(
4587 &self,
4588 venue: Option<&Venue>,
4589 instrument_id: Option<&InstrumentId>,
4590 strategy_id: Option<&StrategyId>,
4591 account_id: Option<&AccountId>,
4592 side: Option<OrderSide>,
4593 ) -> Vec<OrderRef<'_>> {
4594 self.orders_refs(venue, instrument_id, strategy_id, account_id, side)
4595 }
4596
4597 #[must_use]
4599 pub fn orders_open_refs(
4600 &self,
4601 venue: Option<&Venue>,
4602 instrument_id: Option<&InstrumentId>,
4603 strategy_id: Option<&StrategyId>,
4604 account_id: Option<&AccountId>,
4605 side: Option<OrderSide>,
4606 ) -> Vec<OrderRef<'_>> {
4607 let client_order_ids =
4608 self.client_order_ids_open(venue, instrument_id, strategy_id, account_id);
4609 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4610 }
4611
4612 #[must_use]
4616 pub fn orders_open(
4617 &self,
4618 venue: Option<&Venue>,
4619 instrument_id: Option<&InstrumentId>,
4620 strategy_id: Option<&StrategyId>,
4621 account_id: Option<&AccountId>,
4622 side: Option<OrderSide>,
4623 ) -> Vec<OrderRef<'_>> {
4624 self.orders_open_refs(venue, instrument_id, strategy_id, account_id, side)
4625 }
4626
4627 #[must_use]
4629 pub fn orders_closed_refs(
4630 &self,
4631 venue: Option<&Venue>,
4632 instrument_id: Option<&InstrumentId>,
4633 strategy_id: Option<&StrategyId>,
4634 account_id: Option<&AccountId>,
4635 side: Option<OrderSide>,
4636 ) -> Vec<OrderRef<'_>> {
4637 let client_order_ids =
4638 self.client_order_ids_closed(venue, instrument_id, strategy_id, account_id);
4639 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4640 }
4641
4642 #[must_use]
4646 pub fn orders_closed(
4647 &self,
4648 venue: Option<&Venue>,
4649 instrument_id: Option<&InstrumentId>,
4650 strategy_id: Option<&StrategyId>,
4651 account_id: Option<&AccountId>,
4652 side: Option<OrderSide>,
4653 ) -> Vec<OrderRef<'_>> {
4654 self.orders_closed_refs(venue, instrument_id, strategy_id, account_id, side)
4655 }
4656
4657 #[must_use]
4662 pub fn orders_active_local_refs(
4663 &self,
4664 venue: Option<&Venue>,
4665 instrument_id: Option<&InstrumentId>,
4666 strategy_id: Option<&StrategyId>,
4667 account_id: Option<&AccountId>,
4668 side: Option<OrderSide>,
4669 ) -> Vec<OrderRef<'_>> {
4670 let client_order_ids =
4671 self.client_order_ids_active_local(venue, instrument_id, strategy_id, account_id);
4672 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4673 }
4674
4675 #[must_use]
4679 pub fn orders_active_local(
4680 &self,
4681 venue: Option<&Venue>,
4682 instrument_id: Option<&InstrumentId>,
4683 strategy_id: Option<&StrategyId>,
4684 account_id: Option<&AccountId>,
4685 side: Option<OrderSide>,
4686 ) -> Vec<OrderRef<'_>> {
4687 self.orders_active_local_refs(venue, instrument_id, strategy_id, account_id, side)
4688 }
4689
4690 #[must_use]
4692 pub fn orders_emulated_refs(
4693 &self,
4694 venue: Option<&Venue>,
4695 instrument_id: Option<&InstrumentId>,
4696 strategy_id: Option<&StrategyId>,
4697 account_id: Option<&AccountId>,
4698 side: Option<OrderSide>,
4699 ) -> Vec<OrderRef<'_>> {
4700 let client_order_ids =
4701 self.client_order_ids_emulated(venue, instrument_id, strategy_id, account_id);
4702 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4703 }
4704
4705 #[must_use]
4709 pub fn orders_emulated(
4710 &self,
4711 venue: Option<&Venue>,
4712 instrument_id: Option<&InstrumentId>,
4713 strategy_id: Option<&StrategyId>,
4714 account_id: Option<&AccountId>,
4715 side: Option<OrderSide>,
4716 ) -> Vec<OrderRef<'_>> {
4717 self.orders_emulated_refs(venue, instrument_id, strategy_id, account_id, side)
4718 }
4719
4720 #[must_use]
4722 pub fn orders_inflight_refs(
4723 &self,
4724 venue: Option<&Venue>,
4725 instrument_id: Option<&InstrumentId>,
4726 strategy_id: Option<&StrategyId>,
4727 account_id: Option<&AccountId>,
4728 side: Option<OrderSide>,
4729 ) -> Vec<OrderRef<'_>> {
4730 let client_order_ids =
4731 self.client_order_ids_inflight(venue, instrument_id, strategy_id, account_id);
4732 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4733 }
4734
4735 #[must_use]
4739 pub fn orders_inflight(
4740 &self,
4741 venue: Option<&Venue>,
4742 instrument_id: Option<&InstrumentId>,
4743 strategy_id: Option<&StrategyId>,
4744 account_id: Option<&AccountId>,
4745 side: Option<OrderSide>,
4746 ) -> Vec<OrderRef<'_>> {
4747 self.orders_inflight_refs(venue, instrument_id, strategy_id, account_id, side)
4748 }
4749
4750 #[must_use]
4752 pub fn orders_for_position(&self, position_id: &PositionId) -> Vec<OrderRef<'_>> {
4753 match self.index.position_orders.get(position_id) {
4754 Some(client_order_ids) => {
4755 self.get_orders_for_ids(client_order_ids.iter().copied(), None)
4756 }
4757 None => Vec::new(),
4758 }
4759 }
4760
4761 #[must_use]
4763 pub fn order_exists(&self, client_order_id: &ClientOrderId) -> bool {
4764 self.index.orders.contains(client_order_id)
4765 }
4766
4767 #[must_use]
4769 pub fn is_order_open(&self, client_order_id: &ClientOrderId) -> bool {
4770 self.index.orders_open.contains(client_order_id)
4771 }
4772
4773 #[must_use]
4775 pub fn is_order_closed(&self, client_order_id: &ClientOrderId) -> bool {
4776 self.index.orders_closed.contains(client_order_id)
4777 }
4778
4779 #[must_use]
4784 pub fn is_order_active_local(&self, client_order_id: &ClientOrderId) -> bool {
4785 self.index.orders_active_local.contains(client_order_id)
4786 }
4787
4788 #[must_use]
4790 pub fn is_order_emulated(&self, client_order_id: &ClientOrderId) -> bool {
4791 self.index.orders_emulated.contains(client_order_id)
4792 }
4793
4794 #[must_use]
4796 pub fn is_order_inflight(&self, client_order_id: &ClientOrderId) -> bool {
4797 self.index.orders_inflight.contains(client_order_id)
4798 }
4799
4800 #[must_use]
4802 pub fn is_order_pending_cancel_local(&self, client_order_id: &ClientOrderId) -> bool {
4803 self.index.orders_pending_cancel.contains(client_order_id)
4804 }
4805
4806 #[must_use]
4808 pub fn orders_open_count(
4809 &self,
4810 venue: Option<&Venue>,
4811 instrument_id: Option<&InstrumentId>,
4812 strategy_id: Option<&StrategyId>,
4813 account_id: Option<&AccountId>,
4814 side: Option<OrderSide>,
4815 ) -> usize {
4816 self.count_orders_in_bucket(
4817 &self.index.orders_open,
4818 venue,
4819 instrument_id,
4820 strategy_id,
4821 account_id,
4822 side,
4823 )
4824 }
4825
4826 #[must_use]
4828 pub fn orders_closed_count(
4829 &self,
4830 venue: Option<&Venue>,
4831 instrument_id: Option<&InstrumentId>,
4832 strategy_id: Option<&StrategyId>,
4833 account_id: Option<&AccountId>,
4834 side: Option<OrderSide>,
4835 ) -> usize {
4836 self.count_orders_in_bucket(
4837 &self.index.orders_closed,
4838 venue,
4839 instrument_id,
4840 strategy_id,
4841 account_id,
4842 side,
4843 )
4844 }
4845
4846 #[must_use]
4851 pub fn orders_active_local_count(
4852 &self,
4853 venue: Option<&Venue>,
4854 instrument_id: Option<&InstrumentId>,
4855 strategy_id: Option<&StrategyId>,
4856 account_id: Option<&AccountId>,
4857 side: Option<OrderSide>,
4858 ) -> usize {
4859 self.count_orders_in_bucket(
4860 &self.index.orders_active_local,
4861 venue,
4862 instrument_id,
4863 strategy_id,
4864 account_id,
4865 side,
4866 )
4867 }
4868
4869 #[must_use]
4871 pub fn orders_emulated_count(
4872 &self,
4873 venue: Option<&Venue>,
4874 instrument_id: Option<&InstrumentId>,
4875 strategy_id: Option<&StrategyId>,
4876 account_id: Option<&AccountId>,
4877 side: Option<OrderSide>,
4878 ) -> usize {
4879 self.count_orders_in_bucket(
4880 &self.index.orders_emulated,
4881 venue,
4882 instrument_id,
4883 strategy_id,
4884 account_id,
4885 side,
4886 )
4887 }
4888
4889 #[must_use]
4891 pub fn orders_inflight_count(
4892 &self,
4893 venue: Option<&Venue>,
4894 instrument_id: Option<&InstrumentId>,
4895 strategy_id: Option<&StrategyId>,
4896 account_id: Option<&AccountId>,
4897 side: Option<OrderSide>,
4898 ) -> usize {
4899 self.count_orders_in_bucket(
4900 &self.index.orders_inflight,
4901 venue,
4902 instrument_id,
4903 strategy_id,
4904 account_id,
4905 side,
4906 )
4907 }
4908
4909 #[must_use]
4911 pub fn orders_total_count(
4912 &self,
4913 venue: Option<&Venue>,
4914 instrument_id: Option<&InstrumentId>,
4915 strategy_id: Option<&StrategyId>,
4916 account_id: Option<&AccountId>,
4917 side: Option<OrderSide>,
4918 ) -> usize {
4919 self.count_orders_in_bucket(
4920 &self.index.orders,
4921 venue,
4922 instrument_id,
4923 strategy_id,
4924 account_id,
4925 side,
4926 )
4927 }
4928
4929 #[must_use]
4935 pub fn has_orders_open(
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 ) -> bool {
4943 self.any_orders_in_bucket(
4944 &self.index.orders_open,
4945 venue,
4946 instrument_id,
4947 strategy_id,
4948 account_id,
4949 side,
4950 )
4951 }
4952
4953 #[must_use]
4955 pub fn has_orders_closed(
4956 &self,
4957 venue: Option<&Venue>,
4958 instrument_id: Option<&InstrumentId>,
4959 strategy_id: Option<&StrategyId>,
4960 account_id: Option<&AccountId>,
4961 side: Option<OrderSide>,
4962 ) -> bool {
4963 self.any_orders_in_bucket(
4964 &self.index.orders_closed,
4965 venue,
4966 instrument_id,
4967 strategy_id,
4968 account_id,
4969 side,
4970 )
4971 }
4972
4973 #[must_use]
4977 pub fn has_orders_active_local(
4978 &self,
4979 venue: Option<&Venue>,
4980 instrument_id: Option<&InstrumentId>,
4981 strategy_id: Option<&StrategyId>,
4982 account_id: Option<&AccountId>,
4983 side: Option<OrderSide>,
4984 ) -> bool {
4985 self.any_orders_in_bucket(
4986 &self.index.orders_active_local,
4987 venue,
4988 instrument_id,
4989 strategy_id,
4990 account_id,
4991 side,
4992 )
4993 }
4994
4995 #[must_use]
4997 pub fn has_orders_emulated(
4998 &self,
4999 venue: Option<&Venue>,
5000 instrument_id: Option<&InstrumentId>,
5001 strategy_id: Option<&StrategyId>,
5002 account_id: Option<&AccountId>,
5003 side: Option<OrderSide>,
5004 ) -> bool {
5005 self.any_orders_in_bucket(
5006 &self.index.orders_emulated,
5007 venue,
5008 instrument_id,
5009 strategy_id,
5010 account_id,
5011 side,
5012 )
5013 }
5014
5015 #[must_use]
5017 pub fn has_orders_inflight(
5018 &self,
5019 venue: Option<&Venue>,
5020 instrument_id: Option<&InstrumentId>,
5021 strategy_id: Option<&StrategyId>,
5022 account_id: Option<&AccountId>,
5023 side: Option<OrderSide>,
5024 ) -> bool {
5025 self.any_orders_in_bucket(
5026 &self.index.orders_inflight,
5027 venue,
5028 instrument_id,
5029 strategy_id,
5030 account_id,
5031 side,
5032 )
5033 }
5034
5035 #[must_use]
5037 pub fn has_orders(
5038 &self,
5039 venue: Option<&Venue>,
5040 instrument_id: Option<&InstrumentId>,
5041 strategy_id: Option<&StrategyId>,
5042 account_id: Option<&AccountId>,
5043 side: Option<OrderSide>,
5044 ) -> bool {
5045 self.any_orders_in_bucket(
5046 &self.index.orders,
5047 venue,
5048 instrument_id,
5049 strategy_id,
5050 account_id,
5051 side,
5052 )
5053 }
5054
5055 #[must_use]
5057 pub fn order_list(&self, order_list_id: &OrderListId) -> Option<&OrderList> {
5058 self.order_lists.get(order_list_id)
5059 }
5060
5061 pub fn try_order_list(
5067 &self,
5068 order_list_id: &OrderListId,
5069 ) -> Result<&OrderList, OrderListLookupError> {
5070 self.order_lists
5071 .get(order_list_id)
5072 .ok_or_else(|| OrderListLookupError::not_found(*order_list_id))
5073 }
5074
5075 #[must_use]
5077 pub fn order_lists(
5078 &self,
5079 venue: Option<&Venue>,
5080 instrument_id: Option<&InstrumentId>,
5081 strategy_id: Option<&StrategyId>,
5082 account_id: Option<&AccountId>,
5083 ) -> Vec<&OrderList> {
5084 let mut order_lists = self.order_lists.values().collect::<Vec<&OrderList>>();
5085
5086 if let Some(venue) = venue {
5087 order_lists.retain(|ol| &ol.instrument_id.venue == venue);
5088 }
5089
5090 if let Some(instrument_id) = instrument_id {
5091 order_lists.retain(|ol| &ol.instrument_id == instrument_id);
5092 }
5093
5094 if let Some(strategy_id) = strategy_id {
5095 order_lists.retain(|ol| &ol.strategy_id == strategy_id);
5096 }
5097
5098 if let Some(account_id) = account_id {
5099 order_lists.retain(|ol| {
5100 ol.client_order_ids.iter().any(|client_order_id| {
5101 self.orders.get(client_order_id).is_some_and(|order_cell| {
5102 order_cell.borrow().account_id().as_ref() == Some(account_id)
5103 })
5104 })
5105 });
5106 }
5107
5108 order_lists
5109 }
5110
5111 #[must_use]
5113 pub fn order_list_exists(&self, order_list_id: &OrderListId) -> bool {
5114 self.order_lists.contains_key(order_list_id)
5115 }
5116
5117 #[must_use]
5122 pub fn orders_for_exec_algorithm(
5123 &self,
5124 exec_algorithm_id: &ExecAlgorithmId,
5125 venue: Option<&Venue>,
5126 instrument_id: Option<&InstrumentId>,
5127 strategy_id: Option<&StrategyId>,
5128 account_id: Option<&AccountId>,
5129 side: Option<OrderSide>,
5130 ) -> Vec<OrderRef<'_>> {
5131 let Some(exec_algorithm_order_ids) =
5132 self.index.exec_algorithm_orders.get(exec_algorithm_id)
5133 else {
5134 return Vec::new();
5135 };
5136
5137 let filtered = self.query_orders_in_bucket(
5138 exec_algorithm_order_ids,
5139 venue,
5140 instrument_id,
5141 strategy_id,
5142 account_id,
5143 );
5144 self.get_orders_for_ids(filtered.iter().copied(), side)
5145 }
5146
5147 #[must_use]
5149 pub fn orders_for_exec_spawn(&self, exec_spawn_id: &ClientOrderId) -> Vec<OrderRef<'_>> {
5150 match self.index.exec_spawn_orders.get(exec_spawn_id) {
5151 Some(ids) => self.get_orders_for_ids(ids.iter().copied(), None),
5152 None => Vec::new(),
5153 }
5154 }
5155
5156 #[must_use]
5158 pub fn exec_spawn_total_quantity(
5159 &self,
5160 exec_spawn_id: &ClientOrderId,
5161 active_only: bool,
5162 ) -> Option<Quantity> {
5163 self.exec_spawn_total(exec_spawn_id, active_only, Order::quantity)
5164 }
5165
5166 #[must_use]
5168 pub fn exec_spawn_total_filled_qty(
5169 &self,
5170 exec_spawn_id: &ClientOrderId,
5171 active_only: bool,
5172 ) -> Option<Quantity> {
5173 self.exec_spawn_total(exec_spawn_id, active_only, Order::filled_qty)
5174 }
5175
5176 #[must_use]
5178 pub fn exec_spawn_total_leaves_qty(
5179 &self,
5180 exec_spawn_id: &ClientOrderId,
5181 active_only: bool,
5182 ) -> Option<Quantity> {
5183 self.exec_spawn_total(exec_spawn_id, active_only, Order::leaves_qty)
5184 }
5185
5186 fn exec_spawn_total(
5187 &self,
5188 exec_spawn_id: &ClientOrderId,
5189 active_only: bool,
5190 quantity: impl Fn(&OrderAny) -> Quantity,
5191 ) -> Option<Quantity> {
5192 self.orders_for_exec_spawn(exec_spawn_id)
5193 .into_iter()
5194 .filter(|order| !active_only || !order.is_closed())
5195 .map(|order| quantity(&order))
5196 .reduce(|total, quantity| total + quantity)
5197 }
5198
5199 #[must_use]
5203 pub fn position_ref(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
5204 self.positions
5205 .get(position_id)
5206 .map(|position_cell| PositionRef::new(position_cell.borrow()))
5207 }
5208
5209 #[must_use]
5213 pub fn position(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
5214 self.position_ref(position_id)
5215 }
5216
5217 pub fn try_position_ref(
5223 &self,
5224 position_id: &PositionId,
5225 ) -> Result<PositionRef<'_>, PositionLookupError> {
5226 self.positions
5227 .get(position_id)
5228 .map(|position_cell| PositionRef::new(position_cell.borrow()))
5229 .ok_or_else(|| PositionLookupError::not_found(*position_id))
5230 }
5231
5232 pub fn try_position(
5240 &self,
5241 position_id: &PositionId,
5242 ) -> Result<PositionRef<'_>, PositionLookupError> {
5243 self.try_position_ref(position_id)
5244 }
5245
5246 #[must_use]
5256 pub fn position_mut(&mut self, position_id: &PositionId) -> Option<PositionRefMut<'_>> {
5257 self.positions
5258 .get(position_id)
5259 .map(|position_cell| PositionRefMut::new(position_cell.borrow_mut()))
5260 }
5261
5262 #[must_use]
5267 pub fn position_owned(&self, position_id: &PositionId) -> Option<Position> {
5268 self.positions
5269 .get(position_id)
5270 .map(|position_cell| position_cell.borrow().clone())
5271 }
5272
5273 #[must_use]
5275 pub fn position_for_order_ref(
5276 &self,
5277 client_order_id: &ClientOrderId,
5278 ) -> Option<PositionRef<'_>> {
5279 self.index
5280 .order_position
5281 .get(client_order_id)
5282 .and_then(|position_id| self.positions.get(position_id))
5283 .map(|position_cell| PositionRef::new(position_cell.borrow()))
5284 }
5285
5286 #[must_use]
5290 pub fn position_for_order(&self, client_order_id: &ClientOrderId) -> Option<PositionRef<'_>> {
5291 self.position_for_order_ref(client_order_id)
5292 }
5293
5294 #[must_use]
5296 pub fn position_id(&self, client_order_id: &ClientOrderId) -> Option<&PositionId> {
5297 self.index.order_position.get(client_order_id)
5298 }
5299
5300 #[must_use]
5306 pub fn positions_refs(
5307 &self,
5308 venue: Option<&Venue>,
5309 instrument_id: Option<&InstrumentId>,
5310 strategy_id: Option<&StrategyId>,
5311 account_id: Option<&AccountId>,
5312 side: Option<PositionSide>,
5313 ) -> Vec<PositionRef<'_>> {
5314 let position_ids = self.position_ids(venue, instrument_id, strategy_id, account_id);
5315 self.get_positions_for_ids(&position_ids, side)
5316 }
5317
5318 #[must_use]
5322 pub fn positions(
5323 &self,
5324 venue: Option<&Venue>,
5325 instrument_id: Option<&InstrumentId>,
5326 strategy_id: Option<&StrategyId>,
5327 account_id: Option<&AccountId>,
5328 side: Option<PositionSide>,
5329 ) -> Vec<PositionRef<'_>> {
5330 self.positions_refs(venue, instrument_id, strategy_id, account_id, side)
5331 }
5332
5333 #[must_use]
5335 pub fn positions_open_refs(
5336 &self,
5337 venue: Option<&Venue>,
5338 instrument_id: Option<&InstrumentId>,
5339 strategy_id: Option<&StrategyId>,
5340 account_id: Option<&AccountId>,
5341 side: Option<PositionSide>,
5342 ) -> Vec<PositionRef<'_>> {
5343 let position_ids = self.position_open_ids(venue, instrument_id, strategy_id, account_id);
5344 self.get_positions_for_ids(&position_ids, side)
5345 }
5346
5347 #[must_use]
5351 pub fn positions_open(
5352 &self,
5353 venue: Option<&Venue>,
5354 instrument_id: Option<&InstrumentId>,
5355 strategy_id: Option<&StrategyId>,
5356 account_id: Option<&AccountId>,
5357 side: Option<PositionSide>,
5358 ) -> Vec<PositionRef<'_>> {
5359 self.positions_open_refs(venue, instrument_id, strategy_id, account_id, side)
5360 }
5361
5362 #[must_use]
5364 pub fn positions_closed_refs(
5365 &self,
5366 venue: Option<&Venue>,
5367 instrument_id: Option<&InstrumentId>,
5368 strategy_id: Option<&StrategyId>,
5369 account_id: Option<&AccountId>,
5370 side: Option<PositionSide>,
5371 ) -> Vec<PositionRef<'_>> {
5372 let position_ids = self.position_closed_ids(venue, instrument_id, strategy_id, account_id);
5373 self.get_positions_for_ids(&position_ids, side)
5374 }
5375
5376 #[must_use]
5380 pub fn positions_closed(
5381 &self,
5382 venue: Option<&Venue>,
5383 instrument_id: Option<&InstrumentId>,
5384 strategy_id: Option<&StrategyId>,
5385 account_id: Option<&AccountId>,
5386 side: Option<PositionSide>,
5387 ) -> Vec<PositionRef<'_>> {
5388 self.positions_closed_refs(venue, instrument_id, strategy_id, account_id, side)
5389 }
5390
5391 #[must_use]
5393 pub fn position_exists(&self, position_id: &PositionId) -> bool {
5394 self.index.positions.contains(position_id)
5395 }
5396
5397 #[must_use]
5399 pub fn is_position_open(&self, position_id: &PositionId) -> bool {
5400 self.index.positions_open.contains(position_id)
5401 }
5402
5403 #[must_use]
5405 pub fn is_position_closed(&self, position_id: &PositionId) -> bool {
5406 self.index.positions_closed.contains(position_id)
5407 }
5408
5409 #[must_use]
5411 pub fn positions_open_count(
5412 &self,
5413 venue: Option<&Venue>,
5414 instrument_id: Option<&InstrumentId>,
5415 strategy_id: Option<&StrategyId>,
5416 account_id: Option<&AccountId>,
5417 side: Option<PositionSide>,
5418 ) -> usize {
5419 self.count_positions_in_bucket(
5420 &self.index.positions_open,
5421 venue,
5422 instrument_id,
5423 strategy_id,
5424 account_id,
5425 side,
5426 )
5427 }
5428
5429 #[must_use]
5431 pub fn positions_closed_count(
5432 &self,
5433 venue: Option<&Venue>,
5434 instrument_id: Option<&InstrumentId>,
5435 strategy_id: Option<&StrategyId>,
5436 account_id: Option<&AccountId>,
5437 side: Option<PositionSide>,
5438 ) -> usize {
5439 self.count_positions_in_bucket(
5440 &self.index.positions_closed,
5441 venue,
5442 instrument_id,
5443 strategy_id,
5444 account_id,
5445 side,
5446 )
5447 }
5448
5449 #[must_use]
5451 pub fn positions_total_count(
5452 &self,
5453 venue: Option<&Venue>,
5454 instrument_id: Option<&InstrumentId>,
5455 strategy_id: Option<&StrategyId>,
5456 account_id: Option<&AccountId>,
5457 side: Option<PositionSide>,
5458 ) -> usize {
5459 self.count_positions_in_bucket(
5460 &self.index.positions,
5461 venue,
5462 instrument_id,
5463 strategy_id,
5464 account_id,
5465 side,
5466 )
5467 }
5468
5469 #[must_use]
5475 pub fn has_positions_open(
5476 &self,
5477 venue: Option<&Venue>,
5478 instrument_id: Option<&InstrumentId>,
5479 strategy_id: Option<&StrategyId>,
5480 account_id: Option<&AccountId>,
5481 side: Option<PositionSide>,
5482 ) -> bool {
5483 self.any_positions_in_bucket(
5484 &self.index.positions_open,
5485 venue,
5486 instrument_id,
5487 strategy_id,
5488 account_id,
5489 side,
5490 )
5491 }
5492
5493 #[must_use]
5495 pub fn has_positions_closed(
5496 &self,
5497 venue: Option<&Venue>,
5498 instrument_id: Option<&InstrumentId>,
5499 strategy_id: Option<&StrategyId>,
5500 account_id: Option<&AccountId>,
5501 side: Option<PositionSide>,
5502 ) -> bool {
5503 self.any_positions_in_bucket(
5504 &self.index.positions_closed,
5505 venue,
5506 instrument_id,
5507 strategy_id,
5508 account_id,
5509 side,
5510 )
5511 }
5512
5513 #[must_use]
5515 pub fn has_positions(
5516 &self,
5517 venue: Option<&Venue>,
5518 instrument_id: Option<&InstrumentId>,
5519 strategy_id: Option<&StrategyId>,
5520 account_id: Option<&AccountId>,
5521 side: Option<PositionSide>,
5522 ) -> bool {
5523 self.any_positions_in_bucket(
5524 &self.index.positions,
5525 venue,
5526 instrument_id,
5527 strategy_id,
5528 account_id,
5529 side,
5530 )
5531 }
5532
5533 #[must_use]
5537 pub fn strategy_id_for_order(&self, client_order_id: &ClientOrderId) -> Option<&StrategyId> {
5538 self.index.order_strategy.get(client_order_id)
5539 }
5540
5541 #[must_use]
5543 pub fn strategy_id_for_position(&self, position_id: &PositionId) -> Option<&StrategyId> {
5544 self.index.position_strategy.get(position_id)
5545 }
5546
5547 pub fn get(&self, key: &str) -> anyhow::Result<Option<&Bytes>> {
5555 check_valid_string_ascii(key, stringify!(key))?;
5556
5557 Ok(self.general.get(key))
5558 }
5559
5560 #[must_use]
5569 pub fn price(&self, instrument_id: &InstrumentId, price_type: PriceType) -> Option<Price> {
5570 match price_type {
5571 PriceType::Bid => self
5572 .quotes
5573 .get(instrument_id)
5574 .and_then(|quotes| quotes.front().map(|quote| quote.bid_price)),
5575 PriceType::Ask => self
5576 .quotes
5577 .get(instrument_id)
5578 .and_then(|quotes| quotes.front().map(|quote| quote.ask_price)),
5579 PriceType::Mid => self.quotes.get(instrument_id).and_then(|quotes| {
5580 quotes.front().map(|quote| {
5581 let mid = (quote.ask_price.as_decimal() + quote.bid_price.as_decimal())
5582 / Decimal::TWO;
5583
5584 Price::from_decimal_dp(mid, quote.bid_price.precision + 1)
5585 .expect("Invalid mid price for Cache::price")
5586 })
5587 }),
5588 PriceType::Last => self
5589 .trades
5590 .get(instrument_id)
5591 .and_then(|trades| trades.front().map(|trade| trade.price)),
5592 PriceType::Mark => self
5593 .mark_prices
5594 .get(instrument_id)
5595 .and_then(|marks| marks.front().map(|mark| mark.value)),
5596 }
5597 }
5598
5599 #[must_use]
5601 pub fn quotes(&self, instrument_id: &InstrumentId) -> Option<Vec<QuoteTick>> {
5602 self.quotes
5603 .get(instrument_id)
5604 .map(|quotes| quotes.iter().copied().collect())
5605 }
5606
5607 #[must_use]
5609 pub fn trades(&self, instrument_id: &InstrumentId) -> Option<Vec<TradeTick>> {
5610 self.trades
5611 .get(instrument_id)
5612 .map(|trades| trades.iter().copied().collect())
5613 }
5614
5615 #[must_use]
5617 pub fn mark_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<MarkPriceUpdate>> {
5618 self.mark_prices
5619 .get(instrument_id)
5620 .map(|mark_prices| mark_prices.iter().copied().collect())
5621 }
5622
5623 #[must_use]
5625 pub fn index_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<IndexPriceUpdate>> {
5626 self.index_prices
5627 .get(instrument_id)
5628 .map(|index_prices| index_prices.iter().copied().collect())
5629 }
5630
5631 #[must_use]
5633 pub fn funding_rates(&self, instrument_id: &InstrumentId) -> Option<Vec<FundingRateUpdate>> {
5634 self.funding_rates
5635 .get(instrument_id)
5636 .map(|funding_rates| funding_rates.iter().copied().collect())
5637 }
5638
5639 #[must_use]
5641 pub fn instrument_statuses(
5642 &self,
5643 instrument_id: &InstrumentId,
5644 ) -> Option<Vec<InstrumentStatus>> {
5645 self.instrument_statuses
5646 .get(instrument_id)
5647 .map(|statuses| statuses.iter().copied().collect())
5648 }
5649
5650 #[must_use]
5652 pub fn bars(&self, bar_type: &BarType) -> Option<Vec<Bar>> {
5653 self.bars
5654 .get(bar_type)
5655 .map(|bars| bars.iter().copied().collect())
5656 }
5657
5658 #[must_use]
5660 pub fn order_book(&self, instrument_id: &InstrumentId) -> Option<&OrderBook> {
5661 self.books.get(instrument_id)
5662 }
5663
5664 pub fn try_order_book(
5670 &self,
5671 instrument_id: &InstrumentId,
5672 ) -> Result<&OrderBook, OrderBookLookupError> {
5673 self.books
5674 .get(instrument_id)
5675 .ok_or_else(|| OrderBookLookupError::not_found(*instrument_id))
5676 }
5677
5678 #[must_use]
5680 pub fn order_book_mut(&mut self, instrument_id: &InstrumentId) -> Option<&mut OrderBook> {
5681 self.books.get_mut(instrument_id)
5682 }
5683
5684 #[must_use]
5686 pub fn own_order_book(&self, instrument_id: &InstrumentId) -> Option<&OwnOrderBook> {
5687 self.own_books.get(instrument_id)
5688 }
5689
5690 pub fn try_own_order_book(
5697 &self,
5698 instrument_id: &InstrumentId,
5699 ) -> Result<&OwnOrderBook, OwnOrderBookLookupError> {
5700 self.own_books
5701 .get(instrument_id)
5702 .ok_or_else(|| OwnOrderBookLookupError::not_found(*instrument_id))
5703 }
5704
5705 #[must_use]
5707 pub fn own_order_book_mut(
5708 &mut self,
5709 instrument_id: &InstrumentId,
5710 ) -> Option<&mut OwnOrderBook> {
5711 self.own_books.get_mut(instrument_id)
5712 }
5713
5714 #[must_use]
5716 pub fn quote(&self, instrument_id: &InstrumentId) -> Option<&QuoteTick> {
5717 self.quotes
5718 .get(instrument_id)
5719 .and_then(|quotes| quotes.front())
5720 }
5721
5722 #[must_use]
5726 pub fn quote_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&QuoteTick> {
5727 self.quotes
5728 .get(instrument_id)
5729 .and_then(|quotes| quotes.get(index))
5730 }
5731
5732 #[must_use]
5734 pub fn trade(&self, instrument_id: &InstrumentId) -> Option<&TradeTick> {
5735 self.trades
5736 .get(instrument_id)
5737 .and_then(|trades| trades.front())
5738 }
5739
5740 #[must_use]
5744 pub fn trade_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&TradeTick> {
5745 self.trades
5746 .get(instrument_id)
5747 .and_then(|trades| trades.get(index))
5748 }
5749
5750 #[must_use]
5752 pub fn mark_price(&self, instrument_id: &InstrumentId) -> Option<&MarkPriceUpdate> {
5753 self.mark_prices
5754 .get(instrument_id)
5755 .and_then(|mark_prices| mark_prices.front())
5756 }
5757
5758 #[must_use]
5760 pub fn index_price(&self, instrument_id: &InstrumentId) -> Option<&IndexPriceUpdate> {
5761 self.index_prices
5762 .get(instrument_id)
5763 .and_then(|index_prices| index_prices.front())
5764 }
5765
5766 #[must_use]
5768 pub fn funding_rate(&self, instrument_id: &InstrumentId) -> Option<&FundingRateUpdate> {
5769 self.funding_rates
5770 .get(instrument_id)
5771 .and_then(|funding_rates| funding_rates.front())
5772 }
5773
5774 #[must_use]
5776 pub fn instrument_status(&self, instrument_id: &InstrumentId) -> Option<&InstrumentStatus> {
5777 self.instrument_statuses
5778 .get(instrument_id)
5779 .and_then(|statuses| statuses.front())
5780 }
5781
5782 #[must_use]
5784 pub fn instrument_close(&self, instrument_id: &InstrumentId) -> Option<&InstrumentClose> {
5785 self.instrument_closes.get(instrument_id)
5786 }
5787
5788 #[must_use]
5790 pub fn instrument_close_ids(&self) -> Vec<&InstrumentId> {
5791 self.instrument_closes.keys().collect()
5792 }
5793
5794 #[must_use]
5796 pub fn bar(&self, bar_type: &BarType) -> Option<&Bar> {
5797 self.bars.get(bar_type).and_then(|bars| bars.front())
5798 }
5799
5800 #[must_use]
5804 pub fn bar_at_index(&self, bar_type: &BarType, index: usize) -> Option<&Bar> {
5805 self.bars.get(bar_type).and_then(|bars| bars.get(index))
5806 }
5807
5808 #[must_use]
5810 pub fn book_update_count(&self, instrument_id: &InstrumentId) -> usize {
5811 self.books
5812 .get(instrument_id)
5813 .map_or(0, |book| book.update_count) as usize
5814 }
5815
5816 #[must_use]
5818 pub fn quote_count(&self, instrument_id: &InstrumentId) -> usize {
5819 self.quotes
5820 .get(instrument_id)
5821 .map_or(0, BoundedVecDeque::len)
5822 }
5823
5824 #[must_use]
5826 pub fn trade_count(&self, instrument_id: &InstrumentId) -> usize {
5827 self.trades
5828 .get(instrument_id)
5829 .map_or(0, BoundedVecDeque::len)
5830 }
5831
5832 #[must_use]
5834 pub fn mark_price_count(&self, instrument_id: &InstrumentId) -> usize {
5835 self.mark_prices
5836 .get(instrument_id)
5837 .map_or(0, BoundedVecDeque::len)
5838 }
5839
5840 #[must_use]
5842 pub fn index_price_count(&self, instrument_id: &InstrumentId) -> usize {
5843 self.index_prices
5844 .get(instrument_id)
5845 .map_or(0, BoundedVecDeque::len)
5846 }
5847
5848 #[must_use]
5850 pub fn funding_rate_count(&self, instrument_id: &InstrumentId) -> usize {
5851 self.funding_rates
5852 .get(instrument_id)
5853 .map_or(0, BoundedVecDeque::len)
5854 }
5855
5856 #[must_use]
5858 pub fn instrument_status_count(&self, instrument_id: &InstrumentId) -> usize {
5859 self.instrument_statuses
5860 .get(instrument_id)
5861 .map_or(0, BoundedVecDeque::len)
5862 }
5863
5864 #[must_use]
5866 pub fn bar_count(&self, bar_type: &BarType) -> usize {
5867 self.bars.get(bar_type).map_or(0, BoundedVecDeque::len)
5868 }
5869
5870 #[must_use]
5872 pub fn has_order_book(&self, instrument_id: &InstrumentId) -> bool {
5873 self.books.contains_key(instrument_id)
5874 }
5875
5876 #[must_use]
5878 pub fn has_quote_ticks(&self, instrument_id: &InstrumentId) -> bool {
5879 self.quote_count(instrument_id) > 0
5880 }
5881
5882 #[must_use]
5884 pub fn has_trade_ticks(&self, instrument_id: &InstrumentId) -> bool {
5885 self.trade_count(instrument_id) > 0
5886 }
5887
5888 #[must_use]
5890 pub fn has_mark_prices(&self, instrument_id: &InstrumentId) -> bool {
5891 self.mark_price_count(instrument_id) > 0
5892 }
5893
5894 #[must_use]
5896 pub fn has_index_prices(&self, instrument_id: &InstrumentId) -> bool {
5897 self.index_price_count(instrument_id) > 0
5898 }
5899
5900 #[must_use]
5902 pub fn has_funding_rates(&self, instrument_id: &InstrumentId) -> bool {
5903 self.funding_rate_count(instrument_id) > 0
5904 }
5905
5906 #[must_use]
5908 pub fn has_instrument_statuses(&self, instrument_id: &InstrumentId) -> bool {
5909 self.instrument_status_count(instrument_id) > 0
5910 }
5911
5912 #[must_use]
5914 pub fn has_instrument_close(&self, instrument_id: &InstrumentId) -> bool {
5915 self.instrument_closes.contains_key(instrument_id)
5916 }
5917
5918 #[must_use]
5920 pub fn has_bars(&self, bar_type: &BarType) -> bool {
5921 self.bar_count(bar_type) > 0
5922 }
5923
5924 #[must_use]
5925 pub fn get_xrate(
5926 &self,
5927 venue: Venue,
5928 from_currency: Currency,
5929 to_currency: Currency,
5930 price_type: PriceType,
5931 ) -> Option<Decimal> {
5932 match self.try_get_xrate(venue, from_currency, to_currency, price_type) {
5933 Ok(rate) => rate,
5934 Err(e) => {
5935 log::error!("Failed to calculate xrate: {e}");
5936 None
5937 }
5938 }
5939 }
5940
5941 pub fn try_get_xrate(
5948 &self,
5949 venue: Venue,
5950 from_currency: Currency,
5951 to_currency: Currency,
5952 price_type: PriceType,
5953 ) -> anyhow::Result<Option<Decimal>> {
5954 if from_currency == to_currency {
5955 return Ok(Some(Decimal::ONE));
5958 }
5959
5960 let (bid_quote, ask_quote) = self.build_quote_table(&venue);
5961
5962 get_exchange_rate(
5963 from_currency.code,
5964 to_currency.code,
5965 price_type,
5966 bid_quote,
5967 ask_quote,
5968 )
5969 }
5970
5971 fn build_quote_table(
5972 &self,
5973 venue: &Venue,
5974 ) -> (AHashMap<Ustr, Decimal>, AHashMap<Ustr, Decimal>) {
5975 let mut bid_quotes = AHashMap::new();
5976 let mut ask_quotes = AHashMap::new();
5977 let mut quote_sources = AHashMap::new();
5978 let mut bar_quotes = None;
5979 let mut pair_buffer = String::new();
5980
5981 for (instrument_id, instrument) in &self.instruments {
5982 if instrument_id.venue != *venue {
5983 continue;
5984 }
5985
5986 let Some(base_currency) = instrument.base_currency() else {
5987 continue;
5988 };
5989
5990 let (bid_price, ask_price) = if let Some(ticks) = self.quotes.get(instrument_id) {
5991 if let Some(tick) = ticks.front() {
5992 (tick.bid_price, tick.ask_price)
5993 } else {
5994 continue; }
5996 } else {
5997 let quotes = bar_quotes.get_or_insert_with(|| self.build_bar_quote_table(venue));
5998 match (
5999 quotes.get(&(*instrument_id, PriceType::Bid)),
6000 quotes.get(&(*instrument_id, PriceType::Ask)),
6001 ) {
6002 (Some((_, bid_bar)), Some((_, ask_bar))) => (bid_bar.close, ask_bar.close),
6003 _ => continue,
6004 }
6005 };
6006
6007 let base = base_currency.code.as_str();
6008 let quote = instrument.quote_currency().code.as_str();
6009 pair_buffer.clear();
6010 pair_buffer.reserve(base.len() + 1 + quote.len());
6011 pair_buffer.push_str(base);
6012 pair_buffer.push('/');
6013 pair_buffer.push_str(quote);
6014 let pair = Ustr::from(pair_buffer.as_str());
6015 let preference = (
6016 bid_price.is_positive() && ask_price.is_positive(),
6017 instrument.instrument_class() == InstrumentClass::Spot,
6018 Reverse(*instrument_id),
6019 );
6020
6021 if quote_sources
6022 .get(&pair)
6023 .is_some_and(|current| current >= &preference)
6024 {
6025 continue;
6026 }
6027
6028 bid_quotes.insert(pair, bid_price.as_decimal());
6029 ask_quotes.insert(pair, ask_price.as_decimal());
6030 quote_sources.insert(pair, preference);
6031 }
6032
6033 (bid_quotes, ask_quotes)
6034 }
6035
6036 fn build_bar_quote_table(
6037 &self,
6038 venue: &Venue,
6039 ) -> AHashMap<(InstrumentId, PriceType), (&BarType, &Bar)> {
6040 let mut quotes: AHashMap<_, (&BarType, &Bar)> = AHashMap::new();
6041
6042 for (bar_type, bars) in &self.bars {
6043 let instrument_id = bar_type.instrument_id();
6044 let price_type = bar_type.spec().price_type;
6045
6046 if instrument_id.venue != *venue
6047 || !matches!(price_type, PriceType::Bid | PriceType::Ask)
6048 {
6049 continue;
6050 }
6051
6052 let Some(bar) = bars.front() else {
6053 continue;
6054 };
6055
6056 quotes
6058 .entry((instrument_id, price_type))
6059 .and_modify(|current| {
6060 if (current.1.ts_init, current.0) < (bar.ts_init, bar_type) {
6061 *current = (bar_type, bar);
6062 }
6063 })
6064 .or_insert((bar_type, bar));
6065 }
6066
6067 quotes
6068 }
6069
6070 #[must_use]
6072 pub fn get_mark_xrate(&self, from_currency: Currency, to_currency: Currency) -> Option<f64> {
6073 self.mark_xrates.get(&(from_currency, to_currency)).copied()
6074 }
6075
6076 pub fn set_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency, xrate: f64) {
6082 assert!(xrate > 0.0, "xrate was zero");
6083 self.mark_xrates.insert((from_currency, to_currency), xrate);
6084 self.mark_xrates
6085 .insert((to_currency, from_currency), 1.0 / xrate);
6086 }
6087
6088 pub fn clear_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency) {
6094 let _ = self.mark_xrates.remove(&(from_currency, to_currency));
6095 }
6096
6097 pub fn clear_mark_xrates(&mut self) {
6099 self.mark_xrates.clear();
6100 }
6101
6102 #[must_use]
6104 pub fn currency(&self, code: &Ustr) -> Option<&Currency> {
6105 self.currencies.get(code)
6106 }
6107
6108 pub fn try_currency(&self, code: &Ustr) -> Result<&Currency, CurrencyLookupError> {
6114 self.currencies
6115 .get(code)
6116 .ok_or_else(|| CurrencyLookupError::not_found(*code))
6117 }
6118
6119 #[must_use]
6123 pub fn instrument(&self, instrument_id: &InstrumentId) -> Option<&InstrumentAny> {
6124 self.instruments.get(instrument_id)
6125 }
6126
6127 pub fn try_instrument(
6133 &self,
6134 instrument_id: &InstrumentId,
6135 ) -> Result<&InstrumentAny, InstrumentLookupError> {
6136 self.instruments
6137 .get(instrument_id)
6138 .ok_or_else(|| InstrumentLookupError::not_found(*instrument_id))
6139 }
6140
6141 #[must_use]
6143 pub fn instrument_ids(&self, venue: Option<&Venue>) -> Vec<&InstrumentId> {
6144 match venue {
6145 Some(v) => self.instruments.keys().filter(|i| &i.venue == v).collect(),
6146 None => self.instruments.keys().collect(),
6147 }
6148 }
6149
6150 #[must_use]
6152 pub fn instruments(&self, venue: &Venue, underlying: Option<&Ustr>) -> Vec<&InstrumentAny> {
6153 self.instruments
6154 .values()
6155 .filter(|i| &i.id().venue == venue)
6156 .filter(|i| underlying.is_none_or(|u| i.underlying() == Some(*u)))
6157 .collect()
6158 }
6159
6160 #[must_use]
6167 pub fn instruments_by_parent(
6168 &self,
6169 venue: &Venue,
6170 root: &Ustr,
6171 class: InstrumentClass,
6172 ) -> Vec<&InstrumentAny> {
6173 self.instruments
6174 .values()
6175 .filter(|i| &i.id().venue == venue)
6176 .filter(|i| i.underlying() == Some(*root))
6177 .filter(|i| i.instrument_class() == class)
6178 .collect()
6179 }
6180
6181 #[must_use]
6183 pub fn bar_types(
6184 &self,
6185 instrument_id: Option<&InstrumentId>,
6186 price_type: Option<&PriceType>,
6187 aggregation_source: AggregationSource,
6188 ) -> Vec<&BarType> {
6189 let mut bar_types = self
6190 .bars
6191 .keys()
6192 .filter(|bar_type| bar_type.aggregation_source() == aggregation_source)
6193 .collect::<Vec<&BarType>>();
6194
6195 if let Some(instrument_id) = instrument_id {
6196 bar_types.retain(|bar_type| bar_type.instrument_id() == *instrument_id);
6197 }
6198
6199 if let Some(price_type) = price_type {
6200 bar_types.retain(|bar_type| &bar_type.spec().price_type == price_type);
6201 }
6202
6203 bar_types
6204 }
6205
6206 #[must_use]
6210 pub fn synthetic(&self, instrument_id: &InstrumentId) -> Option<&SyntheticInstrument> {
6211 self.synthetics.get(instrument_id)
6212 }
6213
6214 pub fn try_synthetic(
6221 &self,
6222 instrument_id: &InstrumentId,
6223 ) -> Result<&SyntheticInstrument, SyntheticInstrumentLookupError> {
6224 self.synthetics
6225 .get(instrument_id)
6226 .ok_or_else(|| SyntheticInstrumentLookupError::not_found(*instrument_id))
6227 }
6228
6229 #[must_use]
6231 pub fn synthetic_ids(&self) -> Vec<&InstrumentId> {
6232 self.synthetics.keys().collect()
6233 }
6234
6235 #[must_use]
6237 pub fn synthetics(&self) -> Vec<&SyntheticInstrument> {
6238 self.synthetics.values().collect()
6239 }
6240
6241 #[must_use]
6245 pub fn account_ref(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
6246 self.accounts
6247 .get(account_id)
6248 .map(|account_cell| AccountRef::new(account_cell.borrow()))
6249 }
6250
6251 #[must_use]
6255 pub fn account(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
6256 self.account_ref(account_id)
6257 }
6258
6259 pub fn try_account_ref(
6265 &self,
6266 account_id: &AccountId,
6267 ) -> Result<AccountRef<'_>, AccountLookupError> {
6268 self.accounts
6269 .get(account_id)
6270 .map(|account_cell| AccountRef::new(account_cell.borrow()))
6271 .ok_or_else(|| AccountLookupError::not_found(*account_id))
6272 }
6273
6274 pub fn try_account(
6282 &self,
6283 account_id: &AccountId,
6284 ) -> Result<AccountRef<'_>, AccountLookupError> {
6285 self.try_account_ref(account_id)
6286 }
6287
6288 #[must_use]
6298 pub fn account_mut(&mut self, account_id: &AccountId) -> Option<AccountRefMut<'_>> {
6299 self.accounts
6300 .get(account_id)
6301 .map(|account_cell| AccountRefMut::new(account_cell.borrow_mut()))
6302 }
6303
6304 #[must_use]
6310 pub fn account_owned(&self, account_id: &AccountId) -> Option<AccountAny> {
6311 self.accounts.get(account_id).and_then(|account_cell| {
6312 account_cell
6313 .try_borrow()
6314 .ok()
6315 .map(|account| account.clone())
6316 })
6317 }
6318
6319 #[must_use]
6321 pub fn account_for_venue(&self, venue: &Venue) -> Option<AccountRef<'_>> {
6322 self.index
6323 .venue_account
6324 .get(venue)
6325 .and_then(|account_id| self.accounts.get(account_id))
6326 .map(|account_cell| AccountRef::new(account_cell.borrow()))
6327 }
6328
6329 #[must_use]
6334 pub fn account_for_venue_owned(&self, venue: &Venue) -> Option<AccountAny> {
6335 self.index
6336 .venue_account
6337 .get(venue)
6338 .and_then(|account_id| self.accounts.get(account_id))
6339 .map(|account_cell| account_cell.borrow().clone())
6340 }
6341
6342 #[must_use]
6344 pub fn account_id(&self, venue: &Venue) -> Option<&AccountId> {
6345 self.index.venue_account.get(venue)
6346 }
6347
6348 #[must_use]
6354 pub fn accounts(&self, account_id: &AccountId) -> Vec<AccountRef<'_>> {
6355 self.accounts
6356 .values()
6357 .filter(|account_cell| &account_cell.borrow().id() == account_id)
6358 .map(|account_cell| AccountRef::new(account_cell.borrow()))
6359 .collect()
6360 }
6361
6362 #[must_use]
6364 pub fn accounts_all_owned(&self) -> Vec<AccountAny> {
6365 self.accounts
6366 .values()
6367 .map(|account_cell| account_cell.borrow().clone())
6368 .collect()
6369 }
6370
6371 pub fn update_own_order_book(&mut self, order: &OrderAny) {
6379 if !order.has_price() {
6380 return;
6381 }
6382
6383 let instrument_id = order.instrument_id();
6384
6385 if !self.own_books.contains_key(&instrument_id) {
6386 if order.is_closed() {
6387 return;
6388 }
6389
6390 self.own_books
6391 .insert(instrument_id, OwnOrderBook::new(instrument_id));
6392 }
6393
6394 let Some(own_book) = self.own_books.get_mut(&instrument_id) else {
6395 return;
6396 };
6397
6398 let own_book_order = order.to_own_book_order();
6399
6400 if order.is_closed() {
6401 if let Err(e) = own_book.delete(own_book_order) {
6402 log::debug!(
6403 "Failed to delete order {} from own book: {e}",
6404 order.client_order_id(),
6405 );
6406 } else {
6407 log::debug!("Deleted order {} from own book", order.client_order_id());
6408 }
6409 } else {
6410 if let Err(e) = own_book.update(own_book_order) {
6412 log::debug!(
6413 "Failed to update order {} in own book: {e}; inserting instead",
6414 order.client_order_id(),
6415 );
6416 own_book.add(own_book_order);
6417 }
6418 log::debug!("Updated order {} in own book", order.client_order_id());
6419 }
6420 }
6421
6422 pub fn force_remove_from_own_order_book(&mut self, client_order_id: &ClientOrderId) {
6428 let Some(order_cell) = self.orders.get(client_order_id) else {
6429 return;
6430 };
6431 let order = order_cell.borrow();
6432 let instrument_id = order.instrument_id();
6433 let own_book_order = if order.has_price() {
6434 Some(order.to_own_book_order())
6435 } else {
6436 None
6437 };
6438 drop(order);
6439
6440 self.index.orders_open.remove(client_order_id);
6441 self.index.orders_pending_cancel.remove(client_order_id);
6442 self.index.orders_inflight.remove(client_order_id);
6443 self.index.orders_emulated.remove(client_order_id);
6444 self.index.orders_active_local.remove(client_order_id);
6445
6446 if let Some(own_book) = self.own_books.get_mut(&instrument_id)
6447 && let Some(own_book_order) = own_book_order
6448 {
6449 if let Err(e) = own_book.delete(own_book_order) {
6450 log::debug!("Could not force delete {client_order_id} from own book: {e}");
6451 } else {
6452 log::debug!("Force deleted {client_order_id} from own book");
6453 }
6454 }
6455
6456 self.index.orders_closed.insert(*client_order_id);
6457 }
6458
6459 pub fn audit_own_order_books(&mut self) {
6464 log::debug!("Starting own books audit");
6465 let start = std::time::Instant::now();
6466
6467 let valid_order_ids: AHashSet<ClientOrderId> = self
6468 .index
6469 .orders_open
6470 .iter()
6471 .chain(&self.index.orders_inflight)
6472 .chain(&self.index.orders_active_local)
6473 .copied()
6474 .collect();
6475
6476 for own_book in self.own_books.values_mut() {
6477 own_book.audit_open_orders(&valid_order_ids);
6478 }
6479
6480 log::debug!("Completed own books audit in {:?}", start.elapsed());
6481 }
6482}
6483
6484const POSITION_OMS_KEY_PREFIX: &str = "position_oms:";
6485
6486fn position_oms_key(position_id: PositionId) -> String {
6487 format!("{POSITION_OMS_KEY_PREFIX}{position_id}")
6488}