Skip to main content

nautilus_backtest/
exchange.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Provides a `SimulatedExchange` venue for backtesting on historical data.
17
18use 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/// Represents commands with simulated network latency in a min-heap priority queue.
67/// The commands are ordered by timestamp for FIFO processing, with the
68/// earliest timestamp having the highest priority in the queue.
69#[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        // Reverse ordering for min-heap (earliest timestamp first then lowest counter)
89        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/// Simulated exchange venue for realistic trading execution during backtesting.
103///
104/// The `SimulatedExchange` provides a simulation of a trading venue,
105/// including order matching engines, account management, and realistic execution
106/// models. It maintains order books, processes market data, and executes trades
107/// with configurable latency and fill models to accurately simulate real market
108/// conditions during backtesting.
109///
110/// Key features:
111/// - Multi-instrument order matching with realistic execution
112/// - Configurable fee, fill, and latency models
113/// - Support for various order types and execution options
114/// - Account balance and position management
115/// - Market data processing and order book maintenance
116/// - Simulation modules for custom venue behaviors
117#[expect(
118    clippy::struct_excessive_bools,
119    reason = "exchange state mirrors the existing venue configuration flags"
120)]
121pub struct SimulatedExchange {
122    /// The venue identifier.
123    pub id: Venue,
124    /// The order management system type.
125    pub oms_type: OmsType,
126    /// The account type for the venue.
127    pub account_type: AccountType,
128    /// The optional base currency for single-currency accounts.
129    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    /// Creates a new [`SimulatedExchange`] instance from a venue configuration.
183    ///
184    /// # Errors
185    ///
186    /// Returns an error if:
187    /// - `starting_balances` is empty.
188    /// - `base_currency` is `Some` but `starting_balances` contains multiple currencies.
189    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    /// Registers the execution client for the exchange.
259    pub fn register_client(&mut self, client: Rc<dyn ExecutionClient>) {
260        self.exec_client = Some(client);
261    }
262
263    /// Registers the spread quote endpoint used by the data engine.
264    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    /// Sets the fill model for the exchange.
277    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    /// Sets the latency model for the exchange.
290    pub fn set_latency_model(&mut self, latency_model: Box<dyn LatencyModel>) {
291        self.latency_model = Some(latency_model);
292    }
293
294    /// Sets the settlement price for the given instrument.
295    pub fn set_settlement_price(&mut self, instrument_id: InstrumentId, price: Price) {
296        self.settlement_prices.insert(instrument_id, price);
297    }
298
299    /// Returns the configured book type for this venue.
300    #[must_use]
301    pub const fn book_type(&self) -> BookType {
302        self.book_type
303    }
304
305    /// Returns an iterator over the instrument IDs registered with this exchange.
306    pub fn instrument_ids(&self) -> impl Iterator<Item = &InstrumentId> {
307        self.instruments.keys()
308    }
309
310    /// Returns the expiration timestamp for the given instrument, if present.
311    #[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    /// Returns whether an unprocessed instrument remains for the given expiration.
319    #[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    /// Loads non-emulated open orders from the cache into matching engines.
332    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        // Sort for deterministic insertion order
348        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    // panics-doc-ok (transitive via expect_display on venue mismatch)
368    /// Adds an instrument to the simulated exchange and initializes its matching engine.
369    ///
370    /// # Errors
371    ///
372    /// Returns an error if the exchange account type is `Cash` and the instrument is a `CryptoPerpetual` or `CryptoFuture`.
373    ///
374    /// # Panics
375    ///
376    /// Panics if the instrument cannot be added to the exchange.
377    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    /// Returns the best bid price for the given instrument, if available.
440    #[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    /// Returns the best ask price for the given instrument, if available.
448    #[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    /// Returns a reference to the order book for the given instrument, if available.
456    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    /// Returns a reference to the matching engine for the given instrument, if available.
463    #[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    /// Returns a reference to all matching engines keyed by instrument ID.
472    #[must_use]
473    pub const fn get_matching_engines(&self) -> &AHashMap<InstrumentId, OrderMatchingEngine> {
474        &self.matching_engines
475    }
476
477    /// Returns all order books keyed by instrument ID.
478    #[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    /// Returns all open orders, optionally filtered by instrument ID.
488    #[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    /// Returns all open bid orders, optionally filtered by instrument ID.
505    #[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    /// Returns all open ask orders, optionally filtered by instrument ID.
522    #[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    /// Returns the account for this exchange, if an execution client is registered.
539    #[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    /// Returns a reference to the cache.
547    #[must_use]
548    pub fn cache(&self) -> &Rc<RefCell<Cache>> {
549        &self.cache
550    }
551
552    /// Adjusts the account balance by the given amount.
553    ///
554    /// # Panics
555    ///
556    /// Panics if generating account state fails during adjustment.
557    pub fn adjust_account(&mut self, adjustment: Money) {
558        if self.frozen_account {
559            // Nothing to adjust
560            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    /// Returns whether there are pending commands at or before `ts_now`.
606    #[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    /// Returns the latest arrival timestamp across all latency-deferred
617    /// inflight commands, or `None` when the inflight queue is empty.
618    ///
619    /// Used at shutdown to advance the clock past the configured `LatencyModel`
620    /// delay so trailing commands (those emitted on the final data tick or
621    /// in `on_stop`) settle before the engines stop.
622    #[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    /// Iterates all matching engines so newly submitted orders can match
628    /// against the current market state.
629    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    /// Processes instrument expirations due at the given timestamp.
636    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    /// Returns unprocessed instrument expirations for timer scheduling.
649    #[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    /// Advances the exchange clock to the given timestamp so that any event
665    /// generators (modules, account state) see the correct time even when
666    /// no commands are pending.
667    ///
668    /// # Panics
669    ///
670    /// Panics if the clock is not a [`TestClock`].
671    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    /// Sends a trading command to the exchange for processing.
681    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    /// Processes a single order book delta.
723    ///
724    /// # Panics
725    ///
726    /// Panics if adding a missing instrument during delta processing fails.
727    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    /// Processes a batch of order book deltas.
756    ///
757    /// # Panics
758    ///
759    /// Panics if adding a missing instrument during deltas processing fails.
760    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    /// Processes an L2 order book depth snapshot.
789    ///
790    /// # Panics
791    ///
792    /// Panics if adding a missing instrument during depth10 processing fails.
793    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    /// Processes a quote tick and updates the matching engine.
822    ///
823    /// # Panics
824    ///
825    /// Panics if adding a missing instrument during quote tick processing fails.
826    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(&quote.instrument_id) {
832            let instrument = {
833                let cache = self.cache.as_ref().borrow();
834                cache.instrument(&quote.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(&quote.instrument_id) {
848            matching_engine.process_quote_tick(quote);
849        } else {
850            panic!("Matching engine should be initialized");
851        }
852    }
853
854    /// Processes a trade tick and updates the matching engine.
855    ///
856    /// # Panics
857    ///
858    /// Panics if adding a missing instrument during trade tick processing fails.
859    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    /// Processes a bar and updates the matching engine.
888    ///
889    /// # Panics
890    ///
891    /// Panics if adding a missing instrument during bar processing fails.
892    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    /// Processes an instrument status update.
921    ///
922    /// # Panics
923    ///
924    /// Panics if adding a missing instrument during instrument status processing fails.
925    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    /// Processes an instrument close event.
954    ///
955    /// # Panics
956    ///
957    /// Panics if adding a missing instrument during instrument close processing fails.
958    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    /// Processes a funding rate update.
990    ///
991    /// Returns the funding boundary timestamp when the engine should schedule a settlement.
992    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    /// Processes a scheduled funding settlement for the instrument.
1023    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    /// Advances the exchange clock and processes all pending inflight and queued trading commands
1228    /// up to `ts_now`.
1229    ///
1230    /// # Panics
1231    ///
1232    /// Panics if the exchange clock is not a [`TestClock`] or popping an inflight command fails
1233    /// during processing.
1234    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(&timestamp);
1245        }
1246
1247        while let Some(command) = self.message_queue.pop_front() {
1248            self.process_trading_command(command);
1249        }
1250    }
1251
1252    /// Runs all simulation modules for the given timestamp.
1253    ///
1254    /// Must be called once per time step after all command queues have fully
1255    /// settled, not inside the settle loop.
1256    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    /// Resets the exchange to its initial state.
1278    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    /// Logs diagnostic information from all simulation modules.
1300    pub fn log_diagnostics(&self) {
1301        for module in &self.modules {
1302            module.log_diagnostics();
1303        }
1304    }
1305
1306    /// Checks if any margin accounts have breached maintenance margin and liquidates open
1307    /// positions when the trigger threshold is met.
1308    ///
1309    /// Liquidation is scoped to the breached settlement currency: only positions whose
1310    /// instrument settles in the same currency as the breached margin account are closed.
1311    /// Positions settled in other currencies remain open, isolating the liquidation to
1312    /// the currency whose equity fell below the maintenance threshold.
1313    ///
1314    /// > **Note**: A future `cross_margin_mode` venue configuration could extend this to
1315    /// > liquidate all positions across all settlement currencies simultaneously.
1316    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        // Pre-bucket position indices by settlement currency to avoid repeated full scans.
1347        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(&currency) 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; // defer until all positions are priced
1384            }
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}