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