1use std::{
19 cell::RefCell,
20 collections::{BinaryHeap, VecDeque},
21 fmt::Debug,
22 rc::Rc,
23};
24
25use ahash::AHashMap;
26use indexmap::IndexMap;
27use nautilus_common::{
28 cache::Cache,
29 clients::ExecutionClient,
30 clock::{Clock, TestClock},
31 messages::execution::{ModifyOrder, TradingCommand},
32 msgbus::{self, MessagingSwitchboard, TypedHandler, switchboard},
33};
34use nautilus_core::{
35 UUID4, UnixNanos,
36 correctness::{CorrectnessResultExt, FAILED, check_equal},
37};
38use nautilus_execution::{
39 matching_core::RestingOrder,
40 matching_engine::{config::OrderMatchingEngineConfig, engine::OrderMatchingEngine},
41 models::{fee::FeeModelHandle, fill::FillModelAny, latency::LatencyModel},
42};
43use nautilus_model::{
44 accounts::{Account, AccountAny, margin_model::MarginModelAny},
45 data::{
46 Bar, Data, FundingRateUpdate, InstrumentClose, InstrumentStatus, OrderBookDelta,
47 OrderBookDeltas, OrderBookDeltas_API, OrderBookDepth10, QuoteTick, TradeTick,
48 },
49 enums::{AccountType, AggressorSide, BookType, OmsType, OrderStatus, PositionAdjustmentType},
50 events::{FundingSettlement, OrderEventAny, OrderUpdated, PositionAdjusted, PositionEvent},
51 identifiers::{AccountId, InstrumentId, Venue},
52 instruments::{Instrument, InstrumentAny},
53 orderbook::OrderBook,
54 orders::{Order, OrderAny},
55 position::Position,
56 types::{AccountBalance, Currency, Money, Price, Quantity},
57};
58use rust_decimal::Decimal;
59use ustr::Ustr;
60
61use crate::{
62 config::SimulatedVenueConfig,
63 modules::{ExchangeContext, SimulationModule},
64};
65
66#[derive(Debug, Eq, PartialEq)]
70struct InflightCommand {
71 timestamp: UnixNanos,
72 counter: u32,
73 command: TradingCommand,
74}
75
76impl InflightCommand {
77 const fn new(timestamp: UnixNanos, counter: u32, command: TradingCommand) -> Self {
78 Self {
79 timestamp,
80 counter,
81 command,
82 }
83 }
84}
85
86impl Ord for InflightCommand {
87 fn cmp(&self, other: &Self) -> std::cmp::Ordering {
88 other
90 .timestamp
91 .cmp(&self.timestamp)
92 .then_with(|| other.counter.cmp(&self.counter))
93 }
94}
95
96impl PartialOrd for InflightCommand {
97 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
98 Some(self.cmp(other))
99 }
100}
101
102#[expect(
118 clippy::struct_excessive_bools,
119 reason = "exchange state mirrors the existing venue configuration flags"
120)]
121pub struct SimulatedExchange {
122 pub id: Venue,
124 pub oms_type: OmsType,
126 pub account_type: AccountType,
128 pub base_currency: Option<Currency>,
130 starting_balances: Vec<Money>,
131 book_type: BookType,
132 default_leverage: Decimal,
133 exec_client: Option<Rc<dyn ExecutionClient>>,
134 fee_model: FeeModelHandle,
135 fill_model: FillModelAny,
136 latency_model: Option<Box<dyn LatencyModel>>,
137 instruments: AHashMap<InstrumentId, InstrumentAny>,
138 matching_engines: AHashMap<InstrumentId, OrderMatchingEngine>,
139 settlement_prices: AHashMap<InstrumentId, Price>,
140 pending_funding_rates: AHashMap<InstrumentId, FundingRateUpdate>,
141 funding_settled_through: AHashMap<InstrumentId, UnixNanos>,
142 leverages: AHashMap<InstrumentId, Decimal>,
143 margin_model: Option<MarginModelAny>,
144 modules: Vec<Box<dyn SimulationModule>>,
145 clock: Rc<RefCell<dyn Clock>>,
146 cache: Rc<RefCell<Cache>>,
147 message_queue: VecDeque<TradingCommand>,
148 inflight_queue: BinaryHeap<InflightCommand>,
149 inflight_counter: AHashMap<UnixNanos, u32>,
150 bar_execution: bool,
151 bar_adaptive_high_low_ordering: bool,
152 trade_execution: bool,
153 liquidity_consumption: bool,
154 reject_stop_orders: bool,
155 support_gtd_orders: bool,
156 support_contingent_orders: bool,
157 use_position_ids: bool,
158 use_random_ids: bool,
159 use_reduce_only: bool,
160 use_message_queue: bool,
161 use_market_order_acks: bool,
162 allow_cash_borrowing: bool,
163 frozen_account: bool,
164 queue_position: bool,
165 oto_full_trigger: bool,
166 price_protection_points: u32,
167 liquidation_enabled: bool,
168 liquidation_trigger_ratio: f64,
169 liquidation_cancel_open_orders: bool,
170}
171
172impl Debug for SimulatedExchange {
173 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
174 f.debug_struct(stringify!(SimulatedExchange))
175 .field("id", &self.id)
176 .field("account_type", &self.account_type)
177 .finish_non_exhaustive()
178 }
179}
180
181impl SimulatedExchange {
182 pub fn new(
190 config: SimulatedVenueConfig,
191 cache: Rc<RefCell<Cache>>,
192 clock: Rc<RefCell<dyn Clock>>,
193 ) -> anyhow::Result<Self> {
194 if config.starting_balances.is_empty() {
195 anyhow::bail!("Starting balances must be provided")
196 }
197
198 if config.base_currency.is_some() && config.starting_balances.len() > 1 {
199 anyhow::bail!("single-currency account has multiple starting currencies")
200 }
201
202 let default_leverage = config.default_leverage.unwrap_or_else(|| {
203 if config.account_type == AccountType::Margin {
204 Decimal::from(10)
205 } else {
206 Decimal::from(1)
207 }
208 });
209
210 Ok(Self {
211 id: config.venue,
212 oms_type: config.oms_type,
213 account_type: config.account_type,
214 base_currency: config.base_currency,
215 starting_balances: config.starting_balances,
216 book_type: config.book_type,
217 default_leverage,
218 exec_client: None,
219 fee_model: config.fee_model.into(),
220 fill_model: config.fill_model,
221 latency_model: config.latency_model,
222 instruments: AHashMap::new(),
223 matching_engines: AHashMap::new(),
224 settlement_prices: config.settlement_prices,
225 pending_funding_rates: AHashMap::new(),
226 funding_settled_through: AHashMap::new(),
227 leverages: config.leverages,
228 margin_model: config.margin_model,
229 modules: config.modules,
230 clock,
231 cache,
232 message_queue: VecDeque::new(),
233 inflight_queue: BinaryHeap::new(),
234 inflight_counter: AHashMap::new(),
235 bar_execution: config.bar_execution,
236 bar_adaptive_high_low_ordering: config.bar_adaptive_high_low_ordering,
237 trade_execution: config.trade_execution,
238 liquidity_consumption: config.liquidity_consumption,
239 reject_stop_orders: config.reject_stop_orders,
240 support_gtd_orders: config.support_gtd_orders,
241 support_contingent_orders: config.support_contingent_orders,
242 use_position_ids: config.use_position_ids,
243 use_random_ids: config.use_random_ids,
244 use_reduce_only: config.use_reduce_only,
245 use_message_queue: config.use_message_queue,
246 use_market_order_acks: config.use_market_order_acks,
247 allow_cash_borrowing: config.allow_cash_borrowing,
248 frozen_account: config.frozen_account,
249 queue_position: config.queue_position,
250 oto_full_trigger: config.oto_full_trigger,
251 price_protection_points: config.price_protection_points,
252 liquidation_enabled: config.liquidation_enabled,
253 liquidation_trigger_ratio: config.liquidation_trigger_ratio,
254 liquidation_cancel_open_orders: config.liquidation_cancel_open_orders,
255 })
256 }
257
258 pub fn register_client(&mut self, client: Rc<dyn ExecutionClient>) {
260 self.exec_client = Some(client);
261 }
262
263 pub fn register_spread_quote_endpoint(exchange: &Rc<RefCell<Self>>) {
265 let venue = exchange.borrow().id;
266 let endpoint = format!("SimulatedExchange.process_new_quote.{venue}");
267 let handler_id = endpoint.clone();
268 let exchange = Rc::clone(exchange);
269 let handler = TypedHandler::from_with_id(handler_id, move |quote: &QuoteTick| {
270 exchange.borrow_mut().process_quote_tick(quote);
271 });
272
273 msgbus::register_quote_endpoint(endpoint.into(), handler);
274 }
275
276 pub fn set_fill_model(&mut self, fill_model: FillModelAny) {
278 for matching_engine in self.matching_engines.values_mut() {
279 matching_engine.set_fill_model(fill_model.clone().into());
280 log::info!(
281 "Setting fill model for {} to {}",
282 matching_engine.venue,
283 self.fill_model
284 );
285 }
286 self.fill_model = fill_model;
287 }
288
289 pub fn set_latency_model(&mut self, latency_model: Box<dyn LatencyModel>) {
291 self.latency_model = Some(latency_model);
292 }
293
294 pub fn set_settlement_price(&mut self, instrument_id: InstrumentId, price: Price) {
296 self.settlement_prices.insert(instrument_id, price);
297 }
298
299 #[must_use]
301 pub const fn book_type(&self) -> BookType {
302 self.book_type
303 }
304
305 pub fn instrument_ids(&self) -> impl Iterator<Item = &InstrumentId> {
307 self.instruments.keys()
308 }
309
310 #[must_use]
312 pub fn instrument_expiration(&self, instrument_id: InstrumentId) -> Option<UnixNanos> {
313 self.matching_engines
314 .get(&instrument_id)
315 .and_then(|matching_engine| matching_engine.instrument.expiration_ns())
316 }
317
318 #[must_use]
320 pub fn has_unprocessed_instrument_expiration(&self, expiration_ns: UnixNanos) -> bool {
321 self.matching_engines.values().any(|matching_engine| {
322 !matching_engine.is_expiration_processed()
323 && matching_engine.instrument.expiration_ns() == Some(expiration_ns)
324 })
325 }
326
327 pub fn initialize_account(&mut self) {
328 self.generate_fresh_account_state();
329 }
330
331 pub fn load_open_orders(&mut self) {
333 let mut open_orders: Vec<(OrderAny, AccountId)> = {
334 let cache = self.cache.as_ref().borrow();
335 cache
336 .orders_open(Some(&self.id), None, None, None, None)
337 .into_iter()
338 .filter(|order| !order.is_emulated())
339 .filter_map(|order| {
340 order
341 .account_id()
342 .map(|account_id| (order.clone(), account_id))
343 })
344 .collect()
345 };
346
347 open_orders.sort_by(|(a, _), (b, _)| {
349 a.ts_init()
350 .cmp(&b.ts_init())
351 .then_with(|| a.client_order_id().cmp(&b.client_order_id()))
352 });
353
354 for (mut order, account_id) in open_orders {
355 let instrument_id = order.instrument_id();
356 if let Some(matching_engine) = self.matching_engines.get_mut(&instrument_id) {
357 matching_engine.process_order(&mut order, account_id);
358 } else {
359 log::error!(
360 "No matching engine for {instrument_id} to load open order {}",
361 order.client_order_id()
362 );
363 }
364 }
365 }
366
367 pub fn add_instrument(&mut self, instrument: InstrumentAny) -> anyhow::Result<()> {
378 check_equal(
379 &instrument.id().venue,
380 &self.id,
381 "Venue of instrument id",
382 "Venue of simulated exchange",
383 )
384 .expect_display(FAILED);
385
386 if self.account_type == AccountType::Cash
387 && (matches!(instrument, InstrumentAny::CryptoPerpetual(_))
388 || matches!(instrument, InstrumentAny::CryptoFuture(_))
389 || matches!(instrument, InstrumentAny::PerpetualContract(_)))
390 {
391 anyhow::bail!("Cash account cannot trade futures or perpetuals")
392 }
393
394 self.instruments.insert(instrument.id(), instrument.clone());
395
396 let price_protection = if self.price_protection_points == 0 {
397 None
398 } else {
399 Some(self.price_protection_points)
400 };
401
402 let matching_engine_config = OrderMatchingEngineConfig::builder()
403 .bar_execution(self.bar_execution)
404 .bar_adaptive_high_low_ordering(self.bar_adaptive_high_low_ordering)
405 .trade_execution(self.trade_execution)
406 .liquidity_consumption(self.liquidity_consumption)
407 .reject_stop_orders(self.reject_stop_orders)
408 .support_gtd_orders(self.support_gtd_orders)
409 .support_contingent_orders(self.support_contingent_orders)
410 .use_position_ids(self.use_position_ids)
411 .use_random_ids(self.use_random_ids)
412 .use_reduce_only(self.use_reduce_only)
413 .use_market_order_acks(self.use_market_order_acks)
414 .queue_position(self.queue_position)
415 .oto_full_trigger(self.oto_full_trigger)
416 .maybe_price_protection_points(price_protection)
417 .build();
418 let instrument_id = instrument.id();
419 let raw_id = u32::try_from(self.instruments.len())
420 .map_err(|e| anyhow::anyhow!("number of instruments exceeds u32::MAX: {e}"))?;
421 let matching_engine = OrderMatchingEngine::new(
422 instrument,
423 raw_id,
424 self.fill_model.clone().into(),
425 self.fee_model.clone(),
426 self.book_type,
427 self.oms_type,
428 self.account_type,
429 self.clock.clone(),
430 Rc::clone(&self.cache),
431 matching_engine_config,
432 );
433 self.matching_engines.insert(instrument_id, matching_engine);
434
435 log::info!("Added instrument {instrument_id} and created matching engine");
436 Ok(())
437 }
438
439 #[must_use]
441 pub fn best_bid_price(&self, instrument_id: InstrumentId) -> Option<Price> {
442 self.matching_engines
443 .get(&instrument_id)
444 .and_then(OrderMatchingEngine::best_bid_price)
445 }
446
447 #[must_use]
449 pub fn best_ask_price(&self, instrument_id: InstrumentId) -> Option<Price> {
450 self.matching_engines
451 .get(&instrument_id)
452 .and_then(OrderMatchingEngine::best_ask_price)
453 }
454
455 pub fn get_book(&self, instrument_id: InstrumentId) -> Option<&OrderBook> {
457 self.matching_engines
458 .get(&instrument_id)
459 .map(OrderMatchingEngine::get_book)
460 }
461
462 #[must_use]
464 pub fn get_matching_engine(
465 &self,
466 instrument_id: &InstrumentId,
467 ) -> Option<&OrderMatchingEngine> {
468 self.matching_engines.get(instrument_id)
469 }
470
471 #[must_use]
473 pub const fn get_matching_engines(&self) -> &AHashMap<InstrumentId, OrderMatchingEngine> {
474 &self.matching_engines
475 }
476
477 #[must_use]
479 pub fn get_books(&self) -> AHashMap<InstrumentId, OrderBook> {
480 let mut books = AHashMap::new();
481 for (instrument_id, matching_engine) in &self.matching_engines {
482 books.insert(*instrument_id, matching_engine.get_book().clone());
483 }
484 books
485 }
486
487 #[must_use]
489 pub fn get_open_orders(&self, instrument_id: Option<InstrumentId>) -> Vec<RestingOrder> {
490 instrument_id
491 .and_then(|id| {
492 self.matching_engines
493 .get(&id)
494 .map(OrderMatchingEngine::get_open_orders)
495 })
496 .unwrap_or_else(|| {
497 self.matching_engines
498 .values()
499 .flat_map(OrderMatchingEngine::get_open_orders)
500 .collect()
501 })
502 }
503
504 #[must_use]
506 pub fn get_open_bid_orders(&self, instrument_id: Option<InstrumentId>) -> Vec<RestingOrder> {
507 instrument_id
508 .and_then(|id| {
509 self.matching_engines
510 .get(&id)
511 .map(OrderMatchingEngine::get_open_bid_orders)
512 })
513 .unwrap_or_else(|| {
514 self.matching_engines
515 .values()
516 .flat_map(OrderMatchingEngine::get_open_bid_orders)
517 .collect()
518 })
519 }
520
521 #[must_use]
523 pub fn get_open_ask_orders(&self, instrument_id: Option<InstrumentId>) -> Vec<RestingOrder> {
524 instrument_id
525 .and_then(|id| {
526 self.matching_engines
527 .get(&id)
528 .map(OrderMatchingEngine::get_open_ask_orders)
529 })
530 .unwrap_or_else(|| {
531 self.matching_engines
532 .values()
533 .flat_map(OrderMatchingEngine::get_open_ask_orders)
534 .collect()
535 })
536 }
537
538 #[must_use]
540 pub fn get_account(&self) -> Option<AccountAny> {
541 self.exec_client
542 .as_ref()
543 .and_then(|client| client.get_account())
544 }
545
546 #[must_use]
548 pub fn cache(&self) -> &Rc<RefCell<Cache>> {
549 &self.cache
550 }
551
552 pub fn adjust_account(&mut self, adjustment: Money) {
558 if self.frozen_account {
559 return;
561 }
562
563 if let Some(exec_client) = &self.exec_client {
564 let venue = exec_client.venue();
565 log::debug!("Adjusting account for venue {venue}");
566 let account_state = {
567 let cache = self.cache.borrow();
568 if let Some(account) = cache.account_for_venue(&venue) {
569 if let Some(balance) = account.balance(Some(adjustment.currency)) {
570 let mut current_balance = *balance;
571 current_balance.total = current_balance.total + adjustment;
572 current_balance.free = current_balance.free + adjustment;
573
574 let margins = match &*account {
575 AccountAny::Margin(margin_account) => margin_account.margins.clone(),
576 _ => IndexMap::new(),
577 };
578
579 Some((
580 vec![current_balance],
581 margins.values().copied().collect(),
582 self.clock.borrow().timestamp_ns(),
583 ))
584 } else {
585 log::error!(
586 "Cannot adjust account: no balance for currency {}",
587 adjustment.currency
588 );
589 None
590 }
591 } else {
592 log::error!("Cannot adjust account: no account for venue {venue}");
593 None
594 }
595 };
596
597 if let Some((balances, margins, ts_event)) = account_state {
598 exec_client
599 .generate_account_state(balances, margins, true, ts_event)
600 .unwrap();
601 }
602 }
603 }
604
605 #[must_use]
607 pub fn has_pending_commands(&self, ts_now: UnixNanos) -> bool {
608 if !self.message_queue.is_empty() {
609 return true;
610 }
611 self.inflight_queue
612 .peek()
613 .is_some_and(|inflight| inflight.timestamp <= ts_now)
614 }
615
616 #[must_use]
623 pub fn max_inflight_command_ts(&self) -> Option<UnixNanos> {
624 self.inflight_queue.iter().map(|c| c.timestamp).max()
625 }
626
627 pub fn iterate_matching_engines(&mut self, ts_now: UnixNanos) {
630 for matching_engine in self.matching_engines.values_mut() {
631 matching_engine.iterate(ts_now, AggressorSide::NoAggressor);
632 }
633 }
634
635 pub fn process_instrument_expirations(&mut self, ts_now: UnixNanos) {
637 for matching_engine in self.matching_engines.values_mut() {
638 if matching_engine
639 .instrument
640 .expiration_ns()
641 .is_some_and(|expiration_ns| ts_now >= expiration_ns)
642 {
643 matching_engine.process_instrument_expiration(ts_now);
644 }
645 }
646 }
647
648 #[must_use]
650 pub fn instrument_expirations(&self) -> Vec<(InstrumentId, UnixNanos)> {
651 self.matching_engines
652 .values()
653 .filter(|matching_engine| !matching_engine.is_expiration_processed())
654 .filter_map(|matching_engine| {
655 matching_engine
656 .instrument
657 .expiration_ns()
658 .filter(|expiration_ns| *expiration_ns > UnixNanos::default())
659 .map(|expiration_ns| (matching_engine.instrument.id(), expiration_ns))
660 })
661 .collect()
662 }
663
664 pub fn set_clock_time(&self, ts_now: UnixNanos) {
672 let mut clock_ref = self.clock.borrow_mut();
673 let test_clock = clock_ref
674 .as_any_mut()
675 .downcast_mut::<TestClock>()
676 .expect("SimulatedExchange requires TestClock");
677 test_clock.set_time(ts_now);
678 }
679
680 pub fn send(&mut self, command: TradingCommand) {
682 if !self.use_message_queue {
683 self.process_trading_command(command);
684 } else if self.latency_model.is_none() {
685 self.message_queue.push_back(command);
686 } else {
687 let (timestamp, counter) = self.generate_inflight_command(&command);
688 self.inflight_queue
689 .push(InflightCommand::new(timestamp, counter, command));
690 }
691 }
692
693 fn generate_inflight_command(&mut self, command: &TradingCommand) -> (UnixNanos, u32) {
694 if let Some(latency_model) = &self.latency_model {
695 let ts = match command {
696 TradingCommand::SubmitOrder(_) | TradingCommand::SubmitOrderList(_) => {
697 command.ts_init() + latency_model.get_insert_latency()
698 }
699 TradingCommand::ModifyOrder(_) | TradingCommand::ModifyOrders(_) => {
700 command.ts_init() + latency_model.get_update_latency()
701 }
702 TradingCommand::CancelOrder(_)
703 | TradingCommand::CancelOrders(_)
704 | TradingCommand::CancelAllOrders(_) => {
705 command.ts_init() + latency_model.get_delete_latency()
706 }
707 _ => panic!("Cannot handle command: {command:?}"),
708 };
709
710 let counter = self
711 .inflight_counter
712 .entry(ts)
713 .and_modify(|e| *e += 1)
714 .or_insert(1);
715
716 (ts, *counter)
717 } else {
718 panic!("Latency model should be initialized");
719 }
720 }
721
722 pub fn process_order_book_delta(&mut self, delta: OrderBookDelta) {
728 for module in &self.modules {
729 module.pre_process(&Data::Delta(delta));
730 }
731
732 if !self.matching_engines.contains_key(&delta.instrument_id) {
733 let instrument = {
734 let cache = self.cache.as_ref().borrow();
735 cache.instrument(&delta.instrument_id).cloned()
736 };
737
738 if let Some(instrument) = instrument {
739 self.add_instrument(instrument).unwrap();
740 } else {
741 panic!(
742 "No matching engine found for instrument {}",
743 delta.instrument_id
744 );
745 }
746 }
747
748 if let Some(matching_engine) = self.matching_engines.get_mut(&delta.instrument_id) {
749 matching_engine.process_order_book_delta(&delta).unwrap();
750 } else {
751 panic!("Matching engine should be initialized");
752 }
753 }
754
755 pub fn process_order_book_deltas(&mut self, deltas: &OrderBookDeltas) {
761 for module in &self.modules {
762 module.pre_process(&Data::Deltas(OrderBookDeltas_API::new(deltas.clone())));
763 }
764
765 if !self.matching_engines.contains_key(&deltas.instrument_id) {
766 let instrument = {
767 let cache = self.cache.as_ref().borrow();
768 cache.instrument(&deltas.instrument_id).cloned()
769 };
770
771 if let Some(instrument) = instrument {
772 self.add_instrument(instrument).unwrap();
773 } else {
774 panic!(
775 "No matching engine found for instrument {}",
776 deltas.instrument_id
777 );
778 }
779 }
780
781 if let Some(matching_engine) = self.matching_engines.get_mut(&deltas.instrument_id) {
782 matching_engine.process_order_book_deltas(deltas).unwrap();
783 } else {
784 panic!("Matching engine should be initialized");
785 }
786 }
787
788 pub fn process_order_book_depth10(&mut self, depth: &OrderBookDepth10) {
794 for module in &self.modules {
795 module.pre_process(&Data::Depth10(Box::new(*depth)));
796 }
797
798 if !self.matching_engines.contains_key(&depth.instrument_id) {
799 let instrument = {
800 let cache = self.cache.as_ref().borrow();
801 cache.instrument(&depth.instrument_id).cloned()
802 };
803
804 if let Some(instrument) = instrument {
805 self.add_instrument(instrument).unwrap();
806 } else {
807 panic!(
808 "No matching engine found for instrument {}",
809 depth.instrument_id
810 );
811 }
812 }
813
814 if let Some(matching_engine) = self.matching_engines.get_mut(&depth.instrument_id) {
815 matching_engine.process_order_book_depth10(depth).unwrap();
816 } else {
817 panic!("Matching engine should be initialized");
818 }
819 }
820
821 pub fn process_quote_tick(&mut self, quote: &QuoteTick) {
827 for module in &self.modules {
828 module.pre_process(&Data::Quote(*quote));
829 }
830
831 if !self.matching_engines.contains_key("e.instrument_id) {
832 let instrument = {
833 let cache = self.cache.as_ref().borrow();
834 cache.instrument("e.instrument_id).cloned()
835 };
836
837 if let Some(instrument) = instrument {
838 self.add_instrument(instrument).unwrap();
839 } else {
840 panic!(
841 "No matching engine found for instrument {}",
842 quote.instrument_id
843 );
844 }
845 }
846
847 if let Some(matching_engine) = self.matching_engines.get_mut("e.instrument_id) {
848 matching_engine.process_quote_tick(quote);
849 } else {
850 panic!("Matching engine should be initialized");
851 }
852 }
853
854 pub fn process_trade_tick(&mut self, trade: &TradeTick) {
860 for module in &self.modules {
861 module.pre_process(&Data::Trade(*trade));
862 }
863
864 if !self.matching_engines.contains_key(&trade.instrument_id) {
865 let instrument = {
866 let cache = self.cache.as_ref().borrow();
867 cache.instrument(&trade.instrument_id).cloned()
868 };
869
870 if let Some(instrument) = instrument {
871 self.add_instrument(instrument).unwrap();
872 } else {
873 panic!(
874 "No matching engine found for instrument {}",
875 trade.instrument_id
876 );
877 }
878 }
879
880 if let Some(matching_engine) = self.matching_engines.get_mut(&trade.instrument_id) {
881 matching_engine.process_trade_tick(trade);
882 } else {
883 panic!("Matching engine should be initialized");
884 }
885 }
886
887 pub fn process_bar(&mut self, bar: Bar) {
893 for module in &self.modules {
894 module.pre_process(&Data::Bar(bar));
895 }
896
897 if !self.matching_engines.contains_key(&bar.instrument_id()) {
898 let instrument = {
899 let cache = self.cache.as_ref().borrow();
900 cache.instrument(&bar.instrument_id()).cloned()
901 };
902
903 if let Some(instrument) = instrument {
904 self.add_instrument(instrument).unwrap();
905 } else {
906 panic!(
907 "No matching engine found for instrument {}",
908 bar.instrument_id()
909 );
910 }
911 }
912
913 if let Some(matching_engine) = self.matching_engines.get_mut(&bar.instrument_id()) {
914 matching_engine.process_bar(&bar);
915 } else {
916 panic!("Matching engine should be initialized");
917 }
918 }
919
920 pub fn process_instrument_status(&mut self, status: InstrumentStatus) {
926 for module in &self.modules {
927 module.pre_process(&Data::InstrumentStatus(status));
928 }
929
930 if !self.matching_engines.contains_key(&status.instrument_id) {
931 let instrument = {
932 let cache = self.cache.as_ref().borrow();
933 cache.instrument(&status.instrument_id).cloned()
934 };
935
936 if let Some(instrument) = instrument {
937 self.add_instrument(instrument).unwrap();
938 } else {
939 panic!(
940 "No matching engine found for instrument {}",
941 status.instrument_id
942 );
943 }
944 }
945
946 if let Some(matching_engine) = self.matching_engines.get_mut(&status.instrument_id) {
947 matching_engine.process_status(status.action);
948 } else {
949 panic!("Matching engine should be initialized");
950 }
951 }
952
953 pub fn process_instrument_close(&mut self, close: InstrumentClose) {
959 for module in &self.modules {
960 module.pre_process(&Data::InstrumentClose(close));
961 }
962
963 if !self.matching_engines.contains_key(&close.instrument_id) {
964 let instrument = {
965 let cache = self.cache.as_ref().borrow();
966 cache.instrument(&close.instrument_id).cloned()
967 };
968
969 if let Some(instrument) = instrument {
970 self.add_instrument(instrument).unwrap();
971 } else {
972 panic!(
973 "No matching engine found for instrument {}",
974 close.instrument_id
975 );
976 }
977 }
978
979 if let Some(matching_engine) = self.matching_engines.get_mut(&close.instrument_id) {
980 if let Some(price) = self.settlement_prices.get(&close.instrument_id) {
981 matching_engine.set_settlement_price(*price);
982 }
983 matching_engine.process_instrument_close(close);
984 } else {
985 panic!("Matching engine should be initialized");
986 }
987 }
988
989 pub fn process_funding_rate(&mut self, funding_rate: FundingRateUpdate) -> Option<UnixNanos> {
993 for module in &self.modules {
994 module.pre_process(&Data::FundingRateUpdate(funding_rate));
995 }
996
997 if let Some(next_funding_ns) = funding_rate.next_funding_ns {
998 if next_funding_ns <= self.clock.borrow().timestamp_ns() {
999 self.pending_funding_rates
1000 .remove(&funding_rate.instrument_id);
1001 self.settle_funding_rate(funding_rate, next_funding_ns);
1002 return None;
1003 }
1004
1005 self.pending_funding_rates
1006 .insert(funding_rate.instrument_id, funding_rate);
1007 return Some(next_funding_ns);
1008 }
1009
1010 if Self::is_interval_funding_boundary(&funding_rate) {
1011 self.settle_funding_rate(funding_rate, funding_rate.ts_event);
1012 } else {
1013 log::debug!(
1014 "Funding rate update for {} does not define a settlement boundary",
1015 funding_rate.instrument_id
1016 );
1017 }
1018
1019 None
1020 }
1021
1022 pub fn process_funding_settlement(&mut self, instrument_id: InstrumentId, ts_event: UnixNanos) {
1024 let Some(funding_rate) = self.pending_funding_rates.remove(&instrument_id) else {
1025 return;
1026 };
1027
1028 if funding_rate
1029 .next_funding_ns
1030 .is_some_and(|next_funding_ns| next_funding_ns > ts_event)
1031 {
1032 self.pending_funding_rates
1033 .insert(funding_rate.instrument_id, funding_rate);
1034 return;
1035 }
1036
1037 self.settle_funding_rate(funding_rate, ts_event);
1038 }
1039
1040 fn settle_funding_rate(&mut self, funding_rate: FundingRateUpdate, ts_event: UnixNanos) {
1041 if self
1042 .funding_settled_through
1043 .get(&funding_rate.instrument_id)
1044 .is_some_and(|settled_through| *settled_through >= ts_event)
1045 {
1046 return;
1047 }
1048
1049 let Some(exec_client) = &self.exec_client else {
1050 log::warn!(
1051 "Cannot settle funding for {}: execution client is not registered",
1052 funding_rate.instrument_id
1053 );
1054 return;
1055 };
1056 let account_id = exec_client.account_id();
1057
1058 if !self
1059 .matching_engines
1060 .contains_key(&funding_rate.instrument_id)
1061 {
1062 let instrument = {
1063 let cache = self.cache.as_ref().borrow();
1064 cache.instrument(&funding_rate.instrument_id).cloned()
1065 };
1066
1067 if let Some(instrument) = instrument {
1068 if let Err(e) = self.add_instrument(instrument) {
1069 log::error!(
1070 "Cannot settle funding for {}: failed to add instrument: {e}",
1071 funding_rate.instrument_id
1072 );
1073 return;
1074 }
1075 } else {
1076 log::warn!(
1077 "Cannot settle funding for {}: no matching engine or instrument",
1078 funding_rate.instrument_id
1079 );
1080 return;
1081 }
1082 }
1083
1084 let Some(settlement_price) = self.funding_settlement_price(funding_rate.instrument_id)
1085 else {
1086 log::warn!(
1087 "Cannot settle funding for {}: no mark price or top-of-book price",
1088 funding_rate.instrument_id
1089 );
1090 return;
1091 };
1092
1093 let open_positions: Vec<Position> = {
1094 let cache = self.cache.borrow();
1095 cache
1096 .positions_open(
1097 Some(&self.id),
1098 Some(&funding_rate.instrument_id),
1099 None,
1100 Some(&account_id),
1101 None,
1102 )
1103 .into_iter()
1104 .map(|position| position.cloned())
1105 .collect()
1106 };
1107
1108 self.funding_settled_through
1109 .insert(funding_rate.instrument_id, ts_event);
1110
1111 if open_positions.is_empty() {
1112 return;
1113 }
1114
1115 let currency = open_positions[0].settlement_currency;
1116 let ts_init = self.clock.borrow().timestamp_ns();
1117 let settlement = FundingSettlement::new(
1118 msgbus::get_message_bus().borrow().trader_id,
1119 funding_rate.instrument_id,
1120 account_id,
1121 funding_rate.rate,
1122 settlement_price,
1123 currency,
1124 UUID4::new(),
1125 ts_event,
1126 ts_init,
1127 );
1128 let settlement_topic = switchboard::get_funding_settlement_topic(settlement.instrument_id);
1129 msgbus::publish_any(settlement_topic, &settlement);
1130
1131 let mut account_adjustments: AHashMap<Currency, Decimal> = AHashMap::new();
1132 let mut position_events = Vec::new();
1133
1134 {
1135 let mut cache = self.cache.borrow_mut();
1136
1137 for mut position in open_positions {
1138 let notional = position.notional_value(settlement_price);
1139 let side = if position.signed_qty > 0.0 {
1140 -Decimal::ONE
1141 } else {
1142 Decimal::ONE
1143 };
1144 let amount = notional.as_decimal() * funding_rate.rate * side;
1145
1146 let pnl_change = match Money::from_decimal(amount, notional.currency) {
1147 Ok(money) => money,
1148 Err(e) => {
1149 log::error!(
1150 "Cannot settle funding for position {}: invalid funding amount: {e}",
1151 position.id
1152 );
1153 continue;
1154 }
1155 };
1156
1157 let adjustment = PositionAdjusted::new(
1158 settlement.trader_id,
1159 position.strategy_id,
1160 position.instrument_id,
1161 position.id,
1162 position.account_id,
1163 PositionAdjustmentType::Funding,
1164 None,
1165 Some(pnl_change),
1166 Some(Ustr::from(&format!(
1167 "funding_settlement:{}",
1168 settlement.event_id
1169 ))),
1170 UUID4::new(),
1171 settlement.ts_event,
1172 settlement.ts_init,
1173 );
1174 position.apply_adjustment(adjustment);
1175
1176 if let Err(e) = cache.update_position(&position) {
1177 log::error!(
1178 "Cannot update position {} after funding settlement: {e}",
1179 position.id
1180 );
1181 continue;
1182 }
1183
1184 account_adjustments
1185 .entry(pnl_change.currency)
1186 .and_modify(|current| *current += pnl_change.as_decimal())
1187 .or_insert_with(|| pnl_change.as_decimal());
1188 position_events.push(PositionEvent::PositionAdjusted(adjustment));
1189 }
1190 }
1191
1192 for (currency, amount) in account_adjustments {
1193 match Money::from_decimal(amount, currency) {
1194 Ok(adjustment) => self.adjust_account(adjustment),
1195 Err(e) => log::error!("Cannot apply funding account adjustment: {e}"),
1196 }
1197 }
1198
1199 for event in position_events {
1200 let PositionEvent::PositionAdjusted(adjustment) = &event else {
1201 continue;
1202 };
1203 let topic = switchboard::get_event_position_topic(adjustment.strategy_id);
1204 msgbus::publish_position_event(topic, &event);
1205 }
1206 }
1207
1208 fn funding_settlement_price(&self, instrument_id: InstrumentId) -> Option<Price> {
1209 if let Some(mark_price) = self.cache.borrow().mark_price(&instrument_id) {
1210 return Some(mark_price.value);
1211 }
1212
1213 let bid = self.best_bid_price(instrument_id)?;
1214 let ask = self.best_ask_price(instrument_id)?;
1215 let midpoint = (bid.as_decimal() + ask.as_decimal()) / Decimal::from(2);
1216 Price::from_decimal_dp(midpoint, bid.precision.max(ask.precision)).ok()
1217 }
1218
1219 fn is_interval_funding_boundary(funding_rate: &FundingRateUpdate) -> bool {
1220 let Some(interval_mins) = funding_rate.interval else {
1221 return false;
1222 };
1223 let interval_ns = u64::from(interval_mins) * 60 * 1_000_000_000;
1224 interval_ns > 0 && funding_rate.ts_event.as_u64().is_multiple_of(interval_ns)
1225 }
1226
1227 pub fn process(&mut self, ts_now: UnixNanos) {
1235 self.set_clock_time(ts_now);
1236
1237 while let Some(inflight) = self.inflight_queue.peek() {
1238 if inflight.timestamp > ts_now {
1239 break;
1240 }
1241 let inflight = self.inflight_queue.pop().unwrap();
1242 let timestamp = inflight.timestamp;
1243 self.message_queue.push_back(inflight.command);
1244 self.inflight_counter.remove(×tamp);
1245 }
1246
1247 while let Some(command) = self.message_queue.pop_front() {
1248 self.process_trading_command(command);
1249 }
1250 }
1251
1252 pub fn process_modules(&mut self, ts_now: UnixNanos) {
1257 let adjustments = {
1258 let cache = self.cache.borrow();
1259 let ctx = ExchangeContext {
1260 venue: self.id,
1261 base_currency: self.base_currency,
1262 instruments: &self.instruments,
1263 matching_engines: &self.matching_engines,
1264 cache: &cache,
1265 };
1266 self.modules
1267 .iter()
1268 .flat_map(|m| m.process(ts_now, &ctx))
1269 .collect::<Vec<Money>>()
1270 };
1271
1272 for adjustment in adjustments {
1273 self.adjust_account(adjustment);
1274 }
1275 }
1276
1277 pub fn reset(&mut self) {
1279 if !self.account_at_starting_balances() {
1280 self.generate_fresh_account_state();
1281 }
1282
1283 for module in &self.modules {
1284 module.reset();
1285 }
1286
1287 for matching_engine in self.matching_engines.values_mut() {
1288 matching_engine.reset();
1289 }
1290
1291 self.pending_funding_rates.clear();
1292 self.funding_settled_through.clear();
1293 self.message_queue.clear();
1294 self.inflight_queue.clear();
1295
1296 log::info!("Resetting exchange state");
1297 }
1298
1299 pub fn log_diagnostics(&self) {
1301 for module in &self.modules {
1302 module.log_diagnostics();
1303 }
1304 }
1305
1306 pub fn process_liquidations(&mut self, ts_now: UnixNanos) {
1317 if !self.liquidation_enabled {
1318 return;
1319 }
1320
1321 if self.frozen_account {
1322 return;
1323 }
1324
1325 let account = {
1326 let cache = self.cache.borrow();
1327 cache.account_for_venue_owned(&self.id)
1328 };
1329 let Some(account) = account else { return };
1330 let AccountAny::Margin(margin_account) = &account else {
1331 return;
1332 };
1333 let account_id = margin_account.id();
1334
1335 let currencies: Vec<Currency> = margin_account.currencies();
1336
1337 let open_positions: Vec<Position> = {
1338 let cache = self.cache.borrow();
1339 cache
1340 .positions_open(Some(&self.id), None, None, None, None)
1341 .into_iter()
1342 .map(|p| p.cloned())
1343 .collect()
1344 };
1345
1346 let mut positions_by_currency: AHashMap<Currency, Vec<usize>> = AHashMap::new();
1348 for (i, p) in open_positions.iter().enumerate() {
1349 positions_by_currency
1350 .entry(p.settlement_currency)
1351 .or_default()
1352 .push(i);
1353 }
1354
1355 for currency in currencies {
1356 let Some(balance) = margin_account.balance(Some(currency)) else {
1357 continue;
1358 };
1359 let balance_f64 = balance.total.as_f64();
1360
1361 let Some(indices) = positions_by_currency.get(¤cy) else {
1362 continue;
1363 };
1364
1365 let (upnl_f64, all_priced) = {
1366 let cache = self.cache.borrow();
1367 let mut upnl = 0.0_f64;
1368 let mut all_priced = true;
1369
1370 for &i in indices {
1371 let p = &open_positions[i];
1372 if let Some(pnl) = cache.calculate_unrealized_pnl(p) {
1373 upnl += pnl.as_f64();
1374 } else {
1375 all_priced = false;
1376 break;
1377 }
1378 }
1379 (upnl, all_priced)
1380 };
1381
1382 if !all_priced {
1383 continue; }
1385
1386 let equity = balance_f64 + upnl_f64;
1387 let maintenance = margin_account.total_maintenance_margin(currency).as_f64();
1388
1389 if maintenance == 0.0 {
1390 continue;
1391 }
1392
1393 let threshold = maintenance * self.liquidation_trigger_ratio;
1394
1395 if equity > threshold {
1396 continue;
1397 }
1398
1399 log::warn!(
1400 "LIQUIDATION triggered for account {} currency {}: equity={:.4} <= threshold={:.4} (maintenance={:.4} x ratio={})",
1401 account_id,
1402 currency,
1403 equity,
1404 threshold,
1405 maintenance,
1406 self.liquidation_trigger_ratio
1407 );
1408
1409 for matching_engine in self.matching_engines.values_mut() {
1410 matching_engine.liquidate_open_positions(
1411 ts_now,
1412 self.liquidation_cancel_open_orders,
1413 currency,
1414 );
1415 }
1416
1417 break;
1418 }
1419 }
1420
1421 fn process_trading_command(&mut self, command: TradingCommand) {
1422 let instrument_id = command.instrument_id();
1423 assert!(
1424 self.matching_engines.contains_key(&instrument_id),
1425 "Matching engine not found for instrument {instrument_id}",
1426 );
1427
1428 let command = match command {
1429 TradingCommand::ModifyOrder(ref command)
1430 if self.process_modify_submitted_order(command) =>
1431 {
1432 return;
1433 }
1434 TradingCommand::ModifyOrders(mut command) => {
1435 command
1436 .modifies
1437 .retain(|modify| !self.process_modify_submitted_order(modify));
1438
1439 if command.modifies.is_empty() {
1440 return;
1441 }
1442 TradingCommand::ModifyOrders(command)
1443 }
1444 command => command,
1445 };
1446
1447 if let Some(matching_engine) = self.matching_engines.get_mut(&instrument_id) {
1448 let account_id = if let Some(exec_client) = &self.exec_client {
1449 exec_client.account_id()
1450 } else {
1451 panic!("Execution client should be initialized");
1452 };
1453
1454 match command {
1455 TradingCommand::SubmitOrder(command) => {
1456 let mut order = self
1457 .cache
1458 .borrow()
1459 .order(&command.client_order_id)
1460 .map(|o| o.clone())
1461 .expect("Order must exist in cache");
1462 matching_engine.process_order(&mut order, account_id);
1463 }
1464 TradingCommand::ModifyOrder(ref command) => {
1465 matching_engine.process_modify(command, account_id);
1466 }
1467 TradingCommand::ModifyOrders(ref command) => {
1468 matching_engine.process_batch_modify(command, account_id);
1469 }
1470 TradingCommand::CancelOrder(ref command) => {
1471 matching_engine.process_cancel(command, account_id);
1472 }
1473 TradingCommand::CancelOrders(ref command) => {
1474 matching_engine.process_batch_cancel(command, account_id);
1475 }
1476 TradingCommand::CancelAllOrders(ref command) => {
1477 matching_engine.process_cancel_all(command, account_id);
1478 }
1479 TradingCommand::SubmitOrderList(ref command) => {
1480 let mut orders: Vec<OrderAny> = self
1481 .cache
1482 .borrow()
1483 .orders_for_ids(&command.order_list.client_order_ids, command);
1484
1485 for order in &mut orders {
1486 matching_engine.process_order(order, account_id);
1487 }
1488 }
1489 _ => {}
1490 }
1491 } else {
1492 panic!("Matching engine not found for instrument {instrument_id}");
1493 }
1494 }
1495
1496 fn process_modify_submitted_order(&self, command: &ModifyOrder) -> bool {
1497 let Some(order) = self
1498 .cache
1499 .borrow()
1500 .order(&command.client_order_id)
1501 .map(|o| o.clone())
1502 else {
1503 return false;
1504 };
1505
1506 let modifies_submitted_order = matches!(order.status(), OrderStatus::Submitted)
1507 || (matches!(order.status(), OrderStatus::PendingUpdate)
1508 && order
1509 .previous_status()
1510 .is_some_and(|status| matches!(status, OrderStatus::Submitted)));
1511
1512 if !modifies_submitted_order {
1513 return false;
1514 }
1515
1516 self.generate_order_updated(
1517 &order,
1518 command.quantity.unwrap_or_else(|| order.quantity()),
1519 command.price.or_else(|| order.price()),
1520 command.trigger_price.or_else(|| order.trigger_price()),
1521 );
1522 true
1523 }
1524
1525 fn generate_order_updated(
1526 &self,
1527 order: &OrderAny,
1528 quantity: Quantity,
1529 price: Option<Price>,
1530 trigger_price: Option<Price>,
1531 ) {
1532 let ts_now = self.clock.borrow().timestamp_ns();
1533 let event = OrderEventAny::Updated(OrderUpdated::new(
1534 order.trader_id(),
1535 order.strategy_id(),
1536 order.instrument_id(),
1537 order.client_order_id(),
1538 quantity,
1539 UUID4::new(),
1540 ts_now,
1541 ts_now,
1542 false,
1543 order.venue_order_id(),
1544 order.account_id(),
1545 price,
1546 trigger_price,
1547 None,
1548 order.is_quote_quantity(),
1549 ));
1550 Self::dispatch_order_event(event);
1551 }
1552
1553 fn dispatch_order_event(event: OrderEventAny) {
1554 msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), event);
1555 }
1556
1557 fn account_at_starting_balances(&self) -> bool {
1558 let Some(account) = self.get_account() else {
1559 return false;
1560 };
1561
1562 let balances = account.balances();
1563
1564 for starting in &self.starting_balances {
1565 let Some(balance) = balances.get(&starting.currency) else {
1566 return false;
1567 };
1568
1569 if balance.total != *starting || balance.free != *starting {
1570 return false;
1571 }
1572 }
1573
1574 true
1575 }
1576
1577 fn generate_fresh_account_state(&self) {
1578 let balances: Vec<AccountBalance> = self
1579 .starting_balances
1580 .iter()
1581 .map(|money| AccountBalance::new(*money, Money::zero(money.currency), *money))
1582 .collect();
1583
1584 if let Some(exec_client) = &self.exec_client {
1585 exec_client
1586 .generate_account_state(balances, vec![], true, self.clock.borrow().timestamp_ns())
1587 .unwrap();
1588 }
1589
1590 let calculate_account_state = !self.frozen_account;
1591
1592 if let Some(mut account) = self.get_account() {
1593 account.set_calculate_account_state(calculate_account_state);
1594
1595 match &mut account {
1596 AccountAny::Margin(margin_account) => {
1597 margin_account.set_default_leverage(self.default_leverage);
1598 for (instrument_id, leverage) in &self.leverages {
1599 margin_account.set_leverage(*instrument_id, *leverage);
1600 }
1601
1602 if let Some(model) = &self.margin_model {
1603 margin_account.set_margin_model(model.clone());
1604 }
1605 }
1606 AccountAny::Cash(cash_account) => {
1607 cash_account.allow_borrowing = self.allow_cash_borrowing;
1608 }
1609 AccountAny::Betting(_) => {}
1610 }
1611
1612 self.cache.borrow_mut().update_account(&account).unwrap();
1613 }
1614 }
1615}