nautilus_execution/order_emulator/
emulator.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
16use std::{cell::RefCell, fmt::Debug, rc::Rc};
17
18use ahash::{AHashMap, AHashSet};
19use nautilus_common::{
20    cache::Cache,
21    clock::Clock,
22    logging::{CMD, EVT, RECV, SEND},
23    messages::{
24        data::{
25            DataCommand, SubscribeCommand, SubscribeQuotes, SubscribeTrades, UnsubscribeCommand,
26            UnsubscribeQuotes, UnsubscribeTrades,
27        },
28        execution::{
29            BatchModifyOrders, CancelAllOrders, CancelOrder, ModifyOrder, SubmitOrder,
30            SubmitOrderList, TradingCommand,
31        },
32    },
33    msgbus::{
34        self, MessagingSwitchboard, TypedHandler, TypedIntoHandler,
35        switchboard::{
36            get_event_order_topic, get_order_canceled_topic, get_quotes_topic, get_trades_topic,
37        },
38    },
39};
40use nautilus_core::{UUID4, WeakCell};
41use nautilus_model::{
42    data::{OrderBookDeltas, QuoteTick, TradeTick},
43    enums::{ContingencyType, OrderSide, OrderSideSpecified, OrderStatus, OrderType, TriggerType},
44    events::{OrderCanceled, OrderEmulated, OrderEventAny, OrderReleased, OrderUpdated},
45    identifiers::{ClientOrderId, ExecAlgorithmId, InstrumentId, PositionId, StrategyId},
46    instruments::Instrument,
47    orders::{LimitOrder, MarketOrder, Order, OrderAny},
48    types::{Price, Quantity},
49};
50use ustr::Ustr;
51
52use super::handlers::OrderEmulatorOnEventHandler;
53use crate::{
54    matching_core::{MatchAction, OrderMatchingCore, RestingOrder},
55    order_manager::{OrderManagerAction, manager::OrderManager},
56    trailing::trailing_stop_calculate,
57};
58
59pub struct OrderEmulator {
60    clock: Rc<RefCell<dyn Clock>>,
61    cache: Rc<RefCell<Cache>>,
62    manager: OrderManager,
63    matching_cores: AHashMap<InstrumentId, OrderMatchingCore>,
64    subscribed_quotes: AHashSet<InstrumentId>,
65    subscribed_trades: AHashSet<InstrumentId>,
66    subscribed_strategies: AHashSet<StrategyId>,
67    monitored_positions: AHashSet<PositionId>,
68    quote_tick_handler: Option<TypedHandler<QuoteTick>>,
69    trade_tick_handler: Option<TypedHandler<TradeTick>>,
70    quote_handlers: AHashMap<InstrumentId, TypedHandler<QuoteTick>>,
71    trade_handlers: AHashMap<InstrumentId, TypedHandler<TradeTick>>,
72    on_event_handler: Option<TypedHandler<OrderEventAny>>,
73}
74
75impl Debug for OrderEmulator {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        f.debug_struct(stringify!(OrderEmulator))
78            .field("cores", &self.matching_cores.len())
79            .field("subscribed_quotes", &self.subscribed_quotes.len())
80            .field("subscribed_trades", &self.subscribed_trades.len())
81            .field("subscribed_strategies", &self.subscribed_strategies.len())
82            .finish()
83    }
84}
85
86impl OrderEmulator {
87    pub fn new(clock: Rc<RefCell<dyn Clock>>, cache: Rc<RefCell<Cache>>) -> Self {
88        let active_local = true;
89        let manager = OrderManager::new(clock.clone(), cache.clone(), active_local);
90
91        Self {
92            clock,
93            cache,
94            manager,
95            matching_cores: AHashMap::new(),
96            subscribed_quotes: AHashSet::new(),
97            subscribed_trades: AHashSet::new(),
98            subscribed_strategies: AHashSet::new(),
99            monitored_positions: AHashSet::new(),
100            quote_tick_handler: None,
101            trade_tick_handler: None,
102            quote_handlers: AHashMap::new(),
103            trade_handlers: AHashMap::new(),
104            on_event_handler: None,
105        }
106    }
107
108    pub fn register_msgbus_handlers(emulator: &Rc<RefCell<Self>>) {
109        let weak = WeakCell::from(Rc::downgrade(emulator));
110
111        let execute_weak = weak.clone();
112        let execute_handler = TypedIntoHandler::from(move |cmd: TradingCommand| {
113            if let Some(emulator) = execute_weak.upgrade() {
114                emulator.borrow_mut().execute(cmd);
115            }
116        });
117        msgbus::register_trading_command_endpoint(
118            MessagingSwitchboard::order_emulator_execute(),
119            execute_handler,
120        );
121
122        let quote_weak = weak.clone();
123        let quote_handler = TypedHandler::from(move |quote: &QuoteTick| {
124            if let Some(emulator) = quote_weak.upgrade() {
125                emulator.borrow_mut().on_quote_tick(*quote);
126            }
127        });
128
129        let trade_weak = weak.clone();
130        let trade_handler = TypedHandler::from(move |trade: &TradeTick| {
131            if let Some(emulator) = trade_weak.upgrade() {
132                emulator.borrow_mut().on_trade_tick(*trade);
133            }
134        });
135
136        let on_event_handler = TypedHandler::new(OrderEmulatorOnEventHandler::new(
137            Ustr::from(UUID4::new().as_str()),
138            weak,
139        ));
140
141        let mut emulator = emulator.borrow_mut();
142        emulator.quote_tick_handler = Some(quote_handler);
143        emulator.trade_tick_handler = Some(trade_handler);
144        emulator.on_event_handler = Some(on_event_handler);
145    }
146
147    pub fn set_on_event_handler(&mut self, handler: TypedHandler<OrderEventAny>) {
148        self.on_event_handler = Some(handler);
149    }
150
151    /// Caches a submit order command for emulation tracking.
152    pub fn cache_submit_order_command(&mut self, command: SubmitOrder) {
153        self.manager.cache_submit_order_command(command);
154    }
155
156    /// Subscribes to quote data for the given instrument.
157    fn subscribe_quotes_for_instrument(&mut self, instrument_id: InstrumentId) -> bool {
158        if self.quote_handlers.contains_key(&instrument_id) {
159            log::warn!("OrderEmulator attempted duplicate quote subscription for {instrument_id}");
160            return true;
161        }
162
163        let Some(handler) = self.quote_tick_handler.clone() else {
164            log::warn!("Cannot subscribe to quotes: msgbus handlers not registered");
165            return false;
166        };
167
168        let topic = get_quotes_topic(instrument_id);
169        self.quote_handlers.insert(instrument_id, handler.clone());
170        msgbus::subscribe_quotes(topic.into(), handler, None);
171        self.send_data_command(DataCommand::Subscribe(SubscribeCommand::Quotes(
172            SubscribeQuotes::new(
173                instrument_id,
174                None,
175                Some(instrument_id.venue),
176                UUID4::new(),
177                self.clock.borrow().timestamp_ns(),
178                None,
179                None,
180            ),
181        )));
182
183        true
184    }
185
186    /// Subscribes to trade data for the given instrument.
187    fn subscribe_trades_for_instrument(&mut self, instrument_id: InstrumentId) -> bool {
188        if self.trade_handlers.contains_key(&instrument_id) {
189            log::warn!("OrderEmulator attempted duplicate trade subscription for {instrument_id}");
190            return true;
191        }
192
193        let Some(handler) = self.trade_tick_handler.clone() else {
194            log::warn!("Cannot subscribe to trades: msgbus handlers not registered");
195            return false;
196        };
197
198        let topic = get_trades_topic(instrument_id);
199        self.trade_handlers.insert(instrument_id, handler.clone());
200        msgbus::subscribe_trades(topic.into(), handler, None);
201        self.send_data_command(DataCommand::Subscribe(SubscribeCommand::Trades(
202            SubscribeTrades::new(
203                instrument_id,
204                None,
205                Some(instrument_id.venue),
206                UUID4::new(),
207                self.clock.borrow().timestamp_ns(),
208                None,
209                None,
210            ),
211        )));
212
213        true
214    }
215
216    #[must_use]
217    pub fn subscribed_quotes(&self) -> Vec<InstrumentId> {
218        let mut quotes: Vec<InstrumentId> = self.subscribed_quotes.iter().copied().collect();
219        quotes.sort();
220        quotes
221    }
222
223    #[must_use]
224    pub fn subscribed_trades(&self) -> Vec<InstrumentId> {
225        let mut trades: Vec<_> = self.subscribed_trades.iter().copied().collect();
226        trades.sort();
227        trades
228    }
229
230    #[must_use]
231    pub fn subscribed_strategy_count(&self) -> usize {
232        self.subscribed_strategies.len()
233    }
234
235    #[must_use]
236    pub fn monitored_position_count(&self) -> usize {
237        self.monitored_positions.len()
238    }
239
240    #[must_use]
241    pub fn get_submit_order_commands(&self) -> AHashMap<ClientOrderId, SubmitOrder> {
242        self.manager.get_submit_order_commands()
243    }
244
245    #[must_use]
246    pub fn get_matching_core(&self, instrument_id: &InstrumentId) -> Option<OrderMatchingCore> {
247        self.matching_cores.get(instrument_id).cloned()
248    }
249
250    pub fn start(&mut self) {
251        if let Err(e) = self.on_start() {
252            log::error!("{e}");
253        }
254
255        log::info!("Started");
256    }
257
258    pub fn stop(&self) {
259        self.on_stop();
260
261        log::info!("Stopped");
262    }
263
264    pub fn reset(&mut self) {
265        self.on_reset();
266
267        log::info!("Reset");
268    }
269
270    pub fn dispose(&mut self) {
271        self.on_dispose();
272
273        log::info!("Disposed");
274    }
275
276    /// Reactivates emulated orders from cache on start.
277    ///
278    /// # Errors
279    ///
280    /// Returns an error if processing fails.
281    ///
282    pub fn on_start(&mut self) -> anyhow::Result<()> {
283        let emulated_orders: Vec<OrderAny> = self
284            .cache
285            .borrow()
286            .orders_emulated(None, None, None, None, None)
287            .into_iter()
288            .map(|o| o.clone())
289            .collect();
290
291        if emulated_orders.is_empty() {
292            log::debug!("No emulated orders to reactivate");
293            return Ok(());
294        }
295
296        for order in emulated_orders {
297            if !matches!(
298                order.status(),
299                OrderStatus::Initialized | OrderStatus::Emulated
300            ) {
301                continue; // No longer emulated
302            }
303
304            if let Some(parent_order_id) = &order.parent_order_id() {
305                let parent_order = if let Some(order) = self.cache.borrow().order(parent_order_id) {
306                    order.clone()
307                } else {
308                    log::error!("Cannot handle order: parent {parent_order_id} not found");
309                    continue;
310                };
311
312                let is_position_closed = parent_order
313                    .position_id()
314                    .is_some_and(|id| self.cache.borrow().is_position_closed(&id));
315                if parent_order.is_closed() && is_position_closed {
316                    let actions = self.manager.cancel_order(&order);
317                    self.dispatch_manager_actions(actions);
318                    continue; // Parent already closed
319                }
320
321                if parent_order.contingency_type() == Some(ContingencyType::Oto)
322                    && (parent_order.is_active_local()
323                        || parent_order.filled_qty() == Quantity::zero(0))
324                {
325                    continue; // Process contingency order later once parent triggered
326                }
327            }
328
329            let position_id = self
330                .cache
331                .borrow()
332                .position_id(&order.client_order_id())
333                .copied();
334            let client_id = self
335                .cache
336                .borrow()
337                .client_id(&order.client_order_id())
338                .copied();
339
340            let command = SubmitOrder::new(
341                order.trader_id(),
342                client_id,
343                order.strategy_id(),
344                order.instrument_id(),
345                order.client_order_id(),
346                order.init_event().clone(),
347                order.exec_algorithm_id(),
348                position_id,
349                None, // params
350                UUID4::new(),
351                self.clock.borrow().timestamp_ns(),
352                None, // correlation_id
353            );
354
355            self.manager.cache_submit_order_command(command.clone());
356            self.handle_submit_order(&command);
357        }
358
359        Ok(())
360    }
361
362    pub fn on_event(&mut self, event: &OrderEventAny) {
363        log::info!("{RECV}{EVT} {event}");
364
365        let actions = self.manager.handle_event(event);
366        self.dispatch_manager_actions(actions);
367
368        if let Some(order) = self.cache.borrow().order(&event.client_order_id())
369            && order.is_closed()
370            && let Some(matching_core) = self.matching_cores.get_mut(&order.instrument_id())
371            && let Err(e) = matching_core.delete_order(event.client_order_id())
372        {
373            log::debug!("Error deleting order: {e}");
374        }
375        // else: Order not in cache yet
376    }
377
378    pub const fn on_stop(&self) {}
379
380    pub fn on_reset(&mut self) {
381        self.manager.reset();
382        self.matching_cores.clear();
383        self.unsubscribe_all_market_data();
384        self.unsubscribe_strategy_order_events();
385        self.monitored_positions.clear();
386    }
387
388    pub fn on_dispose(&mut self) {
389        self.on_reset();
390    }
391
392    fn unsubscribe_all_market_data(&mut self) {
393        let quote_instrument_ids: Vec<_> = self.subscribed_quotes.drain().collect();
394        for instrument_id in quote_instrument_ids {
395            self.unsubscribe_quotes_for_instrument(instrument_id);
396        }
397
398        let trade_instrument_ids: Vec<_> = self.subscribed_trades.drain().collect();
399        for instrument_id in trade_instrument_ids {
400            self.unsubscribe_trades_for_instrument(instrument_id);
401        }
402    }
403
404    fn unsubscribe_quotes_for_instrument(&mut self, instrument_id: InstrumentId) {
405        if let Some(handler) = self.quote_handlers.remove(&instrument_id) {
406            let topic = get_quotes_topic(instrument_id);
407            msgbus::unsubscribe_quotes(topic.into(), &handler);
408        }
409
410        self.send_data_command(DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(
411            UnsubscribeQuotes::new(
412                instrument_id,
413                None,
414                Some(instrument_id.venue),
415                UUID4::new(),
416                self.clock.borrow().timestamp_ns(),
417                None,
418                None,
419            ),
420        )));
421    }
422
423    fn unsubscribe_trades_for_instrument(&mut self, instrument_id: InstrumentId) {
424        if let Some(handler) = self.trade_handlers.remove(&instrument_id) {
425            let topic = get_trades_topic(instrument_id);
426            msgbus::unsubscribe_trades(topic.into(), &handler);
427        }
428
429        self.send_data_command(DataCommand::Unsubscribe(UnsubscribeCommand::Trades(
430            UnsubscribeTrades::new(
431                instrument_id,
432                None,
433                Some(instrument_id.venue),
434                UUID4::new(),
435                self.clock.borrow().timestamp_ns(),
436                None,
437                None,
438            ),
439        )));
440    }
441
442    fn unsubscribe_strategy_order_events(&mut self) {
443        let strategy_ids: Vec<_> = self.subscribed_strategies.drain().collect();
444        let Some(handler) = &self.on_event_handler else {
445            return;
446        };
447
448        for strategy_id in strategy_ids {
449            msgbus::unsubscribe_order_events(format!("events.order.{strategy_id}").into(), handler);
450        }
451    }
452
453    pub fn execute(&mut self, command: TradingCommand) {
454        log::info!("{RECV}{CMD} {command}");
455
456        match command {
457            TradingCommand::SubmitOrder(command) => self.handle_submit_order(&command),
458            TradingCommand::SubmitOrderList(ref command) => self.handle_submit_order_list(command),
459            TradingCommand::ModifyOrder(ref command) => self.handle_modify_order(command),
460            TradingCommand::ModifyOrders(ref command) => self.handle_batch_modify_orders(command),
461            TradingCommand::CancelOrder(command) => self.handle_cancel_order(command),
462            TradingCommand::CancelAllOrders(ref command) => self.handle_cancel_all_orders(command),
463            _ => log::error!("Cannot handle command: unrecognized {command:?}"),
464        }
465    }
466
467    fn dispatch_manager_actions(&mut self, actions: Vec<OrderManagerAction>) {
468        for action in actions {
469            self.dispatch_manager_action(action);
470        }
471    }
472
473    fn dispatch_manager_action(&mut self, action: OrderManagerAction) {
474        match action {
475            OrderManagerAction::PublishInitialized(event) => publish_order_event(&event),
476            OrderManagerAction::SubmitToEmulator(command) => self.handle_submit_order(&command),
477            OrderManagerAction::SubmitToRisk(command) => {
478                self.send_risk_command(TradingCommand::SubmitOrder(command));
479            }
480            OrderManagerAction::SubmitToAlgorithm {
481                command,
482                exec_algorithm_id,
483            } => self.send_algo_command(command, exec_algorithm_id),
484            OrderManagerAction::CancelLocal(order) => self.cancel_order(&order),
485            OrderManagerAction::ModifyLocalQuantity {
486                mut order,
487                quantity,
488            } => {
489                self.update_order(&mut order, quantity);
490            }
491        }
492    }
493
494    fn create_matching_core(
495        &mut self,
496        instrument_id: InstrumentId,
497        price_increment: Price,
498    ) -> OrderMatchingCore {
499        let matching_core = OrderMatchingCore::new(instrument_id, price_increment);
500        self.matching_cores
501            .insert(instrument_id, matching_core.clone());
502        log::info!("Creating matching core for {instrument_id:?}");
503        matching_core
504    }
505
506    /// # Panics
507    ///
508    /// Panics if the emulation trigger type is `NoTrigger` or if order not in cache.
509    pub fn handle_submit_order(&mut self, command: &SubmitOrder) {
510        let client_order_id = command.client_order_id;
511
512        let mut order = self
513            .cache
514            .borrow()
515            .order(&client_order_id)
516            .map(|o| o.clone())
517            .expect("order must exist in cache");
518
519        let emulation_trigger = order.emulation_trigger();
520
521        if !self
522            .manager
523            .get_submit_order_commands()
524            .contains_key(&client_order_id)
525        {
526            self.manager.cache_submit_order_command(command.clone());
527        }
528
529        assert_ne!(
530            emulation_trigger,
531            Some(TriggerType::NoTrigger),
532            "order.emulation_trigger must not be TriggerType::NoTrigger"
533        );
534
535        if !matches!(
536            emulation_trigger,
537            Some(TriggerType::Default | TriggerType::BidAsk | TriggerType::LastPrice)
538        ) {
539            log::error!("Cannot emulate order: `TriggerType` {emulation_trigger:?} not supported");
540            let actions = self.manager.cancel_order(&order);
541            self.dispatch_manager_actions(actions);
542            return;
543        }
544        let strategy_id = command.strategy_id;
545        let position_id = command.position_id;
546
547        // Get or create matching core
548        let trigger_instrument_id = order
549            .trigger_instrument_id()
550            .unwrap_or_else(|| order.instrument_id());
551
552        let matching_core = self.matching_cores.get(&trigger_instrument_id).cloned();
553
554        let mut matching_core = if let Some(core) = matching_core {
555            core
556        } else {
557            // Handle synthetic instruments
558            let (instrument_id, price_increment) = if trigger_instrument_id.is_synthetic() {
559                let synthetic = self
560                    .cache
561                    .borrow()
562                    .try_synthetic(&trigger_instrument_id)
563                    .cloned();
564
565                match synthetic {
566                    Ok(synthetic) => (synthetic.id, synthetic.price_increment),
567                    Err(e) => {
568                        log::error!("Cannot emulate order: {e}");
569                        let actions = self.manager.cancel_order(&order);
570                        self.dispatch_manager_actions(actions);
571                        return;
572                    }
573                }
574            } else {
575                let instrument = self
576                    .cache
577                    .borrow()
578                    .instrument(&trigger_instrument_id)
579                    .cloned();
580
581                if let Some(instrument) = instrument {
582                    (instrument.id(), instrument.price_increment())
583                } else {
584                    log::error!(
585                        "Cannot emulate order: no instrument {trigger_instrument_id} for trigger"
586                    );
587                    let actions = self.manager.cancel_order(&order);
588                    self.dispatch_manager_actions(actions);
589                    return;
590                }
591            };
592
593            self.create_matching_core(instrument_id, price_increment)
594        };
595
596        // Update trailing stop
597        if matches!(
598            order.order_type(),
599            OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
600        ) {
601            self.update_trailing_stop_order(&mut order);
602            if order.trigger_price().is_none() {
603                log::error!(
604                    "Cannot handle trailing stop order with no trigger_price and no market updates"
605                );
606
607                let actions = self.manager.cancel_order(&order);
608                self.dispatch_manager_actions(actions);
609                return;
610            }
611        }
612
613        // Check if immediately marketable
614        let match_info = RestingOrder::new(
615            order.client_order_id(),
616            order.order_side().as_specified(),
617            order.order_type(),
618            order.trigger_price(),
619            order.price(),
620            true, // is_activated
621        );
622        matching_core.match_order(&match_info);
623
624        // Handle data subscriptions
625        match emulation_trigger.unwrap() {
626            TriggerType::Default | TriggerType::BidAsk => {
627                if !self.subscribed_quotes.contains(&trigger_instrument_id)
628                    && self.subscribe_quotes_for_instrument(trigger_instrument_id)
629                {
630                    self.subscribed_quotes.insert(trigger_instrument_id);
631                }
632            }
633            TriggerType::LastPrice => {
634                if !self.subscribed_trades.contains(&trigger_instrument_id)
635                    && self.subscribe_trades_for_instrument(trigger_instrument_id)
636                {
637                    self.subscribed_trades.insert(trigger_instrument_id);
638                }
639            }
640            _ => {
641                log::error!("Invalid TriggerType: {emulation_trigger:?}");
642                return;
643            }
644        }
645
646        // Check if order was already released
647        if !self
648            .manager
649            .get_submit_order_commands()
650            .contains_key(&order.client_order_id())
651        {
652            return; // Already released
653        }
654
655        // Hold in matching core
656        matching_core.add_order(match_info);
657
658        // Generate emulated event if needed
659        if order.status() == OrderStatus::Initialized {
660            let event = OrderEmulated::new(
661                order.trader_id(),
662                order.strategy_id(),
663                order.instrument_id(),
664                order.client_order_id(),
665                UUID4::new(),
666                self.clock.borrow().timestamp_ns(),
667                self.clock.borrow().timestamp_ns(),
668            );
669
670            let event = OrderEventAny::Emulated(event);
671
672            order = match self.cache.borrow_mut().update_order(&event) {
673                Ok(order) => order,
674                Err(e) => {
675                    log::error!("Cannot apply order event: {e:?}");
676                    return;
677                }
678            };
679
680            self.send_risk_event(event.clone());
681
682            msgbus::publish_order_event(
683                format!("events.order.{}", order.strategy_id()).into(),
684                &event,
685            );
686        }
687
688        self.check_monitoring(strategy_id, position_id);
689
690        // Since we are cloning the matching core, we need to insert it back into the original hashmap
691        self.matching_cores
692            .insert(trigger_instrument_id, matching_core);
693
694        log::info!("Emulating {order}");
695    }
696
697    fn handle_submit_order_list(&mut self, command: &SubmitOrderList) {
698        self.check_monitoring(command.strategy_id, command.position_id);
699
700        let orders: Vec<OrderAny> = self
701            .cache
702            .borrow()
703            .orders_for_ids(&command.order_list.client_order_ids, &command);
704
705        for order in &orders {
706            if let Some(parent_order_id) = order.parent_order_id() {
707                let cache = self.cache.borrow();
708                let parent_order = if let Some(parent_order) = cache.order(&parent_order_id) {
709                    parent_order
710                } else {
711                    log::error!("Parent order for {} not found", order.client_order_id());
712                    continue;
713                };
714
715                if parent_order.contingency_type() == Some(ContingencyType::Oto) {
716                    continue; // Process contingency order later once parent triggered
717                }
718            }
719
720            match self.manager.create_new_submit_order(
721                order,
722                command.position_id,
723                command.client_id,
724                command.correlation_id,
725            ) {
726                Ok(actions) => self.dispatch_manager_actions(actions),
727                Err(e) => log::error!("Error creating new submit order: {e}"),
728            }
729        }
730    }
731
732    fn handle_modify_order(&mut self, command: &ModifyOrder) {
733        if let Some(order) = self.cache.borrow().order(&command.client_order_id) {
734            let price = match command.price {
735                Some(price) => Some(price),
736                None => order.price(),
737            };
738
739            let trigger_price = match command.trigger_price {
740                Some(trigger_price) => Some(trigger_price),
741                None => order.trigger_price(),
742            };
743
744            let ts_now = self.clock.borrow().timestamp_ns();
745            let event = OrderUpdated::new(
746                order.trader_id(),
747                order.strategy_id(),
748                order.instrument_id(),
749                order.client_order_id(),
750                command.quantity.unwrap_or(order.quantity()),
751                UUID4::new(),
752                ts_now,
753                ts_now,
754                false,
755                order.venue_order_id(),
756                order.account_id(),
757                price,
758                trigger_price,
759                None,
760                order.is_quote_quantity(),
761            );
762
763            self.send_exec_event(OrderEventAny::Updated(event));
764
765            let trigger_instrument_id = order
766                .trigger_instrument_id()
767                .unwrap_or_else(|| order.instrument_id());
768
769            if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) {
770                let match_info = RestingOrder::new(
771                    order.client_order_id(),
772                    order.order_side().as_specified(),
773                    order.order_type(),
774                    trigger_price,
775                    price,
776                    true, // is_activated
777                );
778                matching_core.match_order(&match_info);
779            } else {
780                log::error!(
781                    "Cannot handle `ModifyOrder`: no matching core for trigger instrument {trigger_instrument_id}"
782                );
783            }
784        } else {
785            log::error!("Cannot modify order: {} not found", command.client_order_id);
786        }
787    }
788
789    fn handle_batch_modify_orders(&mut self, command: &BatchModifyOrders) {
790        for modify in &command.modifies {
791            self.handle_modify_order(modify);
792        }
793    }
794
795    pub fn handle_cancel_order(&mut self, command: CancelOrder) {
796        let order = if let Some(order) = self.cache.borrow().order(&command.client_order_id) {
797            order.clone()
798        } else {
799            log::error!("Cannot cancel order: {} not found", command.client_order_id);
800            return;
801        };
802
803        let trigger_instrument_id = order
804            .trigger_instrument_id()
805            .unwrap_or_else(|| order.instrument_id());
806
807        let matching_core = if let Some(core) = self.matching_cores.get(&trigger_instrument_id) {
808            core
809        } else {
810            let actions = self.manager.cancel_order(&order);
811            self.dispatch_manager_actions(actions);
812            return;
813        };
814
815        if !matching_core.order_exists(order.client_order_id())
816            && order.is_open()
817            && !order.is_pending_cancel()
818        {
819            // Order not held in the emulator
820            self.send_exec_command(TradingCommand::CancelOrder(command));
821        } else {
822            let actions = self.manager.cancel_order(&order);
823            self.dispatch_manager_actions(actions);
824        }
825    }
826
827    fn handle_cancel_all_orders(&mut self, command: &CancelAllOrders) {
828        let instrument_id = command.instrument_id;
829        let Some(matching_core) = self.matching_cores.get(&instrument_id) else {
830            return; // No orders to cancel
831        };
832
833        // Borrow the iterator and collect just the IDs (8 bytes each) instead
834        // of full RestingOrder snapshots (72 bytes each). The borrow on
835        // matching_core ends here so the manager mutation can proceed.
836        let ids_to_cancel: Vec<ClientOrderId> = match command.order_side {
837            OrderSide::NoOrderSide => matching_core
838                .iter_orders()
839                .map(|o| o.client_order_id)
840                .collect(),
841            OrderSide::Buy => matching_core
842                .iter_bid_orders()
843                .map(|o| o.client_order_id)
844                .collect(),
845            OrderSide::Sell => matching_core
846                .iter_ask_orders()
847                .map(|o| o.client_order_id)
848                .collect(),
849        };
850
851        for id in ids_to_cancel {
852            let order = self.cache.borrow().order(&id).map(|o| o.clone());
853            if let Some(order) = order {
854                let actions = self.manager.cancel_order(&order);
855                self.dispatch_manager_actions(actions);
856            }
857        }
858    }
859
860    pub fn update_order(&mut self, order: &mut OrderAny, new_quantity: Quantity) {
861        log::info!(
862            "Updating order {} quantity to {new_quantity}",
863            order.client_order_id(),
864        );
865
866        let ts_now = self.clock.borrow().timestamp_ns();
867        let event = OrderUpdated::new(
868            order.trader_id(),
869            order.strategy_id(),
870            order.instrument_id(),
871            order.client_order_id(),
872            new_quantity,
873            UUID4::new(),
874            ts_now,
875            ts_now,
876            false,
877            None,
878            order.account_id(),
879            None,
880            None,
881            None,
882            order.is_quote_quantity(),
883        );
884
885        let event = OrderEventAny::Updated(event);
886
887        *order = match self.cache.borrow_mut().update_order(&event) {
888            Ok(order) => order,
889            Err(e) => {
890                log::error!("Cannot apply order event: {e:?}");
891                return;
892            }
893        };
894
895        self.send_risk_event(event);
896    }
897
898    pub fn on_order_book_deltas(&mut self, deltas: &OrderBookDeltas) {
899        log::debug!("Processing {deltas:?}");
900
901        let instrument_id = &deltas.instrument_id;
902        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
903            if let Some(book) = self.cache.borrow().order_book(instrument_id) {
904                let best_bid = book.best_bid_price();
905                let best_ask = book.best_ask_price();
906
907                if let Some(best_bid) = best_bid {
908                    matching_core.set_bid_raw(best_bid);
909                }
910
911                if let Some(best_ask) = best_ask {
912                    matching_core.set_ask_raw(best_ask);
913                }
914            } else {
915                log::error!(
916                    "Cannot handle `OrderBookDeltas`: no book being maintained for {}",
917                    deltas.instrument_id
918                );
919            }
920
921            self.iterate_orders(instrument_id);
922        } else {
923            log::error!(
924                "Cannot handle `OrderBookDeltas`: no matching core for instrument {}",
925                deltas.instrument_id
926            );
927        }
928    }
929
930    pub fn on_quote_tick(&mut self, quote: QuoteTick) {
931        log::debug!("Processing {quote}:?");
932
933        let instrument_id = &quote.instrument_id;
934        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
935            matching_core.set_bid_raw(quote.bid_price);
936            matching_core.set_ask_raw(quote.ask_price);
937
938            self.iterate_orders(instrument_id);
939        } else {
940            log::error!(
941                "Cannot handle `QuoteTick`: no matching core for instrument {}",
942                quote.instrument_id
943            );
944        }
945    }
946
947    pub fn on_trade_tick(&mut self, trade: TradeTick) {
948        log::debug!("Processing {trade:?}");
949
950        let instrument_id = &trade.instrument_id;
951        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
952            matching_core.set_last_raw(trade.price);
953
954            if !self.subscribed_quotes.contains(instrument_id) {
955                matching_core.set_bid_raw(trade.price);
956                matching_core.set_ask_raw(trade.price);
957            }
958
959            self.iterate_orders(instrument_id);
960        } else {
961            log::error!(
962                "Cannot handle `TradeTick`: no matching core for instrument {}",
963                trade.instrument_id
964            );
965        }
966    }
967
968    fn iterate_orders(&mut self, instrument_id: &InstrumentId) {
969        // Process bid actions before ask actions so cross-side
970        // contingencies (OCO/OUO) mutate state between sides
971        let bid_actions = if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
972            matching_core.iterate_bids()
973        } else {
974            log::error!("Cannot iterate orders: no matching core for instrument {instrument_id}");
975            return;
976        };
977
978        for action in bid_actions {
979            match action {
980                MatchAction::FillLimit(id) => self.fill_limit_order(id),
981                MatchAction::TriggerStop(id) => self.trigger_stop_order(id),
982            }
983        }
984
985        let ask_actions = if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
986            matching_core.iterate_asks()
987        } else {
988            return;
989        };
990
991        for action in ask_actions {
992            match action {
993                MatchAction::FillLimit(id) => self.fill_limit_order(id),
994                MatchAction::TriggerStop(id) => self.trigger_stop_order(id),
995            }
996        }
997
998        // Re-snapshot orders after actions to avoid stale trailing stop updates
999        let orders = if let Some(matching_core) = self.matching_cores.get(instrument_id) {
1000            matching_core.get_orders()
1001        } else {
1002            return;
1003        };
1004
1005        for match_info in orders {
1006            if !matches!(
1007                match_info.order_type,
1008                OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
1009            ) {
1010                continue;
1011            }
1012
1013            let mut order = match self
1014                .cache
1015                .borrow()
1016                .order(&match_info.client_order_id)
1017                .map(|o| o.clone())
1018            {
1019                Some(order) => order,
1020                None => continue,
1021            };
1022
1023            if order.is_closed() {
1024                continue;
1025            }
1026
1027            self.update_trailing_stop_order(&mut order);
1028        }
1029    }
1030
1031    pub fn cancel_order(&mut self, order: &OrderAny) {
1032        log::info!("Canceling order {}", order.client_order_id());
1033
1034        let mut order = order.clone();
1035        order.set_emulation_trigger(Some(TriggerType::NoTrigger));
1036
1037        let trigger_instrument_id = order
1038            .trigger_instrument_id()
1039            .unwrap_or(order.instrument_id());
1040
1041        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id)
1042            && let Err(e) = matching_core.delete_order(order.client_order_id())
1043        {
1044            log::debug!("Cannot delete order: {e:?}");
1045        }
1046
1047        self.manager
1048            .pop_submit_order_command(order.client_order_id());
1049
1050        self.cache
1051            .borrow_mut()
1052            .update_order_pending_cancel_local(&order);
1053
1054        let ts_now = self.clock.borrow().timestamp_ns();
1055        let event = OrderCanceled::new(
1056            order.trader_id(),
1057            order.strategy_id(),
1058            order.instrument_id(),
1059            order.client_order_id(),
1060            UUID4::new(),
1061            ts_now,
1062            ts_now,
1063            false,
1064            order.venue_order_id(),
1065            order.account_id(),
1066        );
1067
1068        let event = OrderEventAny::Canceled(event);
1069        if let Err(e) = self.cache.borrow_mut().update_order(&event) {
1070            log::error!("Failed to apply order event: {e}");
1071            return;
1072        }
1073
1074        self.send_portfolio_order_event(event.clone());
1075        publish_order_event(&event);
1076    }
1077
1078    fn check_monitoring(&mut self, strategy_id: StrategyId, position_id: Option<PositionId>) {
1079        if !self.subscribed_strategies.contains(&strategy_id) {
1080            // Subscribe to strategy order events
1081            if let Some(handler) = &self.on_event_handler {
1082                msgbus::subscribe_order_events(
1083                    format!("events.order.{strategy_id}").into(),
1084                    handler.clone(),
1085                    None,
1086                );
1087                self.subscribed_strategies.insert(strategy_id);
1088                log::info!("Subscribed to strategy {strategy_id} order events");
1089            }
1090        }
1091
1092        if let Some(position_id) = position_id
1093            && !self.monitored_positions.contains(&position_id)
1094        {
1095            self.monitored_positions.insert(position_id);
1096        }
1097    }
1098
1099    /// Validates market data availability for order release.
1100    ///
1101    /// Returns `Some(released_price)` if market data is available, `None` otherwise.
1102    /// Logs appropriate warnings when market data is not yet available.
1103    ///
1104    /// Does NOT pop the submit order command - caller must do that and handle missing command
1105    /// according to their contract (panic for market orders, return for limit orders).
1106    fn validate_release(
1107        &self,
1108        order: &OrderAny,
1109        matching_core: &OrderMatchingCore,
1110        trigger_instrument_id: InstrumentId,
1111    ) -> Option<Price> {
1112        let released_price = match order.order_side_specified() {
1113            OrderSideSpecified::Buy => matching_core.ask,
1114            OrderSideSpecified::Sell => matching_core.bid,
1115        };
1116
1117        if released_price.is_none() {
1118            log::warn!(
1119                "Cannot release order {} yet: no market data available for {trigger_instrument_id}, will retry on next update",
1120                order.client_order_id(),
1121            );
1122            return None;
1123        }
1124
1125        Some(released_price.unwrap())
1126    }
1127
1128    /// # Panics
1129    ///
1130    /// Panics if the order type is invalid for a stop order.
1131    pub fn trigger_stop_order(&mut self, client_order_id: ClientOrderId) {
1132        let order = match self
1133            .cache
1134            .borrow()
1135            .order(&client_order_id)
1136            .map(|o| o.clone())
1137        {
1138            Some(order) => order,
1139            None => {
1140                log::error!(
1141                    "Cannot trigger stop order: order {client_order_id} not found in cache"
1142                );
1143                return;
1144            }
1145        };
1146
1147        match order.order_type() {
1148            OrderType::StopLimit | OrderType::LimitIfTouched | OrderType::TrailingStopLimit => {
1149                self.fill_limit_order(client_order_id);
1150            }
1151            OrderType::Market
1152            | OrderType::MarketIfTouched
1153            | OrderType::StopMarket
1154            | OrderType::TrailingStopMarket => self.fill_market_order(client_order_id),
1155            _ => panic!("invalid `OrderType`, was {}", order.order_type()),
1156        }
1157    }
1158
1159    /// # Panics
1160    ///
1161    /// Panics if a limit order has no price.
1162    pub fn fill_limit_order(&mut self, client_order_id: ClientOrderId) {
1163        let order = match self
1164            .cache
1165            .borrow()
1166            .order(&client_order_id)
1167            .map(|o| o.clone())
1168        {
1169            Some(order) => order,
1170            None => {
1171                log::error!("Cannot fill limit order: order {client_order_id} not found in cache");
1172                return;
1173            }
1174        };
1175
1176        if matches!(order.order_type(), OrderType::Limit) {
1177            self.fill_market_order(client_order_id);
1178            return;
1179        }
1180
1181        let trigger_instrument_id = order
1182            .trigger_instrument_id()
1183            .unwrap_or(order.instrument_id());
1184
1185        let matching_core = match self.matching_cores.get(&trigger_instrument_id) {
1186            Some(core) => core,
1187            None => {
1188                log::error!(
1189                    "Cannot fill limit order: no matching core for instrument {trigger_instrument_id}"
1190                );
1191                return; // Order stays queued for retry
1192            }
1193        };
1194
1195        let released_price =
1196            match self.validate_release(&order, matching_core, trigger_instrument_id) {
1197                Some(price) => price,
1198                None => return, // Order stays queued for retry
1199            };
1200
1201        let command = match self
1202            .manager
1203            .pop_submit_order_command(order.client_order_id())
1204        {
1205            Some(command) => command,
1206            None => return, // Order already released
1207        };
1208
1209        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) {
1210            if let Err(e) = matching_core.delete_order(client_order_id) {
1211                log::debug!("Error deleting order: {e:?}");
1212            }
1213
1214            let emulation_trigger = TriggerType::NoTrigger;
1215
1216            // Transform order
1217            let mut transformed = if let Ok(transformed) = LimitOrder::new_checked(
1218                order.trader_id(),
1219                order.strategy_id(),
1220                order.instrument_id(),
1221                order.client_order_id(),
1222                order.order_side(),
1223                order.quantity(),
1224                order.price().unwrap(),
1225                order.time_in_force(),
1226                order.expire_time(),
1227                order.is_post_only(),
1228                order.is_reduce_only(),
1229                order.is_quote_quantity(),
1230                order.display_qty(),
1231                Some(emulation_trigger),
1232                Some(trigger_instrument_id),
1233                order.contingency_type(),
1234                order.order_list_id(),
1235                order.linked_order_ids().map(Vec::from),
1236                order.parent_order_id(),
1237                order.exec_algorithm_id(),
1238                order.exec_algorithm_params().cloned(),
1239                order.exec_spawn_id(),
1240                order.tags().map(Vec::from),
1241                UUID4::new(),
1242                self.clock.borrow().timestamp_ns(),
1243            ) {
1244                transformed
1245            } else {
1246                log::error!("Cannot create limit order");
1247                return;
1248            };
1249            transformed.liquidity_side = order.liquidity_side();
1250
1251            // TODO: fix
1252            // let triggered_price = order.trigger_price();
1253            // if triggered_price.is_some() {
1254            //     transformed.trigger_price() = (triggered_price.unwrap());
1255            // }
1256
1257            let original_events = order.events();
1258
1259            for event in original_events {
1260                transformed.events.insert(0, event.clone());
1261            }
1262
1263            let add_result = {
1264                let mut cache = self.cache.borrow_mut();
1265                cache.add_order(
1266                    OrderAny::Limit(transformed.clone()),
1267                    command.position_id,
1268                    command.client_id,
1269                    true,
1270                )
1271            };
1272
1273            if let Err(e) = add_result {
1274                log::error!("Failed to add order: {e}");
1275            } else {
1276                msgbus::publish_order_event(
1277                    format!("events.order.{}", order.strategy_id()).into(),
1278                    transformed.last_event(),
1279                );
1280            }
1281
1282            let event = OrderReleased::new(
1283                order.trader_id(),
1284                order.strategy_id(),
1285                order.instrument_id(),
1286                order.client_order_id(),
1287                released_price,
1288                UUID4::new(),
1289                self.clock.borrow().timestamp_ns(),
1290                self.clock.borrow().timestamp_ns(),
1291            );
1292
1293            let event = OrderEventAny::Released(event);
1294
1295            let transformed = match self.cache.borrow_mut().update_order(&event) {
1296                Ok(order) => order,
1297                Err(e) => {
1298                    log::error!("Failed to apply order event: {e}");
1299                    return;
1300                }
1301            };
1302
1303            self.send_risk_event(event.clone());
1304
1305            log::info!("Releasing order {}", order.client_order_id());
1306
1307            // Publish event
1308            msgbus::publish_order_event(
1309                format!("events.order.{}", transformed.strategy_id()).into(),
1310                &event,
1311            );
1312
1313            if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
1314                self.send_algo_command(command, exec_algorithm_id);
1315            } else {
1316                self.send_exec_command(TradingCommand::SubmitOrder(command));
1317            }
1318        }
1319    }
1320
1321    /// # Panics
1322    ///
1323    /// Panics if a market order command is missing.
1324    pub fn fill_market_order(&mut self, client_order_id: ClientOrderId) {
1325        let mut order = match self
1326            .cache
1327            .borrow()
1328            .order(&client_order_id)
1329            .map(|o| o.clone())
1330        {
1331            Some(order) => order,
1332            None => {
1333                log::error!("Cannot fill market order: order {client_order_id} not found in cache");
1334                return;
1335            }
1336        };
1337
1338        let trigger_instrument_id = order
1339            .trigger_instrument_id()
1340            .unwrap_or(order.instrument_id());
1341
1342        let matching_core = match self.matching_cores.get(&trigger_instrument_id) {
1343            Some(core) => core,
1344            None => {
1345                log::error!(
1346                    "Cannot fill market order: no matching core for instrument {trigger_instrument_id}"
1347                );
1348                return; // Order stays queued for retry
1349            }
1350        };
1351
1352        let released_price =
1353            match self.validate_release(&order, matching_core, trigger_instrument_id) {
1354                Some(price) => price,
1355                None => return, // Order stays queued for retry
1356            };
1357
1358        let command = self
1359            .manager
1360            .pop_submit_order_command(order.client_order_id())
1361            .expect("invalid operation `fill_market_order` with no command");
1362
1363        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) {
1364            if let Err(e) = matching_core.delete_order(client_order_id) {
1365                log::debug!("Cannot delete order: {e:?}");
1366            }
1367
1368            order.set_emulation_trigger(Some(TriggerType::NoTrigger));
1369
1370            // Transform order
1371            let mut transformed = MarketOrder::new(
1372                order.trader_id(),
1373                order.strategy_id(),
1374                order.instrument_id(),
1375                order.client_order_id(),
1376                order.order_side(),
1377                order.quantity(),
1378                order.time_in_force(),
1379                UUID4::new(),
1380                self.clock.borrow().timestamp_ns(),
1381                order.is_reduce_only(),
1382                order.is_quote_quantity(),
1383                order.contingency_type(),
1384                order.order_list_id(),
1385                order.linked_order_ids().map(Vec::from),
1386                order.parent_order_id(),
1387                order.exec_algorithm_id(),
1388                order.exec_algorithm_params().cloned(),
1389                order.exec_spawn_id(),
1390                order.tags().map(Vec::from),
1391            );
1392
1393            let original_events = order.events();
1394
1395            for event in original_events {
1396                transformed.events.insert(0, event.clone());
1397            }
1398
1399            let add_result = {
1400                let mut cache = self.cache.borrow_mut();
1401                cache.add_order(
1402                    OrderAny::Market(transformed.clone()),
1403                    command.position_id,
1404                    command.client_id,
1405                    true,
1406                )
1407            };
1408
1409            if let Err(e) = add_result {
1410                log::error!("Failed to add order: {e}");
1411            } else {
1412                msgbus::publish_order_event(
1413                    format!("events.order.{}", order.strategy_id()).into(),
1414                    transformed.last_event(),
1415                );
1416            }
1417
1418            let ts_now = self.clock.borrow().timestamp_ns();
1419            let event = OrderReleased::new(
1420                order.trader_id(),
1421                order.strategy_id(),
1422                order.instrument_id(),
1423                order.client_order_id(),
1424                released_price,
1425                UUID4::new(),
1426                ts_now,
1427                ts_now,
1428            );
1429
1430            let event = OrderEventAny::Released(event);
1431
1432            if let Err(e) = self.cache.borrow_mut().update_order(&event) {
1433                log::error!("Failed to apply order event: {e}");
1434                return;
1435            }
1436            self.send_risk_event(event.clone());
1437
1438            log::info!("Releasing order {}", order.client_order_id());
1439
1440            // Publish event
1441            msgbus::publish_order_event(
1442                format!("events.order.{}", order.strategy_id()).into(),
1443                &event,
1444            );
1445
1446            if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
1447                self.send_algo_command(command, exec_algorithm_id);
1448            } else {
1449                self.send_exec_command(TradingCommand::SubmitOrder(command));
1450            }
1451        }
1452    }
1453
1454    fn update_trailing_stop_order(&mut self, order: &mut OrderAny) {
1455        let trigger_instrument_id = order
1456            .trigger_instrument_id()
1457            .unwrap_or_else(|| order.instrument_id());
1458        let Some(matching_core) = self.matching_cores.get(&trigger_instrument_id) else {
1459            log::error!(
1460                "Cannot update trailing-stop order: no matching core for instrument {trigger_instrument_id}"
1461            );
1462            return;
1463        };
1464
1465        let mut bid = matching_core.bid;
1466        let mut ask = matching_core.ask;
1467        let mut last = matching_core.last;
1468        let price_increment = matching_core.price_increment;
1469        let instrument_id = matching_core.instrument_id;
1470
1471        if bid.is_none() || ask.is_none() || last.is_none() {
1472            if let Some(q) = self.cache.borrow().quote(&instrument_id) {
1473                bid.get_or_insert(q.bid_price);
1474                ask.get_or_insert(q.ask_price);
1475            }
1476
1477            if let Some(t) = self.cache.borrow().trade(&instrument_id) {
1478                last.get_or_insert(t.price);
1479            }
1480        }
1481
1482        let (new_trigger_px, new_limit_px) = match trailing_stop_calculate(
1483            price_increment,
1484            order.trigger_price(),
1485            order.activation_price(),
1486            order,
1487            bid,
1488            ask,
1489            last,
1490        ) {
1491            Ok(pair) => pair,
1492            Err(e) => {
1493                log::warn!("Cannot calculate trailing-stop update: {e}");
1494                return;
1495            }
1496        };
1497
1498        if new_trigger_px.is_none() && new_limit_px.is_none() {
1499            return;
1500        }
1501
1502        let ts_now = self.clock.borrow().timestamp_ns();
1503        let update = OrderUpdated::new(
1504            order.trader_id(),
1505            order.strategy_id(),
1506            order.instrument_id(),
1507            order.client_order_id(),
1508            order.quantity(),
1509            UUID4::new(),
1510            ts_now,
1511            ts_now,
1512            false,
1513            order.venue_order_id(),
1514            order.account_id(),
1515            new_limit_px,
1516            new_trigger_px,
1517            None,
1518            order.is_quote_quantity(),
1519        );
1520        let wrapped = OrderEventAny::Updated(update);
1521
1522        *order = match self.cache.borrow_mut().update_order(&wrapped) {
1523            Ok(order) => order,
1524            Err(e) => {
1525                log::error!("Failed to apply order event: {e}");
1526                return;
1527            }
1528        };
1529
1530        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) {
1531            if let Err(e) = matching_core.delete_order(order.client_order_id()) {
1532                log::debug!("Cannot update trailing-stop match info: {e:?}");
1533            }
1534
1535            matching_core.add_order(RestingOrder::new(
1536                order.client_order_id(),
1537                order.order_side().as_specified(),
1538                order.order_type(),
1539                order.trigger_price(),
1540                order.price(),
1541                true,
1542            ));
1543        }
1544
1545        self.send_risk_event(wrapped);
1546    }
1547
1548    fn send_algo_command(&self, command: SubmitOrder, exec_algorithm_id: ExecAlgorithmId) {
1549        let id = command.strategy_id;
1550        log::info!("{id} {CMD}{SEND} {command}");
1551
1552        let endpoint = format!("{exec_algorithm_id}.execute");
1553        msgbus::send_any(endpoint.into(), &TradingCommand::SubmitOrder(command));
1554    }
1555
1556    fn send_risk_command(&self, command: TradingCommand) {
1557        log_cmd_send(&command);
1558        let endpoint = MessagingSwitchboard::risk_engine_queue_execute();
1559        msgbus::send_trading_command(endpoint, command);
1560    }
1561
1562    fn send_exec_command(&self, command: TradingCommand) {
1563        log_cmd_send(&command);
1564        let endpoint = MessagingSwitchboard::exec_engine_queue_execute();
1565        msgbus::send_trading_command(endpoint, command);
1566    }
1567
1568    fn send_risk_event(&self, event: OrderEventAny) {
1569        log_evt_send(&event);
1570        let endpoint = MessagingSwitchboard::risk_engine_process();
1571        msgbus::send_order_event(endpoint, event);
1572    }
1573
1574    fn send_exec_event(&self, event: OrderEventAny) {
1575        log_evt_send(&event);
1576        let endpoint = MessagingSwitchboard::exec_engine_process();
1577        msgbus::send_order_event(endpoint, event);
1578    }
1579
1580    fn send_portfolio_order_event(&self, event: OrderEventAny) {
1581        log_evt_send(&event);
1582        let endpoint = MessagingSwitchboard::portfolio_update_order();
1583        msgbus::send_order_event(endpoint, event);
1584    }
1585
1586    fn send_data_command(&self, command: DataCommand) {
1587        log::info!("{CMD}{SEND} {command:?}");
1588        let endpoint = MessagingSwitchboard::data_engine_queue_execute();
1589        msgbus::send_data_command(endpoint, command);
1590    }
1591}
1592
1593fn publish_order_event(event: &OrderEventAny) {
1594    msgbus::publish_order_event(get_event_order_topic(event.strategy_id()), event);
1595
1596    if let OrderEventAny::Canceled(_) = event {
1597        msgbus::publish_order_event(get_order_canceled_topic(event.instrument_id()), event);
1598    }
1599}
1600
1601fn log_cmd_send(command: &TradingCommand) {
1602    if let Some(id) = command.strategy_id() {
1603        log::info!("{id} {CMD}{SEND} {command}");
1604    } else {
1605        log::info!("{CMD}{SEND} {command}");
1606    }
1607}
1608
1609fn log_evt_send(event: &OrderEventAny) {
1610    let id = event.strategy_id();
1611    log::info!("{id} {EVT}{SEND} {event}");
1612}
1613
1614#[cfg(test)]
1615mod tests {
1616    use std::{cell::RefCell, rc::Rc};
1617
1618    use nautilus_common::{
1619        cache::Cache,
1620        clock::TestClock,
1621        messages::data::{DataCommand, SubscribeCommand, UnsubscribeCommand},
1622        msgbus::{
1623            MessagingSwitchboard,
1624            stubs::{
1625                TypedIntoMessageSavingHandler, get_any_saving_handler,
1626                get_typed_into_message_saving_handler,
1627            },
1628        },
1629    };
1630    use nautilus_core::UUID4;
1631    use nautilus_model::{
1632        data::{QuoteTick, TradeTick},
1633        enums::{AggressorSide, OrderSide, OrderType, TriggerType},
1634        identifiers::{ClientOrderId, OrderListId, StrategyId, TradeId, TraderId},
1635        instruments::{
1636            CryptoPerpetual, Instrument, InstrumentAny, stubs::crypto_perpetual_ethusdt,
1637        },
1638        orders::{OrderList, OrderTestBuilder},
1639        types::{Price, Quantity},
1640    };
1641    use rstest::{fixture, rstest};
1642    use ustr::Ustr;
1643
1644    use super::*;
1645
1646    #[fixture]
1647    fn instrument() -> CryptoPerpetual {
1648        crypto_perpetual_ethusdt()
1649    }
1650
1651    #[expect(clippy::type_complexity)]
1652    fn create_emulator() -> (
1653        Rc<RefCell<dyn Clock>>,
1654        Rc<RefCell<Cache>>,
1655        Rc<RefCell<OrderEmulator>>,
1656    ) {
1657        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1658        let cache = Rc::new(RefCell::new(Cache::new(None, None)));
1659        let emulator = Rc::new(RefCell::new(OrderEmulator::new(
1660            clock.clone(),
1661            cache.clone(),
1662        )));
1663
1664        OrderEmulator::register_msgbus_handlers(&emulator);
1665
1666        (clock, cache, emulator)
1667    }
1668
1669    fn create_stop_market_order(instrument: &CryptoPerpetual, trigger: TriggerType) -> OrderAny {
1670        OrderTestBuilder::new(OrderType::StopMarket)
1671            .instrument_id(instrument.id())
1672            .side(OrderSide::Buy)
1673            .trigger_price(Price::from("5100.00"))
1674            .quantity(Quantity::from(1))
1675            .emulation_trigger(trigger)
1676            .build()
1677    }
1678
1679    fn create_stop_limit_order(instrument: &CryptoPerpetual, trigger: TriggerType) -> OrderAny {
1680        OrderTestBuilder::new(OrderType::StopLimit)
1681            .instrument_id(instrument.id())
1682            .side(OrderSide::Buy)
1683            .price(Price::from("5100.00"))
1684            .trigger_price(Price::from("5100.00"))
1685            .quantity(Quantity::from(1))
1686            .emulation_trigger(trigger)
1687            .build()
1688    }
1689
1690    fn create_list_stop_market_order(
1691        instrument: &CryptoPerpetual,
1692        client_order_id: &str,
1693        order_list_id: OrderListId,
1694    ) -> OrderAny {
1695        OrderTestBuilder::new(OrderType::StopMarket)
1696            .instrument_id(instrument.id())
1697            .client_order_id(ClientOrderId::from(client_order_id))
1698            .order_list_id(order_list_id)
1699            .side(OrderSide::Buy)
1700            .trigger_price(Price::from("5100.00"))
1701            .quantity(Quantity::from(1))
1702            .emulation_trigger(TriggerType::BidAsk)
1703            .build()
1704    }
1705
1706    fn create_submit_order(instrument: &CryptoPerpetual, order: &OrderAny) -> SubmitOrder {
1707        SubmitOrder::new(
1708            TraderId::from("TRADER-001"),
1709            None,
1710            StrategyId::from("STRATEGY-001"),
1711            instrument.id(),
1712            order.client_order_id(),
1713            order.init_event().clone(),
1714            None,
1715            None,
1716            None,
1717            UUID4::new(),
1718            0.into(),
1719            None, // correlation_id
1720        )
1721    }
1722
1723    fn create_quote_tick(instrument: &CryptoPerpetual, bid: &str, ask: &str) -> QuoteTick {
1724        QuoteTick::new(
1725            instrument.id(),
1726            Price::from(bid),
1727            Price::from(ask),
1728            Quantity::from(10),
1729            Quantity::from(10),
1730            0.into(),
1731            0.into(),
1732        )
1733    }
1734
1735    fn create_trade_tick(instrument: &CryptoPerpetual, price: &str) -> TradeTick {
1736        TradeTick::new(
1737            instrument.id(),
1738            Price::from(price),
1739            Quantity::from(1),
1740            AggressorSide::Buyer,
1741            TradeId::from("T-001"),
1742            0.into(),
1743            0.into(),
1744        )
1745    }
1746
1747    fn add_instrument_to_cache(cache: &Rc<RefCell<Cache>>, instrument: &CryptoPerpetual) {
1748        cache
1749            .borrow_mut()
1750            .add_instrument(InstrumentAny::CryptoPerpetual(instrument.clone()))
1751            .unwrap();
1752    }
1753
1754    fn register_risk_event_handler(id: &str) -> TypedIntoMessageSavingHandler<OrderEventAny> {
1755        let (handler, saving_handler) =
1756            get_typed_into_message_saving_handler::<OrderEventAny>(Some(Ustr::from(id)));
1757        msgbus::register_order_event_endpoint(MessagingSwitchboard::risk_engine_process(), handler);
1758        saving_handler
1759    }
1760
1761    fn register_portfolio_event_handler(id: &str) -> TypedIntoMessageSavingHandler<OrderEventAny> {
1762        let (handler, saving_handler) =
1763            get_typed_into_message_saving_handler::<OrderEventAny>(Some(Ustr::from(id)));
1764        msgbus::register_order_event_endpoint(
1765            MessagingSwitchboard::portfolio_update_order(),
1766            handler,
1767        );
1768        saving_handler
1769    }
1770
1771    fn register_data_command_handler(id: &str) -> TypedIntoMessageSavingHandler<DataCommand> {
1772        let (handler, saving_handler) =
1773            get_typed_into_message_saving_handler::<DataCommand>(Some(Ustr::from(id)));
1774        msgbus::register_data_command_endpoint(
1775            MessagingSwitchboard::data_engine_queue_execute(),
1776            handler,
1777        );
1778        saving_handler
1779    }
1780
1781    fn subscribe_order_topic(
1782        strategy_id: StrategyId,
1783    ) -> (TypedHandler<OrderEventAny>, Rc<RefCell<Vec<OrderEventAny>>>) {
1784        let events = Rc::new(RefCell::new(Vec::new()));
1785        let handler = TypedHandler::from({
1786            let events = events.clone();
1787            move |event: &OrderEventAny| {
1788                events.borrow_mut().push(event.clone());
1789            }
1790        });
1791        msgbus::subscribe_order_events(
1792            format!("events.order.{strategy_id}").into(),
1793            handler.clone(),
1794            None,
1795        );
1796        (handler, events)
1797    }
1798
1799    fn subscribe_order_cancel_topic(
1800        instrument_id: InstrumentId,
1801    ) -> (TypedHandler<OrderEventAny>, Rc<RefCell<Vec<OrderEventAny>>>) {
1802        let events = Rc::new(RefCell::new(Vec::new()));
1803        let handler = TypedHandler::from({
1804            let events = events.clone();
1805            move |event: &OrderEventAny| {
1806                events.borrow_mut().push(event.clone());
1807            }
1808        });
1809        msgbus::subscribe_order_events(
1810            get_order_canceled_topic(instrument_id).into(),
1811            handler.clone(),
1812            None,
1813        );
1814        (handler, events)
1815    }
1816
1817    #[rstest]
1818    fn test_dispatch_manager_publish_initialized_publishes_order_event(
1819        instrument: CryptoPerpetual,
1820    ) {
1821        let (_clock, _cache, emulator) = create_emulator();
1822        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
1823        let strategy_id = order.strategy_id();
1824        let client_order_id = order.client_order_id();
1825        let event = OrderEventAny::Initialized(order.init_event().clone());
1826        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
1827
1828        emulator
1829            .borrow_mut()
1830            .dispatch_manager_action(OrderManagerAction::PublishInitialized(event));
1831        msgbus::unsubscribe_order_events(
1832            format!("events.order.{strategy_id}").into(),
1833            &order_handler,
1834        );
1835        let order_events = order_events.borrow();
1836
1837        assert_eq!(order_events.len(), 1);
1838        assert!(matches!(
1839            &order_events[0],
1840            OrderEventAny::Initialized(event) if event.client_order_id == client_order_id
1841        ));
1842    }
1843
1844    #[rstest]
1845    fn test_dispatch_manager_submit_to_emulator_applies_emulated_event(
1846        instrument: CryptoPerpetual,
1847    ) {
1848        let (_clock, cache, emulator) = create_emulator();
1849        let risk_events = register_risk_event_handler("RiskEngine.process.dispatch_emulated");
1850        add_instrument_to_cache(&cache, &instrument);
1851        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
1852        let client_order_id = order.client_order_id();
1853        let command = create_submit_order(&instrument, &order);
1854        cache
1855            .borrow_mut()
1856            .add_order(order, None, None, false)
1857            .unwrap();
1858        emulator
1859            .borrow_mut()
1860            .cache_submit_order_command(command.clone());
1861
1862        emulator
1863            .borrow_mut()
1864            .dispatch_manager_action(OrderManagerAction::SubmitToEmulator(command));
1865        let cache = cache.borrow();
1866        let cached_order = cache.order(&client_order_id).unwrap();
1867        let risk_events = risk_events.get_messages();
1868
1869        assert_eq!(cached_order.status(), OrderStatus::Emulated);
1870        assert_eq!(risk_events.len(), 1);
1871        assert!(matches!(risk_events[0], OrderEventAny::Emulated(_)));
1872    }
1873
1874    #[rstest]
1875    fn test_registered_execute_endpoint_routes_submit_order(instrument: CryptoPerpetual) {
1876        let (_clock, cache, emulator) = create_emulator();
1877        let risk_events = register_risk_event_handler("RiskEngine.process.endpoint_emulated");
1878        let data_commands = register_data_command_handler("DataEngine.queue_execute.endpoint");
1879        add_instrument_to_cache(&cache, &instrument);
1880        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
1881        let client_order_id = order.client_order_id();
1882        let command = create_submit_order(&instrument, &order);
1883        cache
1884            .borrow_mut()
1885            .add_order(order, None, None, false)
1886            .unwrap();
1887        emulator
1888            .borrow_mut()
1889            .cache_submit_order_command(command.clone());
1890
1891        msgbus::send_trading_command(
1892            MessagingSwitchboard::order_emulator_execute(),
1893            TradingCommand::SubmitOrder(command),
1894        );
1895        let cache = cache.borrow();
1896        let cached_order = cache.order(&client_order_id).unwrap();
1897        let risk_events = risk_events.get_messages();
1898        let data_commands = data_commands.get_messages();
1899
1900        assert_eq!(cached_order.status(), OrderStatus::Emulated);
1901        assert_eq!(risk_events.len(), 1);
1902        assert!(matches!(risk_events[0], OrderEventAny::Emulated(_)));
1903        assert_eq!(data_commands.len(), 1);
1904        assert!(matches!(
1905            &data_commands[0],
1906            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
1907                if command.instrument_id == instrument.id()
1908        ));
1909    }
1910
1911    #[rstest]
1912    fn test_dispatch_manager_submit_to_risk_uses_risk_queue(instrument: CryptoPerpetual) {
1913        let (_clock, _cache, emulator) = create_emulator();
1914        let (handler, messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
1915            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
1916        msgbus::register_trading_command_endpoint(
1917            MessagingSwitchboard::risk_engine_queue_execute(),
1918            handler,
1919        );
1920        let order = create_stop_market_order(&instrument, TriggerType::NoTrigger);
1921        let command = create_submit_order(&instrument, &order);
1922        let client_order_id = command.client_order_id;
1923
1924        emulator
1925            .borrow_mut()
1926            .dispatch_manager_action(OrderManagerAction::SubmitToRisk(command));
1927
1928        let messages = messages.get_messages();
1929        assert_eq!(messages.len(), 1);
1930        assert!(matches!(
1931            messages.first(),
1932            Some(TradingCommand::SubmitOrder(command))
1933                if command.client_order_id == client_order_id
1934        ));
1935    }
1936
1937    #[rstest]
1938    fn test_dispatch_manager_submit_to_algorithm_uses_dynamic_endpoint(
1939        instrument: CryptoPerpetual,
1940    ) {
1941        let (_clock, _cache, emulator) = create_emulator();
1942        let exec_algorithm_id = ExecAlgorithmId::from("ALG-001");
1943        let endpoint = format!("{exec_algorithm_id}.execute");
1944        let (handler, messages) =
1945            get_any_saving_handler::<TradingCommand>(Some(Ustr::from("ALG-001.execute")));
1946        msgbus::register_any(endpoint.into(), handler);
1947        let order = create_stop_market_order(&instrument, TriggerType::NoTrigger);
1948        let command = create_submit_order(&instrument, &order);
1949        let client_order_id = command.client_order_id;
1950
1951        emulator
1952            .borrow_mut()
1953            .dispatch_manager_action(OrderManagerAction::SubmitToAlgorithm {
1954                command,
1955                exec_algorithm_id,
1956            });
1957
1958        let messages = messages.get_messages();
1959        assert_eq!(messages.len(), 1);
1960        assert!(matches!(
1961            messages.first(),
1962            Some(TradingCommand::SubmitOrder(command))
1963                if command.client_order_id == client_order_id
1964        ));
1965    }
1966
1967    #[rstest]
1968    fn test_dispatch_manager_cancel_local_applies_and_publishes_event(instrument: CryptoPerpetual) {
1969        let (_clock, cache, emulator) = create_emulator();
1970        let portfolio_events =
1971            register_portfolio_event_handler("Portfolio.update_order.dispatch_canceled");
1972        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
1973        let client_order_id = order.client_order_id();
1974        let strategy_id = order.strategy_id();
1975        let instrument_id = order.instrument_id();
1976        let command = create_submit_order(&instrument, &order);
1977        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
1978        let (cancel_handler, cancel_events) = subscribe_order_cancel_topic(instrument_id);
1979        cache
1980            .borrow_mut()
1981            .add_order(order.clone(), None, None, false)
1982            .unwrap();
1983        emulator.borrow_mut().cache_submit_order_command(command);
1984
1985        emulator
1986            .borrow_mut()
1987            .dispatch_manager_action(OrderManagerAction::CancelLocal(order));
1988        msgbus::unsubscribe_order_events(get_event_order_topic(strategy_id).into(), &order_handler);
1989        msgbus::unsubscribe_order_events(
1990            get_order_canceled_topic(instrument_id).into(),
1991            &cancel_handler,
1992        );
1993        let cached_order = cache
1994            .borrow()
1995            .order(&client_order_id)
1996            .map(|order| order.clone())
1997            .unwrap();
1998        let portfolio_events = portfolio_events.get_messages();
1999        let order_events = order_events.borrow();
2000        let cancel_events = cancel_events.borrow();
2001        let commands = emulator.borrow().get_submit_order_commands();
2002
2003        assert_eq!(cached_order.status(), OrderStatus::Canceled);
2004        assert_eq!(portfolio_events.len(), 1);
2005        assert_eq!(order_events.len(), 1);
2006        assert_eq!(cancel_events.len(), 1);
2007        assert!(matches!(
2008            &portfolio_events[0],
2009            OrderEventAny::Canceled(event) if event.client_order_id == client_order_id
2010        ));
2011        assert_eq!(order_events[0].client_order_id(), client_order_id);
2012        assert_eq!(cancel_events[0].client_order_id(), client_order_id);
2013        assert!(!commands.contains_key(&client_order_id));
2014    }
2015
2016    #[rstest]
2017    fn test_dispatch_manager_modify_local_quantity_updates_order(instrument: CryptoPerpetual) {
2018        let (_clock, cache, emulator) = create_emulator();
2019        let risk_events = register_risk_event_handler("RiskEngine.process.dispatch_updated");
2020        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2021        let client_order_id = order.client_order_id();
2022        let quantity = Quantity::from(2);
2023        cache
2024            .borrow_mut()
2025            .add_order(order.clone(), None, None, false)
2026            .unwrap();
2027
2028        emulator
2029            .borrow_mut()
2030            .dispatch_manager_action(OrderManagerAction::ModifyLocalQuantity { order, quantity });
2031        let cache = cache.borrow();
2032        let cached_order = cache.order(&client_order_id).unwrap();
2033        let risk_events = risk_events.get_messages();
2034
2035        assert_eq!(cached_order.quantity(), quantity);
2036        assert_eq!(risk_events.len(), 1);
2037        assert!(matches!(
2038            &risk_events[0],
2039            OrderEventAny::Updated(event) if event.client_order_id == client_order_id
2040        ));
2041    }
2042
2043    #[rstest]
2044    fn test_subscribed_quotes_initially_empty() {
2045        let (_clock, _cache, emulator) = create_emulator();
2046
2047        assert!(emulator.borrow().subscribed_quotes().is_empty());
2048    }
2049
2050    #[rstest]
2051    fn test_subscribed_trades_initially_empty() {
2052        let (_clock, _cache, emulator) = create_emulator();
2053
2054        assert!(emulator.borrow().subscribed_trades().is_empty());
2055    }
2056
2057    #[rstest]
2058    fn test_get_submit_order_commands_initially_empty() {
2059        let (_clock, _cache, emulator) = create_emulator();
2060
2061        assert!(emulator.borrow().get_submit_order_commands().is_empty());
2062    }
2063
2064    #[rstest]
2065    fn test_get_matching_core_returns_none_when_not_created(instrument: CryptoPerpetual) {
2066        let (_clock, _cache, emulator) = create_emulator();
2067
2068        assert!(
2069            emulator
2070                .borrow()
2071                .get_matching_core(&instrument.id())
2072                .is_none()
2073        );
2074    }
2075
2076    #[rstest]
2077    fn test_create_matching_core(instrument: CryptoPerpetual) {
2078        let (_clock, _cache, emulator) = create_emulator();
2079
2080        emulator
2081            .borrow_mut()
2082            .create_matching_core(instrument.id(), instrument.price_increment);
2083
2084        assert!(
2085            emulator
2086                .borrow()
2087                .get_matching_core(&instrument.id())
2088                .is_some()
2089        );
2090    }
2091
2092    #[rstest]
2093    fn test_on_quote_tick_no_matching_core_does_not_panic(instrument: CryptoPerpetual) {
2094        let (_clock, _cache, emulator) = create_emulator();
2095        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
2096
2097        emulator.borrow_mut().on_quote_tick(quote);
2098    }
2099
2100    #[rstest]
2101    fn test_on_trade_tick_no_matching_core_does_not_panic(instrument: CryptoPerpetual) {
2102        let (_clock, _cache, emulator) = create_emulator();
2103        let trade = create_trade_tick(&instrument, "5065.00");
2104
2105        emulator.borrow_mut().on_trade_tick(trade);
2106    }
2107
2108    #[rstest]
2109    fn test_submit_order_bid_ask_trigger_creates_matching_core(instrument: CryptoPerpetual) {
2110        let (_clock, cache, emulator) = create_emulator();
2111        add_instrument_to_cache(&cache, &instrument);
2112        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2113        let command = create_submit_order(&instrument, &order);
2114        cache
2115            .borrow_mut()
2116            .add_order(order, None, None, false)
2117            .unwrap();
2118
2119        emulator
2120            .borrow_mut()
2121            .cache_submit_order_command(command.clone());
2122        emulator.borrow_mut().handle_submit_order(&command);
2123
2124        assert!(
2125            emulator
2126                .borrow()
2127                .get_matching_core(&instrument.id())
2128                .is_some()
2129        );
2130    }
2131
2132    #[rstest]
2133    fn test_submit_order_bid_ask_trigger_tracks_quote_subscription(instrument: CryptoPerpetual) {
2134        let (_clock, cache, emulator) = create_emulator();
2135        let data_commands = register_data_command_handler("DataEngine.queue_execute.quotes");
2136        add_instrument_to_cache(&cache, &instrument);
2137        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2138        let command = create_submit_order(&instrument, &order);
2139        cache
2140            .borrow_mut()
2141            .add_order(order, None, None, false)
2142            .unwrap();
2143
2144        emulator
2145            .borrow_mut()
2146            .cache_submit_order_command(command.clone());
2147        emulator.borrow_mut().handle_submit_order(&command);
2148
2149        assert_eq!(emulator.borrow().subscribed_quotes(), vec![instrument.id()]);
2150        assert!(emulator.borrow().subscribed_trades().is_empty());
2151
2152        let commands = data_commands.get_messages();
2153        assert_eq!(commands.len(), 1);
2154        assert!(matches!(
2155            &commands[0],
2156            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
2157                if command.instrument_id == instrument.id()
2158        ));
2159
2160        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
2161        msgbus::publish_quote(get_quotes_topic(instrument.id()), &quote);
2162        let core = emulator
2163            .borrow()
2164            .get_matching_core(&instrument.id())
2165            .unwrap();
2166        assert_eq!(core.bid, Some(Price::from("5060.00")));
2167        assert_eq!(core.ask, Some(Price::from("5070.00")));
2168    }
2169
2170    #[rstest]
2171    fn test_submit_order_last_price_trigger_tracks_trade_subscription(instrument: CryptoPerpetual) {
2172        let (_clock, cache, emulator) = create_emulator();
2173        let data_commands = register_data_command_handler("DataEngine.queue_execute.trades");
2174        add_instrument_to_cache(&cache, &instrument);
2175        let order = create_stop_market_order(&instrument, TriggerType::LastPrice);
2176        let command = create_submit_order(&instrument, &order);
2177        cache
2178            .borrow_mut()
2179            .add_order(order, None, None, false)
2180            .unwrap();
2181
2182        emulator
2183            .borrow_mut()
2184            .cache_submit_order_command(command.clone());
2185        emulator.borrow_mut().handle_submit_order(&command);
2186
2187        assert!(emulator.borrow().subscribed_quotes().is_empty());
2188        assert_eq!(emulator.borrow().subscribed_trades(), vec![instrument.id()]);
2189
2190        let commands = data_commands.get_messages();
2191        assert_eq!(commands.len(), 1);
2192        assert!(matches!(
2193            &commands[0],
2194            DataCommand::Subscribe(SubscribeCommand::Trades(command))
2195                if command.instrument_id == instrument.id()
2196        ));
2197
2198        let trade = create_trade_tick(&instrument, "5065.00");
2199        msgbus::publish_trade(get_trades_topic(instrument.id()), &trade);
2200        let core = emulator
2201            .borrow()
2202            .get_matching_core(&instrument.id())
2203            .unwrap();
2204        assert_eq!(core.last, Some(Price::from("5065.00")));
2205    }
2206
2207    #[rstest]
2208    fn test_reset_unsubscribes_market_data_and_clears_state(instrument: CryptoPerpetual) {
2209        let (_clock, cache, emulator) = create_emulator();
2210        let data_commands = register_data_command_handler("DataEngine.queue_execute.reset");
2211        add_instrument_to_cache(&cache, &instrument);
2212        let quote_order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2213        let quote_command = create_submit_order(&instrument, &quote_order);
2214        let trade_order = OrderTestBuilder::new(OrderType::StopMarket)
2215            .instrument_id(instrument.id())
2216            .client_order_id(ClientOrderId::from("O-RESET-TRADE"))
2217            .side(OrderSide::Buy)
2218            .trigger_price(Price::from("5100.00"))
2219            .quantity(Quantity::from(1))
2220            .emulation_trigger(TriggerType::LastPrice)
2221            .build();
2222        let trade_command = create_submit_order(&instrument, &trade_order);
2223        cache
2224            .borrow_mut()
2225            .add_order(quote_order, None, None, false)
2226            .unwrap();
2227        cache
2228            .borrow_mut()
2229            .add_order(trade_order, None, None, false)
2230            .unwrap();
2231        emulator
2232            .borrow_mut()
2233            .cache_submit_order_command(quote_command.clone());
2234        emulator.borrow_mut().handle_submit_order(&quote_command);
2235        emulator
2236            .borrow_mut()
2237            .cache_submit_order_command(trade_command.clone());
2238        emulator.borrow_mut().handle_submit_order(&trade_command);
2239        data_commands.clear();
2240
2241        emulator.borrow_mut().reset();
2242        let commands = data_commands.get_messages();
2243        let emulator_ref = emulator.borrow();
2244
2245        assert!(emulator_ref.subscribed_quotes.is_empty());
2246        assert!(emulator_ref.subscribed_trades.is_empty());
2247        assert!(emulator_ref.subscribed_strategies.is_empty());
2248        assert!(emulator_ref.monitored_positions.is_empty());
2249        assert!(emulator_ref.quote_handlers.is_empty());
2250        assert!(emulator_ref.trade_handlers.is_empty());
2251        assert!(emulator_ref.get_submit_order_commands().is_empty());
2252        assert!(emulator_ref.get_matching_core(&instrument.id()).is_none());
2253        assert!(commands.iter().any(|command| matches!(
2254            command,
2255            DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(command))
2256                if command.instrument_id == instrument.id()
2257        )));
2258        assert!(commands.iter().any(|command| matches!(
2259            command,
2260            DataCommand::Unsubscribe(UnsubscribeCommand::Trades(command))
2261                if command.instrument_id == instrument.id()
2262        )));
2263
2264        drop(emulator_ref);
2265        emulator
2266            .borrow_mut()
2267            .create_matching_core(instrument.id(), instrument.price_increment);
2268        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
2269        let trade = create_trade_tick(&instrument, "5065.00");
2270        msgbus::publish_quote(get_quotes_topic(instrument.id()), &quote);
2271        msgbus::publish_trade(get_trades_topic(instrument.id()), &trade);
2272        let core = emulator
2273            .borrow()
2274            .get_matching_core(&instrument.id())
2275            .unwrap();
2276        assert_eq!(core.bid, None);
2277        assert_eq!(core.ask, None);
2278        assert_eq!(core.last, None);
2279    }
2280
2281    #[rstest]
2282    fn test_submit_order_list_handles_reentrant_order_events(instrument: CryptoPerpetual) {
2283        let (_clock, cache, emulator) = create_emulator();
2284        let data_commands = register_data_command_handler("DataEngine.queue_execute.list");
2285        add_instrument_to_cache(&cache, &instrument);
2286        let order_list_id = OrderListId::from("OL-EMULATOR-001");
2287        let first_order = create_list_stop_market_order(&instrument, "O-LIST-001", order_list_id);
2288        let second_order = create_list_stop_market_order(&instrument, "O-LIST-002", order_list_id);
2289        let orders = vec![first_order.clone(), second_order.clone()];
2290        let order_list = OrderList::from_orders(&orders, 0.into());
2291        let order_inits = orders
2292            .iter()
2293            .map(|order| order.init_event().clone())
2294            .collect();
2295        cache
2296            .borrow_mut()
2297            .add_order(first_order.clone(), None, None, false)
2298            .unwrap();
2299        cache
2300            .borrow_mut()
2301            .add_order(second_order.clone(), None, None, false)
2302            .unwrap();
2303        let command = SubmitOrderList::new(
2304            TraderId::from("TRADER-001"),
2305            None,
2306            StrategyId::from("STRATEGY-001"),
2307            order_list,
2308            order_inits,
2309            None,
2310            None,
2311            None,
2312            UUID4::new(),
2313            0.into(),
2314            None,
2315        );
2316
2317        emulator
2318            .borrow_mut()
2319            .execute(TradingCommand::SubmitOrderList(command));
2320
2321        let commands = data_commands.get_messages();
2322        let cache = cache.borrow();
2323        let first_status = cache
2324            .order(&first_order.client_order_id())
2325            .unwrap()
2326            .status();
2327        let second_status = cache
2328            .order(&second_order.client_order_id())
2329            .unwrap()
2330            .status();
2331        drop(cache);
2332        let emulator = emulator.borrow();
2333
2334        assert_eq!(first_status, OrderStatus::Emulated);
2335        assert_eq!(second_status, OrderStatus::Emulated);
2336        assert!(emulator.get_matching_core(&instrument.id()).is_some());
2337        assert_eq!(emulator.subscribed_quotes(), vec![instrument.id()]);
2338        assert!(commands.iter().any(|command| matches!(
2339            command,
2340            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
2341                if command.instrument_id == instrument.id()
2342        )));
2343    }
2344
2345    #[rstest]
2346    fn test_submit_order_caches_command(instrument: CryptoPerpetual) {
2347        let (_clock, cache, emulator) = create_emulator();
2348        add_instrument_to_cache(&cache, &instrument);
2349        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2350        let client_order_id = order.client_order_id();
2351        let command = create_submit_order(&instrument, &order);
2352        cache
2353            .borrow_mut()
2354            .add_order(order, None, None, false)
2355            .unwrap();
2356
2357        emulator
2358            .borrow_mut()
2359            .cache_submit_order_command(command.clone());
2360        emulator.borrow_mut().handle_submit_order(&command);
2361
2362        let commands = emulator.borrow().get_submit_order_commands();
2363        assert!(commands.contains_key(&client_order_id));
2364    }
2365
2366    #[rstest]
2367    fn test_handle_submit_order_applies_emulated_event_to_cache(instrument: CryptoPerpetual) {
2368        let (_clock, cache, emulator) = create_emulator();
2369        let risk_events = register_risk_event_handler("RiskEngine.process.emulated");
2370        add_instrument_to_cache(&cache, &instrument);
2371        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2372        let client_order_id = order.client_order_id();
2373        let strategy_id = order.strategy_id();
2374        let command = create_submit_order(&instrument, &order);
2375        cache
2376            .borrow_mut()
2377            .add_order(order, None, None, false)
2378            .unwrap();
2379        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
2380
2381        emulator
2382            .borrow_mut()
2383            .cache_submit_order_command(command.clone());
2384        emulator.borrow_mut().handle_submit_order(&command);
2385        msgbus::unsubscribe_order_events(
2386            format!("events.order.{strategy_id}").into(),
2387            &order_handler,
2388        );
2389        let cache = cache.borrow();
2390        let cached_order = cache.order(&client_order_id).unwrap();
2391        let risk_events = risk_events.get_messages();
2392        let order_events = order_events.borrow();
2393
2394        assert_eq!(cached_order.status(), OrderStatus::Emulated);
2395        assert_eq!(cached_order.event_count(), 2);
2396        assert_eq!(risk_events.len(), 1);
2397        assert!(matches!(risk_events[0], OrderEventAny::Emulated(_)));
2398        assert_eq!(order_events.len(), 1);
2399        assert!(matches!(order_events[0], OrderEventAny::Emulated(_)));
2400    }
2401
2402    #[rstest]
2403    fn test_submit_order_missing_synthetic_trigger_cancels_order(instrument: CryptoPerpetual) {
2404        let (_clock, cache, emulator) = create_emulator();
2405        add_instrument_to_cache(&cache, &instrument);
2406        let synthetic_id = InstrumentId::from("BTC-ETH-INDEX.SYNTH");
2407        let order = OrderTestBuilder::new(OrderType::StopMarket)
2408            .instrument_id(instrument.id())
2409            .side(OrderSide::Buy)
2410            .trigger_price(Price::from("5100.00"))
2411            .quantity(Quantity::from(1))
2412            .emulation_trigger(TriggerType::BidAsk)
2413            .trigger_instrument_id(synthetic_id)
2414            .build();
2415        let client_order_id = order.client_order_id();
2416        let command = create_submit_order(&instrument, &order);
2417        cache
2418            .borrow_mut()
2419            .add_order(order, None, None, false)
2420            .unwrap();
2421
2422        emulator
2423            .borrow_mut()
2424            .cache_submit_order_command(command.clone());
2425        emulator.borrow_mut().handle_submit_order(&command);
2426
2427        let cache = cache.borrow();
2428        let cached_order = cache.order(&client_order_id).unwrap();
2429        let emulator = emulator.borrow();
2430        let commands = emulator.get_submit_order_commands();
2431
2432        assert_eq!(cached_order.status(), OrderStatus::Canceled);
2433        assert!(emulator.get_matching_core(&synthetic_id).is_none());
2434        assert!(emulator.subscribed_quotes().is_empty());
2435        assert!(emulator.subscribed_trades().is_empty());
2436        assert!(!commands.contains_key(&client_order_id));
2437    }
2438
2439    #[rstest]
2440    fn test_update_order_applies_updated_event_to_cache(instrument: CryptoPerpetual) {
2441        let (_clock, cache, emulator) = create_emulator();
2442        let risk_events = register_risk_event_handler("RiskEngine.process.updated");
2443        let mut order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2444        let client_order_id = order.client_order_id();
2445        cache
2446            .borrow_mut()
2447            .add_order(order.clone(), None, None, false)
2448            .unwrap();
2449
2450        emulator
2451            .borrow_mut()
2452            .update_order(&mut order, Quantity::from(2));
2453        let cache = cache.borrow();
2454        let cached_order = cache.order(&client_order_id).unwrap();
2455        let risk_events = risk_events.get_messages();
2456
2457        assert_eq!(order.quantity(), Quantity::from(2));
2458        assert_eq!(cached_order.quantity(), Quantity::from(2));
2459        assert_eq!(cached_order.status(), OrderStatus::Initialized);
2460        assert_eq!(risk_events.len(), 1);
2461        assert!(matches!(risk_events[0], OrderEventAny::Updated(_)));
2462    }
2463
2464    #[rstest]
2465    fn test_fill_market_order_applies_released_event_to_cache(instrument: CryptoPerpetual) {
2466        let (_clock, cache, emulator) = create_emulator();
2467        let risk_events = register_risk_event_handler("RiskEngine.process.released");
2468        add_instrument_to_cache(&cache, &instrument);
2469        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2470        let client_order_id = order.client_order_id();
2471        let strategy_id = order.strategy_id();
2472        let command = create_submit_order(&instrument, &order);
2473        cache
2474            .borrow_mut()
2475            .add_order(order, None, None, false)
2476            .unwrap();
2477
2478        emulator
2479            .borrow_mut()
2480            .cache_submit_order_command(command.clone());
2481        emulator.borrow_mut().handle_submit_order(&command);
2482        risk_events.clear();
2483        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
2484        {
2485            let mut emulator = emulator.borrow_mut();
2486            emulator
2487                .matching_cores
2488                .get_mut(&instrument.id())
2489                .unwrap()
2490                .set_ask_raw(Price::from("5100.00"));
2491            emulator.fill_market_order(client_order_id);
2492        }
2493        msgbus::unsubscribe_order_events(
2494            format!("events.order.{strategy_id}").into(),
2495            &order_handler,
2496        );
2497        let cache = cache.borrow();
2498        let cached_order = cache.order(&client_order_id).unwrap();
2499        let risk_events = risk_events.get_messages();
2500        let order_events = order_events.borrow();
2501
2502        assert_eq!(cached_order.status(), OrderStatus::Released);
2503        assert_eq!(risk_events.len(), 1);
2504        assert!(matches!(risk_events[0], OrderEventAny::Released(_)));
2505        assert_eq!(order_events.len(), 2);
2506        assert!(matches!(order_events[0], OrderEventAny::Initialized(_)));
2507        assert!(matches!(order_events[1], OrderEventAny::Released(_)));
2508    }
2509
2510    #[rstest]
2511    fn test_fill_limit_order_publishes_transformed_initialized_before_released(
2512        instrument: CryptoPerpetual,
2513    ) {
2514        let (_clock, cache, emulator) = create_emulator();
2515        let risk_events = register_risk_event_handler("RiskEngine.process.limit_released");
2516        add_instrument_to_cache(&cache, &instrument);
2517        let order = create_stop_limit_order(&instrument, TriggerType::BidAsk);
2518        let client_order_id = order.client_order_id();
2519        let strategy_id = order.strategy_id();
2520        let command = create_submit_order(&instrument, &order);
2521        cache
2522            .borrow_mut()
2523            .add_order(order, None, None, false)
2524            .unwrap();
2525
2526        emulator
2527            .borrow_mut()
2528            .cache_submit_order_command(command.clone());
2529        emulator.borrow_mut().handle_submit_order(&command);
2530        risk_events.clear();
2531        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
2532        {
2533            let mut emulator = emulator.borrow_mut();
2534            emulator
2535                .matching_cores
2536                .get_mut(&instrument.id())
2537                .unwrap()
2538                .set_ask_raw(Price::from("5100.00"));
2539            emulator.fill_limit_order(client_order_id);
2540        }
2541        msgbus::unsubscribe_order_events(
2542            format!("events.order.{strategy_id}").into(),
2543            &order_handler,
2544        );
2545        let cache = cache.borrow();
2546        let cached_order = cache.order(&client_order_id).unwrap();
2547        let risk_events = risk_events.get_messages();
2548        let order_events = order_events.borrow();
2549
2550        assert_eq!(cached_order.status(), OrderStatus::Released);
2551        assert_eq!(risk_events.len(), 1);
2552        assert!(matches!(risk_events[0], OrderEventAny::Released(_)));
2553        assert_eq!(order_events.len(), 2);
2554        assert!(matches!(order_events[0], OrderEventAny::Initialized(_)));
2555        assert!(matches!(order_events[1], OrderEventAny::Released(_)));
2556    }
2557
2558    #[rstest]
2559    fn test_quote_tick_updates_matching_core_prices(instrument: CryptoPerpetual) {
2560        let (_clock, cache, emulator) = create_emulator();
2561        add_instrument_to_cache(&cache, &instrument);
2562        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2563        let command = create_submit_order(&instrument, &order);
2564        cache
2565            .borrow_mut()
2566            .add_order(order, None, None, false)
2567            .unwrap();
2568        emulator
2569            .borrow_mut()
2570            .cache_submit_order_command(command.clone());
2571        emulator.borrow_mut().handle_submit_order(&command);
2572
2573        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
2574        emulator.borrow_mut().on_quote_tick(quote);
2575
2576        let core = emulator
2577            .borrow()
2578            .get_matching_core(&instrument.id())
2579            .unwrap();
2580        assert_eq!(core.bid, Some(Price::from("5060.00")));
2581        assert_eq!(core.ask, Some(Price::from("5070.00")));
2582    }
2583
2584    #[rstest]
2585    fn test_trade_tick_updates_matching_core_last_price(instrument: CryptoPerpetual) {
2586        let (_clock, cache, emulator) = create_emulator();
2587        add_instrument_to_cache(&cache, &instrument);
2588        let order = create_stop_market_order(&instrument, TriggerType::LastPrice);
2589        let command = create_submit_order(&instrument, &order);
2590        cache
2591            .borrow_mut()
2592            .add_order(order, None, None, false)
2593            .unwrap();
2594        emulator
2595            .borrow_mut()
2596            .cache_submit_order_command(command.clone());
2597        emulator.borrow_mut().handle_submit_order(&command);
2598
2599        let trade = create_trade_tick(&instrument, "5065.00");
2600        emulator.borrow_mut().on_trade_tick(trade);
2601
2602        let core = emulator
2603            .borrow()
2604            .get_matching_core(&instrument.id())
2605            .unwrap();
2606        assert_eq!(core.last, Some(Price::from("5065.00")));
2607    }
2608
2609    #[rstest]
2610    fn test_cancel_order_removes_from_matching_core(instrument: CryptoPerpetual) {
2611        let (_clock, cache, emulator) = create_emulator();
2612        add_instrument_to_cache(&cache, &instrument);
2613        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2614        let command = create_submit_order(&instrument, &order);
2615        cache
2616            .borrow_mut()
2617            .add_order(order.clone(), None, None, false)
2618            .unwrap();
2619        emulator
2620            .borrow_mut()
2621            .cache_submit_order_command(command.clone());
2622        emulator.borrow_mut().handle_submit_order(&command);
2623
2624        emulator.borrow_mut().cancel_order(&order);
2625
2626        let core = emulator
2627            .borrow()
2628            .get_matching_core(&instrument.id())
2629            .unwrap();
2630        assert!(core.get_orders().is_empty());
2631    }
2632
2633    #[rstest]
2634    fn test_cancel_order_removes_cached_command(instrument: CryptoPerpetual) {
2635        let (_clock, cache, emulator) = create_emulator();
2636        add_instrument_to_cache(&cache, &instrument);
2637        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2638        let client_order_id = order.client_order_id();
2639        let command = create_submit_order(&instrument, &order);
2640        cache
2641            .borrow_mut()
2642            .add_order(order.clone(), None, None, false)
2643            .unwrap();
2644        emulator
2645            .borrow_mut()
2646            .cache_submit_order_command(command.clone());
2647        emulator.borrow_mut().handle_submit_order(&command);
2648
2649        emulator.borrow_mut().cancel_order(&order);
2650
2651        let commands = emulator.borrow().get_submit_order_commands();
2652        assert!(!commands.contains_key(&client_order_id));
2653    }
2654}
nautilus_execution/order_emulator/emulator.rs

nautilus_execution/order_emulator/
emulator.rs
