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 DurationNanos, SharedCell, UnixNanos,
59 correctness::{
60 check_key_not_in_map, check_predicate_false, check_slice_not_empty,
61 check_valid_string_ascii,
62 },
63};
64#[cfg(feature = "defi")]
65use nautilus_model::defi::{Pool, PoolProfiler};
66use nautilus_model::{
67 accounts::{Account, AccountAny},
68 data::{
69 Bar, BarType, FundingRateUpdate, GreeksData, IndexPriceUpdate, InstrumentClose,
70 InstrumentStatus, MarkPriceUpdate, QuoteTick, TradeTick, YieldCurveData,
71 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 try_borrow(&self) -> Result<Ref<'_, Cache>, std::cell::BorrowError> {
123 self.inner.try_borrow()
124 }
125
126 pub fn borrow(&self) -> Ref<'_, Cache> {
132 self.inner.borrow()
133 }
134}
135
136impl From<Rc<RefCell<Cache>>> for CacheView {
137 fn from(inner: Rc<RefCell<Cache>>) -> Self {
138 Self::new(inner)
139 }
140}
141
142#[derive(Debug)]
149pub struct CacheApi<'a> {
150 cache: &'a RefCell<Cache>,
151}
152
153impl<'a> CacheApi<'a> {
154 pub(crate) fn new(cache: &'a RefCell<Cache>) -> Self {
155 Self { cache }
156 }
157
158 #[must_use]
164 pub fn calculate_unrealized_pnl(&self, position: &Position) -> Option<Money> {
165 self.cache().calculate_unrealized_pnl(position)
166 }
167
168 #[must_use]
174 pub fn oms_type(&self, position_id: &PositionId) -> Option<OmsType> {
175 self.cache().oms_type(position_id)
176 }
177
178 #[must_use]
184 pub fn position_snapshot_bytes(&self, position_id: &PositionId) -> Option<Vec<Vec<u8>>> {
185 self.cache().position_snapshot_bytes(position_id)
186 }
187
188 #[must_use]
194 pub fn position_snapshot_count(&self, position_id: &PositionId) -> usize {
195 self.cache().position_snapshot_count(position_id)
196 }
197
198 #[must_use]
204 pub fn position_snapshots(
205 &self,
206 position_id: Option<&PositionId>,
207 account_id: Option<&AccountId>,
208 ) -> Vec<Position> {
209 self.cache().position_snapshots(position_id, account_id)
210 }
211
212 #[must_use]
218 pub fn position_snapshots_from(&self, position_id: &PositionId, skip: usize) -> Vec<Position> {
219 self.cache().position_snapshots_from(position_id, skip)
220 }
221
222 #[must_use]
228 pub fn position_snapshot_ids(&self, instrument_id: &InstrumentId) -> AHashSet<PositionId> {
229 self.cache().position_snapshot_ids(instrument_id)
230 }
231
232 #[must_use]
238 pub fn client_order_ids(
239 &self,
240 venue: Option<&Venue>,
241 instrument_id: Option<&InstrumentId>,
242 strategy_id: Option<&StrategyId>,
243 account_id: Option<&AccountId>,
244 ) -> AHashSet<ClientOrderId> {
245 self.cache()
246 .client_order_ids(venue, instrument_id, strategy_id, account_id)
247 }
248
249 #[must_use]
255 pub fn client_order_ids_open(
256 &self,
257 venue: Option<&Venue>,
258 instrument_id: Option<&InstrumentId>,
259 strategy_id: Option<&StrategyId>,
260 account_id: Option<&AccountId>,
261 ) -> AHashSet<ClientOrderId> {
262 self.cache()
263 .client_order_ids_open(venue, instrument_id, strategy_id, account_id)
264 }
265
266 #[must_use]
272 pub fn client_order_ids_closed(
273 &self,
274 venue: Option<&Venue>,
275 instrument_id: Option<&InstrumentId>,
276 strategy_id: Option<&StrategyId>,
277 account_id: Option<&AccountId>,
278 ) -> AHashSet<ClientOrderId> {
279 self.cache()
280 .client_order_ids_closed(venue, instrument_id, strategy_id, account_id)
281 }
282
283 #[must_use]
289 pub fn client_order_ids_active_local(
290 &self,
291 venue: Option<&Venue>,
292 instrument_id: Option<&InstrumentId>,
293 strategy_id: Option<&StrategyId>,
294 account_id: Option<&AccountId>,
295 ) -> AHashSet<ClientOrderId> {
296 self.cache()
297 .client_order_ids_active_local(venue, instrument_id, strategy_id, account_id)
298 }
299
300 #[must_use]
306 pub fn client_order_ids_emulated(
307 &self,
308 venue: Option<&Venue>,
309 instrument_id: Option<&InstrumentId>,
310 strategy_id: Option<&StrategyId>,
311 account_id: Option<&AccountId>,
312 ) -> AHashSet<ClientOrderId> {
313 self.cache()
314 .client_order_ids_emulated(venue, instrument_id, strategy_id, account_id)
315 }
316
317 #[must_use]
323 pub fn client_order_ids_inflight(
324 &self,
325 venue: Option<&Venue>,
326 instrument_id: Option<&InstrumentId>,
327 strategy_id: Option<&StrategyId>,
328 account_id: Option<&AccountId>,
329 ) -> AHashSet<ClientOrderId> {
330 self.cache()
331 .client_order_ids_inflight(venue, instrument_id, strategy_id, account_id)
332 }
333
334 #[must_use]
340 pub fn position_ids(
341 &self,
342 venue: Option<&Venue>,
343 instrument_id: Option<&InstrumentId>,
344 strategy_id: Option<&StrategyId>,
345 account_id: Option<&AccountId>,
346 ) -> AHashSet<PositionId> {
347 self.cache()
348 .position_ids(venue, instrument_id, strategy_id, account_id)
349 }
350
351 #[must_use]
357 pub fn position_open_ids(
358 &self,
359 venue: Option<&Venue>,
360 instrument_id: Option<&InstrumentId>,
361 strategy_id: Option<&StrategyId>,
362 account_id: Option<&AccountId>,
363 ) -> AHashSet<PositionId> {
364 self.cache()
365 .position_open_ids(venue, instrument_id, strategy_id, account_id)
366 }
367
368 #[must_use]
374 pub fn position_closed_ids(
375 &self,
376 venue: Option<&Venue>,
377 instrument_id: Option<&InstrumentId>,
378 strategy_id: Option<&StrategyId>,
379 account_id: Option<&AccountId>,
380 ) -> AHashSet<PositionId> {
381 self.cache()
382 .position_closed_ids(venue, instrument_id, strategy_id, account_id)
383 }
384
385 #[must_use]
391 pub fn strategy_ids(&self) -> AHashSet<StrategyId> {
392 self.cache().strategy_ids()
393 }
394
395 #[must_use]
401 pub fn exec_algorithm_ids(&self) -> AHashSet<ExecAlgorithmId> {
402 self.cache().exec_algorithm_ids()
403 }
404
405 #[must_use]
411 pub fn order(&self, client_order_id: &ClientOrderId) -> Option<OrderAny> {
412 self.cache().order_owned(client_order_id)
413 }
414
415 pub fn try_order(&self, client_order_id: &ClientOrderId) -> Result<OrderAny, OrderLookupError> {
426 self.cache().try_order_owned(client_order_id)
427 }
428
429 #[must_use]
435 pub fn orders_for_ids(
436 &self,
437 client_order_ids: &[ClientOrderId],
438 context: &dyn Display,
439 ) -> Vec<OrderAny> {
440 self.cache().orders_for_ids(client_order_ids, context)
441 }
442
443 #[must_use]
449 pub fn client_order_id(&self, venue_order_id: &VenueOrderId) -> Option<ClientOrderId> {
450 self.cache().client_order_id(venue_order_id).copied()
451 }
452
453 #[must_use]
459 pub fn venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
460 self.cache().venue_order_id(client_order_id).copied()
461 }
462
463 #[must_use]
469 pub fn client_id(&self, client_order_id: &ClientOrderId) -> Option<ClientId> {
470 self.cache().client_id(client_order_id).copied()
471 }
472
473 #[must_use]
479 pub fn orders(
480 &self,
481 venue: Option<&Venue>,
482 instrument_id: Option<&InstrumentId>,
483 strategy_id: Option<&StrategyId>,
484 account_id: Option<&AccountId>,
485 side: Option<OrderSide>,
486 ) -> Vec<OrderAny> {
487 self.cache()
488 .orders_refs(venue, instrument_id, strategy_id, account_id, side)
489 .into_iter()
490 .map(|order| order.cloned())
491 .collect()
492 }
493
494 #[must_use]
500 pub fn orders_open(
501 &self,
502 venue: Option<&Venue>,
503 instrument_id: Option<&InstrumentId>,
504 strategy_id: Option<&StrategyId>,
505 account_id: Option<&AccountId>,
506 side: Option<OrderSide>,
507 ) -> Vec<OrderAny> {
508 self.cache()
509 .orders_open_refs(venue, instrument_id, strategy_id, account_id, side)
510 .into_iter()
511 .map(|order| order.cloned())
512 .collect()
513 }
514
515 #[must_use]
521 pub fn orders_closed(
522 &self,
523 venue: Option<&Venue>,
524 instrument_id: Option<&InstrumentId>,
525 strategy_id: Option<&StrategyId>,
526 account_id: Option<&AccountId>,
527 side: Option<OrderSide>,
528 ) -> Vec<OrderAny> {
529 self.cache()
530 .orders_closed_refs(venue, instrument_id, strategy_id, account_id, side)
531 .into_iter()
532 .map(|order| order.cloned())
533 .collect()
534 }
535
536 #[must_use]
542 pub fn orders_active_local(
543 &self,
544 venue: Option<&Venue>,
545 instrument_id: Option<&InstrumentId>,
546 strategy_id: Option<&StrategyId>,
547 account_id: Option<&AccountId>,
548 side: Option<OrderSide>,
549 ) -> Vec<OrderAny> {
550 self.cache()
551 .orders_active_local_refs(venue, instrument_id, strategy_id, account_id, side)
552 .into_iter()
553 .map(|order| order.cloned())
554 .collect()
555 }
556
557 #[must_use]
563 pub fn orders_emulated(
564 &self,
565 venue: Option<&Venue>,
566 instrument_id: Option<&InstrumentId>,
567 strategy_id: Option<&StrategyId>,
568 account_id: Option<&AccountId>,
569 side: Option<OrderSide>,
570 ) -> Vec<OrderAny> {
571 self.cache()
572 .orders_emulated_refs(venue, instrument_id, strategy_id, account_id, side)
573 .into_iter()
574 .map(|order| order.cloned())
575 .collect()
576 }
577
578 #[must_use]
584 pub fn orders_inflight(
585 &self,
586 venue: Option<&Venue>,
587 instrument_id: Option<&InstrumentId>,
588 strategy_id: Option<&StrategyId>,
589 account_id: Option<&AccountId>,
590 side: Option<OrderSide>,
591 ) -> Vec<OrderAny> {
592 self.cache()
593 .orders_inflight_refs(venue, instrument_id, strategy_id, account_id, side)
594 .into_iter()
595 .map(|order| order.cloned())
596 .collect()
597 }
598
599 #[must_use]
605 pub fn orders_for_position(&self, position_id: &PositionId) -> Vec<OrderAny> {
606 self.cache()
607 .orders_for_position(position_id)
608 .into_iter()
609 .map(|order| order.cloned())
610 .collect()
611 }
612
613 #[must_use]
619 pub fn order_exists(&self, client_order_id: &ClientOrderId) -> bool {
620 self.cache().order_exists(client_order_id)
621 }
622
623 #[must_use]
629 pub fn is_order_open(&self, client_order_id: &ClientOrderId) -> bool {
630 self.cache().is_order_open(client_order_id)
631 }
632
633 #[must_use]
639 pub fn is_order_closed(&self, client_order_id: &ClientOrderId) -> bool {
640 self.cache().is_order_closed(client_order_id)
641 }
642
643 #[must_use]
649 pub fn is_order_active_local(&self, client_order_id: &ClientOrderId) -> bool {
650 self.cache().is_order_active_local(client_order_id)
651 }
652
653 #[must_use]
659 pub fn is_order_emulated(&self, client_order_id: &ClientOrderId) -> bool {
660 self.cache().is_order_emulated(client_order_id)
661 }
662
663 #[must_use]
669 pub fn is_order_inflight(&self, client_order_id: &ClientOrderId) -> bool {
670 self.cache().is_order_inflight(client_order_id)
671 }
672
673 #[must_use]
679 pub fn is_order_pending_cancel_local(&self, client_order_id: &ClientOrderId) -> bool {
680 self.cache().is_order_pending_cancel_local(client_order_id)
681 }
682
683 #[must_use]
689 pub fn orders_open_count(
690 &self,
691 venue: Option<&Venue>,
692 instrument_id: Option<&InstrumentId>,
693 strategy_id: Option<&StrategyId>,
694 account_id: Option<&AccountId>,
695 side: Option<OrderSide>,
696 ) -> usize {
697 self.cache()
698 .orders_open_count(venue, instrument_id, strategy_id, account_id, side)
699 }
700
701 #[must_use]
707 pub fn orders_closed_count(
708 &self,
709 venue: Option<&Venue>,
710 instrument_id: Option<&InstrumentId>,
711 strategy_id: Option<&StrategyId>,
712 account_id: Option<&AccountId>,
713 side: Option<OrderSide>,
714 ) -> usize {
715 self.cache()
716 .orders_closed_count(venue, instrument_id, strategy_id, account_id, side)
717 }
718
719 #[must_use]
725 pub fn orders_active_local_count(
726 &self,
727 venue: Option<&Venue>,
728 instrument_id: Option<&InstrumentId>,
729 strategy_id: Option<&StrategyId>,
730 account_id: Option<&AccountId>,
731 side: Option<OrderSide>,
732 ) -> usize {
733 self.cache()
734 .orders_active_local_count(venue, instrument_id, strategy_id, account_id, side)
735 }
736
737 #[must_use]
743 pub fn orders_emulated_count(
744 &self,
745 venue: Option<&Venue>,
746 instrument_id: Option<&InstrumentId>,
747 strategy_id: Option<&StrategyId>,
748 account_id: Option<&AccountId>,
749 side: Option<OrderSide>,
750 ) -> usize {
751 self.cache()
752 .orders_emulated_count(venue, instrument_id, strategy_id, account_id, side)
753 }
754
755 #[must_use]
761 pub fn orders_inflight_count(
762 &self,
763 venue: Option<&Venue>,
764 instrument_id: Option<&InstrumentId>,
765 strategy_id: Option<&StrategyId>,
766 account_id: Option<&AccountId>,
767 side: Option<OrderSide>,
768 ) -> usize {
769 self.cache()
770 .orders_inflight_count(venue, instrument_id, strategy_id, account_id, side)
771 }
772
773 #[must_use]
779 pub fn orders_total_count(
780 &self,
781 venue: Option<&Venue>,
782 instrument_id: Option<&InstrumentId>,
783 strategy_id: Option<&StrategyId>,
784 account_id: Option<&AccountId>,
785 side: Option<OrderSide>,
786 ) -> usize {
787 self.cache()
788 .orders_total_count(venue, instrument_id, strategy_id, account_id, side)
789 }
790
791 #[must_use]
797 pub fn has_orders_open(
798 &self,
799 venue: Option<&Venue>,
800 instrument_id: Option<&InstrumentId>,
801 strategy_id: Option<&StrategyId>,
802 account_id: Option<&AccountId>,
803 side: Option<OrderSide>,
804 ) -> bool {
805 self.cache()
806 .has_orders_open(venue, instrument_id, strategy_id, account_id, side)
807 }
808
809 #[must_use]
815 pub fn has_orders_closed(
816 &self,
817 venue: Option<&Venue>,
818 instrument_id: Option<&InstrumentId>,
819 strategy_id: Option<&StrategyId>,
820 account_id: Option<&AccountId>,
821 side: Option<OrderSide>,
822 ) -> bool {
823 self.cache()
824 .has_orders_closed(venue, instrument_id, strategy_id, account_id, side)
825 }
826
827 #[must_use]
833 pub fn has_orders_active_local(
834 &self,
835 venue: Option<&Venue>,
836 instrument_id: Option<&InstrumentId>,
837 strategy_id: Option<&StrategyId>,
838 account_id: Option<&AccountId>,
839 side: Option<OrderSide>,
840 ) -> bool {
841 self.cache()
842 .has_orders_active_local(venue, instrument_id, strategy_id, account_id, side)
843 }
844
845 #[must_use]
851 pub fn has_orders_emulated(
852 &self,
853 venue: Option<&Venue>,
854 instrument_id: Option<&InstrumentId>,
855 strategy_id: Option<&StrategyId>,
856 account_id: Option<&AccountId>,
857 side: Option<OrderSide>,
858 ) -> bool {
859 self.cache()
860 .has_orders_emulated(venue, instrument_id, strategy_id, account_id, side)
861 }
862
863 #[must_use]
869 pub fn has_orders_inflight(
870 &self,
871 venue: Option<&Venue>,
872 instrument_id: Option<&InstrumentId>,
873 strategy_id: Option<&StrategyId>,
874 account_id: Option<&AccountId>,
875 side: Option<OrderSide>,
876 ) -> bool {
877 self.cache()
878 .has_orders_inflight(venue, instrument_id, strategy_id, account_id, side)
879 }
880
881 #[must_use]
887 pub fn has_orders(
888 &self,
889 venue: Option<&Venue>,
890 instrument_id: Option<&InstrumentId>,
891 strategy_id: Option<&StrategyId>,
892 account_id: Option<&AccountId>,
893 side: Option<OrderSide>,
894 ) -> bool {
895 self.cache()
896 .has_orders(venue, instrument_id, strategy_id, account_id, side)
897 }
898
899 #[must_use]
905 pub fn order_list(&self, order_list_id: &OrderListId) -> Option<OrderList> {
906 self.cache().order_list(order_list_id).cloned()
907 }
908
909 pub fn try_order_list(
920 &self,
921 order_list_id: &OrderListId,
922 ) -> Result<OrderList, OrderListLookupError> {
923 self.cache().try_order_list(order_list_id).cloned()
924 }
925
926 #[must_use]
932 pub fn order_lists(
933 &self,
934 venue: Option<&Venue>,
935 instrument_id: Option<&InstrumentId>,
936 strategy_id: Option<&StrategyId>,
937 account_id: Option<&AccountId>,
938 ) -> Vec<OrderList> {
939 self.cache()
940 .order_lists(venue, instrument_id, strategy_id, account_id)
941 .into_iter()
942 .cloned()
943 .collect()
944 }
945
946 #[must_use]
952 pub fn order_list_exists(&self, order_list_id: &OrderListId) -> bool {
953 self.cache().order_list_exists(order_list_id)
954 }
955
956 #[must_use]
962 pub fn orders_for_exec_algorithm(
963 &self,
964 exec_algorithm_id: &ExecAlgorithmId,
965 venue: Option<&Venue>,
966 instrument_id: Option<&InstrumentId>,
967 strategy_id: Option<&StrategyId>,
968 account_id: Option<&AccountId>,
969 side: Option<OrderSide>,
970 ) -> Vec<OrderAny> {
971 self.cache()
972 .orders_for_exec_algorithm(
973 exec_algorithm_id,
974 venue,
975 instrument_id,
976 strategy_id,
977 account_id,
978 side,
979 )
980 .into_iter()
981 .map(|order| order.cloned())
982 .collect()
983 }
984
985 #[must_use]
991 pub fn orders_for_exec_spawn(&self, exec_spawn_id: &ClientOrderId) -> Vec<OrderAny> {
992 self.cache()
993 .orders_for_exec_spawn(exec_spawn_id)
994 .into_iter()
995 .map(|order| order.cloned())
996 .collect()
997 }
998
999 #[must_use]
1005 pub fn exec_spawn_total_quantity(
1006 &self,
1007 exec_spawn_id: &ClientOrderId,
1008 active_only: bool,
1009 ) -> Option<Quantity> {
1010 self.cache()
1011 .exec_spawn_total_quantity(exec_spawn_id, active_only)
1012 }
1013
1014 #[must_use]
1020 pub fn exec_spawn_total_filled_qty(
1021 &self,
1022 exec_spawn_id: &ClientOrderId,
1023 active_only: bool,
1024 ) -> Option<Quantity> {
1025 self.cache()
1026 .exec_spawn_total_filled_qty(exec_spawn_id, active_only)
1027 }
1028
1029 #[must_use]
1035 pub fn exec_spawn_total_leaves_qty(
1036 &self,
1037 exec_spawn_id: &ClientOrderId,
1038 active_only: bool,
1039 ) -> Option<Quantity> {
1040 self.cache()
1041 .exec_spawn_total_leaves_qty(exec_spawn_id, active_only)
1042 }
1043
1044 #[must_use]
1050 pub fn position(&self, position_id: &PositionId) -> Option<Position> {
1051 self.cache()
1052 .position_ref(position_id)
1053 .map(|position| position.cloned())
1054 }
1055
1056 pub fn try_position(&self, position_id: &PositionId) -> Result<Position, PositionLookupError> {
1067 self.cache()
1068 .try_position_ref(position_id)
1069 .map(|position| position.cloned())
1070 }
1071
1072 #[must_use]
1078 pub fn position_for_order(&self, client_order_id: &ClientOrderId) -> Option<Position> {
1079 self.cache()
1080 .position_for_order_ref(client_order_id)
1081 .map(|position| position.cloned())
1082 }
1083
1084 #[must_use]
1090 pub fn position_id(&self, client_order_id: &ClientOrderId) -> Option<PositionId> {
1091 self.cache().position_id(client_order_id).copied()
1092 }
1093
1094 #[must_use]
1100 pub fn positions(
1101 &self,
1102 venue: Option<&Venue>,
1103 instrument_id: Option<&InstrumentId>,
1104 strategy_id: Option<&StrategyId>,
1105 account_id: Option<&AccountId>,
1106 side: Option<PositionSide>,
1107 ) -> Vec<Position> {
1108 self.cache()
1109 .positions_refs(venue, instrument_id, strategy_id, account_id, side)
1110 .into_iter()
1111 .map(|position| position.cloned())
1112 .collect()
1113 }
1114
1115 #[must_use]
1121 pub fn positions_open(
1122 &self,
1123 venue: Option<&Venue>,
1124 instrument_id: Option<&InstrumentId>,
1125 strategy_id: Option<&StrategyId>,
1126 account_id: Option<&AccountId>,
1127 side: Option<PositionSide>,
1128 ) -> Vec<Position> {
1129 self.cache()
1130 .positions_open_refs(venue, instrument_id, strategy_id, account_id, side)
1131 .into_iter()
1132 .map(|position| position.cloned())
1133 .collect()
1134 }
1135
1136 #[must_use]
1142 pub fn positions_closed(
1143 &self,
1144 venue: Option<&Venue>,
1145 instrument_id: Option<&InstrumentId>,
1146 strategy_id: Option<&StrategyId>,
1147 account_id: Option<&AccountId>,
1148 side: Option<PositionSide>,
1149 ) -> Vec<Position> {
1150 self.cache()
1151 .positions_closed_refs(venue, instrument_id, strategy_id, account_id, side)
1152 .into_iter()
1153 .map(|position| position.cloned())
1154 .collect()
1155 }
1156
1157 #[must_use]
1163 pub fn position_exists(&self, position_id: &PositionId) -> bool {
1164 self.cache().position_exists(position_id)
1165 }
1166
1167 #[must_use]
1173 pub fn is_position_open(&self, position_id: &PositionId) -> bool {
1174 self.cache().is_position_open(position_id)
1175 }
1176
1177 #[must_use]
1183 pub fn is_position_closed(&self, position_id: &PositionId) -> bool {
1184 self.cache().is_position_closed(position_id)
1185 }
1186
1187 #[must_use]
1193 pub fn positions_open_count(
1194 &self,
1195 venue: Option<&Venue>,
1196 instrument_id: Option<&InstrumentId>,
1197 strategy_id: Option<&StrategyId>,
1198 account_id: Option<&AccountId>,
1199 side: Option<PositionSide>,
1200 ) -> usize {
1201 self.cache()
1202 .positions_open_count(venue, instrument_id, strategy_id, account_id, side)
1203 }
1204
1205 #[must_use]
1211 pub fn positions_closed_count(
1212 &self,
1213 venue: Option<&Venue>,
1214 instrument_id: Option<&InstrumentId>,
1215 strategy_id: Option<&StrategyId>,
1216 account_id: Option<&AccountId>,
1217 side: Option<PositionSide>,
1218 ) -> usize {
1219 self.cache()
1220 .positions_closed_count(venue, instrument_id, strategy_id, account_id, side)
1221 }
1222
1223 #[must_use]
1229 pub fn positions_total_count(
1230 &self,
1231 venue: Option<&Venue>,
1232 instrument_id: Option<&InstrumentId>,
1233 strategy_id: Option<&StrategyId>,
1234 account_id: Option<&AccountId>,
1235 side: Option<PositionSide>,
1236 ) -> usize {
1237 self.cache()
1238 .positions_total_count(venue, instrument_id, strategy_id, account_id, side)
1239 }
1240
1241 #[must_use]
1247 pub fn has_positions_open(
1248 &self,
1249 venue: Option<&Venue>,
1250 instrument_id: Option<&InstrumentId>,
1251 strategy_id: Option<&StrategyId>,
1252 account_id: Option<&AccountId>,
1253 side: Option<PositionSide>,
1254 ) -> bool {
1255 self.cache()
1256 .has_positions_open(venue, instrument_id, strategy_id, account_id, side)
1257 }
1258
1259 #[must_use]
1265 pub fn has_positions_closed(
1266 &self,
1267 venue: Option<&Venue>,
1268 instrument_id: Option<&InstrumentId>,
1269 strategy_id: Option<&StrategyId>,
1270 account_id: Option<&AccountId>,
1271 side: Option<PositionSide>,
1272 ) -> bool {
1273 self.cache()
1274 .has_positions_closed(venue, instrument_id, strategy_id, account_id, side)
1275 }
1276
1277 #[must_use]
1283 pub fn has_positions(
1284 &self,
1285 venue: Option<&Venue>,
1286 instrument_id: Option<&InstrumentId>,
1287 strategy_id: Option<&StrategyId>,
1288 account_id: Option<&AccountId>,
1289 side: Option<PositionSide>,
1290 ) -> bool {
1291 self.cache()
1292 .has_positions(venue, instrument_id, strategy_id, account_id, side)
1293 }
1294
1295 #[must_use]
1301 pub fn strategy_id_for_order(&self, client_order_id: &ClientOrderId) -> Option<StrategyId> {
1302 self.cache().strategy_id_for_order(client_order_id).copied()
1303 }
1304
1305 #[must_use]
1311 pub fn strategy_id_for_position(&self, position_id: &PositionId) -> Option<StrategyId> {
1312 self.cache().strategy_id_for_position(position_id).copied()
1313 }
1314
1315 pub fn get(&self, key: &str) -> anyhow::Result<Option<Bytes>> {
1326 let cache = self.cache();
1327 let value = cache.get(key)?;
1328 Ok(value.cloned())
1329 }
1330
1331 #[must_use]
1338 pub fn price(&self, instrument_id: &InstrumentId, price_type: PriceType) -> Option<Price> {
1339 self.cache().price(instrument_id, price_type)
1340 }
1341
1342 #[must_use]
1348 pub fn quotes(&self, instrument_id: &InstrumentId) -> Option<Vec<QuoteTick>> {
1349 self.cache().quotes(instrument_id)
1350 }
1351
1352 #[must_use]
1358 pub fn trades(&self, instrument_id: &InstrumentId) -> Option<Vec<TradeTick>> {
1359 self.cache().trades(instrument_id)
1360 }
1361
1362 #[must_use]
1368 pub fn mark_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<MarkPriceUpdate>> {
1369 self.cache().mark_prices(instrument_id)
1370 }
1371
1372 #[must_use]
1378 pub fn index_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<IndexPriceUpdate>> {
1379 self.cache().index_prices(instrument_id)
1380 }
1381
1382 #[must_use]
1388 pub fn funding_rates(&self, instrument_id: &InstrumentId) -> Option<Vec<FundingRateUpdate>> {
1389 self.cache().funding_rates(instrument_id)
1390 }
1391
1392 #[must_use]
1398 pub fn instrument_statuses(
1399 &self,
1400 instrument_id: &InstrumentId,
1401 ) -> Option<Vec<InstrumentStatus>> {
1402 self.cache().instrument_statuses(instrument_id)
1403 }
1404
1405 #[must_use]
1411 pub fn bars(&self, bar_type: &BarType) -> Option<Vec<Bar>> {
1412 self.cache().bars(bar_type)
1413 }
1414
1415 #[must_use]
1421 pub fn order_book(&self, instrument_id: &InstrumentId) -> Option<OrderBook> {
1422 self.cache().order_book(instrument_id).cloned()
1423 }
1424
1425 pub fn try_order_book(
1436 &self,
1437 instrument_id: &InstrumentId,
1438 ) -> Result<OrderBook, OrderBookLookupError> {
1439 self.cache().try_order_book(instrument_id).cloned()
1440 }
1441
1442 #[must_use]
1448 pub fn own_order_book(&self, instrument_id: &InstrumentId) -> Option<OwnOrderBook> {
1449 self.cache().own_order_book(instrument_id).cloned()
1450 }
1451
1452 pub fn try_own_order_book(
1464 &self,
1465 instrument_id: &InstrumentId,
1466 ) -> Result<OwnOrderBook, OwnOrderBookLookupError> {
1467 self.cache().try_own_order_book(instrument_id).cloned()
1468 }
1469
1470 #[must_use]
1476 pub fn quote(&self, instrument_id: &InstrumentId) -> Option<QuoteTick> {
1477 self.cache().quote(instrument_id).copied()
1478 }
1479
1480 #[must_use]
1488 pub fn quote_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<QuoteTick> {
1489 self.cache().quote_at_index(instrument_id, index).copied()
1490 }
1491
1492 #[must_use]
1498 pub fn trade(&self, instrument_id: &InstrumentId) -> Option<TradeTick> {
1499 self.cache().trade(instrument_id).copied()
1500 }
1501
1502 #[must_use]
1510 pub fn trade_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<TradeTick> {
1511 self.cache().trade_at_index(instrument_id, index).copied()
1512 }
1513
1514 #[must_use]
1520 pub fn mark_price(&self, instrument_id: &InstrumentId) -> Option<MarkPriceUpdate> {
1521 self.cache().mark_price(instrument_id).copied()
1522 }
1523
1524 #[must_use]
1530 pub fn index_price(&self, instrument_id: &InstrumentId) -> Option<IndexPriceUpdate> {
1531 self.cache().index_price(instrument_id).copied()
1532 }
1533
1534 #[must_use]
1540 pub fn funding_rate(&self, instrument_id: &InstrumentId) -> Option<FundingRateUpdate> {
1541 self.cache().funding_rate(instrument_id).copied()
1542 }
1543
1544 #[must_use]
1550 pub fn instrument_status(&self, instrument_id: &InstrumentId) -> Option<InstrumentStatus> {
1551 self.cache().instrument_status(instrument_id).copied()
1552 }
1553
1554 #[must_use]
1560 pub fn instrument_close(&self, instrument_id: &InstrumentId) -> Option<InstrumentClose> {
1561 self.cache().instrument_close(instrument_id).copied()
1562 }
1563
1564 #[must_use]
1570 pub fn bar(&self, bar_type: &BarType) -> Option<Bar> {
1571 self.cache().bar(bar_type).copied()
1572 }
1573
1574 #[must_use]
1582 pub fn bar_at_index(&self, bar_type: &BarType, index: usize) -> Option<Bar> {
1583 self.cache().bar_at_index(bar_type, index).copied()
1584 }
1585
1586 #[must_use]
1592 pub fn book_update_count(&self, instrument_id: &InstrumentId) -> usize {
1593 self.cache().book_update_count(instrument_id)
1594 }
1595
1596 #[must_use]
1602 pub fn quote_count(&self, instrument_id: &InstrumentId) -> usize {
1603 self.cache().quote_count(instrument_id)
1604 }
1605
1606 #[must_use]
1612 pub fn trade_count(&self, instrument_id: &InstrumentId) -> usize {
1613 self.cache().trade_count(instrument_id)
1614 }
1615
1616 #[must_use]
1622 pub fn mark_price_count(&self, instrument_id: &InstrumentId) -> usize {
1623 self.cache().mark_price_count(instrument_id)
1624 }
1625
1626 #[must_use]
1632 pub fn index_price_count(&self, instrument_id: &InstrumentId) -> usize {
1633 self.cache().index_price_count(instrument_id)
1634 }
1635
1636 #[must_use]
1642 pub fn funding_rate_count(&self, instrument_id: &InstrumentId) -> usize {
1643 self.cache().funding_rate_count(instrument_id)
1644 }
1645
1646 #[must_use]
1652 pub fn instrument_status_count(&self, instrument_id: &InstrumentId) -> usize {
1653 self.cache().instrument_status_count(instrument_id)
1654 }
1655
1656 #[must_use]
1662 pub fn bar_count(&self, bar_type: &BarType) -> usize {
1663 self.cache().bar_count(bar_type)
1664 }
1665
1666 #[must_use]
1672 pub fn has_order_book(&self, instrument_id: &InstrumentId) -> bool {
1673 self.cache().has_order_book(instrument_id)
1674 }
1675
1676 #[must_use]
1682 pub fn has_quote_ticks(&self, instrument_id: &InstrumentId) -> bool {
1683 self.cache().has_quote_ticks(instrument_id)
1684 }
1685
1686 #[must_use]
1692 pub fn has_trade_ticks(&self, instrument_id: &InstrumentId) -> bool {
1693 self.cache().has_trade_ticks(instrument_id)
1694 }
1695
1696 #[must_use]
1702 pub fn has_mark_prices(&self, instrument_id: &InstrumentId) -> bool {
1703 self.cache().has_mark_prices(instrument_id)
1704 }
1705
1706 #[must_use]
1712 pub fn has_index_prices(&self, instrument_id: &InstrumentId) -> bool {
1713 self.cache().has_index_prices(instrument_id)
1714 }
1715
1716 #[must_use]
1722 pub fn has_funding_rates(&self, instrument_id: &InstrumentId) -> bool {
1723 self.cache().has_funding_rates(instrument_id)
1724 }
1725
1726 #[must_use]
1732 pub fn has_instrument_statuses(&self, instrument_id: &InstrumentId) -> bool {
1733 self.cache().has_instrument_statuses(instrument_id)
1734 }
1735
1736 #[must_use]
1742 pub fn has_instrument_close(&self, instrument_id: &InstrumentId) -> bool {
1743 self.cache().has_instrument_close(instrument_id)
1744 }
1745
1746 #[must_use]
1752 pub fn has_bars(&self, bar_type: &BarType) -> bool {
1753 self.cache().has_bars(bar_type)
1754 }
1755
1756 #[must_use]
1762 pub fn get_xrate(
1763 &self,
1764 venue: Venue,
1765 from_currency: Currency,
1766 to_currency: Currency,
1767 price_type: PriceType,
1768 ) -> Option<Decimal> {
1769 self.cache()
1770 .get_xrate(venue, from_currency, to_currency, price_type)
1771 }
1772
1773 #[must_use]
1779 pub fn get_mark_xrate(&self, from_currency: Currency, to_currency: Currency) -> Option<f64> {
1780 self.cache().get_mark_xrate(from_currency, to_currency)
1781 }
1782
1783 #[must_use]
1789 pub fn yield_curve(&self, key: &str) -> Option<Box<dyn Fn(f64) -> f64>> {
1790 self.cache().yield_curve(key)
1791 }
1792
1793 #[must_use]
1799 pub fn greeks(&self, instrument_id: &InstrumentId) -> Option<GreeksData> {
1800 self.cache().greeks(instrument_id)
1801 }
1802
1803 #[must_use]
1809 pub fn option_greeks(&self, instrument_id: &InstrumentId) -> Option<OptionGreeks> {
1810 self.cache().option_greeks(instrument_id).copied()
1811 }
1812
1813 #[must_use]
1819 pub fn currency(&self, code: &Ustr) -> Option<Currency> {
1820 self.cache().currency(code).copied()
1821 }
1822
1823 pub fn try_currency(&self, code: &Ustr) -> Result<Currency, CurrencyLookupError> {
1834 self.cache().try_currency(code).copied()
1835 }
1836
1837 #[must_use]
1843 pub fn instrument(&self, instrument_id: &InstrumentId) -> Option<InstrumentAny> {
1844 self.cache().instrument(instrument_id).cloned()
1845 }
1846
1847 pub fn try_instrument(
1858 &self,
1859 instrument_id: &InstrumentId,
1860 ) -> Result<InstrumentAny, InstrumentLookupError> {
1861 self.cache().try_instrument(instrument_id).cloned()
1862 }
1863
1864 #[must_use]
1870 pub fn instrument_ids(&self, venue: Option<&Venue>) -> Vec<InstrumentId> {
1871 self.cache()
1872 .instrument_ids(venue)
1873 .into_iter()
1874 .copied()
1875 .collect()
1876 }
1877
1878 #[must_use]
1884 pub fn instruments(&self, venue: &Venue, underlying: Option<&Ustr>) -> Vec<InstrumentAny> {
1885 self.cache()
1886 .instruments(venue, underlying)
1887 .into_iter()
1888 .cloned()
1889 .collect()
1890 }
1891
1892 #[must_use]
1899 pub fn instruments_by_parent(
1900 &self,
1901 venue: &Venue,
1902 root: &Ustr,
1903 class: InstrumentClass,
1904 ) -> Vec<InstrumentAny> {
1905 self.cache()
1906 .instruments_by_parent(venue, root, class)
1907 .into_iter()
1908 .cloned()
1909 .collect()
1910 }
1911
1912 #[must_use]
1918 pub fn bar_types(
1919 &self,
1920 instrument_id: Option<&InstrumentId>,
1921 price_type: Option<&PriceType>,
1922 aggregation_source: AggregationSource,
1923 ) -> Vec<BarType> {
1924 self.cache()
1925 .bar_types(instrument_id, price_type, aggregation_source)
1926 .into_iter()
1927 .copied()
1928 .collect()
1929 }
1930
1931 #[must_use]
1937 pub fn synthetic(&self, instrument_id: &InstrumentId) -> Option<SyntheticInstrument> {
1938 self.cache().synthetic(instrument_id).cloned()
1939 }
1940
1941 pub fn try_synthetic(
1953 &self,
1954 instrument_id: &InstrumentId,
1955 ) -> Result<SyntheticInstrument, SyntheticInstrumentLookupError> {
1956 self.cache().try_synthetic(instrument_id).cloned()
1957 }
1958
1959 #[must_use]
1965 pub fn synthetic_ids(&self) -> Vec<InstrumentId> {
1966 self.cache().synthetic_ids().into_iter().copied().collect()
1967 }
1968
1969 #[must_use]
1975 pub fn synthetics(&self) -> Vec<SyntheticInstrument> {
1976 self.cache().synthetics().into_iter().cloned().collect()
1977 }
1978
1979 #[cfg(feature = "defi")]
1985 #[must_use]
1986 pub fn pool(&self, instrument_id: &InstrumentId) -> Option<Pool> {
1987 self.cache().pool(instrument_id).cloned()
1988 }
1989
1990 #[cfg(feature = "defi")]
1996 #[must_use]
1997 pub fn pool_ids(&self, venue: Option<&Venue>) -> Vec<InstrumentId> {
1998 self.cache().pool_ids(venue)
1999 }
2000
2001 #[cfg(feature = "defi")]
2007 #[must_use]
2008 pub fn pools(&self, venue: Option<&Venue>) -> Vec<Pool> {
2009 self.cache().pools(venue).into_iter().cloned().collect()
2010 }
2011
2012 #[cfg(feature = "defi")]
2018 #[must_use]
2019 pub fn pool_profiler(&self, instrument_id: &InstrumentId) -> Option<PoolProfiler> {
2020 self.cache().pool_profiler(instrument_id).cloned()
2021 }
2022
2023 #[cfg(feature = "defi")]
2029 #[must_use]
2030 pub fn pool_profiler_ids(&self, venue: Option<&Venue>) -> Vec<InstrumentId> {
2031 self.cache().pool_profiler_ids(venue)
2032 }
2033
2034 #[cfg(feature = "defi")]
2040 #[must_use]
2041 pub fn pool_profilers(&self, venue: Option<&Venue>) -> Vec<PoolProfiler> {
2042 self.cache()
2043 .pool_profilers(venue)
2044 .into_iter()
2045 .cloned()
2046 .collect()
2047 }
2048
2049 #[must_use]
2055 pub fn account(&self, account_id: &AccountId) -> Option<AccountAny> {
2056 self.cache().account_owned(account_id)
2057 }
2058
2059 pub fn try_account(&self, account_id: &AccountId) -> Result<AccountAny, AccountLookupError> {
2070 self.cache()
2071 .try_account(account_id)
2072 .map(|account| account.cloned())
2073 }
2074
2075 #[must_use]
2081 pub fn account_for_venue(&self, venue: &Venue) -> Option<AccountAny> {
2082 self.cache().account_for_venue_owned(venue)
2083 }
2084
2085 #[must_use]
2091 pub fn account_id(&self, venue: &Venue) -> Option<AccountId> {
2092 self.cache().account_id(venue).copied()
2093 }
2094
2095 #[must_use]
2101 pub fn accounts(&self, account_id: &AccountId) -> Vec<AccountAny> {
2102 self.cache()
2103 .accounts(account_id)
2104 .into_iter()
2105 .map(|account| account.cloned())
2106 .collect()
2107 }
2108
2109 #[must_use]
2115 pub fn accounts_all(&self) -> Vec<AccountAny> {
2116 self.cache().accounts_all_owned()
2117 }
2118
2119 fn cache(&self) -> Ref<'_, Cache> {
2120 self.cache.borrow()
2121 }
2122}
2123
2124enum FilterSources<'a, K> {
2131 Unfiltered,
2132 Empty,
2133 Sets(Vec<&'a AHashSet<K>>),
2134}
2135
2136fn intersect_filter_sources<K>(mut sources: Vec<&AHashSet<K>>) -> AHashSet<K>
2142where
2143 K: Copy + Eq + std::hash::Hash,
2144{
2145 debug_assert!(!sources.is_empty());
2146 sources.sort_unstable_by_key(|s| s.len());
2147 let driver = sources[0];
2148 let rest = &sources[1..];
2149
2150 if rest.is_empty() {
2151 return driver.clone();
2152 }
2153
2154 driver
2155 .iter()
2156 .filter(|id| rest.iter().all(|s| s.contains(id)))
2157 .copied()
2158 .collect()
2159}
2160
2161fn intersect_pair_or_many<'a, K>(
2169 bucket: &'a AHashSet<K>,
2170 mut sources: Vec<&'a AHashSet<K>>,
2171) -> AHashSet<K>
2172where
2173 K: Copy + Eq + std::hash::Hash,
2174{
2175 debug_assert!(!sources.is_empty());
2176 if sources.len() == 1 {
2177 let filter = sources[0];
2178 let (larger, smaller) = if bucket.len() >= filter.len() {
2179 (bucket, filter)
2180 } else {
2181 (filter, bucket)
2182 };
2183 return larger.intersection(smaller).copied().collect();
2184 }
2185
2186 sources.push(bucket);
2187 intersect_filter_sources(sources)
2188}
2189
2190#[cfg_attr(
2192 feature = "python",
2193 pyo3::pyclass(module = "nautilus_trader.common", unsendable)
2194)]
2195pub struct Cache {
2196 config: CacheConfig,
2197 index: CacheIndex,
2198 database: Option<Box<dyn CacheDatabaseAdapter>>,
2199 general: AHashMap<String, Bytes>,
2200 currencies: AHashMap<Ustr, Currency>,
2201 instruments: AHashMap<InstrumentId, InstrumentAny>,
2202 instrument_closes: AHashMap<InstrumentId, InstrumentClose>,
2203 synthetics: AHashMap<InstrumentId, SyntheticInstrument>,
2204 books: AHashMap<InstrumentId, OrderBook>,
2205 own_books: AHashMap<InstrumentId, OwnOrderBook>,
2206 quotes: AHashMap<InstrumentId, BoundedVecDeque<QuoteTick>>,
2207 trades: AHashMap<InstrumentId, BoundedVecDeque<TradeTick>>,
2208 mark_xrates: AHashMap<(Currency, Currency), f64>,
2209 mark_prices: AHashMap<InstrumentId, BoundedVecDeque<MarkPriceUpdate>>,
2210 index_prices: AHashMap<InstrumentId, BoundedVecDeque<IndexPriceUpdate>>,
2211 funding_rates: AHashMap<InstrumentId, BoundedVecDeque<FundingRateUpdate>>,
2212 instrument_statuses: AHashMap<InstrumentId, BoundedVecDeque<InstrumentStatus>>,
2213 bars: AHashMap<BarType, BoundedVecDeque<Bar>>,
2214 greeks: AHashMap<InstrumentId, GreeksData>,
2215 option_greeks: AHashMap<InstrumentId, OptionGreeks>,
2216 yield_curves: AHashMap<String, YieldCurveData>,
2217 external_order_claims: AHashMap<InstrumentId, StrategyId>,
2218 accounts: AHashMap<AccountId, SharedCell<AccountAny>>,
2219 orders: AHashMap<ClientOrderId, SharedCell<OrderAny>>,
2220 order_lists: AHashMap<OrderListId, OrderList>,
2221 positions: AHashMap<PositionId, SharedCell<Position>>,
2222 position_snapshots: AHashMap<PositionId, Vec<PositionSnapshotFrame>>,
2223 position_snapshot_revisions: AHashMap<PositionId, u64>,
2224 #[cfg(feature = "defi")]
2225 pub(crate) defi: crate::defi::cache::DefiCache,
2226}
2227
2228impl Debug for Cache {
2229 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2230 f.debug_struct(stringify!(Cache))
2231 .field("config", &self.config)
2232 .field("index", &self.index)
2233 .field("general", &self.general)
2234 .field("currencies", &self.currencies)
2235 .field("instruments", &self.instruments)
2236 .field("synthetics", &self.synthetics)
2237 .field("books", &self.books)
2238 .field("own_books", &self.own_books)
2239 .field("quotes", &self.quotes)
2240 .field("trades", &self.trades)
2241 .field("mark_xrates", &self.mark_xrates)
2242 .field("mark_prices", &self.mark_prices)
2243 .field("index_prices", &self.index_prices)
2244 .field("funding_rates", &self.funding_rates)
2245 .field("instrument_statuses", &self.instrument_statuses)
2246 .field("instrument_closes", &self.instrument_closes)
2247 .field("bars", &self.bars)
2248 .field("greeks", &self.greeks)
2249 .field("option_greeks", &self.option_greeks)
2250 .field("yield_curves", &self.yield_curves)
2251 .field("external_order_claims", &self.external_order_claims)
2252 .field("accounts", &self.accounts)
2253 .field("orders", &self.orders)
2254 .field("order_lists", &self.order_lists)
2255 .field("positions", &self.positions)
2256 .field("position_snapshots", &self.position_snapshots)
2257 .finish()
2258 }
2259}
2260
2261impl Default for Cache {
2262 fn default() -> Self {
2264 Self::new(Some(CacheConfig::default()), None)
2265 }
2266}
2267
2268impl Cache {
2269 #[must_use]
2271 pub fn new(
2279 config: Option<CacheConfig>,
2280 database: Option<Box<dyn CacheDatabaseAdapter>>,
2281 ) -> Self {
2282 Self::try_new(config, database).expect("invalid `CacheConfig`")
2283 }
2284
2285 pub fn try_new(
2291 config: Option<CacheConfig>,
2292 database: Option<Box<dyn CacheDatabaseAdapter>>,
2293 ) -> crate::config::ConfigResult<Self> {
2294 let config = config.unwrap_or_default();
2295 config.validate()?;
2296
2297 Ok(Self {
2298 config,
2299 index: CacheIndex::default(),
2300 database,
2301 general: AHashMap::new(),
2302 currencies: AHashMap::new(),
2303 instruments: AHashMap::new(),
2304 instrument_closes: AHashMap::new(),
2305 synthetics: AHashMap::new(),
2306 books: AHashMap::new(),
2307 own_books: AHashMap::new(),
2308 quotes: AHashMap::new(),
2309 trades: AHashMap::new(),
2310 mark_xrates: AHashMap::new(),
2311 mark_prices: AHashMap::new(),
2312 index_prices: AHashMap::new(),
2313 funding_rates: AHashMap::new(),
2314 instrument_statuses: AHashMap::new(),
2315 bars: AHashMap::new(),
2316 greeks: AHashMap::new(),
2317 option_greeks: AHashMap::new(),
2318 yield_curves: AHashMap::new(),
2319 external_order_claims: AHashMap::new(),
2320 accounts: AHashMap::new(),
2321 orders: AHashMap::new(),
2322 order_lists: AHashMap::new(),
2323 positions: AHashMap::new(),
2324 position_snapshots: AHashMap::new(),
2325 position_snapshot_revisions: AHashMap::new(),
2326 #[cfg(feature = "defi")]
2327 defi: crate::defi::cache::DefiCache::default(),
2328 })
2329 }
2330
2331 #[must_use]
2333 pub fn memory_address(&self) -> String {
2334 format!("{:?}", std::ptr::from_ref(self))
2335 }
2336
2337 #[must_use]
2339 pub fn external_order_claim(&self, instrument_id: &InstrumentId) -> Option<StrategyId> {
2340 self.external_order_claims.get(instrument_id).copied()
2341 }
2342
2343 #[must_use]
2348 pub fn external_order_claim_instrument_ids(
2349 &self,
2350 strategy_id: Option<StrategyId>,
2351 ) -> AHashSet<InstrumentId> {
2352 self.external_order_claims
2353 .iter()
2354 .filter_map(|(instrument_id, owner)| {
2355 strategy_id
2356 .is_none_or(|strategy_id| *owner == strategy_id)
2357 .then_some(*instrument_id)
2358 })
2359 .collect()
2360 }
2361
2362 pub fn set_external_order_claims(
2374 &mut self,
2375 strategy_id: StrategyId,
2376 instrument_ids: &[InstrumentId],
2377 ) -> anyhow::Result<()> {
2378 let mut requested = AHashSet::with_capacity(instrument_ids.len());
2379
2380 for instrument_id in instrument_ids {
2381 if !requested.insert(*instrument_id) {
2382 anyhow::bail!(
2383 "External order claim for {instrument_id} appears more than once for {strategy_id}"
2384 );
2385 }
2386
2387 if let Some(existing) = self.external_order_claims.get(instrument_id)
2388 && *existing != strategy_id
2389 {
2390 anyhow::bail!(
2391 "External order claim for {instrument_id} already exists for {existing}"
2392 );
2393 }
2394 }
2395
2396 self.external_order_claims
2397 .retain(|_, owner| *owner != strategy_id);
2398 self.external_order_claims.extend(
2399 requested
2400 .into_iter()
2401 .map(|instrument_id| (instrument_id, strategy_id)),
2402 );
2403
2404 Ok(())
2405 }
2406
2407 pub fn register_external_order_claims(
2413 &mut self,
2414 strategy_id: StrategyId,
2415 instrument_ids: &[InstrumentId],
2416 ) -> anyhow::Result<()> {
2417 let mut requested = AHashSet::with_capacity(instrument_ids.len());
2418
2419 for instrument_id in instrument_ids {
2420 if !requested.insert(*instrument_id) {
2421 anyhow::bail!(
2422 "External order claim for {instrument_id} appears more than once for {strategy_id}"
2423 );
2424 }
2425
2426 if let Some(existing) = self.external_order_claims.get(instrument_id) {
2427 anyhow::bail!(
2428 "External order claim for {instrument_id} already exists for {existing}"
2429 );
2430 }
2431 }
2432
2433 self.external_order_claims.extend(
2434 requested
2435 .into_iter()
2436 .map(|instrument_id| (instrument_id, strategy_id)),
2437 );
2438
2439 Ok(())
2440 }
2441
2442 pub fn set_database(&mut self, database: Box<dyn CacheDatabaseAdapter>) {
2446 let type_name = std::any::type_name_of_val(&*database);
2447 log::info!("Cache database adapter set: {type_name}");
2448 self.database = Some(database);
2449 }
2450
2451 pub fn cache_general(&mut self) -> anyhow::Result<()> {
2459 self.general = match &mut self.database {
2460 Some(db) => db.load()?,
2461 None => AHashMap::new(),
2462 };
2463
2464 log::info!(
2465 "Cached {} general object(s) from database",
2466 self.general.len()
2467 );
2468 Ok(())
2469 }
2470
2471 pub async fn cache_all(&mut self) -> anyhow::Result<()> {
2480 let cache_map = match &self.database {
2481 Some(db) => db.load_all().await?,
2482 None => CacheMap::default(),
2483 };
2484
2485 self.currencies = cache_map.currencies;
2486 self.instruments = cache_map.instruments;
2487 self.instrument_closes = cache_map.instrument_closes;
2488 self.synthetics = cache_map.synthetics;
2489 self.accounts = cache_map
2490 .accounts
2491 .into_iter()
2492 .map(|(id, account)| (id, SharedCell::new(account)))
2493 .collect();
2494 self.orders = cache_map
2495 .orders
2496 .into_iter()
2497 .map(|(id, order)| (id, SharedCell::new(order)))
2498 .collect();
2499 self.positions = cache_map
2500 .positions
2501 .into_iter()
2502 .map(|(id, position)| (id, SharedCell::new(position)))
2503 .collect();
2504
2505 if let Some(db) = &self.database {
2506 let order_position = db.load_index_order_position()?;
2507 self.index.order_position = self.sanitize_order_position_index(order_position);
2508 self.index.order_client = db.load_index_order_client()?;
2509 }
2510
2511 self.cache_position_oms()?;
2512 self.assign_position_ids_to_contingencies();
2513 Ok(())
2514 }
2515
2516 pub async fn cache_currencies(&mut self) -> anyhow::Result<()> {
2522 self.currencies = match &mut self.database {
2523 Some(db) => db.load_currencies().await?,
2524 None => AHashMap::new(),
2525 };
2526
2527 log::info!("Cached {} currencies from database", self.general.len());
2528 Ok(())
2529 }
2530
2531 pub async fn cache_instruments(&mut self) -> anyhow::Result<()> {
2537 self.instruments = match &mut self.database {
2538 Some(db) => db.load_instruments().await?,
2539 None => AHashMap::new(),
2540 };
2541
2542 log::info!("Cached {} instruments from database", self.general.len());
2543 Ok(())
2544 }
2545
2546 pub async fn cache_synthetics(&mut self) -> anyhow::Result<()> {
2552 self.synthetics = match &mut self.database {
2553 Some(db) => db.load_synthetics().await?,
2554 None => AHashMap::new(),
2555 };
2556
2557 log::info!(
2558 "Cached {} synthetic instruments from database",
2559 self.general.len()
2560 );
2561 Ok(())
2562 }
2563
2564 pub async fn cache_accounts(&mut self) -> anyhow::Result<()> {
2570 self.accounts = match &mut self.database {
2571 Some(db) => db
2572 .load_accounts()
2573 .await?
2574 .into_iter()
2575 .map(|(id, account)| (id, SharedCell::new(account)))
2576 .collect(),
2577 None => AHashMap::new(),
2578 };
2579
2580 log::info!(
2581 "Cached {} synthetic instruments from database",
2582 self.general.len()
2583 );
2584 Ok(())
2585 }
2586
2587 pub async fn cache_orders(&mut self) -> anyhow::Result<()> {
2593 self.orders = match &mut self.database {
2594 Some(db) => db
2595 .load_orders()
2596 .await?
2597 .into_iter()
2598 .map(|(id, order)| (id, SharedCell::new(order)))
2599 .collect(),
2600 None => AHashMap::new(),
2601 };
2602
2603 if let Some(db) = &self.database {
2604 let order_position = db.load_index_order_position()?;
2605 self.index.order_position = self.sanitize_order_position_index(order_position);
2606 self.index.order_client = db.load_index_order_client()?;
2607 }
2608
2609 log::info!("Cached {} orders from database", self.general.len());
2610
2611 self.assign_position_ids_to_contingencies();
2612 Ok(())
2613 }
2614
2615 fn sanitize_order_position_index(
2616 &self,
2617 mut order_position: AHashMap<ClientOrderId, PositionId>,
2618 ) -> AHashMap<ClientOrderId, PositionId> {
2619 let original_len = order_position.len();
2620 order_position.retain(|client_order_id, _| self.orders.contains_key(client_order_id));
2621 let removed = original_len - order_position.len();
2622
2623 if removed > 0 {
2624 log::warn!(
2625 "Filtered {removed} stale order-position index entries without backing orders during cache load"
2626 );
2627 }
2628
2629 order_position
2630 }
2631
2632 pub async fn cache_positions(&mut self) -> anyhow::Result<()> {
2638 self.positions = match &mut self.database {
2639 Some(db) => db
2640 .load_positions()
2641 .await?
2642 .into_iter()
2643 .map(|(id, position)| (id, SharedCell::new(position)))
2644 .collect(),
2645 None => AHashMap::new(),
2646 };
2647
2648 self.cache_position_oms()?;
2649 log::info!("Cached {} positions from database", self.general.len());
2650 Ok(())
2651 }
2652
2653 fn cache_position_oms(&mut self) -> anyhow::Result<()> {
2654 let persisted = match &self.database {
2655 Some(database) => database.load()?,
2656 None => self.general.clone(),
2657 };
2658
2659 self.general
2660 .retain(|key, _| !key.starts_with(POSITION_OMS_KEY_PREFIX));
2661
2662 for (key, value) in persisted {
2663 if !key.starts_with(POSITION_OMS_KEY_PREFIX) {
2664 continue;
2665 }
2666 self.general.insert(key, value);
2667 }
2668
2669 self.index_position_oms();
2670 Ok(())
2671 }
2672
2673 pub fn build_index(&mut self) {
2675 log::debug!("Building index");
2676
2677 for account_id in self.accounts.keys() {
2679 self.index
2680 .venue_account
2681 .insert(account_id.get_issuer(), *account_id);
2682 }
2683
2684 for (client_order_id, order_cell) in &self.orders {
2686 let order = order_cell.borrow();
2687 let instrument_id = order.instrument_id();
2688 let venue = instrument_id.venue;
2689 let strategy_id = order.strategy_id();
2690
2691 self.index
2693 .venue_orders
2694 .entry(venue)
2695 .or_default()
2696 .insert(*client_order_id);
2697
2698 if let Some(venue_order_id) = order.venue_order_id() {
2701 self.index
2702 .venue_order_ids
2703 .insert(venue_order_id, *client_order_id);
2704 self.index
2705 .client_order_ids
2706 .insert(*client_order_id, venue_order_id);
2707 }
2708
2709 if let Some(position_id) = order.position_id() {
2711 self.index
2712 .order_position
2713 .insert(*client_order_id, position_id);
2714 }
2715
2716 self.index
2718 .order_strategy
2719 .insert(*client_order_id, strategy_id);
2720
2721 self.index
2723 .instrument_orders
2724 .entry(instrument_id)
2725 .or_default()
2726 .insert(*client_order_id);
2727
2728 self.index
2730 .strategy_orders
2731 .entry(strategy_id)
2732 .or_default()
2733 .insert(*client_order_id);
2734
2735 if let Some(account_id) = order.account_id() {
2737 self.index
2738 .account_orders
2739 .entry(account_id)
2740 .or_default()
2741 .insert(*client_order_id);
2742 }
2743
2744 if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
2746 self.index
2747 .exec_algorithm_orders
2748 .entry(exec_algorithm_id)
2749 .or_default()
2750 .insert(*client_order_id);
2751 self.index.exec_algorithms.insert(exec_algorithm_id);
2752 }
2753
2754 if let Some(exec_spawn_id) = order.exec_spawn_id() {
2756 self.index
2757 .exec_spawn_orders
2758 .entry(exec_spawn_id)
2759 .or_default()
2760 .insert(*client_order_id);
2761 }
2762
2763 self.index.orders.insert(*client_order_id);
2765
2766 if order.is_active_local() {
2768 self.index.orders_active_local.insert(*client_order_id);
2769 }
2770
2771 if order.is_open() {
2773 self.index.orders_open.insert(*client_order_id);
2774 }
2775
2776 if order.is_closed() {
2778 self.index.orders_closed.insert(*client_order_id);
2779 }
2780
2781 if order.emulation_trigger().is_some() && !order.is_closed() {
2783 self.index.orders_emulated.insert(*client_order_id);
2784 }
2785
2786 if order.is_inflight() {
2788 self.index.orders_inflight.insert(*client_order_id);
2789 }
2790
2791 self.index.strategies.insert(strategy_id);
2793 }
2794
2795 for (position_id, position_cell) in &self.positions {
2797 let position = position_cell.borrow();
2798 let instrument_id = position.instrument_id;
2799 let venue = instrument_id.venue;
2800 let strategy_id = position.strategy_id;
2801
2802 self.index
2804 .venue_positions
2805 .entry(venue)
2806 .or_default()
2807 .insert(*position_id);
2808
2809 self.index
2811 .position_strategy
2812 .insert(*position_id, strategy_id);
2813
2814 let position_orders = self.index.position_orders.entry(*position_id).or_default();
2816 position_orders.extend(
2817 position
2818 .client_order_ids()
2819 .into_iter()
2820 .filter(|client_order_id| self.orders.contains_key(client_order_id)),
2821 );
2822
2823 self.index
2825 .instrument_positions
2826 .entry(instrument_id)
2827 .or_default()
2828 .insert(*position_id);
2829 self.index
2830 .instrument_orders
2831 .entry(instrument_id)
2832 .or_default();
2833
2834 self.index
2836 .strategy_positions
2837 .entry(strategy_id)
2838 .or_default()
2839 .insert(*position_id);
2840 self.index.strategy_orders.entry(strategy_id).or_default();
2841
2842 self.index
2844 .account_positions
2845 .entry(position.account_id)
2846 .or_default()
2847 .insert(*position_id);
2848
2849 self.index.positions.insert(*position_id);
2851
2852 if position.is_open() {
2854 self.index.positions_open.insert(*position_id);
2855 }
2856
2857 if position.is_closed() {
2859 self.index.positions_closed.insert(*position_id);
2860 }
2861
2862 self.index.strategies.insert(strategy_id);
2864 }
2865
2866 self.index_position_oms();
2867 }
2868
2869 fn index_position_oms(&mut self) {
2870 self.index.position_oms.clear();
2871
2872 for (key, value) in &self.general {
2873 let Some(position_id) = key.strip_prefix(POSITION_OMS_KEY_PREFIX) else {
2874 continue;
2875 };
2876 let position_id = PositionId::new(position_id);
2877 if !self.positions.contains_key(&position_id) {
2878 continue;
2879 }
2880
2881 match serde_json::from_slice::<OmsType>(value) {
2882 Ok(oms_type) => {
2883 self.index.position_oms.insert(position_id, oms_type);
2884 }
2885 Err(e) => {
2886 log::error!("Failed to decode position OMS for {position_id}: {e}");
2887 }
2888 }
2889 }
2890
2891 for position in self.positions.values().map(|cell| cell.borrow()) {
2892 if !self.index.position_oms.contains_key(&position.id)
2893 && position.id.as_str()
2894 == format!("{}-{}", position.instrument_id, position.strategy_id)
2895 {
2896 self.index
2897 .position_oms
2898 .insert(position.id, OmsType::Netting);
2899 }
2900 }
2901 }
2902
2903 #[must_use]
2905 pub const fn has_backing(&self) -> bool {
2906 self.database.is_some()
2907 }
2908
2909 pub fn load_actor_state(
2917 &self,
2918 actor_id: &ActorId,
2919 ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
2920 self.database
2921 .as_ref()
2922 .map(|database| database.load_actor(actor_id))
2923 .transpose()
2924 .map(|state| state.map(Self::decode_component_state))
2925 }
2926
2927 pub fn load_strategy_state(
2935 &self,
2936 strategy_id: &StrategyId,
2937 ) -> anyhow::Result<Option<IndexMap<String, Vec<u8>>>> {
2938 self.database
2939 .as_ref()
2940 .map(|database| database.load_strategy(strategy_id))
2941 .transpose()
2942 .map(|state| state.map(Self::decode_component_state))
2943 }
2944
2945 pub fn update_actor_state(
2951 &self,
2952 actor_id: &ActorId,
2953 state: &IndexMap<String, Vec<u8>>,
2954 ) -> anyhow::Result<()> {
2955 if let Some(database) = &self.database {
2956 database.update_actor(actor_id, &Self::encode_component_state(state))?;
2957 }
2958 Ok(())
2959 }
2960
2961 pub fn update_strategy_state(
2967 &self,
2968 strategy_id: &StrategyId,
2969 state: &IndexMap<String, Vec<u8>>,
2970 ) -> anyhow::Result<()> {
2971 if let Some(database) = &self.database {
2972 database.update_strategy(strategy_id, &Self::encode_component_state(state))?;
2973 }
2974 Ok(())
2975 }
2976
2977 fn decode_component_state(state: AHashMap<String, Bytes>) -> IndexMap<String, Vec<u8>> {
2978 state
2979 .into_iter()
2980 .map(|(key, value)| (key, value.to_vec()))
2981 .collect()
2982 }
2983
2984 fn encode_component_state(state: &IndexMap<String, Vec<u8>>) -> AHashMap<String, Bytes> {
2985 state
2986 .iter()
2987 .map(|(key, value)| (key.clone(), Bytes::copy_from_slice(value)))
2988 .collect()
2989 }
2990
2991 #[must_use]
2993 pub fn calculate_unrealized_pnl(&self, position: &Position) -> Option<Money> {
2994 let Some(quote) = self.quote(&position.instrument_id) else {
2995 log::warn!(
2996 "Cannot calculate unrealized PnL for {}, no quotes for {}",
2997 position.id,
2998 position.instrument_id
2999 );
3000 return None;
3001 };
3002
3003 let last = match position.side {
3005 PositionSide::Flat => {
3006 return Some(Money::zero(position.settlement_currency));
3007 }
3008 PositionSide::Long => quote.bid_price,
3009 PositionSide::Short => quote.ask_price,
3010 };
3011
3012 position
3013 .try_unrealized_pnl(last)
3014 .inspect_err(|e| {
3015 log::error!("Cannot calculate unrealized PnL for {}: {e}", position.id);
3016 })
3017 .ok()
3018 }
3019
3020 #[must_use]
3029 pub fn check_integrity(&mut self) -> bool {
3030 let mut error_count = 0;
3031 let failure = "Integrity failure";
3032
3033 let timestamp_us = SystemTime::now()
3035 .duration_since(UNIX_EPOCH)
3036 .expect("Time went backwards")
3037 .as_micros();
3038
3039 log::info!("Checking data integrity");
3040
3041 for account_id in self.accounts.keys() {
3043 if !self
3044 .index
3045 .venue_account
3046 .contains_key(&account_id.get_issuer())
3047 {
3048 log::error!(
3049 "{failure} in accounts: {account_id} not found in `self.index.venue_account`",
3050 );
3051 error_count += 1;
3052 }
3053 }
3054
3055 for (client_order_id, order_cell) in &self.orders {
3056 let order = order_cell.borrow();
3057
3058 if !self.index.order_strategy.contains_key(client_order_id) {
3059 log::error!(
3060 "{failure} in orders: {client_order_id} not found in `self.index.order_strategy`"
3061 );
3062 error_count += 1;
3063 }
3064
3065 if !self.index.orders.contains(client_order_id) {
3066 log::error!(
3067 "{failure} in orders: {client_order_id} not found in `self.index.orders`",
3068 );
3069 error_count += 1;
3070 }
3071
3072 if order.is_inflight() && !self.index.orders_inflight.contains(client_order_id) {
3073 log::error!(
3074 "{failure} in orders: {client_order_id} not found in `self.index.orders_inflight`",
3075 );
3076 error_count += 1;
3077 }
3078
3079 if order.is_active_local() && !self.index.orders_active_local.contains(client_order_id)
3080 {
3081 log::error!(
3082 "{failure} in orders: {client_order_id} not found in `self.index.orders_active_local`",
3083 );
3084 error_count += 1;
3085 }
3086
3087 if order.is_open() && !self.index.orders_open.contains(client_order_id) {
3088 log::error!(
3089 "{failure} in orders: {client_order_id} not found in `self.index.orders_open`",
3090 );
3091 error_count += 1;
3092 }
3093
3094 if order.is_closed() && !self.index.orders_closed.contains(client_order_id) {
3095 log::error!(
3096 "{failure} in orders: {client_order_id} not found in `self.index.orders_closed`",
3097 );
3098 error_count += 1;
3099 }
3100
3101 if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
3102 if !self
3103 .index
3104 .exec_algorithm_orders
3105 .contains_key(&exec_algorithm_id)
3106 {
3107 log::error!(
3108 "{failure} in orders: {client_order_id} not found in `self.index.exec_algorithm_orders`",
3109 );
3110 error_count += 1;
3111 }
3112
3113 if order.exec_spawn_id().is_none()
3114 && !self.index.exec_spawn_orders.contains_key(client_order_id)
3115 {
3116 log::error!(
3117 "{failure} in orders: {client_order_id} not found in `self.index.exec_spawn_orders`",
3118 );
3119 error_count += 1;
3120 }
3121 }
3122 }
3123
3124 for (position_id, position_cell) in &self.positions {
3125 let position = position_cell.borrow();
3126
3127 if !self.index.position_strategy.contains_key(position_id) {
3128 log::error!(
3129 "{failure} in positions: {position_id} not found in `self.index.position_strategy`",
3130 );
3131 error_count += 1;
3132 }
3133
3134 if !self.index.position_orders.contains_key(position_id) {
3135 log::error!(
3136 "{failure} in positions: {position_id} not found in `self.index.position_orders`",
3137 );
3138 error_count += 1;
3139 }
3140
3141 if !self.index.positions.contains(position_id) {
3142 log::error!(
3143 "{failure} in positions: {position_id} not found in `self.index.positions`",
3144 );
3145 error_count += 1;
3146 }
3147
3148 if position.is_open() && !self.index.positions_open.contains(position_id) {
3149 log::error!(
3150 "{failure} in positions: {position_id} not found in `self.index.positions_open`",
3151 );
3152 error_count += 1;
3153 }
3154
3155 if position.is_closed() && !self.index.positions_closed.contains(position_id) {
3156 log::error!(
3157 "{failure} in positions: {position_id} not found in `self.index.positions_closed`",
3158 );
3159 error_count += 1;
3160 }
3161 }
3162
3163 for account_id in self.index.venue_account.values() {
3165 if !self.accounts.contains_key(account_id) {
3166 log::error!(
3167 "{failure} in `index.venue_account`: {account_id} not found in `self.accounts`",
3168 );
3169 error_count += 1;
3170 }
3171 }
3172
3173 for client_order_id in self.index.venue_order_ids.values() {
3174 if !self.orders.contains_key(client_order_id) {
3175 log::error!(
3176 "{failure} in `index.venue_order_ids`: {client_order_id} not found in `self.orders`",
3177 );
3178 error_count += 1;
3179 }
3180 }
3181
3182 for client_order_id in self.index.client_order_ids.keys() {
3183 if !self.orders.contains_key(client_order_id) {
3184 log::error!(
3185 "{failure} in `index.client_order_ids`: {client_order_id} not found in `self.orders`",
3186 );
3187 error_count += 1;
3188 }
3189 }
3190
3191 for client_order_id in self.index.order_position.keys() {
3192 if !self.orders.contains_key(client_order_id) {
3193 log::error!(
3194 "{failure} in `index.order_position`: {client_order_id} not found in `self.orders`",
3195 );
3196 error_count += 1;
3197 }
3198 }
3199
3200 for client_order_id in self.index.order_strategy.keys() {
3202 if !self.orders.contains_key(client_order_id) {
3203 log::error!(
3204 "{failure} in `index.order_strategy`: {client_order_id} not found in `self.orders`",
3205 );
3206 error_count += 1;
3207 }
3208 }
3209
3210 for position_id in self.index.position_strategy.keys() {
3211 if !self.positions.contains_key(position_id) {
3212 log::error!(
3213 "{failure} in `index.position_strategy`: {position_id} not found in `self.positions`",
3214 );
3215 error_count += 1;
3216 }
3217 }
3218
3219 for position_id in self.index.position_orders.keys() {
3220 if !self.positions.contains_key(position_id) {
3221 log::error!(
3222 "{failure} in `index.position_orders`: {position_id} not found in `self.positions`",
3223 );
3224 error_count += 1;
3225 }
3226 }
3227
3228 for (instrument_id, client_order_ids) in &self.index.instrument_orders {
3229 for client_order_id in client_order_ids {
3230 if !self.orders.contains_key(client_order_id) {
3231 log::error!(
3232 "{failure} in `index.instrument_orders`: {instrument_id} not found in `self.orders`",
3233 );
3234 error_count += 1;
3235 }
3236 }
3237 }
3238
3239 for instrument_id in self.index.instrument_positions.keys() {
3240 if !self.index.instrument_orders.contains_key(instrument_id) {
3241 log::error!(
3242 "{failure} in `index.instrument_positions`: {instrument_id} not found in `index.instrument_orders`",
3243 );
3244 error_count += 1;
3245 }
3246 }
3247
3248 for client_order_ids in self.index.strategy_orders.values() {
3249 for client_order_id in client_order_ids {
3250 if !self.orders.contains_key(client_order_id) {
3251 log::error!(
3252 "{failure} in `index.strategy_orders`: {client_order_id} not found in `self.orders`",
3253 );
3254 error_count += 1;
3255 }
3256 }
3257 }
3258
3259 for position_ids in self.index.strategy_positions.values() {
3260 for position_id in position_ids {
3261 if !self.positions.contains_key(position_id) {
3262 log::error!(
3263 "{failure} in `index.strategy_positions`: {position_id} not found in `self.positions`",
3264 );
3265 error_count += 1;
3266 }
3267 }
3268 }
3269
3270 for client_order_id in &self.index.orders {
3271 if !self.orders.contains_key(client_order_id) {
3272 log::error!(
3273 "{failure} in `index.orders`: {client_order_id} not found in `self.orders`",
3274 );
3275 error_count += 1;
3276 }
3277 }
3278
3279 for client_order_id in &self.index.orders_emulated {
3280 if !self.orders.contains_key(client_order_id) {
3281 log::error!(
3282 "{failure} in `index.orders_emulated`: {client_order_id} not found in `self.orders`",
3283 );
3284 error_count += 1;
3285 }
3286 }
3287
3288 for client_order_id in &self.index.orders_active_local {
3289 if !self.orders.contains_key(client_order_id) {
3290 log::error!(
3291 "{failure} in `index.orders_active_local`: {client_order_id} not found in `self.orders`",
3292 );
3293 error_count += 1;
3294 }
3295 }
3296
3297 for client_order_id in &self.index.orders_inflight {
3298 if !self.orders.contains_key(client_order_id) {
3299 log::error!(
3300 "{failure} in `index.orders_inflight`: {client_order_id} not found in `self.orders`",
3301 );
3302 error_count += 1;
3303 }
3304 }
3305
3306 for client_order_id in &self.index.orders_open {
3307 if !self.orders.contains_key(client_order_id) {
3308 log::error!(
3309 "{failure} in `index.orders_open`: {client_order_id} not found in `self.orders`",
3310 );
3311 error_count += 1;
3312 }
3313 }
3314
3315 for client_order_id in &self.index.orders_closed {
3316 if !self.orders.contains_key(client_order_id) {
3317 log::error!(
3318 "{failure} in `index.orders_closed`: {client_order_id} not found in `self.orders`",
3319 );
3320 error_count += 1;
3321 }
3322 }
3323
3324 for position_id in &self.index.positions {
3325 if !self.positions.contains_key(position_id) {
3326 log::error!(
3327 "{failure} in `index.positions`: {position_id} not found in `self.positions`",
3328 );
3329 error_count += 1;
3330 }
3331 }
3332
3333 for position_id in &self.index.positions_open {
3334 if !self.positions.contains_key(position_id) {
3335 log::error!(
3336 "{failure} in `index.positions_open`: {position_id} not found in `self.positions`",
3337 );
3338 error_count += 1;
3339 }
3340 }
3341
3342 for position_id in &self.index.positions_closed {
3343 if !self.positions.contains_key(position_id) {
3344 log::error!(
3345 "{failure} in `index.positions_closed`: {position_id} not found in `self.positions`",
3346 );
3347 error_count += 1;
3348 }
3349 }
3350
3351 for strategy_id in &self.index.strategies {
3352 if !self.index.strategy_orders.contains_key(strategy_id) {
3353 log::error!(
3354 "{failure} in `index.strategies`: {strategy_id} not found in `index.strategy_orders`",
3355 );
3356 error_count += 1;
3357 }
3358 }
3359
3360 for exec_algorithm_id in &self.index.exec_algorithms {
3361 if !self
3362 .index
3363 .exec_algorithm_orders
3364 .contains_key(exec_algorithm_id)
3365 {
3366 log::error!(
3367 "{failure} in `index.exec_algorithms`: {exec_algorithm_id} not found in `index.exec_algorithm_orders`",
3368 );
3369 error_count += 1;
3370 }
3371 }
3372
3373 let total_us = SystemTime::now()
3374 .duration_since(UNIX_EPOCH)
3375 .expect("Time went backwards")
3376 .as_micros()
3377 - timestamp_us;
3378
3379 if error_count == 0 {
3380 log::info!("Integrity check passed in {total_us}μs");
3381 true
3382 } else {
3383 log::error!(
3384 "Integrity check failed with {error_count} error{} in {total_us}μs",
3385 if error_count == 1 { "" } else { "s" },
3386 );
3387 false
3388 }
3389 }
3390
3391 #[must_use]
3395 pub fn check_residuals(&self) -> bool {
3396 log::debug!("Checking residuals");
3397
3398 let mut residuals = false;
3399
3400 for order in self.orders_open(None, None, None, None, None) {
3402 residuals = true;
3403 log::warn!("Residual {order}");
3404 }
3405
3406 for position in self.positions_open(None, None, None, None, None) {
3408 residuals = true;
3409 log::warn!("Residual {position}");
3410 }
3411
3412 residuals
3413 }
3414
3415 pub fn purge_closed_orders(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
3421 log::debug!(
3422 "Purging closed orders{}",
3423 if buffer_secs > 0 {
3424 format!(" with buffer_secs={buffer_secs}")
3425 } else {
3426 String::new()
3427 }
3428 );
3429
3430 let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
3431 log::warn!(
3432 "Cannot purge closed orders: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
3433 );
3434 return;
3435 };
3436 let purge_cutoff = ts_now.checked_sub(buffer_ns);
3437
3438 let mut affected_order_list_ids: AHashSet<OrderListId> = AHashSet::new();
3439 let mut purged_client_order_ids: AHashSet<ClientOrderId> = AHashSet::new();
3440
3441 'outer: for client_order_id in self.index.orders_closed.clone() {
3442 let purge_target = self.orders.get(&client_order_id).and_then(|order_cell| {
3443 let order = order_cell.borrow();
3444 if order.is_closed()
3445 && let Some(ts_closed) = order.ts_closed()
3446 && purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
3447 {
3448 let linked = order.linked_order_ids().map(<[_]>::to_vec);
3449 let order_list_id = order.order_list_id();
3450 Some((linked, order_list_id))
3451 } else {
3452 None
3453 }
3454 });
3455
3456 let Some((linked, order_list_id)) = purge_target else {
3457 continue;
3458 };
3459
3460 if let Some(linked_order_ids) = linked {
3462 for linked_order_id in &linked_order_ids {
3463 if let Some(linked_order_cell) = self.orders.get(linked_order_id)
3464 && linked_order_cell.borrow().is_open()
3465 {
3466 continue 'outer;
3468 }
3469 }
3470 }
3471
3472 if let Some(order_list_id) = order_list_id {
3473 affected_order_list_ids.insert(order_list_id);
3474 }
3475
3476 if self.purge_order_except_aliases(client_order_id) {
3477 purged_client_order_ids.insert(client_order_id);
3478 }
3479 }
3480
3481 if !purged_client_order_ids.is_empty() {
3482 self.index
3483 .venue_order_ids
3484 .retain(|_, owner| !purged_client_order_ids.contains(owner));
3485 }
3486
3487 for order_list_id in affected_order_list_ids {
3488 if let Some(order_list) = self.order_lists.get(&order_list_id) {
3489 let all_purged = order_list
3490 .client_order_ids
3491 .iter()
3492 .all(|id| !self.orders.contains_key(id));
3493
3494 if all_purged {
3495 self.order_lists.remove(&order_list_id);
3496 log::info!("Purged {order_list_id}");
3497 }
3498 }
3499 }
3500 }
3501
3502 pub fn purge_closed_positions(&mut self, ts_now: UnixNanos, buffer_secs: u64) {
3504 log::debug!(
3505 "Purging closed positions{}",
3506 if buffer_secs > 0 {
3507 format!(" with buffer_secs={buffer_secs}")
3508 } else {
3509 String::new()
3510 }
3511 );
3512
3513 let Ok(buffer_ns) = DurationNanos::try_from_secs(buffer_secs) else {
3514 log::warn!(
3515 "Cannot purge closed positions: buffer_secs {buffer_secs} is not representable in `u64` nanoseconds"
3516 );
3517 return;
3518 };
3519 let purge_cutoff = ts_now.checked_sub(buffer_ns);
3520
3521 for position_id in self.index.positions_closed.clone() {
3522 let should_purge = self.positions.get(&position_id).is_some_and(|cell| {
3523 let position = cell.borrow();
3524 position.is_closed()
3525 && position.ts_closed.is_some_and(|ts_closed| {
3526 purge_cutoff.is_some_and(|cutoff| ts_closed <= cutoff)
3527 })
3528 });
3529
3530 if should_purge {
3531 self.purge_position(position_id);
3532 }
3533 }
3534 }
3535
3536 pub fn purge_order(&mut self, client_order_id: ClientOrderId) {
3540 if self.purge_order_except_aliases(client_order_id) {
3541 self.index
3542 .venue_order_ids
3543 .retain(|_, owner| owner != &client_order_id);
3544 }
3545 }
3546
3547 fn purge_order_except_aliases(&mut self, client_order_id: ClientOrderId) -> bool {
3552 struct OrderDetails {
3553 is_open: bool,
3554 instrument_id: InstrumentId,
3555 strategy_id: StrategyId,
3556 account_id: Option<AccountId>,
3557 exec_algorithm_id: Option<ExecAlgorithmId>,
3558 exec_spawn_id: Option<ClientOrderId>,
3559 position_id: Option<PositionId>,
3560 }
3561
3562 let order_cell = self.orders.get(&client_order_id).cloned();
3563 let order_details = order_cell.as_ref().map(|cell| {
3564 let order = cell.borrow();
3565 OrderDetails {
3566 is_open: order.is_open(),
3567 instrument_id: order.instrument_id(),
3568 strategy_id: order.strategy_id(),
3569 account_id: order.account_id(),
3570 exec_algorithm_id: order.exec_algorithm_id(),
3571 exec_spawn_id: order.exec_spawn_id(),
3572 position_id: order.position_id(),
3573 }
3574 });
3575
3576 if order_details
3577 .as_ref()
3578 .is_some_and(|details| details.is_open)
3579 {
3580 log::warn!("Order {client_order_id} found open when purging, skipping purge");
3581 return false;
3582 }
3583
3584 if order_details.is_some() {
3585 self.orders.remove(&client_order_id);
3586 } else {
3587 log::warn!("Order {client_order_id} not found when purging");
3588 }
3589
3590 let indexed_position_id = self.index.order_position.remove(&client_order_id);
3591 let indexed_strategy_id = self.index.order_strategy.remove(&client_order_id);
3592 self.index.order_client.remove(&client_order_id);
3593 self.index.client_order_ids.remove(&client_order_id);
3594
3595 if let Some(details) = &order_details {
3596 if let Some(venue_orders) = self
3597 .index
3598 .venue_orders
3599 .get_mut(&details.instrument_id.venue)
3600 {
3601 venue_orders.remove(&client_order_id);
3602 if venue_orders.is_empty() {
3603 self.index.venue_orders.remove(&details.instrument_id.venue);
3604 }
3605 }
3606
3607 let instrument_orders_became_empty = self
3612 .index
3613 .instrument_orders
3614 .get_mut(&details.instrument_id)
3615 .is_some_and(|instrument_orders| {
3616 instrument_orders.remove(&client_order_id);
3617 instrument_orders.is_empty()
3618 });
3619
3620 let has_instrument_positions = self
3621 .index
3622 .instrument_positions
3623 .get(&details.instrument_id)
3624 .is_some_and(|positions| !positions.is_empty());
3625
3626 if instrument_orders_became_empty && !has_instrument_positions {
3627 self.index.instrument_orders.remove(&details.instrument_id);
3628 }
3629
3630 if let Some(exec_algorithm_id) = details.exec_algorithm_id {
3631 let became_empty = self
3632 .index
3633 .exec_algorithm_orders
3634 .get_mut(&exec_algorithm_id)
3635 .is_some_and(|orders| {
3636 orders.remove(&client_order_id);
3637 orders.is_empty()
3638 });
3639
3640 if became_empty {
3641 self.index.exec_algorithm_orders.remove(&exec_algorithm_id);
3642 self.index.exec_algorithms.remove(&exec_algorithm_id);
3643 }
3644 }
3645
3646 if let Some(account_id) = details.account_id
3647 && let Some(account_orders) = self.index.account_orders.get_mut(&account_id)
3648 {
3649 account_orders.remove(&client_order_id);
3650 if account_orders.is_empty() {
3651 self.index.account_orders.remove(&account_id);
3652 }
3653 }
3654
3655 if let Some(exec_spawn_id) = details.exec_spawn_id
3656 && let Some(spawn_orders) = self.index.exec_spawn_orders.get_mut(&exec_spawn_id)
3657 {
3658 spawn_orders.remove(&client_order_id);
3659 if spawn_orders.is_empty() {
3660 self.index.exec_spawn_orders.remove(&exec_spawn_id);
3661 }
3662 }
3663 }
3664
3665 let mut position_ids = AHashSet::new();
3666 if let Some(position_id) = indexed_position_id {
3667 position_ids.insert(position_id);
3668 }
3669
3670 if let Some(position_id) = order_details
3671 .as_ref()
3672 .and_then(|details| details.position_id)
3673 {
3674 position_ids.insert(position_id);
3675 }
3676
3677 let mut strategy_ids = AHashSet::new();
3678 if let Some(strategy_id) = indexed_strategy_id {
3679 strategy_ids.insert(strategy_id);
3680 }
3681
3682 if let Some(details) = &order_details {
3683 strategy_ids.insert(details.strategy_id);
3684 }
3685
3686 for position_id in position_ids {
3687 if self.positions.contains_key(&position_id) {
3688 if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
3689 position_orders.remove(&client_order_id);
3690 }
3691 continue;
3692 }
3693
3694 let has_other_orders =
3695 if let Some(position_orders) = self.index.position_orders.get_mut(&position_id) {
3696 position_orders.remove(&client_order_id);
3697 !position_orders.is_empty()
3698 } else {
3699 self.index
3700 .order_position
3701 .values()
3702 .any(|candidate| *candidate == position_id)
3703 };
3704
3705 if has_other_orders {
3706 continue;
3707 }
3708
3709 self.index.position_orders.remove(&position_id);
3710 if let Some(strategy_id) = self.index.position_strategy.remove(&position_id) {
3711 strategy_ids.insert(strategy_id);
3712 if let Some(strategy_positions) =
3713 self.index.strategy_positions.get_mut(&strategy_id)
3714 {
3715 strategy_positions.remove(&position_id);
3716 if strategy_positions.is_empty() {
3717 self.index.strategy_positions.remove(&strategy_id);
3718 }
3719 }
3720 }
3721
3722 if let Some(details) = &order_details
3723 && let Some(venue_positions) = self
3724 .index
3725 .venue_positions
3726 .get_mut(&details.instrument_id.venue)
3727 {
3728 venue_positions.remove(&position_id);
3729 if venue_positions.is_empty() {
3730 self.index
3731 .venue_positions
3732 .remove(&details.instrument_id.venue);
3733 }
3734 }
3735 }
3736
3737 for strategy_id in strategy_ids {
3738 let strategy_orders_became_empty = self
3744 .index
3745 .strategy_orders
3746 .get_mut(&strategy_id)
3747 .is_some_and(|strategy_orders| {
3748 strategy_orders.remove(&client_order_id);
3749 strategy_orders.is_empty()
3750 });
3751
3752 let has_positions = self
3753 .index
3754 .strategy_positions
3755 .get(&strategy_id)
3756 .is_some_and(|strategy_positions| !strategy_positions.is_empty());
3757
3758 if strategy_orders_became_empty && !has_positions {
3759 self.index.strategy_orders.remove(&strategy_id);
3760 self.index.strategies.remove(&strategy_id);
3761 }
3762 }
3763
3764 self.index.exec_spawn_orders.remove(&client_order_id);
3765
3766 self.index.orders.remove(&client_order_id);
3767 self.index.orders_active_local.remove(&client_order_id);
3768 self.index.orders_open.remove(&client_order_id);
3769 self.index.orders_closed.remove(&client_order_id);
3770 self.index.orders_emulated.remove(&client_order_id);
3771 self.index.orders_inflight.remove(&client_order_id);
3772 self.index.orders_pending_cancel.remove(&client_order_id);
3773
3774 if order_details.is_some() {
3775 log::info!("Purged order {client_order_id}");
3776 }
3777
3778 true
3779 }
3780
3781 pub fn purge_position(&mut self, position_id: PositionId) {
3785 let position = self
3787 .positions
3788 .get(&position_id)
3789 .map(|cell| cell.borrow().clone());
3790
3791 if let Some(ref pos) = position
3793 && pos.is_open()
3794 {
3795 log::warn!("Position {position_id} found open when purging, skipping purge");
3796 return;
3797 }
3798
3799 if let Some(ref pos) = position {
3801 self.positions.remove(&position_id);
3802
3803 if let Some(venue_positions) =
3805 self.index.venue_positions.get_mut(&pos.instrument_id.venue)
3806 {
3807 venue_positions.remove(&position_id);
3808 if venue_positions.is_empty() {
3809 self.index.venue_positions.remove(&pos.instrument_id.venue);
3810 }
3811 }
3812
3813 let instrument_positions_became_empty = self
3815 .index
3816 .instrument_positions
3817 .get_mut(&pos.instrument_id)
3818 .is_some_and(|positions| {
3819 positions.remove(&position_id);
3820 positions.is_empty()
3821 });
3822
3823 if instrument_positions_became_empty {
3824 self.index.instrument_positions.remove(&pos.instrument_id);
3825 let instrument_orders_empty = self
3826 .index
3827 .instrument_orders
3828 .get(&pos.instrument_id)
3829 .is_some_and(|orders| orders.is_empty());
3830
3831 if instrument_orders_empty {
3832 self.index.instrument_orders.remove(&pos.instrument_id);
3833 }
3834 }
3835
3836 let strategy_positions_became_empty = self
3838 .index
3839 .strategy_positions
3840 .get_mut(&pos.strategy_id)
3841 .is_some_and(|positions| {
3842 positions.remove(&position_id);
3843 positions.is_empty()
3844 });
3845
3846 if strategy_positions_became_empty {
3847 self.index.strategy_positions.remove(&pos.strategy_id);
3848 let strategy_orders_empty = self
3849 .index
3850 .strategy_orders
3851 .get(&pos.strategy_id)
3852 .is_some_and(|orders| orders.is_empty());
3853
3854 if strategy_orders_empty {
3855 self.index.strategy_orders.remove(&pos.strategy_id);
3856 self.index.strategies.remove(&pos.strategy_id);
3857 }
3858 }
3859
3860 if let Some(account_positions) = self.index.account_positions.get_mut(&pos.account_id) {
3862 account_positions.remove(&position_id);
3863 if account_positions.is_empty() {
3864 self.index.account_positions.remove(&pos.account_id);
3865 }
3866 }
3867
3868 for client_order_id in pos.client_order_ids() {
3870 self.index.order_position.remove(&client_order_id);
3871 }
3872
3873 log::info!("Purged position {position_id}");
3874 } else {
3875 log::warn!("Position {position_id} not found when purging");
3876 }
3877
3878 self.index.position_strategy.remove(&position_id);
3880 self.index.position_oms.remove(&position_id);
3881 self.index.position_orders.remove(&position_id);
3882 self.index.positions.remove(&position_id);
3883 self.index.positions_open.remove(&position_id);
3884 self.index.positions_closed.remove(&position_id);
3885
3886 self.position_snapshots.remove(&position_id);
3888 self.bump_position_snapshot_revision(position_id);
3889 }
3890
3891 fn purge_instrument_inner(&mut self, instrument_id: InstrumentId, skip_order_guard: bool) {
3915 #[cfg(feature = "defi")]
3916 let defi_found = self.defi.pools.contains_key(&instrument_id)
3917 || self.defi.pool_profilers.contains_key(&instrument_id);
3918 #[cfg(not(feature = "defi"))]
3919 let defi_found = false;
3920
3921 let found = self.instruments.contains_key(&instrument_id)
3922 || self.synthetics.contains_key(&instrument_id)
3923 || defi_found;
3924
3925 if !found {
3926 log::warn!("Instrument {instrument_id} not found when purging");
3927 return;
3928 }
3929
3930 if !skip_order_guard && let Some(orders) = self.index.instrument_orders.get(&instrument_id)
3931 {
3932 let has_non_terminal = orders
3933 .iter()
3934 .any(|client_order_id| !self.index.orders_closed.contains(client_order_id));
3935
3936 if has_non_terminal {
3937 log::warn!(
3938 "Instrument {instrument_id} has non-terminal orders when purging, skipping purge"
3939 );
3940 return;
3941 }
3942 }
3943
3944 if let Some(positions) = self.index.instrument_positions.get(&instrument_id) {
3945 let has_non_closed = positions
3946 .iter()
3947 .any(|position_id| !self.index.positions_closed.contains(position_id));
3948
3949 if has_non_closed {
3950 log::warn!(
3951 "Instrument {instrument_id} has non-closed positions when purging, skipping purge"
3952 );
3953 return;
3954 }
3955 }
3956
3957 self.instruments.remove(&instrument_id);
3958 self.synthetics.remove(&instrument_id);
3959 self.books.remove(&instrument_id);
3960 self.own_books.remove(&instrument_id);
3961 self.quotes.remove(&instrument_id);
3962 self.trades.remove(&instrument_id);
3963 self.mark_prices.remove(&instrument_id);
3964 self.index_prices.remove(&instrument_id);
3965 self.funding_rates.remove(&instrument_id);
3966 self.instrument_statuses.remove(&instrument_id);
3967 self.instrument_closes.remove(&instrument_id);
3968 self.greeks.remove(&instrument_id);
3969 self.option_greeks.remove(&instrument_id);
3970
3971 self.bars
3972 .retain(|bar_type, _| bar_type.instrument_id() != instrument_id);
3973
3974 #[cfg(feature = "defi")]
3975 {
3976 self.defi.pools.remove(&instrument_id);
3977 self.defi.pool_profilers.remove(&instrument_id);
3978 }
3979
3980 self.index.instrument_orders.remove(&instrument_id);
3981 self.index.instrument_positions.remove(&instrument_id);
3982
3983 log::info!("Purged instrument {instrument_id}");
3984 }
3985
3986 pub fn purge_instrument(&mut self, instrument_id: InstrumentId) {
3991 self.purge_instrument_inner(instrument_id, false);
3992 }
3993
3994 pub fn purge_instrument_skip_order_guard(&mut self, instrument_id: InstrumentId) {
4003 self.purge_instrument_inner(instrument_id, true);
4004 }
4005
4006 pub fn purge_account_events(&mut self, ts_now: UnixNanos, lookback_secs: u64) {
4011 log::debug!(
4012 "Purging account events{}",
4013 if lookback_secs > 0 {
4014 format!(" with lookback_secs={lookback_secs}")
4015 } else {
4016 String::new()
4017 }
4018 );
4019
4020 for account_cell in self.accounts.values() {
4021 let mut account = account_cell.borrow_mut();
4022 let event_count = account.event_count();
4023 account.purge_account_events(ts_now, lookback_secs);
4024 let count_diff = event_count - account.event_count();
4025 if count_diff > 0 {
4026 log::info!(
4027 "Purged {} event(s) from account {}",
4028 count_diff,
4029 account.id()
4030 );
4031 }
4032 }
4033 }
4034
4035 pub fn clear_index(&mut self) {
4037 self.index.clear();
4038 log::debug!("Cleared index");
4039 }
4040
4041 pub fn reset(&mut self) {
4048 log::debug!("Resetting cache");
4049
4050 self.general.clear();
4051 self.books.clear();
4052 self.own_books.clear();
4053 self.quotes.clear();
4054 self.trades.clear();
4055 self.mark_xrates.clear();
4056 self.mark_prices.clear();
4057 self.index_prices.clear();
4058 self.funding_rates.clear();
4059 self.instrument_statuses.clear();
4060 self.instrument_closes.clear();
4061 self.bars.clear();
4062 self.accounts.clear();
4063 self.orders.clear();
4064 self.order_lists.clear();
4065 self.positions.clear();
4066 self.position_snapshots.clear();
4067 self.position_snapshot_revisions.clear();
4068 self.greeks.clear();
4069 self.option_greeks.clear();
4070 self.yield_curves.clear();
4071
4072 if self.config.drop_instruments_on_reset {
4073 self.currencies.clear();
4074 self.instruments.clear();
4075 self.synthetics.clear();
4076 }
4077
4078 #[cfg(feature = "defi")]
4079 {
4080 self.defi.pools.clear();
4081 self.defi.pool_profilers.clear();
4082 }
4083
4084 self.clear_index();
4085
4086 log::info!("Reset cache");
4087 }
4088
4089 pub fn dispose(&mut self) {
4093 self.reset();
4094
4095 if let Some(database) = &mut self.database
4096 && let Err(e) = database.close()
4097 {
4098 log::error!("Failed to close database during dispose: {e}");
4099 }
4100 }
4101
4102 pub fn flush_db(&mut self) {
4106 if let Some(database) = &mut self.database
4107 && let Err(e) = database.flush()
4108 {
4109 log::error!("Failed to flush database: {e}");
4110 }
4111 }
4112
4113 pub fn add(&mut self, key: &str, value: Bytes) -> anyhow::Result<()> {
4121 check_valid_string_ascii(key, stringify!(key))?;
4122 check_predicate_false(value.is_empty(), stringify!(value))?;
4123
4124 log::debug!("Adding general {key}");
4125 self.general.insert(key.to_string(), value.clone());
4126
4127 if let Some(database) = &mut self.database {
4128 database.add(key.to_string(), value)?;
4129 }
4130 Ok(())
4131 }
4132
4133 pub fn add_order_book(&mut self, book: OrderBook) -> anyhow::Result<()> {
4139 log::debug!("Adding `OrderBook` {}", book.instrument_id);
4140
4141 if self.config.save_market_data
4142 && let Some(database) = &mut self.database
4143 {
4144 database.add_order_book(&book)?;
4145 }
4146
4147 self.books.insert(book.instrument_id, book);
4148 Ok(())
4149 }
4150
4151 pub fn add_own_order_book(&mut self, own_book: OwnOrderBook) -> anyhow::Result<()> {
4157 log::debug!("Adding `OwnOrderBook` {}", own_book.instrument_id);
4158
4159 self.own_books.insert(own_book.instrument_id, own_book);
4160 Ok(())
4161 }
4162
4163 pub fn add_mark_price(&mut self, mark_price: MarkPriceUpdate) -> anyhow::Result<()> {
4169 log::debug!("Adding `MarkPriceUpdate` for {}", mark_price.instrument_id);
4170
4171 if self.config.save_market_data {
4172 }
4174
4175 let mark_prices_deque = self
4176 .mark_prices
4177 .entry(mark_price.instrument_id)
4178 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4179 mark_prices_deque.push_front(mark_price);
4180 Ok(())
4181 }
4182
4183 pub fn add_index_price(&mut self, index_price: IndexPriceUpdate) -> anyhow::Result<()> {
4189 log::debug!(
4190 "Adding `IndexPriceUpdate` for {}",
4191 index_price.instrument_id
4192 );
4193
4194 if self.config.save_market_data {
4195 }
4197
4198 let index_prices_deque = self
4199 .index_prices
4200 .entry(index_price.instrument_id)
4201 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4202 index_prices_deque.push_front(index_price);
4203 Ok(())
4204 }
4205
4206 pub fn add_funding_rate(&mut self, funding_rate: FundingRateUpdate) -> anyhow::Result<()> {
4212 log::debug!(
4213 "Adding `FundingRateUpdate` for {}",
4214 funding_rate.instrument_id
4215 );
4216
4217 if self.config.save_market_data {
4218 }
4220
4221 let funding_rates_deque = self
4222 .funding_rates
4223 .entry(funding_rate.instrument_id)
4224 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4225 funding_rates_deque.push_front(funding_rate);
4226 Ok(())
4227 }
4228
4229 pub fn add_funding_rates(&mut self, funding_rates: &[FundingRateUpdate]) -> anyhow::Result<()> {
4235 check_slice_not_empty(funding_rates, stringify!(funding_rates))?;
4236
4237 let instrument_id = funding_rates[0].instrument_id;
4238 log::debug!(
4239 "Adding `FundingRateUpdate`[{}] {instrument_id}",
4240 funding_rates.len()
4241 );
4242
4243 if self.config.save_market_data
4244 && let Some(database) = &mut self.database
4245 {
4246 for funding_rate in funding_rates {
4247 database.add_funding_rate(funding_rate)?;
4248 }
4249 }
4250
4251 let funding_rate_deque = self
4252 .funding_rates
4253 .entry(instrument_id)
4254 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4255
4256 for funding_rate in funding_rates {
4257 funding_rate_deque.push_front(*funding_rate);
4258 }
4259 Ok(())
4260 }
4261
4262 pub fn add_instrument_status(&mut self, status: InstrumentStatus) -> anyhow::Result<()> {
4268 log::debug!("Adding `InstrumentStatus` for {}", status.instrument_id);
4269
4270 if self.config.save_market_data {
4271 }
4273
4274 let statuses_deque = self
4275 .instrument_statuses
4276 .entry(status.instrument_id)
4277 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4278 statuses_deque.push_front(status);
4279 Ok(())
4280 }
4281
4282 pub fn add_instrument_close(&mut self, close: InstrumentClose) -> anyhow::Result<()> {
4292 log::debug!("Adding `InstrumentClose` for {}", close.instrument_id);
4293
4294 if let Some(database) = &self.database {
4295 database.add_instrument_close(&close)?;
4296 }
4297
4298 self.instrument_closes.insert(close.instrument_id, close);
4299 Ok(())
4300 }
4301
4302 pub fn add_quote(&mut self, quote: QuoteTick) -> anyhow::Result<()> {
4308 log::debug!("Adding `QuoteTick` {}", quote.instrument_id);
4309
4310 if self.config.save_market_data
4311 && let Some(database) = &mut self.database
4312 {
4313 database.add_quote("e)?;
4314 }
4315
4316 let quotes_deque = self
4317 .quotes
4318 .entry(quote.instrument_id)
4319 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4320 quotes_deque.push_front(quote);
4321 Ok(())
4322 }
4323
4324 pub fn add_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
4330 check_slice_not_empty(quotes, stringify!(quotes))?;
4331
4332 let instrument_id = quotes[0].instrument_id;
4333 log::debug!("Adding `QuoteTick`[{}] {instrument_id}", quotes.len());
4334
4335 if self.config.save_market_data
4336 && let Some(database) = &mut self.database
4337 {
4338 for quote in quotes {
4339 database.add_quote(quote)?;
4340 }
4341 }
4342
4343 let quotes_deque = self
4344 .quotes
4345 .entry(instrument_id)
4346 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4347
4348 for quote in quotes {
4349 quotes_deque.push_front(*quote);
4350 }
4351 Ok(())
4352 }
4353
4354 pub fn add_trade(&mut self, trade: TradeTick) -> anyhow::Result<()> {
4360 log::debug!("Adding `TradeTick` {}", trade.instrument_id);
4361
4362 if self.config.save_market_data
4363 && let Some(database) = &mut self.database
4364 {
4365 database.add_trade(&trade)?;
4366 }
4367
4368 let trades_deque = self
4369 .trades
4370 .entry(trade.instrument_id)
4371 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4372 trades_deque.push_front(trade);
4373 Ok(())
4374 }
4375
4376 pub fn add_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
4382 check_slice_not_empty(trades, stringify!(trades))?;
4383
4384 let instrument_id = trades[0].instrument_id;
4385 log::debug!("Adding `TradeTick`[{}] {instrument_id}", trades.len());
4386
4387 if self.config.save_market_data
4388 && let Some(database) = &mut self.database
4389 {
4390 for trade in trades {
4391 database.add_trade(trade)?;
4392 }
4393 }
4394
4395 let trades_deque = self
4396 .trades
4397 .entry(instrument_id)
4398 .or_insert_with(|| BoundedVecDeque::new(self.config.tick_capacity));
4399
4400 for trade in trades {
4401 trades_deque.push_front(*trade);
4402 }
4403 Ok(())
4404 }
4405
4406 pub fn add_bar(&mut self, bar: Bar) -> anyhow::Result<()> {
4412 log::debug!("Adding `Bar` {}", bar.bar_type);
4413
4414 if self.config.save_market_data
4415 && let Some(database) = &mut self.database
4416 {
4417 database.add_bar(&bar)?;
4418 }
4419
4420 let bars = self
4421 .bars
4422 .entry(bar.bar_type)
4423 .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
4424 bars.push_front(bar);
4425 Ok(())
4426 }
4427
4428 pub fn add_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
4434 check_slice_not_empty(bars, stringify!(bars))?;
4435
4436 let bar_type = bars[0].bar_type;
4437 log::debug!("Adding `Bar`[{}] {bar_type}", bars.len());
4438
4439 if self.config.save_market_data
4440 && let Some(database) = &mut self.database
4441 {
4442 for bar in bars {
4443 database.add_bar(bar)?;
4444 }
4445 }
4446
4447 let bars_deque = self
4448 .bars
4449 .entry(bar_type)
4450 .or_insert_with(|| BoundedVecDeque::new(self.config.bar_capacity));
4451
4452 for bar in bars {
4453 bars_deque.push_front(*bar);
4454 }
4455 Ok(())
4456 }
4457
4458 pub fn add_greeks(&mut self, greeks: GreeksData) -> anyhow::Result<()> {
4464 log::debug!("Adding `GreeksData` {}", greeks.instrument_id);
4465
4466 if self.config.save_market_data
4467 && let Some(_database) = &mut self.database
4468 {
4469 }
4471
4472 self.greeks.insert(greeks.instrument_id, greeks);
4473 Ok(())
4474 }
4475
4476 pub fn greeks(&self, instrument_id: &InstrumentId) -> Option<GreeksData> {
4478 self.greeks.get(instrument_id).cloned()
4479 }
4480
4481 pub fn add_option_greeks(&mut self, greeks: OptionGreeks) {
4483 log::debug!("Adding `OptionGreeks` {}", greeks.instrument_id);
4484 self.option_greeks.insert(greeks.instrument_id, greeks);
4485 }
4486
4487 #[must_use]
4489 pub fn option_greeks(&self, instrument_id: &InstrumentId) -> Option<&OptionGreeks> {
4490 self.option_greeks.get(instrument_id)
4491 }
4492
4493 pub fn add_yield_curve(&mut self, yield_curve: YieldCurveData) -> anyhow::Result<()> {
4499 log::debug!("Adding `YieldCurveData` {}", yield_curve.curve_name);
4500
4501 if self.config.save_market_data
4502 && let Some(_database) = &mut self.database
4503 {
4504 }
4506
4507 self.yield_curves
4508 .insert(yield_curve.curve_name.clone(), yield_curve);
4509 Ok(())
4510 }
4511
4512 pub fn yield_curve(&self, key: &str) -> Option<Box<dyn Fn(f64) -> f64>> {
4514 self.yield_curves.get(key).map(|curve| {
4515 let curve_clone = curve.clone();
4516 Box::new(move |expiry_in_years: f64| curve_clone.get_rate(expiry_in_years))
4517 as Box<dyn Fn(f64) -> f64>
4518 })
4519 }
4520
4521 pub fn add_currency(&mut self, currency: Currency) -> anyhow::Result<()> {
4527 if self.currencies.contains_key(¤cy.code) {
4528 return Ok(());
4529 }
4530 log::debug!("Adding `Currency` {}", currency.code);
4531
4532 if let Some(database) = &mut self.database {
4533 database.add_currency(¤cy)?;
4534 }
4535
4536 self.currencies.insert(currency.code, currency);
4537 Ok(())
4538 }
4539
4540 pub fn add_instrument(&mut self, instrument: InstrumentAny) -> anyhow::Result<()> {
4546 log::debug!("Adding `Instrument` {}", instrument.id());
4547
4548 if let Some(base_currency) = instrument.base_currency() {
4550 self.add_currency(base_currency)?;
4551 }
4552 self.add_currency(instrument.quote_currency())?;
4553 self.add_currency(instrument.settlement_currency())?;
4554
4555 if let Some(database) = &mut self.database {
4556 database.add_instrument(&instrument)?;
4557 }
4558
4559 self.instruments.insert(instrument.id(), instrument);
4560 Ok(())
4561 }
4562
4563 pub fn add_synthetic(&mut self, synthetic: SyntheticInstrument) -> anyhow::Result<()> {
4569 log::debug!("Adding `SyntheticInstrument` {}", synthetic.id);
4570
4571 if let Some(database) = &mut self.database {
4572 database.add_synthetic(&synthetic)?;
4573 }
4574
4575 self.synthetics.insert(synthetic.id, synthetic);
4576 Ok(())
4577 }
4578
4579 pub fn add_account(&mut self, account: AccountAny) -> anyhow::Result<()> {
4585 log::debug!("Adding `Account` {}", account.id());
4586
4587 if let Some(database) = &mut self.database {
4588 database.add_account(&account)?;
4589 }
4590
4591 let account_id = account.id();
4592 self.accounts.insert(account_id, SharedCell::new(account));
4593 self.index
4594 .venue_account
4595 .insert(account_id.get_issuer(), account_id);
4596 Ok(())
4597 }
4598
4599 pub fn add_venue_order_id(
4608 &mut self,
4609 client_order_id: &ClientOrderId,
4610 venue_order_id: &VenueOrderId,
4611 overwrite: bool,
4612 ) -> anyhow::Result<()> {
4613 self.validate_venue_order_id_claim(client_order_id, venue_order_id, overwrite)?;
4614
4615 self.index
4616 .client_order_ids
4617 .insert(*client_order_id, *venue_order_id);
4618 self.index
4619 .venue_order_ids
4620 .insert(*venue_order_id, *client_order_id);
4621
4622 Ok(())
4623 }
4624
4625 pub fn index_venue_order_id(
4636 &mut self,
4637 client_order_id: &ClientOrderId,
4638 venue_order_id: &VenueOrderId,
4639 ) -> anyhow::Result<()> {
4640 self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
4641
4642 self.index
4643 .venue_order_ids
4644 .insert(*venue_order_id, *client_order_id);
4645 self.index
4646 .client_order_ids
4647 .entry(*client_order_id)
4648 .or_insert(*venue_order_id);
4649
4650 Ok(())
4651 }
4652
4653 fn validate_venue_order_id_claim(
4654 &self,
4655 client_order_id: &ClientOrderId,
4656 venue_order_id: &VenueOrderId,
4657 overwrite: bool,
4658 ) -> anyhow::Result<()> {
4659 self.validate_venue_order_id_ownership(client_order_id, venue_order_id)?;
4660
4661 if let Some(existing_venue_order_id) = self.index.client_order_ids.get(client_order_id)
4662 && !overwrite
4663 && existing_venue_order_id != venue_order_id
4664 {
4665 anyhow::bail!(
4666 "Existing {existing_venue_order_id} for {client_order_id}
4667 did not match the given {venue_order_id}.
4668 If you are writing a test then try a different `venue_order_id`,
4669 otherwise this is probably a bug."
4670 );
4671 }
4672
4673 Ok(())
4674 }
4675
4676 fn validate_venue_order_id_ownership(
4677 &self,
4678 client_order_id: &ClientOrderId,
4679 venue_order_id: &VenueOrderId,
4680 ) -> anyhow::Result<()> {
4681 if let Some(existing_client_order_id) = self.index.venue_order_ids.get(venue_order_id)
4682 && existing_client_order_id != client_order_id
4683 {
4684 return Err(VenueOrderIdOwnershipError {
4685 venue_order_id: *venue_order_id,
4686 existing_client_order_id: *existing_client_order_id,
4687 claimant_client_order_id: *client_order_id,
4688 }
4689 .into());
4690 }
4691
4692 Ok(())
4693 }
4694
4695 pub fn add_order(
4710 &mut self,
4711 order: OrderAny,
4712 position_id: Option<PositionId>,
4713 client_id: Option<ClientId>,
4714 replace_existing: bool,
4715 ) -> anyhow::Result<()> {
4716 let instrument_id = order.instrument_id();
4717 let venue = instrument_id.venue;
4718 let client_order_id = order.client_order_id();
4719 let strategy_id = order.strategy_id();
4720 let exec_algorithm_id = order.exec_algorithm_id();
4721 let exec_spawn_id = order.exec_spawn_id();
4722
4723 if !replace_existing {
4724 check_key_not_in_map(
4725 &client_order_id,
4726 &self.orders,
4727 stringify!(client_order_id),
4728 stringify!(orders),
4729 )?;
4730 }
4731
4732 log::debug!("Adding {order:?}");
4733
4734 self.index.orders.insert(client_order_id);
4735
4736 if order.is_active_local() {
4737 self.index.orders_active_local.insert(client_order_id);
4738 }
4739 self.index
4740 .order_strategy
4741 .insert(client_order_id, strategy_id);
4742 self.index.strategies.insert(strategy_id);
4743
4744 self.index
4746 .venue_orders
4747 .entry(venue)
4748 .or_default()
4749 .insert(client_order_id);
4750
4751 self.index
4753 .instrument_orders
4754 .entry(instrument_id)
4755 .or_default()
4756 .insert(client_order_id);
4757
4758 self.index
4760 .strategy_orders
4761 .entry(strategy_id)
4762 .or_default()
4763 .insert(client_order_id);
4764
4765 if let Some(account_id) = order.account_id() {
4767 self.index
4768 .account_orders
4769 .entry(account_id)
4770 .or_default()
4771 .insert(client_order_id);
4772 }
4773
4774 if let Some(exec_algorithm_id) = exec_algorithm_id {
4776 self.index.exec_algorithms.insert(exec_algorithm_id);
4777
4778 self.index
4779 .exec_algorithm_orders
4780 .entry(exec_algorithm_id)
4781 .or_default()
4782 .insert(client_order_id);
4783 }
4784
4785 if let Some(exec_spawn_id) = exec_spawn_id {
4787 self.index
4788 .exec_spawn_orders
4789 .entry(exec_spawn_id)
4790 .or_default()
4791 .insert(client_order_id);
4792 }
4793
4794 if order.emulation_trigger().is_some() {
4796 self.index.orders_emulated.insert(client_order_id);
4797 }
4798
4799 if let Some(position_id) = position_id {
4801 self.index_position_id_in_memory(&position_id, &venue, &client_order_id, &strategy_id);
4802 }
4803
4804 if let Some(client_id) = client_id {
4806 self.index.order_client.insert(client_order_id, client_id);
4807 log::debug!("Indexed {client_id:?}");
4808 }
4809
4810 let order_cell = if let Some(order_cell) = self.orders.get(&client_order_id) {
4813 *order_cell.borrow_mut() = order;
4814 order_cell.clone()
4815 } else {
4816 let order_cell = SharedCell::new(order);
4817 self.orders.insert(client_order_id, order_cell.clone());
4818 order_cell
4819 };
4820
4821 if let Some(position_id) = position_id {
4822 self.persist_position_id(&position_id, &client_order_id)?;
4823 }
4824
4825 if let Some(database) = &mut self.database {
4826 database.add_order(&order_cell.borrow(), client_id)?;
4827 }
4832
4833 Ok(())
4834 }
4835
4836 pub fn claim_order_clients(
4848 &mut self,
4849 claims: &[(ClientOrderId, ClientId)],
4850 ) -> anyhow::Result<()> {
4851 let mut requested = AHashMap::with_capacity(claims.len());
4852 let mut ordered_claims = Vec::with_capacity(claims.len());
4853
4854 for (client_order_id, client_id) in claims {
4855 if let Some(existing_client_id) = requested.get(client_order_id) {
4856 if existing_client_id != client_id {
4857 anyhow::bail!(
4858 "Conflicting execution client claims for {client_order_id}: \
4859 {existing_client_id} and {client_id}"
4860 );
4861 }
4862 continue;
4863 }
4864
4865 requested.insert(*client_order_id, *client_id);
4866 ordered_claims.push((*client_order_id, *client_id));
4867 }
4868
4869 let mut pending_claims = Vec::with_capacity(ordered_claims.len());
4870 for (client_order_id, client_id) in ordered_claims {
4871 if !self.orders.contains_key(&client_order_id) {
4872 return Err(OrderLookupError::not_found(client_order_id).into());
4873 }
4874
4875 match self.index.order_client.get(&client_order_id) {
4876 Some(existing_client_id) if *existing_client_id == client_id => {}
4877 Some(existing_client_id) => {
4878 anyhow::bail!(
4879 "Order {client_order_id} is already claimed by execution client \
4880 {existing_client_id} and cannot be claimed by {client_id}"
4881 );
4882 }
4883 None => pending_claims.push((client_order_id, client_id)),
4884 }
4885 }
4886
4887 if pending_claims.is_empty() {
4888 return Ok(());
4889 }
4890
4891 if let Some(database) = &self.database {
4892 database.index_order_clients(&pending_claims)?;
4893 }
4894
4895 for (client_order_id, client_id) in pending_claims {
4896 self.index.order_client.insert(client_order_id, client_id);
4897 log::debug!("Claimed {client_order_id} for execution client {client_id}");
4898 }
4899
4900 Ok(())
4901 }
4902
4903 pub fn add_order_list(&mut self, order_list: OrderList) -> anyhow::Result<()> {
4909 let order_list_id = order_list.id;
4910 check_key_not_in_map(
4911 &order_list_id,
4912 &self.order_lists,
4913 stringify!(order_list_id),
4914 stringify!(order_lists),
4915 )?;
4916
4917 log::debug!("Adding {order_list}");
4918 self.order_lists.insert(order_list_id, order_list);
4919 Ok(())
4920 }
4921
4922 pub fn add_position_id(
4933 &mut self,
4934 position_id: &PositionId,
4935 venue: &Venue,
4936 client_order_id: &ClientOrderId,
4937 strategy_id: &StrategyId,
4938 ) -> anyhow::Result<()> {
4939 self.index_position_id_in_memory(position_id, venue, client_order_id, strategy_id);
4940 self.persist_position_id(position_id, client_order_id)
4941 }
4942
4943 fn index_position_id_in_memory(
4944 &mut self,
4945 position_id: &PositionId,
4946 venue: &Venue,
4947 client_order_id: &ClientOrderId,
4948 strategy_id: &StrategyId,
4949 ) {
4950 self.index
4951 .order_position
4952 .insert(*client_order_id, *position_id);
4953 self.index_position(position_id, venue, strategy_id);
4954 self.index
4955 .position_orders
4956 .entry(*position_id)
4957 .or_default()
4958 .insert(*client_order_id);
4959 }
4960
4961 fn persist_position_id(
4962 &mut self,
4963 position_id: &PositionId,
4964 client_order_id: &ClientOrderId,
4965 ) -> anyhow::Result<()> {
4966 if let Some(database) = &mut self.database {
4967 database.index_order_position(*client_order_id, *position_id)?;
4968 }
4969
4970 Ok(())
4971 }
4972
4973 fn index_position(
4974 &mut self,
4975 position_id: &PositionId,
4976 venue: &Venue,
4977 strategy_id: &StrategyId,
4978 ) {
4979 let strategy_id = self
4980 .positions
4981 .get(position_id)
4982 .map(|position| position.borrow().strategy_id)
4983 .filter(StrategyId::is_external)
4984 .unwrap_or(*strategy_id);
4985
4986 self.index
4988 .position_strategy
4989 .insert(*position_id, strategy_id);
4990
4991 self.index.position_orders.entry(*position_id).or_default();
4993
4994 self.index
4996 .strategy_positions
4997 .entry(strategy_id)
4998 .or_default()
4999 .insert(*position_id);
5000
5001 self.index
5003 .venue_positions
5004 .entry(*venue)
5005 .or_default()
5006 .insert(*position_id);
5007 }
5008
5009 fn assign_position_ids_to_contingencies(&mut self) {
5017 let mut assignments: Vec<(PositionId, ClientOrderId)> = Vec::new();
5018
5019 for parent_order_cell in self.orders.values() {
5020 let parent = parent_order_cell.borrow();
5021 if parent.contingency_type() != Some(ContingencyType::Oto) {
5022 continue;
5023 }
5024 let Some(parent_position_id) = parent.position_id() else {
5025 continue;
5026 };
5027 let Some(linked_order_ids) = parent.linked_order_ids() else {
5028 continue;
5029 };
5030
5031 for client_order_id in linked_order_ids {
5032 match self.orders.get(client_order_id) {
5033 None => {
5034 log::error!("Contingency order {client_order_id} not found");
5035 }
5036 Some(contingent_order_cell) => {
5037 if contingent_order_cell.borrow().position_id().is_none() {
5038 assignments.push((parent_position_id, *client_order_id));
5039 }
5040 }
5041 }
5042 }
5043 }
5044
5045 for (position_id, client_order_id) in assignments {
5046 let Some((venue, strategy_id)) = self.orders.get(&client_order_id).map(|order_cell| {
5047 let mut contingent = order_cell.borrow_mut();
5048 contingent.set_position_id(Some(position_id));
5049 (contingent.instrument_id().venue, contingent.strategy_id())
5050 }) else {
5051 continue;
5052 };
5053
5054 if let Err(e) =
5057 self.add_position_id(&position_id, &venue, &client_order_id, &strategy_id)
5058 {
5059 log::error!("Failed to re-index {client_order_id} -> {position_id}: {e}");
5060 }
5061 }
5062 }
5063
5064 pub fn add_position(&mut self, position: &Position, oms_type: OmsType) -> anyhow::Result<()> {
5072 self.add_position_inner(position, oms_type, true)
5073 }
5074
5075 pub fn add_position_without_order(
5083 &mut self,
5084 position: &Position,
5085 oms_type: OmsType,
5086 ) -> anyhow::Result<()> {
5087 self.add_position_inner(position, oms_type, false)
5088 }
5089
5090 fn add_position_inner(
5091 &mut self,
5092 position: &Position,
5093 oms_type: OmsType,
5094 index_order: bool,
5095 ) -> anyhow::Result<()> {
5096 let key = position_oms_key(position.id);
5099 check_valid_string_ascii(&key, stringify!(key))?;
5100 let value = Bytes::from(serde_json::to_vec(&oms_type)?);
5101 check_predicate_false(value.is_empty(), stringify!(value))?;
5102
5103 self.positions
5104 .insert(position.id, SharedCell::new(position.clone()));
5105 self.index.position_oms.insert(position.id, oms_type);
5106 self.index.positions.insert(position.id);
5107 self.index.positions_open.insert(position.id);
5108 self.index.positions_closed.remove(&position.id); self.index.strategies.insert(position.strategy_id);
5110 self.index
5111 .strategy_orders
5112 .entry(position.strategy_id)
5113 .or_default();
5114
5115 log::debug!("Adding {position}");
5116
5117 if index_order {
5118 self.index_position_id_in_memory(
5119 &position.id,
5120 &position.instrument_id.venue,
5121 &position.opening_order_id,
5122 &position.strategy_id,
5123 );
5124 } else {
5125 self.index_position(
5126 &position.id,
5127 &position.instrument_id.venue,
5128 &position.strategy_id,
5129 );
5130 }
5131
5132 let instrument_id = position.instrument_id;
5134 let instrument_positions = self
5135 .index
5136 .instrument_positions
5137 .entry(instrument_id)
5138 .or_default();
5139 instrument_positions.insert(position.id);
5140 self.index
5141 .instrument_orders
5142 .entry(instrument_id)
5143 .or_default();
5144
5145 self.index
5147 .account_positions
5148 .entry(position.account_id)
5149 .or_default()
5150 .insert(position.id);
5151
5152 log::debug!("Adding general {key}");
5153 self.general.insert(key.clone(), value.clone());
5154
5155 if index_order {
5156 self.persist_position_id(&position.id, &position.opening_order_id)?;
5157 }
5158
5159 if let Some(database) = &mut self.database {
5160 database.add_position(position)?;
5161 database.add(key, value)?;
5170 }
5171
5172 Ok(())
5173 }
5174
5175 pub fn update_account(&mut self, account: &AccountAny) -> anyhow::Result<()> {
5184 let account_id = account.id();
5185 match self.accounts.get(&account_id) {
5186 Some(account_cell) => *account_cell.borrow_mut() = account.clone(),
5187 None => {
5188 self.accounts
5189 .insert(account_id, SharedCell::new(account.clone()));
5190 }
5191 }
5192
5193 if let Some(database) = &mut self.database {
5194 database.update_account(account)?;
5195 }
5196 Ok(())
5197 }
5198
5199 #[must_use]
5210 pub fn take_account(&mut self, account_id: &AccountId) -> Option<AccountAny> {
5211 let cell = self.accounts.remove(account_id)?;
5212 let rc: Rc<RefCell<AccountAny>> = cell.into();
5213
5214 match Rc::try_unwrap(rc) {
5215 Ok(cell) => Some(cell.into_inner()),
5216 Err(rc) => {
5217 log::error!(
5218 "Cannot move account {account_id} out of cache: account cell has an outstanding owner"
5219 );
5220 self.accounts.insert(*account_id, rc.into());
5221 None
5222 }
5223 }
5224 }
5225
5226 pub fn cache_account_owned(&mut self, account: AccountAny) {
5228 let account_id = account.id();
5229 self.index
5230 .venue_account
5231 .insert(account_id.get_issuer(), account_id);
5232 match self.accounts.get(&account_id) {
5233 Some(account_cell) => *account_cell.borrow_mut() = account,
5234 None => {
5235 self.accounts.insert(account_id, SharedCell::new(account));
5236 }
5237 }
5238 }
5239
5240 pub fn update_account_owned(&mut self, account: AccountAny) -> anyhow::Result<()> {
5246 let account_id = account.id();
5247 self.cache_account_owned(account);
5248
5249 if let Some(database) = &mut self.database {
5250 let Some(account_cell) = self.accounts.get(&account_id) else {
5251 anyhow::bail!("Account {account_id} not found after cache update");
5252 };
5253 database.update_account(&account_cell.borrow())?;
5254 }
5255 Ok(())
5256 }
5257
5258 pub fn update_account_state(&mut self, event: &AccountState) -> anyhow::Result<()> {
5268 let Some(cell) = self.accounts.get(&event.account_id) else {
5269 return self.add_account(AccountAny::from_events(std::slice::from_ref(event))?);
5270 };
5271
5272 cell.borrow_mut().apply(event.clone())?;
5273
5274 if let Some(database) = &mut self.database {
5275 database.update_account(&cell.borrow())?;
5276 }
5277 Ok(())
5278 }
5279
5280 pub fn replace_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
5291 let client_order_id = order.client_order_id();
5292 if let Some(venue_order_id) = order.venue_order_id() {
5293 self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
5294 }
5295
5296 match self.orders.get(&client_order_id) {
5297 Some(order_cell) => *order_cell.borrow_mut() = order.clone(),
5300 None => {
5301 self.orders
5302 .insert(client_order_id, SharedCell::new(order.clone()));
5303 }
5304 }
5305
5306 self.refresh_order(order)
5307 }
5308
5309 pub fn update_order(&mut self, event: &OrderEventAny) -> anyhow::Result<OrderAny> {
5315 let event_client_order_id = event.client_order_id();
5316 let client_order_id = if self.order_exists(&event_client_order_id) {
5317 event_client_order_id
5318 } else if let Some(venue_order_id) = event.venue_order_id() {
5319 self.index
5320 .venue_order_ids
5321 .get(&venue_order_id)
5322 .copied()
5323 .ok_or(OrderError::NotFound(event_client_order_id))?
5324 } else {
5325 return Err(OrderError::NotFound(event_client_order_id).into());
5326 };
5327
5328 let order_cell = self
5329 .orders
5330 .get(&client_order_id)
5331 .cloned()
5332 .ok_or(OrderError::NotFound(client_order_id))?;
5333
5334 let mut snapshot = order_cell.borrow().clone();
5338 snapshot.apply(event.clone())?;
5339
5340 if let Some(venue_order_id) = snapshot.venue_order_id() {
5344 self.validate_venue_order_id_ownership(&client_order_id, &venue_order_id)?;
5345 }
5346
5347 *order_cell.borrow_mut() = snapshot.clone();
5348
5349 if let Err(e) = self.refresh_order(&snapshot) {
5350 log::error!("Error updating order in cache: {e}");
5351 }
5352
5353 Ok(snapshot)
5354 }
5355
5356 fn refresh_order(&mut self, order: &OrderAny) -> anyhow::Result<()> {
5357 let client_order_id = order.client_order_id();
5358
5359 if let Some(venue_order_id) = order.venue_order_id() {
5362 let overwrite = matches!(order.last_event(), OrderEventAny::Updated(_));
5363 if let Err(e) = self.add_venue_order_id(&client_order_id, &venue_order_id, overwrite) {
5364 if e.is::<VenueOrderIdOwnershipError>() {
5365 return Err(e);
5366 }
5367 log::error!("Error indexing venue order ID in cache: {e}");
5368 }
5369 }
5370
5371 if order.is_active_local() {
5372 self.index.orders_active_local.insert(client_order_id);
5373 } else {
5374 self.index.orders_active_local.remove(&client_order_id);
5375 }
5376
5377 if order.is_inflight() {
5379 self.index.orders_inflight.insert(client_order_id);
5380 } else {
5381 self.index.orders_inflight.remove(&client_order_id);
5382 }
5383
5384 if order.is_open() {
5386 self.index.orders_closed.remove(&client_order_id);
5387 self.index.orders_open.insert(client_order_id);
5388 } else if order.is_closed() {
5389 self.index.orders_open.remove(&client_order_id);
5390 self.index.orders_pending_cancel.remove(&client_order_id);
5391 self.index.orders_closed.insert(client_order_id);
5392 }
5393
5394 if matches!(order.last_event(), OrderEventAny::CancelRejected(_)) {
5396 self.index.orders_pending_cancel.remove(&client_order_id);
5397 }
5398
5399 if order.emulation_trigger().is_some() && !order.is_closed() {
5401 self.index.orders_emulated.insert(client_order_id);
5402 } else {
5403 self.index.orders_emulated.remove(&client_order_id);
5404 }
5405
5406 if let Some(account_id) = order.account_id() {
5408 self.index
5409 .account_orders
5410 .entry(account_id)
5411 .or_default()
5412 .insert(client_order_id);
5413 }
5414
5415 if !self.own_books.is_empty() {
5417 let own_book = self.own_order_book(&order.instrument_id());
5418 if (own_book.is_some() && order.is_closed()) || should_handle_own_book_order(order) {
5419 self.update_own_order_book(order);
5420 }
5421 }
5422
5423 if let Some(database) = &mut self.database {
5424 database.update_order(order.last_event())?;
5425 }
5430
5431 Ok(())
5432 }
5433
5434 pub fn update_order_pending_cancel_local(&mut self, order: &OrderAny) {
5436 self.index
5437 .orders_pending_cancel
5438 .insert(order.client_order_id());
5439 }
5440
5441 pub fn update_position(&mut self, position: &Position) -> anyhow::Result<()> {
5451 let Some(position_cell) = self.positions.get(&position.id).cloned() else {
5452 anyhow::bail!("Cannot update position {}: not found in cache", position.id);
5453 };
5454
5455 self.refresh_position_indexes(position);
5456
5457 *position_cell.borrow_mut() = position.clone();
5458
5459 if let Some(database) = &mut self.database {
5460 database.update_position(position)?;
5461 }
5466
5467 Ok(())
5468 }
5469
5470 pub fn update_position_from_fill(
5480 &mut self,
5481 position_id: PositionId,
5482 fill: &OrderFilled,
5483 ) -> anyhow::Result<Position> {
5484 let Some(position_cell) = self.positions.get(&position_id).cloned() else {
5485 anyhow::bail!("Cannot update position {position_id}: not found in cache");
5486 };
5487
5488 let position = {
5489 let mut position = position_cell.borrow_mut();
5490 position.apply(fill);
5491 position.clone_without_events()
5492 };
5493
5494 self.refresh_position_indexes(&position);
5495
5496 if let Some(database) = &mut self.database {
5497 database.update_position(&position_cell.borrow())?;
5498 }
5499
5500 Ok(position)
5501 }
5502
5503 fn refresh_position_indexes(&mut self, position: &Position) {
5504 if position.is_open() {
5505 self.index.positions_open.insert(position.id);
5506 self.index.positions_closed.remove(&position.id);
5507 } else {
5508 self.index.positions_closed.insert(position.id);
5509 self.index.positions_open.remove(&position.id);
5510 }
5511 }
5512
5513 #[must_use]
5515 pub fn oms_type(&self, position_id: &PositionId) -> Option<OmsType> {
5516 self.index.position_oms.get(position_id).copied()
5517 }
5518
5519 pub fn snapshot_order_state(&self, order: &OrderAny) -> anyhow::Result<()> {
5525 let Some(database) = &self.database else {
5526 log::warn!(
5527 "Cannot snapshot order state for {} (no database configured)",
5528 order.client_order_id()
5529 );
5530 return Ok(());
5531 };
5532
5533 database.snapshot_order_state(order)
5534 }
5535
5536 fn collect_order_filter_sources<'a>(
5547 &'a self,
5548 venue: Option<&Venue>,
5549 instrument_id: Option<&InstrumentId>,
5550 strategy_id: Option<&StrategyId>,
5551 account_id: Option<&AccountId>,
5552 ) -> FilterSources<'a, ClientOrderId> {
5553 let mut sources: Vec<&AHashSet<ClientOrderId>> = Vec::with_capacity(4);
5554
5555 if let Some(venue) = venue {
5556 match self.index.venue_orders.get(venue) {
5557 Some(set) => sources.push(set),
5558 None => return FilterSources::Empty,
5559 }
5560 }
5561
5562 if let Some(instrument_id) = instrument_id {
5563 match self.index.instrument_orders.get(instrument_id) {
5564 Some(set) => sources.push(set),
5565 None => return FilterSources::Empty,
5566 }
5567 }
5568
5569 if let Some(strategy_id) = strategy_id {
5570 match self.index.strategy_orders.get(strategy_id) {
5571 Some(set) => sources.push(set),
5572 None => return FilterSources::Empty,
5573 }
5574 }
5575
5576 if let Some(account_id) = account_id {
5577 match self.index.account_orders.get(account_id) {
5578 Some(set) => sources.push(set),
5579 None => return FilterSources::Empty,
5580 }
5581 }
5582
5583 if sources.is_empty() {
5584 FilterSources::Unfiltered
5585 } else {
5586 FilterSources::Sets(sources)
5587 }
5588 }
5589
5590 fn collect_position_filter_sources<'a>(
5591 &'a self,
5592 venue: Option<&Venue>,
5593 instrument_id: Option<&InstrumentId>,
5594 strategy_id: Option<&StrategyId>,
5595 account_id: Option<&AccountId>,
5596 ) -> FilterSources<'a, PositionId> {
5597 let mut sources: Vec<&AHashSet<PositionId>> = Vec::with_capacity(4);
5598
5599 if let Some(venue) = venue {
5600 match self.index.venue_positions.get(venue) {
5601 Some(set) => sources.push(set),
5602 None => return FilterSources::Empty,
5603 }
5604 }
5605
5606 if let Some(instrument_id) = instrument_id {
5607 match self.index.instrument_positions.get(instrument_id) {
5608 Some(set) => sources.push(set),
5609 None => return FilterSources::Empty,
5610 }
5611 }
5612
5613 if let Some(strategy_id) = strategy_id {
5614 match self.index.strategy_positions.get(strategy_id) {
5615 Some(set) => sources.push(set),
5616 None => return FilterSources::Empty,
5617 }
5618 }
5619
5620 if let Some(account_id) = account_id {
5621 match self.index.account_positions.get(account_id) {
5622 Some(set) => sources.push(set),
5623 None => return FilterSources::Empty,
5624 }
5625 }
5626
5627 if sources.is_empty() {
5628 FilterSources::Unfiltered
5629 } else {
5630 FilterSources::Sets(sources)
5631 }
5632 }
5633
5634 fn query_orders_in_bucket(
5640 &self,
5641 bucket: &AHashSet<ClientOrderId>,
5642 venue: Option<&Venue>,
5643 instrument_id: Option<&InstrumentId>,
5644 strategy_id: Option<&StrategyId>,
5645 account_id: Option<&AccountId>,
5646 ) -> AHashSet<ClientOrderId> {
5647 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5648 FilterSources::Empty => AHashSet::new(),
5649 FilterSources::Unfiltered => bucket.clone(),
5650 FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
5651 }
5652 }
5653
5654 fn query_positions_in_bucket(
5655 &self,
5656 bucket: &AHashSet<PositionId>,
5657 venue: Option<&Venue>,
5658 instrument_id: Option<&InstrumentId>,
5659 strategy_id: Option<&StrategyId>,
5660 account_id: Option<&AccountId>,
5661 ) -> AHashSet<PositionId> {
5662 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5663 FilterSources::Empty => AHashSet::new(),
5664 FilterSources::Unfiltered => bucket.clone(),
5665 FilterSources::Sets(sources) => intersect_pair_or_many(bucket, sources),
5666 }
5667 }
5668
5669 fn view_orders_in_bucket<'a>(
5672 &'a self,
5673 bucket: &'a AHashSet<ClientOrderId>,
5674 venue: Option<&Venue>,
5675 instrument_id: Option<&InstrumentId>,
5676 strategy_id: Option<&StrategyId>,
5677 account_id: Option<&AccountId>,
5678 ) -> Cow<'a, AHashSet<ClientOrderId>> {
5679 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5680 FilterSources::Empty => Cow::Owned(AHashSet::new()),
5681 FilterSources::Unfiltered => Cow::Borrowed(bucket),
5682 FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
5683 }
5684 }
5685
5686 fn view_positions_in_bucket<'a>(
5687 &'a self,
5688 bucket: &'a AHashSet<PositionId>,
5689 venue: Option<&Venue>,
5690 instrument_id: Option<&InstrumentId>,
5691 strategy_id: Option<&StrategyId>,
5692 account_id: Option<&AccountId>,
5693 ) -> Cow<'a, AHashSet<PositionId>> {
5694 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5695 FilterSources::Empty => Cow::Owned(AHashSet::new()),
5696 FilterSources::Unfiltered => Cow::Borrowed(bucket),
5697 FilterSources::Sets(sources) => Cow::Owned(intersect_pair_or_many(bucket, sources)),
5698 }
5699 }
5700
5701 fn iter_orders_in_bucket<'a>(
5706 &'a self,
5707 bucket: &'a AHashSet<ClientOrderId>,
5708 venue: Option<&Venue>,
5709 instrument_id: Option<&InstrumentId>,
5710 strategy_id: Option<&StrategyId>,
5711 account_id: Option<&AccountId>,
5712 ) -> Box<dyn Iterator<Item = ClientOrderId> + 'a> {
5713 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5714 FilterSources::Empty => Box::new(std::iter::empty()),
5715 FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
5716 FilterSources::Sets(mut sources) => {
5717 sources.push(bucket);
5718 sources.sort_unstable_by_key(|s| s.len());
5719 let driver = sources[0];
5720 let rest: Vec<&'a AHashSet<ClientOrderId>> = sources[1..].to_vec();
5721 Box::new(
5722 driver
5723 .iter()
5724 .copied()
5725 .filter(move |id| rest.iter().all(|s| s.contains(id))),
5726 )
5727 }
5728 }
5729 }
5730
5731 fn iter_positions_in_bucket<'a>(
5732 &'a self,
5733 bucket: &'a AHashSet<PositionId>,
5734 venue: Option<&Venue>,
5735 instrument_id: Option<&InstrumentId>,
5736 strategy_id: Option<&StrategyId>,
5737 account_id: Option<&AccountId>,
5738 ) -> Box<dyn Iterator<Item = PositionId> + 'a> {
5739 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5740 FilterSources::Empty => Box::new(std::iter::empty()),
5741 FilterSources::Unfiltered => Box::new(bucket.iter().copied()),
5742 FilterSources::Sets(mut sources) => {
5743 sources.push(bucket);
5744 sources.sort_unstable_by_key(|s| s.len());
5745 let driver = sources[0];
5746 let rest: Vec<&'a AHashSet<PositionId>> = sources[1..].to_vec();
5747 Box::new(
5748 driver
5749 .iter()
5750 .copied()
5751 .filter(move |id| rest.iter().all(|s| s.contains(id))),
5752 )
5753 }
5754 }
5755 }
5756
5757 fn count_orders_in_bucket(
5763 &self,
5764 bucket: &AHashSet<ClientOrderId>,
5765 venue: Option<&Venue>,
5766 instrument_id: Option<&InstrumentId>,
5767 strategy_id: Option<&StrategyId>,
5768 account_id: Option<&AccountId>,
5769 side: Option<OrderSide>,
5770 ) -> usize {
5771 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5772 FilterSources::Empty => 0,
5773 FilterSources::Unfiltered => side.map_or_else(
5774 || bucket.len(),
5775 |side| {
5776 bucket
5777 .iter()
5778 .filter(|id| self.order_side_matches(id, side))
5779 .count()
5780 },
5781 ),
5782 FilterSources::Sets(mut sources) => {
5783 sources.push(bucket);
5784 sources.sort_unstable_by_key(|s| s.len());
5785 let driver = sources[0];
5786 let rest = &sources[1..];
5787
5788 driver
5789 .iter()
5790 .filter(|id| rest.iter().all(|s| s.contains(id)))
5791 .filter(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
5792 .count()
5793 }
5794 }
5795 }
5796
5797 fn count_positions_in_bucket(
5798 &self,
5799 bucket: &AHashSet<PositionId>,
5800 venue: Option<&Venue>,
5801 instrument_id: Option<&InstrumentId>,
5802 strategy_id: Option<&StrategyId>,
5803 account_id: Option<&AccountId>,
5804 side: Option<PositionSide>,
5805 ) -> usize {
5806 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5807 FilterSources::Empty => 0,
5808 FilterSources::Unfiltered => side.map_or_else(
5809 || bucket.len(),
5810 |side| {
5811 bucket
5812 .iter()
5813 .filter(|id| self.position_side_matches(id, side))
5814 .count()
5815 },
5816 ),
5817 FilterSources::Sets(mut sources) => {
5818 sources.push(bucket);
5819 sources.sort_unstable_by_key(|s| s.len());
5820 let driver = sources[0];
5821 let rest = &sources[1..];
5822
5823 driver
5824 .iter()
5825 .filter(|id| rest.iter().all(|s| s.contains(id)))
5826 .filter(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
5827 .count()
5828 }
5829 }
5830 }
5831
5832 fn any_orders_in_bucket(
5838 &self,
5839 bucket: &AHashSet<ClientOrderId>,
5840 venue: Option<&Venue>,
5841 instrument_id: Option<&InstrumentId>,
5842 strategy_id: Option<&StrategyId>,
5843 account_id: Option<&AccountId>,
5844 side: Option<OrderSide>,
5845 ) -> bool {
5846 match self.collect_order_filter_sources(venue, instrument_id, strategy_id, account_id) {
5847 FilterSources::Empty => false,
5848 FilterSources::Unfiltered => side.map_or_else(
5849 || !bucket.is_empty(),
5850 |side| bucket.iter().any(|id| self.order_side_matches(id, side)),
5851 ),
5852 FilterSources::Sets(mut sources) => {
5853 sources.push(bucket);
5854 sources.sort_unstable_by_key(|s| s.len());
5855 let driver = sources[0];
5856 let rest = &sources[1..];
5857
5858 driver
5859 .iter()
5860 .filter(|id| rest.iter().all(|s| s.contains(id)))
5861 .any(|id| side.is_none_or(|side| self.order_side_matches(id, side)))
5862 }
5863 }
5864 }
5865
5866 fn any_positions_in_bucket(
5867 &self,
5868 bucket: &AHashSet<PositionId>,
5869 venue: Option<&Venue>,
5870 instrument_id: Option<&InstrumentId>,
5871 strategy_id: Option<&StrategyId>,
5872 account_id: Option<&AccountId>,
5873 side: Option<PositionSide>,
5874 ) -> bool {
5875 match self.collect_position_filter_sources(venue, instrument_id, strategy_id, account_id) {
5876 FilterSources::Empty => false,
5877 FilterSources::Unfiltered => side.map_or_else(
5878 || !bucket.is_empty(),
5879 |side| bucket.iter().any(|id| self.position_side_matches(id, side)),
5880 ),
5881 FilterSources::Sets(mut sources) => {
5882 sources.push(bucket);
5883 sources.sort_unstable_by_key(|s| s.len());
5884 let driver = sources[0];
5885 let rest = &sources[1..];
5886
5887 driver
5888 .iter()
5889 .filter(|id| rest.iter().all(|s| s.contains(id)))
5890 .any(|id| side.is_none_or(|side| self.position_side_matches(id, side)))
5891 }
5892 }
5893 }
5894
5895 fn order_side_matches(&self, client_order_id: &ClientOrderId, side: OrderSide) -> bool {
5896 self.orders
5897 .get(client_order_id)
5898 .is_some_and(|cell| cell.borrow().order_side() == side)
5899 }
5900
5901 fn position_side_matches(&self, position_id: &PositionId, side: PositionSide) -> bool {
5902 self.positions
5903 .get(position_id)
5904 .is_some_and(|cell| cell.borrow().side == side)
5905 }
5906
5907 fn get_orders_for_ids(
5913 &self,
5914 client_order_ids: &AHashSet<ClientOrderId>,
5915 side: Option<OrderSide>,
5916 ) -> Vec<OrderRef<'_>> {
5917 let mut orders = Vec::new();
5918
5919 for client_order_id in client_order_ids {
5920 let order_cell = self
5921 .orders
5922 .get(client_order_id)
5923 .unwrap_or_else(|| panic!("Order {client_order_id} not found"));
5924 let order = OrderRef::new(order_cell.borrow());
5925
5926 if side.is_none_or(|side| side == order.order_side()) {
5927 orders.push(order);
5928 }
5929 }
5930
5931 orders.sort_by_key(|o| o.client_order_id());
5934 orders
5935 }
5936
5937 fn get_positions_for_ids(
5947 &self,
5948 position_ids: &AHashSet<PositionId>,
5949 side: Option<PositionSide>,
5950 ) -> Vec<PositionRef<'_>> {
5951 let mut positions = Vec::new();
5952
5953 for position_id in position_ids {
5954 let position_cell = self
5955 .positions
5956 .get(position_id)
5957 .unwrap_or_else(|| panic!("Position {position_id} not found"));
5958 let position = PositionRef::new(position_cell.borrow());
5959
5960 if side.is_none_or(|side| side == position.side) {
5961 positions.push(position);
5962 }
5963 }
5964
5965 positions.sort_by_key(|p| p.id);
5968 positions
5969 }
5970
5971 #[must_use]
5973 pub fn client_order_ids(
5974 &self,
5975 venue: Option<&Venue>,
5976 instrument_id: Option<&InstrumentId>,
5977 strategy_id: Option<&StrategyId>,
5978 account_id: Option<&AccountId>,
5979 ) -> AHashSet<ClientOrderId> {
5980 self.query_orders_in_bucket(
5981 &self.index.orders,
5982 venue,
5983 instrument_id,
5984 strategy_id,
5985 account_id,
5986 )
5987 }
5988
5989 #[must_use]
5991 pub fn client_order_ids_open(
5992 &self,
5993 venue: Option<&Venue>,
5994 instrument_id: Option<&InstrumentId>,
5995 strategy_id: Option<&StrategyId>,
5996 account_id: Option<&AccountId>,
5997 ) -> AHashSet<ClientOrderId> {
5998 self.query_orders_in_bucket(
5999 &self.index.orders_open,
6000 venue,
6001 instrument_id,
6002 strategy_id,
6003 account_id,
6004 )
6005 }
6006
6007 #[must_use]
6009 pub fn client_order_ids_closed(
6010 &self,
6011 venue: Option<&Venue>,
6012 instrument_id: Option<&InstrumentId>,
6013 strategy_id: Option<&StrategyId>,
6014 account_id: Option<&AccountId>,
6015 ) -> AHashSet<ClientOrderId> {
6016 self.query_orders_in_bucket(
6017 &self.index.orders_closed,
6018 venue,
6019 instrument_id,
6020 strategy_id,
6021 account_id,
6022 )
6023 }
6024
6025 #[must_use]
6030 pub fn client_order_ids_active_local(
6031 &self,
6032 venue: Option<&Venue>,
6033 instrument_id: Option<&InstrumentId>,
6034 strategy_id: Option<&StrategyId>,
6035 account_id: Option<&AccountId>,
6036 ) -> AHashSet<ClientOrderId> {
6037 self.query_orders_in_bucket(
6038 &self.index.orders_active_local,
6039 venue,
6040 instrument_id,
6041 strategy_id,
6042 account_id,
6043 )
6044 }
6045
6046 #[must_use]
6048 pub fn client_order_ids_emulated(
6049 &self,
6050 venue: Option<&Venue>,
6051 instrument_id: Option<&InstrumentId>,
6052 strategy_id: Option<&StrategyId>,
6053 account_id: Option<&AccountId>,
6054 ) -> AHashSet<ClientOrderId> {
6055 self.query_orders_in_bucket(
6056 &self.index.orders_emulated,
6057 venue,
6058 instrument_id,
6059 strategy_id,
6060 account_id,
6061 )
6062 }
6063
6064 #[must_use]
6066 pub fn client_order_ids_inflight(
6067 &self,
6068 venue: Option<&Venue>,
6069 instrument_id: Option<&InstrumentId>,
6070 strategy_id: Option<&StrategyId>,
6071 account_id: Option<&AccountId>,
6072 ) -> AHashSet<ClientOrderId> {
6073 self.query_orders_in_bucket(
6074 &self.index.orders_inflight,
6075 venue,
6076 instrument_id,
6077 strategy_id,
6078 account_id,
6079 )
6080 }
6081
6082 #[must_use]
6084 pub fn position_ids(
6085 &self,
6086 venue: Option<&Venue>,
6087 instrument_id: Option<&InstrumentId>,
6088 strategy_id: Option<&StrategyId>,
6089 account_id: Option<&AccountId>,
6090 ) -> AHashSet<PositionId> {
6091 self.query_positions_in_bucket(
6092 &self.index.positions,
6093 venue,
6094 instrument_id,
6095 strategy_id,
6096 account_id,
6097 )
6098 }
6099
6100 #[must_use]
6102 pub fn position_open_ids(
6103 &self,
6104 venue: Option<&Venue>,
6105 instrument_id: Option<&InstrumentId>,
6106 strategy_id: Option<&StrategyId>,
6107 account_id: Option<&AccountId>,
6108 ) -> AHashSet<PositionId> {
6109 self.query_positions_in_bucket(
6110 &self.index.positions_open,
6111 venue,
6112 instrument_id,
6113 strategy_id,
6114 account_id,
6115 )
6116 }
6117
6118 #[must_use]
6120 pub fn position_closed_ids(
6121 &self,
6122 venue: Option<&Venue>,
6123 instrument_id: Option<&InstrumentId>,
6124 strategy_id: Option<&StrategyId>,
6125 account_id: Option<&AccountId>,
6126 ) -> AHashSet<PositionId> {
6127 self.query_positions_in_bucket(
6128 &self.index.positions_closed,
6129 venue,
6130 instrument_id,
6131 strategy_id,
6132 account_id,
6133 )
6134 }
6135
6136 #[must_use]
6143 pub fn client_order_ids_view(
6144 &self,
6145 venue: Option<&Venue>,
6146 instrument_id: Option<&InstrumentId>,
6147 strategy_id: Option<&StrategyId>,
6148 account_id: Option<&AccountId>,
6149 ) -> Cow<'_, AHashSet<ClientOrderId>> {
6150 self.view_orders_in_bucket(
6151 &self.index.orders,
6152 venue,
6153 instrument_id,
6154 strategy_id,
6155 account_id,
6156 )
6157 }
6158
6159 #[must_use]
6161 pub fn client_order_ids_open_view(
6162 &self,
6163 venue: Option<&Venue>,
6164 instrument_id: Option<&InstrumentId>,
6165 strategy_id: Option<&StrategyId>,
6166 account_id: Option<&AccountId>,
6167 ) -> Cow<'_, AHashSet<ClientOrderId>> {
6168 self.view_orders_in_bucket(
6169 &self.index.orders_open,
6170 venue,
6171 instrument_id,
6172 strategy_id,
6173 account_id,
6174 )
6175 }
6176
6177 #[must_use]
6179 pub fn client_order_ids_closed_view(
6180 &self,
6181 venue: Option<&Venue>,
6182 instrument_id: Option<&InstrumentId>,
6183 strategy_id: Option<&StrategyId>,
6184 account_id: Option<&AccountId>,
6185 ) -> Cow<'_, AHashSet<ClientOrderId>> {
6186 self.view_orders_in_bucket(
6187 &self.index.orders_closed,
6188 venue,
6189 instrument_id,
6190 strategy_id,
6191 account_id,
6192 )
6193 }
6194
6195 #[must_use]
6197 pub fn client_order_ids_active_local_view(
6198 &self,
6199 venue: Option<&Venue>,
6200 instrument_id: Option<&InstrumentId>,
6201 strategy_id: Option<&StrategyId>,
6202 account_id: Option<&AccountId>,
6203 ) -> Cow<'_, AHashSet<ClientOrderId>> {
6204 self.view_orders_in_bucket(
6205 &self.index.orders_active_local,
6206 venue,
6207 instrument_id,
6208 strategy_id,
6209 account_id,
6210 )
6211 }
6212
6213 #[must_use]
6215 pub fn client_order_ids_emulated_view(
6216 &self,
6217 venue: Option<&Venue>,
6218 instrument_id: Option<&InstrumentId>,
6219 strategy_id: Option<&StrategyId>,
6220 account_id: Option<&AccountId>,
6221 ) -> Cow<'_, AHashSet<ClientOrderId>> {
6222 self.view_orders_in_bucket(
6223 &self.index.orders_emulated,
6224 venue,
6225 instrument_id,
6226 strategy_id,
6227 account_id,
6228 )
6229 }
6230
6231 #[must_use]
6233 pub fn client_order_ids_inflight_view(
6234 &self,
6235 venue: Option<&Venue>,
6236 instrument_id: Option<&InstrumentId>,
6237 strategy_id: Option<&StrategyId>,
6238 account_id: Option<&AccountId>,
6239 ) -> Cow<'_, AHashSet<ClientOrderId>> {
6240 self.view_orders_in_bucket(
6241 &self.index.orders_inflight,
6242 venue,
6243 instrument_id,
6244 strategy_id,
6245 account_id,
6246 )
6247 }
6248
6249 #[must_use]
6251 pub fn position_ids_view(
6252 &self,
6253 venue: Option<&Venue>,
6254 instrument_id: Option<&InstrumentId>,
6255 strategy_id: Option<&StrategyId>,
6256 account_id: Option<&AccountId>,
6257 ) -> Cow<'_, AHashSet<PositionId>> {
6258 self.view_positions_in_bucket(
6259 &self.index.positions,
6260 venue,
6261 instrument_id,
6262 strategy_id,
6263 account_id,
6264 )
6265 }
6266
6267 #[must_use]
6269 pub fn position_open_ids_view(
6270 &self,
6271 venue: Option<&Venue>,
6272 instrument_id: Option<&InstrumentId>,
6273 strategy_id: Option<&StrategyId>,
6274 account_id: Option<&AccountId>,
6275 ) -> Cow<'_, AHashSet<PositionId>> {
6276 self.view_positions_in_bucket(
6277 &self.index.positions_open,
6278 venue,
6279 instrument_id,
6280 strategy_id,
6281 account_id,
6282 )
6283 }
6284
6285 #[must_use]
6287 pub fn position_closed_ids_view(
6288 &self,
6289 venue: Option<&Venue>,
6290 instrument_id: Option<&InstrumentId>,
6291 strategy_id: Option<&StrategyId>,
6292 account_id: Option<&AccountId>,
6293 ) -> Cow<'_, AHashSet<PositionId>> {
6294 self.view_positions_in_bucket(
6295 &self.index.positions_closed,
6296 venue,
6297 instrument_id,
6298 strategy_id,
6299 account_id,
6300 )
6301 }
6302
6303 pub fn iter_client_order_ids(
6309 &self,
6310 venue: Option<&Venue>,
6311 instrument_id: Option<&InstrumentId>,
6312 strategy_id: Option<&StrategyId>,
6313 account_id: Option<&AccountId>,
6314 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6315 self.iter_orders_in_bucket(
6316 &self.index.orders,
6317 venue,
6318 instrument_id,
6319 strategy_id,
6320 account_id,
6321 )
6322 }
6323
6324 pub fn iter_client_order_ids_open(
6326 &self,
6327 venue: Option<&Venue>,
6328 instrument_id: Option<&InstrumentId>,
6329 strategy_id: Option<&StrategyId>,
6330 account_id: Option<&AccountId>,
6331 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6332 self.iter_orders_in_bucket(
6333 &self.index.orders_open,
6334 venue,
6335 instrument_id,
6336 strategy_id,
6337 account_id,
6338 )
6339 }
6340
6341 pub fn iter_client_order_ids_closed(
6343 &self,
6344 venue: Option<&Venue>,
6345 instrument_id: Option<&InstrumentId>,
6346 strategy_id: Option<&StrategyId>,
6347 account_id: Option<&AccountId>,
6348 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6349 self.iter_orders_in_bucket(
6350 &self.index.orders_closed,
6351 venue,
6352 instrument_id,
6353 strategy_id,
6354 account_id,
6355 )
6356 }
6357
6358 pub fn iter_client_order_ids_active_local(
6360 &self,
6361 venue: Option<&Venue>,
6362 instrument_id: Option<&InstrumentId>,
6363 strategy_id: Option<&StrategyId>,
6364 account_id: Option<&AccountId>,
6365 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6366 self.iter_orders_in_bucket(
6367 &self.index.orders_active_local,
6368 venue,
6369 instrument_id,
6370 strategy_id,
6371 account_id,
6372 )
6373 }
6374
6375 pub fn iter_client_order_ids_emulated(
6377 &self,
6378 venue: Option<&Venue>,
6379 instrument_id: Option<&InstrumentId>,
6380 strategy_id: Option<&StrategyId>,
6381 account_id: Option<&AccountId>,
6382 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6383 self.iter_orders_in_bucket(
6384 &self.index.orders_emulated,
6385 venue,
6386 instrument_id,
6387 strategy_id,
6388 account_id,
6389 )
6390 }
6391
6392 pub fn iter_client_order_ids_inflight(
6394 &self,
6395 venue: Option<&Venue>,
6396 instrument_id: Option<&InstrumentId>,
6397 strategy_id: Option<&StrategyId>,
6398 account_id: Option<&AccountId>,
6399 ) -> Box<dyn Iterator<Item = ClientOrderId> + '_> {
6400 self.iter_orders_in_bucket(
6401 &self.index.orders_inflight,
6402 venue,
6403 instrument_id,
6404 strategy_id,
6405 account_id,
6406 )
6407 }
6408
6409 pub fn iter_position_ids(
6411 &self,
6412 venue: Option<&Venue>,
6413 instrument_id: Option<&InstrumentId>,
6414 strategy_id: Option<&StrategyId>,
6415 account_id: Option<&AccountId>,
6416 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
6417 self.iter_positions_in_bucket(
6418 &self.index.positions,
6419 venue,
6420 instrument_id,
6421 strategy_id,
6422 account_id,
6423 )
6424 }
6425
6426 pub fn iter_position_open_ids(
6428 &self,
6429 venue: Option<&Venue>,
6430 instrument_id: Option<&InstrumentId>,
6431 strategy_id: Option<&StrategyId>,
6432 account_id: Option<&AccountId>,
6433 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
6434 self.iter_positions_in_bucket(
6435 &self.index.positions_open,
6436 venue,
6437 instrument_id,
6438 strategy_id,
6439 account_id,
6440 )
6441 }
6442
6443 pub fn iter_position_closed_ids(
6445 &self,
6446 venue: Option<&Venue>,
6447 instrument_id: Option<&InstrumentId>,
6448 strategy_id: Option<&StrategyId>,
6449 account_id: Option<&AccountId>,
6450 ) -> Box<dyn Iterator<Item = PositionId> + '_> {
6451 self.iter_positions_in_bucket(
6452 &self.index.positions_closed,
6453 venue,
6454 instrument_id,
6455 strategy_id,
6456 account_id,
6457 )
6458 }
6459
6460 #[must_use]
6462 pub fn strategy_ids(&self) -> AHashSet<StrategyId> {
6463 self.index.strategies.clone()
6464 }
6465
6466 #[must_use]
6468 pub fn exec_algorithm_ids(&self) -> AHashSet<ExecAlgorithmId> {
6469 self.index.exec_algorithms.clone()
6470 }
6471
6472 #[must_use]
6481 pub fn order_ref(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
6482 self.orders
6483 .get(client_order_id)
6484 .map(|order_cell| OrderRef::new(order_cell.borrow()))
6485 }
6486
6487 #[must_use]
6491 pub fn order(&self, client_order_id: &ClientOrderId) -> Option<OrderRef<'_>> {
6492 self.order_ref(client_order_id)
6493 }
6494
6495 pub fn try_order_ref(
6501 &self,
6502 client_order_id: &ClientOrderId,
6503 ) -> Result<OrderRef<'_>, OrderLookupError> {
6504 self.orders
6505 .get(client_order_id)
6506 .map(|order_cell| OrderRef::new(order_cell.borrow()))
6507 .ok_or_else(|| OrderLookupError::not_found(*client_order_id))
6508 }
6509
6510 pub fn try_order(
6518 &self,
6519 client_order_id: &ClientOrderId,
6520 ) -> Result<OrderRef<'_>, OrderLookupError> {
6521 self.try_order_ref(client_order_id)
6522 }
6523
6524 #[must_use]
6534 pub fn order_mut(&mut self, client_order_id: &ClientOrderId) -> Option<OrderRefMut<'_>> {
6535 self.orders
6536 .get(client_order_id)
6537 .map(|order_cell| OrderRefMut::new(order_cell.borrow_mut()))
6538 }
6539
6540 #[must_use]
6545 pub fn order_owned(&self, client_order_id: &ClientOrderId) -> Option<OrderAny> {
6546 self.orders
6547 .get(client_order_id)
6548 .map(|order_cell| order_cell.borrow().clone())
6549 }
6550
6551 pub fn try_order_owned(
6557 &self,
6558 client_order_id: &ClientOrderId,
6559 ) -> Result<OrderAny, OrderLookupError> {
6560 self.try_order_ref(client_order_id)
6561 .map(|order| order.cloned())
6562 }
6563
6564 #[must_use]
6566 pub fn orders_for_ids(
6567 &self,
6568 client_order_ids: &[ClientOrderId],
6569 context: &dyn Display,
6570 ) -> Vec<OrderAny> {
6571 let mut orders = Vec::with_capacity(client_order_ids.len());
6572 for id in client_order_ids {
6573 match self.orders.get(id) {
6574 Some(order_cell) => orders.push(order_cell.borrow().clone()),
6575 None => log::error!("Order {id} not found in cache for {context}"),
6576 }
6577 }
6578 orders
6579 }
6580
6581 #[must_use]
6583 pub fn client_order_id(&self, venue_order_id: &VenueOrderId) -> Option<&ClientOrderId> {
6584 self.index.venue_order_ids.get(venue_order_id)
6585 }
6586
6587 #[must_use]
6589 pub fn venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<&VenueOrderId> {
6590 self.index.client_order_ids.get(client_order_id)
6591 }
6592
6593 #[must_use]
6595 pub fn client_id(&self, client_order_id: &ClientOrderId) -> Option<&ClientId> {
6596 self.index.order_client.get(client_order_id)
6597 }
6598
6599 #[must_use]
6605 pub fn orders_refs(
6606 &self,
6607 venue: Option<&Venue>,
6608 instrument_id: Option<&InstrumentId>,
6609 strategy_id: Option<&StrategyId>,
6610 account_id: Option<&AccountId>,
6611 side: Option<OrderSide>,
6612 ) -> Vec<OrderRef<'_>> {
6613 let client_order_ids = self.client_order_ids(venue, instrument_id, strategy_id, account_id);
6614 self.get_orders_for_ids(&client_order_ids, side)
6615 }
6616
6617 #[must_use]
6621 pub fn orders(
6622 &self,
6623 venue: Option<&Venue>,
6624 instrument_id: Option<&InstrumentId>,
6625 strategy_id: Option<&StrategyId>,
6626 account_id: Option<&AccountId>,
6627 side: Option<OrderSide>,
6628 ) -> Vec<OrderRef<'_>> {
6629 self.orders_refs(venue, instrument_id, strategy_id, account_id, side)
6630 }
6631
6632 #[must_use]
6634 pub fn orders_open_refs(
6635 &self,
6636 venue: Option<&Venue>,
6637 instrument_id: Option<&InstrumentId>,
6638 strategy_id: Option<&StrategyId>,
6639 account_id: Option<&AccountId>,
6640 side: Option<OrderSide>,
6641 ) -> Vec<OrderRef<'_>> {
6642 let client_order_ids =
6643 self.client_order_ids_open(venue, instrument_id, strategy_id, account_id);
6644 self.get_orders_for_ids(&client_order_ids, side)
6645 }
6646
6647 #[must_use]
6651 pub fn orders_open(
6652 &self,
6653 venue: Option<&Venue>,
6654 instrument_id: Option<&InstrumentId>,
6655 strategy_id: Option<&StrategyId>,
6656 account_id: Option<&AccountId>,
6657 side: Option<OrderSide>,
6658 ) -> Vec<OrderRef<'_>> {
6659 self.orders_open_refs(venue, instrument_id, strategy_id, account_id, side)
6660 }
6661
6662 #[must_use]
6664 pub fn orders_closed_refs(
6665 &self,
6666 venue: Option<&Venue>,
6667 instrument_id: Option<&InstrumentId>,
6668 strategy_id: Option<&StrategyId>,
6669 account_id: Option<&AccountId>,
6670 side: Option<OrderSide>,
6671 ) -> Vec<OrderRef<'_>> {
6672 let client_order_ids =
6673 self.client_order_ids_closed(venue, instrument_id, strategy_id, account_id);
6674 self.get_orders_for_ids(&client_order_ids, side)
6675 }
6676
6677 #[must_use]
6681 pub fn orders_closed(
6682 &self,
6683 venue: Option<&Venue>,
6684 instrument_id: Option<&InstrumentId>,
6685 strategy_id: Option<&StrategyId>,
6686 account_id: Option<&AccountId>,
6687 side: Option<OrderSide>,
6688 ) -> Vec<OrderRef<'_>> {
6689 self.orders_closed_refs(venue, instrument_id, strategy_id, account_id, side)
6690 }
6691
6692 #[must_use]
6697 pub fn orders_active_local_refs(
6698 &self,
6699 venue: Option<&Venue>,
6700 instrument_id: Option<&InstrumentId>,
6701 strategy_id: Option<&StrategyId>,
6702 account_id: Option<&AccountId>,
6703 side: Option<OrderSide>,
6704 ) -> Vec<OrderRef<'_>> {
6705 let client_order_ids =
6706 self.client_order_ids_active_local(venue, instrument_id, strategy_id, account_id);
6707 self.get_orders_for_ids(&client_order_ids, side)
6708 }
6709
6710 #[must_use]
6714 pub fn orders_active_local(
6715 &self,
6716 venue: Option<&Venue>,
6717 instrument_id: Option<&InstrumentId>,
6718 strategy_id: Option<&StrategyId>,
6719 account_id: Option<&AccountId>,
6720 side: Option<OrderSide>,
6721 ) -> Vec<OrderRef<'_>> {
6722 self.orders_active_local_refs(venue, instrument_id, strategy_id, account_id, side)
6723 }
6724
6725 #[must_use]
6727 pub fn orders_emulated_refs(
6728 &self,
6729 venue: Option<&Venue>,
6730 instrument_id: Option<&InstrumentId>,
6731 strategy_id: Option<&StrategyId>,
6732 account_id: Option<&AccountId>,
6733 side: Option<OrderSide>,
6734 ) -> Vec<OrderRef<'_>> {
6735 let client_order_ids =
6736 self.client_order_ids_emulated(venue, instrument_id, strategy_id, account_id);
6737 self.get_orders_for_ids(&client_order_ids, side)
6738 }
6739
6740 #[must_use]
6744 pub fn orders_emulated(
6745 &self,
6746 venue: Option<&Venue>,
6747 instrument_id: Option<&InstrumentId>,
6748 strategy_id: Option<&StrategyId>,
6749 account_id: Option<&AccountId>,
6750 side: Option<OrderSide>,
6751 ) -> Vec<OrderRef<'_>> {
6752 self.orders_emulated_refs(venue, instrument_id, strategy_id, account_id, side)
6753 }
6754
6755 #[must_use]
6757 pub fn orders_inflight_refs(
6758 &self,
6759 venue: Option<&Venue>,
6760 instrument_id: Option<&InstrumentId>,
6761 strategy_id: Option<&StrategyId>,
6762 account_id: Option<&AccountId>,
6763 side: Option<OrderSide>,
6764 ) -> Vec<OrderRef<'_>> {
6765 let client_order_ids =
6766 self.client_order_ids_inflight(venue, instrument_id, strategy_id, account_id);
6767 self.get_orders_for_ids(&client_order_ids, side)
6768 }
6769
6770 #[must_use]
6774 pub fn orders_inflight(
6775 &self,
6776 venue: Option<&Venue>,
6777 instrument_id: Option<&InstrumentId>,
6778 strategy_id: Option<&StrategyId>,
6779 account_id: Option<&AccountId>,
6780 side: Option<OrderSide>,
6781 ) -> Vec<OrderRef<'_>> {
6782 self.orders_inflight_refs(venue, instrument_id, strategy_id, account_id, side)
6783 }
6784
6785 #[must_use]
6787 pub fn orders_for_position(&self, position_id: &PositionId) -> Vec<OrderRef<'_>> {
6788 match self.index.position_orders.get(position_id) {
6789 Some(client_order_ids) => self.get_orders_for_ids(client_order_ids, None),
6790 None => Vec::new(),
6791 }
6792 }
6793
6794 #[must_use]
6796 pub fn order_exists(&self, client_order_id: &ClientOrderId) -> bool {
6797 self.index.orders.contains(client_order_id)
6798 }
6799
6800 #[must_use]
6802 pub fn is_order_open(&self, client_order_id: &ClientOrderId) -> bool {
6803 self.index.orders_open.contains(client_order_id)
6804 }
6805
6806 #[must_use]
6808 pub fn is_order_closed(&self, client_order_id: &ClientOrderId) -> bool {
6809 self.index.orders_closed.contains(client_order_id)
6810 }
6811
6812 #[must_use]
6817 pub fn is_order_active_local(&self, client_order_id: &ClientOrderId) -> bool {
6818 self.index.orders_active_local.contains(client_order_id)
6819 }
6820
6821 #[must_use]
6823 pub fn is_order_emulated(&self, client_order_id: &ClientOrderId) -> bool {
6824 self.index.orders_emulated.contains(client_order_id)
6825 }
6826
6827 #[must_use]
6829 pub fn is_order_inflight(&self, client_order_id: &ClientOrderId) -> bool {
6830 self.index.orders_inflight.contains(client_order_id)
6831 }
6832
6833 #[must_use]
6835 pub fn is_order_pending_cancel_local(&self, client_order_id: &ClientOrderId) -> bool {
6836 self.index.orders_pending_cancel.contains(client_order_id)
6837 }
6838
6839 #[must_use]
6841 pub fn orders_open_count(
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 ) -> usize {
6849 self.count_orders_in_bucket(
6850 &self.index.orders_open,
6851 venue,
6852 instrument_id,
6853 strategy_id,
6854 account_id,
6855 side,
6856 )
6857 }
6858
6859 #[must_use]
6861 pub fn orders_closed_count(
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 ) -> usize {
6869 self.count_orders_in_bucket(
6870 &self.index.orders_closed,
6871 venue,
6872 instrument_id,
6873 strategy_id,
6874 account_id,
6875 side,
6876 )
6877 }
6878
6879 #[must_use]
6884 pub fn orders_active_local_count(
6885 &self,
6886 venue: Option<&Venue>,
6887 instrument_id: Option<&InstrumentId>,
6888 strategy_id: Option<&StrategyId>,
6889 account_id: Option<&AccountId>,
6890 side: Option<OrderSide>,
6891 ) -> usize {
6892 self.count_orders_in_bucket(
6893 &self.index.orders_active_local,
6894 venue,
6895 instrument_id,
6896 strategy_id,
6897 account_id,
6898 side,
6899 )
6900 }
6901
6902 #[must_use]
6904 pub fn orders_emulated_count(
6905 &self,
6906 venue: Option<&Venue>,
6907 instrument_id: Option<&InstrumentId>,
6908 strategy_id: Option<&StrategyId>,
6909 account_id: Option<&AccountId>,
6910 side: Option<OrderSide>,
6911 ) -> usize {
6912 self.count_orders_in_bucket(
6913 &self.index.orders_emulated,
6914 venue,
6915 instrument_id,
6916 strategy_id,
6917 account_id,
6918 side,
6919 )
6920 }
6921
6922 #[must_use]
6924 pub fn orders_inflight_count(
6925 &self,
6926 venue: Option<&Venue>,
6927 instrument_id: Option<&InstrumentId>,
6928 strategy_id: Option<&StrategyId>,
6929 account_id: Option<&AccountId>,
6930 side: Option<OrderSide>,
6931 ) -> usize {
6932 self.count_orders_in_bucket(
6933 &self.index.orders_inflight,
6934 venue,
6935 instrument_id,
6936 strategy_id,
6937 account_id,
6938 side,
6939 )
6940 }
6941
6942 #[must_use]
6944 pub fn orders_total_count(
6945 &self,
6946 venue: Option<&Venue>,
6947 instrument_id: Option<&InstrumentId>,
6948 strategy_id: Option<&StrategyId>,
6949 account_id: Option<&AccountId>,
6950 side: Option<OrderSide>,
6951 ) -> usize {
6952 self.count_orders_in_bucket(
6953 &self.index.orders,
6954 venue,
6955 instrument_id,
6956 strategy_id,
6957 account_id,
6958 side,
6959 )
6960 }
6961
6962 #[must_use]
6968 pub fn has_orders_open(
6969 &self,
6970 venue: Option<&Venue>,
6971 instrument_id: Option<&InstrumentId>,
6972 strategy_id: Option<&StrategyId>,
6973 account_id: Option<&AccountId>,
6974 side: Option<OrderSide>,
6975 ) -> bool {
6976 self.any_orders_in_bucket(
6977 &self.index.orders_open,
6978 venue,
6979 instrument_id,
6980 strategy_id,
6981 account_id,
6982 side,
6983 )
6984 }
6985
6986 #[must_use]
6988 pub fn has_orders_closed(
6989 &self,
6990 venue: Option<&Venue>,
6991 instrument_id: Option<&InstrumentId>,
6992 strategy_id: Option<&StrategyId>,
6993 account_id: Option<&AccountId>,
6994 side: Option<OrderSide>,
6995 ) -> bool {
6996 self.any_orders_in_bucket(
6997 &self.index.orders_closed,
6998 venue,
6999 instrument_id,
7000 strategy_id,
7001 account_id,
7002 side,
7003 )
7004 }
7005
7006 #[must_use]
7010 pub fn has_orders_active_local(
7011 &self,
7012 venue: Option<&Venue>,
7013 instrument_id: Option<&InstrumentId>,
7014 strategy_id: Option<&StrategyId>,
7015 account_id: Option<&AccountId>,
7016 side: Option<OrderSide>,
7017 ) -> bool {
7018 self.any_orders_in_bucket(
7019 &self.index.orders_active_local,
7020 venue,
7021 instrument_id,
7022 strategy_id,
7023 account_id,
7024 side,
7025 )
7026 }
7027
7028 #[must_use]
7030 pub fn has_orders_emulated(
7031 &self,
7032 venue: Option<&Venue>,
7033 instrument_id: Option<&InstrumentId>,
7034 strategy_id: Option<&StrategyId>,
7035 account_id: Option<&AccountId>,
7036 side: Option<OrderSide>,
7037 ) -> bool {
7038 self.any_orders_in_bucket(
7039 &self.index.orders_emulated,
7040 venue,
7041 instrument_id,
7042 strategy_id,
7043 account_id,
7044 side,
7045 )
7046 }
7047
7048 #[must_use]
7050 pub fn has_orders_inflight(
7051 &self,
7052 venue: Option<&Venue>,
7053 instrument_id: Option<&InstrumentId>,
7054 strategy_id: Option<&StrategyId>,
7055 account_id: Option<&AccountId>,
7056 side: Option<OrderSide>,
7057 ) -> bool {
7058 self.any_orders_in_bucket(
7059 &self.index.orders_inflight,
7060 venue,
7061 instrument_id,
7062 strategy_id,
7063 account_id,
7064 side,
7065 )
7066 }
7067
7068 #[must_use]
7070 pub fn has_orders(
7071 &self,
7072 venue: Option<&Venue>,
7073 instrument_id: Option<&InstrumentId>,
7074 strategy_id: Option<&StrategyId>,
7075 account_id: Option<&AccountId>,
7076 side: Option<OrderSide>,
7077 ) -> bool {
7078 self.any_orders_in_bucket(
7079 &self.index.orders,
7080 venue,
7081 instrument_id,
7082 strategy_id,
7083 account_id,
7084 side,
7085 )
7086 }
7087
7088 #[must_use]
7090 pub fn order_list(&self, order_list_id: &OrderListId) -> Option<&OrderList> {
7091 self.order_lists.get(order_list_id)
7092 }
7093
7094 pub fn try_order_list(
7100 &self,
7101 order_list_id: &OrderListId,
7102 ) -> Result<&OrderList, OrderListLookupError> {
7103 self.order_lists
7104 .get(order_list_id)
7105 .ok_or_else(|| OrderListLookupError::not_found(*order_list_id))
7106 }
7107
7108 #[must_use]
7110 pub fn order_lists(
7111 &self,
7112 venue: Option<&Venue>,
7113 instrument_id: Option<&InstrumentId>,
7114 strategy_id: Option<&StrategyId>,
7115 account_id: Option<&AccountId>,
7116 ) -> Vec<&OrderList> {
7117 let mut order_lists = self.order_lists.values().collect::<Vec<&OrderList>>();
7118
7119 if let Some(venue) = venue {
7120 order_lists.retain(|ol| &ol.instrument_id.venue == venue);
7121 }
7122
7123 if let Some(instrument_id) = instrument_id {
7124 order_lists.retain(|ol| &ol.instrument_id == instrument_id);
7125 }
7126
7127 if let Some(strategy_id) = strategy_id {
7128 order_lists.retain(|ol| &ol.strategy_id == strategy_id);
7129 }
7130
7131 if let Some(account_id) = account_id {
7132 order_lists.retain(|ol| {
7133 ol.client_order_ids.iter().any(|client_order_id| {
7134 self.orders.get(client_order_id).is_some_and(|order_cell| {
7135 order_cell.borrow().account_id().as_ref() == Some(account_id)
7136 })
7137 })
7138 });
7139 }
7140
7141 order_lists
7142 }
7143
7144 #[must_use]
7146 pub fn order_list_exists(&self, order_list_id: &OrderListId) -> bool {
7147 self.order_lists.contains_key(order_list_id)
7148 }
7149
7150 #[must_use]
7155 pub fn orders_for_exec_algorithm(
7156 &self,
7157 exec_algorithm_id: &ExecAlgorithmId,
7158 venue: Option<&Venue>,
7159 instrument_id: Option<&InstrumentId>,
7160 strategy_id: Option<&StrategyId>,
7161 account_id: Option<&AccountId>,
7162 side: Option<OrderSide>,
7163 ) -> Vec<OrderRef<'_>> {
7164 let Some(exec_algorithm_order_ids) =
7165 self.index.exec_algorithm_orders.get(exec_algorithm_id)
7166 else {
7167 return Vec::new();
7168 };
7169
7170 let filtered = self.query_orders_in_bucket(
7171 exec_algorithm_order_ids,
7172 venue,
7173 instrument_id,
7174 strategy_id,
7175 account_id,
7176 );
7177 self.get_orders_for_ids(&filtered, side)
7178 }
7179
7180 #[must_use]
7182 pub fn orders_for_exec_spawn(&self, exec_spawn_id: &ClientOrderId) -> Vec<OrderRef<'_>> {
7183 match self.index.exec_spawn_orders.get(exec_spawn_id) {
7184 Some(ids) => self.get_orders_for_ids(ids, None),
7185 None => Vec::new(),
7186 }
7187 }
7188
7189 #[must_use]
7191 pub fn exec_spawn_total_quantity(
7192 &self,
7193 exec_spawn_id: &ClientOrderId,
7194 active_only: bool,
7195 ) -> Option<Quantity> {
7196 self.exec_spawn_total(exec_spawn_id, active_only, Order::quantity)
7197 }
7198
7199 #[must_use]
7201 pub fn exec_spawn_total_filled_qty(
7202 &self,
7203 exec_spawn_id: &ClientOrderId,
7204 active_only: bool,
7205 ) -> Option<Quantity> {
7206 self.exec_spawn_total(exec_spawn_id, active_only, Order::filled_qty)
7207 }
7208
7209 #[must_use]
7211 pub fn exec_spawn_total_leaves_qty(
7212 &self,
7213 exec_spawn_id: &ClientOrderId,
7214 active_only: bool,
7215 ) -> Option<Quantity> {
7216 self.exec_spawn_total(exec_spawn_id, active_only, Order::leaves_qty)
7217 }
7218
7219 fn exec_spawn_total(
7220 &self,
7221 exec_spawn_id: &ClientOrderId,
7222 active_only: bool,
7223 quantity: impl Fn(&OrderAny) -> Quantity,
7224 ) -> Option<Quantity> {
7225 self.orders_for_exec_spawn(exec_spawn_id)
7226 .into_iter()
7227 .filter(|order| !active_only || !order.is_closed())
7228 .map(|order| quantity(&order))
7229 .reduce(|total, quantity| total + quantity)
7230 }
7231
7232 #[must_use]
7236 pub fn position_ref(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
7237 self.positions
7238 .get(position_id)
7239 .map(|position_cell| PositionRef::new(position_cell.borrow()))
7240 }
7241
7242 #[must_use]
7246 pub fn position(&self, position_id: &PositionId) -> Option<PositionRef<'_>> {
7247 self.position_ref(position_id)
7248 }
7249
7250 pub fn try_position_ref(
7256 &self,
7257 position_id: &PositionId,
7258 ) -> Result<PositionRef<'_>, PositionLookupError> {
7259 self.positions
7260 .get(position_id)
7261 .map(|position_cell| PositionRef::new(position_cell.borrow()))
7262 .ok_or_else(|| PositionLookupError::not_found(*position_id))
7263 }
7264
7265 pub fn try_position(
7273 &self,
7274 position_id: &PositionId,
7275 ) -> Result<PositionRef<'_>, PositionLookupError> {
7276 self.try_position_ref(position_id)
7277 }
7278
7279 #[must_use]
7289 pub fn position_mut(&mut self, position_id: &PositionId) -> Option<PositionRefMut<'_>> {
7290 self.positions
7291 .get(position_id)
7292 .map(|position_cell| PositionRefMut::new(position_cell.borrow_mut()))
7293 }
7294
7295 #[must_use]
7300 pub fn position_owned(&self, position_id: &PositionId) -> Option<Position> {
7301 self.positions
7302 .get(position_id)
7303 .map(|position_cell| position_cell.borrow().clone())
7304 }
7305
7306 #[must_use]
7308 pub fn position_for_order_ref(
7309 &self,
7310 client_order_id: &ClientOrderId,
7311 ) -> Option<PositionRef<'_>> {
7312 self.index
7313 .order_position
7314 .get(client_order_id)
7315 .and_then(|position_id| self.positions.get(position_id))
7316 .map(|position_cell| PositionRef::new(position_cell.borrow()))
7317 }
7318
7319 #[must_use]
7323 pub fn position_for_order(&self, client_order_id: &ClientOrderId) -> Option<PositionRef<'_>> {
7324 self.position_for_order_ref(client_order_id)
7325 }
7326
7327 #[must_use]
7329 pub fn position_id(&self, client_order_id: &ClientOrderId) -> Option<&PositionId> {
7330 self.index.order_position.get(client_order_id)
7331 }
7332
7333 #[must_use]
7339 pub fn positions_refs(
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 ) -> Vec<PositionRef<'_>> {
7347 let position_ids = self.position_ids(venue, instrument_id, strategy_id, account_id);
7348 self.get_positions_for_ids(&position_ids, side)
7349 }
7350
7351 #[must_use]
7355 pub fn positions(
7356 &self,
7357 venue: Option<&Venue>,
7358 instrument_id: Option<&InstrumentId>,
7359 strategy_id: Option<&StrategyId>,
7360 account_id: Option<&AccountId>,
7361 side: Option<PositionSide>,
7362 ) -> Vec<PositionRef<'_>> {
7363 self.positions_refs(venue, instrument_id, strategy_id, account_id, side)
7364 }
7365
7366 #[must_use]
7368 pub fn positions_open_refs(
7369 &self,
7370 venue: Option<&Venue>,
7371 instrument_id: Option<&InstrumentId>,
7372 strategy_id: Option<&StrategyId>,
7373 account_id: Option<&AccountId>,
7374 side: Option<PositionSide>,
7375 ) -> Vec<PositionRef<'_>> {
7376 let position_ids = self.position_open_ids(venue, instrument_id, strategy_id, account_id);
7377 self.get_positions_for_ids(&position_ids, side)
7378 }
7379
7380 #[must_use]
7384 pub fn positions_open(
7385 &self,
7386 venue: Option<&Venue>,
7387 instrument_id: Option<&InstrumentId>,
7388 strategy_id: Option<&StrategyId>,
7389 account_id: Option<&AccountId>,
7390 side: Option<PositionSide>,
7391 ) -> Vec<PositionRef<'_>> {
7392 self.positions_open_refs(venue, instrument_id, strategy_id, account_id, side)
7393 }
7394
7395 #[must_use]
7397 pub fn positions_closed_refs(
7398 &self,
7399 venue: Option<&Venue>,
7400 instrument_id: Option<&InstrumentId>,
7401 strategy_id: Option<&StrategyId>,
7402 account_id: Option<&AccountId>,
7403 side: Option<PositionSide>,
7404 ) -> Vec<PositionRef<'_>> {
7405 let position_ids = self.position_closed_ids(venue, instrument_id, strategy_id, account_id);
7406 self.get_positions_for_ids(&position_ids, side)
7407 }
7408
7409 #[must_use]
7413 pub fn positions_closed(
7414 &self,
7415 venue: Option<&Venue>,
7416 instrument_id: Option<&InstrumentId>,
7417 strategy_id: Option<&StrategyId>,
7418 account_id: Option<&AccountId>,
7419 side: Option<PositionSide>,
7420 ) -> Vec<PositionRef<'_>> {
7421 self.positions_closed_refs(venue, instrument_id, strategy_id, account_id, side)
7422 }
7423
7424 #[must_use]
7426 pub fn position_exists(&self, position_id: &PositionId) -> bool {
7427 self.index.positions.contains(position_id)
7428 }
7429
7430 #[must_use]
7432 pub fn is_position_open(&self, position_id: &PositionId) -> bool {
7433 self.index.positions_open.contains(position_id)
7434 }
7435
7436 #[must_use]
7438 pub fn is_position_closed(&self, position_id: &PositionId) -> bool {
7439 self.index.positions_closed.contains(position_id)
7440 }
7441
7442 #[must_use]
7444 pub fn positions_open_count(
7445 &self,
7446 venue: Option<&Venue>,
7447 instrument_id: Option<&InstrumentId>,
7448 strategy_id: Option<&StrategyId>,
7449 account_id: Option<&AccountId>,
7450 side: Option<PositionSide>,
7451 ) -> usize {
7452 self.count_positions_in_bucket(
7453 &self.index.positions_open,
7454 venue,
7455 instrument_id,
7456 strategy_id,
7457 account_id,
7458 side,
7459 )
7460 }
7461
7462 #[must_use]
7464 pub fn positions_closed_count(
7465 &self,
7466 venue: Option<&Venue>,
7467 instrument_id: Option<&InstrumentId>,
7468 strategy_id: Option<&StrategyId>,
7469 account_id: Option<&AccountId>,
7470 side: Option<PositionSide>,
7471 ) -> usize {
7472 self.count_positions_in_bucket(
7473 &self.index.positions_closed,
7474 venue,
7475 instrument_id,
7476 strategy_id,
7477 account_id,
7478 side,
7479 )
7480 }
7481
7482 #[must_use]
7484 pub fn positions_total_count(
7485 &self,
7486 venue: Option<&Venue>,
7487 instrument_id: Option<&InstrumentId>,
7488 strategy_id: Option<&StrategyId>,
7489 account_id: Option<&AccountId>,
7490 side: Option<PositionSide>,
7491 ) -> usize {
7492 self.count_positions_in_bucket(
7493 &self.index.positions,
7494 venue,
7495 instrument_id,
7496 strategy_id,
7497 account_id,
7498 side,
7499 )
7500 }
7501
7502 #[must_use]
7508 pub fn has_positions_open(
7509 &self,
7510 venue: Option<&Venue>,
7511 instrument_id: Option<&InstrumentId>,
7512 strategy_id: Option<&StrategyId>,
7513 account_id: Option<&AccountId>,
7514 side: Option<PositionSide>,
7515 ) -> bool {
7516 self.any_positions_in_bucket(
7517 &self.index.positions_open,
7518 venue,
7519 instrument_id,
7520 strategy_id,
7521 account_id,
7522 side,
7523 )
7524 }
7525
7526 #[must_use]
7528 pub fn has_positions_closed(
7529 &self,
7530 venue: Option<&Venue>,
7531 instrument_id: Option<&InstrumentId>,
7532 strategy_id: Option<&StrategyId>,
7533 account_id: Option<&AccountId>,
7534 side: Option<PositionSide>,
7535 ) -> bool {
7536 self.any_positions_in_bucket(
7537 &self.index.positions_closed,
7538 venue,
7539 instrument_id,
7540 strategy_id,
7541 account_id,
7542 side,
7543 )
7544 }
7545
7546 #[must_use]
7548 pub fn has_positions(
7549 &self,
7550 venue: Option<&Venue>,
7551 instrument_id: Option<&InstrumentId>,
7552 strategy_id: Option<&StrategyId>,
7553 account_id: Option<&AccountId>,
7554 side: Option<PositionSide>,
7555 ) -> bool {
7556 self.any_positions_in_bucket(
7557 &self.index.positions,
7558 venue,
7559 instrument_id,
7560 strategy_id,
7561 account_id,
7562 side,
7563 )
7564 }
7565
7566 #[must_use]
7570 pub fn strategy_id_for_order(&self, client_order_id: &ClientOrderId) -> Option<&StrategyId> {
7571 self.index.order_strategy.get(client_order_id)
7572 }
7573
7574 #[must_use]
7576 pub fn strategy_id_for_position(&self, position_id: &PositionId) -> Option<&StrategyId> {
7577 self.index.position_strategy.get(position_id)
7578 }
7579
7580 pub fn get(&self, key: &str) -> anyhow::Result<Option<&Bytes>> {
7588 check_valid_string_ascii(key, stringify!(key))?;
7589
7590 Ok(self.general.get(key))
7591 }
7592
7593 #[must_use]
7602 pub fn price(&self, instrument_id: &InstrumentId, price_type: PriceType) -> Option<Price> {
7603 match price_type {
7604 PriceType::Bid => self
7605 .quotes
7606 .get(instrument_id)
7607 .and_then(|quotes| quotes.front().map(|quote| quote.bid_price)),
7608 PriceType::Ask => self
7609 .quotes
7610 .get(instrument_id)
7611 .and_then(|quotes| quotes.front().map(|quote| quote.ask_price)),
7612 PriceType::Mid => self.quotes.get(instrument_id).and_then(|quotes| {
7613 quotes.front().map(|quote| {
7614 let mid = (quote.ask_price.as_decimal() + quote.bid_price.as_decimal())
7615 / Decimal::TWO;
7616
7617 Price::from_decimal_dp(mid, quote.bid_price.precision + 1)
7618 .expect("Invalid mid price for Cache::price")
7619 })
7620 }),
7621 PriceType::Last => self
7622 .trades
7623 .get(instrument_id)
7624 .and_then(|trades| trades.front().map(|trade| trade.price)),
7625 PriceType::Mark => self
7626 .mark_prices
7627 .get(instrument_id)
7628 .and_then(|marks| marks.front().map(|mark| mark.value)),
7629 }
7630 }
7631
7632 #[must_use]
7634 pub fn quotes(&self, instrument_id: &InstrumentId) -> Option<Vec<QuoteTick>> {
7635 self.quotes
7636 .get(instrument_id)
7637 .map(|quotes| quotes.iter().copied().collect())
7638 }
7639
7640 #[must_use]
7642 pub fn trades(&self, instrument_id: &InstrumentId) -> Option<Vec<TradeTick>> {
7643 self.trades
7644 .get(instrument_id)
7645 .map(|trades| trades.iter().copied().collect())
7646 }
7647
7648 #[must_use]
7650 pub fn mark_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<MarkPriceUpdate>> {
7651 self.mark_prices
7652 .get(instrument_id)
7653 .map(|mark_prices| mark_prices.iter().copied().collect())
7654 }
7655
7656 #[must_use]
7658 pub fn index_prices(&self, instrument_id: &InstrumentId) -> Option<Vec<IndexPriceUpdate>> {
7659 self.index_prices
7660 .get(instrument_id)
7661 .map(|index_prices| index_prices.iter().copied().collect())
7662 }
7663
7664 #[must_use]
7666 pub fn funding_rates(&self, instrument_id: &InstrumentId) -> Option<Vec<FundingRateUpdate>> {
7667 self.funding_rates
7668 .get(instrument_id)
7669 .map(|funding_rates| funding_rates.iter().copied().collect())
7670 }
7671
7672 #[must_use]
7674 pub fn instrument_statuses(
7675 &self,
7676 instrument_id: &InstrumentId,
7677 ) -> Option<Vec<InstrumentStatus>> {
7678 self.instrument_statuses
7679 .get(instrument_id)
7680 .map(|statuses| statuses.iter().copied().collect())
7681 }
7682
7683 #[must_use]
7685 pub fn bars(&self, bar_type: &BarType) -> Option<Vec<Bar>> {
7686 self.bars
7687 .get(bar_type)
7688 .map(|bars| bars.iter().copied().collect())
7689 }
7690
7691 #[must_use]
7693 pub fn order_book(&self, instrument_id: &InstrumentId) -> Option<&OrderBook> {
7694 self.books.get(instrument_id)
7695 }
7696
7697 pub fn try_order_book(
7703 &self,
7704 instrument_id: &InstrumentId,
7705 ) -> Result<&OrderBook, OrderBookLookupError> {
7706 self.books
7707 .get(instrument_id)
7708 .ok_or_else(|| OrderBookLookupError::not_found(*instrument_id))
7709 }
7710
7711 #[must_use]
7713 pub fn order_book_mut(&mut self, instrument_id: &InstrumentId) -> Option<&mut OrderBook> {
7714 self.books.get_mut(instrument_id)
7715 }
7716
7717 #[must_use]
7719 pub fn own_order_book(&self, instrument_id: &InstrumentId) -> Option<&OwnOrderBook> {
7720 self.own_books.get(instrument_id)
7721 }
7722
7723 pub fn try_own_order_book(
7730 &self,
7731 instrument_id: &InstrumentId,
7732 ) -> Result<&OwnOrderBook, OwnOrderBookLookupError> {
7733 self.own_books
7734 .get(instrument_id)
7735 .ok_or_else(|| OwnOrderBookLookupError::not_found(*instrument_id))
7736 }
7737
7738 #[must_use]
7740 pub fn own_order_book_mut(
7741 &mut self,
7742 instrument_id: &InstrumentId,
7743 ) -> Option<&mut OwnOrderBook> {
7744 self.own_books.get_mut(instrument_id)
7745 }
7746
7747 #[must_use]
7749 pub fn quote(&self, instrument_id: &InstrumentId) -> Option<&QuoteTick> {
7750 self.quotes
7751 .get(instrument_id)
7752 .and_then(|quotes| quotes.front())
7753 }
7754
7755 #[must_use]
7759 pub fn quote_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&QuoteTick> {
7760 self.quotes
7761 .get(instrument_id)
7762 .and_then(|quotes| quotes.get(index))
7763 }
7764
7765 #[must_use]
7767 pub fn trade(&self, instrument_id: &InstrumentId) -> Option<&TradeTick> {
7768 self.trades
7769 .get(instrument_id)
7770 .and_then(|trades| trades.front())
7771 }
7772
7773 #[must_use]
7777 pub fn trade_at_index(&self, instrument_id: &InstrumentId, index: usize) -> Option<&TradeTick> {
7778 self.trades
7779 .get(instrument_id)
7780 .and_then(|trades| trades.get(index))
7781 }
7782
7783 #[must_use]
7785 pub fn mark_price(&self, instrument_id: &InstrumentId) -> Option<&MarkPriceUpdate> {
7786 self.mark_prices
7787 .get(instrument_id)
7788 .and_then(|mark_prices| mark_prices.front())
7789 }
7790
7791 #[must_use]
7793 pub fn index_price(&self, instrument_id: &InstrumentId) -> Option<&IndexPriceUpdate> {
7794 self.index_prices
7795 .get(instrument_id)
7796 .and_then(|index_prices| index_prices.front())
7797 }
7798
7799 #[must_use]
7801 pub fn funding_rate(&self, instrument_id: &InstrumentId) -> Option<&FundingRateUpdate> {
7802 self.funding_rates
7803 .get(instrument_id)
7804 .and_then(|funding_rates| funding_rates.front())
7805 }
7806
7807 #[must_use]
7809 pub fn instrument_status(&self, instrument_id: &InstrumentId) -> Option<&InstrumentStatus> {
7810 self.instrument_statuses
7811 .get(instrument_id)
7812 .and_then(|statuses| statuses.front())
7813 }
7814
7815 #[must_use]
7817 pub fn instrument_close(&self, instrument_id: &InstrumentId) -> Option<&InstrumentClose> {
7818 self.instrument_closes.get(instrument_id)
7819 }
7820
7821 #[must_use]
7823 pub fn instrument_close_ids(&self) -> Vec<&InstrumentId> {
7824 self.instrument_closes.keys().collect()
7825 }
7826
7827 #[must_use]
7829 pub fn bar(&self, bar_type: &BarType) -> Option<&Bar> {
7830 self.bars.get(bar_type).and_then(|bars| bars.front())
7831 }
7832
7833 #[must_use]
7837 pub fn bar_at_index(&self, bar_type: &BarType, index: usize) -> Option<&Bar> {
7838 self.bars.get(bar_type).and_then(|bars| bars.get(index))
7839 }
7840
7841 #[must_use]
7843 pub fn book_update_count(&self, instrument_id: &InstrumentId) -> usize {
7844 self.books
7845 .get(instrument_id)
7846 .map_or(0, |book| book.update_count) as usize
7847 }
7848
7849 #[must_use]
7851 pub fn quote_count(&self, instrument_id: &InstrumentId) -> usize {
7852 self.quotes
7853 .get(instrument_id)
7854 .map_or(0, BoundedVecDeque::len)
7855 }
7856
7857 #[must_use]
7859 pub fn trade_count(&self, instrument_id: &InstrumentId) -> usize {
7860 self.trades
7861 .get(instrument_id)
7862 .map_or(0, BoundedVecDeque::len)
7863 }
7864
7865 #[must_use]
7867 pub fn mark_price_count(&self, instrument_id: &InstrumentId) -> usize {
7868 self.mark_prices
7869 .get(instrument_id)
7870 .map_or(0, BoundedVecDeque::len)
7871 }
7872
7873 #[must_use]
7875 pub fn index_price_count(&self, instrument_id: &InstrumentId) -> usize {
7876 self.index_prices
7877 .get(instrument_id)
7878 .map_or(0, BoundedVecDeque::len)
7879 }
7880
7881 #[must_use]
7883 pub fn funding_rate_count(&self, instrument_id: &InstrumentId) -> usize {
7884 self.funding_rates
7885 .get(instrument_id)
7886 .map_or(0, BoundedVecDeque::len)
7887 }
7888
7889 #[must_use]
7891 pub fn instrument_status_count(&self, instrument_id: &InstrumentId) -> usize {
7892 self.instrument_statuses
7893 .get(instrument_id)
7894 .map_or(0, BoundedVecDeque::len)
7895 }
7896
7897 #[must_use]
7899 pub fn bar_count(&self, bar_type: &BarType) -> usize {
7900 self.bars.get(bar_type).map_or(0, BoundedVecDeque::len)
7901 }
7902
7903 #[must_use]
7905 pub fn has_order_book(&self, instrument_id: &InstrumentId) -> bool {
7906 self.books.contains_key(instrument_id)
7907 }
7908
7909 #[must_use]
7911 pub fn has_quote_ticks(&self, instrument_id: &InstrumentId) -> bool {
7912 self.quote_count(instrument_id) > 0
7913 }
7914
7915 #[must_use]
7917 pub fn has_trade_ticks(&self, instrument_id: &InstrumentId) -> bool {
7918 self.trade_count(instrument_id) > 0
7919 }
7920
7921 #[must_use]
7923 pub fn has_mark_prices(&self, instrument_id: &InstrumentId) -> bool {
7924 self.mark_price_count(instrument_id) > 0
7925 }
7926
7927 #[must_use]
7929 pub fn has_index_prices(&self, instrument_id: &InstrumentId) -> bool {
7930 self.index_price_count(instrument_id) > 0
7931 }
7932
7933 #[must_use]
7935 pub fn has_funding_rates(&self, instrument_id: &InstrumentId) -> bool {
7936 self.funding_rate_count(instrument_id) > 0
7937 }
7938
7939 #[must_use]
7941 pub fn has_instrument_statuses(&self, instrument_id: &InstrumentId) -> bool {
7942 self.instrument_status_count(instrument_id) > 0
7943 }
7944
7945 #[must_use]
7947 pub fn has_instrument_close(&self, instrument_id: &InstrumentId) -> bool {
7948 self.instrument_closes.contains_key(instrument_id)
7949 }
7950
7951 #[must_use]
7953 pub fn has_bars(&self, bar_type: &BarType) -> bool {
7954 self.bar_count(bar_type) > 0
7955 }
7956
7957 #[must_use]
7958 pub fn get_xrate(
7959 &self,
7960 venue: Venue,
7961 from_currency: Currency,
7962 to_currency: Currency,
7963 price_type: PriceType,
7964 ) -> Option<Decimal> {
7965 match self.try_get_xrate(venue, from_currency, to_currency, price_type) {
7966 Ok(rate) => rate,
7967 Err(e) => {
7968 log::error!("Failed to calculate xrate: {e}");
7969 None
7970 }
7971 }
7972 }
7973
7974 pub fn try_get_xrate(
7981 &self,
7982 venue: Venue,
7983 from_currency: Currency,
7984 to_currency: Currency,
7985 price_type: PriceType,
7986 ) -> anyhow::Result<Option<Decimal>> {
7987 if from_currency == to_currency {
7988 return Ok(Some(Decimal::ONE));
7991 }
7992
7993 let (bid_quote, ask_quote) = self.build_quote_table(&venue);
7994
7995 get_exchange_rate(
7996 from_currency.code,
7997 to_currency.code,
7998 price_type,
7999 bid_quote,
8000 ask_quote,
8001 )
8002 }
8003
8004 fn build_quote_table(
8005 &self,
8006 venue: &Venue,
8007 ) -> (AHashMap<Ustr, Decimal>, AHashMap<Ustr, Decimal>) {
8008 let mut bid_quotes = AHashMap::new();
8009 let mut ask_quotes = AHashMap::new();
8010 let mut quote_sources = AHashMap::new();
8011
8012 for (instrument_id, instrument) in &self.instruments {
8013 if instrument_id.venue != *venue {
8014 continue;
8015 }
8016
8017 let Some(base_currency) = instrument.base_currency() else {
8018 continue;
8019 };
8020 let pair = Ustr::from(&format!(
8021 "{}/{}",
8022 base_currency.code,
8023 instrument.quote_currency().code
8024 ));
8025
8026 let (bid_price, ask_price) = if let Some(ticks) = self.quotes.get(instrument_id) {
8027 if let Some(tick) = ticks.front() {
8028 (tick.bid_price, tick.ask_price)
8029 } else {
8030 continue; }
8032 } else {
8033 let mut latest_bid: Option<(&BarType, &Bar)> = None;
8037 let mut latest_ask: Option<(&BarType, &Bar)> = None;
8038
8039 for (bar_type, bars) in &self.bars {
8040 if bar_type.instrument_id() != *instrument_id {
8041 continue;
8042 }
8043
8044 let Some(bar) = bars.front() else {
8045 continue;
8046 };
8047
8048 let slot = match bar_type.spec().price_type {
8049 PriceType::Bid => &mut latest_bid,
8050 PriceType::Ask => &mut latest_ask,
8051 _ => continue,
8052 };
8053
8054 if slot.is_none_or(|(current_type, current)| {
8055 (current.ts_init, current_type) < (bar.ts_init, bar_type)
8056 }) {
8057 *slot = Some((bar_type, bar));
8058 }
8059 }
8060
8061 match (latest_bid, latest_ask) {
8062 (Some((_, bid_bar)), Some((_, ask_bar))) => (bid_bar.close, ask_bar.close),
8063 _ => continue,
8064 }
8065 };
8066
8067 let preference = (
8068 bid_price.is_positive() && ask_price.is_positive(),
8069 instrument.instrument_class() == InstrumentClass::Spot,
8070 Reverse(*instrument_id),
8071 );
8072
8073 if quote_sources
8074 .get(&pair)
8075 .is_some_and(|current| current >= &preference)
8076 {
8077 continue;
8078 }
8079
8080 bid_quotes.insert(pair, bid_price.as_decimal());
8081 ask_quotes.insert(pair, ask_price.as_decimal());
8082 quote_sources.insert(pair, preference);
8083 }
8084
8085 (bid_quotes, ask_quotes)
8086 }
8087
8088 #[must_use]
8090 pub fn get_mark_xrate(&self, from_currency: Currency, to_currency: Currency) -> Option<f64> {
8091 self.mark_xrates.get(&(from_currency, to_currency)).copied()
8092 }
8093
8094 pub fn set_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency, xrate: f64) {
8100 assert!(xrate > 0.0, "xrate was zero");
8101 self.mark_xrates.insert((from_currency, to_currency), xrate);
8102 self.mark_xrates
8103 .insert((to_currency, from_currency), 1.0 / xrate);
8104 }
8105
8106 pub fn clear_mark_xrate(&mut self, from_currency: Currency, to_currency: Currency) {
8112 let _ = self.mark_xrates.remove(&(from_currency, to_currency));
8113 }
8114
8115 pub fn clear_mark_xrates(&mut self) {
8117 self.mark_xrates.clear();
8118 }
8119
8120 #[must_use]
8122 pub fn currency(&self, code: &Ustr) -> Option<&Currency> {
8123 self.currencies.get(code)
8124 }
8125
8126 pub fn try_currency(&self, code: &Ustr) -> Result<&Currency, CurrencyLookupError> {
8132 self.currencies
8133 .get(code)
8134 .ok_or_else(|| CurrencyLookupError::not_found(*code))
8135 }
8136
8137 #[must_use]
8141 pub fn instrument(&self, instrument_id: &InstrumentId) -> Option<&InstrumentAny> {
8142 self.instruments.get(instrument_id)
8143 }
8144
8145 pub fn try_instrument(
8151 &self,
8152 instrument_id: &InstrumentId,
8153 ) -> Result<&InstrumentAny, InstrumentLookupError> {
8154 self.instruments
8155 .get(instrument_id)
8156 .ok_or_else(|| InstrumentLookupError::not_found(*instrument_id))
8157 }
8158
8159 #[must_use]
8161 pub fn instrument_ids(&self, venue: Option<&Venue>) -> Vec<&InstrumentId> {
8162 match venue {
8163 Some(v) => self.instruments.keys().filter(|i| &i.venue == v).collect(),
8164 None => self.instruments.keys().collect(),
8165 }
8166 }
8167
8168 #[must_use]
8170 pub fn instruments(&self, venue: &Venue, underlying: Option<&Ustr>) -> Vec<&InstrumentAny> {
8171 self.instruments
8172 .values()
8173 .filter(|i| &i.id().venue == venue)
8174 .filter(|i| underlying.is_none_or(|u| i.underlying() == Some(*u)))
8175 .collect()
8176 }
8177
8178 #[must_use]
8185 pub fn instruments_by_parent(
8186 &self,
8187 venue: &Venue,
8188 root: &Ustr,
8189 class: InstrumentClass,
8190 ) -> Vec<&InstrumentAny> {
8191 self.instruments
8192 .values()
8193 .filter(|i| &i.id().venue == venue)
8194 .filter(|i| i.underlying() == Some(*root))
8195 .filter(|i| i.instrument_class() == class)
8196 .collect()
8197 }
8198
8199 #[must_use]
8201 pub fn bar_types(
8202 &self,
8203 instrument_id: Option<&InstrumentId>,
8204 price_type: Option<&PriceType>,
8205 aggregation_source: AggregationSource,
8206 ) -> Vec<&BarType> {
8207 let mut bar_types = self
8208 .bars
8209 .keys()
8210 .filter(|bar_type| bar_type.aggregation_source() == aggregation_source)
8211 .collect::<Vec<&BarType>>();
8212
8213 if let Some(instrument_id) = instrument_id {
8214 bar_types.retain(|bar_type| bar_type.instrument_id() == *instrument_id);
8215 }
8216
8217 if let Some(price_type) = price_type {
8218 bar_types.retain(|bar_type| &bar_type.spec().price_type == price_type);
8219 }
8220
8221 bar_types
8222 }
8223
8224 #[must_use]
8228 pub fn synthetic(&self, instrument_id: &InstrumentId) -> Option<&SyntheticInstrument> {
8229 self.synthetics.get(instrument_id)
8230 }
8231
8232 pub fn try_synthetic(
8239 &self,
8240 instrument_id: &InstrumentId,
8241 ) -> Result<&SyntheticInstrument, SyntheticInstrumentLookupError> {
8242 self.synthetics
8243 .get(instrument_id)
8244 .ok_or_else(|| SyntheticInstrumentLookupError::not_found(*instrument_id))
8245 }
8246
8247 #[must_use]
8249 pub fn synthetic_ids(&self) -> Vec<&InstrumentId> {
8250 self.synthetics.keys().collect()
8251 }
8252
8253 #[must_use]
8255 pub fn synthetics(&self) -> Vec<&SyntheticInstrument> {
8256 self.synthetics.values().collect()
8257 }
8258
8259 #[must_use]
8263 pub fn account_ref(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
8264 self.accounts
8265 .get(account_id)
8266 .map(|account_cell| AccountRef::new(account_cell.borrow()))
8267 }
8268
8269 #[must_use]
8273 pub fn account(&self, account_id: &AccountId) -> Option<AccountRef<'_>> {
8274 self.account_ref(account_id)
8275 }
8276
8277 pub fn try_account_ref(
8283 &self,
8284 account_id: &AccountId,
8285 ) -> Result<AccountRef<'_>, AccountLookupError> {
8286 self.accounts
8287 .get(account_id)
8288 .map(|account_cell| AccountRef::new(account_cell.borrow()))
8289 .ok_or_else(|| AccountLookupError::not_found(*account_id))
8290 }
8291
8292 pub fn try_account(
8300 &self,
8301 account_id: &AccountId,
8302 ) -> Result<AccountRef<'_>, AccountLookupError> {
8303 self.try_account_ref(account_id)
8304 }
8305
8306 #[must_use]
8316 pub fn account_mut(&mut self, account_id: &AccountId) -> Option<AccountRefMut<'_>> {
8317 self.accounts
8318 .get(account_id)
8319 .map(|account_cell| AccountRefMut::new(account_cell.borrow_mut()))
8320 }
8321
8322 #[must_use]
8328 pub fn account_owned(&self, account_id: &AccountId) -> Option<AccountAny> {
8329 self.accounts.get(account_id).and_then(|account_cell| {
8330 account_cell
8331 .try_borrow()
8332 .ok()
8333 .map(|account| account.clone())
8334 })
8335 }
8336
8337 #[must_use]
8339 pub fn account_for_venue(&self, venue: &Venue) -> Option<AccountRef<'_>> {
8340 self.index
8341 .venue_account
8342 .get(venue)
8343 .and_then(|account_id| self.accounts.get(account_id))
8344 .map(|account_cell| AccountRef::new(account_cell.borrow()))
8345 }
8346
8347 #[must_use]
8352 pub fn account_for_venue_owned(&self, venue: &Venue) -> Option<AccountAny> {
8353 self.index
8354 .venue_account
8355 .get(venue)
8356 .and_then(|account_id| self.accounts.get(account_id))
8357 .map(|account_cell| account_cell.borrow().clone())
8358 }
8359
8360 #[must_use]
8362 pub fn account_id(&self, venue: &Venue) -> Option<&AccountId> {
8363 self.index.venue_account.get(venue)
8364 }
8365
8366 #[must_use]
8372 pub fn accounts(&self, account_id: &AccountId) -> Vec<AccountRef<'_>> {
8373 self.accounts
8374 .values()
8375 .filter(|account_cell| &account_cell.borrow().id() == account_id)
8376 .map(|account_cell| AccountRef::new(account_cell.borrow()))
8377 .collect()
8378 }
8379
8380 #[must_use]
8382 pub fn accounts_all_owned(&self) -> Vec<AccountAny> {
8383 self.accounts
8384 .values()
8385 .map(|account_cell| account_cell.borrow().clone())
8386 .collect()
8387 }
8388
8389 pub fn update_own_order_book(&mut self, order: &OrderAny) {
8397 if !order.has_price() {
8398 return;
8399 }
8400
8401 let instrument_id = order.instrument_id();
8402
8403 if !self.own_books.contains_key(&instrument_id) {
8404 if order.is_closed() {
8405 return;
8406 }
8407
8408 self.own_books
8409 .insert(instrument_id, OwnOrderBook::new(instrument_id));
8410 }
8411
8412 let Some(own_book) = self.own_books.get_mut(&instrument_id) else {
8413 return;
8414 };
8415
8416 let own_book_order = order.to_own_book_order();
8417
8418 if order.is_closed() {
8419 if let Err(e) = own_book.delete(own_book_order) {
8420 log::debug!(
8421 "Failed to delete order {} from own book: {e}",
8422 order.client_order_id(),
8423 );
8424 } else {
8425 log::debug!("Deleted order {} from own book", order.client_order_id());
8426 }
8427 } else {
8428 if let Err(e) = own_book.update(own_book_order) {
8430 log::debug!(
8431 "Failed to update order {} in own book: {e}; inserting instead",
8432 order.client_order_id(),
8433 );
8434 own_book.add(own_book_order);
8435 }
8436 log::debug!("Updated order {} in own book", order.client_order_id());
8437 }
8438 }
8439
8440 pub fn force_remove_from_own_order_book(&mut self, client_order_id: &ClientOrderId) {
8446 let Some(order_cell) = self.orders.get(client_order_id) else {
8447 return;
8448 };
8449 let order = order_cell.borrow();
8450 let instrument_id = order.instrument_id();
8451 let own_book_order = if order.has_price() {
8452 Some(order.to_own_book_order())
8453 } else {
8454 None
8455 };
8456 drop(order);
8457
8458 self.index.orders_open.remove(client_order_id);
8459 self.index.orders_pending_cancel.remove(client_order_id);
8460 self.index.orders_inflight.remove(client_order_id);
8461 self.index.orders_emulated.remove(client_order_id);
8462 self.index.orders_active_local.remove(client_order_id);
8463
8464 if let Some(own_book) = self.own_books.get_mut(&instrument_id)
8465 && let Some(own_book_order) = own_book_order
8466 {
8467 if let Err(e) = own_book.delete(own_book_order) {
8468 log::debug!("Could not force delete {client_order_id} from own book: {e}");
8469 } else {
8470 log::debug!("Force deleted {client_order_id} from own book");
8471 }
8472 }
8473
8474 self.index.orders_closed.insert(*client_order_id);
8475 }
8476
8477 pub fn audit_own_order_books(&mut self) {
8482 log::debug!("Starting own books audit");
8483 let start = std::time::Instant::now();
8484
8485 let valid_order_ids: AHashSet<ClientOrderId> = self
8486 .index
8487 .orders_open
8488 .iter()
8489 .chain(&self.index.orders_inflight)
8490 .chain(&self.index.orders_active_local)
8491 .copied()
8492 .collect();
8493
8494 for own_book in self.own_books.values_mut() {
8495 own_book.audit_open_orders(&valid_order_ids);
8496 }
8497
8498 log::debug!("Completed own books audit in {:?}", start.elapsed());
8499 }
8500}
8501
8502const POSITION_OMS_KEY_PREFIX: &str = "position_oms:";
8503
8504fn position_oms_key(position_id: PositionId) -> String {
8505 format!("{POSITION_OMS_KEY_PREFIX}{position_id}")
8506}