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::{Ordering, 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, register_loaded_currencies};
52pub use error::{
53 ACCOUNT_NOT_FOUND, AccountLookupError, CURRENCY_NOT_FOUND, CurrencyLookupError,
54 INSTRUMENT_NOT_FOUND, InstrumentLookupError, ORDER_BOOK_NOT_FOUND, ORDER_LIST_NOT_FOUND,
55 ORDER_NOT_FOUND, OWN_ORDER_BOOK_NOT_FOUND, OrderBookLookupError, OrderListLookupError,
56 OrderLookupError, OwnOrderBookLookupError, POSITION_NOT_FOUND, PositionLookupError,
57 SYNTHETIC_INSTRUMENT_NOT_FOUND, SyntheticInstrumentLookupError, VenueOrderIdOwnershipError,
58};
59use filter::{FilterSources, intersect_pair_or_many};
60use index::CacheIndex;
61use indexmap::IndexMap;
62use nautilus_core::{
63 DurationNanos, SharedCell, UnixNanos,
64 correctness::{
65 check_key_not_in_map, check_predicate_false, check_slice_not_empty, check_valid_string_utf8,
66 },
67};
68use nautilus_model::{
69 accounts::{Account, AccountAny},
70 data::{
71 Bar, BarType, FundingRateUpdate, GreeksData, IndexPriceUpdate, InstrumentClose,
72 InstrumentStatus, MarkPriceUpdate, QuoteTick, TradeTick, YieldCurve,
73 option_chain::OptionGreeks,
74 },
75 enums::{
76 AggregationSource, ContingencyType, InstrumentClass, OmsType, OrderSide, PositionSide,
77 PriceType,
78 },
79 events::{AccountState, OrderEventAny, OrderFilled, PositionAdjusted},
80 identifiers::{
81 AccountId, ActorId, ClientId, ClientOrderId, ExecAlgorithmId, InstrumentId, OrderListId,
82 PositionId, StrategyId, Venue, VenueOrderId,
83 },
84 instruments::{Instrument, InstrumentAny, SyntheticInstrument},
85 orderbook::{
86 OrderBook,
87 own::{OwnOrderBook, should_handle_own_book_order},
88 },
89 orders::{Order, OrderAny, OrderError, OrderList},
90 position::{Position, PositionReplayEvent},
91 types::{
92 Currency, Money, Price, Quantity,
93 fixed::{FIXED_PRECISION, MAX_FLOAT_PRECISION, check_fixed_precision},
94 price::PriceRaw,
95 },
96};
97pub use position::CacheSnapshotRef;
98use position::PositionSnapshotFrame;
99pub use refs::{AccountRef, AccountRefMut, OrderRef, OrderRefMut, PositionRef, PositionRefMut};
100use rust_decimal::Decimal;
101use ustr::Ustr;
102pub use view::CacheView; use crate::xrate::get_exchange_rate;
105
106#[cfg_attr(
108 feature = "python",
109 pyo3::pyclass(module = "nautilus_trader.common", unsendable)
110)]
111pub struct Cache {
112 config: CacheConfig,
113 index: CacheIndex,
114 database: Option<Box<dyn CacheDatabaseAdapter>>,
115 general: AHashMap<String, Bytes>,
116 currencies: AHashMap<Ustr, Currency>,
117 instruments: AHashMap<InstrumentId, InstrumentAny>,
118 instrument_closes: AHashMap<InstrumentId, InstrumentClose>,
119 synthetics: AHashMap<InstrumentId, SyntheticInstrument>,
120 books: AHashMap<InstrumentId, OrderBook>,
121 own_books: AHashMap<InstrumentId, OwnOrderBook>,
122 quotes: AHashMap<InstrumentId, BoundedVecDeque<QuoteTick>>,
123 trades: AHashMap<InstrumentId, BoundedVecDeque<TradeTick>>,
124 mark_xrates: AHashMap<(Currency, Currency), f64>,
125 mark_prices: AHashMap<InstrumentId, BoundedVecDeque<MarkPriceUpdate>>,
126 index_prices: AHashMap<InstrumentId, BoundedVecDeque<IndexPriceUpdate>>,
127 funding_rates: AHashMap<InstrumentId, BoundedVecDeque<FundingRateUpdate>>,
128 instrument_statuses: AHashMap<InstrumentId, BoundedVecDeque<InstrumentStatus>>,
129 bars: AHashMap<BarType, BoundedVecDeque<Bar>>,
130 greeks: AHashMap<InstrumentId, GreeksData>,
131 option_greeks: AHashMap<InstrumentId, OptionGreeks>,
132 yield_curves: AHashMap<String, YieldCurve>,
133 external_order_claims: AHashMap<InstrumentId, StrategyId>,
134 client_accounts: AHashMap<ClientId, AccountId>,
135 client_routes: AHashMap<Venue, ClientId>,
136 default_client_id: Option<ClientId>,
137 external_clients: AHashSet<ClientId>,
138 accounts: AHashMap<AccountId, SharedCell<AccountAny>>,
139 orders: AHashMap<ClientOrderId, SharedCell<OrderAny>>,
140 order_lists: AHashMap<OrderListId, OrderList>,
141 positions: AHashMap<PositionId, SharedCell<Position>>,
142 position_snapshots: AHashMap<PositionId, Vec<PositionSnapshotFrame>>,
143 position_snapshot_revisions: AHashMap<PositionId, u64>,
144 #[cfg(feature = "defi")]
145 pub(crate) defi: crate::defi::cache::DefiCache,
146}
147
148impl Debug for Cache {
149 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
150 f.debug_struct(stringify!(Cache))
151 .field("config", &self.config)
152 .field("index", &self.index)
153 .field("general", &self.general)
154 .field("currencies", &self.currencies)
155 .field("instruments", &self.instruments)
156 .field("synthetics", &self.synthetics)
157 .field("books", &self.books)
158 .field("own_books", &self.own_books)
159 .field("quotes", &self.quotes)
160 .field("trades", &self.trades)
161 .field("mark_xrates", &self.mark_xrates)
162 .field("mark_prices", &self.mark_prices)
163 .field("index_prices", &self.index_prices)
164 .field("funding_rates", &self.funding_rates)
165 .field("instrument_statuses", &self.instrument_statuses)
166 .field("instrument_closes", &self.instrument_closes)
167 .field("bars", &self.bars)
168 .field("greeks", &self.greeks)
169 .field("option_greeks", &self.option_greeks)
170 .field("yield_curves", &self.yield_curves)
171 .field("external_order_claims", &self.external_order_claims)
172 .field("client_accounts", &self.client_accounts)
173 .field("client_routes", &self.client_routes)
174 .field("default_client_id", &self.default_client_id)
175 .field("external_clients", &self.external_clients)
176 .field("accounts", &self.accounts)
177 .field("orders", &self.orders)
178 .field("order_lists", &self.order_lists)
179 .field("positions", &self.positions)
180 .field("position_snapshots", &self.position_snapshots)
181 .finish()
182 }
183}
184
185impl Default for Cache {
186 fn default() -> Self {
188 Self::new(Some(CacheConfig::default()), None)
189 }
190}
191
192impl Cache {
193 #[must_use]
195 pub fn new(
203 config: Option<CacheConfig>,
204 database: Option<Box<dyn CacheDatabaseAdapter>>,
205 ) -> Self {
206 Self::try_new(config, database).expect("invalid `CacheConfig`")
207 }
208
209 pub fn try_new(
215 config: Option<CacheConfig>,
216 database: Option<Box<dyn CacheDatabaseAdapter>>,
217 ) -> crate::config::ConfigResult<Self> {
218 let config = config.unwrap_or_default();
219 config.validate()?;
220
221 Ok(Self {
222 config,
223 index: CacheIndex::default(),
224 database,
225 general: AHashMap::new(),
226 currencies: AHashMap::new(),
227 instruments: AHashMap::new(),
228 instrument_closes: AHashMap::new(),
229 synthetics: AHashMap::new(),
230 books: AHashMap::new(),
231 own_books: AHashMap::new(),
232 quotes: AHashMap::new(),
233 trades: AHashMap::new(),
234 mark_xrates: AHashMap::new(),
235 mark_prices: AHashMap::new(),
236 index_prices: AHashMap::new(),
237 funding_rates: AHashMap::new(),
238 instrument_statuses: AHashMap::new(),
239 bars: AHashMap::new(),
240 greeks: AHashMap::new(),
241 option_greeks: AHashMap::new(),
242 yield_curves: AHashMap::new(),
243 external_order_claims: AHashMap::new(),
244 client_accounts: AHashMap::new(),
245 client_routes: AHashMap::new(),
246 default_client_id: None,
247 external_clients: AHashSet::new(),
248 accounts: AHashMap::new(),
249 orders: AHashMap::new(),
250 order_lists: AHashMap::new(),
251 positions: AHashMap::new(),
252 position_snapshots: AHashMap::new(),
253 position_snapshot_revisions: AHashMap::new(),
254 #[cfg(feature = "defi")]
255 defi: crate::defi::cache::DefiCache::default(),
256 })
257 }
258
259 #[must_use]
261 pub fn memory_address(&self) -> String {
262 format!("{:?}", std::ptr::from_ref(self))
263 }
264
265 #[must_use]
267 pub fn external_order_claim(&self, instrument_id: &InstrumentId) -> Option<StrategyId> {
268 self.external_order_claims.get(instrument_id).copied()
269 }
270
271 #[must_use]
276 pub fn external_order_claim_instrument_ids(
277 &self,
278 strategy_id: Option<StrategyId>,
279 ) -> AHashSet<InstrumentId> {
280 self.external_order_claims
281 .iter()
282 .filter_map(|(instrument_id, owner)| {
283 strategy_id
284 .is_none_or(|strategy_id| *owner == strategy_id)
285 .then_some(*instrument_id)
286 })
287 .collect()
288 }
289
290 pub fn set_external_order_claims(
302 &mut self,
303 strategy_id: StrategyId,
304 instrument_ids: &[InstrumentId],
305 ) -> anyhow::Result<()> {
306 let mut requested = AHashSet::with_capacity(instrument_ids.len());
307
308 for instrument_id in instrument_ids {
309 if !requested.insert(*instrument_id) {
310 anyhow::bail!(
311 "External order claim for {instrument_id} appears more than once for {strategy_id}"
312 );
313 }
314
315 if let Some(existing) = self.external_order_claims.get(instrument_id)
316 && *existing != strategy_id
317 {
318 anyhow::bail!(
319 "External order claim for {instrument_id} already exists for {existing}"
320 );
321 }
322 }
323
324 self.external_order_claims
325 .retain(|_, owner| *owner != strategy_id);
326 self.external_order_claims.extend(
327 requested
328 .into_iter()
329 .map(|instrument_id| (instrument_id, strategy_id)),
330 );
331
332 Ok(())
333 }
334
335 pub fn register_external_order_claims(
341 &mut self,
342 strategy_id: StrategyId,
343 instrument_ids: &[InstrumentId],
344 ) -> anyhow::Result<()> {
345 let mut requested = AHashSet::with_capacity(instrument_ids.len());
346
347 for instrument_id in instrument_ids {
348 if !requested.insert(*instrument_id) {
349 anyhow::bail!(
350 "External order claim for {instrument_id} appears more than once for {strategy_id}"
351 );
352 }
353
354 if let Some(existing) = self.external_order_claims.get(instrument_id) {
355 anyhow::bail!(
356 "External order claim for {instrument_id} already exists for {existing}"
357 );
358 }
359 }
360
361 self.external_order_claims.extend(
362 requested
363 .into_iter()
364 .map(|instrument_id| (instrument_id, strategy_id)),
365 );
366
367 Ok(())
368 }
369
370 #[must_use]
375 pub fn account_id_for_client(&self, client_id: &ClientId) -> Option<&AccountId> {
376 self.client_accounts.get(client_id)
377 }
378
379 pub fn add_client_account(&mut self, client_id: ClientId, account_id: AccountId) {
384 self.client_accounts.insert(client_id, account_id);
385 }
386
387 pub fn remove_client_account(&mut self, client_id: &ClientId) {
389 self.client_accounts.remove(client_id);
390 }
391
392 #[must_use]
394 pub fn client_id_for_venue(&self, venue: &Venue) -> Option<&ClientId> {
395 self.client_routes
396 .get(venue)
397 .or(self.default_client_id.as_ref())
398 }
399
400 pub fn add_client_route(&mut self, client_id: ClientId, venue: Venue) {
405 self.client_routes.insert(venue, client_id);
406 }
407
408 pub fn set_default_client(&mut self, client_id: ClientId) {
411 self.default_client_id = Some(client_id);
412 }
413
414 pub fn remove_client_routes(&mut self, client_id: &ClientId) {
416 self.client_routes.retain(|_, routed| routed != client_id);
417
418 if self.default_client_id.as_ref() == Some(client_id) {
419 self.default_client_id = None;
420 }
421 }
422
423 #[must_use]
426 pub fn is_external_client(&self, client_id: &ClientId) -> bool {
427 self.external_clients.contains(client_id)
428 }
429
430 pub fn add_external_client(&mut self, client_id: ClientId) {
434 self.external_clients.insert(client_id);
435 }
436
437 pub fn set_database(&mut self, database: Box<dyn CacheDatabaseAdapter>) {
441 let type_name = std::any::type_name_of_val(&*database);
442 log::info!("Cache database adapter set: {type_name}");
443 self.database = Some(database);
444 }
445
446 pub fn cache_general(&mut self) -> anyhow::Result<()> {
454 self.general = match &mut self.database {
455 Some(db) => db.load()?,
456 None => AHashMap::new(),
457 };
458
459 log::info!(
460 "Cached {} general object(s) from database",
461 self.general.len()
462 );
463 Ok(())
464 }
465
466 pub async fn cache_all(&mut self) -> anyhow::Result<()> {
475 let mut cache_map = match &self.database {
476 Some(db) => db.load_all().await?,
477 None => CacheMap::default(),
478 };
479
480 register_loaded_currencies(&mut cache_map.currencies)?;
483 self.currencies = cache_map.currencies;
484 self.instruments = cache_map.instruments;
485 self.instrument_closes = cache_map.instrument_closes;
486 self.synthetics = cache_map.synthetics;
487 self.accounts = cache_map
488 .accounts
489 .into_iter()
490 .map(|(id, account)| (id, SharedCell::new(account)))
491 .collect();
492 self.orders = cache_map
493 .orders
494 .into_iter()
495 .map(|(id, order)| (id, SharedCell::new(order)))
496 .collect();
497 self.positions = cache_map
498 .positions
499 .into_iter()
500 .map(|(id, position)| (id, SharedCell::new(position)))
501 .collect();
502
503 if let Some(db) = &self.database {
504 let order_position = db.load_index_order_position()?;
505 self.index.order_position = self.sanitize_order_position_index(order_position);
506 self.index.order_client = db.load_index_order_client()?;
507 }
508
509 self.cache_position_oms()?;
510 self.assign_position_ids_to_contingencies();
511 Ok(())
512 }
513
514 pub async fn cache_currencies(&mut self) -> anyhow::Result<()> {
520 self.currencies = match &mut self.database {
521 Some(db) => db.load_currencies().await?,
522 None => AHashMap::new(),
523 };
524
525 register_loaded_currencies(&mut self.currencies)?;
526
527 log::info!("Cached {} currencies from database", self.currencies.len());
528 Ok(())
529 }
530
531 pub async fn cache_instruments(&mut self) -> anyhow::Result<()> {
537 self.instruments = match &mut self.database {
538 Some(db) => db.load_instruments().await?,
539 None => AHashMap::new(),
540 };
541
542 log::info!("Cached {} instruments from database", self.general.len());
543 Ok(())
544 }
545
546 pub async fn cache_synthetics(&mut self) -> anyhow::Result<()> {
552 self.synthetics = match &mut self.database {
553 Some(db) => db.load_synthetics().await?,
554 None => AHashMap::new(),
555 };
556
557 log::info!(
558 "Cached {} synthetic instruments from database",
559 self.general.len()
560 );
561 Ok(())
562 }
563
564 pub async fn cache_accounts(&mut self) -> anyhow::Result<()> {
570 self.accounts = match &mut self.database {
571 Some(db) => db
572 .load_accounts()
573 .await?
574 .into_iter()
575 .map(|(id, account)| (id, SharedCell::new(account)))
576 .collect(),
577 None => AHashMap::new(),
578 };
579
580 log::info!(
581 "Cached {} synthetic instruments from database",
582 self.general.len()
583 );
584 Ok(())
585 }
586
587 pub async fn cache_orders(&mut self) -> anyhow::Result<()> {
593 self.orders = match &mut self.database {
594 Some(db) => db
595 .load_orders()
596 .await?
597 .into_iter()
598 .map(|(id, order)| (id, SharedCell::new(order)))
599 .collect(),
600 None => AHashMap::new(),
601 };
602
603 if let Some(db) = &self.database {
604 let order_position = db.load_index_order_position()?;
605 self.index.order_position = self.sanitize_order_position_index(order_position);
606 self.index.order_client = db.load_index_order_client()?;
607 }
608
609 log::info!("Cached {} orders from database", self.general.len());
610
611 self.assign_position_ids_to_contingencies();
612 Ok(())
613 }
614
615 fn sanitize_order_position_index(
616 &self,
617 mut order_position: AHashMap<ClientOrderId, PositionId>,
618 ) -> AHashMap<ClientOrderId, PositionId> {
619 let original_len = order_position.len();
620 order_position.retain(|client_order_id, _| self.orders.contains_key(client_order_id));
621 let removed = original_len - order_position.len();
622
623 if removed > 0 {
624 log::warn!(
625 "Filtered {removed} stale order-position index entries without backing orders during cache load"
626 );
627 }
628
629 order_position
630 }
631
632 pub async fn cache_positions(&mut self) -> anyhow::Result<()> {
638 self.positions = match &mut self.database {
639 Some(db) => db
640 .load_positions()
641 .await?
642 .into_iter()
643 .map(|(id, position)| (id, SharedCell::new(position)))
644 .collect(),
645 None => AHashMap::new(),
646 };
647
648 self.cache_position_oms()?;
649 log::info!("Cached {} positions from database", self.general.len());
650 Ok(())
651 }
652
653 fn cache_position_oms(&mut self) -> anyhow::Result<()> {
654 let persisted = match &self.database {
655 Some(database) => database.load()?,
656 None => self.general.clone(),
657 };
658
659 self.general
660 .retain(|key, _| !key.starts_with(POSITION_OMS_KEY_PREFIX));
661
662 for (key, value) in persisted {
663 if !key.starts_with(POSITION_OMS_KEY_PREFIX) {
664 continue;
665 }
666 self.general.insert(key, value);
667 }
668
669 self.index_position_oms();
670 Ok(())
671 }
672
673 pub fn build_index(&mut self) {
675 log::debug!("Building index");
676
677 for account_id in self.accounts.keys() {
679 self.index.add_venue_account(*account_id);
680 }
681
682 for (client_order_id, order_cell) in &self.orders {
684 let order = order_cell.borrow();
685 let instrument_id = order.instrument_id();
686 let venue = instrument_id.venue;
687 let strategy_id = order.strategy_id();
688
689 self.index
691 .venue_orders
692 .entry(venue)
693 .or_default()
694 .insert(*client_order_id);
695
696 if let Some(venue_order_id) = order.venue_order_id() {
699 self.index
700 .venue_order_ids
701 .insert(venue_order_id, *client_order_id);
702 self.index
703 .client_order_ids
704 .insert(*client_order_id, venue_order_id);
705 }
706
707 self.index.order_position.extend(
709 order
710 .position_id()
711 .map(|position_id| (*client_order_id, position_id)),
712 );
713
714 self.index
716 .order_strategy
717 .insert(*client_order_id, strategy_id);
718
719 self.index
721 .instrument_orders
722 .entry(instrument_id)
723 .or_default()
724 .insert(*client_order_id);
725
726 self.index
728 .strategy_orders
729 .entry(strategy_id)
730 .or_default()
731 .insert(*client_order_id);
732
733 if let Some(account_id) = order.account_id() {
735 self.index
736 .account_orders
737 .entry(account_id)
738 .or_default()
739 .insert(*client_order_id);
740 }
741
742 if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
744 self.index
745 .exec_algorithm_orders
746 .entry(exec_algorithm_id)
747 .or_default()
748 .insert(*client_order_id);
749 self.index.exec_algorithms.insert(exec_algorithm_id);
750 }
751
752 if let Some(exec_spawn_id) = order.exec_spawn_id() {
754 self.index
755 .exec_spawn_orders
756 .entry(exec_spawn_id)
757 .or_default()
758 .insert(*client_order_id);
759 }
760
761 self.index.orders.insert(*client_order_id);
763
764 if order.is_active_local() {
766 self.index.orders_active_local.insert(*client_order_id);
767 }
768
769 if order.is_open() {
771 self.index.orders_open.insert(*client_order_id);
772 }
773
774 if order.is_closed() {
776 self.index.orders_closed.insert(*client_order_id);
777 }
778
779 if order.emulation_trigger().is_some() && !order.is_closed() {
781 self.index.orders_emulated.insert(*client_order_id);
782 }
783
784 if order.is_inflight() {
786 self.index.orders_inflight.insert(*client_order_id);
787 }
788
789 self.index.strategies.insert(strategy_id);
791 }
792
793 for (position_id, position_cell) in &self.positions {
795 let position = position_cell.borrow();
796 let instrument_id = position.instrument_id;
797 let venue = instrument_id.venue;
798 let strategy_id = position.strategy_id;
799
800 self.index
802 .venue_positions
803 .entry(venue)
804 .or_default()
805 .insert(*position_id);
806
807 self.index
809 .position_strategy
810 .insert(*position_id, strategy_id);
811
812 let position_orders = self.index.position_orders.entry(*position_id).or_default();
814 position_orders.extend(
815 position
816 .client_order_ids()
817 .into_iter()
818 .filter(|client_order_id| self.orders.contains_key(client_order_id)),
819 );
820
821 self.index
823 .instrument_positions
824 .entry(instrument_id)
825 .or_default()
826 .insert(*position_id);
827 self.index
828 .instrument_orders
829 .entry(instrument_id)
830 .or_default();
831
832 self.index
834 .strategy_positions
835 .entry(strategy_id)
836 .or_default()
837 .insert(*position_id);
838 self.index.strategy_orders.entry(strategy_id).or_default();
839
840 self.index
842 .account_positions
843 .entry(position.account_id)
844 .or_default()
845 .insert(*position_id);
846
847 self.index.positions.insert(*position_id);
849
850 if position.is_open() {
852 self.index.positions_open.insert(*position_id);
853 }
854
855 if position.is_closed() {
857 self.index.positions_closed.insert(*position_id);
858 }
859
860 self.index.strategies.insert(strategy_id);
862 }
863
864 self.index_position_oms();
865 let client_order_ids: Vec<_> = self.orders.keys().copied().collect();
866 for client_order_id in client_order_ids {
867 self.reindex_hedging_position(client_order_id);
868 }
869 }
870
871 fn index_position_oms(&mut self) {
872 self.index.position_oms.clear();
873
874 for (key, value) in &self.general {
875 let Some(position_id) = key.strip_prefix(POSITION_OMS_KEY_PREFIX) else {
876 continue;
877 };
878 let position_id = PositionId::new(position_id);
879 if !self.positions.contains_key(&position_id) {
880 continue;
881 }
882
883 match serde_json::from_slice::<OmsType>(value) {
884 Ok(oms_type) => {
885 self.index.position_oms.insert(position_id, oms_type);
886 }
887 Err(e) => {
888 log::error!("Failed to decode position OMS for {position_id}: {e}");
889 }
890 }
891 }
892
893 for position in self.positions.values().map(|cell| cell.borrow()) {
894 if !self.index.position_oms.contains_key(&position.id)
895 && position.id.as_str()
896 == format!("{}-{}", position.instrument_id, position.strategy_id)
897 {
898 self.index
899 .position_oms
900 .insert(position.id, OmsType::Netting);
901 }
902 }
903 }
904
905 #[must_use]
907 pub const fn has_backing(&self) -> bool {
908 self.database.is_some()
909 }
910
911 pub fn load_actor_state(
919 &self,
920 actor_id: &ActorId,
921 ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
922 self.database
923 .as_ref()
924 .map(|database| database.load_actor(actor_id))
925 .transpose()
926 .map(|state| state.map(Self::decode_component_state))
927 }
928
929 pub fn load_strategy_state(
937 &self,
938 strategy_id: &StrategyId,
939 ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
940 self.database
941 .as_ref()
942 .map(|database| database.load_strategy(strategy_id))
943 .transpose()
944 .map(|state| state.map(Self::decode_component_state))
945 }
946
947 pub fn update_actor_state(
953 &self,
954 actor_id: &ActorId,
955 state: &IndexMap<String, Vec<u8>>,
956 ) -> anyhow::Result<()> {
957 if let Some(database) = &self.database {
958 database.update_actor(actor_id, &Self::encode_component_state(state))?;
959 }
960 Ok(())
961 }
962
963 pub fn update_strategy_state(
969 &self,
970 strategy_id: &StrategyId,
971 state: &IndexMap<String, Vec<u8>>,
972 ) -> anyhow::Result<()> {
973 if let Some(database) = &self.database {
974 database.update_strategy(strategy_id, &Self::encode_component_state(state))?;
975 }
976 Ok(())
977 }
978
979 fn decode_component_state(state: AHashMap<String, Bytes>) -> IndexMap<String, Vec<u8>> {
980 state
981 .into_iter()
982 .map(|(key, value)| (key, value.to_vec()))
983 .collect()
984 }
985
986 fn encode_component_state(state: &IndexMap<String, Vec<u8>>) -> AHashMap<String, Bytes> {
987 state
988 .iter()
989 .map(|(key, value)| (key.clone(), Bytes::copy_from_slice(value)))
990 .collect()
991 }
992
993 #[must_use]
995 pub fn calculate_unrealized_pnl(&self, position: &Position) -> Option<Money> {
996 let Some(quote) = self.quote(&position.instrument_id) else {
997 log::warn!(
998 "Cannot calculate unrealized PnL for {}, no quotes for {}",
999 position.id,
1000 position.instrument_id
1001 );
1002 return None;
1003 };
1004
1005 let last = match position.side {
1007 PositionSide::Flat => {
1008 return Some(Money::zero(position.settlement_currency));
1009 }
1010 PositionSide::Long => quote.bid_price,
1011 PositionSide::Short => quote.ask_price,
1012 };
1013
1014 position
1015 .try_unrealized_pnl(last)
1016 .inspect_err(|e| {
1017 log::error!("Cannot calculate unrealized PnL for {}: {e}", position.id);
1018 })
1019 .ok()
1020 }
1021
1022 #[must_use]
1031 pub fn check_integrity(&mut self) -> bool {
1032 let mut error_count = 0;
1033 let failure = "Integrity failure";
1034
1035 let timestamp_us = SystemTime::now()
1037 .duration_since(UNIX_EPOCH)
1038 .expect("Time went backwards")
1039 .as_micros();
1040
1041 log::info!("Checking data integrity");
1042
1043 for account_id in self.accounts.keys() {
1045 if !self
1046 .index
1047 .venue_accounts
1048 .get(&account_id.get_issuer())
1049 .is_some_and(|account_ids| account_ids.contains(account_id))
1050 {
1051 log::error!(
1052 "{failure} in accounts: {account_id} not found in `self.index.venue_accounts`",
1053 );
1054 error_count += 1;
1055 }
1056 }
1057
1058 for (client_order_id, order_cell) in &self.orders {
1059 let order = order_cell.borrow();
1060
1061 if !self.index.order_strategy.contains_key(client_order_id) {
1062 log::error!(
1063 "{failure} in orders: {client_order_id} not found in `self.index.order_strategy`"
1064 );
1065 error_count += 1;
1066 }
1067
1068 if !self.index.orders.contains(client_order_id) {
1069 log::error!(
1070 "{failure} in orders: {client_order_id} not found in `self.index.orders`",
1071 );
1072 error_count += 1;
1073 }
1074
1075 if order.is_inflight() && !self.index.orders_inflight.contains(client_order_id) {
1076 log::error!(
1077 "{failure} in orders: {client_order_id} not found in `self.index.orders_inflight`",
1078 );
1079 error_count += 1;
1080 }
1081
1082 if order.is_active_local() && !self.index.orders_active_local.contains(client_order_id)
1083 {
1084 log::error!(
1085 "{failure} in orders: {client_order_id} not found in `self.index.orders_active_local`",
1086 );
1087 error_count += 1;
1088 }
1089
1090 if order.is_open() && !self.index.orders_open.contains(client_order_id) {
1091 log::error!(
1092 "{failure} in orders: {client_order_id} not found in `self.index.orders_open`",
1093 );
1094 error_count += 1;
1095 }
1096
1097 if order.is_closed() && !self.index.orders_closed.contains(client_order_id) {
1098 log::error!(
1099 "{failure} in orders: {client_order_id} not found in `self.index.orders_closed`",
1100 );
1101 error_count += 1;
1102 }
1103
1104 if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
1105 if !self
1106 .index
1107 .exec_algorithm_orders
1108 .contains_key(&exec_algorithm_id)
1109 {
1110 log::error!(
1111 "{failure} in orders: {client_order_id} not found in `self.index.exec_algorithm_orders`",
1112 );
1113 error_count += 1;
1114 }
1115
1116 if order.exec_spawn_id().is_none()
1117 && !self.index.exec_spawn_orders.contains_key(client_order_id)
1118 {
1119 log::error!(
1120 "{failure} in orders: {client_order_id} not found in `self.index.exec_spawn_orders`",
1121 );
1122 error_count += 1;
1123 }
1124 }
1125 }
1126
1127 for (position_id, position_cell) in &self.positions {
1128 let position = position_cell.borrow();
1129
1130 if !self.index.position_strategy.contains_key(position_id) {
1131 log::error!(
1132 "{failure} in positions: {position_id} not found in `self.index.position_strategy`",
1133 );
1134 error_count += 1;
1135 }
1136
1137 if !self.index.position_orders.contains_key(position_id) {
1138 log::error!(
1139 "{failure} in positions: {position_id} not found in `self.index.position_orders`",
1140 );
1141 error_count += 1;
1142 }
1143
1144 if !self.index.positions.contains(position_id) {
1145 log::error!(
1146 "{failure} in positions: {position_id} not found in `self.index.positions`",
1147 );
1148 error_count += 1;
1149 }
1150
1151 if position.is_open() && !self.index.positions_open.contains(position_id) {
1152 log::error!(
1153 "{failure} in positions: {position_id} not found in `self.index.positions_open`",
1154 );
1155 error_count += 1;
1156 }
1157
1158 if position.is_closed() && !self.index.positions_closed.contains(position_id) {
1159 log::error!(
1160 "{failure} in positions: {position_id} not found in `self.index.positions_closed`",
1161 );
1162 error_count += 1;
1163 }
1164 }
1165
1166 for account_id in self.index.venue_accounts.values().flatten() {
1168 if !self.accounts.contains_key(account_id) {
1169 log::error!(
1170 "{failure} in `index.venue_accounts`: {account_id} not found in `self.accounts`",
1171 );
1172 error_count += 1;
1173 }
1174 }
1175
1176 for client_order_id in self.index.venue_order_ids.values() {
1177 if !self.orders.contains_key(client_order_id) {
1178 log::error!(
1179 "{failure} in `index.venue_order_ids`: {client_order_id} not found in `self.orders`",
1180 );
1181 error_count += 1;
1182 }
1183 }
1184
1185 for client_order_id in self.index.client_order_ids.keys() {
1186 if !self.orders.contains_key(client_order_id) {
1187 log::error!(
1188 "{failure} in `index.client_order_ids`: {client_order_id} not found in `self.orders`",
1189 );
1190 error_count += 1;
1191 }
1192 }
1193
1194 for client_order_id in self.index.order_position.keys() {
1195 if !self.orders.contains_key(client_order_id) {
1196 log::error!(
1197 "{failure} in `index.order_position`: {client_order_id} not found in `self.orders`",
1198 );
1199 error_count += 1;
1200 }
1201 }
1202
1203 for client_order_id in self.index.order_strategy.keys() {
1205 if !self.orders.contains_key(client_order_id) {
1206 log::error!(
1207 "{failure} in `index.order_strategy`: {client_order_id} not found in `self.orders`",
1208 );
1209 error_count += 1;
1210 }
1211 }
1212
1213 for position_id in self.index.position_strategy.keys() {
1214 if !self.positions.contains_key(position_id) {
1215 log::error!(
1216 "{failure} in `index.position_strategy`: {position_id} not found in `self.positions`",
1217 );
1218 error_count += 1;
1219 }
1220 }
1221
1222 for position_id in self.index.position_orders.keys() {
1223 if !self.positions.contains_key(position_id) {
1224 log::error!(
1225 "{failure} in `index.position_orders`: {position_id} not found in `self.positions`",
1226 );
1227 error_count += 1;
1228 }
1229 }
1230
1231 for (instrument_id, client_order_ids) in &self.index.instrument_orders {
1232 for client_order_id in client_order_ids {
1233 if !self.orders.contains_key(client_order_id) {
1234 log::error!(
1235 "{failure} in `index.instrument_orders`: {instrument_id} not found in `self.orders`",
1236 );
1237 error_count += 1;
1238 }
1239 }
1240 }
1241
1242 for instrument_id in self.index.instrument_positions.keys() {
1243 if !self.index.instrument_orders.contains_key(instrument_id) {
1244 log::error!(
1245 "{failure} in `index.instrument_positions`: {instrument_id} not found in `index.instrument_orders`",
1246 );
1247 error_count += 1;
1248 }
1249 }
1250
1251 for client_order_ids in self.index.strategy_orders.values() {
1252 for client_order_id in client_order_ids {
1253 if !self.orders.contains_key(client_order_id) {
1254 log::error!(
1255 "{failure} in `index.strategy_orders`: {client_order_id} not found in `self.orders`",
1256 );
1257 error_count += 1;
1258 }
1259 }
1260 }
1261
1262 for position_ids in self.index.strategy_positions.values() {
1263 for position_id in position_ids {
1264 if !self.positions.contains_key(position_id) {
1265 log::error!(
1266 "{failure} in `index.strategy_positions`: {position_id} not found in `self.positions`",
1267 );
1268 error_count += 1;
1269 }
1270 }
1271 }
1272
1273 for client_order_id in &self.index.orders {
1274 if !self.orders.contains_key(client_order_id) {
1275 log::error!(
1276 "{failure} in `index.orders`: {client_order_id} not found in `self.orders`",
1277 );
1278 error_count += 1;
1279 }
1280 }
1281
1282 for client_order_id in &self.index.orders_emulated {
1283 if !self.orders.contains_key(client_order_id) {
1284 log::error!(
1285 "{failure} in `index.orders_emulated`: {client_order_id} not found in `self.orders`",
1286 );
1287 error_count += 1;
1288 }
1289 }
1290
1291 for client_order_id in &self.index.orders_active_local {
1292 if !self.orders.contains_key(client_order_id) {
1293 log::error!(
1294 "{failure} in `index.orders_active_local`: {client_order_id} not found in `self.orders`",
1295 );
1296 error_count += 1;
1297 }
1298 }
1299
1300 for client_order_id in &self.index.orders_inflight {
1301 if !self.orders.contains_key(client_order_id) {
1302 log::error!(
1303 "{failure} in `index.orders_inflight`: {client_order_id} not found in `self.orders`",
1304 );
1305 error_count += 1;
1306 }
1307 }
1308
1309 for client_order_id in &self.index.orders_open {
1310 if !self.orders.contains_key(client_order_id) {
1311 log::error!(
1312 "{failure} in `index.orders_open`: {client_order_id} not found in `self.orders`",
1313 );
1314 error_count += 1;
1315 }
1316 }
1317
1318 for client_order_id in &self.index.orders_closed {
1319 if !self.orders.contains_key(client_order_id) {
1320 log::error!(
1321 "{failure} in `index.orders_closed`: {client_order_id} not found in `self.orders`",
1322 );
1323 error_count += 1;
1324 }
1325 }
1326
1327 for position_id in &self.index.positions {
1328 if !self.positions.contains_key(position_id) {
1329 log::error!(
1330 "{failure} in `index.positions`: {position_id} not found in `self.positions`",
1331 );
1332 error_count += 1;
1333 }
1334 }
1335
1336 for position_id in &self.index.positions_open {
1337 if !self.positions.contains_key(position_id) {
1338 log::error!(
1339 "{failure} in `index.positions_open`: {position_id} not found in `self.positions`",
1340 );
1341 error_count += 1;
1342 }
1343 }
1344
1345 for position_id in &self.index.positions_closed {
1346 if !self.positions.contains_key(position_id) {
1347 log::error!(
1348 "{failure} in `index.positions_closed`: {position_id} not found in `self.positions`",
1349 );
1350 error_count += 1;
1351 }
1352 }
1353
1354 for strategy_id in &self.index.strategies {
1355 if !self.index.strategy_orders.contains_key(strategy_id) {
1356 log::error!(
1357 "{failure} in `index.strategies`: {strategy_id} not found in `index.strategy_orders`",
1358 );
1359 error_count += 1;
1360 }
1361 }
1362
1363 for exec_algorithm_id in &self.index.exec_algorithms {
1364 if !self
1365 .index
1366 .exec_algorithm_orders
1367 .contains_key(exec_algorithm_id)
1368 {
1369 log::error!(
1370 "{failure} in `index.exec_algorithms`: {exec_algorithm_id} not found in `index.exec_algorithm_orders`",
1371 );
1372 error_count += 1;
1373 }
1374 }
1375
1376 let total_us = SystemTime::now()
1377 .duration_since(UNIX_EPOCH)
1378 .expect("Time went backwards")
1379 .as_micros()
1380 - timestamp_us;
1381
1382 if error_count == 0 {
1383 log::info!("Integrity check passed in {total_us}μs");
1384 true
1385 } else {
1386 log::error!(
1387 "Integrity check failed with {error_count} error{} in {total_us}μs",
1388 if error_count == 1 { "" } else { "s" },
1389 );
1390 false
1391 }
1392 }
1393
1394 #[must_use]
1398 pub fn check_residuals(&self) -> bool {
1399 log::debug!("Checking residuals");
1400
1401 let mut residuals = false;
1402
1403 for order in self.orders_open(None, None, None, None, None) {
1405 residuals = true;
1406 log::warn!("Residual {order}");
1407 }
1408
1409 for position in self.positions_open(None, None, None, None, None) {
1411 residuals = true;
1412 log::warn!("Residual {position}");
1413 }
1414
1415 residuals
1416 }
1417
1418 pub fn purge_closed_orders(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
1424 log::debug!(
1425 "Purging closed orders{}",
1426 if buffer_secs > 0 {
1427 format!(" with buffer_secs={buffer_secs}")
1428 } else {
1429 String::new()
1430 }
1431 );
1432
1433 let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
1434 log::warn!(
1435 "Cannot purge closed orders: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
1436 );
1437 return;
1438 };
1439 let purge_cutoff = ts_now.checked_sub(buffer_ns);
1440
1441 let mut affected_order_list_ids: AHashSet<OrderListId> = AHashSet::new();
1442 let mut purged_client_order_ids: AHashSet<ClientOrderId> = AHashSet::new();
1443
1444 'outer: for client_order_id in self.index.orders_closed.clone() {
1445 let purge_target = self.orders.get(&client_order_id).and_then(|order_cell| {
1446 let order = order_cell.borrow();
1447 if order.is_closed()
1448 && let Some(ts_closed) = order.ts_closed()
1449 && purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
1450 {
1451 let linked = order.linked_order_ids().map(<[_]>::to_vec);
1452 let order_list_id = order.order_list_id();
1453 Some((linked, order_list_id))
1454 } else {
1455 None
1456 }
1457 });
1458
1459 let Some((linked, order_list_id)) = purge_target else {
1460 continue;
1461 };
1462
1463 if let Some(linked_order_ids) = linked {
1465 for linked_order_id in &linked_order_ids {
1466 if let Some(linked_order_cell) = self.orders.get(linked_order_id)
1467 && linked_order_cell.borrow().is_open()
1468 {
1469 continue 'outer;
1471 }
1472 }
1473 }
1474
1475 if let Some(order_list_id) = order_list_id {
1476 affected_order_list_ids.insert(order_list_id);
1477 }
1478
1479 if self.purge_order_except_aliases(client_order_id) {
1480 purged_client_order_ids.insert(client_order_id);
1481 }
1482 }
1483
1484 if !purged_client_order_ids.is_empty() {
1485 self.index
1486 .venue_order_ids
1487 .retain(|_, owner| !purged_client_order_ids.contains(owner));
1488 }
1489
1490 for order_list_id in affected_order_list_ids {
1491 if let Some(order_list) = self.order_lists.get(&order_list_id) {
1492 let all_purged = order_list
1493 .client_order_ids
1494 .iter()
1495 .all(|id| !self.orders.contains_key(id));
1496
1497 if all_purged {
1498 self.order_lists.remove(&order_list_id);
1499 log::info!("Purged {order_list_id}");
1500 }
1501 }
1502 }
1503 }
1504
1505 pub fn purge_closed_positions(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
1507 log::debug!(
1508 "Purging closed positions{}",
1509 if buffer_secs > 0 {
1510 format!(" with buffer_secs={buffer_secs}")
1511 } else {
1512 String::new()
1513 }
1514 );
1515
1516 let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
1517 log::warn!(
1518 "Cannot purge closed positions: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
1519 );
1520 return;
1521 };
1522 let purge_cutoff = ts_now.checked_sub(buffer_ns);
1523
1524 for position_id in self.index.positions_closed.clone() {
1525 let should_purge = self.positions.get(&position_id).is_some_and(|cell| {
1526 let position = cell.borrow();
1527 position.is_closed()
1528 && position.ts_closed.is_some_and(|ts_closed| {
1529 purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
1530 })
1531 });
1532
1533 if should_purge {
1534 self.purge_position(position_id);
1535 }
1536 }
1537 }
1538
1539 pub fn purge_order(&mut self, client_order_id: ClientOrderId) {
1543 if self.purge_order_except_aliases(client_order_id) {
1544 self.index
1545 .venue_order_ids
1546 .retain(|_, owner| owner != &client_order_id);
1547 }
1548 }
1549
1550 fn purge_order_except_aliases(&mut self, client_order_id: ClientOrderId) -> bool {
1555 struct OrderDetails {
1556 is_open: bool,
1557 instrument_id: InstrumentId,
1558 strategy_id: StrategyId,
1559 account_id: Option<AccountId>,
1560 exec_algorithm_id: Option<ExecAlgorithmId>,
1561 exec_spawn_id: Option<ClientOrderId>,
1562 position_id: Option<PositionId>,
1563 }
1564
1565 let order_cell = self.orders.get(&client_order_id).cloned();
1566 let order_details = order_cell.as_ref().map(|cell| {
1567 let order = cell.borrow();
1568 OrderDetails {
1569 is_open: order.is_open(),
1570 instrument_id: order.instrument_id(),
1571 strategy_id: order.strategy_id(),
1572 account_id: order.account_id(),
1573 exec_algorithm_id: order.exec_algorithm_id(),
1574 exec_spawn_id: order.exec_spawn_id(),
1575 position_id: order.position_id(),
1576 }
1577 });
1578
1579 if order_details
1580 .as_ref()
1581 .is_some_and(|details| details.is_open)
1582 {
1583 log::warn!("Order {client_order_id} found open when purging, skipping purge");
1584 return false;
1585 }
1586
1587 if order_details.is_some() {
1588 self.orders.remove(&client_order_id);
1589 } else {
1590 log::warn!("Order {client_order_id} not found when purging");
1591 }
1592
1593 let indexed_position_id = self.index.order_position.remove(&client_order_id);
1594 let indexed_strategy_id = self.index.order_strategy.remove(&client_order_id);
1595 self.index.order_client.remove(&client_order_id);
1596 self.index.client_order_ids.remove(&client_order_id);
1597
1598 if let Some(details) = &order_details {
1599 if let Some(venue_orders) = self
1600 .index
1601 .venue_orders
1602 .get_mut(&details.instrument_id.venue)
1603 {
1604 venue_orders.remove(&client_order_id);
1605 if venue_orders.is_empty() {
1606 self.index.venue_orders.remove(&details.instrument_id.venue);
1607 }
1608 }
1609
1610 let instrument_orders_became_empty = self
1615 .index
1616 .instrument_orders
1617 .get_mut(&details.instrument_id)
1618 .is_some_and(|instrument_orders| {
1619 instrument_orders.remove(&client_order_id);
1620 instrument_orders.is_empty()
1621 });
1622
1623 let has_instrument_positions = self
1624 .index
1625 .instrument_positions
1626 .get(&details.instrument_id)
1627 .is_some_and(|positions| !positions.is_empty());
1628
1629 if instrument_orders_became_empty && !has_instrument_positions {
1630 self.index.instrument_orders.remove(&details.instrument_id);
1631 }
1632
1633 if let Some(exec_algorithm_id) = details.exec_algorithm_id {
1634 let became_empty = self
1635 .index
1636 .exec_algorithm_orders
1637 .get_mut(&exec_algorithm_id)
1638 .is_some_and(|orders| {
1639 orders.remove(&client_order_id);
1640 orders.is_empty()
1641 });
1642
1643 if became_empty {
1644 self.index.exec_algorithm_orders.remove(&exec_algorithm_id);
1645 self.index.exec_algorithms.remove(&exec_algorithm_id);
1646 }
1647 }
1648
1649 if let Some(account_id) = details.account_id
1650 && let Some(account_orders) = self.index.account_orders.get_mut(&account_id)
1651 {
1652 account_orders.remove(&client_order_id);
1653 if account_orders.is_empty() {
1654 self.index.account_orders.remove(&account_id);
1655 }
1656 }
1657
1658 if let Some(exec_spawn_id) = details.exec_spawn_id
1659 && let Some(spawn_orders) = self.index.exec_spawn_orders.get_mut(&exec_spawn_id)
1660 {
1661 spawn_orders.remove(&client_order_id);
1662 if spawn_orders.is_empty() {
1663 self.index.exec_spawn_orders.remove(&exec_spawn_id);
1664 }
1665 }
1666 }
1667
1668 let mut position_ids = AHashSet::new();
1669 if let Some(position_id) = indexed_position_id {
1670 position_ids.insert(position_id);
1671 }
1672
1673 if let Some(position_id) = order_details
1674 .as_ref()
1675 .and_then(|details| details.position_id)
1676 {
1677 position_ids.insert(position_id);
1678 }
1679
1680 let mut strategy_ids = AHashSet::new();
1681 if let Some(strategy_id) = indexed_strategy_id {
1682 strategy_ids.insert(strategy_id);
1683 }
1684
1685 if let Some(details) = &order_details {
1686 strategy_ids.insert(details.strategy_id);
1687 }
1688
1689 for position_id in position_ids {
1690 if self.positions.contains_key(&position_id) {
1691 if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
1692 position_orders.remove(&client_order_id);
1693 }
1694 continue;
1695 }
1696
1697 let has_other_orders =
1698 if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
1699 position_orders.remove(&client_order_id);
1700 !position_orders.is_empty()
1701 } else {
1702 self.index
1703 .order_position
1704 .values()
1705 .any(|candidate| *candidate == position_id)
1706 };
1707
1708 if has_other_orders {
1709 continue;
1710 }
1711
1712 self.index.position_orders.remove(&position_id);
1713 if let Some(strategy_id) = self.index.position_strategy.remove(&position_id) {
1714 strategy_ids.insert(strategy_id);
1715 if let Some(strategy_positions) =
1716 self.index.strategy_positions.get_mut(&strategy_id)
1717 {
1718 strategy_positions.remove(&position_id);
1719 if strategy_positions.is_empty() {
1720 self.index.strategy_positions.remove(&strategy_id);
1721 }
1722 }
1723 }
1724
1725 if let Some(details) = &order_details
1726 && let Some(venue_positions) = self
1727 .index
1728 .venue_positions
1729 .get_mut(&details.instrument_id.venue)
1730 {
1731 venue_positions.remove(&position_id);
1732 if venue_positions.is_empty() {
1733 self.index
1734 .venue_positions
1735 .remove(&details.instrument_id.venue);
1736 }
1737 }
1738 }
1739
1740 for strategy_id in strategy_ids {
1741 let strategy_orders_became_empty = self
1747 .index
1748 .strategy_orders
1749 .get_mut(&strategy_id)
1750 .is_some_and(|strategy_orders| {
1751 strategy_orders.remove(&client_order_id);
1752 strategy_orders.is_empty()
1753 });
1754
1755 let has_positions = self
1756 .index
1757 .strategy_positions
1758 .get(&strategy_id)
1759 .is_some_and(|strategy_positions| !strategy_positions.is_empty());
1760
1761 if strategy_orders_became_empty && !has_positions {
1762 self.index.strategy_orders.remove(&strategy_id);
1763 self.index.strategies.remove(&strategy_id);
1764 }
1765 }
1766
1767 self.index.exec_spawn_orders.remove(&client_order_id);
1768
1769 self.index.orders.remove(&client_order_id);
1770 self.index.orders_active_local.remove(&client_order_id);
1771 self.index.orders_open.remove(&client_order_id);
1772 self.index.orders_closed.remove(&client_order_id);
1773 self.index.orders_emulated.remove(&client_order_id);
1774 self.index.orders_inflight.remove(&client_order_id);
1775 self.index.orders_pending_cancel.remove(&client_order_id);
1776
1777 if order_details.is_some() {
1778 log::info!("Purged order {client_order_id}");
1779 }
1780
1781 true
1782 }
1783
1784 pub fn purge_position(&mut self, position_id: PositionId) {
1788 let position = self
1790 .positions
1791 .get(&position_id)
1792 .map(|cell| cell.borrow().clone());
1793
1794 if let Some(ref pos) = position
1796 && pos.is_open()
1797 {
1798 log::warn!("Position {position_id} found open when purging, skipping purge");
1799 return;
1800 }
1801
1802 if let Some(ref pos) = position {
1804 self.positions.remove(&position_id);
1805
1806 if let Some(venue_positions) =
1808 self.index.venue_positions.get_mut(&pos.instrument_id.venue)
1809 {
1810 venue_positions.remove(&position_id);
1811 if venue_positions.is_empty() {
1812 self.index.venue_positions.remove(&pos.instrument_id.venue);
1813 }
1814 }
1815
1816 let instrument_positions_became_empty = self
1818 .index
1819 .instrument_positions
1820 .get_mut(&pos.instrument_id)
1821 .is_some_and(|positions| {
1822 positions.remove(&position_id);
1823 positions.is_empty()
1824 });
1825
1826 if instrument_positions_became_empty {
1827 self.index.instrument_positions.remove(&pos.instrument_id);
1828 let instrument_orders_empty = self
1829 .index
1830 .instrument_orders
1831 .get(&pos.instrument_id)
1832 .is_some_and(|orders| orders.is_empty());
1833
1834 if instrument_orders_empty {
1835 self.index.instrument_orders.remove(&pos.instrument_id);
1836 }
1837 }
1838
1839 let strategy_positions_became_empty = self
1841 .index
1842 .strategy_positions
1843 .get_mut(&pos.strategy_id)
1844 .is_some_and(|positions| {
1845 positions.remove(&position_id);
1846 positions.is_empty()
1847 });
1848
1849 if strategy_positions_became_empty {
1850 self.index.strategy_positions.remove(&pos.strategy_id);
1851 let strategy_orders_empty = self
1852 .index
1853 .strategy_orders
1854 .get(&pos.strategy_id)
1855 .is_some_and(|orders| orders.is_empty());
1856
1857 if strategy_orders_empty {
1858 self.index.strategy_orders.remove(&pos.strategy_id);
1859 self.index.strategies.remove(&pos.strategy_id);
1860 }
1861 }
1862
1863 if let Some(account_positions) = self.index.account_positions.get_mut(&pos.account_id) {
1865 account_positions.remove(&position_id);
1866 if account_positions.is_empty() {
1867 self.index.account_positions.remove(&pos.account_id);
1868 }
1869 }
1870
1871 for client_order_id in pos.client_order_ids() {
1873 self.index.order_position.remove(&client_order_id);
1874 }
1875
1876 log::info!("Purged position {position_id}");
1877 } else {
1878 log::warn!("Position {position_id} not found when purging");
1879 }
1880
1881 self.index.position_strategy.remove(&position_id);
1883 self.index.position_oms.remove(&position_id);
1884 self.index.position_orders.remove(&position_id);
1885 self.index.positions.remove(&position_id);
1886 self.index.positions_open.remove(&position_id);
1887 self.index.positions_closed.remove(&position_id);
1888
1889 self.position_snapshots.remove(&position_id);
1891 self.bump_position_snapshot_revision(position_id);
1892 }
1893
1894 fn purge_instrument_inner(&mut self, instrument_id: InstrumentId, skip_order_guard: bool) {
1918 #[cfg(feature = "defi")]
1919 let defi_found = self.defi.pools.contains_key(&instrument_id)
1920 || self.defi.pool_profilers.contains_key(&instrument_id);
1921 #[cfg(not(feature = "defi"))]
1922 let defi_found = false;
1923
1924 let found = self.instruments.contains_key(&instrument_id)
1925 || self.synthetics.contains_key(&instrument_id)
1926 || defi_found;
1927
1928 if !found {
1929 log::warn!("Instrument {instrument_id} not found when purging");
1930 return;
1931 }
1932
1933 if !skip_order_guard && let Some(orders) = self.index.instrument_orders.get(&instrument_id)
1934 {
1935 let has_non_terminal = orders
1936 .iter()
1937 .any(|client_order_id| !self.index.orders_closed.contains(client_order_id));
1938
1939 if has_non_terminal {
1940 log::warn!(
1941 "Instrument {instrument_id} has non-terminal orders when purging, skipping purge"
1942 );
1943 return;
1944 }
1945 }
1946
1947 if let Some(positions) = self.index.instrument_positions.get(&instrument_id) {
1948 let has_non_closed = positions
1949 .iter()
1950 .any(|position_id| !self.index.positions_closed.contains(position_id));
1951
1952 if has_non_closed {
1953 log::warn!(
1954 "Instrument {instrument_id} has non-closed positions when purging, skipping purge"
1955 );
1956 return;
1957 }
1958 }
1959
1960 self.instruments.remove(&instrument_id);
1961 self.synthetics.remove(&instrument_id);
1962 self.books.remove(&instrument_id);
1963 self.own_books.remove(&instrument_id);
1964 self.quotes.remove(&instrument_id);
1965 self.trades.remove(&instrument_id);
1966 self.mark_prices.remove(&instrument_id);
1967 self.index_prices.remove(&instrument_id);
1968 self.funding_rates.remove(&instrument_id);
1969 self.instrument_statuses.remove(&instrument_id);
1970 self.instrument_closes.remove(&instrument_id);
1971 self.greeks.remove(&instrument_id);
1972 self.option_greeks.remove(&instrument_id);
1973
1974 self.bars
1975 .retain(|bar_type, _| bar_type.instrument_id() != instrument_id);
1976
1977 #[cfg(feature = "defi")]
1978 {
1979 self.defi.pools.remove(&instrument_id);
1980 self.defi.pool_profilers.remove(&instrument_id);
1981 }
1982
1983 log::info!("Purged instrument {instrument_id}");
1984 }
1985
1986 pub fn purge_instrument(&mut self, instrument_id: InstrumentId) {
1991 self.purge_instrument_inner(instrument_id, false);
1992 }
1993
1994 pub fn purge_instrument_skip_order_guard(&mut self, instrument_id: InstrumentId) {
2003 self.purge_instrument_inner(instrument_id, true);
2004 }
2005
2006 pub fn purge_account_events(&mut self, ts_now: UnixNanos, lookback_secs: u64) {
2011 log::debug!(
2012 "Purging account events{}",
2013 if lookback_secs > 0 {
2014 format!(" with lookback_secs={lookback_secs}")
2015 } else {
2016 String::new()
2017 }
2018 );
2019
2020 for account_cell in self.accounts.values() {
2021 let mut account = account_cell.borrow_mut();
2022 let event_count = account.event_count();
2023 account.purge_account_events(ts_now, lookback_secs);
2024 let count_diff = event_count - account.event_count();
2025 if count_diff > 0 {
2026 log::info!(
2027 "Purged {} event(s) from account {}",
2028 count_diff,
2029 account.id()
2030 );
2031 }
2032 }
2033 }
2034
2035 pub fn clear_index(&mut self) {
2037 self.index.clear();
2038 log::debug!("Cleared index");
2039 }
2040
2041 pub fn reset(&mut self) {
2049 log::debug!("Resetting cache");
2050
2051 self.general.clear();
2052 self.books.clear();
2053 self.own_books.clear();
2054 self.quotes.clear();
2055 self.trades.clear();
2056 self.mark_xrates.clear();
2057 self.mark_prices.clear();
2058 self.index_prices.clear();
2059 self.funding_rates.clear();
2060 self.instrument_statuses.clear();
2061 self.instrument_closes.clear();
2062 self.bars.clear();
2063 self.accounts.clear();
2064 self.orders.clear();
2065 self.order_lists.clear();
2066 self.positions.clear();
2067 self.position_snapshots.clear();
2068 self.position_snapshot_revisions.clear();
2069 self.greeks.clear();
2070 self.option_greeks.clear();
2071 self.yield_curves.clear();
2072
2073 if self.config.drop_instruments_on_reset {
2074 self.currencies.clear();
2075 self.instruments.clear();
2076 self.synthetics.clear();
2077 }
2078
2079 #[cfg(feature = "defi")]
2080 {
2081 self.defi.pools.clear();
2082 self.defi.pool_profilers.clear();
2083 }
2084
2085 self.clear_index();
2086
2087 log::info!("Reset cache");
2088 }
2089
2090 pub fn dispose(&mut self) {
2094 self.reset();
2095
2096 if let Some(database) = &mut self.database
2097 && let Err(e) = database.close()
2098 {
2099 log::error!("Failed to close database during dispose: {e}");
2100 }
2101 }
2102
2103 pub fn flush_db(&mut self) -> anyhow::Result<()> {
2111 match &mut self.database {
2112 Some(database) => database.flush(),
2113 None => Ok(()),
2114 }
2115 }
2116
2117 pub fn add(&mut self, key: &str, value: Bytes) -> anyhow::Result<()> {
2125 check_valid_string_utf8(key, stringify!(key))?;
2127 check_predicate_false(value.is_empty(), stringify!(value))?;
2128
2129 log::debug!("Adding general {key}");
2130 self.general.insert(key.to_string(), value.clone());
2131
2132 if let Some(database) = &mut self.database {
2133 database.add(key.to_string(), value)?;
2134 }
2135 Ok(())
2136 }
2137
2138 pub fn add_order_book(&mut self, book: OrderBook) -> anyhow::Result<()> {
2144 log::debug!("Adding `OrderBook` {}", book.instrument_id);
2145
2146 if self.config.save_market_data
2147 && let Some(database) = &mut self.database
2148 {
2149 database.add_order_book(&book)?;
2150 }
2151
2152 self.books.insert(book.instrument_id, book);
2153 Ok(())
2154 }
2155
2156 pub fn add_own_order_book(&mut self, own_book: OwnOrderBook) -> anyhow::Result<()> {
2162 log::debug!("Adding `OwnOrderBook` {}", own_book.instrument_id);
2163
2164 self.own_books.insert(own_book.instrument_id, own_book);
2165 Ok(())
2166 }
2167
2168 pub fn add_mark_price(&mut self, mark_price: MarkPriceUpdate) -> anyhow::Result<()> {
2174 log::debug!("Adding `MarkPriceUpdate` for {}", mark_price.instrument_id);
2175
2176 if self.config.save_market_data {
2177 }
2179
2180 let mark_prices_deque = self
2181 .mark_prices
2182 .entry(mark_price.instrument_id)
2183 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2184 mark_prices_deque.push_front(mark_price);
2185 Ok(())
2186 }
2187
2188 pub fn add_index_price(&mut self, index_price: IndexPriceUpdate) -> anyhow::Result<()> {
2194 log::debug!(
2195 "Adding `IndexPriceUpdate` for {}",
2196 index_price.instrument_id
2197 );
2198
2199 if self.config.save_market_data {
2200 }
2202
2203 let index_prices_deque = self
2204 .index_prices
2205 .entry(index_price.instrument_id)
2206 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2207 index_prices_deque.push_front(index_price);
2208 Ok(())
2209 }
2210
2211 pub fn add_funding_rate(&mut self, funding_rate: FundingRateUpdate) -> anyhow::Result<()> {
2217 log::debug!(
2218 "Adding `FundingRateUpdate` for {}",
2219 funding_rate.instrument_id
2220 );
2221
2222 if self.config.save_market_data {
2223 }
2225
2226 let funding_rates_deque = self
2227 .funding_rates
2228 .entry(funding_rate.instrument_id)
2229 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2230 funding_rates_deque.push_front(funding_rate);
2231 Ok(())
2232 }
2233
2234 pub fn add_funding_rates(&mut self, funding_rates: &[FundingRateUpdate]) -> anyhow::Result<()> {
2240 check_slice_not_empty(funding_rates, stringify!(funding_rates))?;
2241
2242 let instrument_id = funding_rates[0].instrument_id;
2243 log::debug!(
2244 "Adding `FundingRateUpdate`[{}] {instrument_id}",
2245 funding_rates.len()
2246 );
2247
2248 if self.config.save_market_data
2249 && let Some(database) = &mut self.database
2250 {
2251 for funding_rate in funding_rates {
2252 database.add_funding_rate(funding_rate)?;
2253 }
2254 }
2255
2256 let funding_rate_deque = self
2257 .funding_rates
2258 .entry(instrument_id)
2259 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2260
2261 for funding_rate in funding_rates {
2262 funding_rate_deque.push_front(*funding_rate);
2263 }
2264 Ok(())
2265 }
2266
2267 pub fn add_instrument_status(&mut self, status: InstrumentStatus) -> anyhow::Result<()> {
2273 log::debug!("Adding `InstrumentStatus` for {}", status.instrument_id);
2274
2275 if self.config.save_market_data {
2276 }
2278
2279 let statuses_deque = self
2280 .instrument_statuses
2281 .entry(status.instrument_id)
2282 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2283 statuses_deque.push_front(status);
2284 Ok(())
2285 }
2286
2287 pub fn add_instrument_close(&mut self, close: InstrumentClose) -> anyhow::Result<()> {
2297 log::debug!("Adding `InstrumentClose` for {}", close.instrument_id);
2298
2299 if let Some(database) = &self.database {
2300 database.add_instrument_close(&close)?;
2301 }
2302
2303 self.instrument_closes.insert(close.instrument_id, close);
2304 Ok(())
2305 }
2306
2307 pub fn add_quote(&mut self, quote: QuoteTick) -> anyhow::Result<()> {
2313 log::debug!("Adding `QuoteTick` {}", quote.instrument_id);
2314
2315 if self.config.save_market_data
2316 && let Some(database) = &mut self.database
2317 {
2318 database.add_quote("e)?;
2319 }
2320
2321 let quotes_deque = self
2322 .quotes
2323 .entry(quote.instrument_id)
2324 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2325 quotes_deque.push_front(quote);
2326 Ok(())
2327 }
2328
2329 pub fn add_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
2335 check_slice_not_empty(quotes, stringify!(quotes))?;
2336
2337 let instrument_id = quotes[0].instrument_id;
2338 log::debug!("Adding `QuoteTick`[{}] {instrument_id}", quotes.len());
2339
2340 if self.config.save_market_data
2341 && let Some(database) = &mut self.database
2342 {
2343 for quote in quotes {
2344 database.add_quote(quote)?;
2345 }
2346 }
2347
2348 let quotes_deque = self
2349 .quotes
2350 .entry(instrument_id)
2351 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2352
2353 for quote in quotes {
2354 quotes_deque.push_front(*quote);
2355 }
2356 Ok(())
2357 }
2358
2359 pub fn add_trade(&mut self, trade: TradeTick) -> anyhow::Result<()> {
2365 log::debug!("Adding `TradeTick` {}", trade.instrument_id);
2366
2367 if self.config.save_market_data
2368 && let Some(database) = &mut self.database
2369 {
2370 database.add_trade(&trade)?;
2371 }
2372
2373 let trades_deque = self
2374 .trades
2375 .entry(trade.instrument_id)
2376 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2377 trades_deque.push_front(trade);
2378 Ok(())
2379 }
2380
2381 pub fn add_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
2387 check_slice_not_empty(trades, stringify!(trades))?;
2388
2389 let instrument_id = trades[0].instrument_id;
2390 log::debug!("Adding `TradeTick`[{}] {instrument_id}", trades.len());
2391
2392 if self.config.save_market_data
2393 && let Some(database) = &mut self.database
2394 {
2395 for trade in trades {
2396 database.add_trade(trade)?;
2397 }
2398 }
2399
2400 let trades_deque = self
2401 .trades
2402 .entry(instrument_id)
2403 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
2404
2405 for trade in trades {
2406 trades_deque.push_front(*trade);
2407 }
2408 Ok(())
2409 }
2410
2411 pub fn add_bar(&mut self, bar: Bar) -> anyhow::Result<()> {
2420 log::debug!("Adding `Bar` {}", bar.bar_type);
2421
2422 if self.config.save_market_data
2423 && let Some(database) = &mut self.database
2424 {
2425 database.add_bar(&bar)?;
2426 }
2427
2428 let bars = self
2429 .bars
2430 .entry(bar.bar_type)
2431 .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
2432 insert_bar(bars, bar);
2433 Ok(())
2434 }
2435
2436 pub fn add_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
2442 check_slice_not_empty(bars, stringify!(bars))?;
2443
2444 let bar_type = bars[0].bar_type;
2445 log::debug!("Adding `Bar`[{}] {bar_type}", bars.len());
2446
2447 if self.config.save_market_data
2448 && let Some(database) = &mut self.database
2449 {
2450 for bar in bars {
2451 database.add_bar(bar)?;
2452 }
2453 }
2454
2455 let bars_deque = self
2456 .bars
2457 .entry(bar_type)
2458 .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
2459
2460 for bar in bars {
2461 insert_bar(bars_deque, *bar);
2462 }
2463 Ok(())
2464 }
2465
2466 pub fn add_bar_historical(&mut self, bar: Bar) -> anyhow::Result<()> {
2475 log::debug!("Adding historical `Bar` {}", bar.bar_type);
2476
2477 if self.config.save_market_data
2478 && let Some(database) = &mut self.database
2479 {
2480 database.add_bar(&bar)?;
2481 }
2482
2483 let bars = self
2484 .bars
2485 .entry(bar.bar_type)
2486 .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
2487 insert_bar_historical(bars, bar);
2488 Ok(())
2489 }
2490
2491 pub fn add_greeks(&mut self, greeks: GreeksData) -> anyhow::Result<()> {
2497 log::debug!("Adding `GreeksData` {}", greeks.instrument_id);
2498
2499 if self.config.save_market_data
2500 && let Some(_database) = &mut self.database
2501 {
2502 }
2504
2505 self.greeks.insert(greeks.instrument_id, greeks);
2506 Ok(())
2507 }
2508
2509 pub fn greeks(&self, instrument_id: &InstrumentId) -> Option<GreeksData> {
2511 self.greeks.get(instrument_id).cloned()
2512 }
2513
2514 pub fn add_option_greeks(&mut self, greeks: OptionGreeks) {
2516 log::debug!("Adding `OptionGreeks` {}", greeks.instrument_id);
2517 self.option_greeks.insert(greeks.instrument_id, greeks);
2518 }
2519
2520 #[must_use]
2522 pub fn option_greeks(&self, instrument_id: &InstrumentId) -> Option<&OptionGreeks> {
2523 self.option_greeks.get(instrument_id)
2524 }
2525
2526 pub fn add_yield_curve(&mut self, yield_curve: YieldCurve) -> anyhow::Result<()> {
2534 yield_curve.validate()?;
2535
2536 log::debug!("Adding `YieldCurve` {}", yield_curve.curve_name);
2537
2538 if self.config.save_market_data
2539 && let Some(_database) = &mut self.database
2540 {
2541 }
2543
2544 self.yield_curves
2545 .insert(yield_curve.curve_name.clone(), yield_curve);
2546 Ok(())
2547 }
2548
2549 pub fn yield_curve(&self, key: &str) -> Option<Box<dyn Fn(f64) -> f64>> {
2551 self.yield_curves.get(key).map(|curve| {
2552 let curve_clone = curve.clone();
2553 Box::new(move |expiry_in_years: f64| curve_clone.get_rate(expiry_in_years))
2554 as Box<dyn Fn(f64) -> f64>
2555 })
2556 }
2557
2558 pub fn add_currency(&mut self, currency: Currency) -> anyhow::Result<()> {
2564 if self.currencies.contains_key(¤cy.code) {
2565 return Ok(());
2566 }
2567 log::debug!("Adding `Currency` {}", currency.code);
2568
2569 if let Some(database) = &mut self.database {
2570 database.add_currency(¤cy)?;
2571 }
2572
2573 self.currencies.insert(currency.code, currency);
2574 Ok(())
2575 }
2576
2577 pub fn add_instrument(&mut self, instrument: InstrumentAny) -> anyhow::Result<()> {
2583 log::debug!("Adding `Instrument` {}", instrument.id());
2584
2585 if let Some(base_currency) = instrument.base_currency() {
2587 self.add_currency(base_currency)?;
2588 }
2589 self.add_currency(instrument.quote_currency())?;
2590 self.add_currency(instrument.settlement_currency())?;
2591
2592 if let Some(database) = &mut self.database {
2593 database.add_instrument(&instrument)?;
2594 }
2595
2596 self.instruments.insert(instrument.id(), instrument);
2597 Ok(())
2598 }
2599
2600 pub fn add_synthetic(&mut self, synthetic: SyntheticInstrument) -> anyhow::Result<()> {
2606 log::debug!("Adding `SyntheticInstrument` {}", synthetic.id);
2607
2608 if let Some(database) = &mut self.database {
2609 database.add_synthetic(&synthetic)?;
2610 }
2611
2612 self.synthetics.insert(synthetic.id, synthetic);
2613 Ok(())
2614 }
2615
2616 fn state_currencies(state: &AccountState) -> impl Iterator<Item = Currency> + '_ {
2618 state
2619 .balances
2620 .iter()
2621 .map(|balance| balance.currency)
2622 .chain(state.margins.iter().map(|margin| margin.currency))
2623 .chain(state.base_currency)
2624 }
2625
2626 fn add_account_currencies(&mut self, account: &AccountAny) -> anyhow::Result<()> {
2633 let Some(state) = account.last_event() else {
2634 return Ok(());
2635 };
2636
2637 for currency in Self::state_currencies(&state) {
2638 self.add_currency(currency)?;
2639 }
2640 Ok(())
2641 }
2642
2643 pub fn add_account(&mut self, account: AccountAny) -> anyhow::Result<()> {
2649 log::debug!("Adding `Account` {}", account.id());
2650
2651 self.add_account_currencies(&account)?;
2652
2653 if let Some(database) = &mut self.database {
2654 database.add_account(&account)?;
2655 }
2656
2657 let account_id = account.id();
2658 self.accounts.insert(account_id, SharedCell::new(account));
2659 self.index.add_venue_account(account_id);
2660 Ok(())
2661 }
2662
2663 pub fn add_venue_order_id(
2672 &mut self,
2673 client_order_id: &ClientOrderId,
2674 venue_order_id: &VenueOrderId,
2675 overwrite: bool,
2676 ) -> anyhow::Result<()> {
2677 self.validate_venue_order_id_claim(client_order_id, venue_order_id, overwrite)?;
2678
2679 self.index
2680 .client_order_ids
2681 .insert(*client_order_id, *venue_order_id);
2682 self.index
2683 .venue_order_ids
2684 .insert(*venue_order_id, *client_order_id);
2685
2686 Ok(())
2687 }
2688
2689 pub fn index_venue_order_id(
2700 &mut self,
2701 client_order_id: &ClientOrderId,
2702 venue_order_id: &VenueOrderId,
2703 ) -> anyhow::Result<()> {
2704 self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
2705
2706 self.index
2707 .venue_order_ids
2708 .insert(*venue_order_id, *client_order_id);
2709 self.index
2710 .client_order_ids
2711 .entry(*client_order_id)
2712 .or_insert(*venue_order_id);
2713
2714 Ok(())
2715 }
2716
2717 fn validate_venue_order_id_claim(
2718 &self,
2719 client_order_id: &ClientOrderId,
2720 venue_order_id: &VenueOrderId,
2721 overwrite: bool,
2722 ) -> anyhow::Result<()> {
2723 self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
2724
2725 if let Some(existing_venue_order_id) = self.index.client_order_ids.get(client_order_id)
2726 && !overwrite
2727 && existing_venue_order_id != venue_order_id
2728 {
2729 anyhow::bail!(
2730 "Existing {existing_venue_order_id} for {client_order_id}
2731 did not match the given {venue_order_id}.
2732 If you are writing a test then try a different `venue_order_id`,
2733 otherwise this is probably a bug."
2734 );
2735 }
2736
2737 Ok(())
2738 }
2739
2740 fn validate_venue_order_id_ownership(
2741 &self,
2742 client_order_id: &ClientOrderId,
2743 venue_order_id: &VenueOrderId,
2744 ) -> anyhow::Result<()> {
2745 if let Some(existing_client_order_id) = self.index.venue_order_ids.get(venue_order_id)
2746 && existing_client_order_id != client_order_id
2747 {
2748 return Err(VenueOrderIdOwnershipError {
2749 venue_order_id: *venue_order_id,
2750 existing_client_order_id: *existing_client_order_id,
2751 claimant_client_order_id: *client_order_id,
2752 }
2753 .into());
2754 }
2755
2756 Ok(())
2757 }
2758
2759 pub fn add_order(
2774 &mut self,
2775 order: OrderAny,
2776 position_id: Option<PositionId>,
2777 client_id: Option<ClientId>,
2778 replace_existing: bool,
2779 ) -> anyhow::Result<()> {
2780 let instrument_id = order.instrument_id();
2781 let venue = instrument_id.venue;
2782 let client_order_id = order.client_order_id();
2783 let strategy_id = order.strategy_id();
2784 let exec_algorithm_id = order.exec_algorithm_id();
2785 let exec_spawn_id = order.exec_spawn_id();
2786
2787 if !replace_existing {
2788 check_key_not_in_map(
2789 &client_order_id,
2790 &self.orders,
2791 stringify!(client_order_id),
2792 stringify!(orders),
2793 )?;
2794 }
2795
2796 log::debug!("Adding {order:?}");
2797
2798 self.index.orders.insert(client_order_id);
2799
2800 if order.is_active_local() {
2801 self.index.orders_active_local.insert(client_order_id);
2802 }
2803 self.index
2804 .order_strategy
2805 .insert(client_order_id, strategy_id);
2806 self.index.strategies.insert(strategy_id);
2807
2808 self.index
2810 .venue_orders
2811 .entry(venue)
2812 .or_default()
2813 .insert(client_order_id);
2814
2815 self.index
2817 .instrument_orders
2818 .entry(instrument_id)
2819 .or_default()
2820 .insert(client_order_id);
2821
2822 self.index
2824 .strategy_orders
2825 .entry(strategy_id)
2826 .or_default()
2827 .insert(client_order_id);
2828
2829 if let Some(account_id) = order.account_id() {
2831 self.index
2832 .account_orders
2833 .entry(account_id)
2834 .or_default()
2835 .insert(client_order_id);
2836 }
2837
2838 if let Some(exec_algorithm_id) = exec_algorithm_id {
2840 self.index.exec_algorithms.insert(exec_algorithm_id);
2841
2842 self.index
2843 .exec_algorithm_orders
2844 .entry(exec_algorithm_id)
2845 .or_default()
2846 .insert(client_order_id);
2847 }
2848
2849 if let Some(exec_spawn_id) = exec_spawn_id {
2851 self.index
2852 .exec_spawn_orders
2853 .entry(exec_spawn_id)
2854 .or_default()
2855 .insert(client_order_id);
2856 }
2857
2858 if order.emulation_trigger().is_some() {
2860 self.index.orders_emulated.insert(client_order_id);
2861 }
2862
2863 if let Some(position_id) = position_id {
2865 self.index_position_id_in_memory(&position_id, &venue, &client_order_id, &strategy_id);
2866 }
2867
2868 if let Some(client_id) = client_id {
2870 self.index.order_client.insert(client_order_id, client_id);
2871 log::debug!("Indexed {client_id:?}");
2872 }
2873
2874 let order_cell = if let Some(order_cell) = self.orders.get(&client_order_id) {
2877 *order_cell.borrow_mut() = order;
2878 order_cell.clone()
2879 } else {
2880 let order_cell = SharedCell::new(order);
2881 self.orders.insert(client_order_id, order_cell.clone());
2882 order_cell
2883 };
2884
2885 if let Some(position_id) = position_id {
2886 self.persist_position_id(&position_id, &client_order_id)?;
2887 }
2888
2889 if let Some(database) = &mut self.database {
2890 database.add_order(&order_cell.borrow(), client_id)?;
2891 }
2896
2897 Ok(())
2898 }
2899
2900 pub fn claim_order_clients(
2912 &mut self,
2913 claims: &[(ClientOrderId, ClientId)],
2914 ) -> anyhow::Result<()> {
2915 let mut requested = AHashMap::with_capacity(claims.len());
2916 let mut ordered_claims = Vec::with_capacity(claims.len());
2917
2918 for (client_order_id, client_id) in claims {
2919 if let Some(existing_client_id) = requested.get(client_order_id) {
2920 if existing_client_id != client_id {
2921 anyhow::bail!(
2922 "Conflicting execution client claims for {client_order_id}: \
2923 {existing_client_id} and {client_id}"
2924 );
2925 }
2926 continue;
2927 }
2928
2929 requested.insert(*client_order_id, *client_id);
2930 ordered_claims.push((*client_order_id, *client_id));
2931 }
2932
2933 let mut pending_claims = Vec::with_capacity(ordered_claims.len());
2934 for (client_order_id, client_id) in ordered_claims {
2935 if !self.orders.contains_key(&client_order_id) {
2936 return Err(OrderLookupError::not_found(client_order_id).into());
2937 }
2938
2939 match self.index.order_client.get(&client_order_id) {
2940 Some(existing_client_id) if *existing_client_id == client_id => {}
2941 Some(existing_client_id) => {
2942 anyhow::bail!(
2943 "Order {client_order_id} is already claimed by execution client \
2944 {existing_client_id} and cannot be claimed by {client_id}"
2945 );
2946 }
2947 None => pending_claims.push((client_order_id, client_id)),
2948 }
2949 }
2950
2951 if pending_claims.is_empty() {
2952 return Ok(());
2953 }
2954
2955 if let Some(database) = &self.database {
2956 database.index_order_clients(&pending_claims)?;
2957 }
2958
2959 for (client_order_id, client_id) in pending_claims {
2960 self.index.order_client.insert(client_order_id, client_id);
2961 log::debug!("Claimed {client_order_id} for execution client {client_id}");
2962 }
2963
2964 Ok(())
2965 }
2966
2967 pub fn add_order_list(&mut self, order_list: OrderList) -> anyhow::Result<()> {
2973 let order_list_id = order_list.id;
2974 check_key_not_in_map(
2975 &order_list_id,
2976 &self.order_lists,
2977 stringify!(order_list_id),
2978 stringify!(order_lists),
2979 )?;
2980
2981 log::debug!("Adding {order_list}");
2982 self.order_lists.insert(order_list_id, order_list);
2983 Ok(())
2984 }
2985
2986 pub fn add_position_id(
2997 &mut self,
2998 position_id: &PositionId,
2999 venue: &Venue,
3000 client_order_id: &ClientOrderId,
3001 strategy_id: &StrategyId,
3002 ) -> anyhow::Result<()> {
3003 self.index_position_id_in_memory(position_id, venue, client_order_id, strategy_id);
3004 self.persist_position_id(position_id, client_order_id)
3005 }
3006
3007 fn index_position_id_in_memory(
3008 &mut self,
3009 position_id: &PositionId,
3010 venue: &Venue,
3011 client_order_id: &ClientOrderId,
3012 strategy_id: &StrategyId,
3013 ) {
3014 self.index
3015 .order_position
3016 .insert(*client_order_id, *position_id);
3017 self.index_position(position_id, venue, strategy_id);
3018 self.index
3019 .position_orders
3020 .entry(*position_id)
3021 .or_default()
3022 .insert(*client_order_id);
3023 }
3024
3025 fn persist_position_id(
3026 &mut self,
3027 position_id: &PositionId,
3028 client_order_id: &ClientOrderId,
3029 ) -> anyhow::Result<()> {
3030 if let Some(database) = &mut self.database {
3031 database.index_order_position(*client_order_id, *position_id)?;
3032 }
3033
3034 Ok(())
3035 }
3036
3037 fn index_position(
3038 &mut self,
3039 position_id: &PositionId,
3040 venue: &Venue,
3041 strategy_id: &StrategyId,
3042 ) {
3043 let strategy_id = self
3044 .positions
3045 .get(position_id)
3046 .map(|position| position.borrow().strategy_id)
3047 .filter(StrategyId::is_external)
3048 .unwrap_or(*strategy_id);
3049
3050 self.index
3052 .position_strategy
3053 .insert(*position_id, strategy_id);
3054
3055 self.index.position_orders.entry(*position_id).or_default();
3057
3058 self.index
3060 .strategy_positions
3061 .entry(strategy_id)
3062 .or_default()
3063 .insert(*position_id);
3064
3065 self.index
3067 .venue_positions
3068 .entry(*venue)
3069 .or_default()
3070 .insert(*position_id);
3071 }
3072
3073 fn assign_position_ids_to_contingencies(&mut self) {
3081 let mut assignments: Vec<(PositionId, ClientOrderId)> = Vec::new();
3082
3083 for parent_order_cell in self.orders.values() {
3084 let parent = parent_order_cell.borrow();
3085 if parent.contingency_type() != Some(ContingencyType::Oto) {
3086 continue;
3087 }
3088 let Some(parent_position_id) = parent.position_id() else {
3089 continue;
3090 };
3091 let Some(linked_order_ids) = parent.linked_order_ids() else {
3092 continue;
3093 };
3094
3095 for client_order_id in linked_order_ids {
3096 match self.orders.get(client_order_id) {
3097 None => {
3098 log::error!("Contingency order {client_order_id} not found");
3099 }
3100 Some(contingent_order_cell) => {
3101 if contingent_order_cell.borrow().position_id().is_none() {
3102 assignments.push((parent_position_id, *client_order_id));
3103 }
3104 }
3105 }
3106 }
3107 }
3108
3109 for (position_id, client_order_id) in assignments {
3110 let Some((venue, strategy_id)) = self.orders.get(&client_order_id).map(|order_cell| {
3111 let mut contingent = order_cell.borrow_mut();
3112 contingent.set_position_id(Some(position_id));
3113 (contingent.instrument_id().venue, contingent.strategy_id())
3114 }) else {
3115 continue;
3116 };
3117
3118 if let Err(e) =
3121 self.add_position_id(&position_id, &venue, &client_order_id, &strategy_id)
3122 {
3123 log::error!("Failed to re-index {client_order_id} -> {position_id}: {e}");
3124 }
3125 }
3126 }
3127
3128 pub fn add_position(&mut self, position: &Position, oms_type: OmsType) -> anyhow::Result<()> {
3136 self.add_position_inner(position.clone(), oms_type, true, false)
3137 }
3138
3139 pub fn add_position_without_order(
3147 &mut self,
3148 position: &Position,
3149 oms_type: OmsType,
3150 ) -> anyhow::Result<()> {
3151 self.add_position_inner(position.clone(), oms_type, false, false)
3152 }
3153
3154 pub fn replace_position(
3165 &mut self,
3166 position: &Position,
3167 oms_type: OmsType,
3168 index_order: bool,
3169 carry_replay_state: bool,
3170 ) -> anyhow::Result<()> {
3171 self.add_position_inner(position.clone(), oms_type, index_order, carry_replay_state)
3172 }
3173
3174 fn add_position_inner(
3175 &mut self,
3176 mut position: Position,
3177 oms_type: OmsType,
3178 index_order: bool,
3179 carry_replay_state: bool,
3180 ) -> anyhow::Result<()> {
3181 let key = position_oms_key(position.id);
3186 check_valid_string_utf8(&key, stringify!(key))?;
3187 let value = Bytes::from(serde_json::to_vec(&oms_type)?);
3188 check_predicate_false(value.is_empty(), stringify!(value))?;
3189
3190 let position_id = position.id;
3191 let strategy_id = position.strategy_id;
3192 let instrument_id = position.instrument_id;
3193 let account_id = position.account_id;
3194 let opening_order_id = position.opening_order_id;
3195
3196 log::debug!("Adding {position}");
3197
3198 let position_cell = if let Some(position_cell) = self.positions.get(&position_id).cloned() {
3202 let mut prior = position_cell.borrow_mut();
3203 if carry_replay_state {
3204 position.transfer_replay_state_from(&mut prior);
3205 }
3206 *prior = position;
3207 drop(prior);
3208 position_cell
3209 } else {
3210 let position_cell = SharedCell::new(position);
3211 self.positions.insert(position_id, position_cell.clone());
3212 position_cell
3213 };
3214
3215 self.index.position_oms.insert(position_id, oms_type);
3216 self.index.positions.insert(position_id);
3217 self.index.positions_open.insert(position_id);
3218 self.index.positions_closed.remove(&position_id); self.index.strategies.insert(strategy_id);
3220 self.index.strategy_orders.entry(strategy_id).or_default();
3221
3222 if index_order {
3223 self.index_position_id_in_memory(
3224 &position_id,
3225 &instrument_id.venue,
3226 &opening_order_id,
3227 &strategy_id,
3228 );
3229 } else {
3230 self.index_position(&position_id, &instrument_id.venue, &strategy_id);
3231 }
3232
3233 let instrument_positions = self
3235 .index
3236 .instrument_positions
3237 .entry(instrument_id)
3238 .or_default();
3239 instrument_positions.insert(position_id);
3240 self.index
3241 .instrument_orders
3242 .entry(instrument_id)
3243 .or_default();
3244
3245 self.index
3247 .account_positions
3248 .entry(account_id)
3249 .or_default()
3250 .insert(position_id);
3251
3252 log::debug!("Adding general {key}");
3253 self.general.insert(key.clone(), value.clone());
3254
3255 if index_order {
3256 self.persist_position_id(&position_id, &opening_order_id)?;
3257 }
3258
3259 if let Some(database) = &mut self.database {
3260 database.add_position(&position_cell.borrow())?;
3261 database.add(key, value)?;
3270 }
3271
3272 Ok(())
3273 }
3274
3275 pub fn update_account(&mut self, account: &AccountAny) -> anyhow::Result<()> {
3284 let account_id = account.id();
3285 match self.accounts.get(&account_id) {
3286 Some(account_cell) => *account_cell.borrow_mut() = account.clone(),
3287 None => {
3288 self.accounts
3289 .insert(account_id, SharedCell::new(account.clone()));
3290 }
3291 }
3292
3293 self.add_account_currencies(account)?;
3294
3295 if let Some(database) = &mut self.database {
3296 database.update_account(account)?;
3297 }
3298 Ok(())
3299 }
3300
3301 #[must_use]
3312 pub fn take_account(&mut self, account_id: &AccountId) -> Option<AccountAny> {
3313 let cell = self.accounts.remove(account_id)?;
3314 let rc: Rc<RefCell<AccountAny>> = cell.into();
3315
3316 match Rc::try_unwrap(rc) {
3317 Ok(cell) => Some(cell.into_inner()),
3318 Err(rc) => {
3319 log::error!(
3320 "Cannot move account {account_id} out of cache: account cell has an outstanding owner"
3321 );
3322 self.accounts.insert(*account_id, rc.into());
3323 None
3324 }
3325 }
3326 }
3327
3328 pub fn cache_account_owned(&mut self, account: AccountAny) {
3330 let account_id = account.id();
3331 self.index.add_venue_account(account_id);
3332 match self.accounts.get(&account_id) {
3333 Some(account_cell) => *account_cell.borrow_mut() = account,
3334 None => {
3335 self.accounts.insert(account_id, SharedCell::new(account));
3336 }
3337 }
3338 }
3339
3340 pub fn update_account_owned(&mut self, account: AccountAny) -> anyhow::Result<()> {
3346 let account_id = account.id();
3347 let currencies: Vec<Currency> = account
3350 .last_event()
3351 .map(|state| Self::state_currencies(&state).collect())
3352 .unwrap_or_default();
3353 self.cache_account_owned(account);
3354
3355 for currency in currencies {
3356 self.add_currency(currency)?;
3357 }
3358
3359 if let Some(database) = &mut self.database {
3360 let Some(account_cell) = self.accounts.get(&account_id) else {
3361 anyhow::bail!("Account {account_id} not found after cache update");
3362 };
3363 database.update_account(&account_cell.borrow())?;
3364 }
3365 Ok(())
3366 }
3367
3368 pub fn update_account_state(&mut self, event: &AccountState) -> anyhow::Result<()> {
3378 let Some(cell) = self.accounts.get(&event.account_id) else {
3379 return self.add_account(AccountAny::from_events(std::slice::from_ref(event))?);
3380 };
3381
3382 cell.borrow_mut().apply(event.clone())?;
3385
3386 for currency in Self::state_currencies(event) {
3389 self.add_currency(currency)?;
3390 }
3391
3392 let Some(cell) = self.accounts.get(&event.account_id) else {
3393 anyhow::bail!("Account {} not found after apply", event.account_id);
3394 };
3395
3396 if let Some(database) = &mut self.database {
3397 database.update_account(&cell.borrow())?;
3398 }
3399 Ok(())
3400 }
3401
3402 pub fn replace_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
3413 let client_order_id = order.client_order_id();
3414 if let Some(venue_order_id) = order.venue_order_id() {
3415 self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
3416 }
3417
3418 match self.orders.get(&client_order_id) {
3419 Some(order_cell) => *order_cell.borrow_mut() = order.clone(),
3422 None => {
3423 self.orders
3424 .insert(client_order_id, SharedCell::new(order.clone()));
3425 }
3426 }
3427
3428 self.refresh_order(order)
3429 }
3430
3431 pub fn update_order(&mut self, event: &OrderEventAny) -> anyhow::Result<OrderAny> {
3437 let event_client_order_id = event.client_order_id();
3438 let client_order_id = if self.order_exists(&event_client_order_id) {
3439 event_client_order_id
3440 } else if let Some(venue_order_id) = event.venue_order_id() {
3441 self.index
3442 .venue_order_ids
3443 .get(&venue_order_id)
3444 .copied()
3445 .ok_or(OrderError::NotFound(event_client_order_id))?
3446 } else {
3447 return Err(OrderError::NotFound(event_client_order_id).into());
3448 };
3449
3450 let order_cell = self
3451 .orders
3452 .get(&client_order_id)
3453 .cloned()
3454 .ok_or(OrderError::NotFound(client_order_id))?;
3455
3456 let mut snapshot = order_cell.borrow().clone();
3460 snapshot.apply(event.clone())?;
3461
3462 if let Some(venue_order_id) = snapshot.venue_order_id() {
3466 self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
3467 }
3468
3469 *order_cell.borrow_mut() = snapshot.clone();
3470
3471 if matches!(event, OrderEventAny::FillVoided(_)) {
3472 self.reindex_hedging_position(client_order_id);
3473 }
3474
3475 if let Err(e) = self.refresh_order(&snapshot) {
3476 log::error!("Error updating order in cache: {e}");
3477 }
3478
3479 Ok(snapshot)
3480 }
3481
3482 fn refresh_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
3483 let client_order_id = order.client_order_id();
3484
3485 if let Some(venue_order_id) = order.venue_order_id() {
3488 let overwrite = matches!(order.last_event(), OrderEventAny::Updated(_));
3489 if let Err(e) = self.add_venue_order_id(&client_order_id, &venue_order_id, overwrite) {
3490 if e.is::<VenueOrderIdOwnershipError>() {
3491 return Err(e);
3492 }
3493 log::error!("Error indexing venue order ID in cache: {e}");
3494 }
3495 }
3496
3497 if order.is_active_local() {
3498 self.index.orders_active_local.insert(client_order_id);
3499 } else {
3500 self.index.orders_active_local.remove(&client_order_id);
3501 }
3502
3503 if order.is_inflight() {
3505 self.index.orders_inflight.insert(client_order_id);
3506 } else {
3507 self.index.orders_inflight.remove(&client_order_id);
3508 }
3509
3510 if order.is_open() {
3512 self.index.orders_closed.remove(&client_order_id);
3513 self.index.orders_open.insert(client_order_id);
3514 } else {
3515 self.index.orders_open.remove(&client_order_id);
3516 }
3517
3518 if order.is_closed() {
3519 self.index.orders_pending_cancel.remove(&client_order_id);
3520 self.index.orders_closed.insert(client_order_id);
3521 }
3522
3523 if matches!(order.last_event(), OrderEventAny::CancelRejected(_)) {
3525 self.index.orders_pending_cancel.remove(&client_order_id);
3526 }
3527
3528 if order.emulation_trigger().is_some() && !order.is_closed() {
3530 self.index.orders_emulated.insert(client_order_id);
3531 } else {
3532 self.index.orders_emulated.remove(&client_order_id);
3533 }
3534
3535 if let Some(account_id) = order.account_id() {
3537 self.index
3538 .account_orders
3539 .entry(account_id)
3540 .or_default()
3541 .insert(client_order_id);
3542 }
3543
3544 if !self.own_books.is_empty() {
3546 let own_book = self.own_order_book(&order.instrument_id());
3547 if (own_book.is_some() && order.is_closed()) || should_handle_own_book_order(order) {
3548 self.update_own_order_book(order);
3549 }
3550 }
3551
3552 if let Some(database) = &mut self.database {
3553 database.update_order(order.last_event())?;
3554 }
3559
3560 Ok(())
3561 }
3562
3563 pub fn update_order_pending_cancel_local(&mut self, order: &OrderAny) {
3565 self.index
3566 .orders_pending_cancel
3567 .insert(order.client_order_id());
3568 }
3569
3570 pub fn update_position(&mut self, position: &Position) -> anyhow::Result<()> {
3580 let Some(position_cell) = self.positions.get(&position.id).cloned() else {
3581 anyhow::bail!("Cannot update position {}: not found in cache", position.id);
3582 };
3583
3584 self.refresh_position_indexes(position);
3585
3586 *position_cell.borrow_mut() = position.clone();
3587
3588 if let Some(database) = &mut self.database {
3589 database.update_position(position)?;
3590 }
3595
3596 Ok(())
3597 }
3598
3599 pub fn update_position_from_fill(
3609 &mut self,
3610 position_id: PositionId,
3611 fill: &OrderFilled,
3612 ) -> anyhow::Result<Position> {
3613 let Some(position_cell) = self.positions.get(&position_id).cloned() else {
3614 anyhow::bail!("Cannot update position {position_id}: not found in cache");
3615 };
3616
3617 let position = {
3618 let mut position = position_cell.borrow_mut();
3619 position.apply(fill);
3620 position.clone_without_events()
3621 };
3622
3623 self.refresh_position_indexes(&position);
3624
3625 if let Some(database) = &mut self.database {
3626 database.update_position(&position_cell.borrow())?;
3627 }
3628
3629 Ok(position)
3630 }
3631
3632 pub fn update_position_from_adjustment(
3643 &mut self,
3644 position_id: PositionId,
3645 adjustment: PositionAdjusted,
3646 ) -> anyhow::Result<()> {
3647 let Some(position_cell) = self.positions.get(&position_id).cloned() else {
3648 anyhow::bail!("Cannot update position {position_id}: not found in cache");
3649 };
3650
3651 position_cell.borrow_mut().apply_adjustment(adjustment);
3652 self.refresh_position_indexes(&position_cell.borrow());
3653
3654 if let Some(database) = &mut self.database {
3655 database.update_position(&position_cell.borrow())?;
3656 }
3657
3658 Ok(())
3659 }
3660
3661 pub fn revert_position_adjustment(&mut self, prior: &Position) -> anyhow::Result<()> {
3672 let Some(position_cell) = self.positions.get(&prior.id).cloned() else {
3673 anyhow::bail!("Cannot revert position {}: not found in cache", prior.id);
3674 };
3675
3676 position_cell.borrow_mut().revert_last_adjustment(prior)?;
3677 self.refresh_position_indexes(&position_cell.borrow());
3678
3679 if let Some(database) = &mut self.database {
3680 database.update_position(&position_cell.borrow())?;
3681 }
3682
3683 Ok(())
3684 }
3685
3686 pub fn update_position_from_instrument_close(
3698 &mut self,
3699 position_id: PositionId,
3700 close: InstrumentClose,
3701 ) -> anyhow::Result<Position> {
3702 let Some(position_cell) = self.positions.get(&position_id).cloned() else {
3703 anyhow::bail!("Cannot update position {position_id}: not found in cache");
3704 };
3705
3706 let position = {
3707 let mut position = position_cell.borrow_mut();
3708 position.apply_instrument_close(close)?;
3709 position.clone_without_events()
3710 };
3711
3712 self.refresh_position_indexes(&position);
3713
3714 if let Some(database) = &mut self.database
3715 && let Err(e) = database.update_position(&position_cell.borrow())
3716 {
3717 log::error!("Failed to persist settled position {position_id}: {e}");
3718 }
3719
3720 Ok(position)
3721 }
3722
3723 fn refresh_position_indexes(&mut self, position: &Position) {
3724 if position.is_open() {
3725 self.index.positions_open.insert(position.id);
3726 self.index.positions_closed.remove(&position.id);
3727 } else {
3728 self.index.positions_closed.insert(position.id);
3729 self.index.positions_open.remove(&position.id);
3730 }
3731 }
3732
3733 #[must_use]
3735 pub fn oms_type(&self, position_id: &PositionId) -> Option<OmsType> {
3736 self.index.position_oms.get(position_id).copied()
3737 }
3738
3739 pub fn snapshot_order_state(&self, order: &OrderAny) -> anyhow::Result<()> {
3745 let Some(database) = &self.database else {
3746 log::warn!(
3747 "Cannot snapshot order state for {} (no database configured)",
3748 order.client_order_id()
3749 );
3750 return Ok(());
3751 };
3752
3753 database.snapshot_order_state(order)
3754 }
3755
3756 fn collect_order_filter_sources<'a>(
3767 &'a self,
3768 venue: Option<&Venue>,
3769 instrument_id: Option<&InstrumentId>,
3770 strategy_id: Option<&StrategyId>,
3771 account_id: Option<&AccountId>,
3772 ) -> FilterSources<'a, ClientOrderId> {
3773 let mut sources: Vec<&AHashSet<ClientOrderId>> = Vec::with_capacity(4);
3774
3775 if let Some(venue) = venue {
3776 match self.index.venue_orders.get(venue) {
3777 Some(set) => sources.push(set),
3778 None => return FilterSources::Empty,
3779 }
3780 }
3781
3782 if let Some(instrument_id) = instrument_id {
3783 match self.index.instrument_orders.get(instrument_id) {
3784 Some(set) => sources.push(set),
3785 None => return FilterSources::Empty,
3786 }
3787 }
3788
3789 if let Some(strategy_id) = strategy_id {
3790 match self.index.strategy_orders.get(strategy_id) {
3791 Some(set) => sources.push(set),
3792 None => return FilterSources::Empty,
3793 }
3794 }
3795
3796 if let Some(account_id) = account_id {
3797 match self.index.account_orders.get(account_id) {
3798 Some(set) => sources.push(set),
3799 None => return FilterSources::Empty,
3800 }
3801 }
3802
3803 if sources.is_empty() {
3804 FilterSources::Unfiltered
3805 } else {
3806 FilterSources::Sets(sources)
3807 }
3808 }
3809
3810 fn collect_position_filter_sources<'a>(
3811 &'a self,
3812 venue: Option<&Venue>,
3813 instrument_id: Option<&InstrumentId>,
3814 strategy_id: Option<&StrategyId>,
3815 account_id: Option<&AccountId>,
3816 ) -> FilterSources<'a, PositionId> {
3817 let mut sources: Vec<&AHashSet<PositionId>> = Vec::with_capacity(4);
3818
3819 if let Some(venue) = venue {
3820 match self.index.venue_positions.get(venue) {
3821 Some(set) => sources.push(set),
3822 None => return FilterSources::Empty,
3823 }
3824 }
3825
3826 if let Some(instrument_id) = instrument_id {
3827 match self.index.instrument_positions.get(instrument_id) {
3828 Some(set) => sources.push(set),
3829 None => return FilterSources::Empty,
3830 }
3831 }
3832
3833 if let Some(strategy_id) = strategy_id {
3834 match self.index.strategy_positions.get(strategy_id) {
3835 Some(set) => sources.push(set),
3836 None => return FilterSources::Empty,
3837 }
3838 }
3839
3840 if let Some(account_id) = account_id {
3841 match self.index.account_positions.get(account_id) {
3842 Some(set) => sources.push(set),
3843 None => return FilterSources::Empty,
3844 }
3845 }
3846
3847 if sources.is_empty() {
3848 FilterSources::Unfiltered
3849 } else {
3850 FilterSources::Sets(sources)
3851 }
3852 }
3853
3854 fn query_orders_in_bucket(
3860 &self,
3861 bucket: &AHashSet<ClientOrderId>,
3862 venue: Option<&Venue>,
3863 instrument_id: Option<&InstrumentId>,
3864 strategy_id: Option<&StrategyId>,
3865 account_id: Option<&AccountId>,
3866 ) -> AHashSet<ClientOrderId> {
3867 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3868 FilterSources::Empty => AHashSet::new(),
3869 FilterSources::Unfiltered => bucket.clone(),
3870 FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
3871 }
3872 }
3873
3874 fn query_positions_in_bucket(
3875 &self,
3876 bucket: &AHashSet<PositionId>,
3877 venue: Option<&Venue>,
3878 instrument_id: Option<&InstrumentId>,
3879 strategy_id: Option<&StrategyId>,
3880 account_id: Option<&AccountId>,
3881 ) -> AHashSet<PositionId> {
3882 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3883 FilterSources::Empty => AHashSet::new(),
3884 FilterSources::Unfiltered => bucket.clone(),
3885 FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
3886 }
3887 }
3888
3889 fn view_orders_in_bucket<'a>(
3892 &'a self,
3893 bucket: &'a AHashSet<ClientOrderId>,
3894 venue: Option<&Venue>,
3895 instrument_id: Option<&InstrumentId>,
3896 strategy_id: Option<&StrategyId>,
3897 account_id: Option<&AccountId>,
3898 ) -> Cow<'a, AHashSet<ClientOrderId>> {
3899 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3900 FilterSources::Empty => Cow::Owned(AHashSet::new()),
3901 FilterSources::Unfiltered => Cow::Borrowed(bucket),
3902 FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
3903 }
3904 }
3905
3906 fn view_positions_in_bucket<'a>(
3907 &'a self,
3908 bucket: &'a AHashSet<PositionId>,
3909 venue: Option<&Venue>,
3910 instrument_id: Option<&InstrumentId>,
3911 strategy_id: Option<&StrategyId>,
3912 account_id: Option<&AccountId>,
3913 ) -> Cow<'a, AHashSet<PositionId>> {
3914 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3915 FilterSources::Empty => Cow::Owned(AHashSet::new()),
3916 FilterSources::Unfiltered => Cow::Borrowed(bucket),
3917 FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
3918 }
3919 }
3920
3921 fn iter_orders_in_bucket<'a>(
3926 &'a self,
3927 bucket: &'a AHashSet<ClientOrderId>,
3928 venue: Option<&Venue>,
3929 instrument_id: Option<&InstrumentId>,
3930 strategy_id: Option<&StrategyId>,
3931 account_id: Option<&AccountId>,
3932 ) -> Box<dyn Iterator<Item = ClientOrderId> + 'a> {
3933 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3934 FilterSources::Empty => Box::new(std::iter::empty()),
3935 FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
3936 FilterSources::Sets(mut sources) => {
3937 sources.push(bucket);
3938 sources.sort_unstable_by_key(|s| s.len());
3939 let driver = sources[0];
3940 let rest: Vec<&'a AHashSet<ClientOrderId>> = sources[1..].to_vec();
3941 Box::new(
3942 driver
3943 .iter()
3944 .copied()
3945 .filter(move |id| rest.iter().all(|s| s.contains(id))),
3946 )
3947 }
3948 }
3949 }
3950
3951 fn iter_positions_in_bucket<'a>(
3952 &'a self,
3953 bucket: &'a AHashSet<PositionId>,
3954 venue: Option<&Venue>,
3955 instrument_id: Option<&InstrumentId>,
3956 strategy_id: Option<&StrategyId>,
3957 account_id: Option<&AccountId>,
3958 ) -> Box<dyn Iterator<Item = PositionId> + 'a> {
3959 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
3960 FilterSources::Empty => Box::new(std::iter::empty()),
3961 FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
3962 FilterSources::Sets(mut sources) => {
3963 sources.push(bucket);
3964 sources.sort_unstable_by_key(|s| s.len());
3965 let driver = sources[0];
3966 let rest: Vec<&'a AHashSet<PositionId>> = sources[1..].to_vec();
3967 Box::new(
3968 driver
3969 .iter()
3970 .copied()
3971 .filter(move |id| rest.iter().all(|s| s.contains(id))),
3972 )
3973 }
3974 }
3975 }
3976
3977 fn count_orders_in_bucket(
3983 &self,
3984 bucket: &AHashSet<ClientOrderId>,
3985 venue: Option<&Venue>,
3986 instrument_id: Option<&InstrumentId>,
3987 strategy_id: Option<&StrategyId>,
3988 account_id: Option<&AccountId>,
3989 side: Option<OrderSide>,
3990 ) -> usize {
3991 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
3992 FilterSources::Empty => 0,
3993 FilterSources::Unfiltered => side.map_or_else(
3994 || bucket.len(),
3995 |side| {
3996 bucket
3997 .iter()
3998 .filter(|id| self.order_side_matches(id, side))
3999 .count()
4000 },
4001 ),
4002 FilterSources::Sets(mut sources) => {
4003 sources.push(bucket);
4004 sources.sort_unstable_by_key(|s| s.len());
4005 let driver = sources[0];
4006 let rest = &sources[1..];
4007
4008 driver
4009 .iter()
4010 .filter(|id| rest.iter().all(|s| s.contains(id)))
4011 .filter(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
4012 .count()
4013 }
4014 }
4015 }
4016
4017 fn count_positions_in_bucket(
4018 &self,
4019 bucket: &AHashSet<PositionId>,
4020 venue: Option<&Venue>,
4021 instrument_id: Option<&InstrumentId>,
4022 strategy_id: Option<&StrategyId>,
4023 account_id: Option<&AccountId>,
4024 side: Option<PositionSide>,
4025 ) -> usize {
4026 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
4027 FilterSources::Empty => 0,
4028 FilterSources::Unfiltered => side.map_or_else(
4029 || bucket.len(),
4030 |side| {
4031 bucket
4032 .iter()
4033 .filter(|id| self.position_side_matches(id, side))
4034 .count()
4035 },
4036 ),
4037 FilterSources::Sets(mut sources) => {
4038 sources.push(bucket);
4039 sources.sort_unstable_by_key(|s| s.len());
4040 let driver = sources[0];
4041 let rest = &sources[1..];
4042
4043 driver
4044 .iter()
4045 .filter(|id| rest.iter().all(|s| s.contains(id)))
4046 .filter(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
4047 .count()
4048 }
4049 }
4050 }
4051
4052 fn any_orders_in_bucket(
4058 &self,
4059 bucket: &AHashSet<ClientOrderId>,
4060 venue: Option<&Venue>,
4061 instrument_id: Option<&InstrumentId>,
4062 strategy_id: Option<&StrategyId>,
4063 account_id: Option<&AccountId>,
4064 side: Option<OrderSide>,
4065 ) -> bool {
4066 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
4067 FilterSources::Empty => false,
4068 FilterSources::Unfiltered => side.map_or_else(
4069 || !bucket.is_empty(),
4070 |side| bucket.iter().any(|id| self.order_side_matches(id, side)),
4071 ),
4072 FilterSources::Sets(mut sources) => {
4073 sources.push(bucket);
4074 sources.sort_unstable_by_key(|s| s.len());
4075 let driver = sources[0];
4076 let rest = &sources[1..];
4077
4078 driver
4079 .iter()
4080 .filter(|id| rest.iter().all(|s| s.contains(id)))
4081 .any(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
4082 }
4083 }
4084 }
4085
4086 fn any_positions_in_bucket(
4087 &self,
4088 bucket: &AHashSet<PositionId>,
4089 venue: Option<&Venue>,
4090 instrument_id: Option<&InstrumentId>,
4091 strategy_id: Option<&StrategyId>,
4092 account_id: Option<&AccountId>,
4093 side: Option<PositionSide>,
4094 ) -> bool {
4095 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
4096 FilterSources::Empty => false,
4097 FilterSources::Unfiltered => side.map_or_else(
4098 || !bucket.is_empty(),
4099 |side| bucket.iter().any(|id| self.position_side_matches(id, side)),
4100 ),
4101 FilterSources::Sets(mut sources) => {
4102 sources.push(bucket);
4103 sources.sort_unstable_by_key(|s| s.len());
4104 let driver = sources[0];
4105 let rest = &sources[1..];
4106
4107 driver
4108 .iter()
4109 .filter(|id| rest.iter().all(|s| s.contains(id)))
4110 .any(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
4111 }
4112 }
4113 }
4114
4115 fn order_side_matches(&self, client_order_id: &ClientOrderId, side: OrderSide) -> bool {
4116 self.orders
4117 .get(client_order_id)
4118 .is_some_and(|cell| cell.borrow().order_side() == side)
4119 }
4120
4121 fn position_side_matches(&self, position_id: &PositionId, side: PositionSide) -> bool {
4122 self.positions
4123 .get(position_id)
4124 .is_some_and(|cell| cell.borrow().side == side)
4125 }
4126
4127 fn get_orders_for_ids(
4133 &self,
4134 client_order_ids: impl IntoIterator<Item = ClientOrderId>,
4135 side: Option<OrderSide>,
4136 ) -> Vec<OrderRef<'_>> {
4137 const UNCACHED_SORT_MAX_LEN: usize = 32;
4138
4139 let mut orders = Vec::new();
4140
4141 for client_order_id in client_order_ids {
4142 let order_cell = self
4143 .orders
4144 .get(&client_order_id)
4145 .unwrap_or_else(|| panic!("Order {client_order_id} not found"));
4146 let order = OrderRef::new(order_cell.borrow());
4147
4148 if side.is_none_or(|side| side == order.order_side()) {
4149 orders.push(order);
4150 }
4151 }
4152
4153 let key = |order: &OrderRef<'_>| order.client_order_id();
4156
4157 if orders.len() <= UNCACHED_SORT_MAX_LEN {
4158 orders.sort_by_key(key);
4159 } else {
4160 orders.sort_by_cached_key(key);
4161 }
4162
4163 orders
4164 }
4165
4166 fn get_positions_for_ids(
4176 &self,
4177 position_ids: &AHashSet<PositionId>,
4178 side: Option<PositionSide>,
4179 ) -> Vec<PositionRef<'_>> {
4180 let mut positions = Vec::new();
4181
4182 for position_id in position_ids {
4183 let position_cell = self
4184 .positions
4185 .get(position_id)
4186 .unwrap_or_else(|| panic!("Position {position_id} not found"));
4187 let position = PositionRef::new(position_cell.borrow());
4188
4189 if side.is_none_or(|side| side == position.side) {
4190 positions.push(position);
4191 }
4192 }
4193
4194 positions.sort_by_key(|p| p.id);
4197 positions
4198 }
4199
4200 #[must_use]
4202 pub fn client_order_ids(
4203 &self,
4204 venue: Option<&Venue>,
4205 instrument_id: Option<&InstrumentId>,
4206 strategy_id: Option<&StrategyId>,
4207 account_id: Option<&AccountId>,
4208 ) -> AHashSet<ClientOrderId> {
4209 self.query_orders_in_bucket(
4210 &self.index.orders,
4211 venue,
4212 instrument_id,
4213 strategy_id,
4214 account_id,
4215 )
4216 }
4217
4218 #[must_use]
4220 pub fn client_order_ids_open(
4221 &self,
4222 venue: Option<&Venue>,
4223 instrument_id: Option<&InstrumentId>,
4224 strategy_id: Option<&StrategyId>,
4225 account_id: Option<&AccountId>,
4226 ) -> AHashSet<ClientOrderId> {
4227 self.query_orders_in_bucket(
4228 &self.index.orders_open,
4229 venue,
4230 instrument_id,
4231 strategy_id,
4232 account_id,
4233 )
4234 }
4235
4236 #[must_use]
4238 pub fn client_order_ids_closed(
4239 &self,
4240 venue: Option<&Venue>,
4241 instrument_id: Option<&InstrumentId>,
4242 strategy_id: Option<&StrategyId>,
4243 account_id: Option<&AccountId>,
4244 ) -> AHashSet<ClientOrderId> {
4245 self.query_orders_in_bucket(
4246 &self.index.orders_closed,
4247 venue,
4248 instrument_id,
4249 strategy_id,
4250 account_id,
4251 )
4252 }
4253
4254 #[must_use]
4259 pub fn client_order_ids_active_local(
4260 &self,
4261 venue: Option<&Venue>,
4262 instrument_id: Option<&InstrumentId>,
4263 strategy_id: Option<&StrategyId>,
4264 account_id: Option<&AccountId>,
4265 ) -> AHashSet<ClientOrderId> {
4266 self.query_orders_in_bucket(
4267 &self.index.orders_active_local,
4268 venue,
4269 instrument_id,
4270 strategy_id,
4271 account_id,
4272 )
4273 }
4274
4275 #[must_use]
4277 pub fn client_order_ids_emulated(
4278 &self,
4279 venue: Option<&Venue>,
4280 instrument_id: Option<&InstrumentId>,
4281 strategy_id: Option<&StrategyId>,
4282 account_id: Option<&AccountId>,
4283 ) -> AHashSet<ClientOrderId> {
4284 self.query_orders_in_bucket(
4285 &self.index.orders_emulated,
4286 venue,
4287 instrument_id,
4288 strategy_id,
4289 account_id,
4290 )
4291 }
4292
4293 #[must_use]
4295 pub fn client_order_ids_inflight(
4296 &self,
4297 venue: Option<&Venue>,
4298 instrument_id: Option<&InstrumentId>,
4299 strategy_id: Option<&StrategyId>,
4300 account_id: Option<&AccountId>,
4301 ) -> AHashSet<ClientOrderId> {
4302 self.query_orders_in_bucket(
4303 &self.index.orders_inflight,
4304 venue,
4305 instrument_id,
4306 strategy_id,
4307 account_id,
4308 )
4309 }
4310
4311 #[must_use]
4313 pub fn position_ids(
4314 &self,
4315 venue: Option<&Venue>,
4316 instrument_id: Option<&InstrumentId>,
4317 strategy_id: Option<&StrategyId>,
4318 account_id: Option<&AccountId>,
4319 ) -> AHashSet<PositionId> {
4320 self.query_positions_in_bucket(
4321 &self.index.positions,
4322 venue,
4323 instrument_id,
4324 strategy_id,
4325 account_id,
4326 )
4327 }
4328
4329 #[must_use]
4331 pub fn position_open_ids(
4332 &self,
4333 venue: Option<&Venue>,
4334 instrument_id: Option<&InstrumentId>,
4335 strategy_id: Option<&StrategyId>,
4336 account_id: Option<&AccountId>,
4337 ) -> AHashSet<PositionId> {
4338 self.query_positions_in_bucket(
4339 &self.index.positions_open,
4340 venue,
4341 instrument_id,
4342 strategy_id,
4343 account_id,
4344 )
4345 }
4346
4347 #[must_use]
4349 pub fn position_closed_ids(
4350 &self,
4351 venue: Option<&Venue>,
4352 instrument_id: Option<&InstrumentId>,
4353 strategy_id: Option<&StrategyId>,
4354 account_id: Option<&AccountId>,
4355 ) -> AHashSet<PositionId> {
4356 self.query_positions_in_bucket(
4357 &self.index.positions_closed,
4358 venue,
4359 instrument_id,
4360 strategy_id,
4361 account_id,
4362 )
4363 }
4364
4365 #[must_use]
4372 pub fn client_order_ids_view(
4373 &self,
4374 venue: Option<&Venue>,
4375 instrument_id: Option<&InstrumentId>,
4376 strategy_id: Option<&StrategyId>,
4377 account_id: Option<&AccountId>,
4378 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4379 self.view_orders_in_bucket(
4380 &self.index.orders,
4381 venue,
4382 instrument_id,
4383 strategy_id,
4384 account_id,
4385 )
4386 }
4387
4388 #[must_use]
4390 pub fn client_order_ids_open_view(
4391 &self,
4392 venue: Option<&Venue>,
4393 instrument_id: Option<&InstrumentId>,
4394 strategy_id: Option<&StrategyId>,
4395 account_id: Option<&AccountId>,
4396 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4397 self.view_orders_in_bucket(
4398 &self.index.orders_open,
4399 venue,
4400 instrument_id,
4401 strategy_id,
4402 account_id,
4403 )
4404 }
4405
4406 #[must_use]
4408 pub fn client_order_ids_closed_view(
4409 &self,
4410 venue: Option<&Venue>,
4411 instrument_id: Option<&InstrumentId>,
4412 strategy_id: Option<&StrategyId>,
4413 account_id: Option<&AccountId>,
4414 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4415 self.view_orders_in_bucket(
4416 &self.index.orders_closed,
4417 venue,
4418 instrument_id,
4419 strategy_id,
4420 account_id,
4421 )
4422 }
4423
4424 #[must_use]
4426 pub fn client_order_ids_active_local_view(
4427 &self,
4428 venue: Option<&Venue>,
4429 instrument_id: Option<&InstrumentId>,
4430 strategy_id: Option<&StrategyId>,
4431 account_id: Option<&AccountId>,
4432 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4433 self.view_orders_in_bucket(
4434 &self.index.orders_active_local,
4435 venue,
4436 instrument_id,
4437 strategy_id,
4438 account_id,
4439 )
4440 }
4441
4442 #[must_use]
4444 pub fn client_order_ids_emulated_view(
4445 &self,
4446 venue: Option<&Venue>,
4447 instrument_id: Option<&InstrumentId>,
4448 strategy_id: Option<&StrategyId>,
4449 account_id: Option<&AccountId>,
4450 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4451 self.view_orders_in_bucket(
4452 &self.index.orders_emulated,
4453 venue,
4454 instrument_id,
4455 strategy_id,
4456 account_id,
4457 )
4458 }
4459
4460 #[must_use]
4462 pub fn client_order_ids_inflight_view(
4463 &self,
4464 venue: Option<&Venue>,
4465 instrument_id: Option<&InstrumentId>,
4466 strategy_id: Option<&StrategyId>,
4467 account_id: Option<&AccountId>,
4468 ) -> Cow<'_, AHashSet<ClientOrderId>> {
4469 self.view_orders_in_bucket(
4470 &self.index.orders_inflight,
4471 venue,
4472 instrument_id,
4473 strategy_id,
4474 account_id,
4475 )
4476 }
4477
4478 #[must_use]
4480 pub fn position_ids_view(
4481 &self,
4482 venue: Option<&Venue>,
4483 instrument_id: Option<&InstrumentId>,
4484 strategy_id: Option<&StrategyId>,
4485 account_id: Option<&AccountId>,
4486 ) -> Cow<'_, AHashSet<PositionId>> {
4487 self.view_positions_in_bucket(
4488 &self.index.positions,
4489 venue,
4490 instrument_id,
4491 strategy_id,
4492 account_id,
4493 )
4494 }
4495
4496 #[must_use]
4498 pub fn position_open_ids_view(
4499 &self,
4500 venue: Option<&Venue>,
4501 instrument_id: Option<&InstrumentId>,
4502 strategy_id: Option<&StrategyId>,
4503 account_id: Option<&AccountId>,
4504 ) -> Cow<'_, AHashSet<PositionId>> {
4505 self.view_positions_in_bucket(
4506 &self.index.positions_open,
4507 venue,
4508 instrument_id,
4509 strategy_id,
4510 account_id,
4511 )
4512 }
4513
4514 #[must_use]
4516 pub fn position_closed_ids_view(
4517 &self,
4518 venue: Option<&Venue>,
4519 instrument_id: Option<&InstrumentId>,
4520 strategy_id: Option<&StrategyId>,
4521 account_id: Option<&AccountId>,
4522 ) -> Cow<'_, AHashSet<PositionId>> {
4523 self.view_positions_in_bucket(
4524 &self.index.positions_closed,
4525 venue,
4526 instrument_id,
4527 strategy_id,
4528 account_id,
4529 )
4530 }
4531
4532 pub fn iter_client_order_ids(
4538 &self,
4539 venue: Option<&Venue>,
4540 instrument_id: Option<&InstrumentId>,
4541 strategy_id: Option<&StrategyId>,
4542 account_id: Option<&AccountId>,
4543 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4544 self.iter_orders_in_bucket(
4545 &self.index.orders,
4546 venue,
4547 instrument_id,
4548 strategy_id,
4549 account_id,
4550 )
4551 }
4552
4553 pub fn iter_client_order_ids_open(
4555 &self,
4556 venue: Option<&Venue>,
4557 instrument_id: Option<&InstrumentId>,
4558 strategy_id: Option<&StrategyId>,
4559 account_id: Option<&AccountId>,
4560 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4561 self.iter_orders_in_bucket(
4562 &self.index.orders_open,
4563 venue,
4564 instrument_id,
4565 strategy_id,
4566 account_id,
4567 )
4568 }
4569
4570 pub fn iter_client_order_ids_closed(
4572 &self,
4573 venue: Option<&Venue>,
4574 instrument_id: Option<&InstrumentId>,
4575 strategy_id: Option<&StrategyId>,
4576 account_id: Option<&AccountId>,
4577 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4578 self.iter_orders_in_bucket(
4579 &self.index.orders_closed,
4580 venue,
4581 instrument_id,
4582 strategy_id,
4583 account_id,
4584 )
4585 }
4586
4587 pub fn iter_client_order_ids_active_local(
4589 &self,
4590 venue: Option<&Venue>,
4591 instrument_id: Option<&InstrumentId>,
4592 strategy_id: Option<&StrategyId>,
4593 account_id: Option<&AccountId>,
4594 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4595 self.iter_orders_in_bucket(
4596 &self.index.orders_active_local,
4597 venue,
4598 instrument_id,
4599 strategy_id,
4600 account_id,
4601 )
4602 }
4603
4604 pub fn iter_client_order_ids_emulated(
4606 &self,
4607 venue: Option<&Venue>,
4608 instrument_id: Option<&InstrumentId>,
4609 strategy_id: Option<&StrategyId>,
4610 account_id: Option<&AccountId>,
4611 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4612 self.iter_orders_in_bucket(
4613 &self.index.orders_emulated,
4614 venue,
4615 instrument_id,
4616 strategy_id,
4617 account_id,
4618 )
4619 }
4620
4621 pub fn iter_client_order_ids_inflight(
4623 &self,
4624 venue: Option<&Venue>,
4625 instrument_id: Option<&InstrumentId>,
4626 strategy_id: Option<&StrategyId>,
4627 account_id: Option<&AccountId>,
4628 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
4629 self.iter_orders_in_bucket(
4630 &self.index.orders_inflight,
4631 venue,
4632 instrument_id,
4633 strategy_id,
4634 account_id,
4635 )
4636 }
4637
4638 pub fn iter_position_ids(
4640 &self,
4641 venue: Option<&Venue>,
4642 instrument_id: Option<&InstrumentId>,
4643 strategy_id: Option<&StrategyId>,
4644 account_id: Option<&AccountId>,
4645 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
4646 self.iter_positions_in_bucket(
4647 &self.index.positions,
4648 venue,
4649 instrument_id,
4650 strategy_id,
4651 account_id,
4652 )
4653 }
4654
4655 pub fn iter_position_open_ids(
4657 &self,
4658 venue: Option<&Venue>,
4659 instrument_id: Option<&InstrumentId>,
4660 strategy_id: Option<&StrategyId>,
4661 account_id: Option<&AccountId>,
4662 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
4663 self.iter_positions_in_bucket(
4664 &self.index.positions_open,
4665 venue,
4666 instrument_id,
4667 strategy_id,
4668 account_id,
4669 )
4670 }
4671
4672 pub fn iter_position_closed_ids(
4674 &self,
4675 venue: Option<&Venue>,
4676 instrument_id: Option<&InstrumentId>,
4677 strategy_id: Option<&StrategyId>,
4678 account_id: Option<&AccountId>,
4679 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
4680 self.iter_positions_in_bucket(
4681 &self.index.positions_closed,
4682 venue,
4683 instrument_id,
4684 strategy_id,
4685 account_id,
4686 )
4687 }
4688
4689 #[must_use]
4691 pub fn strategy_ids(&self) -> AHashSet<StrategyId> {
4692 self.index.strategies.clone()
4693 }
4694
4695 #[must_use]
4697 pub fn exec_algorithm_ids(&self) -> AHashSet<ExecAlgorithmId> {
4698 self.index.exec_algorithms.clone()
4699 }
4700
4701 #[must_use]
4710 pub fn order_ref(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
4711 self.orders
4712 .get(client_order_id)
4713 .map(|order_cell| OrderRef::new(order_cell.borrow()))
4714 }
4715
4716 #[must_use]
4720 pub fn order(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
4721 self.order_ref(client_order_id)
4722 }
4723
4724 pub fn try_order_ref(
4730 &self,
4731 client_order_id: &ClientOrderId,
4732 ) -> Result<OrderRef<'_>, OrderLookupError> {
4733 self.orders
4734 .get(client_order_id)
4735 .map(|order_cell| OrderRef::new(order_cell.borrow()))
4736 .ok_or_else(|| OrderLookupError::not_found(*client_order_id))
4737 }
4738
4739 pub fn try_order(
4747 &self,
4748 client_order_id: &ClientOrderId,
4749 ) -> Result<OrderRef<'_>, OrderLookupError> {
4750 self.try_order_ref(client_order_id)
4751 }
4752
4753 #[must_use]
4763 pub fn order_mut(&mut self, client_order_id: &ClientOrderId) -> Option<OrderRefMut<'_>> {
4764 self.orders
4765 .get(client_order_id)
4766 .map(|order_cell| OrderRefMut::new(order_cell.borrow_mut()))
4767 }
4768
4769 #[must_use]
4774 pub fn order_owned(&self, client_order_id: &ClientOrderId) -> Option<OrderAny> {
4775 self.orders
4776 .get(client_order_id)
4777 .map(|order_cell| order_cell.borrow().clone())
4778 }
4779
4780 pub fn try_order_owned(
4786 &self,
4787 client_order_id: &ClientOrderId,
4788 ) -> Result<OrderAny, OrderLookupError> {
4789 self.try_order_ref(client_order_id)
4790 .map(|order| order.cloned())
4791 }
4792
4793 #[must_use]
4795 pub fn orders_for_ids(
4796 &self,
4797 client_order_ids: &[ClientOrderId],
4798 context: &dyn Display,
4799 ) -> Vec<OrderAny> {
4800 let mut orders = Vec::with_capacity(client_order_ids.len());
4801 for id in client_order_ids {
4802 match self.orders.get(id) {
4803 Some(order_cell) => orders.push(order_cell.borrow().clone()),
4804 None => log::error!("Order {id} not found in cache for {context}"),
4805 }
4806 }
4807 orders
4808 }
4809
4810 #[must_use]
4812 pub fn client_order_id(&self, venue_order_id: &VenueOrderId) -> Option<&ClientOrderId> {
4813 self.index.venue_order_ids.get(venue_order_id)
4814 }
4815
4816 #[must_use]
4818 pub fn venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<&VenueOrderId> {
4819 self.index.client_order_ids.get(client_order_id)
4820 }
4821
4822 #[must_use]
4824 pub fn client_id(&self, client_order_id: &ClientOrderId) -> Option<&ClientId> {
4825 self.index.order_client.get(client_order_id)
4826 }
4827
4828 #[must_use]
4834 pub fn orders_refs(
4835 &self,
4836 venue: Option<&Venue>,
4837 instrument_id: Option<&InstrumentId>,
4838 strategy_id: Option<&StrategyId>,
4839 account_id: Option<&AccountId>,
4840 side: Option<OrderSide>,
4841 ) -> Vec<OrderRef<'_>> {
4842 if venue.is_none()
4843 && instrument_id.is_none()
4844 && strategy_id.is_none()
4845 && account_id.is_none()
4846 {
4847 return self.get_orders_for_ids(self.index.orders.iter().copied(), side);
4848 }
4849
4850 let client_order_ids =
4851 self.iter_client_order_ids(venue, instrument_id, strategy_id, account_id);
4852 self.get_orders_for_ids(client_order_ids, side)
4853 }
4854
4855 #[must_use]
4859 pub fn orders(
4860 &self,
4861 venue: Option<&Venue>,
4862 instrument_id: Option<&InstrumentId>,
4863 strategy_id: Option<&StrategyId>,
4864 account_id: Option<&AccountId>,
4865 side: Option<OrderSide>,
4866 ) -> Vec<OrderRef<'_>> {
4867 self.orders_refs(venue, instrument_id, strategy_id, account_id, side)
4868 }
4869
4870 #[must_use]
4872 pub fn orders_open_refs(
4873 &self,
4874 venue: Option<&Venue>,
4875 instrument_id: Option<&InstrumentId>,
4876 strategy_id: Option<&StrategyId>,
4877 account_id: Option<&AccountId>,
4878 side: Option<OrderSide>,
4879 ) -> Vec<OrderRef<'_>> {
4880 let client_order_ids =
4881 self.client_order_ids_open(venue, instrument_id, strategy_id, account_id);
4882 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4883 }
4884
4885 #[must_use]
4889 pub fn orders_open(
4890 &self,
4891 venue: Option<&Venue>,
4892 instrument_id: Option<&InstrumentId>,
4893 strategy_id: Option<&StrategyId>,
4894 account_id: Option<&AccountId>,
4895 side: Option<OrderSide>,
4896 ) -> Vec<OrderRef<'_>> {
4897 self.orders_open_refs(venue, instrument_id, strategy_id, account_id, side)
4898 }
4899
4900 #[must_use]
4902 pub fn orders_closed_refs(
4903 &self,
4904 venue: Option<&Venue>,
4905 instrument_id: Option<&InstrumentId>,
4906 strategy_id: Option<&StrategyId>,
4907 account_id: Option<&AccountId>,
4908 side: Option<OrderSide>,
4909 ) -> Vec<OrderRef<'_>> {
4910 let client_order_ids =
4911 self.client_order_ids_closed(venue, instrument_id, strategy_id, account_id);
4912 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4913 }
4914
4915 #[must_use]
4919 pub fn orders_closed(
4920 &self,
4921 venue: Option<&Venue>,
4922 instrument_id: Option<&InstrumentId>,
4923 strategy_id: Option<&StrategyId>,
4924 account_id: Option<&AccountId>,
4925 side: Option<OrderSide>,
4926 ) -> Vec<OrderRef<'_>> {
4927 self.orders_closed_refs(venue, instrument_id, strategy_id, account_id, side)
4928 }
4929
4930 #[must_use]
4935 pub fn orders_active_local_refs(
4936 &self,
4937 venue: Option<&Venue>,
4938 instrument_id: Option<&InstrumentId>,
4939 strategy_id: Option<&StrategyId>,
4940 account_id: Option<&AccountId>,
4941 side: Option<OrderSide>,
4942 ) -> Vec<OrderRef<'_>> {
4943 let client_order_ids =
4944 self.client_order_ids_active_local(venue, instrument_id, strategy_id, account_id);
4945 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4946 }
4947
4948 #[must_use]
4952 pub fn orders_active_local(
4953 &self,
4954 venue: Option<&Venue>,
4955 instrument_id: Option<&InstrumentId>,
4956 strategy_id: Option<&StrategyId>,
4957 account_id: Option<&AccountId>,
4958 side: Option<OrderSide>,
4959 ) -> Vec<OrderRef<'_>> {
4960 self.orders_active_local_refs(venue, instrument_id, strategy_id, account_id, side)
4961 }
4962
4963 #[must_use]
4965 pub fn orders_emulated_refs(
4966 &self,
4967 venue: Option<&Venue>,
4968 instrument_id: Option<&InstrumentId>,
4969 strategy_id: Option<&StrategyId>,
4970 account_id: Option<&AccountId>,
4971 side: Option<OrderSide>,
4972 ) -> Vec<OrderRef<'_>> {
4973 let client_order_ids =
4974 self.client_order_ids_emulated(venue, instrument_id, strategy_id, account_id);
4975 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
4976 }
4977
4978 #[must_use]
4982 pub fn orders_emulated(
4983 &self,
4984 venue: Option<&Venue>,
4985 instrument_id: Option<&InstrumentId>,
4986 strategy_id: Option<&StrategyId>,
4987 account_id: Option<&AccountId>,
4988 side: Option<OrderSide>,
4989 ) -> Vec<OrderRef<'_>> {
4990 self.orders_emulated_refs(venue, instrument_id, strategy_id, account_id, side)
4991 }
4992
4993 #[must_use]
4995 pub fn orders_inflight_refs(
4996 &self,
4997 venue: Option<&Venue>,
4998 instrument_id: Option<&InstrumentId>,
4999 strategy_id: Option<&StrategyId>,
5000 account_id: Option<&AccountId>,
5001 side: Option<OrderSide>,
5002 ) -> Vec<OrderRef<'_>> {
5003 let client_order_ids =
5004 self.client_order_ids_inflight(venue, instrument_id, strategy_id, account_id);
5005 self.get_orders_for_ids(client_order_ids.iter().copied(), side)
5006 }
5007
5008 #[must_use]
5012 pub fn orders_inflight(
5013 &self,
5014 venue: Option<&Venue>,
5015 instrument_id: Option<&InstrumentId>,
5016 strategy_id: Option<&StrategyId>,
5017 account_id: Option<&AccountId>,
5018 side: Option<OrderSide>,
5019 ) -> Vec<OrderRef<'_>> {
5020 self.orders_inflight_refs(venue, instrument_id, strategy_id, account_id, side)
5021 }
5022
5023 #[must_use]
5025 pub fn orders_for_position(&self, position_id: &PositionId) -> Vec<OrderRef<'_>> {
5026 match self.index.position_orders.get(position_id) {
5027 Some(client_order_ids) => {
5028 self.get_orders_for_ids(client_order_ids.iter().copied(), None)
5029 }
5030 None => Vec::new(),
5031 }
5032 }
5033
5034 #[must_use]
5036 pub fn order_exists(&self, client_order_id: &ClientOrderId) -> bool {
5037 self.index.orders.contains(client_order_id)
5038 }
5039
5040 #[must_use]
5042 pub fn is_order_open(&self, client_order_id: &ClientOrderId) -> bool {
5043 self.index.orders_open.contains(client_order_id)
5044 }
5045
5046 #[must_use]
5048 pub fn is_order_closed(&self, client_order_id: &ClientOrderId) -> bool {
5049 self.index.orders_closed.contains(client_order_id)
5050 }
5051
5052 #[must_use]
5057 pub fn is_order_active_local(&self, client_order_id: &ClientOrderId) -> bool {
5058 self.index.orders_active_local.contains(client_order_id)
5059 }
5060
5061 #[must_use]
5063 pub fn is_order_emulated(&self, client_order_id: &ClientOrderId) -> bool {
5064 self.index.orders_emulated.contains(client_order_id)
5065 }
5066
5067 #[must_use]
5069 pub fn is_order_inflight(&self, client_order_id: &ClientOrderId) -> bool {
5070 self.index.orders_inflight.contains(client_order_id)
5071 }
5072
5073 #[must_use]
5075 pub fn is_order_pending_cancel_local(&self, client_order_id: &ClientOrderId) -> bool {
5076 self.index.orders_pending_cancel.contains(client_order_id)
5077 }
5078
5079 #[must_use]
5081 pub fn orders_open_count(
5082 &self,
5083 venue: Option<&Venue>,
5084 instrument_id: Option<&InstrumentId>,
5085 strategy_id: Option<&StrategyId>,
5086 account_id: Option<&AccountId>,
5087 side: Option<OrderSide>,
5088 ) -> usize {
5089 self.count_orders_in_bucket(
5090 &self.index.orders_open,
5091 venue,
5092 instrument_id,
5093 strategy_id,
5094 account_id,
5095 side,
5096 )
5097 }
5098
5099 #[must_use]
5101 pub fn orders_closed_count(
5102 &self,
5103 venue: Option<&Venue>,
5104 instrument_id: Option<&InstrumentId>,
5105 strategy_id: Option<&StrategyId>,
5106 account_id: Option<&AccountId>,
5107 side: Option<OrderSide>,
5108 ) -> usize {
5109 self.count_orders_in_bucket(
5110 &self.index.orders_closed,
5111 venue,
5112 instrument_id,
5113 strategy_id,
5114 account_id,
5115 side,
5116 )
5117 }
5118
5119 #[must_use]
5124 pub fn orders_active_local_count(
5125 &self,
5126 venue: Option<&Venue>,
5127 instrument_id: Option<&InstrumentId>,
5128 strategy_id: Option<&StrategyId>,
5129 account_id: Option<&AccountId>,
5130 side: Option<OrderSide>,
5131 ) -> usize {
5132 self.count_orders_in_bucket(
5133 &self.index.orders_active_local,
5134 venue,
5135 instrument_id,
5136 strategy_id,
5137 account_id,
5138 side,
5139 )
5140 }
5141
5142 #[must_use]
5144 pub fn orders_emulated_count(
5145 &self,
5146 venue: Option<&Venue>,
5147 instrument_id: Option<&InstrumentId>,
5148 strategy_id: Option<&StrategyId>,
5149 account_id: Option<&AccountId>,
5150 side: Option<OrderSide>,
5151 ) -> usize {
5152 self.count_orders_in_bucket(
5153 &self.index.orders_emulated,
5154 venue,
5155 instrument_id,
5156 strategy_id,
5157 account_id,
5158 side,
5159 )
5160 }
5161
5162 #[must_use]
5164 pub fn orders_inflight_count(
5165 &self,
5166 venue: Option<&Venue>,
5167 instrument_id: Option<&InstrumentId>,
5168 strategy_id: Option<&StrategyId>,
5169 account_id: Option<&AccountId>,
5170 side: Option<OrderSide>,
5171 ) -> usize {
5172 self.count_orders_in_bucket(
5173 &self.index.orders_inflight,
5174 venue,
5175 instrument_id,
5176 strategy_id,
5177 account_id,
5178 side,
5179 )
5180 }
5181
5182 #[must_use]
5184 pub fn orders_total_count(
5185 &self,
5186 venue: Option<&Venue>,
5187 instrument_id: Option<&InstrumentId>,
5188 strategy_id: Option<&StrategyId>,
5189 account_id: Option<&AccountId>,
5190 side: Option<OrderSide>,
5191 ) -> usize {
5192 self.count_orders_in_bucket(
5193 &self.index.orders,
5194 venue,
5195 instrument_id,
5196 strategy_id,
5197 account_id,
5198 side,
5199 )
5200 }
5201
5202 #[must_use]
5208 pub fn has_orders_open(
5209 &self,
5210 venue: Option<&Venue>,
5211 instrument_id: Option<&InstrumentId>,
5212 strategy_id: Option<&StrategyId>,
5213 account_id: Option<&AccountId>,
5214 side: Option<OrderSide>,
5215 ) -> bool {
5216 self.any_orders_in_bucket(
5217 &self.index.orders_open,
5218 venue,
5219 instrument_id,
5220 strategy_id,
5221 account_id,
5222 side,
5223 )
5224 }
5225
5226 #[must_use]
5228 pub fn has_orders_closed(
5229 &self,
5230 venue: Option<&Venue>,
5231 instrument_id: Option<&InstrumentId>,
5232 strategy_id: Option<&StrategyId>,
5233 account_id: Option<&AccountId>,
5234 side: Option<OrderSide>,
5235 ) -> bool {
5236 self.any_orders_in_bucket(
5237 &self.index.orders_closed,
5238 venue,
5239 instrument_id,
5240 strategy_id,
5241 account_id,
5242 side,
5243 )
5244 }
5245
5246 #[must_use]
5250 pub fn has_orders_active_local(
5251 &self,
5252 venue: Option<&Venue>,
5253 instrument_id: Option<&InstrumentId>,
5254 strategy_id: Option<&StrategyId>,
5255 account_id: Option<&AccountId>,
5256 side: Option<OrderSide>,
5257 ) -> bool {
5258 self.any_orders_in_bucket(
5259 &self.index.orders_active_local,
5260 venue,
5261 instrument_id,
5262 strategy_id,
5263 account_id,
5264 side,
5265 )
5266 }
5267
5268 #[must_use]
5270 pub fn has_orders_emulated(
5271 &self,
5272 venue: Option<&Venue>,
5273 instrument_id: Option<&InstrumentId>,
5274 strategy_id: Option<&StrategyId>,
5275 account_id: Option<&AccountId>,
5276 side: Option<OrderSide>,
5277 ) -> bool {
5278 self.any_orders_in_bucket(
5279 &self.index.orders_emulated,
5280 venue,
5281 instrument_id,
5282 strategy_id,
5283 account_id,
5284 side,
5285 )
5286 }
5287
5288 #[must_use]
5290 pub fn has_orders_inflight(
5291 &self,
5292 venue: Option<&Venue>,
5293 instrument_id: Option<&InstrumentId>,
5294 strategy_id: Option<&StrategyId>,
5295 account_id: Option<&AccountId>,
5296 side: Option<OrderSide>,
5297 ) -> bool {
5298 self.any_orders_in_bucket(
5299 &self.index.orders_inflight,
5300 venue,
5301 instrument_id,
5302 strategy_id,
5303 account_id,
5304 side,
5305 )
5306 }
5307
5308 #[must_use]
5310 pub fn has_orders(
5311 &self,
5312 venue: Option<&Venue>,
5313 instrument_id: Option<&InstrumentId>,
5314 strategy_id: Option<&StrategyId>,
5315 account_id: Option<&AccountId>,
5316 side: Option<OrderSide>,
5317 ) -> bool {
5318 self.any_orders_in_bucket(
5319 &self.index.orders,
5320 venue,
5321 instrument_id,
5322 strategy_id,
5323 account_id,
5324 side,
5325 )
5326 }
5327
5328 #[must_use]
5330 pub fn order_list(&self, order_list_id: &OrderListId) -> Option<&OrderList> {
5331 self.order_lists.get(order_list_id)
5332 }
5333
5334 pub fn try_order_list(
5340 &self,
5341 order_list_id: &OrderListId,
5342 ) -> Result<&OrderList, OrderListLookupError> {
5343 self.order_lists
5344 .get(order_list_id)
5345 .ok_or_else(|| OrderListLookupError::not_found(*order_list_id))
5346 }
5347
5348 #[must_use]
5350 pub fn order_lists(
5351 &self,
5352 venue: Option<&Venue>,
5353 instrument_id: Option<&InstrumentId>,
5354 strategy_id: Option<&StrategyId>,
5355 account_id: Option<&AccountId>,
5356 ) -> Vec<&OrderList> {
5357 let mut order_lists = self.order_lists.values().collect::<Vec<&OrderList>>();
5358
5359 if let Some(venue) = venue {
5360 order_lists.retain(|ol| &ol.instrument_id.venue == venue);
5361 }
5362
5363 if let Some(instrument_id) = instrument_id {
5364 order_lists.retain(|ol| &ol.instrument_id == instrument_id);
5365 }
5366
5367 if let Some(strategy_id) = strategy_id {
5368 order_lists.retain(|ol| &ol.strategy_id == strategy_id);
5369 }
5370
5371 if let Some(account_id) = account_id {
5372 order_lists.retain(|ol| {
5373 ol.client_order_ids.iter().any(|client_order_id| {
5374 self.orders.get(client_order_id).is_some_and(|order_cell| {
5375 order_cell.borrow().account_id().as_ref() == Some(account_id)
5376 })
5377 })
5378 });
5379 }
5380
5381 order_lists
5382 }
5383
5384 #[must_use]
5386 pub fn order_list_exists(&self, order_list_id: &OrderListId) -> bool {
5387 self.order_lists.contains_key(order_list_id)
5388 }
5389
5390 #[must_use]
5395 pub fn orders_for_exec_algorithm(
5396 &self,
5397 exec_algorithm_id: &ExecAlgorithmId,
5398 venue: Option<&Venue>,
5399 instrument_id: Option<&InstrumentId>,
5400 strategy_id: Option<&StrategyId>,
5401 account_id: Option<&AccountId>,
5402 side: Option<OrderSide>,
5403 ) -> Vec<OrderRef<'_>> {
5404 let Some(exec_algorithm_order_ids) =
5405 self.index.exec_algorithm_orders.get(exec_algorithm_id)
5406 else {
5407 return Vec::new();
5408 };
5409
5410 let filtered = self.query_orders_in_bucket(
5411 exec_algorithm_order_ids,
5412 venue,
5413 instrument_id,
5414 strategy_id,
5415 account_id,
5416 );
5417 self.get_orders_for_ids(filtered.iter().copied(), side)
5418 }
5419
5420 #[must_use]
5422 pub fn orders_for_exec_spawn(&self, exec_spawn_id: &ClientOrderId) -> Vec<OrderRef<'_>> {
5423 match self.index.exec_spawn_orders.get(exec_spawn_id) {
5424 Some(ids) => self.get_orders_for_ids(ids.iter().copied(), None),
5425 None => Vec::new(),
5426 }
5427 }
5428
5429 #[must_use]
5431 pub fn exec_spawn_total_quantity(
5432 &self,
5433 exec_spawn_id: &ClientOrderId,
5434 active_only: bool,
5435 ) -> Option<Quantity> {
5436 self.exec_spawn_total(exec_spawn_id, active_only, Order::quantity)
5437 }
5438
5439 #[must_use]
5441 pub fn exec_spawn_total_filled_qty(
5442 &self,
5443 exec_spawn_id: &ClientOrderId,
5444 active_only: bool,
5445 ) -> Option<Quantity> {
5446 self.exec_spawn_total(exec_spawn_id, active_only, Order::filled_qty)
5447 }
5448
5449 #[must_use]
5451 pub fn exec_spawn_total_leaves_qty(
5452 &self,
5453 exec_spawn_id: &ClientOrderId,
5454 active_only: bool,
5455 ) -> Option<Quantity> {
5456 self.exec_spawn_total(exec_spawn_id, active_only, Order::leaves_qty)
5457 }
5458
5459 fn exec_spawn_total(
5460 &self,
5461 exec_spawn_id: &ClientOrderId,
5462 active_only: bool,
5463 quantity: impl Fn(&OrderAny) -> Quantity,
5464 ) -> Option<Quantity> {
5465 self.orders_for_exec_spawn(exec_spawn_id)
5466 .into_iter()
5467 .filter(|order| !active_only || !order.is_closed())
5468 .map(|order| quantity(&order))
5469 .reduce(|total, quantity| total + quantity)
5470 }
5471
5472 #[must_use]
5476 pub fn position_ref(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
5477 self.positions
5478 .get(position_id)
5479 .map(|position_cell| PositionRef::new(position_cell.borrow()))
5480 }
5481
5482 #[must_use]
5486 pub fn position(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
5487 self.position_ref(position_id)
5488 }
5489
5490 pub fn try_position_ref(
5496 &self,
5497 position_id: &PositionId,
5498 ) -> Result<PositionRef<'_>, PositionLookupError> {
5499 self.positions
5500 .get(position_id)
5501 .map(|position_cell| PositionRef::new(position_cell.borrow()))
5502 .ok_or_else(|| PositionLookupError::not_found(*position_id))
5503 }
5504
5505 pub fn try_position(
5513 &self,
5514 position_id: &PositionId,
5515 ) -> Result<PositionRef<'_>, PositionLookupError> {
5516 self.try_position_ref(position_id)
5517 }
5518
5519 #[must_use]
5529 pub fn position_mut(&mut self, position_id: &PositionId) -> Option<PositionRefMut<'_>> {
5530 self.positions
5531 .get(position_id)
5532 .map(|position_cell| PositionRefMut::new(position_cell.borrow_mut()))
5533 }
5534
5535 #[must_use]
5540 pub fn position_owned(&self, position_id: &PositionId) -> Option<Position> {
5541 self.positions
5542 .get(position_id)
5543 .map(|position_cell| position_cell.borrow().clone())
5544 }
5545
5546 #[must_use]
5548 pub fn position_for_order_ref(
5549 &self,
5550 client_order_id: &ClientOrderId,
5551 ) -> Option<PositionRef<'_>> {
5552 self.index
5553 .order_position
5554 .get(client_order_id)
5555 .and_then(|position_id| self.positions.get(position_id))
5556 .map(|position_cell| PositionRef::new(position_cell.borrow()))
5557 }
5558
5559 #[must_use]
5563 pub fn position_for_order(&self, client_order_id: &ClientOrderId) -> Option<PositionRef<'_>> {
5564 self.position_for_order_ref(client_order_id)
5565 }
5566
5567 #[must_use]
5569 pub fn position_id(&self, client_order_id: &ClientOrderId) -> Option<&PositionId> {
5570 self.index.order_position.get(client_order_id)
5571 }
5572
5573 pub fn hedging_flip_position_ids(
5583 &self,
5584 client_order_id: ClientOrderId,
5585 ) -> anyhow::Result<Option<[PositionId; 2]>> {
5586 let Some(order) = self.order_ref(&client_order_id) else {
5587 return Ok(None);
5588 };
5589
5590 let mut flip = None;
5591 let mut corrected_source = None;
5592
5593 for event in order.events() {
5594 let OrderEventAny::Filled(fill) = event else {
5595 continue;
5596 };
5597
5598 let Some(source_id) = fill.position_id.filter(PositionId::is_virtual) else {
5599 continue;
5600 };
5601
5602 let source = self.position_ref(&source_id);
5603 if source.is_some() && self.oms_type(&source_id) != Some(OmsType::Hedging) {
5604 continue;
5605 }
5606
5607 if source.as_ref().is_some_and(|source| {
5608 source.replay_events.iter().any(|event| {
5609 matches!(event, PositionReplayEvent::Filled(fragment) if fragment == fill)
5610 })
5611 }) {
5612 continue;
5613 }
5614
5615 let closing = source.as_ref().and_then(|source| {
5616 source
5617 .fill_fragments(client_order_id, fill.trade_id)
5618 .into_iter()
5619 .find(|fragment| {
5620 fragment.event_id == fill.event_id && fragment.last_qty < fill.last_qty
5621 })
5622 .map(|closing| (source, closing))
5623 });
5624
5625 let Some((source, closing)) = closing else {
5626 let has_opening = self
5627 .positions_refs(
5628 None,
5629 Some(&fill.instrument_id),
5630 Some(&fill.strategy_id),
5631 Some(&fill.account_id),
5632 None,
5633 )
5634 .iter()
5635 .any(|position| {
5636 position.id != source_id
5637 && position.id.is_virtual()
5638 && self.oms_type(&position.id) == Some(OmsType::Hedging)
5639 && position.replay_events.iter().any(|event| {
5640 matches!(event, PositionReplayEvent::Filled(opening)
5641 if opening.causation_id == Some(fill.event_id)
5642 && opening.client_order_id == client_order_id
5643 && opening.trade_id == fill.trade_id)
5644 })
5645 });
5646
5647 anyhow::ensure!(
5648 !has_opening,
5649 "missing virtual flip source fragment for order {client_order_id}"
5650 );
5651 continue;
5652 };
5653
5654 let (positions, corrected_closing) =
5655 self.hedging_flip_destination(fill, source, closing)?;
5656 anyhow::ensure!(
5657 flip.is_none() || corrected_source == Some(source.id),
5658 "ambiguous virtual flip history for order {client_order_id}"
5659 );
5660 flip = Some(positions);
5661 corrected_source = corrected_closing.then_some(source.id);
5662 }
5663
5664 Ok(flip)
5665 }
5666
5667 fn hedging_flip_destination(
5668 &self,
5669 fill: &OrderFilled,
5670 source: &Position,
5671 closing: &OrderFilled,
5672 ) -> anyhow::Result<([PositionId; 2], bool)> {
5673 let positions = self.positions_refs(
5674 None,
5675 Some(&fill.instrument_id),
5676 Some(&fill.strategy_id),
5677 Some(&fill.account_id),
5678 None,
5679 );
5680
5681 let mut openings = positions.iter().flat_map(|position| {
5682 position
5683 .replay_events
5684 .iter()
5685 .filter_map(move |event| match event {
5686 PositionReplayEvent::Filled(opening)
5687 if opening.causation_id == Some(fill.event_id)
5688 && position.id != source.id =>
5689 {
5690 Some((position, opening))
5691 }
5692 _ => None,
5693 })
5694 });
5695
5696 let (destination, opening) = openings.next().ok_or_else(|| {
5697 anyhow::anyhow!(
5698 "missing virtual flip fragment for order {}",
5699 fill.client_order_id
5700 )
5701 })?;
5702
5703 anyhow::ensure!(
5704 openings.next().is_none(),
5705 "ambiguous virtual flip fragments for order {}",
5706 fill.client_order_id
5707 );
5708 anyhow::ensure!(
5709 source.instrument_id == fill.instrument_id
5710 && source.account_id == fill.account_id
5711 && source.strategy_id == fill.strategy_id
5712 && destination.id.is_virtual()
5713 && self.oms_type(&destination.id) == Some(OmsType::Hedging),
5714 "conflicting virtual flip positions for order {}",
5715 fill.client_order_id
5716 );
5717 let (expected_closing, expected_opening) =
5718 fill.split_for_position_flip(closing.last_qty, Some(destination.id), opening.event_id)?;
5719 anyhow::ensure!(
5720 *closing == expected_closing && *opening == expected_opening,
5721 "conflicting virtual flip provenance for order {}",
5722 fill.client_order_id
5723 );
5724
5725 let corrected_closing = source.fill_voids.iter().any(|correction| {
5726 correction.event.client_order_id == fill.client_order_id
5727 && correction.event.trade_id == fill.trade_id
5728 && correction.voided_qty.non_zero()
5729 });
5730
5731 if corrected_closing && source.is_open() {
5732 anyhow::ensure!(
5733 !destination.is_open(),
5734 "ambiguous corrected virtual flip for order {}",
5735 fill.client_order_id
5736 );
5737 return Ok(([destination.id, source.id], corrected_closing));
5738 }
5739
5740 Ok(([source.id, destination.id], corrected_closing))
5741 }
5742
5743 fn reindex_hedging_position(&mut self, client_order_id: ClientOrderId) {
5744 match self.hedging_flip_position_ids(client_order_id) {
5745 Ok(Some([_, position_id])) => {
5746 let position = self
5747 .position_ref(&position_id)
5748 .expect("proven position is cached");
5749 let venue = position.instrument_id.venue;
5750 let strategy_id = position.strategy_id;
5751 drop(position);
5752
5753 if let Err(e) =
5754 self.add_position_id(&position_id, &venue, &client_order_id, &strategy_id)
5755 {
5756 log::error!("Failed to index virtual flip for order {client_order_id}: {e}");
5757 }
5758 }
5759 Ok(None) => {}
5760 Err(e) => {
5761 self.index.order_position.remove(&client_order_id);
5762
5763 log::error!("Cannot index virtual flip for order {client_order_id}: {e}");
5764 }
5765 }
5766 }
5767
5768 #[must_use]
5774 pub fn positions_refs(
5775 &self,
5776 venue: Option<&Venue>,
5777 instrument_id: Option<&InstrumentId>,
5778 strategy_id: Option<&StrategyId>,
5779 account_id: Option<&AccountId>,
5780 side: Option<PositionSide>,
5781 ) -> Vec<PositionRef<'_>> {
5782 let position_ids = self.position_ids(venue, instrument_id, strategy_id, account_id);
5783 self.get_positions_for_ids(&position_ids, side)
5784 }
5785
5786 #[must_use]
5790 pub fn positions(
5791 &self,
5792 venue: Option<&Venue>,
5793 instrument_id: Option<&InstrumentId>,
5794 strategy_id: Option<&StrategyId>,
5795 account_id: Option<&AccountId>,
5796 side: Option<PositionSide>,
5797 ) -> Vec<PositionRef<'_>> {
5798 self.positions_refs(venue, instrument_id, strategy_id, account_id, side)
5799 }
5800
5801 #[must_use]
5803 pub fn positions_open_refs(
5804 &self,
5805 venue: Option<&Venue>,
5806 instrument_id: Option<&InstrumentId>,
5807 strategy_id: Option<&StrategyId>,
5808 account_id: Option<&AccountId>,
5809 side: Option<PositionSide>,
5810 ) -> Vec<PositionRef<'_>> {
5811 let position_ids = self.position_open_ids(venue, instrument_id, strategy_id, account_id);
5812 self.get_positions_for_ids(&position_ids, side)
5813 }
5814
5815 #[must_use]
5819 pub fn positions_open(
5820 &self,
5821 venue: Option<&Venue>,
5822 instrument_id: Option<&InstrumentId>,
5823 strategy_id: Option<&StrategyId>,
5824 account_id: Option<&AccountId>,
5825 side: Option<PositionSide>,
5826 ) -> Vec<PositionRef<'_>> {
5827 self.positions_open_refs(venue, instrument_id, strategy_id, account_id, side)
5828 }
5829
5830 #[must_use]
5832 pub fn positions_closed_refs(
5833 &self,
5834 venue: Option<&Venue>,
5835 instrument_id: Option<&InstrumentId>,
5836 strategy_id: Option<&StrategyId>,
5837 account_id: Option<&AccountId>,
5838 side: Option<PositionSide>,
5839 ) -> Vec<PositionRef<'_>> {
5840 let position_ids = self.position_closed_ids(venue, instrument_id, strategy_id, account_id);
5841 self.get_positions_for_ids(&position_ids, side)
5842 }
5843
5844 #[must_use]
5848 pub fn positions_closed(
5849 &self,
5850 venue: Option<&Venue>,
5851 instrument_id: Option<&InstrumentId>,
5852 strategy_id: Option<&StrategyId>,
5853 account_id: Option<&AccountId>,
5854 side: Option<PositionSide>,
5855 ) -> Vec<PositionRef<'_>> {
5856 self.positions_closed_refs(venue, instrument_id, strategy_id, account_id, side)
5857 }
5858
5859 #[must_use]
5861 pub fn position_exists(&self, position_id: &PositionId) -> bool {
5862 self.index.positions.contains(position_id)
5863 }
5864
5865 #[must_use]
5867 pub fn is_position_open(&self, position_id: &PositionId) -> bool {
5868 self.index.positions_open.contains(position_id)
5869 }
5870
5871 #[must_use]
5873 pub fn is_position_closed(&self, position_id: &PositionId) -> bool {
5874 self.index.positions_closed.contains(position_id)
5875 }
5876
5877 #[must_use]
5879 pub fn positions_open_count(
5880 &self,
5881 venue: Option<&Venue>,
5882 instrument_id: Option<&InstrumentId>,
5883 strategy_id: Option<&StrategyId>,
5884 account_id: Option<&AccountId>,
5885 side: Option<PositionSide>,
5886 ) -> usize {
5887 self.count_positions_in_bucket(
5888 &self.index.positions_open,
5889 venue,
5890 instrument_id,
5891 strategy_id,
5892 account_id,
5893 side,
5894 )
5895 }
5896
5897 #[must_use]
5899 pub fn positions_closed_count(
5900 &self,
5901 venue: Option<&Venue>,
5902 instrument_id: Option<&InstrumentId>,
5903 strategy_id: Option<&StrategyId>,
5904 account_id: Option<&AccountId>,
5905 side: Option<PositionSide>,
5906 ) -> usize {
5907 self.count_positions_in_bucket(
5908 &self.index.positions_closed,
5909 venue,
5910 instrument_id,
5911 strategy_id,
5912 account_id,
5913 side,
5914 )
5915 }
5916
5917 #[must_use]
5919 pub fn positions_total_count(
5920 &self,
5921 venue: Option<&Venue>,
5922 instrument_id: Option<&InstrumentId>,
5923 strategy_id: Option<&StrategyId>,
5924 account_id: Option<&AccountId>,
5925 side: Option<PositionSide>,
5926 ) -> usize {
5927 self.count_positions_in_bucket(
5928 &self.index.positions,
5929 venue,
5930 instrument_id,
5931 strategy_id,
5932 account_id,
5933 side,
5934 )
5935 }
5936
5937 #[must_use]
5943 pub fn has_positions_open(
5944 &self,
5945 venue: Option<&Venue>,
5946 instrument_id: Option<&InstrumentId>,
5947 strategy_id: Option<&StrategyId>,
5948 account_id: Option<&AccountId>,
5949 side: Option<PositionSide>,
5950 ) -> bool {
5951 self.any_positions_in_bucket(
5952 &self.index.positions_open,
5953 venue,
5954 instrument_id,
5955 strategy_id,
5956 account_id,
5957 side,
5958 )
5959 }
5960
5961 #[must_use]
5963 pub fn has_positions_closed(
5964 &self,
5965 venue: Option<&Venue>,
5966 instrument_id: Option<&InstrumentId>,
5967 strategy_id: Option<&StrategyId>,
5968 account_id: Option<&AccountId>,
5969 side: Option<PositionSide>,
5970 ) -> bool {
5971 self.any_positions_in_bucket(
5972 &self.index.positions_closed,
5973 venue,
5974 instrument_id,
5975 strategy_id,
5976 account_id,
5977 side,
5978 )
5979 }
5980
5981 #[must_use]
5983 pub fn has_positions(
5984 &self,
5985 venue: Option<&Venue>,
5986 instrument_id: Option<&InstrumentId>,
5987 strategy_id: Option<&StrategyId>,
5988 account_id: Option<&AccountId>,
5989 side: Option<PositionSide>,
5990 ) -> bool {
5991 self.any_positions_in_bucket(
5992 &self.index.positions,
5993 venue,
5994 instrument_id,
5995 strategy_id,
5996 account_id,
5997 side,
5998 )
5999 }
6000
6001 #[must_use]
6005 pub fn strategy_id_for_order(&self, client_order_id: &ClientOrderId) -> Option<&StrategyId> {
6006 self.index.order_strategy.get(client_order_id)
6007 }
6008
6009 #[must_use]
6011 pub fn strategy_id_for_position(&self, position_id: &PositionId) -> Option<&StrategyId> {
6012 self.index.position_strategy.get(position_id)
6013 }
6014
6015 pub fn get(&self, key: &str) -> anyhow::Result<Option<&Bytes>> {
6023 check_valid_string_utf8(key, stringify!(key))?;
6024
6025 Ok(self.general.get(key))
6026 }
6027
6028 #[must_use]
6038 pub fn price(&self, instrument_id: &InstrumentId, price_type: PriceType) -> Option<Price> {
6039 match price_type {
6040 PriceType::Bid => self
6041 .quotes
6042 .get(instrument_id)
6043 .and_then(|quotes| quotes.front().map(|quote| quote.bid_price)),
6044 PriceType::Ask => self
6045 .quotes
6046 .get(instrument_id)
6047 .and_then(|quotes| quotes.front().map(|quote| quote.ask_price)),
6048 PriceType::Mid => self.quotes.get(instrument_id).and_then(|quotes| {
6049 let quote = quotes.front()?;
6050 let bid = quote.bid_price;
6051 let ask = quote.ask_price;
6052 if bid.is_undefined() || bid.is_error() || ask.is_undefined() || ask.is_error() {
6053 return None;
6054 }
6055
6056 let precision = bid.precision;
6058 check_fixed_precision(precision).ok()?;
6059
6060 #[allow(
6061 clippy::useless_conversion,
6062 reason = "i128::from is a widening conversion when PriceRaw is i64"
6063 )]
6064 let (mut bid_raw, mut ask_raw) = (i128::from(bid.raw()), i128::from(ask.raw()));
6065
6066 if precision < FIXED_PRECISION {
6067 let factor = 10_i128.pow(u32::from(FIXED_PRECISION - precision));
6068 bid_raw = (bid_raw / factor) * factor;
6069 ask_raw = (ask_raw / factor) * factor;
6070 }
6071 let sum = bid_raw.checked_add(ask_raw)?;
6072
6073 if precision < FIXED_PRECISION {
6074 let raw = PriceRaw::try_from(sum / 2).ok()?;
6075 return Price::from_raw_checked(raw, precision + 1).ok();
6076 }
6077
6078 if precision != MAX_FLOAT_PRECISION
6079 && let Some(price) = sum
6080 .checked_mul(5)
6081 .and_then(|raw| PriceRaw::try_from(raw).ok())
6082 .and_then(|raw| Price::from_raw_checked(raw, precision + 1).ok())
6083 {
6084 return Some(price);
6085 }
6086
6087 let mut raw = sum / 2;
6088 if sum % 2 != 0 && raw % 2 != 0 {
6089 raw += sum.signum();
6090 }
6091 let raw = PriceRaw::try_from(raw).ok()?;
6092 Price::from_raw_checked(raw, precision).ok()
6093 }),
6094 PriceType::Last => self
6095 .trades
6096 .get(instrument_id)
6097 .and_then(|trades| trades.front().map(|trade| trade.price)),
6098 PriceType::Mark => self
6099 .mark_prices
6100 .get(instrument_id)
6101 .and_then(|marks| marks.front().map(|mark| mark.value)),
6102 }
6103 }
6104
6105 #[must_use]
6107 pub fn quotes(&self, instrument_id: &InstrumentId) -> Option<Vec<QuoteTick>> {
6108 self.quotes
6109 .get(instrument_id)
6110 .map(|quotes| quotes.iter().copied().collect())
6111 }
6112
6113 #[must_use]
6115 pub fn trades(&self, instrument_id: &InstrumentId) -> Option<Vec<TradeTick>> {
6116 self.trades
6117 .get(instrument_id)
6118 .map(|trades| trades.iter().copied().collect())
6119 }
6120
6121 #[must_use]
6123 pub fn mark_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<MarkPriceUpdate>> {
6124 self.mark_prices
6125 .get(instrument_id)
6126 .map(|mark_prices| mark_prices.iter().copied().collect())
6127 }
6128
6129 #[must_use]
6131 pub fn index_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<IndexPriceUpdate>> {
6132 self.index_prices
6133 .get(instrument_id)
6134 .map(|index_prices| index_prices.iter().copied().collect())
6135 }
6136
6137 #[must_use]
6139 pub fn funding_rates(&self, instrument_id: &InstrumentId) -> Option<Vec<FundingRateUpdate>> {
6140 self.funding_rates
6141 .get(instrument_id)
6142 .map(|funding_rates| funding_rates.iter().copied().collect())
6143 }
6144
6145 #[must_use]
6147 pub fn instrument_statuses(
6148 &self,
6149 instrument_id: &InstrumentId,
6150 ) -> Option<Vec<InstrumentStatus>> {
6151 self.instrument_statuses
6152 .get(instrument_id)
6153 .map(|statuses| statuses.iter().copied().collect())
6154 }
6155
6156 #[must_use]
6158 pub fn bars(&self, bar_type: &BarType) -> Option<Vec<Bar>> {
6159 self.bars
6160 .get(bar_type)
6161 .map(|bars| bars.iter().copied().collect())
6162 }
6163
6164 #[must_use]
6166 pub fn order_book(&self, instrument_id: &InstrumentId) -> Option<&OrderBook> {
6167 self.books.get(instrument_id)
6168 }
6169
6170 pub fn try_order_book(
6176 &self,
6177 instrument_id: &InstrumentId,
6178 ) -> Result<&OrderBook, OrderBookLookupError> {
6179 self.books
6180 .get(instrument_id)
6181 .ok_or_else(|| OrderBookLookupError::not_found(*instrument_id))
6182 }
6183
6184 #[must_use]
6186 pub fn order_book_mut(&mut self, instrument_id: &InstrumentId) -> Option<&mut OrderBook> {
6187 self.books.get_mut(instrument_id)
6188 }
6189
6190 #[must_use]
6192 pub fn own_order_book(&self, instrument_id: &InstrumentId) -> Option<&OwnOrderBook> {
6193 self.own_books.get(instrument_id)
6194 }
6195
6196 pub fn try_own_order_book(
6202 &self,
6203 instrument_id: &InstrumentId,
6204 ) -> Result<&OwnOrderBook, OwnOrderBookLookupError> {
6205 self.own_books
6206 .get(instrument_id)
6207 .ok_or_else(|| OwnOrderBookLookupError::not_found(*instrument_id))
6208 }
6209
6210 #[must_use]
6212 pub fn own_order_book_mut(
6213 &mut self,
6214 instrument_id: &InstrumentId,
6215 ) -> Option<&mut OwnOrderBook> {
6216 self.own_books.get_mut(instrument_id)
6217 }
6218
6219 #[must_use]
6221 pub fn quote(&self, instrument_id: &InstrumentId) -> Option<&QuoteTick> {
6222 self.quotes
6223 .get(instrument_id)
6224 .and_then(|quotes| quotes.front())
6225 }
6226
6227 #[must_use]
6231 pub fn quote_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&QuoteTick> {
6232 self.quotes
6233 .get(instrument_id)
6234 .and_then(|quotes| quotes.get(index))
6235 }
6236
6237 #[must_use]
6239 pub fn trade(&self, instrument_id: &InstrumentId) -> Option<&TradeTick> {
6240 self.trades
6241 .get(instrument_id)
6242 .and_then(|trades| trades.front())
6243 }
6244
6245 #[must_use]
6249 pub fn trade_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&TradeTick> {
6250 self.trades
6251 .get(instrument_id)
6252 .and_then(|trades| trades.get(index))
6253 }
6254
6255 #[must_use]
6257 pub fn mark_price(&self, instrument_id: &InstrumentId) -> Option<&MarkPriceUpdate> {
6258 self.mark_prices
6259 .get(instrument_id)
6260 .and_then(|mark_prices| mark_prices.front())
6261 }
6262
6263 #[must_use]
6265 pub fn index_price(&self, instrument_id: &InstrumentId) -> Option<&IndexPriceUpdate> {
6266 self.index_prices
6267 .get(instrument_id)
6268 .and_then(|index_prices| index_prices.front())
6269 }
6270
6271 #[must_use]
6273 pub fn funding_rate(&self, instrument_id: &InstrumentId) -> Option<&FundingRateUpdate> {
6274 self.funding_rates
6275 .get(instrument_id)
6276 .and_then(|funding_rates| funding_rates.front())
6277 }
6278
6279 #[must_use]
6281 pub fn instrument_status(&self, instrument_id: &InstrumentId) -> Option<&InstrumentStatus> {
6282 self.instrument_statuses
6283 .get(instrument_id)
6284 .and_then(|statuses| statuses.front())
6285 }
6286
6287 #[must_use]
6289 pub fn instrument_close(&self, instrument_id: &InstrumentId) -> Option<&InstrumentClose> {
6290 self.instrument_closes.get(instrument_id)
6291 }
6292
6293 #[must_use]
6295 pub fn instrument_close_ids(&self) -> Vec<&InstrumentId> {
6296 self.instrument_closes.keys().collect()
6297 }
6298
6299 #[must_use]
6301 pub fn bar(&self, bar_type: &BarType) -> Option<&Bar> {
6302 self.bars.get(bar_type).and_then(|bars| bars.front())
6303 }
6304
6305 #[must_use]
6309 pub fn bar_at_index(&self, bar_type: &BarType, index: usize) -> Option<&Bar> {
6310 self.bars.get(bar_type).and_then(|bars| bars.get(index))
6311 }
6312
6313 #[must_use]
6315 pub fn book_update_count(&self, instrument_id: &InstrumentId) -> usize {
6316 self.books
6317 .get(instrument_id)
6318 .map_or(0, |book| book.update_count) as usize
6319 }
6320
6321 #[must_use]
6323 pub fn quote_count(&self, instrument_id: &InstrumentId) -> usize {
6324 self.quotes
6325 .get(instrument_id)
6326 .map_or(0, BoundedVecDeque::len)
6327 }
6328
6329 #[must_use]
6331 pub fn trade_count(&self, instrument_id: &InstrumentId) -> usize {
6332 self.trades
6333 .get(instrument_id)
6334 .map_or(0, BoundedVecDeque::len)
6335 }
6336
6337 #[must_use]
6339 pub fn mark_price_count(&self, instrument_id: &InstrumentId) -> usize {
6340 self.mark_prices
6341 .get(instrument_id)
6342 .map_or(0, BoundedVecDeque::len)
6343 }
6344
6345 #[must_use]
6347 pub fn index_price_count(&self, instrument_id: &InstrumentId) -> usize {
6348 self.index_prices
6349 .get(instrument_id)
6350 .map_or(0, BoundedVecDeque::len)
6351 }
6352
6353 #[must_use]
6355 pub fn funding_rate_count(&self, instrument_id: &InstrumentId) -> usize {
6356 self.funding_rates
6357 .get(instrument_id)
6358 .map_or(0, BoundedVecDeque::len)
6359 }
6360
6361 #[must_use]
6363 pub fn instrument_status_count(&self, instrument_id: &InstrumentId) -> usize {
6364 self.instrument_statuses
6365 .get(instrument_id)
6366 .map_or(0, BoundedVecDeque::len)
6367 }
6368
6369 #[must_use]
6371 pub fn bar_count(&self, bar_type: &BarType) -> usize {
6372 self.bars.get(bar_type).map_or(0, BoundedVecDeque::len)
6373 }
6374
6375 #[must_use]
6377 pub fn has_order_book(&self, instrument_id: &InstrumentId) -> bool {
6378 self.books.contains_key(instrument_id)
6379 }
6380
6381 #[must_use]
6383 pub fn has_quote_ticks(&self, instrument_id: &InstrumentId) -> bool {
6384 self.quote_count(instrument_id) > 0
6385 }
6386
6387 #[must_use]
6389 pub fn has_trade_ticks(&self, instrument_id: &InstrumentId) -> bool {
6390 self.trade_count(instrument_id) > 0
6391 }
6392
6393 #[must_use]
6395 pub fn has_mark_prices(&self, instrument_id: &InstrumentId) -> bool {
6396 self.mark_price_count(instrument_id) > 0
6397 }
6398
6399 #[must_use]
6401 pub fn has_index_prices(&self, instrument_id: &InstrumentId) -> bool {
6402 self.index_price_count(instrument_id) > 0
6403 }
6404
6405 #[must_use]
6407 pub fn has_funding_rates(&self, instrument_id: &InstrumentId) -> bool {
6408 self.funding_rate_count(instrument_id) > 0
6409 }
6410
6411 #[must_use]
6413 pub fn has_instrument_statuses(&self, instrument_id: &InstrumentId) -> bool {
6414 self.instrument_status_count(instrument_id) > 0
6415 }
6416
6417 #[must_use]
6419 pub fn has_instrument_close(&self, instrument_id: &InstrumentId) -> bool {
6420 self.instrument_closes.contains_key(instrument_id)
6421 }
6422
6423 #[must_use]
6425 pub fn has_bars(&self, bar_type: &BarType) -> bool {
6426 self.bar_count(bar_type) > 0
6427 }
6428
6429 #[must_use]
6430 pub fn get_xrate(
6431 &self,
6432 venue: Venue,
6433 from_currency: Currency,
6434 to_currency: Currency,
6435 price_type: PriceType,
6436 ) -> Option<Decimal> {
6437 match self.try_get_xrate(venue, from_currency, to_currency, price_type) {
6438 Ok(rate) => rate,
6439 Err(e) => {
6440 log::error!("Failed to calculate xrate: {e}");
6441 None
6442 }
6443 }
6444 }
6445
6446 pub fn try_get_xrate(
6452 &self,
6453 venue: Venue,
6454 from_currency: Currency,
6455 to_currency: Currency,
6456 price_type: PriceType,
6457 ) -> anyhow::Result<Option<Decimal>> {
6458 if from_currency == to_currency {
6459 return Ok(Some(Decimal::ONE));
6462 }
6463
6464 let (bid_quote, ask_quote) = self.build_quote_table(&venue);
6465
6466 get_exchange_rate(
6467 from_currency.code,
6468 to_currency.code,
6469 price_type,
6470 bid_quote,
6471 ask_quote,
6472 )
6473 }
6474
6475 fn build_quote_table(
6476 &self,
6477 venue: &Venue,
6478 ) -> (AHashMap<Ustr, Decimal>, AHashMap<Ustr, Decimal>) {
6479 let mut bid_quotes = AHashMap::new();
6480 let mut ask_quotes = AHashMap::new();
6481 let mut quote_sources = AHashMap::new();
6482 let mut bar_quotes = None;
6483 let mut pair_buffer = String::new();
6484
6485 for (instrument_id, instrument) in &self.instruments {
6486 if instrument_id.venue != *venue {
6487 continue;
6488 }
6489
6490 let Some(base_currency) = instrument.base_currency() else {
6491 continue;
6492 };
6493
6494 let (bid_price, ask_price) = if let Some(ticks) = self.quotes.get(instrument_id) {
6495 if let Some(tick) = ticks.front() {
6496 (tick.bid_price, tick.ask_price)
6497 } else {
6498 continue; }
6500 } else {
6501 let quotes = bar_quotes.get_or_insert_with(|| self.build_bar_quote_table(venue));
6502 match (
6503 quotes.get(&(*instrument_id, PriceType::Bid)),
6504 quotes.get(&(*instrument_id, PriceType::Ask)),
6505 ) {
6506 (Some((_, bid_bar)), Some((_, ask_bar))) => (bid_bar.close, ask_bar.close),
6507 _ => continue,
6508 }
6509 };
6510
6511 let base = base_currency.code.as_str();
6512 let quote = instrument.quote_currency().code.as_str();
6513 pair_buffer.clear();
6514 pair_buffer.reserve(base.len() + 1 + quote.len());
6515 pair_buffer.push_str(base);
6516 pair_buffer.push('/');
6517 pair_buffer.push_str(quote);
6518 let pair = Ustr::from(pair_buffer.as_str());
6519 let preference = (
6520 bid_price.is_positive() && ask_price.is_positive(),
6521 instrument.instrument_class() == InstrumentClass::Spot,
6522 Reverse(*instrument_id),
6523 );
6524
6525 if quote_sources
6526 .get(&pair)
6527 .is_some_and(|current| current >= &preference)
6528 {
6529 continue;
6530 }
6531
6532 bid_quotes.insert(pair, bid_price.as_decimal());
6533 ask_quotes.insert(pair, ask_price.as_decimal());
6534 quote_sources.insert(pair, preference);
6535 }
6536
6537 (bid_quotes, ask_quotes)
6538 }
6539
6540 fn build_bar_quote_table(
6541 &self,
6542 venue: &Venue,
6543 ) -> AHashMap<(InstrumentId, PriceType), (&BarType, &Bar)> {
6544 let mut quotes: AHashMap<_, (&BarType, &Bar)> = AHashMap::new();
6545
6546 for (bar_type, bars) in &self.bars {
6547 let instrument_id = bar_type.instrument_id();
6548 let price_type = bar_type.spec().price_type;
6549
6550 if instrument_id.venue != *venue
6551 || !matches!(price_type, PriceType::Bid | PriceType::Ask)
6552 {
6553 continue;
6554 }
6555
6556 let Some(bar) = bars.front() else {
6557 continue;
6558 };
6559
6560 quotes
6562 .entry((instrument_id, price_type))
6563 .and_modify(|current| {
6564 if (current.1.ts_init, current.0) < (bar.ts_init, bar_type) {
6565 *current = (bar_type, bar);
6566 }
6567 })
6568 .or_insert((bar_type, bar));
6569 }
6570
6571 quotes
6572 }
6573
6574 #[must_use]
6576 pub fn get_mark_xrate(&self, from_currency: Currency, to_currency: Currency) -> Option<f64> {
6577 self.mark_xrates.get(&(from_currency, to_currency)).copied()
6578 }
6579
6580 pub fn set_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency, xrate: f64) {
6586 assert!(xrate > 0.0, "xrate was zero");
6587 self.mark_xrates.insert((from_currency, to_currency), xrate);
6588 self.mark_xrates
6589 .insert((to_currency, from_currency), 1.0 / xrate);
6590 }
6591
6592 pub fn clear_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency) {
6598 let _ = self.mark_xrates.remove(&(from_currency, to_currency));
6599 }
6600
6601 pub fn clear_mark_xrates(&mut self) {
6603 self.mark_xrates.clear();
6604 }
6605
6606 #[must_use]
6608 pub fn currency(&self, code: &Ustr) -> Option<&Currency> {
6609 self.currencies.get(code)
6610 }
6611
6612 pub fn try_currency(&self, code: &Ustr) -> Result<&Currency, CurrencyLookupError> {
6618 self.currencies
6619 .get(code)
6620 .ok_or_else(|| CurrencyLookupError::not_found(*code))
6621 }
6622
6623 #[must_use]
6627 pub fn instrument(&self, instrument_id: &InstrumentId) -> Option<&InstrumentAny> {
6628 self.instruments.get(instrument_id)
6629 }
6630
6631 pub fn try_instrument(
6637 &self,
6638 instrument_id: &InstrumentId,
6639 ) -> Result<&InstrumentAny, InstrumentLookupError> {
6640 self.instruments
6641 .get(instrument_id)
6642 .ok_or_else(|| InstrumentLookupError::not_found(*instrument_id))
6643 }
6644
6645 #[must_use]
6647 pub fn instrument_ids(&self, venue: Option<&Venue>) -> Vec<&InstrumentId> {
6648 match venue {
6649 Some(v) => self.instruments.keys().filter(|i| &i.venue == v).collect(),
6650 None => self.instruments.keys().collect(),
6651 }
6652 }
6653
6654 #[must_use]
6656 pub fn instruments(&self, venue: &Venue, underlying: Option<&Ustr>) -> Vec<&InstrumentAny> {
6657 self.instruments
6658 .values()
6659 .filter(|i| &i.id().venue == venue)
6660 .filter(|i| underlying.is_none_or(|u| i.underlying() == Some(*u)))
6661 .collect()
6662 }
6663
6664 #[must_use]
6671 pub fn instruments_by_parent(
6672 &self,
6673 venue: &Venue,
6674 root: &Ustr,
6675 class: InstrumentClass,
6676 ) -> Vec<&InstrumentAny> {
6677 self.instruments
6678 .values()
6679 .filter(|i| &i.id().venue == venue)
6680 .filter(|i| i.underlying() == Some(*root))
6681 .filter(|i| i.instrument_class() == class)
6682 .collect()
6683 }
6684
6685 #[must_use]
6687 pub fn bar_types(
6688 &self,
6689 instrument_id: Option<&InstrumentId>,
6690 price_type: Option<&PriceType>,
6691 aggregation_source: AggregationSource,
6692 ) -> Vec<&BarType> {
6693 let mut bar_types = self
6694 .bars
6695 .keys()
6696 .filter(|bar_type| bar_type.aggregation_source() == aggregation_source)
6697 .collect::<Vec<&BarType>>();
6698
6699 if let Some(instrument_id) = instrument_id {
6700 bar_types.retain(|bar_type| bar_type.instrument_id() == *instrument_id);
6701 }
6702
6703 if let Some(price_type) = price_type {
6704 bar_types.retain(|bar_type| &bar_type.spec().price_type == price_type);
6705 }
6706
6707 bar_types
6708 }
6709
6710 #[must_use]
6714 pub fn synthetic(&self, instrument_id: &InstrumentId) -> Option<&SyntheticInstrument> {
6715 self.synthetics.get(instrument_id)
6716 }
6717
6718 pub fn try_synthetic(
6724 &self,
6725 instrument_id: &InstrumentId,
6726 ) -> Result<&SyntheticInstrument, SyntheticInstrumentLookupError> {
6727 self.synthetics
6728 .get(instrument_id)
6729 .ok_or_else(|| SyntheticInstrumentLookupError::not_found(*instrument_id))
6730 }
6731
6732 #[must_use]
6734 pub fn synthetic_ids(&self) -> Vec<&InstrumentId> {
6735 self.synthetics.keys().collect()
6736 }
6737
6738 #[must_use]
6740 pub fn synthetics(&self) -> Vec<&SyntheticInstrument> {
6741 self.synthetics.values().collect()
6742 }
6743
6744 #[must_use]
6748 pub fn account_ref(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
6749 self.accounts
6750 .get(account_id)
6751 .map(|account_cell| AccountRef::new(account_cell.borrow()))
6752 }
6753
6754 #[must_use]
6758 pub fn account(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
6759 self.account_ref(account_id)
6760 }
6761
6762 pub fn try_account_ref(
6768 &self,
6769 account_id: &AccountId,
6770 ) -> Result<AccountRef<'_>, AccountLookupError> {
6771 self.accounts
6772 .get(account_id)
6773 .map(|account_cell| AccountRef::new(account_cell.borrow()))
6774 .ok_or_else(|| AccountLookupError::not_found(*account_id))
6775 }
6776
6777 pub fn try_account(
6785 &self,
6786 account_id: &AccountId,
6787 ) -> Result<AccountRef<'_>, AccountLookupError> {
6788 self.try_account_ref(account_id)
6789 }
6790
6791 #[must_use]
6801 pub fn account_mut(&mut self, account_id: &AccountId) -> Option<AccountRefMut<'_>> {
6802 self.accounts
6803 .get(account_id)
6804 .map(|account_cell| AccountRefMut::new(account_cell.borrow_mut()))
6805 }
6806
6807 #[must_use]
6813 pub fn account_owned(&self, account_id: &AccountId) -> Option<AccountAny> {
6814 self.accounts.get(account_id).and_then(|account_cell| {
6815 account_cell
6816 .try_borrow()
6817 .ok()
6818 .map(|account| account.clone())
6819 })
6820 }
6821
6822 #[must_use]
6827 pub fn account_for_venue(&self, venue: &Venue) -> Option<AccountRef<'_>> {
6828 self.account_id(venue)
6829 .and_then(|account_id| self.accounts.get(account_id))
6830 .map(|account_cell| AccountRef::new(account_cell.borrow()))
6831 }
6832
6833 #[must_use]
6839 pub fn account_for_venue_owned(&self, venue: &Venue) -> Option<AccountAny> {
6840 self.account_id(venue)
6841 .and_then(|account_id| self.accounts.get(account_id))
6842 .map(|account_cell| account_cell.borrow().clone())
6843 }
6844
6845 #[must_use]
6849 pub fn account_id(&self, venue: &Venue) -> Option<&AccountId> {
6850 let mut account_ids = self.index.venue_accounts.get(venue)?.iter();
6851 let account_id = account_ids.next()?;
6852 account_ids.next().is_none().then_some(account_id)
6853 }
6854
6855 #[must_use]
6861 pub fn accounts(&self, account_id: &AccountId) -> Vec<AccountRef<'_>> {
6862 self.accounts
6863 .values()
6864 .filter(|account_cell| &account_cell.borrow().id() == account_id)
6865 .map(|account_cell| AccountRef::new(account_cell.borrow()))
6866 .collect()
6867 }
6868
6869 #[must_use]
6871 pub fn accounts_all_owned(&self) -> Vec<AccountAny> {
6872 self.accounts
6873 .values()
6874 .map(|account_cell| account_cell.borrow().clone())
6875 .collect()
6876 }
6877
6878 pub fn update_own_order_book(&mut self, order: &OrderAny) {
6886 if !order.has_price() {
6887 return;
6888 }
6889
6890 let instrument_id = order.instrument_id();
6891
6892 if !self.own_books.contains_key(&instrument_id) {
6893 if order.is_closed() {
6894 return;
6895 }
6896
6897 self.own_books
6898 .insert(instrument_id, OwnOrderBook::new(instrument_id));
6899 }
6900
6901 let Some(own_book) = self.own_books.get_mut(&instrument_id) else {
6902 return;
6903 };
6904
6905 let own_book_order = order.to_own_book_order();
6906
6907 if order.is_closed() {
6908 if let Err(e) = own_book.delete(own_book_order) {
6909 log::debug!(
6910 "Failed to delete order {} from own book: {e}",
6911 order.client_order_id(),
6912 );
6913 } else {
6914 log::debug!("Deleted order {} from own book", order.client_order_id());
6915 }
6916 } else {
6917 if let Err(e) = own_book.update(own_book_order) {
6919 log::debug!(
6920 "Failed to update order {} in own book: {e}; inserting instead",
6921 order.client_order_id(),
6922 );
6923 own_book.add(own_book_order);
6924 }
6925 log::debug!("Updated order {} in own book", order.client_order_id());
6926 }
6927 }
6928
6929 pub fn force_remove_from_own_order_book(&mut self, client_order_id: &ClientOrderId) {
6935 let Some(order_cell) = self.orders.get(client_order_id) else {
6936 return;
6937 };
6938 let order = order_cell.borrow();
6939 let instrument_id = order.instrument_id();
6940 let own_book_order = if order.has_price() {
6941 Some(order.to_own_book_order())
6942 } else {
6943 None
6944 };
6945 drop(order);
6946
6947 self.index.orders_open.remove(client_order_id);
6948 self.index.orders_pending_cancel.remove(client_order_id);
6949 self.index.orders_inflight.remove(client_order_id);
6950 self.index.orders_emulated.remove(client_order_id);
6951 self.index.orders_active_local.remove(client_order_id);
6952
6953 if let Some(own_book) = self.own_books.get_mut(&instrument_id)
6954 && let Some(own_book_order) = own_book_order
6955 {
6956 if let Err(e) = own_book.delete(own_book_order) {
6957 log::debug!("Could not force delete {client_order_id} from own book: {e}");
6958 } else {
6959 log::debug!("Force deleted {client_order_id} from own book");
6960 }
6961 }
6962
6963 self.index.orders_closed.insert(*client_order_id);
6964 }
6965
6966 pub fn audit_own_order_books(&mut self) {
6971 log::debug!("Starting own books audit");
6972 let start = std::time::Instant::now();
6973
6974 let valid_order_ids: AHashSet<ClientOrderId> = self
6975 .index
6976 .orders_open
6977 .iter()
6978 .chain(&self.index.orders_inflight)
6979 .chain(&self.index.orders_active_local)
6980 .copied()
6981 .collect();
6982
6983 for own_book in self.own_books.values_mut() {
6984 own_book.audit_open_orders(&valid_order_ids);
6985 }
6986
6987 log::debug!("Completed own books audit in {:?}", start.elapsed());
6988 }
6989}
6990
6991const POSITION_OMS_KEY_PREFIX: &str = "position_oms:";
6992
6993fn position_oms_key(position_id: PositionId) -> String {
6994 format!("{POSITION_OMS_KEY_PREFIX}{position_id}")
6995}
6996
6997fn insert_bar(bars: &mut BoundedVecDeque<Bar>, bar: Bar) {
7002 match bars.front() {
7003 None => bars.push_front(bar),
7004 Some(front) => match bar.ts_event.cmp(&front.ts_event) {
7005 Ordering::Greater => bars.push_front(bar),
7006 Ordering::Equal => {
7007 if bar.bar_type.spec().is_time_aggregated() {
7008 bars.replace_front(bar);
7009 } else {
7010 bars.push_front(bar);
7011 }
7012 }
7013 Ordering::Less => log::debug!(
7014 "Skipping bar {bar} with `ts_event` older than last bar `ts_event` {}",
7015 front.ts_event,
7016 ),
7017 },
7018 }
7019}
7020
7021fn insert_bar_historical(bars: &mut BoundedVecDeque<Bar>, bar: Bar) {
7026 let index = bars.partition_point(|cached| cached.ts_event > bar.ts_event);
7027
7028 if bar.bar_type.spec().is_time_aggregated()
7029 && bars
7030 .get(index)
7031 .is_some_and(|cached| cached.ts_event == bar.ts_event)
7032 {
7033 bars.replace(index, bar);
7034 } else {
7035 bars.insert(index, bar);
7036 }
7037}