1pub mod config;
21pub mod database;
22pub mod fifo;
23pub mod quote;
24pub mod refs;
25
26mod bounded;
27mod error;
28mod index;
29mod position;
30
31#[cfg(test)]
32mod tests;
33
34use std::{
35 borrow::Cow,
36 cell::{Ref, RefCell},
37 cmp::Reverse,
38 fmt::{Debug, Display},
39 rc::Rc,
40 time::{SystemTime, UNIX_EPOCH},
41};
42
43use ahash::{AHashMap, AHashSet};
44use bounded::BoundedVecDeque;
45use bytes::Bytes;
46pub use config::CacheConfig; use database::{CacheDatabaseAdapter, CacheMap};
48pub use error::{
49 ACCOUNT_NOT_FOUND, AccountLookupError, CURRENCY_NOT_FOUND, CurrencyLookupError,
50 INSTRUMENT_NOT_FOUND, InstrumentLookupError, ORDER_BOOK_NOT_FOUND, ORDER_LIST_NOT_FOUND,
51 ORDER_NOT_FOUND, OWN_ORDER_BOOK_NOT_FOUND, OrderBookLookupError, OrderListLookupError,
52 OrderLookupError, OwnOrderBookLookupError, POSITION_NOT_FOUND, PositionLookupError,
53 SYNTHETIC_INSTRUMENT_NOT_FOUND, SyntheticInstrumentLookupError, VenueOrderIdOwnershipError,
54};
55use index::CacheIndex;
56use indexmap::IndexMap;
57use nautilus_core::{
58 SharedCell, UnixNanos,
59 correctness::{
60 check_key_not_in_map, check_predicate_false, check_slice_not_empty,
61 check_valid_string_ascii,
62 },
63 datetime::secs_to_nanos,
64};
65#[cfg(feature = "defi")]
66use nautilus_model::defi::{Pool, PoolProfiler};
67use nautilus_model::{
68 accounts::{Account, AccountAny},
69 data::{
70 Bar, BarType, FundingRateUpdate, GreeksData, IndexPriceUpdate, InstrumentStatus,
71 MarkPriceUpdate, QuoteTick, TradeTick, YieldCurveData, option_chain::OptionGreeks,
72 },
73 enums::{
74 AggregationSource, ContingencyType, InstrumentClass, OmsType, OrderSide, PositionSide,
75 PriceType,
76 },
77 events::{AccountState, OrderEventAny, OrderFilled},
78 identifiers::{
79 AccountId, ActorId, ClientId, ClientOrderId, ExecAlgorithmId, InstrumentId, OrderListId,
80 PositionId, StrategyId, Venue, VenueOrderId,
81 },
82 instruments::{Instrument, InstrumentAny, SyntheticInstrument},
83 orderbook::{
84 OrderBook,
85 own::{OwnOrderBook, should_handle_own_book_order},
86 },
87 orders::{Order, OrderAny, OrderError, OrderList},
88 position::Position,
89 types::{Currency, Money, Price, Quantity},
90};
91pub use position::CacheSnapshotRef;
92use position::PositionSnapshotFrame;
93pub use refs::{AccountRef, AccountRefMut, OrderRef, OrderRefMut, PositionRef, PositionRefMut};
94use rust_decimal::Decimal;
95use ustr::Ustr;
96
97use crate::xrate::get_exchange_rate;
98
99#[derive(Clone, Debug)]
106pub struct CacheView {
107 inner: Rc<RefCell<Cache>>,
108}
109
110impl CacheView {
111 #[must_use]
113 pub fn new(inner: Rc<RefCell<Cache>>) -> Self {
114 Self { inner }
115 }
116
117 pub fn borrow(&self) -> Ref<'_, Cache> {
123 self.inner.borrow()
124 }
125}
126
127impl From<Rc<RefCell<Cache>>> for CacheView {
128 fn from(inner: Rc<RefCell<Cache>>) -> Self {
129 Self::new(inner)
130 }
131}
132
133#[derive(Debug)]
140pub struct CacheApi<'a> {
141 cache: &'a RefCell<Cache>,
142}
143
144impl<'a> CacheApi<'a> {
145 pub(crate) fn new(cache: &'a RefCell<Cache>) -> Self {
146 Self { cache }
147 }
148
149 #[must_use]
155 pub fn calculate_unrealized_pnl(&self, position: &Position) -> Option<Money> {
156 self.cache().calculate_unrealized_pnl(position)
157 }
158
159 #[must_use]
165 pub fn oms_type(&self, position_id: &PositionId) -> Option<OmsType> {
166 self.cache().oms_type(position_id)
167 }
168
169 #[must_use]
175 pub fn position_snapshot_bytes(&self, position_id: &PositionId) -> Option<Vec<Vec<u8>>> {
176 self.cache().position_snapshot_bytes(position_id)
177 }
178
179 #[must_use]
185 pub fn position_snapshot_count(&self, position_id: &PositionId) -> usize {
186 self.cache().position_snapshot_count(position_id)
187 }
188
189 #[must_use]
195 pub fn position_snapshots(
196 &self,
197 position_id: Option<&PositionId>,
198 account_id: Option<&AccountId>,
199 ) -> Vec<Position> {
200 self.cache().position_snapshots(position_id, account_id)
201 }
202
203 #[must_use]
209 pub fn position_snapshots_from(&self, position_id: &PositionId, skip: usize) -> Vec<Position> {
210 self.cache().position_snapshots_from(position_id, skip)
211 }
212
213 #[must_use]
219 pub fn position_snapshot_ids(&self, instrument_id: &InstrumentId) -> AHashSet<PositionId> {
220 self.cache().position_snapshot_ids(instrument_id)
221 }
222
223 #[must_use]
229 pub fn client_order_ids(
230 &self,
231 venue: Option<&Venue>,
232 instrument_id: Option<&InstrumentId>,
233 strategy_id: Option<&StrategyId>,
234 account_id: Option<&AccountId>,
235 ) -> AHashSet<ClientOrderId> {
236 self.cache()
237 .client_order_ids(venue, instrument_id, strategy_id, account_id)
238 }
239
240 #[must_use]
246 pub fn client_order_ids_open(
247 &self,
248 venue: Option<&Venue>,
249 instrument_id: Option<&InstrumentId>,
250 strategy_id: Option<&StrategyId>,
251 account_id: Option<&AccountId>,
252 ) -> AHashSet<ClientOrderId> {
253 self.cache()
254 .client_order_ids_open(venue, instrument_id, strategy_id, account_id)
255 }
256
257 #[must_use]
263 pub fn client_order_ids_closed(
264 &self,
265 venue: Option<&Venue>,
266 instrument_id: Option<&InstrumentId>,
267 strategy_id: Option<&StrategyId>,
268 account_id: Option<&AccountId>,
269 ) -> AHashSet<ClientOrderId> {
270 self.cache()
271 .client_order_ids_closed(venue, instrument_id, strategy_id, account_id)
272 }
273
274 #[must_use]
280 pub fn client_order_ids_active_local(
281 &self,
282 venue: Option<&Venue>,
283 instrument_id: Option<&InstrumentId>,
284 strategy_id: Option<&StrategyId>,
285 account_id: Option<&AccountId>,
286 ) -> AHashSet<ClientOrderId> {
287 self.cache()
288 .client_order_ids_active_local(venue, instrument_id, strategy_id, account_id)
289 }
290
291 #[must_use]
297 pub fn client_order_ids_emulated(
298 &self,
299 venue: Option<&Venue>,
300 instrument_id: Option<&InstrumentId>,
301 strategy_id: Option<&StrategyId>,
302 account_id: Option<&AccountId>,
303 ) -> AHashSet<ClientOrderId> {
304 self.cache()
305 .client_order_ids_emulated(venue, instrument_id, strategy_id, account_id)
306 }
307
308 #[must_use]
314 pub fn client_order_ids_inflight(
315 &self,
316 venue: Option<&Venue>,
317 instrument_id: Option<&InstrumentId>,
318 strategy_id: Option<&StrategyId>,
319 account_id: Option<&AccountId>,
320 ) -> AHashSet<ClientOrderId> {
321 self.cache()
322 .client_order_ids_inflight(venue, instrument_id, strategy_id, account_id)
323 }
324
325 #[must_use]
331 pub fn position_ids(
332 &self,
333 venue: Option<&Venue>,
334 instrument_id: Option<&InstrumentId>,
335 strategy_id: Option<&StrategyId>,
336 account_id: Option<&AccountId>,
337 ) -> AHashSet<PositionId> {
338 self.cache()
339 .position_ids(venue, instrument_id, strategy_id, account_id)
340 }
341
342 #[must_use]
348 pub fn position_open_ids(
349 &self,
350 venue: Option<&Venue>,
351 instrument_id: Option<&InstrumentId>,
352 strategy_id: Option<&StrategyId>,
353 account_id: Option<&AccountId>,
354 ) -> AHashSet<PositionId> {
355 self.cache()
356 .position_open_ids(venue, instrument_id, strategy_id, account_id)
357 }
358
359 #[must_use]
365 pub fn position_closed_ids(
366 &self,
367 venue: Option<&Venue>,
368 instrument_id: Option<&InstrumentId>,
369 strategy_id: Option<&StrategyId>,
370 account_id: Option<&AccountId>,
371 ) -> AHashSet<PositionId> {
372 self.cache()
373 .position_closed_ids(venue, instrument_id, strategy_id, account_id)
374 }
375
376 #[must_use]
382 pub fn strategy_ids(&self) -> AHashSet<StrategyId> {
383 self.cache().strategy_ids()
384 }
385
386 #[must_use]
392 pub fn exec_algorithm_ids(&self) -> AHashSet<ExecAlgorithmId> {
393 self.cache().exec_algorithm_ids()
394 }
395
396 #[must_use]
402 pub fn order(&self, client_order_id: &ClientOrderId) -> Option<OrderAny> {
403 self.cache().order_owned(client_order_id)
404 }
405
406 pub fn try_order(&self, client_order_id: &ClientOrderId) -> Result<OrderAny, OrderLookupError> {
417 self.cache().try_order_owned(client_order_id)
418 }
419
420 #[must_use]
426 pub fn orders_for_ids(
427 &self,
428 client_order_ids: &[ClientOrderId],
429 context: &dyn Display,
430 ) -> Vec<OrderAny> {
431 self.cache().orders_for_ids(client_order_ids, context)
432 }
433
434 #[must_use]
440 pub fn client_order_id(&self, venue_order_id: &VenueOrderId) -> Option<ClientOrderId> {
441 self.cache().client_order_id(venue_order_id).copied()
442 }
443
444 #[must_use]
450 pub fn venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
451 self.cache().venue_order_id(client_order_id).copied()
452 }
453
454 #[must_use]
460 pub fn client_id(&self, client_order_id: &ClientOrderId) -> Option<ClientId> {
461 self.cache().client_id(client_order_id).copied()
462 }
463
464 #[must_use]
470 pub fn orders(
471 &self,
472 venue: Option<&Venue>,
473 instrument_id: Option<&InstrumentId>,
474 strategy_id: Option<&StrategyId>,
475 account_id: Option<&AccountId>,
476 side: Option<OrderSide>,
477 ) -> Vec<OrderAny> {
478 self.cache()
479 .orders_refs(venue, instrument_id, strategy_id, account_id, side)
480 .into_iter()
481 .map(|order| order.cloned())
482 .collect()
483 }
484
485 #[must_use]
491 pub fn orders_open(
492 &self,
493 venue: Option<&Venue>,
494 instrument_id: Option<&InstrumentId>,
495 strategy_id: Option<&StrategyId>,
496 account_id: Option<&AccountId>,
497 side: Option<OrderSide>,
498 ) -> Vec<OrderAny> {
499 self.cache()
500 .orders_open_refs(venue, instrument_id, strategy_id, account_id, side)
501 .into_iter()
502 .map(|order| order.cloned())
503 .collect()
504 }
505
506 #[must_use]
512 pub fn orders_closed(
513 &self,
514 venue: Option<&Venue>,
515 instrument_id: Option<&InstrumentId>,
516 strategy_id: Option<&StrategyId>,
517 account_id: Option<&AccountId>,
518 side: Option<OrderSide>,
519 ) -> Vec<OrderAny> {
520 self.cache()
521 .orders_closed_refs(venue, instrument_id, strategy_id, account_id, side)
522 .into_iter()
523 .map(|order| order.cloned())
524 .collect()
525 }
526
527 #[must_use]
533 pub fn orders_active_local(
534 &self,
535 venue: Option<&Venue>,
536 instrument_id: Option<&InstrumentId>,
537 strategy_id: Option<&StrategyId>,
538 account_id: Option<&AccountId>,
539 side: Option<OrderSide>,
540 ) -> Vec<OrderAny> {
541 self.cache()
542 .orders_active_local_refs(venue, instrument_id, strategy_id, account_id, side)
543 .into_iter()
544 .map(|order| order.cloned())
545 .collect()
546 }
547
548 #[must_use]
554 pub fn orders_emulated(
555 &self,
556 venue: Option<&Venue>,
557 instrument_id: Option<&InstrumentId>,
558 strategy_id: Option<&StrategyId>,
559 account_id: Option<&AccountId>,
560 side: Option<OrderSide>,
561 ) -> Vec<OrderAny> {
562 self.cache()
563 .orders_emulated_refs(venue, instrument_id, strategy_id, account_id, side)
564 .into_iter()
565 .map(|order| order.cloned())
566 .collect()
567 }
568
569 #[must_use]
575 pub fn orders_inflight(
576 &self,
577 venue: Option<&Venue>,
578 instrument_id: Option<&InstrumentId>,
579 strategy_id: Option<&StrategyId>,
580 account_id: Option<&AccountId>,
581 side: Option<OrderSide>,
582 ) -> Vec<OrderAny> {
583 self.cache()
584 .orders_inflight_refs(venue, instrument_id, strategy_id, account_id, side)
585 .into_iter()
586 .map(|order| order.cloned())
587 .collect()
588 }
589
590 #[must_use]
596 pub fn orders_for_position(&self, position_id: &PositionId) -> Vec<OrderAny> {
597 self.cache()
598 .orders_for_position(position_id)
599 .into_iter()
600 .map(|order| order.cloned())
601 .collect()
602 }
603
604 #[must_use]
610 pub fn order_exists(&self, client_order_id: &ClientOrderId) -> bool {
611 self.cache().order_exists(client_order_id)
612 }
613
614 #[must_use]
620 pub fn is_order_open(&self, client_order_id: &ClientOrderId) -> bool {
621 self.cache().is_order_open(client_order_id)
622 }
623
624 #[must_use]
630 pub fn is_order_closed(&self, client_order_id: &ClientOrderId) -> bool {
631 self.cache().is_order_closed(client_order_id)
632 }
633
634 #[must_use]
640 pub fn is_order_active_local(&self, client_order_id: &ClientOrderId) -> bool {
641 self.cache().is_order_active_local(client_order_id)
642 }
643
644 #[must_use]
650 pub fn is_order_emulated(&self, client_order_id: &ClientOrderId) -> bool {
651 self.cache().is_order_emulated(client_order_id)
652 }
653
654 #[must_use]
660 pub fn is_order_inflight(&self, client_order_id: &ClientOrderId) -> bool {
661 self.cache().is_order_inflight(client_order_id)
662 }
663
664 #[must_use]
670 pub fn is_order_pending_cancel_local(&self, client_order_id: &ClientOrderId) -> bool {
671 self.cache().is_order_pending_cancel_local(client_order_id)
672 }
673
674 #[must_use]
680 pub fn orders_open_count(
681 &self,
682 venue: Option<&Venue>,
683 instrument_id: Option<&InstrumentId>,
684 strategy_id: Option<&StrategyId>,
685 account_id: Option<&AccountId>,
686 side: Option<OrderSide>,
687 ) -> usize {
688 self.cache()
689 .orders_open_count(venue, instrument_id, strategy_id, account_id, side)
690 }
691
692 #[must_use]
698 pub fn orders_closed_count(
699 &self,
700 venue: Option<&Venue>,
701 instrument_id: Option<&InstrumentId>,
702 strategy_id: Option<&StrategyId>,
703 account_id: Option<&AccountId>,
704 side: Option<OrderSide>,
705 ) -> usize {
706 self.cache()
707 .orders_closed_count(venue, instrument_id, strategy_id, account_id, side)
708 }
709
710 #[must_use]
716 pub fn orders_active_local_count(
717 &self,
718 venue: Option<&Venue>,
719 instrument_id: Option<&InstrumentId>,
720 strategy_id: Option<&StrategyId>,
721 account_id: Option<&AccountId>,
722 side: Option<OrderSide>,
723 ) -> usize {
724 self.cache()
725 .orders_active_local_count(venue, instrument_id, strategy_id, account_id, side)
726 }
727
728 #[must_use]
734 pub fn orders_emulated_count(
735 &self,
736 venue: Option<&Venue>,
737 instrument_id: Option<&InstrumentId>,
738 strategy_id: Option<&StrategyId>,
739 account_id: Option<&AccountId>,
740 side: Option<OrderSide>,
741 ) -> usize {
742 self.cache()
743 .orders_emulated_count(venue, instrument_id, strategy_id, account_id, side)
744 }
745
746 #[must_use]
752 pub fn orders_inflight_count(
753 &self,
754 venue: Option<&Venue>,
755 instrument_id: Option<&InstrumentId>,
756 strategy_id: Option<&StrategyId>,
757 account_id: Option<&AccountId>,
758 side: Option<OrderSide>,
759 ) -> usize {
760 self.cache()
761 .orders_inflight_count(venue, instrument_id, strategy_id, account_id, side)
762 }
763
764 #[must_use]
770 pub fn orders_total_count(
771 &self,
772 venue: Option<&Venue>,
773 instrument_id: Option<&InstrumentId>,
774 strategy_id: Option<&StrategyId>,
775 account_id: Option<&AccountId>,
776 side: Option<OrderSide>,
777 ) -> usize {
778 self.cache()
779 .orders_total_count(venue, instrument_id, strategy_id, account_id, side)
780 }
781
782 #[must_use]
788 pub fn has_orders_open(
789 &self,
790 venue: Option<&Venue>,
791 instrument_id: Option<&InstrumentId>,
792 strategy_id: Option<&StrategyId>,
793 account_id: Option<&AccountId>,
794 side: Option<OrderSide>,
795 ) -> bool {
796 self.cache()
797 .has_orders_open(venue, instrument_id, strategy_id, account_id, side)
798 }
799
800 #[must_use]
806 pub fn has_orders_closed(
807 &self,
808 venue: Option<&Venue>,
809 instrument_id: Option<&InstrumentId>,
810 strategy_id: Option<&StrategyId>,
811 account_id: Option<&AccountId>,
812 side: Option<OrderSide>,
813 ) -> bool {
814 self.cache()
815 .has_orders_closed(venue, instrument_id, strategy_id, account_id, side)
816 }
817
818 #[must_use]
824 pub fn has_orders_active_local(
825 &self,
826 venue: Option<&Venue>,
827 instrument_id: Option<&InstrumentId>,
828 strategy_id: Option<&StrategyId>,
829 account_id: Option<&AccountId>,
830 side: Option<OrderSide>,
831 ) -> bool {
832 self.cache()
833 .has_orders_active_local(venue, instrument_id, strategy_id, account_id, side)
834 }
835
836 #[must_use]
842 pub fn has_orders_emulated(
843 &self,
844 venue: Option<&Venue>,
845 instrument_id: Option<&InstrumentId>,
846 strategy_id: Option<&StrategyId>,
847 account_id: Option<&AccountId>,
848 side: Option<OrderSide>,
849 ) -> bool {
850 self.cache()
851 .has_orders_emulated(venue, instrument_id, strategy_id, account_id, side)
852 }
853
854 #[must_use]
860 pub fn has_orders_inflight(
861 &self,
862 venue: Option<&Venue>,
863 instrument_id: Option<&InstrumentId>,
864 strategy_id: Option<&StrategyId>,
865 account_id: Option<&AccountId>,
866 side: Option<OrderSide>,
867 ) -> bool {
868 self.cache()
869 .has_orders_inflight(venue, instrument_id, strategy_id, account_id, side)
870 }
871
872 #[must_use]
878 pub fn has_orders(
879 &self,
880 venue: Option<&Venue>,
881 instrument_id: Option<&InstrumentId>,
882 strategy_id: Option<&StrategyId>,
883 account_id: Option<&AccountId>,
884 side: Option<OrderSide>,
885 ) -> bool {
886 self.cache()
887 .has_orders(venue, instrument_id, strategy_id, account_id, side)
888 }
889
890 #[must_use]
896 pub fn order_list(&self, order_list_id: &OrderListId) -> Option<OrderList> {
897 self.cache().order_list(order_list_id).cloned()
898 }
899
900 pub fn try_order_list(
911 &self,
912 order_list_id: &OrderListId,
913 ) -> Result<OrderList, OrderListLookupError> {
914 self.cache().try_order_list(order_list_id).cloned()
915 }
916
917 #[must_use]
923 pub fn order_lists(
924 &self,
925 venue: Option<&Venue>,
926 instrument_id: Option<&InstrumentId>,
927 strategy_id: Option<&StrategyId>,
928 account_id: Option<&AccountId>,
929 ) -> Vec<OrderList> {
930 self.cache()
931 .order_lists(venue, instrument_id, strategy_id, account_id)
932 .into_iter()
933 .cloned()
934 .collect()
935 }
936
937 #[must_use]
943 pub fn order_list_exists(&self, order_list_id: &OrderListId) -> bool {
944 self.cache().order_list_exists(order_list_id)
945 }
946
947 #[must_use]
953 pub fn orders_for_exec_algorithm(
954 &self,
955 exec_algorithm_id: &ExecAlgorithmId,
956 venue: Option<&Venue>,
957 instrument_id: Option<&InstrumentId>,
958 strategy_id: Option<&StrategyId>,
959 account_id: Option<&AccountId>,
960 side: Option<OrderSide>,
961 ) -> Vec<OrderAny> {
962 self.cache()
963 .orders_for_exec_algorithm(
964 exec_algorithm_id,
965 venue,
966 instrument_id,
967 strategy_id,
968 account_id,
969 side,
970 )
971 .into_iter()
972 .map(|order| order.cloned())
973 .collect()
974 }
975
976 #[must_use]
982 pub fn orders_for_exec_spawn(&self, exec_spawn_id: &ClientOrderId) -> Vec<OrderAny> {
983 self.cache()
984 .orders_for_exec_spawn(exec_spawn_id)
985 .into_iter()
986 .map(|order| order.cloned())
987 .collect()
988 }
989
990 #[must_use]
996 pub fn exec_spawn_total_quantity(
997 &self,
998 exec_spawn_id: &ClientOrderId,
999 active_only: bool,
1000 ) -> Option<Quantity> {
1001 self.cache()
1002 .exec_spawn_total_quantity(exec_spawn_id, active_only)
1003 }
1004
1005 #[must_use]
1011 pub fn exec_spawn_total_filled_qty(
1012 &self,
1013 exec_spawn_id: &ClientOrderId,
1014 active_only: bool,
1015 ) -> Option<Quantity> {
1016 self.cache()
1017 .exec_spawn_total_filled_qty(exec_spawn_id, active_only)
1018 }
1019
1020 #[must_use]
1026 pub fn exec_spawn_total_leaves_qty(
1027 &self,
1028 exec_spawn_id: &ClientOrderId,
1029 active_only: bool,
1030 ) -> Option<Quantity> {
1031 self.cache()
1032 .exec_spawn_total_leaves_qty(exec_spawn_id, active_only)
1033 }
1034
1035 #[must_use]
1041 pub fn position(&self, position_id: &PositionId) -> Option<Position> {
1042 self.cache()
1043 .position_ref(position_id)
1044 .map(|position| position.cloned())
1045 }
1046
1047 pub fn try_position(&self, position_id: &PositionId) -> Result<Position, PositionLookupError> {
1058 self.cache()
1059 .try_position_ref(position_id)
1060 .map(|position| position.cloned())
1061 }
1062
1063 #[must_use]
1069 pub fn position_for_order(&self, client_order_id: &ClientOrderId) -> Option<Position> {
1070 self.cache()
1071 .position_for_order_ref(client_order_id)
1072 .map(|position| position.cloned())
1073 }
1074
1075 #[must_use]
1081 pub fn position_id(&self, client_order_id: &ClientOrderId) -> Option<PositionId> {
1082 self.cache().position_id(client_order_id).copied()
1083 }
1084
1085 #[must_use]
1091 pub fn positions(
1092 &self,
1093 venue: Option<&Venue>,
1094 instrument_id: Option<&InstrumentId>,
1095 strategy_id: Option<&StrategyId>,
1096 account_id: Option<&AccountId>,
1097 side: Option<PositionSide>,
1098 ) -> Vec<Position> {
1099 self.cache()
1100 .positions_refs(venue, instrument_id, strategy_id, account_id, side)
1101 .into_iter()
1102 .map(|position| position.cloned())
1103 .collect()
1104 }
1105
1106 #[must_use]
1112 pub fn positions_open(
1113 &self,
1114 venue: Option<&Venue>,
1115 instrument_id: Option<&InstrumentId>,
1116 strategy_id: Option<&StrategyId>,
1117 account_id: Option<&AccountId>,
1118 side: Option<PositionSide>,
1119 ) -> Vec<Position> {
1120 self.cache()
1121 .positions_open_refs(venue, instrument_id, strategy_id, account_id, side)
1122 .into_iter()
1123 .map(|position| position.cloned())
1124 .collect()
1125 }
1126
1127 #[must_use]
1133 pub fn positions_closed(
1134 &self,
1135 venue: Option<&Venue>,
1136 instrument_id: Option<&InstrumentId>,
1137 strategy_id: Option<&StrategyId>,
1138 account_id: Option<&AccountId>,
1139 side: Option<PositionSide>,
1140 ) -> Vec<Position> {
1141 self.cache()
1142 .positions_closed_refs(venue, instrument_id, strategy_id, account_id, side)
1143 .into_iter()
1144 .map(|position| position.cloned())
1145 .collect()
1146 }
1147
1148 #[must_use]
1154 pub fn position_exists(&self, position_id: &PositionId) -> bool {
1155 self.cache().position_exists(position_id)
1156 }
1157
1158 #[must_use]
1164 pub fn is_position_open(&self, position_id: &PositionId) -> bool {
1165 self.cache().is_position_open(position_id)
1166 }
1167
1168 #[must_use]
1174 pub fn is_position_closed(&self, position_id: &PositionId) -> bool {
1175 self.cache().is_position_closed(position_id)
1176 }
1177
1178 #[must_use]
1184 pub fn positions_open_count(
1185 &self,
1186 venue: Option<&Venue>,
1187 instrument_id: Option<&InstrumentId>,
1188 strategy_id: Option<&StrategyId>,
1189 account_id: Option<&AccountId>,
1190 side: Option<PositionSide>,
1191 ) -> usize {
1192 self.cache()
1193 .positions_open_count(venue, instrument_id, strategy_id, account_id, side)
1194 }
1195
1196 #[must_use]
1202 pub fn positions_closed_count(
1203 &self,
1204 venue: Option<&Venue>,
1205 instrument_id: Option<&InstrumentId>,
1206 strategy_id: Option<&StrategyId>,
1207 account_id: Option<&AccountId>,
1208 side: Option<PositionSide>,
1209 ) -> usize {
1210 self.cache()
1211 .positions_closed_count(venue, instrument_id, strategy_id, account_id, side)
1212 }
1213
1214 #[must_use]
1220 pub fn positions_total_count(
1221 &self,
1222 venue: Option<&Venue>,
1223 instrument_id: Option<&InstrumentId>,
1224 strategy_id: Option<&StrategyId>,
1225 account_id: Option<&AccountId>,
1226 side: Option<PositionSide>,
1227 ) -> usize {
1228 self.cache()
1229 .positions_total_count(venue, instrument_id, strategy_id, account_id, side)
1230 }
1231
1232 #[must_use]
1238 pub fn has_positions_open(
1239 &self,
1240 venue: Option<&Venue>,
1241 instrument_id: Option<&InstrumentId>,
1242 strategy_id: Option<&StrategyId>,
1243 account_id: Option<&AccountId>,
1244 side: Option<PositionSide>,
1245 ) -> bool {
1246 self.cache()
1247 .has_positions_open(venue, instrument_id, strategy_id, account_id, side)
1248 }
1249
1250 #[must_use]
1256 pub fn has_positions_closed(
1257 &self,
1258 venue: Option<&Venue>,
1259 instrument_id: Option<&InstrumentId>,
1260 strategy_id: Option<&StrategyId>,
1261 account_id: Option<&AccountId>,
1262 side: Option<PositionSide>,
1263 ) -> bool {
1264 self.cache()
1265 .has_positions_closed(venue, instrument_id, strategy_id, account_id, side)
1266 }
1267
1268 #[must_use]
1274 pub fn has_positions(
1275 &self,
1276 venue: Option<&Venue>,
1277 instrument_id: Option<&InstrumentId>,
1278 strategy_id: Option<&StrategyId>,
1279 account_id: Option<&AccountId>,
1280 side: Option<PositionSide>,
1281 ) -> bool {
1282 self.cache()
1283 .has_positions(venue, instrument_id, strategy_id, account_id, side)
1284 }
1285
1286 #[must_use]
1292 pub fn strategy_id_for_order(&self, client_order_id: &ClientOrderId) -> Option<StrategyId> {
1293 self.cache().strategy_id_for_order(client_order_id).copied()
1294 }
1295
1296 #[must_use]
1302 pub fn strategy_id_for_position(&self, position_id: &PositionId) -> Option<StrategyId> {
1303 self.cache().strategy_id_for_position(position_id).copied()
1304 }
1305
1306 pub fn get(&self, key: &str) -> anyhow::Result<Option<Bytes>> {
1317 let cache = self.cache();
1318 let value = cache.get(key)?;
1319 Ok(value.cloned())
1320 }
1321
1322 #[must_use]
1329 pub fn price(&self, instrument_id: &InstrumentId, price_type: PriceType) -> Option<Price> {
1330 self.cache().price(instrument_id, price_type)
1331 }
1332
1333 #[must_use]
1339 pub fn quotes(&self, instrument_id: &InstrumentId) -> Option<Vec<QuoteTick>> {
1340 self.cache().quotes(instrument_id)
1341 }
1342
1343 #[must_use]
1349 pub fn trades(&self, instrument_id: &InstrumentId) -> Option<Vec<TradeTick>> {
1350 self.cache().trades(instrument_id)
1351 }
1352
1353 #[must_use]
1359 pub fn mark_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<MarkPriceUpdate>> {
1360 self.cache().mark_prices(instrument_id)
1361 }
1362
1363 #[must_use]
1369 pub fn index_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<IndexPriceUpdate>> {
1370 self.cache().index_prices(instrument_id)
1371 }
1372
1373 #[must_use]
1379 pub fn funding_rates(&self, instrument_id: &InstrumentId) -> Option<Vec<FundingRateUpdate>> {
1380 self.cache().funding_rates(instrument_id)
1381 }
1382
1383 #[must_use]
1389 pub fn instrument_statuses(
1390 &self,
1391 instrument_id: &InstrumentId,
1392 ) -> Option<Vec<InstrumentStatus>> {
1393 self.cache().instrument_statuses(instrument_id)
1394 }
1395
1396 #[must_use]
1402 pub fn bars(&self, bar_type: &BarType) -> Option<Vec<Bar>> {
1403 self.cache().bars(bar_type)
1404 }
1405
1406 #[must_use]
1412 pub fn order_book(&self, instrument_id: &InstrumentId) -> Option<OrderBook> {
1413 self.cache().order_book(instrument_id).cloned()
1414 }
1415
1416 pub fn try_order_book(
1427 &self,
1428 instrument_id: &InstrumentId,
1429 ) -> Result<OrderBook, OrderBookLookupError> {
1430 self.cache().try_order_book(instrument_id).cloned()
1431 }
1432
1433 #[must_use]
1439 pub fn own_order_book(&self, instrument_id: &InstrumentId) -> Option<OwnOrderBook> {
1440 self.cache().own_order_book(instrument_id).cloned()
1441 }
1442
1443 pub fn try_own_order_book(
1455 &self,
1456 instrument_id: &InstrumentId,
1457 ) -> Result<OwnOrderBook, OwnOrderBookLookupError> {
1458 self.cache().try_own_order_book(instrument_id).cloned()
1459 }
1460
1461 #[must_use]
1467 pub fn quote(&self, instrument_id: &InstrumentId) -> Option<QuoteTick> {
1468 self.cache().quote(instrument_id).copied()
1469 }
1470
1471 #[must_use]
1479 pub fn quote_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<QuoteTick> {
1480 self.cache().quote_at_index(instrument_id, index).copied()
1481 }
1482
1483 #[must_use]
1489 pub fn trade(&self, instrument_id: &InstrumentId) -> Option<TradeTick> {
1490 self.cache().trade(instrument_id).copied()
1491 }
1492
1493 #[must_use]
1501 pub fn trade_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<TradeTick> {
1502 self.cache().trade_at_index(instrument_id, index).copied()
1503 }
1504
1505 #[must_use]
1511 pub fn mark_price(&self, instrument_id: &InstrumentId) -> Option<MarkPriceUpdate> {
1512 self.cache().mark_price(instrument_id).copied()
1513 }
1514
1515 #[must_use]
1521 pub fn index_price(&self, instrument_id: &InstrumentId) -> Option<IndexPriceUpdate> {
1522 self.cache().index_price(instrument_id).copied()
1523 }
1524
1525 #[must_use]
1531 pub fn funding_rate(&self, instrument_id: &InstrumentId) -> Option<FundingRateUpdate> {
1532 self.cache().funding_rate(instrument_id).copied()
1533 }
1534
1535 #[must_use]
1541 pub fn instrument_status(&self, instrument_id: &InstrumentId) -> Option<InstrumentStatus> {
1542 self.cache().instrument_status(instrument_id).copied()
1543 }
1544
1545 #[must_use]
1551 pub fn bar(&self, bar_type: &BarType) -> Option<Bar> {
1552 self.cache().bar(bar_type).copied()
1553 }
1554
1555 #[must_use]
1563 pub fn bar_at_index(&self, bar_type: &BarType, index: usize) -> Option<Bar> {
1564 self.cache().bar_at_index(bar_type, index).copied()
1565 }
1566
1567 #[must_use]
1573 pub fn book_update_count(&self, instrument_id: &InstrumentId) -> usize {
1574 self.cache().book_update_count(instrument_id)
1575 }
1576
1577 #[must_use]
1583 pub fn quote_count(&self, instrument_id: &InstrumentId) -> usize {
1584 self.cache().quote_count(instrument_id)
1585 }
1586
1587 #[must_use]
1593 pub fn trade_count(&self, instrument_id: &InstrumentId) -> usize {
1594 self.cache().trade_count(instrument_id)
1595 }
1596
1597 #[must_use]
1603 pub fn mark_price_count(&self, instrument_id: &InstrumentId) -> usize {
1604 self.cache().mark_price_count(instrument_id)
1605 }
1606
1607 #[must_use]
1613 pub fn index_price_count(&self, instrument_id: &InstrumentId) -> usize {
1614 self.cache().index_price_count(instrument_id)
1615 }
1616
1617 #[must_use]
1623 pub fn funding_rate_count(&self, instrument_id: &InstrumentId) -> usize {
1624 self.cache().funding_rate_count(instrument_id)
1625 }
1626
1627 #[must_use]
1633 pub fn instrument_status_count(&self, instrument_id: &InstrumentId) -> usize {
1634 self.cache().instrument_status_count(instrument_id)
1635 }
1636
1637 #[must_use]
1643 pub fn bar_count(&self, bar_type: &BarType) -> usize {
1644 self.cache().bar_count(bar_type)
1645 }
1646
1647 #[must_use]
1653 pub fn has_order_book(&self, instrument_id: &InstrumentId) -> bool {
1654 self.cache().has_order_book(instrument_id)
1655 }
1656
1657 #[must_use]
1663 pub fn has_quote_ticks(&self, instrument_id: &InstrumentId) -> bool {
1664 self.cache().has_quote_ticks(instrument_id)
1665 }
1666
1667 #[must_use]
1673 pub fn has_trade_ticks(&self, instrument_id: &InstrumentId) -> bool {
1674 self.cache().has_trade_ticks(instrument_id)
1675 }
1676
1677 #[must_use]
1683 pub fn has_mark_prices(&self, instrument_id: &InstrumentId) -> bool {
1684 self.cache().has_mark_prices(instrument_id)
1685 }
1686
1687 #[must_use]
1693 pub fn has_index_prices(&self, instrument_id: &InstrumentId) -> bool {
1694 self.cache().has_index_prices(instrument_id)
1695 }
1696
1697 #[must_use]
1703 pub fn has_funding_rates(&self, instrument_id: &InstrumentId) -> bool {
1704 self.cache().has_funding_rates(instrument_id)
1705 }
1706
1707 #[must_use]
1713 pub fn has_instrument_statuses(&self, instrument_id: &InstrumentId) -> bool {
1714 self.cache().has_instrument_statuses(instrument_id)
1715 }
1716
1717 #[must_use]
1723 pub fn has_bars(&self, bar_type: &BarType) -> bool {
1724 self.cache().has_bars(bar_type)
1725 }
1726
1727 #[must_use]
1733 pub fn get_xrate(
1734 &self,
1735 venue: Venue,
1736 from_currency: Currency,
1737 to_currency: Currency,
1738 price_type: PriceType,
1739 ) -> Option<Decimal> {
1740 self.cache()
1741 .get_xrate(venue, from_currency, to_currency, price_type)
1742 }
1743
1744 #[must_use]
1750 pub fn get_mark_xrate(&self, from_currency: Currency, to_currency: Currency) -> Option<f64> {
1751 self.cache().get_mark_xrate(from_currency, to_currency)
1752 }
1753
1754 #[must_use]
1760 pub fn yield_curve(&self, key: &str) -> Option<Box<dyn Fn(f64) -> f64>> {
1761 self.cache().yield_curve(key)
1762 }
1763
1764 #[must_use]
1770 pub fn greeks(&self, instrument_id: &InstrumentId) -> Option<GreeksData> {
1771 self.cache().greeks(instrument_id)
1772 }
1773
1774 #[must_use]
1780 pub fn option_greeks(&self, instrument_id: &InstrumentId) -> Option<OptionGreeks> {
1781 self.cache().option_greeks(instrument_id).copied()
1782 }
1783
1784 #[must_use]
1790 pub fn currency(&self, code: &Ustr) -> Option<Currency> {
1791 self.cache().currency(code).copied()
1792 }
1793
1794 pub fn try_currency(&self, code: &Ustr) -> Result<Currency, CurrencyLookupError> {
1805 self.cache().try_currency(code).copied()
1806 }
1807
1808 #[must_use]
1814 pub fn instrument(&self, instrument_id: &InstrumentId) -> Option<InstrumentAny> {
1815 self.cache().instrument(instrument_id).cloned()
1816 }
1817
1818 pub fn try_instrument(
1829 &self,
1830 instrument_id: &InstrumentId,
1831 ) -> Result<InstrumentAny, InstrumentLookupError> {
1832 self.cache().try_instrument(instrument_id).cloned()
1833 }
1834
1835 #[must_use]
1841 pub fn instrument_ids(&self, venue: Option<&Venue>) -> Vec<InstrumentId> {
1842 self.cache()
1843 .instrument_ids(venue)
1844 .into_iter()
1845 .copied()
1846 .collect()
1847 }
1848
1849 #[must_use]
1855 pub fn instruments(&self, venue: &Venue, underlying: Option<&Ustr>) -> Vec<InstrumentAny> {
1856 self.cache()
1857 .instruments(venue, underlying)
1858 .into_iter()
1859 .cloned()
1860 .collect()
1861 }
1862
1863 #[must_use]
1870 pub fn instruments_by_parent(
1871 &self,
1872 venue: &Venue,
1873 root: &Ustr,
1874 class: InstrumentClass,
1875 ) -> Vec<InstrumentAny> {
1876 self.cache()
1877 .instruments_by_parent(venue, root, class)
1878 .into_iter()
1879 .cloned()
1880 .collect()
1881 }
1882
1883 #[must_use]
1889 pub fn bar_types(
1890 &self,
1891 instrument_id: Option<&InstrumentId>,
1892 price_type: Option<&PriceType>,
1893 aggregation_source: AggregationSource,
1894 ) -> Vec<BarType> {
1895 self.cache()
1896 .bar_types(instrument_id, price_type, aggregation_source)
1897 .into_iter()
1898 .copied()
1899 .collect()
1900 }
1901
1902 #[must_use]
1908 pub fn synthetic(&self, instrument_id: &InstrumentId) -> Option<SyntheticInstrument> {
1909 self.cache().synthetic(instrument_id).cloned()
1910 }
1911
1912 pub fn try_synthetic(
1924 &self,
1925 instrument_id: &InstrumentId,
1926 ) -> Result<SyntheticInstrument, SyntheticInstrumentLookupError> {
1927 self.cache().try_synthetic(instrument_id).cloned()
1928 }
1929
1930 #[must_use]
1936 pub fn synthetic_ids(&self) -> Vec<InstrumentId> {
1937 self.cache().synthetic_ids().into_iter().copied().collect()
1938 }
1939
1940 #[must_use]
1946 pub fn synthetics(&self) -> Vec<SyntheticInstrument> {
1947 self.cache().synthetics().into_iter().cloned().collect()
1948 }
1949
1950 #[cfg(feature = "defi")]
1956 #[must_use]
1957 pub fn pool(&self, instrument_id: &InstrumentId) -> Option<Pool> {
1958 self.cache().pool(instrument_id).cloned()
1959 }
1960
1961 #[cfg(feature = "defi")]
1967 #[must_use]
1968 pub fn pool_ids(&self, venue: Option<&Venue>) -> Vec<InstrumentId> {
1969 self.cache().pool_ids(venue)
1970 }
1971
1972 #[cfg(feature = "defi")]
1978 #[must_use]
1979 pub fn pools(&self, venue: Option<&Venue>) -> Vec<Pool> {
1980 self.cache().pools(venue).into_iter().cloned().collect()
1981 }
1982
1983 #[cfg(feature = "defi")]
1989 #[must_use]
1990 pub fn pool_profiler(&self, instrument_id: &InstrumentId) -> Option<PoolProfiler> {
1991 self.cache().pool_profiler(instrument_id).cloned()
1992 }
1993
1994 #[cfg(feature = "defi")]
2000 #[must_use]
2001 pub fn pool_profiler_ids(&self, venue: Option<&Venue>) -> Vec<InstrumentId> {
2002 self.cache().pool_profiler_ids(venue)
2003 }
2004
2005 #[cfg(feature = "defi")]
2011 #[must_use]
2012 pub fn pool_profilers(&self, venue: Option<&Venue>) -> Vec<PoolProfiler> {
2013 self.cache()
2014 .pool_profilers(venue)
2015 .into_iter()
2016 .cloned()
2017 .collect()
2018 }
2019
2020 #[must_use]
2026 pub fn account(&self, account_id: &AccountId) -> Option<AccountAny> {
2027 self.cache().account_owned(account_id)
2028 }
2029
2030 pub fn try_account(&self, account_id: &AccountId) -> Result<AccountAny, AccountLookupError> {
2041 self.cache()
2042 .try_account(account_id)
2043 .map(|account| account.cloned())
2044 }
2045
2046 #[must_use]
2052 pub fn account_for_venue(&self, venue: &Venue) -> Option<AccountAny> {
2053 self.cache().account_for_venue_owned(venue)
2054 }
2055
2056 #[must_use]
2062 pub fn account_id(&self, venue: &Venue) -> Option<AccountId> {
2063 self.cache().account_id(venue).copied()
2064 }
2065
2066 #[must_use]
2072 pub fn accounts(&self, account_id: &AccountId) -> Vec<AccountAny> {
2073 self.cache()
2074 .accounts(account_id)
2075 .into_iter()
2076 .map(|account| account.cloned())
2077 .collect()
2078 }
2079
2080 #[must_use]
2086 pub fn accounts_all(&self) -> Vec<AccountAny> {
2087 self.cache().accounts_all_owned()
2088 }
2089
2090 fn cache(&self) -> Ref<'_, Cache> {
2091 self.cache.borrow()
2092 }
2093}
2094
2095enum FilterSources<'a, K> {
2102 Unfiltered,
2103 Empty,
2104 Sets(Vec<&'a AHashSet<K>>),
2105}
2106
2107fn intersect_filter_sources<K>(mut sources: Vec<&AHashSet<K>>) -> AHashSet<K>
2113where
2114 K: Copy + Eq + std::hash::Hash,
2115{
2116 debug_assert!(!sources.is_empty());
2117 sources.sort_unstable_by_key(|s| s.len());
2118 let driver = sources[0];
2119 let rest = &sources[1..];
2120
2121 if rest.is_empty() {
2122 return driver.clone();
2123 }
2124
2125 driver
2126 .iter()
2127 .filter(|id| rest.iter().all(|s| s.contains(id)))
2128 .copied()
2129 .collect()
2130}
2131
2132fn intersect_pair_or_many<'a, K>(
2140 bucket: &'a AHashSet<K>,
2141 mut sources: Vec<&'a AHashSet<K>>,
2142) -> AHashSet<K>
2143where
2144 K: Copy + Eq + std::hash::Hash,
2145{
2146 debug_assert!(!sources.is_empty());
2147 if sources.len() == 1 {
2148 let filter = sources[0];
2149 let (larger, smaller) = if bucket.len() >= filter.len() {
2150 (bucket, filter)
2151 } else {
2152 (filter, bucket)
2153 };
2154 return larger.intersection(smaller).copied().collect();
2155 }
2156
2157 sources.push(bucket);
2158 intersect_filter_sources(sources)
2159}
2160
2161#[cfg_attr(
2163 feature = "python",
2164 pyo3::pyclass(module = "nautilus_trader.common", unsendable)
2165)]
2166pub struct Cache {
2167 config: CacheConfig,
2168 index: CacheIndex,
2169 database: Option<Box<dyn CacheDatabaseAdapter>>,
2170 general: AHashMap<String, Bytes>,
2171 currencies: AHashMap<Ustr, Currency>,
2172 instruments: AHashMap<InstrumentId, InstrumentAny>,
2173 synthetics: AHashMap<InstrumentId, SyntheticInstrument>,
2174 books: AHashMap<InstrumentId, OrderBook>,
2175 own_books: AHashMap<InstrumentId, OwnOrderBook>,
2176 quotes: AHashMap<InstrumentId, BoundedVecDeque<QuoteTick>>,
2177 trades: AHashMap<InstrumentId, BoundedVecDeque<TradeTick>>,
2178 mark_xrates: AHashMap<(Currency, Currency), f64>,
2179 mark_prices: AHashMap<InstrumentId, BoundedVecDeque<MarkPriceUpdate>>,
2180 index_prices: AHashMap<InstrumentId, BoundedVecDeque<IndexPriceUpdate>>,
2181 funding_rates: AHashMap<InstrumentId, BoundedVecDeque<FundingRateUpdate>>,
2182 instrument_statuses: AHashMap<InstrumentId, BoundedVecDeque<InstrumentStatus>>,
2183 bars: AHashMap<BarType, BoundedVecDeque<Bar>>,
2184 greeks: AHashMap<InstrumentId, GreeksData>,
2185 option_greeks: AHashMap<InstrumentId, OptionGreeks>,
2186 yield_curves: AHashMap<String, YieldCurveData>,
2187 accounts: AHashMap<AccountId, SharedCell<AccountAny>>,
2188 orders: AHashMap<ClientOrderId, SharedCell<OrderAny>>,
2189 order_lists: AHashMap<OrderListId, OrderList>,
2190 positions: AHashMap<PositionId, SharedCell<Position>>,
2191 position_snapshots: AHashMap<PositionId, Vec<PositionSnapshotFrame>>,
2192 position_snapshot_revisions: AHashMap<PositionId, u64>,
2193 #[cfg(feature = "defi")]
2194 pub(crate) defi: crate::defi::cache::DefiCache,
2195}
2196
2197impl Debug for Cache {
2198 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2199 f.debug_struct(stringify!(Cache))
2200 .field("config", &self.config)
2201 .field("index", &self.index)
2202 .field("general", &self.general)
2203 .field("currencies", &self.currencies)
2204 .field("instruments", &self.instruments)
2205 .field("synthetics", &self.synthetics)
2206 .field("books", &self.books)
2207 .field("own_books", &self.own_books)
2208 .field("quotes", &self.quotes)
2209 .field("trades", &self.trades)
2210 .field("mark_xrates", &self.mark_xrates)
2211 .field("mark_prices", &self.mark_prices)
2212 .field("index_prices", &self.index_prices)
2213 .field("funding_rates", &self.funding_rates)
2214 .field("instrument_statuses", &self.instrument_statuses)
2215 .field("bars", &self.bars)
2216 .field("greeks", &self.greeks)
2217 .field("option_greeks", &self.option_greeks)
2218 .field("yield_curves", &self.yield_curves)
2219 .field("accounts", &self.accounts)
2220 .field("orders", &self.orders)
2221 .field("order_lists", &self.order_lists)
2222 .field("positions", &self.positions)
2223 .field("position_snapshots", &self.position_snapshots)
2224 .finish()
2225 }
2226}
2227
2228impl Default for Cache {
2229 fn default() -> Self {
2231 Self::new(Some(CacheConfig::default()), None)
2232 }
2233}
2234
2235impl Cache {
2236 #[must_use]
2238 pub fn new(
2246 config: Option<CacheConfig>,
2247 database: Option<Box<dyn CacheDatabaseAdapter>>,
2248 ) -> Self {
2249 let config = config.unwrap_or_default();
2250 config.validate().expect("invalid `CacheConfig`");
2251
2252 Self {
2253 config,
2254 index: CacheIndex::default(),
2255 database,
2256 general: AHashMap::new(),
2257 currencies: AHashMap::new(),
2258 instruments: AHashMap::new(),
2259 synthetics: AHashMap::new(),
2260 books: AHashMap::new(),
2261 own_books: AHashMap::new(),
2262 quotes: AHashMap::new(),
2263 trades: AHashMap::new(),
2264 mark_xrates: AHashMap::new(),
2265 mark_prices: AHashMap::new(),
2266 index_prices: AHashMap::new(),
2267 funding_rates: AHashMap::new(),
2268 instrument_statuses: AHashMap::new(),
2269 bars: AHashMap::new(),
2270 greeks: AHashMap::new(),
2271 option_greeks: AHashMap::new(),
2272 yield_curves: AHashMap::new(),
2273 accounts: AHashMap::new(),
2274 orders: AHashMap::new(),
2275 order_lists: AHashMap::new(),
2276 positions: AHashMap::new(),
2277 position_snapshots: AHashMap::new(),
2278 position_snapshot_revisions: AHashMap::new(),
2279 #[cfg(feature = "defi")]
2280 defi: crate::defi::cache::DefiCache::default(),
2281 }
2282 }
2283
2284 #[must_use]
2286 pub fn memory_address(&self) -> String {
2287 format!("{:?}", std::ptr::from_ref(self))
2288 }
2289
2290 pub fn set_database(&mut self, database: Box<dyn CacheDatabaseAdapter>) {
2294 let type_name = std::any::type_name_of_val(&*database);
2295 log::info!("Cache database adapter set: {type_name}");
2296 self.database = Some(database);
2297 }
2298
2299 pub fn cache_general(&mut self) -> anyhow::Result<()> {
2307 self.general = match &mut self.database {
2308 Some(db) => db.load()?,
2309 None => AHashMap::new(),
2310 };
2311
2312 log::info!(
2313 "Cached {} general object(s) from database",
2314 self.general.len()
2315 );
2316 Ok(())
2317 }
2318
2319 pub async fn cache_all(&mut self) -> anyhow::Result<()> {
2325 let cache_map = match &self.database {
2326 Some(db) => db.load_all().await?,
2327 None => CacheMap::default(),
2328 };
2329
2330 self.currencies = cache_map.currencies;
2331 self.instruments = cache_map.instruments;
2332 self.synthetics = cache_map.synthetics;
2333 self.accounts = cache_map
2334 .accounts
2335 .into_iter()
2336 .map(|(id, account)| (id, SharedCell::new(account)))
2337 .collect();
2338 self.orders = cache_map
2339 .orders
2340 .into_iter()
2341 .map(|(id, order)| (id, SharedCell::new(order)))
2342 .collect();
2343 self.positions = cache_map
2344 .positions
2345 .into_iter()
2346 .map(|(id, position)| (id, SharedCell::new(position)))
2347 .collect();
2348
2349 if let Some(db) = &self.database {
2350 let order_position = db.load_index_order_position()?;
2351 self.index.order_position = self.sanitize_order_position_index(order_position);
2352 self.index.order_client = db.load_index_order_client()?;
2353 }
2354
2355 self.cache_position_oms()?;
2356 self.assign_position_ids_to_contingencies();
2357 Ok(())
2358 }
2359
2360 pub async fn cache_currencies(&mut self) -> anyhow::Result<()> {
2366 self.currencies = match &mut self.database {
2367 Some(db) => db.load_currencies().await?,
2368 None => AHashMap::new(),
2369 };
2370
2371 log::info!("Cached {} currencies from database", self.general.len());
2372 Ok(())
2373 }
2374
2375 pub async fn cache_instruments(&mut self) -> anyhow::Result<()> {
2381 self.instruments = match &mut self.database {
2382 Some(db) => db.load_instruments().await?,
2383 None => AHashMap::new(),
2384 };
2385
2386 log::info!("Cached {} instruments from database", self.general.len());
2387 Ok(())
2388 }
2389
2390 pub async fn cache_synthetics(&mut self) -> anyhow::Result<()> {
2396 self.synthetics = match &mut self.database {
2397 Some(db) => db.load_synthetics().await?,
2398 None => AHashMap::new(),
2399 };
2400
2401 log::info!(
2402 "Cached {} synthetic instruments from database",
2403 self.general.len()
2404 );
2405 Ok(())
2406 }
2407
2408 pub async fn cache_accounts(&mut self) -> anyhow::Result<()> {
2414 self.accounts = match &mut self.database {
2415 Some(db) => db
2416 .load_accounts()
2417 .await?
2418 .into_iter()
2419 .map(|(id, account)| (id, SharedCell::new(account)))
2420 .collect(),
2421 None => AHashMap::new(),
2422 };
2423
2424 log::info!(
2425 "Cached {} synthetic instruments from database",
2426 self.general.len()
2427 );
2428 Ok(())
2429 }
2430
2431 pub async fn cache_orders(&mut self) -> anyhow::Result<()> {
2437 self.orders = match &mut self.database {
2438 Some(db) => db
2439 .load_orders()
2440 .await?
2441 .into_iter()
2442 .map(|(id, order)| (id, SharedCell::new(order)))
2443 .collect(),
2444 None => AHashMap::new(),
2445 };
2446
2447 if let Some(db) = &self.database {
2448 let order_position = db.load_index_order_position()?;
2449 self.index.order_position = self.sanitize_order_position_index(order_position);
2450 self.index.order_client = db.load_index_order_client()?;
2451 }
2452
2453 log::info!("Cached {} orders from database", self.general.len());
2454
2455 self.assign_position_ids_to_contingencies();
2456 Ok(())
2457 }
2458
2459 fn sanitize_order_position_index(
2460 &self,
2461 mut order_position: AHashMap<ClientOrderId, PositionId>,
2462 ) -> AHashMap<ClientOrderId, PositionId> {
2463 let original_len = order_position.len();
2464 order_position.retain(|client_order_id, _| self.orders.contains_key(client_order_id));
2465 let removed = original_len - order_position.len();
2466
2467 if removed > 0 {
2468 log::warn!(
2469 "Filtered {removed} stale order-position index entries without backing orders during cache load"
2470 );
2471 }
2472
2473 order_position
2474 }
2475
2476 pub async fn cache_positions(&mut self) -> anyhow::Result<()> {
2482 self.positions = match &mut self.database {
2483 Some(db) => db
2484 .load_positions()
2485 .await?
2486 .into_iter()
2487 .map(|(id, position)| (id, SharedCell::new(position)))
2488 .collect(),
2489 None => AHashMap::new(),
2490 };
2491
2492 self.cache_position_oms()?;
2493 log::info!("Cached {} positions from database", self.general.len());
2494 Ok(())
2495 }
2496
2497 fn cache_position_oms(&mut self) -> anyhow::Result<()> {
2498 let persisted = match &self.database {
2499 Some(database) => database.load()?,
2500 None => self.general.clone(),
2501 };
2502
2503 self.general
2504 .retain(|key, _| !key.starts_with(POSITION_OMS_KEY_PREFIX));
2505
2506 for (key, value) in persisted {
2507 if !key.starts_with(POSITION_OMS_KEY_PREFIX) {
2508 continue;
2509 }
2510 self.general.insert(key, value);
2511 }
2512
2513 self.index_position_oms();
2514 Ok(())
2515 }
2516
2517 pub fn build_index(&mut self) {
2519 log::debug!("Building index");
2520
2521 for account_id in self.accounts.keys() {
2523 self.index
2524 .venue_account
2525 .insert(account_id.get_issuer(), *account_id);
2526 }
2527
2528 for (client_order_id, order_cell) in &self.orders {
2530 let order = order_cell.borrow();
2531 let instrument_id = order.instrument_id();
2532 let venue = instrument_id.venue;
2533 let strategy_id = order.strategy_id();
2534
2535 self.index
2537 .venue_orders
2538 .entry(venue)
2539 .or_default()
2540 .insert(*client_order_id);
2541
2542 if let Some(venue_order_id) = order.venue_order_id() {
2545 self.index
2546 .venue_order_ids
2547 .insert(venue_order_id, *client_order_id);
2548 self.index
2549 .client_order_ids
2550 .insert(*client_order_id, venue_order_id);
2551 }
2552
2553 if let Some(position_id) = order.position_id() {
2555 self.index
2556 .order_position
2557 .insert(*client_order_id, position_id);
2558 }
2559
2560 self.index
2562 .order_strategy
2563 .insert(*client_order_id, strategy_id);
2564
2565 self.index
2567 .instrument_orders
2568 .entry(instrument_id)
2569 .or_default()
2570 .insert(*client_order_id);
2571
2572 self.index
2574 .strategy_orders
2575 .entry(strategy_id)
2576 .or_default()
2577 .insert(*client_order_id);
2578
2579 if let Some(account_id) = order.account_id() {
2581 self.index
2582 .account_orders
2583 .entry(account_id)
2584 .or_default()
2585 .insert(*client_order_id);
2586 }
2587
2588 if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
2590 self.index
2591 .exec_algorithm_orders
2592 .entry(exec_algorithm_id)
2593 .or_default()
2594 .insert(*client_order_id);
2595 self.index.exec_algorithms.insert(exec_algorithm_id);
2596 }
2597
2598 if let Some(exec_spawn_id) = order.exec_spawn_id() {
2600 self.index
2601 .exec_spawn_orders
2602 .entry(exec_spawn_id)
2603 .or_default()
2604 .insert(*client_order_id);
2605 }
2606
2607 self.index.orders.insert(*client_order_id);
2609
2610 if order.is_active_local() {
2612 self.index.orders_active_local.insert(*client_order_id);
2613 }
2614
2615 if order.is_open() {
2617 self.index.orders_open.insert(*client_order_id);
2618 }
2619
2620 if order.is_closed() {
2622 self.index.orders_closed.insert(*client_order_id);
2623 }
2624
2625 if order.emulation_trigger().is_some() && !order.is_closed() {
2627 self.index.orders_emulated.insert(*client_order_id);
2628 }
2629
2630 if order.is_inflight() {
2632 self.index.orders_inflight.insert(*client_order_id);
2633 }
2634
2635 self.index.strategies.insert(strategy_id);
2637 }
2638
2639 for (position_id, position_cell) in &self.positions {
2641 let position = position_cell.borrow();
2642 let instrument_id = position.instrument_id;
2643 let venue = instrument_id.venue;
2644 let strategy_id = position.strategy_id;
2645
2646 self.index
2648 .venue_positions
2649 .entry(venue)
2650 .or_default()
2651 .insert(*position_id);
2652
2653 self.index
2655 .position_strategy
2656 .insert(*position_id, strategy_id);
2657
2658 let position_orders = self.index.position_orders.entry(*position_id).or_default();
2660 position_orders.extend(
2661 position
2662 .client_order_ids()
2663 .into_iter()
2664 .filter(|client_order_id| self.orders.contains_key(client_order_id)),
2665 );
2666
2667 self.index
2669 .instrument_positions
2670 .entry(instrument_id)
2671 .or_default()
2672 .insert(*position_id);
2673 self.index
2674 .instrument_orders
2675 .entry(instrument_id)
2676 .or_default();
2677
2678 self.index
2680 .strategy_positions
2681 .entry(strategy_id)
2682 .or_default()
2683 .insert(*position_id);
2684 self.index.strategy_orders.entry(strategy_id).or_default();
2685
2686 self.index
2688 .account_positions
2689 .entry(position.account_id)
2690 .or_default()
2691 .insert(*position_id);
2692
2693 self.index.positions.insert(*position_id);
2695
2696 if position.is_open() {
2698 self.index.positions_open.insert(*position_id);
2699 }
2700
2701 if position.is_closed() {
2703 self.index.positions_closed.insert(*position_id);
2704 }
2705
2706 self.index.strategies.insert(strategy_id);
2708 }
2709
2710 self.index_position_oms();
2711 }
2712
2713 fn index_position_oms(&mut self) {
2714 self.index.position_oms.clear();
2715
2716 for (key, value) in &self.general {
2717 let Some(position_id) = key.strip_prefix(POSITION_OMS_KEY_PREFIX) else {
2718 continue;
2719 };
2720 let position_id = PositionId::new(position_id);
2721 if !self.positions.contains_key(&position_id) {
2722 continue;
2723 }
2724
2725 match serde_json::from_slice::<OmsType>(value) {
2726 Ok(oms_type) => {
2727 self.index.position_oms.insert(position_id, oms_type);
2728 }
2729 Err(e) => {
2730 log::error!("Failed to decode position OMS for {position_id}: {e}");
2731 }
2732 }
2733 }
2734
2735 for position in self.positions.values().map(|cell| cell.borrow()) {
2736 if !self.index.position_oms.contains_key(&position.id)
2737 && position.id.as_str()
2738 == format!("{}-{}", position.instrument_id, position.strategy_id)
2739 {
2740 self.index
2741 .position_oms
2742 .insert(position.id, OmsType::Netting);
2743 }
2744 }
2745 }
2746
2747 #[must_use]
2749 pub const fn has_backing(&self) -> bool {
2750 self.database.is_some()
2751 }
2752
2753 pub fn load_actor_state(
2761 &self,
2762 actor_id: &ActorId,
2763 ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
2764 self.database
2765 .as_ref()
2766 .map(|database| database.load_actor(actor_id))
2767 .transpose()
2768 .map(|state| state.map(Self::decode_component_state))
2769 }
2770
2771 pub fn load_strategy_state(
2779 &self,
2780 strategy_id: &StrategyId,
2781 ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
2782 self.database
2783 .as_ref()
2784 .map(|database| database.load_strategy(strategy_id))
2785 .transpose()
2786 .map(|state| state.map(Self::decode_component_state))
2787 }
2788
2789 pub fn update_actor_state(
2795 &self,
2796 actor_id: &ActorId,
2797 state: &IndexMap<String, Vec<u8>>,
2798 ) -> anyhow::Result<()> {
2799 if let Some(database) = &self.database {
2800 database.update_actor(actor_id, &Self::encode_component_state(state))?;
2801 }
2802 Ok(())
2803 }
2804
2805 pub fn update_strategy_state(
2811 &self,
2812 strategy_id: &StrategyId,
2813 state: &IndexMap<String, Vec<u8>>,
2814 ) -> anyhow::Result<()> {
2815 if let Some(database) = &self.database {
2816 database.update_strategy(strategy_id, &Self::encode_component_state(state))?;
2817 }
2818 Ok(())
2819 }
2820
2821 fn decode_component_state(state: AHashMap<String, Bytes>) -> IndexMap<String, Vec<u8>> {
2822 state
2823 .into_iter()
2824 .map(|(key, value)| (key, value.to_vec()))
2825 .collect()
2826 }
2827
2828 fn encode_component_state(state: &IndexMap<String, Vec<u8>>) -> AHashMap<String, Bytes> {
2829 state
2830 .iter()
2831 .map(|(key, value)| (key.clone(), Bytes::copy_from_slice(value)))
2832 .collect()
2833 }
2834
2835 #[must_use]
2837 pub fn calculate_unrealized_pnl(&self, position: &Position) -> Option<Money> {
2838 let Some(quote) = self.quote(&position.instrument_id) else {
2839 log::warn!(
2840 "Cannot calculate unrealized PnL for {}, no quotes for {}",
2841 position.id,
2842 position.instrument_id
2843 );
2844 return None;
2845 };
2846
2847 let last = match position.side {
2849 PositionSide::Flat => {
2850 return Some(Money::zero(position.settlement_currency));
2851 }
2852 PositionSide::Long => quote.bid_price,
2853 PositionSide::Short => quote.ask_price,
2854 };
2855
2856 position
2857 .try_unrealized_pnl(last)
2858 .inspect_err(|e| {
2859 log::error!("Cannot calculate unrealized PnL for {}: {e}", position.id);
2860 })
2861 .ok()
2862 }
2863
2864 #[must_use]
2873 pub fn check_integrity(&mut self) -> bool {
2874 let mut error_count = 0;
2875 let failure = "Integrity failure";
2876
2877 let timestamp_us = SystemTime::now()
2879 .duration_since(UNIX_EPOCH)
2880 .expect("Time went backwards")
2881 .as_micros();
2882
2883 log::info!("Checking data integrity");
2884
2885 for account_id in self.accounts.keys() {
2887 if !self
2888 .index
2889 .venue_account
2890 .contains_key(&account_id.get_issuer())
2891 {
2892 log::error!(
2893 "{failure} in accounts: {account_id} not found in `self.index.venue_account`",
2894 );
2895 error_count += 1;
2896 }
2897 }
2898
2899 for (client_order_id, order_cell) in &self.orders {
2900 let order = order_cell.borrow();
2901
2902 if !self.index.order_strategy.contains_key(client_order_id) {
2903 log::error!(
2904 "{failure} in orders: {client_order_id} not found in `self.index.order_strategy`"
2905 );
2906 error_count += 1;
2907 }
2908
2909 if !self.index.orders.contains(client_order_id) {
2910 log::error!(
2911 "{failure} in orders: {client_order_id} not found in `self.index.orders`",
2912 );
2913 error_count += 1;
2914 }
2915
2916 if order.is_inflight() && !self.index.orders_inflight.contains(client_order_id) {
2917 log::error!(
2918 "{failure} in orders: {client_order_id} not found in `self.index.orders_inflight`",
2919 );
2920 error_count += 1;
2921 }
2922
2923 if order.is_active_local() && !self.index.orders_active_local.contains(client_order_id)
2924 {
2925 log::error!(
2926 "{failure} in orders: {client_order_id} not found in `self.index.orders_active_local`",
2927 );
2928 error_count += 1;
2929 }
2930
2931 if order.is_open() && !self.index.orders_open.contains(client_order_id) {
2932 log::error!(
2933 "{failure} in orders: {client_order_id} not found in `self.index.orders_open`",
2934 );
2935 error_count += 1;
2936 }
2937
2938 if order.is_closed() && !self.index.orders_closed.contains(client_order_id) {
2939 log::error!(
2940 "{failure} in orders: {client_order_id} not found in `self.index.orders_closed`",
2941 );
2942 error_count += 1;
2943 }
2944
2945 if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
2946 if !self
2947 .index
2948 .exec_algorithm_orders
2949 .contains_key(&exec_algorithm_id)
2950 {
2951 log::error!(
2952 "{failure} in orders: {client_order_id} not found in `self.index.exec_algorithm_orders`",
2953 );
2954 error_count += 1;
2955 }
2956
2957 if order.exec_spawn_id().is_none()
2958 && !self.index.exec_spawn_orders.contains_key(client_order_id)
2959 {
2960 log::error!(
2961 "{failure} in orders: {client_order_id} not found in `self.index.exec_spawn_orders`",
2962 );
2963 error_count += 1;
2964 }
2965 }
2966 }
2967
2968 for (position_id, position_cell) in &self.positions {
2969 let position = position_cell.borrow();
2970
2971 if !self.index.position_strategy.contains_key(position_id) {
2972 log::error!(
2973 "{failure} in positions: {position_id} not found in `self.index.position_strategy`",
2974 );
2975 error_count += 1;
2976 }
2977
2978 if !self.index.position_orders.contains_key(position_id) {
2979 log::error!(
2980 "{failure} in positions: {position_id} not found in `self.index.position_orders`",
2981 );
2982 error_count += 1;
2983 }
2984
2985 if !self.index.positions.contains(position_id) {
2986 log::error!(
2987 "{failure} in positions: {position_id} not found in `self.index.positions`",
2988 );
2989 error_count += 1;
2990 }
2991
2992 if position.is_open() && !self.index.positions_open.contains(position_id) {
2993 log::error!(
2994 "{failure} in positions: {position_id} not found in `self.index.positions_open`",
2995 );
2996 error_count += 1;
2997 }
2998
2999 if position.is_closed() && !self.index.positions_closed.contains(position_id) {
3000 log::error!(
3001 "{failure} in positions: {position_id} not found in `self.index.positions_closed`",
3002 );
3003 error_count += 1;
3004 }
3005 }
3006
3007 for account_id in self.index.venue_account.values() {
3009 if !self.accounts.contains_key(account_id) {
3010 log::error!(
3011 "{failure} in `index.venue_account`: {account_id} not found in `self.accounts`",
3012 );
3013 error_count += 1;
3014 }
3015 }
3016
3017 for client_order_id in self.index.venue_order_ids.values() {
3018 if !self.orders.contains_key(client_order_id) {
3019 log::error!(
3020 "{failure} in `index.venue_order_ids`: {client_order_id} not found in `self.orders`",
3021 );
3022 error_count += 1;
3023 }
3024 }
3025
3026 for client_order_id in self.index.client_order_ids.keys() {
3027 if !self.orders.contains_key(client_order_id) {
3028 log::error!(
3029 "{failure} in `index.client_order_ids`: {client_order_id} not found in `self.orders`",
3030 );
3031 error_count += 1;
3032 }
3033 }
3034
3035 for client_order_id in self.index.order_position.keys() {
3036 if !self.orders.contains_key(client_order_id) {
3037 log::error!(
3038 "{failure} in `index.order_position`: {client_order_id} not found in `self.orders`",
3039 );
3040 error_count += 1;
3041 }
3042 }
3043
3044 for client_order_id in self.index.order_strategy.keys() {
3046 if !self.orders.contains_key(client_order_id) {
3047 log::error!(
3048 "{failure} in `index.order_strategy`: {client_order_id} not found in `self.orders`",
3049 );
3050 error_count += 1;
3051 }
3052 }
3053
3054 for position_id in self.index.position_strategy.keys() {
3055 if !self.positions.contains_key(position_id) {
3056 log::error!(
3057 "{failure} in `index.position_strategy`: {position_id} not found in `self.positions`",
3058 );
3059 error_count += 1;
3060 }
3061 }
3062
3063 for position_id in self.index.position_orders.keys() {
3064 if !self.positions.contains_key(position_id) {
3065 log::error!(
3066 "{failure} in `index.position_orders`: {position_id} not found in `self.positions`",
3067 );
3068 error_count += 1;
3069 }
3070 }
3071
3072 for (instrument_id, client_order_ids) in &self.index.instrument_orders {
3073 for client_order_id in client_order_ids {
3074 if !self.orders.contains_key(client_order_id) {
3075 log::error!(
3076 "{failure} in `index.instrument_orders`: {instrument_id} not found in `self.orders`",
3077 );
3078 error_count += 1;
3079 }
3080 }
3081 }
3082
3083 for instrument_id in self.index.instrument_positions.keys() {
3084 if !self.index.instrument_orders.contains_key(instrument_id) {
3085 log::error!(
3086 "{failure} in `index.instrument_positions`: {instrument_id} not found in `index.instrument_orders`",
3087 );
3088 error_count += 1;
3089 }
3090 }
3091
3092 for client_order_ids in self.index.strategy_orders.values() {
3093 for client_order_id in client_order_ids {
3094 if !self.orders.contains_key(client_order_id) {
3095 log::error!(
3096 "{failure} in `index.strategy_orders`: {client_order_id} not found in `self.orders`",
3097 );
3098 error_count += 1;
3099 }
3100 }
3101 }
3102
3103 for position_ids in self.index.strategy_positions.values() {
3104 for position_id in position_ids {
3105 if !self.positions.contains_key(position_id) {
3106 log::error!(
3107 "{failure} in `index.strategy_positions`: {position_id} not found in `self.positions`",
3108 );
3109 error_count += 1;
3110 }
3111 }
3112 }
3113
3114 for client_order_id in &self.index.orders {
3115 if !self.orders.contains_key(client_order_id) {
3116 log::error!(
3117 "{failure} in `index.orders`: {client_order_id} not found in `self.orders`",
3118 );
3119 error_count += 1;
3120 }
3121 }
3122
3123 for client_order_id in &self.index.orders_emulated {
3124 if !self.orders.contains_key(client_order_id) {
3125 log::error!(
3126 "{failure} in `index.orders_emulated`: {client_order_id} not found in `self.orders`",
3127 );
3128 error_count += 1;
3129 }
3130 }
3131
3132 for client_order_id in &self.index.orders_active_local {
3133 if !self.orders.contains_key(client_order_id) {
3134 log::error!(
3135 "{failure} in `index.orders_active_local`: {client_order_id} not found in `self.orders`",
3136 );
3137 error_count += 1;
3138 }
3139 }
3140
3141 for client_order_id in &self.index.orders_inflight {
3142 if !self.orders.contains_key(client_order_id) {
3143 log::error!(
3144 "{failure} in `index.orders_inflight`: {client_order_id} not found in `self.orders`",
3145 );
3146 error_count += 1;
3147 }
3148 }
3149
3150 for client_order_id in &self.index.orders_open {
3151 if !self.orders.contains_key(client_order_id) {
3152 log::error!(
3153 "{failure} in `index.orders_open`: {client_order_id} not found in `self.orders`",
3154 );
3155 error_count += 1;
3156 }
3157 }
3158
3159 for client_order_id in &self.index.orders_closed {
3160 if !self.orders.contains_key(client_order_id) {
3161 log::error!(
3162 "{failure} in `index.orders_closed`: {client_order_id} not found in `self.orders`",
3163 );
3164 error_count += 1;
3165 }
3166 }
3167
3168 for position_id in &self.index.positions {
3169 if !self.positions.contains_key(position_id) {
3170 log::error!(
3171 "{failure} in `index.positions`: {position_id} not found in `self.positions`",
3172 );
3173 error_count += 1;
3174 }
3175 }
3176
3177 for position_id in &self.index.positions_open {
3178 if !self.positions.contains_key(position_id) {
3179 log::error!(
3180 "{failure} in `index.positions_open`: {position_id} not found in `self.positions`",
3181 );
3182 error_count += 1;
3183 }
3184 }
3185
3186 for position_id in &self.index.positions_closed {
3187 if !self.positions.contains_key(position_id) {
3188 log::error!(
3189 "{failure} in `index.positions_closed`: {position_id} not found in `self.positions`",
3190 );
3191 error_count += 1;
3192 }
3193 }
3194
3195 for strategy_id in &self.index.strategies {
3196 if !self.index.strategy_orders.contains_key(strategy_id) {
3197 log::error!(
3198 "{failure} in `index.strategies`: {strategy_id} not found in `index.strategy_orders`",
3199 );
3200 error_count += 1;
3201 }
3202 }
3203
3204 for exec_algorithm_id in &self.index.exec_algorithms {
3205 if !self
3206 .index
3207 .exec_algorithm_orders
3208 .contains_key(exec_algorithm_id)
3209 {
3210 log::error!(
3211 "{failure} in `index.exec_algorithms`: {exec_algorithm_id} not found in `index.exec_algorithm_orders`",
3212 );
3213 error_count += 1;
3214 }
3215 }
3216
3217 let total_us = SystemTime::now()
3218 .duration_since(UNIX_EPOCH)
3219 .expect("Time went backwards")
3220 .as_micros()
3221 - timestamp_us;
3222
3223 if error_count == 0 {
3224 log::info!("Integrity check passed in {total_us}μs");
3225 true
3226 } else {
3227 log::error!(
3228 "Integrity check failed with {error_count} error{} in {total_us}μs",
3229 if error_count == 1 { "" } else { "s" },
3230 );
3231 false
3232 }
3233 }
3234
3235 #[must_use]
3239 pub fn check_residuals(&self) -> bool {
3240 log::debug!("Checking residuals");
3241
3242 let mut residuals = false;
3243
3244 for order in self.orders_open(None, None, None, None, None) {
3246 residuals = true;
3247 log::warn!("Residual {order}");
3248 }
3249
3250 for position in self.positions_open(None, None, None, None, None) {
3252 residuals = true;
3253 log::warn!("Residual {position}");
3254 }
3255
3256 residuals
3257 }
3258
3259 pub fn purge_closed_orders(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
3265 log::debug!(
3266 "Purging closed orders{}",
3267 if buffer_secs > 0 {
3268 format!(" with buffer_secs={buffer_secs}")
3269 } else {
3270 String::new()
3271 }
3272 );
3273
3274 let Ok(buffer_ns) = secs_to_nanos(buffer_secs as f64) else {
3275 log::warn!(
3276 "Cannot purge closed orders: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
3277 );
3278 return;
3279 };
3280 let purge_cutoff = ts_now.checked_sub(buffer_ns);
3281
3282 let mut affected_order_list_ids: AHashSet<OrderListId> = AHashSet::new();
3283 let mut purged_client_order_ids: AHashSet<ClientOrderId> = AHashSet::new();
3284
3285 'outer: for client_order_id in self.index.orders_closed.clone() {
3286 let purge_target = self.orders.get(&client_order_id).and_then(|order_cell| {
3287 let order = order_cell.borrow();
3288 if order.is_closed()
3289 && let Some(ts_closed) = order.ts_closed()
3290 && purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
3291 {
3292 let linked = order.linked_order_ids().map(<[_]>::to_vec);
3293 let order_list_id = order.order_list_id();
3294 Some((linked, order_list_id))
3295 } else {
3296 None
3297 }
3298 });
3299
3300 let Some((linked, order_list_id)) = purge_target else {
3301 continue;
3302 };
3303
3304 if let Some(linked_order_ids) = linked {
3306 for linked_order_id in &linked_order_ids {
3307 if let Some(linked_order_cell) = self.orders.get(linked_order_id)
3308 && linked_order_cell.borrow().is_open()
3309 {
3310 continue 'outer;
3312 }
3313 }
3314 }
3315
3316 if let Some(order_list_id) = order_list_id {
3317 affected_order_list_ids.insert(order_list_id);
3318 }
3319
3320 if self.purge_order_except_aliases(client_order_id) {
3321 purged_client_order_ids.insert(client_order_id);
3322 }
3323 }
3324
3325 if !purged_client_order_ids.is_empty() {
3326 self.index
3327 .venue_order_ids
3328 .retain(|_, owner| !purged_client_order_ids.contains(owner));
3329 }
3330
3331 for order_list_id in affected_order_list_ids {
3332 if let Some(order_list) = self.order_lists.get(&order_list_id) {
3333 let all_purged = order_list
3334 .client_order_ids
3335 .iter()
3336 .all(|id| !self.orders.contains_key(id));
3337
3338 if all_purged {
3339 self.order_lists.remove(&order_list_id);
3340 log::info!("Purged {order_list_id}");
3341 }
3342 }
3343 }
3344 }
3345
3346 pub fn purge_closed_positions(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
3348 log::debug!(
3349 "Purging closed positions{}",
3350 if buffer_secs > 0 {
3351 format!(" with buffer_secs={buffer_secs}")
3352 } else {
3353 String::new()
3354 }
3355 );
3356
3357 let Ok(buffer_ns) = secs_to_nanos(buffer_secs as f64) else {
3358 log::warn!(
3359 "Cannot purge closed positions: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
3360 );
3361 return;
3362 };
3363 let purge_cutoff = ts_now.checked_sub(buffer_ns);
3364
3365 for position_id in self.index.positions_closed.clone() {
3366 let should_purge = self.positions.get(&position_id).is_some_and(|cell| {
3367 let position = cell.borrow();
3368 position.is_closed()
3369 && position.ts_closed.is_some_and(|ts_closed| {
3370 purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
3371 })
3372 });
3373
3374 if should_purge {
3375 self.purge_position(position_id);
3376 }
3377 }
3378 }
3379
3380 pub fn purge_order(&mut self, client_order_id: ClientOrderId) {
3384 if self.purge_order_except_aliases(client_order_id) {
3385 self.index
3386 .venue_order_ids
3387 .retain(|_, owner| owner != &client_order_id);
3388 }
3389 }
3390
3391 fn purge_order_except_aliases(&mut self, client_order_id: ClientOrderId) -> bool {
3396 struct OrderDetails {
3397 is_open: bool,
3398 instrument_id: InstrumentId,
3399 strategy_id: StrategyId,
3400 account_id: Option<AccountId>,
3401 exec_algorithm_id: Option<ExecAlgorithmId>,
3402 exec_spawn_id: Option<ClientOrderId>,
3403 position_id: Option<PositionId>,
3404 }
3405
3406 let order_cell = self.orders.get(&client_order_id).cloned();
3407 let order_details = order_cell.as_ref().map(|cell| {
3408 let order = cell.borrow();
3409 OrderDetails {
3410 is_open: order.is_open(),
3411 instrument_id: order.instrument_id(),
3412 strategy_id: order.strategy_id(),
3413 account_id: order.account_id(),
3414 exec_algorithm_id: order.exec_algorithm_id(),
3415 exec_spawn_id: order.exec_spawn_id(),
3416 position_id: order.position_id(),
3417 }
3418 });
3419
3420 if order_details
3421 .as_ref()
3422 .is_some_and(|details| details.is_open)
3423 {
3424 log::warn!("Order {client_order_id} found open when purging, skipping purge");
3425 return false;
3426 }
3427
3428 if order_details.is_some() {
3429 self.orders.remove(&client_order_id);
3430 } else {
3431 log::warn!("Order {client_order_id} not found when purging");
3432 }
3433
3434 let indexed_position_id = self.index.order_position.remove(&client_order_id);
3435 let indexed_strategy_id = self.index.order_strategy.remove(&client_order_id);
3436 self.index.order_client.remove(&client_order_id);
3437 self.index.client_order_ids.remove(&client_order_id);
3438
3439 if let Some(details) = &order_details {
3440 if let Some(venue_orders) = self
3441 .index
3442 .venue_orders
3443 .get_mut(&details.instrument_id.venue)
3444 {
3445 venue_orders.remove(&client_order_id);
3446 if venue_orders.is_empty() {
3447 self.index.venue_orders.remove(&details.instrument_id.venue);
3448 }
3449 }
3450
3451 let instrument_orders_became_empty = self
3456 .index
3457 .instrument_orders
3458 .get_mut(&details.instrument_id)
3459 .is_some_and(|instrument_orders| {
3460 instrument_orders.remove(&client_order_id);
3461 instrument_orders.is_empty()
3462 });
3463
3464 let has_instrument_positions = self
3465 .index
3466 .instrument_positions
3467 .get(&details.instrument_id)
3468 .is_some_and(|positions| !positions.is_empty());
3469
3470 if instrument_orders_became_empty && !has_instrument_positions {
3471 self.index.instrument_orders.remove(&details.instrument_id);
3472 }
3473
3474 if let Some(exec_algorithm_id) = details.exec_algorithm_id {
3475 let became_empty = self
3476 .index
3477 .exec_algorithm_orders
3478 .get_mut(&exec_algorithm_id)
3479 .is_some_and(|orders| {
3480 orders.remove(&client_order_id);
3481 orders.is_empty()
3482 });
3483
3484 if became_empty {
3485 self.index.exec_algorithm_orders.remove(&exec_algorithm_id);
3486 self.index.exec_algorithms.remove(&exec_algorithm_id);
3487 }
3488 }
3489
3490 if let Some(account_id) = details.account_id
3491 && let Some(account_orders) = self.index.account_orders.get_mut(&account_id)
3492 {
3493 account_orders.remove(&client_order_id);
3494 if account_orders.is_empty() {
3495 self.index.account_orders.remove(&account_id);
3496 }
3497 }
3498
3499 if let Some(exec_spawn_id) = details.exec_spawn_id
3500 && let Some(spawn_orders) = self.index.exec_spawn_orders.get_mut(&exec_spawn_id)
3501 {
3502 spawn_orders.remove(&client_order_id);
3503 if spawn_orders.is_empty() {
3504 self.index.exec_spawn_orders.remove(&exec_spawn_id);
3505 }
3506 }
3507 }
3508
3509 let mut position_ids = AHashSet::new();
3510 if let Some(position_id) = indexed_position_id {
3511 position_ids.insert(position_id);
3512 }
3513
3514 if let Some(position_id) = order_details
3515 .as_ref()
3516 .and_then(|details| details.position_id)
3517 {
3518 position_ids.insert(position_id);
3519 }
3520
3521 let mut strategy_ids = AHashSet::new();
3522 if let Some(strategy_id) = indexed_strategy_id {
3523 strategy_ids.insert(strategy_id);
3524 }
3525
3526 if let Some(details) = &order_details {
3527 strategy_ids.insert(details.strategy_id);
3528 }
3529
3530 for position_id in position_ids {
3531 if self.positions.contains_key(&position_id) {
3532 if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
3533 position_orders.remove(&client_order_id);
3534 }
3535 continue;
3536 }
3537
3538 let has_other_orders =
3539 if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
3540 position_orders.remove(&client_order_id);
3541 !position_orders.is_empty()
3542 } else {
3543 self.index
3544 .order_position
3545 .values()
3546 .any(|candidate| *candidate == position_id)
3547 };
3548
3549 if has_other_orders {
3550 continue;
3551 }
3552
3553 self.index.position_orders.remove(&position_id);
3554 if let Some(strategy_id) = self.index.position_strategy.remove(&position_id) {
3555 strategy_ids.insert(strategy_id);
3556 if let Some(strategy_positions) =
3557 self.index.strategy_positions.get_mut(&strategy_id)
3558 {
3559 strategy_positions.remove(&position_id);
3560 if strategy_positions.is_empty() {
3561 self.index.strategy_positions.remove(&strategy_id);
3562 }
3563 }
3564 }
3565
3566 if let Some(details) = &order_details
3567 && let Some(venue_positions) = self
3568 .index
3569 .venue_positions
3570 .get_mut(&details.instrument_id.venue)
3571 {
3572 venue_positions.remove(&position_id);
3573 if venue_positions.is_empty() {
3574 self.index
3575 .venue_positions
3576 .remove(&details.instrument_id.venue);
3577 }
3578 }
3579 }
3580
3581 for strategy_id in strategy_ids {
3582 let strategy_orders_became_empty = self
3588 .index
3589 .strategy_orders
3590 .get_mut(&strategy_id)
3591 .is_some_and(|strategy_orders| {
3592 strategy_orders.remove(&client_order_id);
3593 strategy_orders.is_empty()
3594 });
3595
3596 let has_positions = self
3597 .index
3598 .strategy_positions
3599 .get(&strategy_id)
3600 .is_some_and(|strategy_positions| !strategy_positions.is_empty());
3601
3602 if strategy_orders_became_empty && !has_positions {
3603 self.index.strategy_orders.remove(&strategy_id);
3604 self.index.strategies.remove(&strategy_id);
3605 }
3606 }
3607
3608 self.index.exec_spawn_orders.remove(&client_order_id);
3609
3610 self.index.orders.remove(&client_order_id);
3611 self.index.orders_active_local.remove(&client_order_id);
3612 self.index.orders_open.remove(&client_order_id);
3613 self.index.orders_closed.remove(&client_order_id);
3614 self.index.orders_emulated.remove(&client_order_id);
3615 self.index.orders_inflight.remove(&client_order_id);
3616 self.index.orders_pending_cancel.remove(&client_order_id);
3617
3618 if order_details.is_some() {
3619 log::info!("Purged order {client_order_id}");
3620 }
3621
3622 true
3623 }
3624
3625 pub fn purge_position(&mut self, position_id: PositionId) {
3629 let position = self
3631 .positions
3632 .get(&position_id)
3633 .map(|cell| cell.borrow().clone());
3634
3635 if let Some(ref pos) = position
3637 && pos.is_open()
3638 {
3639 log::warn!("Position {position_id} found open when purging, skipping purge");
3640 return;
3641 }
3642
3643 if let Some(ref pos) = position {
3645 self.positions.remove(&position_id);
3646
3647 if let Some(venue_positions) =
3649 self.index.venue_positions.get_mut(&pos.instrument_id.venue)
3650 {
3651 venue_positions.remove(&position_id);
3652 if venue_positions.is_empty() {
3653 self.index.venue_positions.remove(&pos.instrument_id.venue);
3654 }
3655 }
3656
3657 let instrument_positions_became_empty = self
3659 .index
3660 .instrument_positions
3661 .get_mut(&pos.instrument_id)
3662 .is_some_and(|positions| {
3663 positions.remove(&position_id);
3664 positions.is_empty()
3665 });
3666
3667 if instrument_positions_became_empty {
3668 self.index.instrument_positions.remove(&pos.instrument_id);
3669 let instrument_orders_empty = self
3670 .index
3671 .instrument_orders
3672 .get(&pos.instrument_id)
3673 .is_some_and(|orders| orders.is_empty());
3674
3675 if instrument_orders_empty {
3676 self.index.instrument_orders.remove(&pos.instrument_id);
3677 }
3678 }
3679
3680 let strategy_positions_became_empty = self
3682 .index
3683 .strategy_positions
3684 .get_mut(&pos.strategy_id)
3685 .is_some_and(|positions| {
3686 positions.remove(&position_id);
3687 positions.is_empty()
3688 });
3689
3690 if strategy_positions_became_empty {
3691 self.index.strategy_positions.remove(&pos.strategy_id);
3692 let strategy_orders_empty = self
3693 .index
3694 .strategy_orders
3695 .get(&pos.strategy_id)
3696 .is_some_and(|orders| orders.is_empty());
3697
3698 if strategy_orders_empty {
3699 self.index.strategy_orders.remove(&pos.strategy_id);
3700 self.index.strategies.remove(&pos.strategy_id);
3701 }
3702 }
3703
3704 if let Some(account_positions) = self.index.account_positions.get_mut(&pos.account_id) {
3706 account_positions.remove(&position_id);
3707 if account_positions.is_empty() {
3708 self.index.account_positions.remove(&pos.account_id);
3709 }
3710 }
3711
3712 for client_order_id in pos.client_order_ids() {
3714 self.index.order_position.remove(&client_order_id);
3715 }
3716
3717 log::info!("Purged position {position_id}");
3718 } else {
3719 log::warn!("Position {position_id} not found when purging");
3720 }
3721
3722 self.index.position_strategy.remove(&position_id);
3724 self.index.position_oms.remove(&position_id);
3725 self.index.position_orders.remove(&position_id);
3726 self.index.positions.remove(&position_id);
3727 self.index.positions_open.remove(&position_id);
3728 self.index.positions_closed.remove(&position_id);
3729
3730 self.position_snapshots.remove(&position_id);
3732 self.bump_position_snapshot_revision(position_id);
3733 }
3734
3735 fn purge_instrument_inner(&mut self, instrument_id: InstrumentId, skip_order_guard: bool) {
3759 #[cfg(feature = "defi")]
3760 let defi_found = self.defi.pools.contains_key(&instrument_id)
3761 || self.defi.pool_profilers.contains_key(&instrument_id);
3762 #[cfg(not(feature = "defi"))]
3763 let defi_found = false;
3764
3765 let found = self.instruments.contains_key(&instrument_id)
3766 || self.synthetics.contains_key(&instrument_id)
3767 || defi_found;
3768
3769 if !found {
3770 log::warn!("Instrument {instrument_id} not found when purging");
3771 return;
3772 }
3773
3774 if !skip_order_guard && let Some(orders) = self.index.instrument_orders.get(&instrument_id)
3775 {
3776 let has_non_terminal = orders
3777 .iter()
3778 .any(|client_order_id| !self.index.orders_closed.contains(client_order_id));
3779
3780 if has_non_terminal {
3781 log::warn!(
3782 "Instrument {instrument_id} has non-terminal orders when purging, skipping purge"
3783 );
3784 return;
3785 }
3786 }
3787
3788 if let Some(positions) = self.index.instrument_positions.get(&instrument_id) {
3789 let has_non_closed = positions
3790 .iter()
3791 .any(|position_id| !self.index.positions_closed.contains(position_id));
3792
3793 if has_non_closed {
3794 log::warn!(
3795 "Instrument {instrument_id} has non-closed positions when purging, skipping purge"
3796 );
3797 return;
3798 }
3799 }
3800
3801 self.instruments.remove(&instrument_id);
3802 self.synthetics.remove(&instrument_id);
3803 self.books.remove(&instrument_id);
3804 self.own_books.remove(&instrument_id);
3805 self.quotes.remove(&instrument_id);
3806 self.trades.remove(&instrument_id);
3807 self.mark_prices.remove(&instrument_id);
3808 self.index_prices.remove(&instrument_id);
3809 self.funding_rates.remove(&instrument_id);
3810 self.instrument_statuses.remove(&instrument_id);
3811 self.greeks.remove(&instrument_id);
3812 self.option_greeks.remove(&instrument_id);
3813
3814 self.bars
3815 .retain(|bar_type, _| bar_type.instrument_id() != instrument_id);
3816
3817 #[cfg(feature = "defi")]
3818 {
3819 self.defi.pools.remove(&instrument_id);
3820 self.defi.pool_profilers.remove(&instrument_id);
3821 }
3822
3823 self.index.instrument_orders.remove(&instrument_id);
3824 self.index.instrument_positions.remove(&instrument_id);
3825
3826 log::info!("Purged instrument {instrument_id}");
3827 }
3828
3829 pub fn purge_instrument(&mut self, instrument_id: InstrumentId) {
3834 self.purge_instrument_inner(instrument_id, false);
3835 }
3836
3837 pub fn purge_instrument_skip_order_guard(&mut self, instrument_id: InstrumentId) {
3846 self.purge_instrument_inner(instrument_id, true);
3847 }
3848
3849 pub fn purge_account_events(&mut self, ts_now: UnixNanos, lookback_secs: u64) {
3854 log::debug!(
3855 "Purging account events{}",
3856 if lookback_secs > 0 {
3857 format!(" with lookback_secs={lookback_secs}")
3858 } else {
3859 String::new()
3860 }
3861 );
3862
3863 for account_cell in self.accounts.values() {
3864 let mut account = account_cell.borrow_mut();
3865 let event_count = account.event_count();
3866 account.purge_account_events(ts_now, lookback_secs);
3867 let count_diff = event_count - account.event_count();
3868 if count_diff > 0 {
3869 log::info!(
3870 "Purged {} event(s) from account {}",
3871 count_diff,
3872 account.id()
3873 );
3874 }
3875 }
3876 }
3877
3878 pub fn clear_index(&mut self) {
3880 self.index.clear();
3881 log::debug!("Cleared index");
3882 }
3883
3884 pub fn reset(&mut self) {
3890 log::debug!("Resetting cache");
3891
3892 self.general.clear();
3893 self.books.clear();
3894 self.own_books.clear();
3895 self.quotes.clear();
3896 self.trades.clear();
3897 self.mark_xrates.clear();
3898 self.mark_prices.clear();
3899 self.index_prices.clear();
3900 self.funding_rates.clear();
3901 self.instrument_statuses.clear();
3902 self.bars.clear();
3903 self.accounts.clear();
3904 self.orders.clear();
3905 self.order_lists.clear();
3906 self.positions.clear();
3907 self.position_snapshots.clear();
3908 self.position_snapshot_revisions.clear();
3909 self.greeks.clear();
3910 self.option_greeks.clear();
3911 self.yield_curves.clear();
3912
3913 if self.config.drop_instruments_on_reset {
3914 self.currencies.clear();
3915 self.instruments.clear();
3916 self.synthetics.clear();
3917 }
3918
3919 #[cfg(feature = "defi")]
3920 {
3921 self.defi.pools.clear();
3922 self.defi.pool_profilers.clear();
3923 }
3924
3925 self.clear_index();
3926
3927 log::info!("Reset cache");
3928 }
3929
3930 pub fn dispose(&mut self) {
3934 self.reset();
3935
3936 if let Some(database) = &mut self.database
3937 && let Err(e) = database.close()
3938 {
3939 log::error!("Failed to close database during dispose: {e}");
3940 }
3941 }
3942
3943 pub fn flush_db(&mut self) {
3947 if let Some(database) = &mut self.database
3948 && let Err(e) = database.flush()
3949 {
3950 log::error!("Failed to flush database: {e}");
3951 }
3952 }
3953
3954 pub fn add(&mut self, key: &str, value: Bytes) -> anyhow::Result<()> {
3962 check_valid_string_ascii(key, stringify!(key))?;
3963 check_predicate_false(value.is_empty(), stringify!(value))?;
3964
3965 log::debug!("Adding general {key}");
3966 self.general.insert(key.to_string(), value.clone());
3967
3968 if let Some(database) = &mut self.database {
3969 database.add(key.to_string(), value)?;
3970 }
3971 Ok(())
3972 }
3973
3974 pub fn add_order_book(&mut self, book: OrderBook) -> anyhow::Result<()> {
3980 log::debug!("Adding `OrderBook` {}", book.instrument_id);
3981
3982 if self.config.save_market_data
3983 && let Some(database) = &mut self.database
3984 {
3985 database.add_order_book(&book)?;
3986 }
3987
3988 self.books.insert(book.instrument_id, book);
3989 Ok(())
3990 }
3991
3992 pub fn add_own_order_book(&mut self, own_book: OwnOrderBook) -> anyhow::Result<()> {
3998 log::debug!("Adding `OwnOrderBook` {}", own_book.instrument_id);
3999
4000 self.own_books.insert(own_book.instrument_id, own_book);
4001 Ok(())
4002 }
4003
4004 pub fn add_mark_price(&mut self, mark_price: MarkPriceUpdate) -> anyhow::Result<()> {
4010 log::debug!("Adding `MarkPriceUpdate` for {}", mark_price.instrument_id);
4011
4012 if self.config.save_market_data {
4013 }
4015
4016 let mark_prices_deque = self
4017 .mark_prices
4018 .entry(mark_price.instrument_id)
4019 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4020 mark_prices_deque.push_front(mark_price);
4021 Ok(())
4022 }
4023
4024 pub fn add_index_price(&mut self, index_price: IndexPriceUpdate) -> anyhow::Result<()> {
4030 log::debug!(
4031 "Adding `IndexPriceUpdate` for {}",
4032 index_price.instrument_id
4033 );
4034
4035 if self.config.save_market_data {
4036 }
4038
4039 let index_prices_deque = self
4040 .index_prices
4041 .entry(index_price.instrument_id)
4042 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4043 index_prices_deque.push_front(index_price);
4044 Ok(())
4045 }
4046
4047 pub fn add_funding_rate(&mut self, funding_rate: FundingRateUpdate) -> anyhow::Result<()> {
4053 log::debug!(
4054 "Adding `FundingRateUpdate` for {}",
4055 funding_rate.instrument_id
4056 );
4057
4058 if self.config.save_market_data {
4059 }
4061
4062 let funding_rates_deque = self
4063 .funding_rates
4064 .entry(funding_rate.instrument_id)
4065 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4066 funding_rates_deque.push_front(funding_rate);
4067 Ok(())
4068 }
4069
4070 pub fn add_funding_rates(&mut self, funding_rates: &[FundingRateUpdate]) -> anyhow::Result<()> {
4076 check_slice_not_empty(funding_rates, stringify!(funding_rates))?;
4077
4078 let instrument_id = funding_rates[0].instrument_id;
4079 log::debug!(
4080 "Adding `FundingRateUpdate`[{}] {instrument_id}",
4081 funding_rates.len()
4082 );
4083
4084 if self.config.save_market_data
4085 && let Some(database) = &mut self.database
4086 {
4087 for funding_rate in funding_rates {
4088 database.add_funding_rate(funding_rate)?;
4089 }
4090 }
4091
4092 let funding_rate_deque = self
4093 .funding_rates
4094 .entry(instrument_id)
4095 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4096
4097 for funding_rate in funding_rates {
4098 funding_rate_deque.push_front(*funding_rate);
4099 }
4100 Ok(())
4101 }
4102
4103 pub fn add_instrument_status(&mut self, status: InstrumentStatus) -> anyhow::Result<()> {
4109 log::debug!("Adding `InstrumentStatus` for {}", status.instrument_id);
4110
4111 if self.config.save_market_data {
4112 }
4114
4115 let statuses_deque = self
4116 .instrument_statuses
4117 .entry(status.instrument_id)
4118 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4119 statuses_deque.push_front(status);
4120 Ok(())
4121 }
4122
4123 pub fn add_quote(&mut self, quote: QuoteTick) -> anyhow::Result<()> {
4129 log::debug!("Adding `QuoteTick` {}", quote.instrument_id);
4130
4131 if self.config.save_market_data
4132 && let Some(database) = &mut self.database
4133 {
4134 database.add_quote("e)?;
4135 }
4136
4137 let quotes_deque = self
4138 .quotes
4139 .entry(quote.instrument_id)
4140 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4141 quotes_deque.push_front(quote);
4142 Ok(())
4143 }
4144
4145 pub fn add_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
4151 check_slice_not_empty(quotes, stringify!(quotes))?;
4152
4153 let instrument_id = quotes[0].instrument_id;
4154 log::debug!("Adding `QuoteTick`[{}] {instrument_id}", quotes.len());
4155
4156 if self.config.save_market_data
4157 && let Some(database) = &mut self.database
4158 {
4159 for quote in quotes {
4160 database.add_quote(quote)?;
4161 }
4162 }
4163
4164 let quotes_deque = self
4165 .quotes
4166 .entry(instrument_id)
4167 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4168
4169 for quote in quotes {
4170 quotes_deque.push_front(*quote);
4171 }
4172 Ok(())
4173 }
4174
4175 pub fn add_trade(&mut self, trade: TradeTick) -> anyhow::Result<()> {
4181 log::debug!("Adding `TradeTick` {}", trade.instrument_id);
4182
4183 if self.config.save_market_data
4184 && let Some(database) = &mut self.database
4185 {
4186 database.add_trade(&trade)?;
4187 }
4188
4189 let trades_deque = self
4190 .trades
4191 .entry(trade.instrument_id)
4192 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4193 trades_deque.push_front(trade);
4194 Ok(())
4195 }
4196
4197 pub fn add_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
4203 check_slice_not_empty(trades, stringify!(trades))?;
4204
4205 let instrument_id = trades[0].instrument_id;
4206 log::debug!("Adding `TradeTick`[{}] {instrument_id}", trades.len());
4207
4208 if self.config.save_market_data
4209 && let Some(database) = &mut self.database
4210 {
4211 for trade in trades {
4212 database.add_trade(trade)?;
4213 }
4214 }
4215
4216 let trades_deque = self
4217 .trades
4218 .entry(instrument_id)
4219 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4220
4221 for trade in trades {
4222 trades_deque.push_front(*trade);
4223 }
4224 Ok(())
4225 }
4226
4227 pub fn add_bar(&mut self, bar: Bar) -> anyhow::Result<()> {
4233 log::debug!("Adding `Bar` {}", bar.bar_type);
4234
4235 if self.config.save_market_data
4236 && let Some(database) = &mut self.database
4237 {
4238 database.add_bar(&bar)?;
4239 }
4240
4241 let bars = self
4242 .bars
4243 .entry(bar.bar_type)
4244 .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
4245 bars.push_front(bar);
4246 Ok(())
4247 }
4248
4249 pub fn add_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
4255 check_slice_not_empty(bars, stringify!(bars))?;
4256
4257 let bar_type = bars[0].bar_type;
4258 log::debug!("Adding `Bar`[{}] {bar_type}", bars.len());
4259
4260 if self.config.save_market_data
4261 && let Some(database) = &mut self.database
4262 {
4263 for bar in bars {
4264 database.add_bar(bar)?;
4265 }
4266 }
4267
4268 let bars_deque = self
4269 .bars
4270 .entry(bar_type)
4271 .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
4272
4273 for bar in bars {
4274 bars_deque.push_front(*bar);
4275 }
4276 Ok(())
4277 }
4278
4279 pub fn add_greeks(&mut self, greeks: GreeksData) -> anyhow::Result<()> {
4285 log::debug!("Adding `GreeksData` {}", greeks.instrument_id);
4286
4287 if self.config.save_market_data
4288 && let Some(_database) = &mut self.database
4289 {
4290 }
4292
4293 self.greeks.insert(greeks.instrument_id, greeks);
4294 Ok(())
4295 }
4296
4297 pub fn greeks(&self, instrument_id: &InstrumentId) -> Option<GreeksData> {
4299 self.greeks.get(instrument_id).cloned()
4300 }
4301
4302 pub fn add_option_greeks(&mut self, greeks: OptionGreeks) {
4304 log::debug!("Adding `OptionGreeks` {}", greeks.instrument_id);
4305 self.option_greeks.insert(greeks.instrument_id, greeks);
4306 }
4307
4308 #[must_use]
4310 pub fn option_greeks(&self, instrument_id: &InstrumentId) -> Option<&OptionGreeks> {
4311 self.option_greeks.get(instrument_id)
4312 }
4313
4314 pub fn add_yield_curve(&mut self, yield_curve: YieldCurveData) -> anyhow::Result<()> {
4320 log::debug!("Adding `YieldCurveData` {}", yield_curve.curve_name);
4321
4322 if self.config.save_market_data
4323 && let Some(_database) = &mut self.database
4324 {
4325 }
4327
4328 self.yield_curves
4329 .insert(yield_curve.curve_name.clone(), yield_curve);
4330 Ok(())
4331 }
4332
4333 pub fn yield_curve(&self, key: &str) -> Option<Box<dyn Fn(f64) -> f64>> {
4335 self.yield_curves.get(key).map(|curve| {
4336 let curve_clone = curve.clone();
4337 Box::new(move |expiry_in_years: f64| curve_clone.get_rate(expiry_in_years))
4338 as Box<dyn Fn(f64) -> f64>
4339 })
4340 }
4341
4342 pub fn add_currency(&mut self, currency: Currency) -> anyhow::Result<()> {
4348 if self.currencies.contains_key(¤cy.code) {
4349 return Ok(());
4350 }
4351 log::debug!("Adding `Currency` {}", currency.code);
4352
4353 if let Some(database) = &mut self.database {
4354 database.add_currency(¤cy)?;
4355 }
4356
4357 self.currencies.insert(currency.code, currency);
4358 Ok(())
4359 }
4360
4361 pub fn add_instrument(&mut self, instrument: InstrumentAny) -> anyhow::Result<()> {
4367 log::debug!("Adding `Instrument` {}", instrument.id());
4368
4369 if let Some(base_currency) = instrument.base_currency() {
4371 self.add_currency(base_currency)?;
4372 }
4373 self.add_currency(instrument.quote_currency())?;
4374 self.add_currency(instrument.settlement_currency())?;
4375
4376 if let Some(database) = &mut self.database {
4377 database.add_instrument(&instrument)?;
4378 }
4379
4380 self.instruments.insert(instrument.id(), instrument);
4381 Ok(())
4382 }
4383
4384 pub fn add_synthetic(&mut self, synthetic: SyntheticInstrument) -> anyhow::Result<()> {
4390 log::debug!("Adding `SyntheticInstrument` {}", synthetic.id);
4391
4392 if let Some(database) = &mut self.database {
4393 database.add_synthetic(&synthetic)?;
4394 }
4395
4396 self.synthetics.insert(synthetic.id, synthetic);
4397 Ok(())
4398 }
4399
4400 pub fn add_account(&mut self, account: AccountAny) -> anyhow::Result<()> {
4406 log::debug!("Adding `Account` {}", account.id());
4407
4408 if let Some(database) = &mut self.database {
4409 database.add_account(&account)?;
4410 }
4411
4412 let account_id = account.id();
4413 self.accounts.insert(account_id, SharedCell::new(account));
4414 self.index
4415 .venue_account
4416 .insert(account_id.get_issuer(), account_id);
4417 Ok(())
4418 }
4419
4420 pub fn add_venue_order_id(
4429 &mut self,
4430 client_order_id: &ClientOrderId,
4431 venue_order_id: &VenueOrderId,
4432 overwrite: bool,
4433 ) -> anyhow::Result<()> {
4434 self.validate_venue_order_id_claim(client_order_id, venue_order_id, overwrite)?;
4435
4436 self.index
4437 .client_order_ids
4438 .insert(*client_order_id, *venue_order_id);
4439 self.index
4440 .venue_order_ids
4441 .insert(*venue_order_id, *client_order_id);
4442
4443 Ok(())
4444 }
4445
4446 pub fn index_venue_order_id(
4457 &mut self,
4458 client_order_id: &ClientOrderId,
4459 venue_order_id: &VenueOrderId,
4460 ) -> anyhow::Result<()> {
4461 self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
4462
4463 self.index
4464 .venue_order_ids
4465 .insert(*venue_order_id, *client_order_id);
4466 self.index
4467 .client_order_ids
4468 .entry(*client_order_id)
4469 .or_insert(*venue_order_id);
4470
4471 Ok(())
4472 }
4473
4474 fn validate_venue_order_id_claim(
4475 &self,
4476 client_order_id: &ClientOrderId,
4477 venue_order_id: &VenueOrderId,
4478 overwrite: bool,
4479 ) -> anyhow::Result<()> {
4480 self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
4481
4482 if let Some(existing_venue_order_id) = self.index.client_order_ids.get(client_order_id)
4483 && !overwrite
4484 && existing_venue_order_id != venue_order_id
4485 {
4486 anyhow::bail!(
4487 "Existing {existing_venue_order_id} for {client_order_id}
4488 did not match the given {venue_order_id}.
4489 If you are writing a test then try a different `venue_order_id`,
4490 otherwise this is probably a bug."
4491 );
4492 }
4493
4494 Ok(())
4495 }
4496
4497 fn validate_venue_order_id_ownership(
4498 &self,
4499 client_order_id: &ClientOrderId,
4500 venue_order_id: &VenueOrderId,
4501 ) -> anyhow::Result<()> {
4502 if let Some(existing_client_order_id) = self.index.venue_order_ids.get(venue_order_id)
4503 && existing_client_order_id != client_order_id
4504 {
4505 return Err(VenueOrderIdOwnershipError {
4506 venue_order_id: *venue_order_id,
4507 existing_client_order_id: *existing_client_order_id,
4508 claimant_client_order_id: *client_order_id,
4509 }
4510 .into());
4511 }
4512
4513 Ok(())
4514 }
4515
4516 pub fn add_order(
4531 &mut self,
4532 order: OrderAny,
4533 position_id: Option<PositionId>,
4534 client_id: Option<ClientId>,
4535 replace_existing: bool,
4536 ) -> anyhow::Result<()> {
4537 let instrument_id = order.instrument_id();
4538 let venue = instrument_id.venue;
4539 let client_order_id = order.client_order_id();
4540 let strategy_id = order.strategy_id();
4541 let exec_algorithm_id = order.exec_algorithm_id();
4542 let exec_spawn_id = order.exec_spawn_id();
4543
4544 if !replace_existing {
4545 check_key_not_in_map(
4546 &client_order_id,
4547 &self.orders,
4548 stringify!(client_order_id),
4549 stringify!(orders),
4550 )?;
4551 }
4552
4553 log::debug!("Adding {order:?}");
4554
4555 self.index.orders.insert(client_order_id);
4556
4557 if order.is_active_local() {
4558 self.index.orders_active_local.insert(client_order_id);
4559 }
4560 self.index
4561 .order_strategy
4562 .insert(client_order_id, strategy_id);
4563 self.index.strategies.insert(strategy_id);
4564
4565 self.index
4567 .venue_orders
4568 .entry(venue)
4569 .or_default()
4570 .insert(client_order_id);
4571
4572 self.index
4574 .instrument_orders
4575 .entry(instrument_id)
4576 .or_default()
4577 .insert(client_order_id);
4578
4579 self.index
4581 .strategy_orders
4582 .entry(strategy_id)
4583 .or_default()
4584 .insert(client_order_id);
4585
4586 if let Some(account_id) = order.account_id() {
4588 self.index
4589 .account_orders
4590 .entry(account_id)
4591 .or_default()
4592 .insert(client_order_id);
4593 }
4594
4595 if let Some(exec_algorithm_id) = exec_algorithm_id {
4597 self.index.exec_algorithms.insert(exec_algorithm_id);
4598
4599 self.index
4600 .exec_algorithm_orders
4601 .entry(exec_algorithm_id)
4602 .or_default()
4603 .insert(client_order_id);
4604 }
4605
4606 if let Some(exec_spawn_id) = exec_spawn_id {
4608 self.index
4609 .exec_spawn_orders
4610 .entry(exec_spawn_id)
4611 .or_default()
4612 .insert(client_order_id);
4613 }
4614
4615 if order.emulation_trigger().is_some() {
4617 self.index.orders_emulated.insert(client_order_id);
4618 }
4619
4620 if let Some(position_id) = position_id {
4622 self.index_position_id_in_memory(&position_id, &venue, &client_order_id, &strategy_id);
4623 }
4624
4625 if let Some(client_id) = client_id {
4627 self.index.order_client.insert(client_order_id, client_id);
4628 log::debug!("Indexed {client_id:?}");
4629 }
4630
4631 let order_cell = if let Some(order_cell) = self.orders.get(&client_order_id) {
4634 *order_cell.borrow_mut() = order;
4635 order_cell.clone()
4636 } else {
4637 let order_cell = SharedCell::new(order);
4638 self.orders.insert(client_order_id, order_cell.clone());
4639 order_cell
4640 };
4641
4642 if let Some(position_id) = position_id {
4643 self.persist_position_id(&position_id, &client_order_id)?;
4644 }
4645
4646 if let Some(database) = &mut self.database {
4647 database.add_order(&order_cell.borrow(), client_id)?;
4648 }
4653
4654 Ok(())
4655 }
4656
4657 pub fn claim_order_clients(
4669 &mut self,
4670 claims: &[(ClientOrderId, ClientId)],
4671 ) -> anyhow::Result<()> {
4672 let mut requested = AHashMap::with_capacity(claims.len());
4673 let mut ordered_claims = Vec::with_capacity(claims.len());
4674
4675 for (client_order_id, client_id) in claims {
4676 if let Some(existing_client_id) = requested.get(client_order_id) {
4677 if existing_client_id != client_id {
4678 anyhow::bail!(
4679 "Conflicting execution client claims for {client_order_id}: \
4680 {existing_client_id} and {client_id}"
4681 );
4682 }
4683 continue;
4684 }
4685
4686 requested.insert(*client_order_id, *client_id);
4687 ordered_claims.push((*client_order_id, *client_id));
4688 }
4689
4690 let mut pending_claims = Vec::with_capacity(ordered_claims.len());
4691 for (client_order_id, client_id) in ordered_claims {
4692 if !self.orders.contains_key(&client_order_id) {
4693 return Err(OrderLookupError::not_found(client_order_id).into());
4694 }
4695
4696 match self.index.order_client.get(&client_order_id) {
4697 Some(existing_client_id) if *existing_client_id == client_id => {}
4698 Some(existing_client_id) => {
4699 anyhow::bail!(
4700 "Order {client_order_id} is already claimed by execution client \
4701 {existing_client_id} and cannot be claimed by {client_id}"
4702 );
4703 }
4704 None => pending_claims.push((client_order_id, client_id)),
4705 }
4706 }
4707
4708 if pending_claims.is_empty() {
4709 return Ok(());
4710 }
4711
4712 if let Some(database) = &self.database {
4713 database.index_order_clients(&pending_claims)?;
4714 }
4715
4716 for (client_order_id, client_id) in pending_claims {
4717 self.index.order_client.insert(client_order_id, client_id);
4718 log::debug!("Claimed {client_order_id} for execution client {client_id}");
4719 }
4720
4721 Ok(())
4722 }
4723
4724 pub fn add_order_list(&mut self, order_list: OrderList) -> anyhow::Result<()> {
4730 let order_list_id = order_list.id;
4731 check_key_not_in_map(
4732 &order_list_id,
4733 &self.order_lists,
4734 stringify!(order_list_id),
4735 stringify!(order_lists),
4736 )?;
4737
4738 log::debug!("Adding {order_list:?}");
4739 self.order_lists.insert(order_list_id, order_list);
4740 Ok(())
4741 }
4742
4743 pub fn add_position_id(
4751 &mut self,
4752 position_id: &PositionId,
4753 venue: &Venue,
4754 client_order_id: &ClientOrderId,
4755 strategy_id: &StrategyId,
4756 ) -> anyhow::Result<()> {
4757 self.index_position_id_in_memory(position_id, venue, client_order_id, strategy_id);
4758 self.persist_position_id(position_id, client_order_id)
4759 }
4760
4761 fn index_position_id_in_memory(
4762 &mut self,
4763 position_id: &PositionId,
4764 venue: &Venue,
4765 client_order_id: &ClientOrderId,
4766 strategy_id: &StrategyId,
4767 ) {
4768 self.index
4769 .order_position
4770 .insert(*client_order_id, *position_id);
4771 self.index_position(position_id, venue, strategy_id);
4772 self.index
4773 .position_orders
4774 .entry(*position_id)
4775 .or_default()
4776 .insert(*client_order_id);
4777 }
4778
4779 fn persist_position_id(
4780 &mut self,
4781 position_id: &PositionId,
4782 client_order_id: &ClientOrderId,
4783 ) -> anyhow::Result<()> {
4784 if let Some(database) = &mut self.database {
4785 database.index_order_position(*client_order_id, *position_id)?;
4786 }
4787
4788 Ok(())
4789 }
4790
4791 fn index_position(
4792 &mut self,
4793 position_id: &PositionId,
4794 venue: &Venue,
4795 strategy_id: &StrategyId,
4796 ) {
4797 self.index
4799 .position_strategy
4800 .insert(*position_id, *strategy_id);
4801
4802 self.index.position_orders.entry(*position_id).or_default();
4804
4805 self.index
4807 .strategy_positions
4808 .entry(*strategy_id)
4809 .or_default()
4810 .insert(*position_id);
4811
4812 self.index
4814 .venue_positions
4815 .entry(*venue)
4816 .or_default()
4817 .insert(*position_id);
4818 }
4819
4820 fn assign_position_ids_to_contingencies(&mut self) {
4828 let mut assignments: Vec<(PositionId, ClientOrderId)> = Vec::new();
4829
4830 for parent_order_cell in self.orders.values() {
4831 let parent = parent_order_cell.borrow();
4832 if parent.contingency_type() != Some(ContingencyType::Oto) {
4833 continue;
4834 }
4835 let Some(parent_position_id) = parent.position_id() else {
4836 continue;
4837 };
4838 let Some(linked_order_ids) = parent.linked_order_ids() else {
4839 continue;
4840 };
4841
4842 for client_order_id in linked_order_ids {
4843 match self.orders.get(client_order_id) {
4844 None => {
4845 log::error!("Contingency order {client_order_id} not found");
4846 }
4847 Some(contingent_order_cell) => {
4848 if contingent_order_cell.borrow().position_id().is_none() {
4849 assignments.push((parent_position_id, *client_order_id));
4850 }
4851 }
4852 }
4853 }
4854 }
4855
4856 for (position_id, client_order_id) in assignments {
4857 let Some((venue, strategy_id)) = self.orders.get(&client_order_id).map(|order_cell| {
4858 let mut contingent = order_cell.borrow_mut();
4859 contingent.set_position_id(Some(position_id));
4860 (contingent.instrument_id().venue, contingent.strategy_id())
4861 }) else {
4862 continue;
4863 };
4864
4865 if let Err(e) =
4868 self.add_position_id(&position_id, &venue, &client_order_id, &strategy_id)
4869 {
4870 log::error!("Failed to re-index {client_order_id} -> {position_id}: {e}");
4871 }
4872 }
4873 }
4874
4875 pub fn add_position(&mut self, position: &Position, oms_type: OmsType) -> anyhow::Result<()> {
4883 self.add_position_inner(position, oms_type, true)
4884 }
4885
4886 pub fn add_position_without_order(
4894 &mut self,
4895 position: &Position,
4896 oms_type: OmsType,
4897 ) -> anyhow::Result<()> {
4898 self.add_position_inner(position, oms_type, false)
4899 }
4900
4901 fn add_position_inner(
4902 &mut self,
4903 position: &Position,
4904 oms_type: OmsType,
4905 index_order: bool,
4906 ) -> anyhow::Result<()> {
4907 let key = position_oms_key(position.id);
4910 check_valid_string_ascii(&key, stringify!(key))?;
4911 let value = Bytes::from(serde_json::to_vec(&oms_type)?);
4912 check_predicate_false(value.is_empty(), stringify!(value))?;
4913
4914 self.positions
4915 .insert(position.id, SharedCell::new(position.clone()));
4916 self.index.position_oms.insert(position.id, oms_type);
4917 self.index.positions.insert(position.id);
4918 self.index.positions_open.insert(position.id);
4919 self.index.positions_closed.remove(&position.id); self.index.strategies.insert(position.strategy_id);
4921 self.index
4922 .strategy_orders
4923 .entry(position.strategy_id)
4924 .or_default();
4925
4926 log::debug!("Adding {position}");
4927
4928 if index_order {
4929 self.index_position_id_in_memory(
4930 &position.id,
4931 &position.instrument_id.venue,
4932 &position.opening_order_id,
4933 &position.strategy_id,
4934 );
4935 } else {
4936 self.index_position(
4937 &position.id,
4938 &position.instrument_id.venue,
4939 &position.strategy_id,
4940 );
4941 }
4942
4943 let instrument_id = position.instrument_id;
4945 let instrument_positions = self
4946 .index
4947 .instrument_positions
4948 .entry(instrument_id)
4949 .or_default();
4950 instrument_positions.insert(position.id);
4951 self.index
4952 .instrument_orders
4953 .entry(instrument_id)
4954 .or_default();
4955
4956 self.index
4958 .account_positions
4959 .entry(position.account_id)
4960 .or_default()
4961 .insert(position.id);
4962
4963 log::debug!("Adding general {key}");
4964 self.general.insert(key.clone(), value.clone());
4965
4966 if index_order {
4967 self.persist_position_id(&position.id, &position.opening_order_id)?;
4968 }
4969
4970 if let Some(database) = &mut self.database {
4971 database.add_position(position)?;
4972 database.add(key, value)?;
4981 }
4982
4983 Ok(())
4984 }
4985
4986 pub fn update_account(&mut self, account: &AccountAny) -> anyhow::Result<()> {
4995 let account_id = account.id();
4996 match self.accounts.get(&account_id) {
4997 Some(account_cell) => *account_cell.borrow_mut() = account.clone(),
4998 None => {
4999 self.accounts
5000 .insert(account_id, SharedCell::new(account.clone()));
5001 }
5002 }
5003
5004 if let Some(database) = &mut self.database {
5005 database.update_account(account)?;
5006 }
5007 Ok(())
5008 }
5009
5010 #[must_use]
5021 pub fn take_account(&mut self, account_id: &AccountId) -> Option<AccountAny> {
5022 let cell = self.accounts.remove(account_id)?;
5023 let rc: Rc<RefCell<AccountAny>> = cell.into();
5024
5025 match Rc::try_unwrap(rc) {
5026 Ok(cell) => Some(cell.into_inner()),
5027 Err(rc) => {
5028 log::error!(
5029 "Cannot move account {account_id} out of cache: account cell has an outstanding owner"
5030 );
5031 self.accounts.insert(*account_id, rc.into());
5032 None
5033 }
5034 }
5035 }
5036
5037 pub fn cache_account_owned(&mut self, account: AccountAny) {
5039 let account_id = account.id();
5040 self.index
5041 .venue_account
5042 .insert(account_id.get_issuer(), account_id);
5043 match self.accounts.get(&account_id) {
5044 Some(account_cell) => *account_cell.borrow_mut() = account,
5045 None => {
5046 self.accounts.insert(account_id, SharedCell::new(account));
5047 }
5048 }
5049 }
5050
5051 pub fn update_account_owned(&mut self, account: AccountAny) -> anyhow::Result<()> {
5057 let account_id = account.id();
5058 self.cache_account_owned(account);
5059
5060 if let Some(database) = &mut self.database {
5061 let Some(account_cell) = self.accounts.get(&account_id) else {
5062 anyhow::bail!("Account {account_id} not found after cache update");
5063 };
5064 database.update_account(&account_cell.borrow())?;
5065 }
5066 Ok(())
5067 }
5068
5069 pub fn update_account_state(&mut self, event: &AccountState) -> anyhow::Result<()> {
5079 let Some(cell) = self.accounts.get(&event.account_id) else {
5080 return self.add_account(AccountAny::from_events(std::slice::from_ref(event))?);
5081 };
5082
5083 cell.borrow_mut().apply(event.clone())?;
5084
5085 if let Some(database) = &mut self.database {
5086 database.update_account(&cell.borrow())?;
5087 }
5088 Ok(())
5089 }
5090
5091 pub fn replace_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
5102 let client_order_id = order.client_order_id();
5103 if let Some(venue_order_id) = order.venue_order_id() {
5104 self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
5105 }
5106
5107 match self.orders.get(&client_order_id) {
5108 Some(order_cell) => *order_cell.borrow_mut() = order.clone(),
5111 None => {
5112 self.orders
5113 .insert(client_order_id, SharedCell::new(order.clone()));
5114 }
5115 }
5116
5117 self.refresh_order(order)
5118 }
5119
5120 pub fn update_order(&mut self, event: &OrderEventAny) -> anyhow::Result<OrderAny> {
5126 let event_client_order_id = event.client_order_id();
5127 let client_order_id = if self.order_exists(&event_client_order_id) {
5128 event_client_order_id
5129 } else if let Some(venue_order_id) = event.venue_order_id() {
5130 self.index
5131 .venue_order_ids
5132 .get(&venue_order_id)
5133 .copied()
5134 .ok_or(OrderError::NotFound(event_client_order_id))?
5135 } else {
5136 return Err(OrderError::NotFound(event_client_order_id).into());
5137 };
5138
5139 let order_cell = self
5140 .orders
5141 .get(&client_order_id)
5142 .cloned()
5143 .ok_or(OrderError::NotFound(client_order_id))?;
5144
5145 let mut snapshot = order_cell.borrow().clone();
5149 snapshot.apply(event.clone())?;
5150
5151 if let Some(venue_order_id) = snapshot.venue_order_id() {
5155 self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
5156 }
5157
5158 *order_cell.borrow_mut() = snapshot.clone();
5159
5160 if let Err(e) = self.refresh_order(&snapshot) {
5161 log::error!("Error updating order in cache: {e}");
5162 }
5163
5164 Ok(snapshot)
5165 }
5166
5167 fn refresh_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
5168 let client_order_id = order.client_order_id();
5169
5170 if let Some(venue_order_id) = order.venue_order_id() {
5173 let overwrite = matches!(order.last_event(), OrderEventAny::Updated(_));
5174 if let Err(e) = self.add_venue_order_id(&client_order_id, &venue_order_id, overwrite) {
5175 if e.is::<VenueOrderIdOwnershipError>() {
5176 return Err(e);
5177 }
5178 log::error!("Error indexing venue order ID in cache: {e}");
5179 }
5180 }
5181
5182 if order.is_active_local() {
5183 self.index.orders_active_local.insert(client_order_id);
5184 } else {
5185 self.index.orders_active_local.remove(&client_order_id);
5186 }
5187
5188 if order.is_inflight() {
5190 self.index.orders_inflight.insert(client_order_id);
5191 } else {
5192 self.index.orders_inflight.remove(&client_order_id);
5193 }
5194
5195 if order.is_open() {
5197 self.index.orders_closed.remove(&client_order_id);
5198 self.index.orders_open.insert(client_order_id);
5199 } else if order.is_closed() {
5200 self.index.orders_open.remove(&client_order_id);
5201 self.index.orders_pending_cancel.remove(&client_order_id);
5202 self.index.orders_closed.insert(client_order_id);
5203 }
5204
5205 if matches!(order.last_event(), OrderEventAny::CancelRejected(_)) {
5207 self.index.orders_pending_cancel.remove(&client_order_id);
5208 }
5209
5210 if order.emulation_trigger().is_some() && !order.is_closed() {
5212 self.index.orders_emulated.insert(client_order_id);
5213 } else {
5214 self.index.orders_emulated.remove(&client_order_id);
5215 }
5216
5217 if let Some(account_id) = order.account_id() {
5219 self.index
5220 .account_orders
5221 .entry(account_id)
5222 .or_default()
5223 .insert(client_order_id);
5224 }
5225
5226 if !self.own_books.is_empty() {
5228 let own_book = self.own_order_book(&order.instrument_id());
5229 if (own_book.is_some() && order.is_closed()) || should_handle_own_book_order(order) {
5230 self.update_own_order_book(order);
5231 }
5232 }
5233
5234 if let Some(database) = &mut self.database {
5235 database.update_order(order.last_event())?;
5236 }
5241
5242 Ok(())
5243 }
5244
5245 pub fn update_order_pending_cancel_local(&mut self, order: &OrderAny) {
5247 self.index
5248 .orders_pending_cancel
5249 .insert(order.client_order_id());
5250 }
5251
5252 pub fn update_position(&mut self, position: &Position) -> anyhow::Result<()> {
5262 let Some(position_cell) = self.positions.get(&position.id).cloned() else {
5263 anyhow::bail!("Cannot update position {}: not found in cache", position.id);
5264 };
5265
5266 self.refresh_position_indexes(position);
5267
5268 *position_cell.borrow_mut() = position.clone();
5269
5270 if let Some(database) = &mut self.database {
5271 database.update_position(position)?;
5272 }
5277
5278 Ok(())
5279 }
5280
5281 pub fn update_position_from_fill(
5291 &mut self,
5292 position_id: PositionId,
5293 fill: &OrderFilled,
5294 ) -> anyhow::Result<Position> {
5295 let Some(position_cell) = self.positions.get(&position_id).cloned() else {
5296 anyhow::bail!("Cannot update position {position_id}: not found in cache");
5297 };
5298
5299 let position = {
5300 let mut position = position_cell.borrow_mut();
5301 position.apply(fill);
5302 position.clone_without_events()
5303 };
5304
5305 self.refresh_position_indexes(&position);
5306
5307 if let Some(database) = &mut self.database {
5308 database.update_position(&position_cell.borrow())?;
5309 }
5310
5311 Ok(position)
5312 }
5313
5314 fn refresh_position_indexes(&mut self, position: &Position) {
5315 if position.is_open() {
5316 self.index.positions_open.insert(position.id);
5317 self.index.positions_closed.remove(&position.id);
5318 } else {
5319 self.index.positions_closed.insert(position.id);
5320 self.index.positions_open.remove(&position.id);
5321 }
5322 }
5323
5324 #[must_use]
5326 pub fn oms_type(&self, position_id: &PositionId) -> Option<OmsType> {
5327 self.index.position_oms.get(position_id).copied()
5328 }
5329
5330 pub fn snapshot_order_state(&self, order: &OrderAny) -> anyhow::Result<()> {
5336 let Some(database) = &self.database else {
5337 log::warn!(
5338 "Cannot snapshot order state for {} (no database configured)",
5339 order.client_order_id()
5340 );
5341 return Ok(());
5342 };
5343
5344 database.snapshot_order_state(order)
5345 }
5346
5347 fn collect_order_filter_sources<'a>(
5358 &'a self,
5359 venue: Option<&Venue>,
5360 instrument_id: Option<&InstrumentId>,
5361 strategy_id: Option<&StrategyId>,
5362 account_id: Option<&AccountId>,
5363 ) -> FilterSources<'a, ClientOrderId> {
5364 let mut sources: Vec<&AHashSet<ClientOrderId>> = Vec::with_capacity(4);
5365
5366 if let Some(venue) = venue {
5367 match self.index.venue_orders.get(venue) {
5368 Some(set) => sources.push(set),
5369 None => return FilterSources::Empty,
5370 }
5371 }
5372
5373 if let Some(instrument_id) = instrument_id {
5374 match self.index.instrument_orders.get(instrument_id) {
5375 Some(set) => sources.push(set),
5376 None => return FilterSources::Empty,
5377 }
5378 }
5379
5380 if let Some(strategy_id) = strategy_id {
5381 match self.index.strategy_orders.get(strategy_id) {
5382 Some(set) => sources.push(set),
5383 None => return FilterSources::Empty,
5384 }
5385 }
5386
5387 if let Some(account_id) = account_id {
5388 match self.index.account_orders.get(account_id) {
5389 Some(set) => sources.push(set),
5390 None => return FilterSources::Empty,
5391 }
5392 }
5393
5394 if sources.is_empty() {
5395 FilterSources::Unfiltered
5396 } else {
5397 FilterSources::Sets(sources)
5398 }
5399 }
5400
5401 fn collect_position_filter_sources<'a>(
5402 &'a self,
5403 venue: Option<&Venue>,
5404 instrument_id: Option<&InstrumentId>,
5405 strategy_id: Option<&StrategyId>,
5406 account_id: Option<&AccountId>,
5407 ) -> FilterSources<'a, PositionId> {
5408 let mut sources: Vec<&AHashSet<PositionId>> = Vec::with_capacity(4);
5409
5410 if let Some(venue) = venue {
5411 match self.index.venue_positions.get(venue) {
5412 Some(set) => sources.push(set),
5413 None => return FilterSources::Empty,
5414 }
5415 }
5416
5417 if let Some(instrument_id) = instrument_id {
5418 match self.index.instrument_positions.get(instrument_id) {
5419 Some(set) => sources.push(set),
5420 None => return FilterSources::Empty,
5421 }
5422 }
5423
5424 if let Some(strategy_id) = strategy_id {
5425 match self.index.strategy_positions.get(strategy_id) {
5426 Some(set) => sources.push(set),
5427 None => return FilterSources::Empty,
5428 }
5429 }
5430
5431 if let Some(account_id) = account_id {
5432 match self.index.account_positions.get(account_id) {
5433 Some(set) => sources.push(set),
5434 None => return FilterSources::Empty,
5435 }
5436 }
5437
5438 if sources.is_empty() {
5439 FilterSources::Unfiltered
5440 } else {
5441 FilterSources::Sets(sources)
5442 }
5443 }
5444
5445 fn query_orders_in_bucket(
5451 &self,
5452 bucket: &AHashSet<ClientOrderId>,
5453 venue: Option<&Venue>,
5454 instrument_id: Option<&InstrumentId>,
5455 strategy_id: Option<&StrategyId>,
5456 account_id: Option<&AccountId>,
5457 ) -> AHashSet<ClientOrderId> {
5458 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5459 FilterSources::Empty => AHashSet::new(),
5460 FilterSources::Unfiltered => bucket.clone(),
5461 FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
5462 }
5463 }
5464
5465 fn query_positions_in_bucket(
5466 &self,
5467 bucket: &AHashSet<PositionId>,
5468 venue: Option<&Venue>,
5469 instrument_id: Option<&InstrumentId>,
5470 strategy_id: Option<&StrategyId>,
5471 account_id: Option<&AccountId>,
5472 ) -> AHashSet<PositionId> {
5473 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5474 FilterSources::Empty => AHashSet::new(),
5475 FilterSources::Unfiltered => bucket.clone(),
5476 FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
5477 }
5478 }
5479
5480 fn view_orders_in_bucket<'a>(
5483 &'a self,
5484 bucket: &'a AHashSet<ClientOrderId>,
5485 venue: Option<&Venue>,
5486 instrument_id: Option<&InstrumentId>,
5487 strategy_id: Option<&StrategyId>,
5488 account_id: Option<&AccountId>,
5489 ) -> Cow<'a, AHashSet<ClientOrderId>> {
5490 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5491 FilterSources::Empty => Cow::Owned(AHashSet::new()),
5492 FilterSources::Unfiltered => Cow::Borrowed(bucket),
5493 FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
5494 }
5495 }
5496
5497 fn view_positions_in_bucket<'a>(
5498 &'a self,
5499 bucket: &'a AHashSet<PositionId>,
5500 venue: Option<&Venue>,
5501 instrument_id: Option<&InstrumentId>,
5502 strategy_id: Option<&StrategyId>,
5503 account_id: Option<&AccountId>,
5504 ) -> Cow<'a, AHashSet<PositionId>> {
5505 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5506 FilterSources::Empty => Cow::Owned(AHashSet::new()),
5507 FilterSources::Unfiltered => Cow::Borrowed(bucket),
5508 FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
5509 }
5510 }
5511
5512 fn iter_orders_in_bucket<'a>(
5517 &'a self,
5518 bucket: &'a AHashSet<ClientOrderId>,
5519 venue: Option<&Venue>,
5520 instrument_id: Option<&InstrumentId>,
5521 strategy_id: Option<&StrategyId>,
5522 account_id: Option<&AccountId>,
5523 ) -> Box<dyn Iterator<Item = ClientOrderId> + 'a> {
5524 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5525 FilterSources::Empty => Box::new(std::iter::empty()),
5526 FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
5527 FilterSources::Sets(mut sources) => {
5528 sources.push(bucket);
5529 sources.sort_unstable_by_key(|s| s.len());
5530 let driver = sources[0];
5531 let rest: Vec<&'a AHashSet<ClientOrderId>> = sources[1..].to_vec();
5532 Box::new(
5533 driver
5534 .iter()
5535 .copied()
5536 .filter(move |id| rest.iter().all(|s| s.contains(id))),
5537 )
5538 }
5539 }
5540 }
5541
5542 fn iter_positions_in_bucket<'a>(
5543 &'a self,
5544 bucket: &'a AHashSet<PositionId>,
5545 venue: Option<&Venue>,
5546 instrument_id: Option<&InstrumentId>,
5547 strategy_id: Option<&StrategyId>,
5548 account_id: Option<&AccountId>,
5549 ) -> Box<dyn Iterator<Item = PositionId> + 'a> {
5550 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5551 FilterSources::Empty => Box::new(std::iter::empty()),
5552 FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
5553 FilterSources::Sets(mut sources) => {
5554 sources.push(bucket);
5555 sources.sort_unstable_by_key(|s| s.len());
5556 let driver = sources[0];
5557 let rest: Vec<&'a AHashSet<PositionId>> = sources[1..].to_vec();
5558 Box::new(
5559 driver
5560 .iter()
5561 .copied()
5562 .filter(move |id| rest.iter().all(|s| s.contains(id))),
5563 )
5564 }
5565 }
5566 }
5567
5568 fn count_orders_in_bucket(
5574 &self,
5575 bucket: &AHashSet<ClientOrderId>,
5576 venue: Option<&Venue>,
5577 instrument_id: Option<&InstrumentId>,
5578 strategy_id: Option<&StrategyId>,
5579 account_id: Option<&AccountId>,
5580 side: Option<OrderSide>,
5581 ) -> usize {
5582 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5583 FilterSources::Empty => 0,
5584 FilterSources::Unfiltered => side.map_or_else(
5585 || bucket.len(),
5586 |side| {
5587 bucket
5588 .iter()
5589 .filter(|id| self.order_side_matches(id, side))
5590 .count()
5591 },
5592 ),
5593 FilterSources::Sets(mut sources) => {
5594 sources.push(bucket);
5595 sources.sort_unstable_by_key(|s| s.len());
5596 let driver = sources[0];
5597 let rest = &sources[1..];
5598
5599 driver
5600 .iter()
5601 .filter(|id| rest.iter().all(|s| s.contains(id)))
5602 .filter(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
5603 .count()
5604 }
5605 }
5606 }
5607
5608 fn count_positions_in_bucket(
5609 &self,
5610 bucket: &AHashSet<PositionId>,
5611 venue: Option<&Venue>,
5612 instrument_id: Option<&InstrumentId>,
5613 strategy_id: Option<&StrategyId>,
5614 account_id: Option<&AccountId>,
5615 side: Option<PositionSide>,
5616 ) -> usize {
5617 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5618 FilterSources::Empty => 0,
5619 FilterSources::Unfiltered => side.map_or_else(
5620 || bucket.len(),
5621 |side| {
5622 bucket
5623 .iter()
5624 .filter(|id| self.position_side_matches(id, side))
5625 .count()
5626 },
5627 ),
5628 FilterSources::Sets(mut sources) => {
5629 sources.push(bucket);
5630 sources.sort_unstable_by_key(|s| s.len());
5631 let driver = sources[0];
5632 let rest = &sources[1..];
5633
5634 driver
5635 .iter()
5636 .filter(|id| rest.iter().all(|s| s.contains(id)))
5637 .filter(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
5638 .count()
5639 }
5640 }
5641 }
5642
5643 fn any_orders_in_bucket(
5649 &self,
5650 bucket: &AHashSet<ClientOrderId>,
5651 venue: Option<&Venue>,
5652 instrument_id: Option<&InstrumentId>,
5653 strategy_id: Option<&StrategyId>,
5654 account_id: Option<&AccountId>,
5655 side: Option<OrderSide>,
5656 ) -> bool {
5657 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5658 FilterSources::Empty => false,
5659 FilterSources::Unfiltered => side.map_or_else(
5660 || !bucket.is_empty(),
5661 |side| bucket.iter().any(|id| self.order_side_matches(id, side)),
5662 ),
5663 FilterSources::Sets(mut sources) => {
5664 sources.push(bucket);
5665 sources.sort_unstable_by_key(|s| s.len());
5666 let driver = sources[0];
5667 let rest = &sources[1..];
5668
5669 driver
5670 .iter()
5671 .filter(|id| rest.iter().all(|s| s.contains(id)))
5672 .any(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
5673 }
5674 }
5675 }
5676
5677 fn any_positions_in_bucket(
5678 &self,
5679 bucket: &AHashSet<PositionId>,
5680 venue: Option<&Venue>,
5681 instrument_id: Option<&InstrumentId>,
5682 strategy_id: Option<&StrategyId>,
5683 account_id: Option<&AccountId>,
5684 side: Option<PositionSide>,
5685 ) -> bool {
5686 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5687 FilterSources::Empty => false,
5688 FilterSources::Unfiltered => side.map_or_else(
5689 || !bucket.is_empty(),
5690 |side| bucket.iter().any(|id| self.position_side_matches(id, side)),
5691 ),
5692 FilterSources::Sets(mut sources) => {
5693 sources.push(bucket);
5694 sources.sort_unstable_by_key(|s| s.len());
5695 let driver = sources[0];
5696 let rest = &sources[1..];
5697
5698 driver
5699 .iter()
5700 .filter(|id| rest.iter().all(|s| s.contains(id)))
5701 .any(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
5702 }
5703 }
5704 }
5705
5706 fn order_side_matches(&self, client_order_id: &ClientOrderId, side: OrderSide) -> bool {
5707 self.orders
5708 .get(client_order_id)
5709 .is_some_and(|cell| cell.borrow().order_side() == side)
5710 }
5711
5712 fn position_side_matches(&self, position_id: &PositionId, side: PositionSide) -> bool {
5713 self.positions
5714 .get(position_id)
5715 .is_some_and(|cell| cell.borrow().side == side)
5716 }
5717
5718 fn get_orders_for_ids(
5724 &self,
5725 client_order_ids: &AHashSet<ClientOrderId>,
5726 side: Option<OrderSide>,
5727 ) -> Vec<OrderRef<'_>> {
5728 let mut orders = Vec::new();
5729
5730 for client_order_id in client_order_ids {
5731 let order_cell = self
5732 .orders
5733 .get(client_order_id)
5734 .unwrap_or_else(|| panic!("Order {client_order_id} not found"));
5735 let order = OrderRef::new(order_cell.borrow());
5736
5737 if side.is_none_or(|side| side == order.order_side()) {
5738 orders.push(order);
5739 }
5740 }
5741
5742 orders.sort_by_key(|o| o.client_order_id());
5745 orders
5746 }
5747
5748 fn get_positions_for_ids(
5758 &self,
5759 position_ids: &AHashSet<PositionId>,
5760 side: Option<PositionSide>,
5761 ) -> Vec<PositionRef<'_>> {
5762 let mut positions = Vec::new();
5763
5764 for position_id in position_ids {
5765 let position_cell = self
5766 .positions
5767 .get(position_id)
5768 .unwrap_or_else(|| panic!("Position {position_id} not found"));
5769 let position = PositionRef::new(position_cell.borrow());
5770
5771 if side.is_none_or(|side| side == position.side) {
5772 positions.push(position);
5773 }
5774 }
5775
5776 positions.sort_by_key(|p| p.id);
5779 positions
5780 }
5781
5782 #[must_use]
5784 pub fn client_order_ids(
5785 &self,
5786 venue: Option<&Venue>,
5787 instrument_id: Option<&InstrumentId>,
5788 strategy_id: Option<&StrategyId>,
5789 account_id: Option<&AccountId>,
5790 ) -> AHashSet<ClientOrderId> {
5791 self.query_orders_in_bucket(
5792 &self.index.orders,
5793 venue,
5794 instrument_id,
5795 strategy_id,
5796 account_id,
5797 )
5798 }
5799
5800 #[must_use]
5802 pub fn client_order_ids_open(
5803 &self,
5804 venue: Option<&Venue>,
5805 instrument_id: Option<&InstrumentId>,
5806 strategy_id: Option<&StrategyId>,
5807 account_id: Option<&AccountId>,
5808 ) -> AHashSet<ClientOrderId> {
5809 self.query_orders_in_bucket(
5810 &self.index.orders_open,
5811 venue,
5812 instrument_id,
5813 strategy_id,
5814 account_id,
5815 )
5816 }
5817
5818 #[must_use]
5820 pub fn client_order_ids_closed(
5821 &self,
5822 venue: Option<&Venue>,
5823 instrument_id: Option<&InstrumentId>,
5824 strategy_id: Option<&StrategyId>,
5825 account_id: Option<&AccountId>,
5826 ) -> AHashSet<ClientOrderId> {
5827 self.query_orders_in_bucket(
5828 &self.index.orders_closed,
5829 venue,
5830 instrument_id,
5831 strategy_id,
5832 account_id,
5833 )
5834 }
5835
5836 #[must_use]
5841 pub fn client_order_ids_active_local(
5842 &self,
5843 venue: Option<&Venue>,
5844 instrument_id: Option<&InstrumentId>,
5845 strategy_id: Option<&StrategyId>,
5846 account_id: Option<&AccountId>,
5847 ) -> AHashSet<ClientOrderId> {
5848 self.query_orders_in_bucket(
5849 &self.index.orders_active_local,
5850 venue,
5851 instrument_id,
5852 strategy_id,
5853 account_id,
5854 )
5855 }
5856
5857 #[must_use]
5859 pub fn client_order_ids_emulated(
5860 &self,
5861 venue: Option<&Venue>,
5862 instrument_id: Option<&InstrumentId>,
5863 strategy_id: Option<&StrategyId>,
5864 account_id: Option<&AccountId>,
5865 ) -> AHashSet<ClientOrderId> {
5866 self.query_orders_in_bucket(
5867 &self.index.orders_emulated,
5868 venue,
5869 instrument_id,
5870 strategy_id,
5871 account_id,
5872 )
5873 }
5874
5875 #[must_use]
5877 pub fn client_order_ids_inflight(
5878 &self,
5879 venue: Option<&Venue>,
5880 instrument_id: Option<&InstrumentId>,
5881 strategy_id: Option<&StrategyId>,
5882 account_id: Option<&AccountId>,
5883 ) -> AHashSet<ClientOrderId> {
5884 self.query_orders_in_bucket(
5885 &self.index.orders_inflight,
5886 venue,
5887 instrument_id,
5888 strategy_id,
5889 account_id,
5890 )
5891 }
5892
5893 #[must_use]
5895 pub fn position_ids(
5896 &self,
5897 venue: Option<&Venue>,
5898 instrument_id: Option<&InstrumentId>,
5899 strategy_id: Option<&StrategyId>,
5900 account_id: Option<&AccountId>,
5901 ) -> AHashSet<PositionId> {
5902 self.query_positions_in_bucket(
5903 &self.index.positions,
5904 venue,
5905 instrument_id,
5906 strategy_id,
5907 account_id,
5908 )
5909 }
5910
5911 #[must_use]
5913 pub fn position_open_ids(
5914 &self,
5915 venue: Option<&Venue>,
5916 instrument_id: Option<&InstrumentId>,
5917 strategy_id: Option<&StrategyId>,
5918 account_id: Option<&AccountId>,
5919 ) -> AHashSet<PositionId> {
5920 self.query_positions_in_bucket(
5921 &self.index.positions_open,
5922 venue,
5923 instrument_id,
5924 strategy_id,
5925 account_id,
5926 )
5927 }
5928
5929 #[must_use]
5931 pub fn position_closed_ids(
5932 &self,
5933 venue: Option<&Venue>,
5934 instrument_id: Option<&InstrumentId>,
5935 strategy_id: Option<&StrategyId>,
5936 account_id: Option<&AccountId>,
5937 ) -> AHashSet<PositionId> {
5938 self.query_positions_in_bucket(
5939 &self.index.positions_closed,
5940 venue,
5941 instrument_id,
5942 strategy_id,
5943 account_id,
5944 )
5945 }
5946
5947 #[must_use]
5954 pub fn client_order_ids_view(
5955 &self,
5956 venue: Option<&Venue>,
5957 instrument_id: Option<&InstrumentId>,
5958 strategy_id: Option<&StrategyId>,
5959 account_id: Option<&AccountId>,
5960 ) -> Cow<'_, AHashSet<ClientOrderId>> {
5961 self.view_orders_in_bucket(
5962 &self.index.orders,
5963 venue,
5964 instrument_id,
5965 strategy_id,
5966 account_id,
5967 )
5968 }
5969
5970 #[must_use]
5972 pub fn client_order_ids_open_view(
5973 &self,
5974 venue: Option<&Venue>,
5975 instrument_id: Option<&InstrumentId>,
5976 strategy_id: Option<&StrategyId>,
5977 account_id: Option<&AccountId>,
5978 ) -> Cow<'_, AHashSet<ClientOrderId>> {
5979 self.view_orders_in_bucket(
5980 &self.index.orders_open,
5981 venue,
5982 instrument_id,
5983 strategy_id,
5984 account_id,
5985 )
5986 }
5987
5988 #[must_use]
5990 pub fn client_order_ids_closed_view(
5991 &self,
5992 venue: Option<&Venue>,
5993 instrument_id: Option<&InstrumentId>,
5994 strategy_id: Option<&StrategyId>,
5995 account_id: Option<&AccountId>,
5996 ) -> Cow<'_, AHashSet<ClientOrderId>> {
5997 self.view_orders_in_bucket(
5998 &self.index.orders_closed,
5999 venue,
6000 instrument_id,
6001 strategy_id,
6002 account_id,
6003 )
6004 }
6005
6006 #[must_use]
6008 pub fn client_order_ids_active_local_view(
6009 &self,
6010 venue: Option<&Venue>,
6011 instrument_id: Option<&InstrumentId>,
6012 strategy_id: Option<&StrategyId>,
6013 account_id: Option<&AccountId>,
6014 ) -> Cow<'_, AHashSet<ClientOrderId>> {
6015 self.view_orders_in_bucket(
6016 &self.index.orders_active_local,
6017 venue,
6018 instrument_id,
6019 strategy_id,
6020 account_id,
6021 )
6022 }
6023
6024 #[must_use]
6026 pub fn client_order_ids_emulated_view(
6027 &self,
6028 venue: Option<&Venue>,
6029 instrument_id: Option<&InstrumentId>,
6030 strategy_id: Option<&StrategyId>,
6031 account_id: Option<&AccountId>,
6032 ) -> Cow<'_, AHashSet<ClientOrderId>> {
6033 self.view_orders_in_bucket(
6034 &self.index.orders_emulated,
6035 venue,
6036 instrument_id,
6037 strategy_id,
6038 account_id,
6039 )
6040 }
6041
6042 #[must_use]
6044 pub fn client_order_ids_inflight_view(
6045 &self,
6046 venue: Option<&Venue>,
6047 instrument_id: Option<&InstrumentId>,
6048 strategy_id: Option<&StrategyId>,
6049 account_id: Option<&AccountId>,
6050 ) -> Cow<'_, AHashSet<ClientOrderId>> {
6051 self.view_orders_in_bucket(
6052 &self.index.orders_inflight,
6053 venue,
6054 instrument_id,
6055 strategy_id,
6056 account_id,
6057 )
6058 }
6059
6060 #[must_use]
6062 pub fn position_ids_view(
6063 &self,
6064 venue: Option<&Venue>,
6065 instrument_id: Option<&InstrumentId>,
6066 strategy_id: Option<&StrategyId>,
6067 account_id: Option<&AccountId>,
6068 ) -> Cow<'_, AHashSet<PositionId>> {
6069 self.view_positions_in_bucket(
6070 &self.index.positions,
6071 venue,
6072 instrument_id,
6073 strategy_id,
6074 account_id,
6075 )
6076 }
6077
6078 #[must_use]
6080 pub fn position_open_ids_view(
6081 &self,
6082 venue: Option<&Venue>,
6083 instrument_id: Option<&InstrumentId>,
6084 strategy_id: Option<&StrategyId>,
6085 account_id: Option<&AccountId>,
6086 ) -> Cow<'_, AHashSet<PositionId>> {
6087 self.view_positions_in_bucket(
6088 &self.index.positions_open,
6089 venue,
6090 instrument_id,
6091 strategy_id,
6092 account_id,
6093 )
6094 }
6095
6096 #[must_use]
6098 pub fn position_closed_ids_view(
6099 &self,
6100 venue: Option<&Venue>,
6101 instrument_id: Option<&InstrumentId>,
6102 strategy_id: Option<&StrategyId>,
6103 account_id: Option<&AccountId>,
6104 ) -> Cow<'_, AHashSet<PositionId>> {
6105 self.view_positions_in_bucket(
6106 &self.index.positions_closed,
6107 venue,
6108 instrument_id,
6109 strategy_id,
6110 account_id,
6111 )
6112 }
6113
6114 pub fn iter_client_order_ids(
6120 &self,
6121 venue: Option<&Venue>,
6122 instrument_id: Option<&InstrumentId>,
6123 strategy_id: Option<&StrategyId>,
6124 account_id: Option<&AccountId>,
6125 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6126 self.iter_orders_in_bucket(
6127 &self.index.orders,
6128 venue,
6129 instrument_id,
6130 strategy_id,
6131 account_id,
6132 )
6133 }
6134
6135 pub fn iter_client_order_ids_open(
6137 &self,
6138 venue: Option<&Venue>,
6139 instrument_id: Option<&InstrumentId>,
6140 strategy_id: Option<&StrategyId>,
6141 account_id: Option<&AccountId>,
6142 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6143 self.iter_orders_in_bucket(
6144 &self.index.orders_open,
6145 venue,
6146 instrument_id,
6147 strategy_id,
6148 account_id,
6149 )
6150 }
6151
6152 pub fn iter_client_order_ids_closed(
6154 &self,
6155 venue: Option<&Venue>,
6156 instrument_id: Option<&InstrumentId>,
6157 strategy_id: Option<&StrategyId>,
6158 account_id: Option<&AccountId>,
6159 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6160 self.iter_orders_in_bucket(
6161 &self.index.orders_closed,
6162 venue,
6163 instrument_id,
6164 strategy_id,
6165 account_id,
6166 )
6167 }
6168
6169 pub fn iter_client_order_ids_active_local(
6171 &self,
6172 venue: Option<&Venue>,
6173 instrument_id: Option<&InstrumentId>,
6174 strategy_id: Option<&StrategyId>,
6175 account_id: Option<&AccountId>,
6176 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6177 self.iter_orders_in_bucket(
6178 &self.index.orders_active_local,
6179 venue,
6180 instrument_id,
6181 strategy_id,
6182 account_id,
6183 )
6184 }
6185
6186 pub fn iter_client_order_ids_emulated(
6188 &self,
6189 venue: Option<&Venue>,
6190 instrument_id: Option<&InstrumentId>,
6191 strategy_id: Option<&StrategyId>,
6192 account_id: Option<&AccountId>,
6193 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6194 self.iter_orders_in_bucket(
6195 &self.index.orders_emulated,
6196 venue,
6197 instrument_id,
6198 strategy_id,
6199 account_id,
6200 )
6201 }
6202
6203 pub fn iter_client_order_ids_inflight(
6205 &self,
6206 venue: Option<&Venue>,
6207 instrument_id: Option<&InstrumentId>,
6208 strategy_id: Option<&StrategyId>,
6209 account_id: Option<&AccountId>,
6210 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6211 self.iter_orders_in_bucket(
6212 &self.index.orders_inflight,
6213 venue,
6214 instrument_id,
6215 strategy_id,
6216 account_id,
6217 )
6218 }
6219
6220 pub fn iter_position_ids(
6222 &self,
6223 venue: Option<&Venue>,
6224 instrument_id: Option<&InstrumentId>,
6225 strategy_id: Option<&StrategyId>,
6226 account_id: Option<&AccountId>,
6227 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
6228 self.iter_positions_in_bucket(
6229 &self.index.positions,
6230 venue,
6231 instrument_id,
6232 strategy_id,
6233 account_id,
6234 )
6235 }
6236
6237 pub fn iter_position_open_ids(
6239 &self,
6240 venue: Option<&Venue>,
6241 instrument_id: Option<&InstrumentId>,
6242 strategy_id: Option<&StrategyId>,
6243 account_id: Option<&AccountId>,
6244 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
6245 self.iter_positions_in_bucket(
6246 &self.index.positions_open,
6247 venue,
6248 instrument_id,
6249 strategy_id,
6250 account_id,
6251 )
6252 }
6253
6254 pub fn iter_position_closed_ids(
6256 &self,
6257 venue: Option<&Venue>,
6258 instrument_id: Option<&InstrumentId>,
6259 strategy_id: Option<&StrategyId>,
6260 account_id: Option<&AccountId>,
6261 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
6262 self.iter_positions_in_bucket(
6263 &self.index.positions_closed,
6264 venue,
6265 instrument_id,
6266 strategy_id,
6267 account_id,
6268 )
6269 }
6270
6271 #[must_use]
6273 pub fn strategy_ids(&self) -> AHashSet<StrategyId> {
6274 self.index.strategies.clone()
6275 }
6276
6277 #[must_use]
6279 pub fn exec_algorithm_ids(&self) -> AHashSet<ExecAlgorithmId> {
6280 self.index.exec_algorithms.clone()
6281 }
6282
6283 #[must_use]
6292 pub fn order_ref(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
6293 self.orders
6294 .get(client_order_id)
6295 .map(|order_cell| OrderRef::new(order_cell.borrow()))
6296 }
6297
6298 #[must_use]
6302 pub fn order(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
6303 self.order_ref(client_order_id)
6304 }
6305
6306 pub fn try_order_ref(
6312 &self,
6313 client_order_id: &ClientOrderId,
6314 ) -> Result<OrderRef<'_>, OrderLookupError> {
6315 self.orders
6316 .get(client_order_id)
6317 .map(|order_cell| OrderRef::new(order_cell.borrow()))
6318 .ok_or_else(|| OrderLookupError::not_found(*client_order_id))
6319 }
6320
6321 pub fn try_order(
6329 &self,
6330 client_order_id: &ClientOrderId,
6331 ) -> Result<OrderRef<'_>, OrderLookupError> {
6332 self.try_order_ref(client_order_id)
6333 }
6334
6335 #[must_use]
6345 pub fn order_mut(&mut self, client_order_id: &ClientOrderId) -> Option<OrderRefMut<'_>> {
6346 self.orders
6347 .get(client_order_id)
6348 .map(|order_cell| OrderRefMut::new(order_cell.borrow_mut()))
6349 }
6350
6351 #[must_use]
6356 pub fn order_owned(&self, client_order_id: &ClientOrderId) -> Option<OrderAny> {
6357 self.orders
6358 .get(client_order_id)
6359 .map(|order_cell| order_cell.borrow().clone())
6360 }
6361
6362 pub fn try_order_owned(
6368 &self,
6369 client_order_id: &ClientOrderId,
6370 ) -> Result<OrderAny, OrderLookupError> {
6371 self.try_order_ref(client_order_id)
6372 .map(|order| order.cloned())
6373 }
6374
6375 #[must_use]
6377 pub fn orders_for_ids(
6378 &self,
6379 client_order_ids: &[ClientOrderId],
6380 context: &dyn Display,
6381 ) -> Vec<OrderAny> {
6382 let mut orders = Vec::with_capacity(client_order_ids.len());
6383 for id in client_order_ids {
6384 match self.orders.get(id) {
6385 Some(order_cell) => orders.push(order_cell.borrow().clone()),
6386 None => log::error!("Order {id} not found in cache for {context}"),
6387 }
6388 }
6389 orders
6390 }
6391
6392 #[must_use]
6394 pub fn client_order_id(&self, venue_order_id: &VenueOrderId) -> Option<&ClientOrderId> {
6395 self.index.venue_order_ids.get(venue_order_id)
6396 }
6397
6398 #[must_use]
6400 pub fn venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<&VenueOrderId> {
6401 self.index.client_order_ids.get(client_order_id)
6402 }
6403
6404 #[must_use]
6406 pub fn client_id(&self, client_order_id: &ClientOrderId) -> Option<&ClientId> {
6407 self.index.order_client.get(client_order_id)
6408 }
6409
6410 #[must_use]
6416 pub fn orders_refs(
6417 &self,
6418 venue: Option<&Venue>,
6419 instrument_id: Option<&InstrumentId>,
6420 strategy_id: Option<&StrategyId>,
6421 account_id: Option<&AccountId>,
6422 side: Option<OrderSide>,
6423 ) -> Vec<OrderRef<'_>> {
6424 let client_order_ids = self.client_order_ids(venue, instrument_id, strategy_id, account_id);
6425 self.get_orders_for_ids(&client_order_ids, side)
6426 }
6427
6428 #[must_use]
6432 pub fn orders(
6433 &self,
6434 venue: Option<&Venue>,
6435 instrument_id: Option<&InstrumentId>,
6436 strategy_id: Option<&StrategyId>,
6437 account_id: Option<&AccountId>,
6438 side: Option<OrderSide>,
6439 ) -> Vec<OrderRef<'_>> {
6440 self.orders_refs(venue, instrument_id, strategy_id, account_id, side)
6441 }
6442
6443 #[must_use]
6445 pub fn orders_open_refs(
6446 &self,
6447 venue: Option<&Venue>,
6448 instrument_id: Option<&InstrumentId>,
6449 strategy_id: Option<&StrategyId>,
6450 account_id: Option<&AccountId>,
6451 side: Option<OrderSide>,
6452 ) -> Vec<OrderRef<'_>> {
6453 let client_order_ids =
6454 self.client_order_ids_open(venue, instrument_id, strategy_id, account_id);
6455 self.get_orders_for_ids(&client_order_ids, side)
6456 }
6457
6458 #[must_use]
6462 pub fn orders_open(
6463 &self,
6464 venue: Option<&Venue>,
6465 instrument_id: Option<&InstrumentId>,
6466 strategy_id: Option<&StrategyId>,
6467 account_id: Option<&AccountId>,
6468 side: Option<OrderSide>,
6469 ) -> Vec<OrderRef<'_>> {
6470 self.orders_open_refs(venue, instrument_id, strategy_id, account_id, side)
6471 }
6472
6473 #[must_use]
6475 pub fn orders_closed_refs(
6476 &self,
6477 venue: Option<&Venue>,
6478 instrument_id: Option<&InstrumentId>,
6479 strategy_id: Option<&StrategyId>,
6480 account_id: Option<&AccountId>,
6481 side: Option<OrderSide>,
6482 ) -> Vec<OrderRef<'_>> {
6483 let client_order_ids =
6484 self.client_order_ids_closed(venue, instrument_id, strategy_id, account_id);
6485 self.get_orders_for_ids(&client_order_ids, side)
6486 }
6487
6488 #[must_use]
6492 pub fn orders_closed(
6493 &self,
6494 venue: Option<&Venue>,
6495 instrument_id: Option<&InstrumentId>,
6496 strategy_id: Option<&StrategyId>,
6497 account_id: Option<&AccountId>,
6498 side: Option<OrderSide>,
6499 ) -> Vec<OrderRef<'_>> {
6500 self.orders_closed_refs(venue, instrument_id, strategy_id, account_id, side)
6501 }
6502
6503 #[must_use]
6508 pub fn orders_active_local_refs(
6509 &self,
6510 venue: Option<&Venue>,
6511 instrument_id: Option<&InstrumentId>,
6512 strategy_id: Option<&StrategyId>,
6513 account_id: Option<&AccountId>,
6514 side: Option<OrderSide>,
6515 ) -> Vec<OrderRef<'_>> {
6516 let client_order_ids =
6517 self.client_order_ids_active_local(venue, instrument_id, strategy_id, account_id);
6518 self.get_orders_for_ids(&client_order_ids, side)
6519 }
6520
6521 #[must_use]
6525 pub fn orders_active_local(
6526 &self,
6527 venue: Option<&Venue>,
6528 instrument_id: Option<&InstrumentId>,
6529 strategy_id: Option<&StrategyId>,
6530 account_id: Option<&AccountId>,
6531 side: Option<OrderSide>,
6532 ) -> Vec<OrderRef<'_>> {
6533 self.orders_active_local_refs(venue, instrument_id, strategy_id, account_id, side)
6534 }
6535
6536 #[must_use]
6538 pub fn orders_emulated_refs(
6539 &self,
6540 venue: Option<&Venue>,
6541 instrument_id: Option<&InstrumentId>,
6542 strategy_id: Option<&StrategyId>,
6543 account_id: Option<&AccountId>,
6544 side: Option<OrderSide>,
6545 ) -> Vec<OrderRef<'_>> {
6546 let client_order_ids =
6547 self.client_order_ids_emulated(venue, instrument_id, strategy_id, account_id);
6548 self.get_orders_for_ids(&client_order_ids, side)
6549 }
6550
6551 #[must_use]
6555 pub fn orders_emulated(
6556 &self,
6557 venue: Option<&Venue>,
6558 instrument_id: Option<&InstrumentId>,
6559 strategy_id: Option<&StrategyId>,
6560 account_id: Option<&AccountId>,
6561 side: Option<OrderSide>,
6562 ) -> Vec<OrderRef<'_>> {
6563 self.orders_emulated_refs(venue, instrument_id, strategy_id, account_id, side)
6564 }
6565
6566 #[must_use]
6568 pub fn orders_inflight_refs(
6569 &self,
6570 venue: Option<&Venue>,
6571 instrument_id: Option<&InstrumentId>,
6572 strategy_id: Option<&StrategyId>,
6573 account_id: Option<&AccountId>,
6574 side: Option<OrderSide>,
6575 ) -> Vec<OrderRef<'_>> {
6576 let client_order_ids =
6577 self.client_order_ids_inflight(venue, instrument_id, strategy_id, account_id);
6578 self.get_orders_for_ids(&client_order_ids, side)
6579 }
6580
6581 #[must_use]
6585 pub fn orders_inflight(
6586 &self,
6587 venue: Option<&Venue>,
6588 instrument_id: Option<&InstrumentId>,
6589 strategy_id: Option<&StrategyId>,
6590 account_id: Option<&AccountId>,
6591 side: Option<OrderSide>,
6592 ) -> Vec<OrderRef<'_>> {
6593 self.orders_inflight_refs(venue, instrument_id, strategy_id, account_id, side)
6594 }
6595
6596 #[must_use]
6598 pub fn orders_for_position(&self, position_id: &PositionId) -> Vec<OrderRef<'_>> {
6599 match self.index.position_orders.get(position_id) {
6600 Some(client_order_ids) => self.get_orders_for_ids(client_order_ids, None),
6601 None => Vec::new(),
6602 }
6603 }
6604
6605 #[must_use]
6607 pub fn order_exists(&self, client_order_id: &ClientOrderId) -> bool {
6608 self.index.orders.contains(client_order_id)
6609 }
6610
6611 #[must_use]
6613 pub fn is_order_open(&self, client_order_id: &ClientOrderId) -> bool {
6614 self.index.orders_open.contains(client_order_id)
6615 }
6616
6617 #[must_use]
6619 pub fn is_order_closed(&self, client_order_id: &ClientOrderId) -> bool {
6620 self.index.orders_closed.contains(client_order_id)
6621 }
6622
6623 #[must_use]
6628 pub fn is_order_active_local(&self, client_order_id: &ClientOrderId) -> bool {
6629 self.index.orders_active_local.contains(client_order_id)
6630 }
6631
6632 #[must_use]
6634 pub fn is_order_emulated(&self, client_order_id: &ClientOrderId) -> bool {
6635 self.index.orders_emulated.contains(client_order_id)
6636 }
6637
6638 #[must_use]
6640 pub fn is_order_inflight(&self, client_order_id: &ClientOrderId) -> bool {
6641 self.index.orders_inflight.contains(client_order_id)
6642 }
6643
6644 #[must_use]
6646 pub fn is_order_pending_cancel_local(&self, client_order_id: &ClientOrderId) -> bool {
6647 self.index.orders_pending_cancel.contains(client_order_id)
6648 }
6649
6650 #[must_use]
6652 pub fn orders_open_count(
6653 &self,
6654 venue: Option<&Venue>,
6655 instrument_id: Option<&InstrumentId>,
6656 strategy_id: Option<&StrategyId>,
6657 account_id: Option<&AccountId>,
6658 side: Option<OrderSide>,
6659 ) -> usize {
6660 self.count_orders_in_bucket(
6661 &self.index.orders_open,
6662 venue,
6663 instrument_id,
6664 strategy_id,
6665 account_id,
6666 side,
6667 )
6668 }
6669
6670 #[must_use]
6672 pub fn orders_closed_count(
6673 &self,
6674 venue: Option<&Venue>,
6675 instrument_id: Option<&InstrumentId>,
6676 strategy_id: Option<&StrategyId>,
6677 account_id: Option<&AccountId>,
6678 side: Option<OrderSide>,
6679 ) -> usize {
6680 self.count_orders_in_bucket(
6681 &self.index.orders_closed,
6682 venue,
6683 instrument_id,
6684 strategy_id,
6685 account_id,
6686 side,
6687 )
6688 }
6689
6690 #[must_use]
6695 pub fn orders_active_local_count(
6696 &self,
6697 venue: Option<&Venue>,
6698 instrument_id: Option<&InstrumentId>,
6699 strategy_id: Option<&StrategyId>,
6700 account_id: Option<&AccountId>,
6701 side: Option<OrderSide>,
6702 ) -> usize {
6703 self.count_orders_in_bucket(
6704 &self.index.orders_active_local,
6705 venue,
6706 instrument_id,
6707 strategy_id,
6708 account_id,
6709 side,
6710 )
6711 }
6712
6713 #[must_use]
6715 pub fn orders_emulated_count(
6716 &self,
6717 venue: Option<&Venue>,
6718 instrument_id: Option<&InstrumentId>,
6719 strategy_id: Option<&StrategyId>,
6720 account_id: Option<&AccountId>,
6721 side: Option<OrderSide>,
6722 ) -> usize {
6723 self.count_orders_in_bucket(
6724 &self.index.orders_emulated,
6725 venue,
6726 instrument_id,
6727 strategy_id,
6728 account_id,
6729 side,
6730 )
6731 }
6732
6733 #[must_use]
6735 pub fn orders_inflight_count(
6736 &self,
6737 venue: Option<&Venue>,
6738 instrument_id: Option<&InstrumentId>,
6739 strategy_id: Option<&StrategyId>,
6740 account_id: Option<&AccountId>,
6741 side: Option<OrderSide>,
6742 ) -> usize {
6743 self.count_orders_in_bucket(
6744 &self.index.orders_inflight,
6745 venue,
6746 instrument_id,
6747 strategy_id,
6748 account_id,
6749 side,
6750 )
6751 }
6752
6753 #[must_use]
6755 pub fn orders_total_count(
6756 &self,
6757 venue: Option<&Venue>,
6758 instrument_id: Option<&InstrumentId>,
6759 strategy_id: Option<&StrategyId>,
6760 account_id: Option<&AccountId>,
6761 side: Option<OrderSide>,
6762 ) -> usize {
6763 self.count_orders_in_bucket(
6764 &self.index.orders,
6765 venue,
6766 instrument_id,
6767 strategy_id,
6768 account_id,
6769 side,
6770 )
6771 }
6772
6773 #[must_use]
6779 pub fn has_orders_open(
6780 &self,
6781 venue: Option<&Venue>,
6782 instrument_id: Option<&InstrumentId>,
6783 strategy_id: Option<&StrategyId>,
6784 account_id: Option<&AccountId>,
6785 side: Option<OrderSide>,
6786 ) -> bool {
6787 self.any_orders_in_bucket(
6788 &self.index.orders_open,
6789 venue,
6790 instrument_id,
6791 strategy_id,
6792 account_id,
6793 side,
6794 )
6795 }
6796
6797 #[must_use]
6799 pub fn has_orders_closed(
6800 &self,
6801 venue: Option<&Venue>,
6802 instrument_id: Option<&InstrumentId>,
6803 strategy_id: Option<&StrategyId>,
6804 account_id: Option<&AccountId>,
6805 side: Option<OrderSide>,
6806 ) -> bool {
6807 self.any_orders_in_bucket(
6808 &self.index.orders_closed,
6809 venue,
6810 instrument_id,
6811 strategy_id,
6812 account_id,
6813 side,
6814 )
6815 }
6816
6817 #[must_use]
6821 pub fn has_orders_active_local(
6822 &self,
6823 venue: Option<&Venue>,
6824 instrument_id: Option<&InstrumentId>,
6825 strategy_id: Option<&StrategyId>,
6826 account_id: Option<&AccountId>,
6827 side: Option<OrderSide>,
6828 ) -> bool {
6829 self.any_orders_in_bucket(
6830 &self.index.orders_active_local,
6831 venue,
6832 instrument_id,
6833 strategy_id,
6834 account_id,
6835 side,
6836 )
6837 }
6838
6839 #[must_use]
6841 pub fn has_orders_emulated(
6842 &self,
6843 venue: Option<&Venue>,
6844 instrument_id: Option<&InstrumentId>,
6845 strategy_id: Option<&StrategyId>,
6846 account_id: Option<&AccountId>,
6847 side: Option<OrderSide>,
6848 ) -> bool {
6849 self.any_orders_in_bucket(
6850 &self.index.orders_emulated,
6851 venue,
6852 instrument_id,
6853 strategy_id,
6854 account_id,
6855 side,
6856 )
6857 }
6858
6859 #[must_use]
6861 pub fn has_orders_inflight(
6862 &self,
6863 venue: Option<&Venue>,
6864 instrument_id: Option<&InstrumentId>,
6865 strategy_id: Option<&StrategyId>,
6866 account_id: Option<&AccountId>,
6867 side: Option<OrderSide>,
6868 ) -> bool {
6869 self.any_orders_in_bucket(
6870 &self.index.orders_inflight,
6871 venue,
6872 instrument_id,
6873 strategy_id,
6874 account_id,
6875 side,
6876 )
6877 }
6878
6879 #[must_use]
6881 pub fn has_orders(
6882 &self,
6883 venue: Option<&Venue>,
6884 instrument_id: Option<&InstrumentId>,
6885 strategy_id: Option<&StrategyId>,
6886 account_id: Option<&AccountId>,
6887 side: Option<OrderSide>,
6888 ) -> bool {
6889 self.any_orders_in_bucket(
6890 &self.index.orders,
6891 venue,
6892 instrument_id,
6893 strategy_id,
6894 account_id,
6895 side,
6896 )
6897 }
6898
6899 #[must_use]
6901 pub fn order_list(&self, order_list_id: &OrderListId) -> Option<&OrderList> {
6902 self.order_lists.get(order_list_id)
6903 }
6904
6905 pub fn try_order_list(
6911 &self,
6912 order_list_id: &OrderListId,
6913 ) -> Result<&OrderList, OrderListLookupError> {
6914 self.order_lists
6915 .get(order_list_id)
6916 .ok_or_else(|| OrderListLookupError::not_found(*order_list_id))
6917 }
6918
6919 #[must_use]
6921 pub fn order_lists(
6922 &self,
6923 venue: Option<&Venue>,
6924 instrument_id: Option<&InstrumentId>,
6925 strategy_id: Option<&StrategyId>,
6926 account_id: Option<&AccountId>,
6927 ) -> Vec<&OrderList> {
6928 let mut order_lists = self.order_lists.values().collect::<Vec<&OrderList>>();
6929
6930 if let Some(venue) = venue {
6931 order_lists.retain(|ol| &ol.instrument_id.venue == venue);
6932 }
6933
6934 if let Some(instrument_id) = instrument_id {
6935 order_lists.retain(|ol| &ol.instrument_id == instrument_id);
6936 }
6937
6938 if let Some(strategy_id) = strategy_id {
6939 order_lists.retain(|ol| &ol.strategy_id == strategy_id);
6940 }
6941
6942 if let Some(account_id) = account_id {
6943 order_lists.retain(|ol| {
6944 ol.client_order_ids.iter().any(|client_order_id| {
6945 self.orders.get(client_order_id).is_some_and(|order_cell| {
6946 order_cell.borrow().account_id().as_ref() == Some(account_id)
6947 })
6948 })
6949 });
6950 }
6951
6952 order_lists
6953 }
6954
6955 #[must_use]
6957 pub fn order_list_exists(&self, order_list_id: &OrderListId) -> bool {
6958 self.order_lists.contains_key(order_list_id)
6959 }
6960
6961 #[must_use]
6966 pub fn orders_for_exec_algorithm(
6967 &self,
6968 exec_algorithm_id: &ExecAlgorithmId,
6969 venue: Option<&Venue>,
6970 instrument_id: Option<&InstrumentId>,
6971 strategy_id: Option<&StrategyId>,
6972 account_id: Option<&AccountId>,
6973 side: Option<OrderSide>,
6974 ) -> Vec<OrderRef<'_>> {
6975 let Some(exec_algorithm_order_ids) =
6976 self.index.exec_algorithm_orders.get(exec_algorithm_id)
6977 else {
6978 return Vec::new();
6979 };
6980
6981 let filtered = self.query_orders_in_bucket(
6982 exec_algorithm_order_ids,
6983 venue,
6984 instrument_id,
6985 strategy_id,
6986 account_id,
6987 );
6988 self.get_orders_for_ids(&filtered, side)
6989 }
6990
6991 #[must_use]
6993 pub fn orders_for_exec_spawn(&self, exec_spawn_id: &ClientOrderId) -> Vec<OrderRef<'_>> {
6994 match self.index.exec_spawn_orders.get(exec_spawn_id) {
6995 Some(ids) => self.get_orders_for_ids(ids, None),
6996 None => Vec::new(),
6997 }
6998 }
6999
7000 #[must_use]
7002 pub fn exec_spawn_total_quantity(
7003 &self,
7004 exec_spawn_id: &ClientOrderId,
7005 active_only: bool,
7006 ) -> Option<Quantity> {
7007 self.exec_spawn_total(exec_spawn_id, active_only, Order::quantity)
7008 }
7009
7010 #[must_use]
7012 pub fn exec_spawn_total_filled_qty(
7013 &self,
7014 exec_spawn_id: &ClientOrderId,
7015 active_only: bool,
7016 ) -> Option<Quantity> {
7017 self.exec_spawn_total(exec_spawn_id, active_only, Order::filled_qty)
7018 }
7019
7020 #[must_use]
7022 pub fn exec_spawn_total_leaves_qty(
7023 &self,
7024 exec_spawn_id: &ClientOrderId,
7025 active_only: bool,
7026 ) -> Option<Quantity> {
7027 self.exec_spawn_total(exec_spawn_id, active_only, Order::leaves_qty)
7028 }
7029
7030 fn exec_spawn_total(
7031 &self,
7032 exec_spawn_id: &ClientOrderId,
7033 active_only: bool,
7034 quantity: impl Fn(&OrderAny) -> Quantity,
7035 ) -> Option<Quantity> {
7036 self.orders_for_exec_spawn(exec_spawn_id)
7037 .into_iter()
7038 .filter(|order| !active_only || !order.is_closed())
7039 .map(|order| quantity(&order))
7040 .reduce(|total, quantity| total + quantity)
7041 }
7042
7043 #[must_use]
7047 pub fn position_ref(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
7048 self.positions
7049 .get(position_id)
7050 .map(|position_cell| PositionRef::new(position_cell.borrow()))
7051 }
7052
7053 #[must_use]
7057 pub fn position(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
7058 self.position_ref(position_id)
7059 }
7060
7061 pub fn try_position_ref(
7067 &self,
7068 position_id: &PositionId,
7069 ) -> Result<PositionRef<'_>, PositionLookupError> {
7070 self.positions
7071 .get(position_id)
7072 .map(|position_cell| PositionRef::new(position_cell.borrow()))
7073 .ok_or_else(|| PositionLookupError::not_found(*position_id))
7074 }
7075
7076 pub fn try_position(
7084 &self,
7085 position_id: &PositionId,
7086 ) -> Result<PositionRef<'_>, PositionLookupError> {
7087 self.try_position_ref(position_id)
7088 }
7089
7090 #[must_use]
7100 pub fn position_mut(&mut self, position_id: &PositionId) -> Option<PositionRefMut<'_>> {
7101 self.positions
7102 .get(position_id)
7103 .map(|position_cell| PositionRefMut::new(position_cell.borrow_mut()))
7104 }
7105
7106 #[must_use]
7111 pub fn position_owned(&self, position_id: &PositionId) -> Option<Position> {
7112 self.positions
7113 .get(position_id)
7114 .map(|position_cell| position_cell.borrow().clone())
7115 }
7116
7117 #[must_use]
7119 pub fn position_for_order_ref(
7120 &self,
7121 client_order_id: &ClientOrderId,
7122 ) -> Option<PositionRef<'_>> {
7123 self.index
7124 .order_position
7125 .get(client_order_id)
7126 .and_then(|position_id| self.positions.get(position_id))
7127 .map(|position_cell| PositionRef::new(position_cell.borrow()))
7128 }
7129
7130 #[must_use]
7134 pub fn position_for_order(&self, client_order_id: &ClientOrderId) -> Option<PositionRef<'_>> {
7135 self.position_for_order_ref(client_order_id)
7136 }
7137
7138 #[must_use]
7140 pub fn position_id(&self, client_order_id: &ClientOrderId) -> Option<&PositionId> {
7141 self.index.order_position.get(client_order_id)
7142 }
7143
7144 #[must_use]
7150 pub fn positions_refs(
7151 &self,
7152 venue: Option<&Venue>,
7153 instrument_id: Option<&InstrumentId>,
7154 strategy_id: Option<&StrategyId>,
7155 account_id: Option<&AccountId>,
7156 side: Option<PositionSide>,
7157 ) -> Vec<PositionRef<'_>> {
7158 let position_ids = self.position_ids(venue, instrument_id, strategy_id, account_id);
7159 self.get_positions_for_ids(&position_ids, side)
7160 }
7161
7162 #[must_use]
7166 pub fn positions(
7167 &self,
7168 venue: Option<&Venue>,
7169 instrument_id: Option<&InstrumentId>,
7170 strategy_id: Option<&StrategyId>,
7171 account_id: Option<&AccountId>,
7172 side: Option<PositionSide>,
7173 ) -> Vec<PositionRef<'_>> {
7174 self.positions_refs(venue, instrument_id, strategy_id, account_id, side)
7175 }
7176
7177 #[must_use]
7179 pub fn positions_open_refs(
7180 &self,
7181 venue: Option<&Venue>,
7182 instrument_id: Option<&InstrumentId>,
7183 strategy_id: Option<&StrategyId>,
7184 account_id: Option<&AccountId>,
7185 side: Option<PositionSide>,
7186 ) -> Vec<PositionRef<'_>> {
7187 let position_ids = self.position_open_ids(venue, instrument_id, strategy_id, account_id);
7188 self.get_positions_for_ids(&position_ids, side)
7189 }
7190
7191 #[must_use]
7195 pub fn positions_open(
7196 &self,
7197 venue: Option<&Venue>,
7198 instrument_id: Option<&InstrumentId>,
7199 strategy_id: Option<&StrategyId>,
7200 account_id: Option<&AccountId>,
7201 side: Option<PositionSide>,
7202 ) -> Vec<PositionRef<'_>> {
7203 self.positions_open_refs(venue, instrument_id, strategy_id, account_id, side)
7204 }
7205
7206 #[must_use]
7208 pub fn positions_closed_refs(
7209 &self,
7210 venue: Option<&Venue>,
7211 instrument_id: Option<&InstrumentId>,
7212 strategy_id: Option<&StrategyId>,
7213 account_id: Option<&AccountId>,
7214 side: Option<PositionSide>,
7215 ) -> Vec<PositionRef<'_>> {
7216 let position_ids = self.position_closed_ids(venue, instrument_id, strategy_id, account_id);
7217 self.get_positions_for_ids(&position_ids, side)
7218 }
7219
7220 #[must_use]
7224 pub fn positions_closed(
7225 &self,
7226 venue: Option<&Venue>,
7227 instrument_id: Option<&InstrumentId>,
7228 strategy_id: Option<&StrategyId>,
7229 account_id: Option<&AccountId>,
7230 side: Option<PositionSide>,
7231 ) -> Vec<PositionRef<'_>> {
7232 self.positions_closed_refs(venue, instrument_id, strategy_id, account_id, side)
7233 }
7234
7235 #[must_use]
7237 pub fn position_exists(&self, position_id: &PositionId) -> bool {
7238 self.index.positions.contains(position_id)
7239 }
7240
7241 #[must_use]
7243 pub fn is_position_open(&self, position_id: &PositionId) -> bool {
7244 self.index.positions_open.contains(position_id)
7245 }
7246
7247 #[must_use]
7249 pub fn is_position_closed(&self, position_id: &PositionId) -> bool {
7250 self.index.positions_closed.contains(position_id)
7251 }
7252
7253 #[must_use]
7255 pub fn positions_open_count(
7256 &self,
7257 venue: Option<&Venue>,
7258 instrument_id: Option<&InstrumentId>,
7259 strategy_id: Option<&StrategyId>,
7260 account_id: Option<&AccountId>,
7261 side: Option<PositionSide>,
7262 ) -> usize {
7263 self.count_positions_in_bucket(
7264 &self.index.positions_open,
7265 venue,
7266 instrument_id,
7267 strategy_id,
7268 account_id,
7269 side,
7270 )
7271 }
7272
7273 #[must_use]
7275 pub fn positions_closed_count(
7276 &self,
7277 venue: Option<&Venue>,
7278 instrument_id: Option<&InstrumentId>,
7279 strategy_id: Option<&StrategyId>,
7280 account_id: Option<&AccountId>,
7281 side: Option<PositionSide>,
7282 ) -> usize {
7283 self.count_positions_in_bucket(
7284 &self.index.positions_closed,
7285 venue,
7286 instrument_id,
7287 strategy_id,
7288 account_id,
7289 side,
7290 )
7291 }
7292
7293 #[must_use]
7295 pub fn positions_total_count(
7296 &self,
7297 venue: Option<&Venue>,
7298 instrument_id: Option<&InstrumentId>,
7299 strategy_id: Option<&StrategyId>,
7300 account_id: Option<&AccountId>,
7301 side: Option<PositionSide>,
7302 ) -> usize {
7303 self.count_positions_in_bucket(
7304 &self.index.positions,
7305 venue,
7306 instrument_id,
7307 strategy_id,
7308 account_id,
7309 side,
7310 )
7311 }
7312
7313 #[must_use]
7319 pub fn has_positions_open(
7320 &self,
7321 venue: Option<&Venue>,
7322 instrument_id: Option<&InstrumentId>,
7323 strategy_id: Option<&StrategyId>,
7324 account_id: Option<&AccountId>,
7325 side: Option<PositionSide>,
7326 ) -> bool {
7327 self.any_positions_in_bucket(
7328 &self.index.positions_open,
7329 venue,
7330 instrument_id,
7331 strategy_id,
7332 account_id,
7333 side,
7334 )
7335 }
7336
7337 #[must_use]
7339 pub fn has_positions_closed(
7340 &self,
7341 venue: Option<&Venue>,
7342 instrument_id: Option<&InstrumentId>,
7343 strategy_id: Option<&StrategyId>,
7344 account_id: Option<&AccountId>,
7345 side: Option<PositionSide>,
7346 ) -> bool {
7347 self.any_positions_in_bucket(
7348 &self.index.positions_closed,
7349 venue,
7350 instrument_id,
7351 strategy_id,
7352 account_id,
7353 side,
7354 )
7355 }
7356
7357 #[must_use]
7359 pub fn has_positions(
7360 &self,
7361 venue: Option<&Venue>,
7362 instrument_id: Option<&InstrumentId>,
7363 strategy_id: Option<&StrategyId>,
7364 account_id: Option<&AccountId>,
7365 side: Option<PositionSide>,
7366 ) -> bool {
7367 self.any_positions_in_bucket(
7368 &self.index.positions,
7369 venue,
7370 instrument_id,
7371 strategy_id,
7372 account_id,
7373 side,
7374 )
7375 }
7376
7377 #[must_use]
7381 pub fn strategy_id_for_order(&self, client_order_id: &ClientOrderId) -> Option<&StrategyId> {
7382 self.index.order_strategy.get(client_order_id)
7383 }
7384
7385 #[must_use]
7387 pub fn strategy_id_for_position(&self, position_id: &PositionId) -> Option<&StrategyId> {
7388 self.index.position_strategy.get(position_id)
7389 }
7390
7391 pub fn get(&self, key: &str) -> anyhow::Result<Option<&Bytes>> {
7399 check_valid_string_ascii(key, stringify!(key))?;
7400
7401 Ok(self.general.get(key))
7402 }
7403
7404 #[must_use]
7413 pub fn price(&self, instrument_id: &InstrumentId, price_type: PriceType) -> Option<Price> {
7414 match price_type {
7415 PriceType::Bid => self
7416 .quotes
7417 .get(instrument_id)
7418 .and_then(|quotes| quotes.front().map(|quote| quote.bid_price)),
7419 PriceType::Ask => self
7420 .quotes
7421 .get(instrument_id)
7422 .and_then(|quotes| quotes.front().map(|quote| quote.ask_price)),
7423 PriceType::Mid => self.quotes.get(instrument_id).and_then(|quotes| {
7424 quotes.front().map(|quote| {
7425 let mid = (quote.ask_price.as_decimal() + quote.bid_price.as_decimal())
7426 / Decimal::TWO;
7427
7428 Price::from_decimal_dp(mid, quote.bid_price.precision + 1)
7429 .expect("Invalid mid price for Cache::price")
7430 })
7431 }),
7432 PriceType::Last => self
7433 .trades
7434 .get(instrument_id)
7435 .and_then(|trades| trades.front().map(|trade| trade.price)),
7436 PriceType::Mark => self
7437 .mark_prices
7438 .get(instrument_id)
7439 .and_then(|marks| marks.front().map(|mark| mark.value)),
7440 }
7441 }
7442
7443 #[must_use]
7445 pub fn quotes(&self, instrument_id: &InstrumentId) -> Option<Vec<QuoteTick>> {
7446 self.quotes
7447 .get(instrument_id)
7448 .map(|quotes| quotes.iter().copied().collect())
7449 }
7450
7451 #[must_use]
7453 pub fn trades(&self, instrument_id: &InstrumentId) -> Option<Vec<TradeTick>> {
7454 self.trades
7455 .get(instrument_id)
7456 .map(|trades| trades.iter().copied().collect())
7457 }
7458
7459 #[must_use]
7461 pub fn mark_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<MarkPriceUpdate>> {
7462 self.mark_prices
7463 .get(instrument_id)
7464 .map(|mark_prices| mark_prices.iter().copied().collect())
7465 }
7466
7467 #[must_use]
7469 pub fn index_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<IndexPriceUpdate>> {
7470 self.index_prices
7471 .get(instrument_id)
7472 .map(|index_prices| index_prices.iter().copied().collect())
7473 }
7474
7475 #[must_use]
7477 pub fn funding_rates(&self, instrument_id: &InstrumentId) -> Option<Vec<FundingRateUpdate>> {
7478 self.funding_rates
7479 .get(instrument_id)
7480 .map(|funding_rates| funding_rates.iter().copied().collect())
7481 }
7482
7483 #[must_use]
7485 pub fn instrument_statuses(
7486 &self,
7487 instrument_id: &InstrumentId,
7488 ) -> Option<Vec<InstrumentStatus>> {
7489 self.instrument_statuses
7490 .get(instrument_id)
7491 .map(|statuses| statuses.iter().copied().collect())
7492 }
7493
7494 #[must_use]
7496 pub fn bars(&self, bar_type: &BarType) -> Option<Vec<Bar>> {
7497 self.bars
7498 .get(bar_type)
7499 .map(|bars| bars.iter().copied().collect())
7500 }
7501
7502 #[must_use]
7504 pub fn order_book(&self, instrument_id: &InstrumentId) -> Option<&OrderBook> {
7505 self.books.get(instrument_id)
7506 }
7507
7508 pub fn try_order_book(
7514 &self,
7515 instrument_id: &InstrumentId,
7516 ) -> Result<&OrderBook, OrderBookLookupError> {
7517 self.books
7518 .get(instrument_id)
7519 .ok_or_else(|| OrderBookLookupError::not_found(*instrument_id))
7520 }
7521
7522 #[must_use]
7524 pub fn order_book_mut(&mut self, instrument_id: &InstrumentId) -> Option<&mut OrderBook> {
7525 self.books.get_mut(instrument_id)
7526 }
7527
7528 #[must_use]
7530 pub fn own_order_book(&self, instrument_id: &InstrumentId) -> Option<&OwnOrderBook> {
7531 self.own_books.get(instrument_id)
7532 }
7533
7534 pub fn try_own_order_book(
7541 &self,
7542 instrument_id: &InstrumentId,
7543 ) -> Result<&OwnOrderBook, OwnOrderBookLookupError> {
7544 self.own_books
7545 .get(instrument_id)
7546 .ok_or_else(|| OwnOrderBookLookupError::not_found(*instrument_id))
7547 }
7548
7549 #[must_use]
7551 pub fn own_order_book_mut(
7552 &mut self,
7553 instrument_id: &InstrumentId,
7554 ) -> Option<&mut OwnOrderBook> {
7555 self.own_books.get_mut(instrument_id)
7556 }
7557
7558 #[must_use]
7560 pub fn quote(&self, instrument_id: &InstrumentId) -> Option<&QuoteTick> {
7561 self.quotes
7562 .get(instrument_id)
7563 .and_then(|quotes| quotes.front())
7564 }
7565
7566 #[must_use]
7570 pub fn quote_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&QuoteTick> {
7571 self.quotes
7572 .get(instrument_id)
7573 .and_then(|quotes| quotes.get(index))
7574 }
7575
7576 #[must_use]
7578 pub fn trade(&self, instrument_id: &InstrumentId) -> Option<&TradeTick> {
7579 self.trades
7580 .get(instrument_id)
7581 .and_then(|trades| trades.front())
7582 }
7583
7584 #[must_use]
7588 pub fn trade_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&TradeTick> {
7589 self.trades
7590 .get(instrument_id)
7591 .and_then(|trades| trades.get(index))
7592 }
7593
7594 #[must_use]
7596 pub fn mark_price(&self, instrument_id: &InstrumentId) -> Option<&MarkPriceUpdate> {
7597 self.mark_prices
7598 .get(instrument_id)
7599 .and_then(|mark_prices| mark_prices.front())
7600 }
7601
7602 #[must_use]
7604 pub fn index_price(&self, instrument_id: &InstrumentId) -> Option<&IndexPriceUpdate> {
7605 self.index_prices
7606 .get(instrument_id)
7607 .and_then(|index_prices| index_prices.front())
7608 }
7609
7610 #[must_use]
7612 pub fn funding_rate(&self, instrument_id: &InstrumentId) -> Option<&FundingRateUpdate> {
7613 self.funding_rates
7614 .get(instrument_id)
7615 .and_then(|funding_rates| funding_rates.front())
7616 }
7617
7618 #[must_use]
7620 pub fn instrument_status(&self, instrument_id: &InstrumentId) -> Option<&InstrumentStatus> {
7621 self.instrument_statuses
7622 .get(instrument_id)
7623 .and_then(|statuses| statuses.front())
7624 }
7625
7626 #[must_use]
7628 pub fn bar(&self, bar_type: &BarType) -> Option<&Bar> {
7629 self.bars.get(bar_type).and_then(|bars| bars.front())
7630 }
7631
7632 #[must_use]
7636 pub fn bar_at_index(&self, bar_type: &BarType, index: usize) -> Option<&Bar> {
7637 self.bars.get(bar_type).and_then(|bars| bars.get(index))
7638 }
7639
7640 #[must_use]
7642 pub fn book_update_count(&self, instrument_id: &InstrumentId) -> usize {
7643 self.books
7644 .get(instrument_id)
7645 .map_or(0, |book| book.update_count) as usize
7646 }
7647
7648 #[must_use]
7650 pub fn quote_count(&self, instrument_id: &InstrumentId) -> usize {
7651 self.quotes
7652 .get(instrument_id)
7653 .map_or(0, BoundedVecDeque::len)
7654 }
7655
7656 #[must_use]
7658 pub fn trade_count(&self, instrument_id: &InstrumentId) -> usize {
7659 self.trades
7660 .get(instrument_id)
7661 .map_or(0, BoundedVecDeque::len)
7662 }
7663
7664 #[must_use]
7666 pub fn mark_price_count(&self, instrument_id: &InstrumentId) -> usize {
7667 self.mark_prices
7668 .get(instrument_id)
7669 .map_or(0, BoundedVecDeque::len)
7670 }
7671
7672 #[must_use]
7674 pub fn index_price_count(&self, instrument_id: &InstrumentId) -> usize {
7675 self.index_prices
7676 .get(instrument_id)
7677 .map_or(0, BoundedVecDeque::len)
7678 }
7679
7680 #[must_use]
7682 pub fn funding_rate_count(&self, instrument_id: &InstrumentId) -> usize {
7683 self.funding_rates
7684 .get(instrument_id)
7685 .map_or(0, BoundedVecDeque::len)
7686 }
7687
7688 #[must_use]
7690 pub fn instrument_status_count(&self, instrument_id: &InstrumentId) -> usize {
7691 self.instrument_statuses
7692 .get(instrument_id)
7693 .map_or(0, BoundedVecDeque::len)
7694 }
7695
7696 #[must_use]
7698 pub fn bar_count(&self, bar_type: &BarType) -> usize {
7699 self.bars.get(bar_type).map_or(0, BoundedVecDeque::len)
7700 }
7701
7702 #[must_use]
7704 pub fn has_order_book(&self, instrument_id: &InstrumentId) -> bool {
7705 self.books.contains_key(instrument_id)
7706 }
7707
7708 #[must_use]
7710 pub fn has_quote_ticks(&self, instrument_id: &InstrumentId) -> bool {
7711 self.quote_count(instrument_id) > 0
7712 }
7713
7714 #[must_use]
7716 pub fn has_trade_ticks(&self, instrument_id: &InstrumentId) -> bool {
7717 self.trade_count(instrument_id) > 0
7718 }
7719
7720 #[must_use]
7722 pub fn has_mark_prices(&self, instrument_id: &InstrumentId) -> bool {
7723 self.mark_price_count(instrument_id) > 0
7724 }
7725
7726 #[must_use]
7728 pub fn has_index_prices(&self, instrument_id: &InstrumentId) -> bool {
7729 self.index_price_count(instrument_id) > 0
7730 }
7731
7732 #[must_use]
7734 pub fn has_funding_rates(&self, instrument_id: &InstrumentId) -> bool {
7735 self.funding_rate_count(instrument_id) > 0
7736 }
7737
7738 #[must_use]
7740 pub fn has_instrument_statuses(&self, instrument_id: &InstrumentId) -> bool {
7741 self.instrument_status_count(instrument_id) > 0
7742 }
7743
7744 #[must_use]
7746 pub fn has_bars(&self, bar_type: &BarType) -> bool {
7747 self.bar_count(bar_type) > 0
7748 }
7749
7750 #[must_use]
7751 pub fn get_xrate(
7752 &self,
7753 venue: Venue,
7754 from_currency: Currency,
7755 to_currency: Currency,
7756 price_type: PriceType,
7757 ) -> Option<Decimal> {
7758 match self.try_get_xrate(venue, from_currency, to_currency, price_type) {
7759 Ok(rate) => rate,
7760 Err(e) => {
7761 log::error!("Failed to calculate xrate: {e}");
7762 None
7763 }
7764 }
7765 }
7766
7767 pub fn try_get_xrate(
7774 &self,
7775 venue: Venue,
7776 from_currency: Currency,
7777 to_currency: Currency,
7778 price_type: PriceType,
7779 ) -> anyhow::Result<Option<Decimal>> {
7780 if from_currency == to_currency {
7781 return Ok(Some(Decimal::ONE));
7784 }
7785
7786 let (bid_quote, ask_quote) = self.build_quote_table(&venue);
7787
7788 get_exchange_rate(
7789 from_currency.code,
7790 to_currency.code,
7791 price_type,
7792 bid_quote,
7793 ask_quote,
7794 )
7795 }
7796
7797 fn build_quote_table(
7798 &self,
7799 venue: &Venue,
7800 ) -> (AHashMap<Ustr, Decimal>, AHashMap<Ustr, Decimal>) {
7801 let mut bid_quotes = AHashMap::new();
7802 let mut ask_quotes = AHashMap::new();
7803 let mut quote_sources = AHashMap::new();
7804
7805 for (instrument_id, instrument) in &self.instruments {
7806 if instrument_id.venue != *venue {
7807 continue;
7808 }
7809
7810 let Some(base_currency) = instrument.base_currency() else {
7811 continue;
7812 };
7813 let pair = Ustr::from(&format!(
7814 "{}/{}",
7815 base_currency.code,
7816 instrument.quote_currency().code
7817 ));
7818
7819 let (bid_price, ask_price) = if let Some(ticks) = self.quotes.get(instrument_id) {
7820 if let Some(tick) = ticks.front() {
7821 (tick.bid_price, tick.ask_price)
7822 } else {
7823 continue; }
7825 } else {
7826 let mut latest_bid: Option<(&BarType, &Bar)> = None;
7830 let mut latest_ask: Option<(&BarType, &Bar)> = None;
7831
7832 for (bar_type, bars) in &self.bars {
7833 if bar_type.instrument_id() != *instrument_id {
7834 continue;
7835 }
7836
7837 let Some(bar) = bars.front() else {
7838 continue;
7839 };
7840
7841 let slot = match bar_type.spec().price_type {
7842 PriceType::Bid => &mut latest_bid,
7843 PriceType::Ask => &mut latest_ask,
7844 _ => continue,
7845 };
7846
7847 if slot.is_none_or(|(current_type, current)| {
7848 (current.ts_init, current_type) < (bar.ts_init, bar_type)
7849 }) {
7850 *slot = Some((bar_type, bar));
7851 }
7852 }
7853
7854 match (latest_bid, latest_ask) {
7855 (Some((_, bid_bar)), Some((_, ask_bar))) => (bid_bar.close, ask_bar.close),
7856 _ => continue,
7857 }
7858 };
7859
7860 let preference = (
7861 bid_price.is_positive() && ask_price.is_positive(),
7862 instrument.instrument_class() == InstrumentClass::Spot,
7863 Reverse(*instrument_id),
7864 );
7865
7866 if quote_sources
7867 .get(&pair)
7868 .is_some_and(|current| current >= &preference)
7869 {
7870 continue;
7871 }
7872
7873 bid_quotes.insert(pair, bid_price.as_decimal());
7874 ask_quotes.insert(pair, ask_price.as_decimal());
7875 quote_sources.insert(pair, preference);
7876 }
7877
7878 (bid_quotes, ask_quotes)
7879 }
7880
7881 #[must_use]
7883 pub fn get_mark_xrate(&self, from_currency: Currency, to_currency: Currency) -> Option<f64> {
7884 self.mark_xrates.get(&(from_currency, to_currency)).copied()
7885 }
7886
7887 pub fn set_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency, xrate: f64) {
7893 assert!(xrate > 0.0, "xrate was zero");
7894 self.mark_xrates.insert((from_currency, to_currency), xrate);
7895 self.mark_xrates
7896 .insert((to_currency, from_currency), 1.0 / xrate);
7897 }
7898
7899 pub fn clear_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency) {
7905 let _ = self.mark_xrates.remove(&(from_currency, to_currency));
7906 }
7907
7908 pub fn clear_mark_xrates(&mut self) {
7910 self.mark_xrates.clear();
7911 }
7912
7913 #[must_use]
7915 pub fn currency(&self, code: &Ustr) -> Option<&Currency> {
7916 self.currencies.get(code)
7917 }
7918
7919 pub fn try_currency(&self, code: &Ustr) -> Result<&Currency, CurrencyLookupError> {
7925 self.currencies
7926 .get(code)
7927 .ok_or_else(|| CurrencyLookupError::not_found(*code))
7928 }
7929
7930 #[must_use]
7934 pub fn instrument(&self, instrument_id: &InstrumentId) -> Option<&InstrumentAny> {
7935 self.instruments.get(instrument_id)
7936 }
7937
7938 pub fn try_instrument(
7944 &self,
7945 instrument_id: &InstrumentId,
7946 ) -> Result<&InstrumentAny, InstrumentLookupError> {
7947 self.instruments
7948 .get(instrument_id)
7949 .ok_or_else(|| InstrumentLookupError::not_found(*instrument_id))
7950 }
7951
7952 #[must_use]
7954 pub fn instrument_ids(&self, venue: Option<&Venue>) -> Vec<&InstrumentId> {
7955 match venue {
7956 Some(v) => self.instruments.keys().filter(|i| &i.venue == v).collect(),
7957 None => self.instruments.keys().collect(),
7958 }
7959 }
7960
7961 #[must_use]
7963 pub fn instruments(&self, venue: &Venue, underlying: Option<&Ustr>) -> Vec<&InstrumentAny> {
7964 self.instruments
7965 .values()
7966 .filter(|i| &i.id().venue == venue)
7967 .filter(|i| underlying.is_none_or(|u| i.underlying() == Some(*u)))
7968 .collect()
7969 }
7970
7971 #[must_use]
7978 pub fn instruments_by_parent(
7979 &self,
7980 venue: &Venue,
7981 root: &Ustr,
7982 class: InstrumentClass,
7983 ) -> Vec<&InstrumentAny> {
7984 self.instruments
7985 .values()
7986 .filter(|i| &i.id().venue == venue)
7987 .filter(|i| i.underlying() == Some(*root))
7988 .filter(|i| i.instrument_class() == class)
7989 .collect()
7990 }
7991
7992 #[must_use]
7994 pub fn bar_types(
7995 &self,
7996 instrument_id: Option<&InstrumentId>,
7997 price_type: Option<&PriceType>,
7998 aggregation_source: AggregationSource,
7999 ) -> Vec<&BarType> {
8000 let mut bar_types = self
8001 .bars
8002 .keys()
8003 .filter(|bar_type| bar_type.aggregation_source() == aggregation_source)
8004 .collect::<Vec<&BarType>>();
8005
8006 if let Some(instrument_id) = instrument_id {
8007 bar_types.retain(|bar_type| bar_type.instrument_id() == *instrument_id);
8008 }
8009
8010 if let Some(price_type) = price_type {
8011 bar_types.retain(|bar_type| &bar_type.spec().price_type == price_type);
8012 }
8013
8014 bar_types
8015 }
8016
8017 #[must_use]
8021 pub fn synthetic(&self, instrument_id: &InstrumentId) -> Option<&SyntheticInstrument> {
8022 self.synthetics.get(instrument_id)
8023 }
8024
8025 pub fn try_synthetic(
8032 &self,
8033 instrument_id: &InstrumentId,
8034 ) -> Result<&SyntheticInstrument, SyntheticInstrumentLookupError> {
8035 self.synthetics
8036 .get(instrument_id)
8037 .ok_or_else(|| SyntheticInstrumentLookupError::not_found(*instrument_id))
8038 }
8039
8040 #[must_use]
8042 pub fn synthetic_ids(&self) -> Vec<&InstrumentId> {
8043 self.synthetics.keys().collect()
8044 }
8045
8046 #[must_use]
8048 pub fn synthetics(&self) -> Vec<&SyntheticInstrument> {
8049 self.synthetics.values().collect()
8050 }
8051
8052 #[must_use]
8056 pub fn account_ref(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
8057 self.accounts
8058 .get(account_id)
8059 .map(|account_cell| AccountRef::new(account_cell.borrow()))
8060 }
8061
8062 #[must_use]
8066 pub fn account(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
8067 self.account_ref(account_id)
8068 }
8069
8070 pub fn try_account_ref(
8076 &self,
8077 account_id: &AccountId,
8078 ) -> Result<AccountRef<'_>, AccountLookupError> {
8079 self.accounts
8080 .get(account_id)
8081 .map(|account_cell| AccountRef::new(account_cell.borrow()))
8082 .ok_or_else(|| AccountLookupError::not_found(*account_id))
8083 }
8084
8085 pub fn try_account(
8093 &self,
8094 account_id: &AccountId,
8095 ) -> Result<AccountRef<'_>, AccountLookupError> {
8096 self.try_account_ref(account_id)
8097 }
8098
8099 #[must_use]
8109 pub fn account_mut(&mut self, account_id: &AccountId) -> Option<AccountRefMut<'_>> {
8110 self.accounts
8111 .get(account_id)
8112 .map(|account_cell| AccountRefMut::new(account_cell.borrow_mut()))
8113 }
8114
8115 #[must_use]
8121 pub fn account_owned(&self, account_id: &AccountId) -> Option<AccountAny> {
8122 self.accounts.get(account_id).and_then(|account_cell| {
8123 account_cell
8124 .try_borrow()
8125 .ok()
8126 .map(|account| account.clone())
8127 })
8128 }
8129
8130 #[must_use]
8132 pub fn account_for_venue(&self, venue: &Venue) -> Option<AccountRef<'_>> {
8133 self.index
8134 .venue_account
8135 .get(venue)
8136 .and_then(|account_id| self.accounts.get(account_id))
8137 .map(|account_cell| AccountRef::new(account_cell.borrow()))
8138 }
8139
8140 #[must_use]
8145 pub fn account_for_venue_owned(&self, venue: &Venue) -> Option<AccountAny> {
8146 self.index
8147 .venue_account
8148 .get(venue)
8149 .and_then(|account_id| self.accounts.get(account_id))
8150 .map(|account_cell| account_cell.borrow().clone())
8151 }
8152
8153 #[must_use]
8155 pub fn account_id(&self, venue: &Venue) -> Option<&AccountId> {
8156 self.index.venue_account.get(venue)
8157 }
8158
8159 #[must_use]
8165 pub fn accounts(&self, account_id: &AccountId) -> Vec<AccountRef<'_>> {
8166 self.accounts
8167 .values()
8168 .filter(|account_cell| &account_cell.borrow().id() == account_id)
8169 .map(|account_cell| AccountRef::new(account_cell.borrow()))
8170 .collect()
8171 }
8172
8173 #[must_use]
8175 pub fn accounts_all_owned(&self) -> Vec<AccountAny> {
8176 self.accounts
8177 .values()
8178 .map(|account_cell| account_cell.borrow().clone())
8179 .collect()
8180 }
8181
8182 pub fn update_own_order_book(&mut self, order: &OrderAny) {
8190 if !order.has_price() {
8191 return;
8192 }
8193
8194 let instrument_id = order.instrument_id();
8195
8196 if !self.own_books.contains_key(&instrument_id) {
8197 if order.is_closed() {
8198 return;
8199 }
8200
8201 self.own_books
8202 .insert(instrument_id, OwnOrderBook::new(instrument_id));
8203 }
8204
8205 let Some(own_book) = self.own_books.get_mut(&instrument_id) else {
8206 return;
8207 };
8208
8209 let own_book_order = order.to_own_book_order();
8210
8211 if order.is_closed() {
8212 if let Err(e) = own_book.delete(own_book_order) {
8213 log::debug!(
8214 "Failed to delete order {} from own book: {e}",
8215 order.client_order_id(),
8216 );
8217 } else {
8218 log::debug!("Deleted order {} from own book", order.client_order_id());
8219 }
8220 } else {
8221 if let Err(e) = own_book.update(own_book_order) {
8223 log::debug!(
8224 "Failed to update order {} in own book: {e}; inserting instead",
8225 order.client_order_id(),
8226 );
8227 own_book.add(own_book_order);
8228 }
8229 log::debug!("Updated order {} in own book", order.client_order_id());
8230 }
8231 }
8232
8233 pub fn force_remove_from_own_order_book(&mut self, client_order_id: &ClientOrderId) {
8239 let Some(order_cell) = self.orders.get(client_order_id) else {
8240 return;
8241 };
8242 let order = order_cell.borrow();
8243 let instrument_id = order.instrument_id();
8244 let own_book_order = if order.has_price() {
8245 Some(order.to_own_book_order())
8246 } else {
8247 None
8248 };
8249 drop(order);
8250
8251 self.index.orders_open.remove(client_order_id);
8252 self.index.orders_pending_cancel.remove(client_order_id);
8253 self.index.orders_inflight.remove(client_order_id);
8254 self.index.orders_emulated.remove(client_order_id);
8255 self.index.orders_active_local.remove(client_order_id);
8256
8257 if let Some(own_book) = self.own_books.get_mut(&instrument_id)
8258 && let Some(own_book_order) = own_book_order
8259 {
8260 if let Err(e) = own_book.delete(own_book_order) {
8261 log::debug!("Could not force delete {client_order_id} from own book: {e}");
8262 } else {
8263 log::debug!("Force deleted {client_order_id} from own book");
8264 }
8265 }
8266
8267 self.index.orders_closed.insert(*client_order_id);
8268 }
8269
8270 pub fn audit_own_order_books(&mut self) {
8275 log::debug!("Starting own books audit");
8276 let start = std::time::Instant::now();
8277
8278 let valid_order_ids: AHashSet<ClientOrderId> = self
8279 .index
8280 .orders_open
8281 .iter()
8282 .chain(&self.index.orders_inflight)
8283 .chain(&self.index.orders_active_local)
8284 .copied()
8285 .collect();
8286
8287 for own_book in self.own_books.values_mut() {
8288 own_book.audit_open_orders(&valid_order_ids);
8289 }
8290
8291 log::debug!("Completed own books audit in {:?}", start.elapsed());
8292 }
8293}
8294
8295const POSITION_OMS_KEY_PREFIX: &str = "position_oms:";
8296
8297fn position_oms_key(position_id: PositionId) -> String {
8298 format!("{POSITION_OMS_KEY_PREFIX}{position_id}")
8299}