Skip to main content

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, collections::VecDeque, 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, OrderStatus, OrderType, TimeInForce, TriggerType},
44    events::{
45        OrderCanceled, OrderEmulated, OrderEventAny, OrderInitialized, OrderReleased, OrderUpdated,
46    },
47    identifiers::{ClientOrderId, ExecAlgorithmId, InstrumentId, PositionId, StrategyId},
48    instruments::Instrument,
49    orders::{LimitOrder, MarketOrder, Order, OrderAny},
50    types::{Price, Quantity},
51};
52use ustr::Ustr;
53
54use super::{PendingMessage, handlers::OrderEmulatorOnEventHandler};
55use crate::{
56    matching::{MatchAction, OrderMatchingCore, RestingOrder},
57    order_manager::{OrderManagerAction, manager::OrderManager},
58    trailing::{store_trailing_stop_activation, trailing_stop_calculate},
59};
60
61pub struct OrderEmulator {
62    clock: Rc<RefCell<dyn Clock>>,
63    cache: Rc<RefCell<Cache>>,
64    manager: OrderManager,
65    matching_cores: AHashMap<InstrumentId, OrderMatchingCore>,
66    subscribed_quotes: AHashSet<InstrumentId>,
67    subscribed_trades: AHashSet<InstrumentId>,
68    subscribed_strategies: AHashSet<StrategyId>,
69    monitored_positions: AHashSet<PositionId>,
70    quote_tick_handler: Option<TypedHandler<QuoteTick>>,
71    trade_tick_handler: Option<TypedHandler<TradeTick>>,
72    quote_handlers: AHashMap<InstrumentId, TypedHandler<QuoteTick>>,
73    trade_handlers: AHashMap<InstrumentId, TypedHandler<TradeTick>>,
74    on_event_handler: Option<TypedHandler<OrderEventAny>>,
75    pending_messages: Rc<RefCell<VecDeque<PendingMessage>>>,
76}
77
78impl Debug for OrderEmulator {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        f.debug_struct(stringify!(OrderEmulator))
81            .field("cores", &self.matching_cores.len())
82            .field("subscribed_quotes", &self.subscribed_quotes.len())
83            .field("subscribed_trades", &self.subscribed_trades.len())
84            .field("subscribed_strategies", &self.subscribed_strategies.len())
85            .finish()
86    }
87}
88
89impl OrderEmulator {
90    pub fn new(clock: Rc<RefCell<dyn Clock>>, cache: Rc<RefCell<Cache>>) -> Self {
91        let active_local = true;
92        let manager = OrderManager::new(clock.clone(), cache.clone(), active_local);
93
94        Self {
95            clock,
96            cache,
97            manager,
98            matching_cores: AHashMap::new(),
99            subscribed_quotes: AHashSet::new(),
100            subscribed_trades: AHashSet::new(),
101            subscribed_strategies: AHashSet::new(),
102            monitored_positions: AHashSet::new(),
103            quote_tick_handler: None,
104            trade_tick_handler: None,
105            quote_handlers: AHashMap::new(),
106            trade_handlers: AHashMap::new(),
107            on_event_handler: None,
108            pending_messages: Rc::new(RefCell::new(VecDeque::new())),
109        }
110    }
111
112    pub fn register_msgbus_handlers(emulator: &Rc<RefCell<Self>>) {
113        let weak = WeakCell::from(Rc::downgrade(emulator));
114        let pending_messages = emulator.borrow().pending_messages.clone();
115
116        let execute_weak = weak.clone();
117        let execute_pending = pending_messages.clone();
118        let execute_handler = TypedIntoHandler::from(move |cmd: TradingCommand| {
119            if let Some(emulator) = execute_weak.upgrade() {
120                match emulator.try_borrow_mut() {
121                    Ok(mut emulator) => emulator.execute(cmd),
122                    Err(_) => execute_pending
123                        .borrow_mut()
124                        .push_back(PendingMessage::Command(Box::new(cmd))),
125                }
126            }
127        });
128        msgbus::register_trading_command_endpoint(
129            MessagingSwitchboard::order_emulator_execute(),
130            execute_handler,
131        );
132
133        let quote_weak = weak.clone();
134        let quote_handler = TypedHandler::from(move |quote: &QuoteTick| {
135            if let Some(emulator) = quote_weak.upgrade() {
136                emulator.borrow_mut().on_quote_tick(*quote);
137            }
138        });
139
140        let trade_weak = weak.clone();
141        let trade_handler = TypedHandler::from(move |trade: &TradeTick| {
142            if let Some(emulator) = trade_weak.upgrade() {
143                emulator.borrow_mut().on_trade_tick(*trade);
144            }
145        });
146
147        let on_event_handler = TypedHandler::new(OrderEmulatorOnEventHandler::new(
148            Ustr::from(UUID4::new().as_str()),
149            weak,
150            WeakCell::from(Rc::downgrade(&pending_messages)),
151        ));
152
153        let mut emulator = emulator.borrow_mut();
154        emulator.quote_tick_handler = Some(quote_handler);
155        emulator.trade_tick_handler = Some(trade_handler);
156        emulator.on_event_handler = Some(on_event_handler);
157    }
158
159    pub fn set_on_event_handler(&mut self, handler: TypedHandler<OrderEventAny>) {
160        self.on_event_handler = Some(handler);
161    }
162
163    /// Caches a submit order command for emulation tracking.
164    pub fn cache_submit_order_command(&mut self, command: SubmitOrder) {
165        self.manager.cache_submit_order_command(command);
166    }
167
168    /// Subscribes to quote data for the given instrument.
169    fn subscribe_quotes_for_instrument(&mut self, instrument_id: InstrumentId) -> bool {
170        if self.quote_handlers.contains_key(&instrument_id) {
171            log::warn!("OrderEmulator attempted duplicate quote subscription for {instrument_id}");
172            return true;
173        }
174
175        let Some(handler) = self.quote_tick_handler.clone() else {
176            log::warn!("Cannot subscribe to quotes: msgbus handlers not registered");
177            return false;
178        };
179
180        let topic = get_quotes_topic(instrument_id);
181        self.quote_handlers.insert(instrument_id, handler.clone());
182        msgbus::subscribe_quotes(topic.into(), handler, None);
183        self.send_data_command(DataCommand::Subscribe(SubscribeCommand::Quotes(
184            SubscribeQuotes::new(
185                instrument_id,
186                None,
187                Some(instrument_id.venue),
188                UUID4::new(),
189                self.clock.borrow().timestamp_ns(),
190                None,
191                None,
192            ),
193        )));
194
195        true
196    }
197
198    /// Subscribes to trade data for the given instrument.
199    fn subscribe_trades_for_instrument(&mut self, instrument_id: InstrumentId) -> bool {
200        if self.trade_handlers.contains_key(&instrument_id) {
201            log::warn!("OrderEmulator attempted duplicate trade subscription for {instrument_id}");
202            return true;
203        }
204
205        let Some(handler) = self.trade_tick_handler.clone() else {
206            log::warn!("Cannot subscribe to trades: msgbus handlers not registered");
207            return false;
208        };
209
210        let topic = get_trades_topic(instrument_id);
211        self.trade_handlers.insert(instrument_id, handler.clone());
212        msgbus::subscribe_trades(topic.into(), handler, None);
213        self.send_data_command(DataCommand::Subscribe(SubscribeCommand::Trades(
214            SubscribeTrades::new(
215                instrument_id,
216                None,
217                Some(instrument_id.venue),
218                UUID4::new(),
219                self.clock.borrow().timestamp_ns(),
220                None,
221                None,
222            ),
223        )));
224
225        true
226    }
227
228    #[must_use]
229    pub fn subscribed_quotes(&self) -> Vec<InstrumentId> {
230        let mut quotes: Vec<InstrumentId> = self.subscribed_quotes.iter().copied().collect();
231        quotes.sort();
232        quotes
233    }
234
235    #[must_use]
236    pub fn subscribed_trades(&self) -> Vec<InstrumentId> {
237        let mut trades: Vec<_> = self.subscribed_trades.iter().copied().collect();
238        trades.sort();
239        trades
240    }
241
242    #[must_use]
243    pub fn subscribed_strategy_count(&self) -> usize {
244        self.subscribed_strategies.len()
245    }
246
247    #[must_use]
248    pub fn monitored_position_count(&self) -> usize {
249        self.monitored_positions.len()
250    }
251
252    #[must_use]
253    pub fn get_submit_order_commands(&self) -> AHashMap<ClientOrderId, SubmitOrder> {
254        self.manager.get_submit_order_commands()
255    }
256
257    #[must_use]
258    pub fn get_matching_core(&self, instrument_id: &InstrumentId) -> Option<OrderMatchingCore> {
259        self.matching_cores.get(instrument_id).cloned()
260    }
261
262    pub fn start(&mut self) {
263        if let Err(e) = self.on_start() {
264            log::error!("{e}");
265        }
266
267        log::info!("Started");
268    }
269
270    pub fn stop(&self) {
271        self.on_stop();
272
273        log::info!("Stopped");
274    }
275
276    pub fn reset(&mut self) {
277        self.on_reset();
278
279        log::info!("Reset");
280    }
281
282    pub fn dispose(&mut self) {
283        self.on_dispose();
284
285        log::info!("Disposed");
286    }
287
288    /// Reactivates emulated orders from cache on start.
289    ///
290    /// # Errors
291    ///
292    /// Returns an error if processing fails.
293    pub fn on_start(&mut self) -> anyhow::Result<()> {
294        let emulated_orders: Vec<ClientOrderId> = self
295            .cache
296            .borrow()
297            .orders_emulated(None, None, None, None, None)
298            .into_iter()
299            .map(|order| order.client_order_id())
300            .collect();
301
302        if emulated_orders.is_empty() {
303            log::debug!("No emulated orders to reactivate");
304            return Ok(());
305        }
306
307        for client_order_id in emulated_orders {
308            self.reactivate_order(client_order_id);
309        }
310
311        self.drain_pending_messages();
312
313        Ok(())
314    }
315
316    fn reactivate_order(&mut self, client_order_id: ClientOrderId) {
317        let Some(order) = self.cache.borrow().order_owned(&client_order_id) else {
318            return;
319        };
320
321        if !matches!(
322            order.status(),
323            OrderStatus::Initialized | OrderStatus::Emulated
324        ) {
325            return; // No longer emulated
326        }
327
328        if let Some(parent_order_id) = &order.parent_order_id() {
329            let parent_order = if let Some(order) = self.cache.borrow().order(parent_order_id) {
330                order.clone()
331            } else {
332                log::error!("Cannot handle order: parent {parent_order_id} not found");
333                return;
334            };
335
336            let is_position_closed = parent_order
337                .position_id()
338                .is_none_or(|id| self.cache.borrow().is_position_closed(&id));
339
340            if parent_order.is_closed() && is_position_closed {
341                let actions = self.manager.cancel_order(&order);
342                self.dispatch_manager_actions(actions);
343                return; // Parent already closed
344            }
345
346            if parent_order.contingency_type() == Some(ContingencyType::Oto)
347                && (parent_order.is_active_local()
348                    || parent_order.filled_qty() == Quantity::zero(0))
349            {
350                return; // Process contingency order later once parent triggered
351            }
352        }
353
354        let position_id = self
355            .cache
356            .borrow()
357            .position_id(&order.client_order_id())
358            .copied();
359        let client_id = self
360            .cache
361            .borrow()
362            .client_id(&order.client_order_id())
363            .copied();
364
365        let command = SubmitOrder::new(
366            order.trader_id(),
367            client_id,
368            order.strategy_id(),
369            order.instrument_id(),
370            order.client_order_id(),
371            order.init_event().clone(),
372            order.exec_algorithm_id(),
373            position_id,
374            None, // params
375            UUID4::new(),
376            self.clock.borrow().timestamp_ns(),
377            None, // correlation_id
378        );
379
380        self.manager.cache_submit_order_command(command.clone());
381        self.handle_submit_order(&command);
382    }
383
384    pub fn on_event(&mut self, event: &OrderEventAny) {
385        log::info!("{RECV}{EVT} {event}");
386
387        let actions = self.manager.handle_event(event);
388        self.dispatch_manager_actions(actions);
389
390        if let Some(order) = self.cache.borrow().order(&event.client_order_id())
391            && order.is_closed()
392            && let Some(matching_core) = self.matching_cores.get_mut(&order.instrument_id())
393            && let Err(e) = matching_core.delete_order(event.client_order_id())
394        {
395            log::debug!("Error deleting order: {e}");
396        }
397        // else: Order not in cache yet
398
399        self.drain_pending_messages();
400    }
401
402    fn drain_pending_messages(&mut self) {
403        loop {
404            let message = self.pending_messages.borrow_mut().pop_front();
405            match message {
406                Some(PendingMessage::Command(command)) => self.execute(*command),
407                Some(PendingMessage::Event(event)) => self.on_event(&event),
408                None => break,
409            }
410        }
411    }
412
413    pub const fn on_stop(&self) {}
414
415    pub fn on_reset(&mut self) {
416        self.manager.reset();
417        self.pending_messages.borrow_mut().clear();
418        self.matching_cores.clear();
419        self.unsubscribe_all_market_data();
420        self.unsubscribe_strategy_order_events();
421        self.monitored_positions.clear();
422    }
423
424    pub fn on_dispose(&mut self) {
425        self.on_reset();
426    }
427
428    fn unsubscribe_all_market_data(&mut self) {
429        // Sort so the emitted unsubscribe commands, and the UUID4 draws they consume,
430        // follow the same sequence across runs; the drained sets are AHash-backed.
431        let mut quote_instrument_ids: Vec<_> = self.subscribed_quotes.drain().collect();
432        quote_instrument_ids.sort();
433
434        for instrument_id in quote_instrument_ids {
435            self.unsubscribe_quotes_for_instrument(instrument_id);
436        }
437
438        let mut trade_instrument_ids: Vec<_> = self.subscribed_trades.drain().collect();
439        trade_instrument_ids.sort();
440
441        for instrument_id in trade_instrument_ids {
442            self.unsubscribe_trades_for_instrument(instrument_id);
443        }
444    }
445
446    fn unsubscribe_quotes_for_instrument(&mut self, instrument_id: InstrumentId) {
447        if let Some(handler) = self.quote_handlers.remove(&instrument_id) {
448            let topic = get_quotes_topic(instrument_id);
449            msgbus::unsubscribe_quotes(topic.into(), &handler);
450        }
451
452        self.send_data_command(DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(
453            UnsubscribeQuotes::new(
454                instrument_id,
455                None,
456                Some(instrument_id.venue),
457                UUID4::new(),
458                self.clock.borrow().timestamp_ns(),
459                None,
460                None,
461            ),
462        )));
463    }
464
465    fn unsubscribe_trades_for_instrument(&mut self, instrument_id: InstrumentId) {
466        if let Some(handler) = self.trade_handlers.remove(&instrument_id) {
467            let topic = get_trades_topic(instrument_id);
468            msgbus::unsubscribe_trades(topic.into(), &handler);
469        }
470
471        self.send_data_command(DataCommand::Unsubscribe(UnsubscribeCommand::Trades(
472            UnsubscribeTrades::new(
473                instrument_id,
474                None,
475                Some(instrument_id.venue),
476                UUID4::new(),
477                self.clock.borrow().timestamp_ns(),
478                None,
479                None,
480            ),
481        )));
482    }
483
484    fn unsubscribe_strategy_order_events(&mut self) {
485        let mut strategy_ids: Vec<_> = self.subscribed_strategies.drain().collect();
486        strategy_ids.sort();
487        let Some(handler) = &self.on_event_handler else {
488            return;
489        };
490
491        for strategy_id in strategy_ids {
492            msgbus::unsubscribe_order_events(format!("events.order.{strategy_id}").into(), handler);
493        }
494    }
495
496    pub fn execute(&mut self, command: TradingCommand) {
497        log::info!("{RECV}{CMD} {command}");
498
499        match command {
500            TradingCommand::SubmitOrder(command) => self.handle_submit_order(&command),
501            TradingCommand::SubmitOrderList(ref command) => self.handle_submit_order_list(command),
502            TradingCommand::ModifyOrder(ref command) => self.handle_modify_order(command),
503            TradingCommand::ModifyOrders(ref command) => self.handle_batch_modify_orders(command),
504            TradingCommand::CancelOrder(command) => self.handle_cancel_order(command),
505            TradingCommand::CancelAllOrders(ref command) => self.handle_cancel_all_orders(command),
506            _ => log::error!("Cannot handle command: unrecognized {command}"),
507        }
508
509        self.drain_pending_messages();
510    }
511
512    fn dispatch_manager_actions(&mut self, actions: Vec<OrderManagerAction>) {
513        for action in actions {
514            self.dispatch_manager_action(action);
515        }
516    }
517
518    fn dispatch_manager_action(&mut self, action: OrderManagerAction) {
519        match action {
520            OrderManagerAction::PublishInitialized(event) => publish_order_event(&event),
521            OrderManagerAction::SubmitToEmulator(command) => self.handle_submit_order(&command),
522            OrderManagerAction::SubmitToRisk(command) => {
523                self.send_risk_command(TradingCommand::SubmitOrder(command));
524            }
525            OrderManagerAction::SubmitToAlgorithm {
526                command,
527                exec_algorithm_id,
528            } => self.send_algo_command(command, exec_algorithm_id),
529            OrderManagerAction::CancelLocal(order) => self.cancel_order(&order),
530            OrderManagerAction::ModifyLocalQuantity {
531                mut order,
532                quantity,
533            } => {
534                self.update_order(&mut order, quantity);
535            }
536        }
537    }
538
539    fn create_matching_core(
540        &mut self,
541        instrument_id: InstrumentId,
542        price_increment: Price,
543        emulation_trigger: TriggerType,
544    ) -> OrderMatchingCore {
545        let mut matching_core = OrderMatchingCore::new(instrument_id, price_increment);
546        let cache = self.cache.borrow();
547        if emulation_trigger == TriggerType::LastPrice {
548            if let Some(trade) = cache.trade(&instrument_id) {
549                matching_core.set_bid_raw(trade.price);
550                matching_core.set_ask_raw(trade.price);
551                matching_core.set_last_raw(trade.price);
552            }
553        } else if let Some(quote) = cache.quote(&instrument_id) {
554            matching_core.set_bid_raw(quote.bid_price);
555            matching_core.set_ask_raw(quote.ask_price);
556        }
557
558        drop(cache);
559        self.matching_cores
560            .insert(instrument_id, matching_core.clone());
561
562        log::info!("Creating matching core for {instrument_id:?}");
563        matching_core
564    }
565
566    /// # Panics
567    ///
568    /// Panics if the emulation trigger type is not set or if the order is not in the cache.
569    pub fn handle_submit_order(&mut self, command: &SubmitOrder) {
570        let client_order_id = command.client_order_id;
571
572        let mut order = self
573            .cache
574            .borrow()
575            .order(&client_order_id)
576            .map(|o| o.clone())
577            .expect("order must exist in cache");
578
579        if order.is_closed() {
580            return;
581        }
582
583        let emulation_trigger = order.emulation_trigger();
584
585        if !self
586            .manager
587            .get_submit_order_commands()
588            .contains_key(&client_order_id)
589        {
590            self.manager.cache_submit_order_command(command.clone());
591        }
592
593        assert!(
594            emulation_trigger.is_some(),
595            "order.emulation_trigger must be set"
596        );
597
598        if !matches!(
599            emulation_trigger,
600            Some(TriggerType::Default | TriggerType::BidAsk | TriggerType::LastPrice)
601        ) {
602            log::error!("Cannot emulate order: `TriggerType` {emulation_trigger:?} not supported");
603            let actions = self.manager.cancel_order(&order);
604            self.dispatch_manager_actions(actions);
605            return;
606        }
607
608        let emulation_trigger = emulation_trigger.expect("emulation trigger must be set");
609        let strategy_id = command.strategy_id;
610        let position_id = command.position_id;
611
612        let trigger_instrument_id = order
613            .trigger_instrument_id()
614            .unwrap_or_else(|| order.instrument_id());
615
616        let matching_core = self.matching_cores.get(&trigger_instrument_id).cloned();
617
618        let mut matching_core = if let Some(core) = matching_core {
619            core
620        } else {
621            // Handle synthetic instruments
622            let (instrument_id, price_increment) = if trigger_instrument_id.is_synthetic() {
623                let synthetic = self
624                    .cache
625                    .borrow()
626                    .try_synthetic(&trigger_instrument_id)
627                    .cloned();
628
629                match synthetic {
630                    Ok(synthetic) => (synthetic.id, synthetic.price_increment),
631                    Err(e) => {
632                        log::error!("Cannot emulate order: {e}");
633                        let actions = self.manager.cancel_order(&order);
634                        self.dispatch_manager_actions(actions);
635                        return;
636                    }
637                }
638            } else {
639                let instrument = self
640                    .cache
641                    .borrow()
642                    .instrument(&trigger_instrument_id)
643                    .cloned();
644
645                if let Some(instrument) = instrument {
646                    (instrument.id(), instrument.price_increment())
647                } else {
648                    log::error!(
649                        "Cannot emulate order: no instrument {trigger_instrument_id} for trigger"
650                    );
651                    let actions = self.manager.cancel_order(&order);
652                    self.dispatch_manager_actions(actions);
653                    return;
654                }
655            };
656
657            self.create_matching_core(instrument_id, price_increment, emulation_trigger)
658        };
659
660        // Update trailing stop
661        if matches!(
662            order.order_type(),
663            OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
664        ) {
665            self.update_trailing_stop_order(&mut order);
666            if order.trigger_price().is_none() && is_order_activated(&order) {
667                log::error!(
668                    "Cannot handle trailing stop order with no trigger_price and no market updates"
669                );
670
671                let actions = self.manager.cancel_order(&order);
672                self.dispatch_manager_actions(actions);
673                return;
674            }
675            // Not yet activated: held inert until the activation price is touched
676        }
677
678        self.check_monitoring(strategy_id, position_id);
679
680        // Check if immediately marketable
681        let is_activated = is_order_activated(&order);
682
683        let match_info = RestingOrder::new_with_trigger_type(
684            order.client_order_id(),
685            order.order_side(),
686            order.order_type(),
687            order.emulation_trigger(),
688            if is_activated {
689                order.trigger_price()
690            } else {
691                None
692            },
693            if is_activated { order.price() } else { None },
694            is_activated,
695        );
696
697        if let Some(action) = matching_core.match_order(&match_info) {
698            self.dispatch_match_action(action);
699        }
700
701        // Handle data subscriptions
702        match emulation_trigger {
703            TriggerType::Default | TriggerType::BidAsk => {
704                if !self.subscribed_quotes.contains(&trigger_instrument_id)
705                    && self.subscribe_quotes_for_instrument(trigger_instrument_id)
706                {
707                    self.subscribed_quotes.insert(trigger_instrument_id);
708                }
709            }
710            TriggerType::LastPrice => {
711                if !self.subscribed_trades.contains(&trigger_instrument_id)
712                    && self.subscribe_trades_for_instrument(trigger_instrument_id)
713                {
714                    self.subscribed_trades.insert(trigger_instrument_id);
715                }
716            }
717            _ => {
718                log::error!("Invalid TriggerType: {emulation_trigger:?}");
719                return;
720            }
721        }
722
723        // Check if order was already released
724        if !self
725            .manager
726            .get_submit_order_commands()
727            .contains_key(&order.client_order_id())
728        {
729            return; // Already released
730        }
731
732        // Hold in matching core
733        matching_core.add_order(match_info);
734
735        if order.status() == OrderStatus::Initialized {
736            let event = OrderEmulated::new(
737                order.trader_id(),
738                order.strategy_id(),
739                order.instrument_id(),
740                order.client_order_id(),
741                UUID4::new(),
742                self.clock.borrow().timestamp_ns(),
743                self.clock.borrow().timestamp_ns(),
744            );
745
746            let event = OrderEventAny::Emulated(event);
747
748            order = match self.cache.borrow_mut().update_order(&event) {
749                Ok(order) => order,
750                Err(e) => {
751                    log::error!("Cannot apply order event: {e:?}");
752                    return;
753                }
754            };
755
756            self.send_risk_event(event.clone());
757
758            let topic = get_event_order_topic(order.strategy_id());
759            msgbus::publish_order_event(topic, &event);
760        }
761
762        // Since we are cloning the matching core, we need to insert it back into the original hashmap
763        self.matching_cores
764            .insert(trigger_instrument_id, matching_core);
765
766        log::info!("Emulating {order}");
767    }
768
769    fn handle_submit_order_list(&mut self, command: &SubmitOrderList) {
770        self.check_monitoring(command.strategy_id, command.position_id);
771
772        let orders: Vec<OrderAny> = self
773            .cache
774            .borrow()
775            .orders_for_ids(&command.order_list.client_order_ids, &command);
776
777        for order in &orders {
778            let order = self.cache.borrow().order_owned(&order.client_order_id());
779
780            let Some(order) = order else {
781                continue;
782            };
783
784            if order.is_closed() {
785                continue;
786            }
787
788            if let Some(parent_order_id) = order.parent_order_id() {
789                let cache = self.cache.borrow();
790                let parent_order = if let Some(parent_order) = cache.order(&parent_order_id) {
791                    parent_order
792                } else {
793                    log::error!("Parent order for {} not found", order.client_order_id());
794                    continue;
795                };
796
797                if parent_order.contingency_type() == Some(ContingencyType::Oto) {
798                    continue; // Process contingency order later once parent triggered
799                }
800            }
801
802            match self.manager.create_new_submit_order(
803                &order,
804                command.position_id,
805                command.client_id,
806                command.correlation_id,
807            ) {
808                Ok(actions) => self.dispatch_manager_actions(actions),
809                Err(e) => log::error!("Error creating new submit order: {e}"),
810            }
811        }
812    }
813
814    fn handle_modify_order(&mut self, command: &ModifyOrder) {
815        let order = self
816            .cache
817            .borrow()
818            .order(&command.client_order_id)
819            .map(|order| order.clone());
820
821        if let Some(order) = order {
822            let price = match command.price {
823                Some(price) => Some(price),
824                None => order.price(),
825            };
826
827            let trigger_price = match command.trigger_price {
828                Some(trigger_price) => Some(trigger_price),
829                None => order.trigger_price(),
830            };
831
832            let ts_now = self.clock.borrow().timestamp_ns();
833            let event = OrderUpdated::new(
834                order.trader_id(),
835                order.strategy_id(),
836                order.instrument_id(),
837                order.client_order_id(),
838                command.quantity.unwrap_or(order.quantity()),
839                UUID4::new(),
840                ts_now,
841                ts_now,
842                false,
843                order.venue_order_id(),
844                order.account_id(),
845                price,
846                trigger_price,
847                None,
848                order.is_quote_quantity(),
849            );
850
851            let event = OrderEventAny::Updated(event);
852
853            // Apply locally as for cancels, sending to the execution engine would re-enter it
854            // when a strategy modifies from an order or position event handler.
855            self.apply_order_event(&event);
856
857            // A synchronous event handler may supersede this update before dispatch returns
858            let order = self
859                .cache
860                .borrow()
861                .order(&command.client_order_id)
862                .filter(|order| order.last_event() == &event)
863                .map(|order| order.clone());
864
865            let Some(order) = order else {
866                return;
867            };
868
869            if !self
870                .manager
871                .get_submit_order_commands()
872                .contains_key(&command.client_order_id)
873            {
874                return;
875            }
876
877            let trigger_instrument_id = order
878                .trigger_instrument_id()
879                .unwrap_or_else(|| order.instrument_id());
880
881            let (action, match_info) = if let Some(matching_core) =
882                self.matching_cores.get_mut(&trigger_instrument_id)
883            {
884                if let Err(e) = matching_core.delete_order(order.client_order_id()) {
885                    log::debug!("Cannot update order match info: {e:?}");
886                }
887
888                let is_activated = is_order_activated(&order);
889
890                let match_info = RestingOrder::new_with_trigger_type(
891                    order.client_order_id(),
892                    order.order_side(),
893                    order.order_type(),
894                    order.emulation_trigger(),
895                    if is_activated {
896                        order.trigger_price()
897                    } else {
898                        None
899                    },
900                    if is_activated { order.price() } else { None },
901                    is_activated,
902                );
903
904                (matching_core.match_order(&match_info), match_info)
905            } else {
906                log::error!(
907                    "Cannot handle `ModifyOrder`: no matching core for trigger instrument {trigger_instrument_id}"
908                );
909                return;
910            };
911
912            if let Some(action) = action {
913                self.dispatch_match_action(action);
914            }
915
916            // Hold in matching core unless the match released the order
917            if self
918                .manager
919                .get_submit_order_commands()
920                .contains_key(&command.client_order_id)
921                && let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id)
922            {
923                matching_core.add_order(match_info);
924            }
925        } else {
926            log::error!("Cannot modify order: {} not found", command.client_order_id);
927        }
928    }
929
930    fn handle_batch_modify_orders(&mut self, command: &BatchModifyOrders) {
931        for modify in &command.modifies {
932            self.handle_modify_order(modify);
933        }
934    }
935
936    pub fn handle_cancel_order(&mut self, command: CancelOrder) {
937        let order = if let Some(order) = self.cache.borrow().order(&command.client_order_id) {
938            order.clone()
939        } else {
940            log::error!("Cannot cancel order: {} not found", command.client_order_id);
941            return;
942        };
943
944        let trigger_instrument_id = order
945            .trigger_instrument_id()
946            .unwrap_or_else(|| order.instrument_id());
947
948        let matching_core = if let Some(core) = self.matching_cores.get(&trigger_instrument_id) {
949            core
950        } else {
951            let actions = self.manager.cancel_order(&order);
952            self.dispatch_manager_actions(actions);
953            return;
954        };
955
956        if !matching_core.order_exists(order.client_order_id())
957            && order.is_open()
958            && !order.is_pending_cancel()
959        {
960            // Order not held in the emulator
961            self.send_exec_command(TradingCommand::CancelOrder(command));
962        } else {
963            let actions = self.manager.cancel_order(&order);
964            self.dispatch_manager_actions(actions);
965        }
966    }
967
968    fn handle_cancel_all_orders(&mut self, command: &CancelAllOrders) {
969        let mut ids_to_cancel: Vec<ClientOrderId> = self
970            .matching_cores
971            .values()
972            .flat_map(OrderMatchingCore::iter_orders)
973            .map(|order| order.client_order_id)
974            .collect();
975        ids_to_cancel.sort_unstable();
976        ids_to_cancel.dedup();
977
978        {
979            let cache = self.cache.borrow();
980            ids_to_cancel.retain(|client_order_id| {
981                let Some(order) = cache.order(client_order_id) else {
982                    return false;
983                };
984
985                order.instrument_id() == command.instrument_id
986                    && command
987                        .order_side
988                        .is_none_or(|side| order.order_side() == side)
989                    && command.client_id.is_none_or(|client_id| {
990                        cache
991                            .client_id(client_order_id)
992                            .is_some_and(|order_client_id| *order_client_id == client_id)
993                    })
994            });
995        }
996
997        for id in ids_to_cancel {
998            let Some(order) = self.cache.borrow().order_owned(&id) else {
999                continue;
1000            };
1001            let actions = self.manager.cancel_order(&order);
1002            self.dispatch_manager_actions(actions);
1003        }
1004    }
1005
1006    pub fn update_order(&mut self, order: &mut OrderAny, new_quantity: Quantity) {
1007        log::info!(
1008            "Updating order {} quantity to {new_quantity}",
1009            order.client_order_id(),
1010        );
1011
1012        let ts_now = self.clock.borrow().timestamp_ns();
1013        let event = OrderUpdated::new(
1014            order.trader_id(),
1015            order.strategy_id(),
1016            order.instrument_id(),
1017            order.client_order_id(),
1018            new_quantity,
1019            UUID4::new(),
1020            ts_now,
1021            ts_now,
1022            false,
1023            None,
1024            order.account_id(),
1025            None,
1026            None,
1027            None,
1028            order.is_quote_quantity(),
1029        );
1030
1031        let event = OrderEventAny::Updated(event);
1032
1033        *order = match self.cache.borrow_mut().update_order(&event) {
1034            Ok(order) => order,
1035            Err(e) => {
1036                log::error!("Cannot apply order event: {e:?}");
1037                return;
1038            }
1039        };
1040
1041        self.send_risk_event(event);
1042    }
1043
1044    pub fn on_order_book_deltas(&mut self, deltas: &OrderBookDeltas) {
1045        log::debug!("Processing {deltas:?}");
1046
1047        let instrument_id = &deltas.instrument_id;
1048        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
1049            if let Some(book) = self.cache.borrow().order_book(instrument_id) {
1050                let best_bid = book.best_bid_price();
1051                let best_ask = book.best_ask_price();
1052
1053                if let Some(best_bid) = best_bid {
1054                    matching_core.set_bid_raw(best_bid);
1055                }
1056
1057                if let Some(best_ask) = best_ask {
1058                    matching_core.set_ask_raw(best_ask);
1059                }
1060            } else {
1061                log::error!(
1062                    "Cannot handle `OrderBookDeltas`: no book being maintained for {}",
1063                    deltas.instrument_id
1064                );
1065            }
1066
1067            self.iterate_orders(instrument_id);
1068        } else {
1069            log::error!(
1070                "Cannot handle `OrderBookDeltas`: no matching core for instrument {}",
1071                deltas.instrument_id
1072            );
1073        }
1074    }
1075
1076    pub fn on_quote_tick(&mut self, quote: QuoteTick) {
1077        log::debug!("Processing {quote}:?");
1078
1079        let instrument_id = &quote.instrument_id;
1080        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
1081            matching_core.set_bid_raw(quote.bid_price);
1082            matching_core.set_ask_raw(quote.ask_price);
1083
1084            self.iterate_orders(instrument_id);
1085        } else {
1086            log::error!(
1087                "Cannot handle `QuoteTick`: no matching core for instrument {}",
1088                quote.instrument_id
1089            );
1090        }
1091    }
1092
1093    pub fn on_trade_tick(&mut self, trade: TradeTick) {
1094        log::debug!("Processing {trade:?}");
1095
1096        let instrument_id = &trade.instrument_id;
1097        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
1098            matching_core.set_last_raw(trade.price);
1099
1100            if !self.subscribed_quotes.contains(instrument_id) {
1101                matching_core.set_bid_raw(trade.price);
1102                matching_core.set_ask_raw(trade.price);
1103            }
1104
1105            self.iterate_orders(instrument_id);
1106        } else {
1107            log::error!(
1108                "Cannot handle `TradeTick`: no matching core for instrument {}",
1109                trade.instrument_id
1110            );
1111        }
1112    }
1113
1114    fn iterate_orders(&mut self, instrument_id: &InstrumentId) {
1115        // Process bid actions before ask actions so cross-side
1116        // contingencies (OCO/OUO) mutate state between sides
1117        let bid_actions = if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
1118            matching_core.iterate_bids()
1119        } else {
1120            log::error!("Cannot iterate orders: no matching core for instrument {instrument_id}");
1121            return;
1122        };
1123
1124        for action in bid_actions {
1125            self.dispatch_held_match(action, instrument_id);
1126        }
1127
1128        let ask_actions = if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
1129            matching_core.iterate_asks()
1130        } else {
1131            return;
1132        };
1133
1134        for action in ask_actions {
1135            self.dispatch_held_match(action, instrument_id);
1136        }
1137
1138        // Re-snapshot orders after actions to avoid stale trailing stop updates
1139        let orders = if let Some(matching_core) = self.matching_cores.get(instrument_id) {
1140            matching_core.get_orders()
1141        } else {
1142            return;
1143        };
1144
1145        for match_info in orders {
1146            if !matches!(
1147                match_info.order_type,
1148                OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
1149            ) {
1150                continue;
1151            }
1152
1153            let mut order = match self
1154                .cache
1155                .borrow()
1156                .order(&match_info.client_order_id)
1157                .map(|o| o.clone())
1158            {
1159                Some(order) => order,
1160                None => continue,
1161            };
1162
1163            if order.is_closed() {
1164                continue;
1165            }
1166
1167            self.update_trailing_stop_order(&mut order);
1168        }
1169
1170        self.drain_pending_messages();
1171    }
1172
1173    fn dispatch_held_match(&mut self, action: MatchAction, instrument_id: &InstrumentId) {
1174        let client_order_id = match action {
1175            MatchAction::FillLimit(id) | MatchAction::TriggerStop(id) => id,
1176        };
1177
1178        let Some(core) = self.matching_cores.get(instrument_id) else {
1179            return;
1180        };
1181
1182        let Some(order) = core.get_order(client_order_id) else {
1183            return;
1184        };
1185
1186        let Some(action) = core.match_order(order) else {
1187            return;
1188        };
1189
1190        self.dispatch_match_action(action);
1191    }
1192
1193    fn dispatch_match_action(&mut self, action: MatchAction) {
1194        match action {
1195            MatchAction::FillLimit(id) => self.fill_limit_order(id),
1196            MatchAction::TriggerStop(id) => self.trigger_stop_order(id),
1197        }
1198
1199        self.drain_pending_messages();
1200    }
1201
1202    pub fn cancel_order(&mut self, order: &OrderAny) {
1203        log::info!("Canceling order {}", order.client_order_id());
1204
1205        let mut order = order.clone();
1206        order.set_emulation_trigger(None);
1207
1208        let trigger_instrument_id = order
1209            .trigger_instrument_id()
1210            .unwrap_or(order.instrument_id());
1211
1212        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id)
1213            && let Err(e) = matching_core.delete_order(order.client_order_id())
1214        {
1215            log::debug!("Cannot delete order: {e:?}");
1216        }
1217
1218        self.manager
1219            .pop_submit_order_command(order.client_order_id());
1220
1221        self.cache
1222            .borrow_mut()
1223            .update_order_pending_cancel_local(&order);
1224
1225        let ts_now = self.clock.borrow().timestamp_ns();
1226        let event = OrderCanceled::new(
1227            order.trader_id(),
1228            order.strategy_id(),
1229            order.instrument_id(),
1230            order.client_order_id(),
1231            UUID4::new(),
1232            ts_now,
1233            ts_now,
1234            false,
1235            order.venue_order_id(),
1236            order.account_id(),
1237            None,
1238        );
1239
1240        self.apply_order_event(&OrderEventAny::Canceled(event));
1241    }
1242
1243    fn check_monitoring(&mut self, strategy_id: StrategyId, position_id: Option<PositionId>) {
1244        if !self.subscribed_strategies.contains(&strategy_id) {
1245            // Subscribe to strategy order events
1246            if let Some(handler) = &self.on_event_handler {
1247                msgbus::subscribe_order_events(
1248                    format!("events.order.{strategy_id}").into(),
1249                    handler.clone(),
1250                    None,
1251                );
1252                self.subscribed_strategies.insert(strategy_id);
1253                log::info!("Subscribed to strategy {strategy_id} order events");
1254            }
1255        }
1256
1257        if let Some(position_id) = position_id
1258            && !self.monitored_positions.contains(&position_id)
1259        {
1260            self.monitored_positions.insert(position_id);
1261        }
1262    }
1263
1264    /// # Panics
1265    ///
1266    /// Panics if the order type is invalid for a stop order.
1267    pub fn trigger_stop_order(&mut self, client_order_id: ClientOrderId) {
1268        let order = match self
1269            .cache
1270            .borrow()
1271            .order(&client_order_id)
1272            .map(|o| o.clone())
1273        {
1274            Some(order) => order,
1275            None => {
1276                log::error!(
1277                    "Cannot trigger stop order: order {client_order_id} not found in cache"
1278                );
1279                return;
1280            }
1281        };
1282
1283        match order.order_type() {
1284            OrderType::StopLimit | OrderType::LimitIfTouched | OrderType::TrailingStopLimit => {
1285                self.fill_limit_order(client_order_id);
1286            }
1287            OrderType::Market
1288            | OrderType::MarketIfTouched
1289            | OrderType::StopMarket
1290            | OrderType::TrailingStopMarket => self.fill_market_order(client_order_id),
1291            _ => panic!("invalid `OrderType`, was {}", order.order_type()),
1292        }
1293    }
1294
1295    /// Releases an emulated limit-type order as a limit order,
1296    /// or a plain limit order as a market order.
1297    ///
1298    /// Retains the command when market data is unavailable or the limit order cannot be built.
1299    /// Returns without releasing a limit-type order when its command is missing.
1300    ///
1301    /// # Panics
1302    ///
1303    /// Panics if a plain limit order is ready for market release but its command is missing.
1304    pub fn fill_limit_order(&mut self, client_order_id: ClientOrderId) {
1305        self.release_order(client_order_id, ReleaseType::Limit);
1306    }
1307
1308    /// Releases an emulated order as a market order, converting GTD to GTC.
1309    ///
1310    /// Retains the command when market data is unavailable.
1311    ///
1312    /// # Panics
1313    ///
1314    /// Panics if an order is ready for market release but its command is missing.
1315    pub fn fill_market_order(&mut self, client_order_id: ClientOrderId) {
1316        self.release_order(client_order_id, ReleaseType::Market);
1317    }
1318
1319    fn release_order(&mut self, client_order_id: ClientOrderId, release_type: ReleaseType) {
1320        let kind = match release_type {
1321            ReleaseType::Limit => "limit",
1322            ReleaseType::Market => "market",
1323        };
1324
1325        let Some(mut order) = self.cache.borrow().order_owned(&client_order_id) else {
1326            log::error!("Cannot fill {kind} order: order {client_order_id} not found in cache");
1327            return;
1328        };
1329
1330        if matches!(
1331            (release_type, order.order_type()),
1332            (ReleaseType::Limit, OrderType::Limit)
1333        ) {
1334            self.fill_market_order(client_order_id);
1335            return;
1336        }
1337
1338        let trigger_instrument_id = order
1339            .trigger_instrument_id()
1340            .unwrap_or(order.instrument_id());
1341
1342        let Some(matching_core) = self.matching_cores.get(&trigger_instrument_id) else {
1343            log::error!(
1344                "Cannot fill {kind} order: no matching core for instrument {trigger_instrument_id}"
1345            );
1346            return;
1347        };
1348
1349        let Some(released_price) =
1350            self.validate_release(&order, matching_core, trigger_instrument_id)
1351        else {
1352            return;
1353        };
1354
1355        // Build before taking the command, so a failed transform leaves the order emulated
1356        let limit_order = match release_type {
1357            ReleaseType::Limit => {
1358                let Some(transformed) = self.transform_limit_order(&order, trigger_instrument_id)
1359                else {
1360                    return;
1361                };
1362
1363                Some(transformed)
1364            }
1365            ReleaseType::Market => None,
1366        };
1367
1368        let command = self.manager.pop_submit_order_command(client_order_id);
1369
1370        let command = match release_type {
1371            ReleaseType::Limit => {
1372                let Some(command) = command else {
1373                    return;
1374                };
1375
1376                command
1377            }
1378            ReleaseType::Market => {
1379                command.expect("invalid operation `fill_market_order` with no command")
1380            }
1381        };
1382
1383        let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) else {
1384            return;
1385        };
1386
1387        if let Err(e) = matching_core.delete_order(client_order_id) {
1388            match release_type {
1389                ReleaseType::Limit => log::debug!("Error deleting order: {e:?}"),
1390                ReleaseType::Market => log::debug!("Cannot delete order: {e:?}"),
1391            }
1392        }
1393
1394        let transformed = match limit_order {
1395            Some(mut transformed) => {
1396                transformed.liquidity_side = order.liquidity_side();
1397
1398                // TODO: fix
1399                // let triggered_price = order.trigger_price();
1400                // if triggered_price.is_some() {
1401                //     transformed.trigger_price() = (triggered_price.unwrap());
1402                // }
1403
1404                transformed.prepend_events(order.events().into_iter().cloned());
1405                OrderAny::Limit(transformed)
1406            }
1407            None => {
1408                order.set_emulation_trigger(None);
1409                let mut transformed = self.transform_market_order(&order);
1410                transformed.prepend_events(order.events().into_iter().cloned());
1411                OrderAny::Market(transformed)
1412            }
1413        };
1414
1415        self.submit_released_order(&order, &transformed, command, released_price, release_type);
1416    }
1417
1418    fn validate_release(
1419        &self,
1420        order: &OrderAny,
1421        matching_core: &OrderMatchingCore,
1422        trigger_instrument_id: InstrumentId,
1423    ) -> Option<Price> {
1424        let released_price = match order.order_side() {
1425            OrderSide::Buy => matching_core.ask,
1426            OrderSide::Sell => matching_core.bid,
1427        };
1428
1429        if released_price.is_none() {
1430            log::warn!(
1431                "Cannot release order {} yet: no market data available for {trigger_instrument_id}, will retry on next update",
1432                order.client_order_id(),
1433            );
1434            return None;
1435        }
1436
1437        released_price
1438    }
1439
1440    fn transform_limit_order(
1441        &self,
1442        order: &OrderAny,
1443        trigger_instrument_id: InstrumentId,
1444    ) -> Option<LimitOrder> {
1445        let client_order_id = order.client_order_id();
1446
1447        // A trailing stop limit has no limit price until its trailing calculation succeeds
1448        let Some(price) = order.price() else {
1449            log::warn!(
1450                "Cannot release order {client_order_id} yet: no limit price, will retry on next update"
1451            );
1452            return None;
1453        };
1454
1455        let transformed = match LimitOrder::new_checked(
1456            order.trader_id(),
1457            order.strategy_id(),
1458            order.instrument_id(),
1459            order.client_order_id(),
1460            order.order_side(),
1461            order.quantity(),
1462            price,
1463            order.time_in_force(),
1464            order.expire_time(),
1465            order.is_post_only(),
1466            order.is_reduce_only(),
1467            order.is_quote_quantity(),
1468            order.display_qty(),
1469            None,
1470            Some(trigger_instrument_id),
1471            order.contingency_type(),
1472            order.order_list_id(),
1473            order.linked_order_ids().map(Vec::from),
1474            order.parent_order_id(),
1475            order.exec_algorithm_id(),
1476            order.exec_algorithm_params().cloned(),
1477            order.exec_spawn_id(),
1478            order.tags().map(Vec::from),
1479            UUID4::new(),
1480            self.clock.borrow().timestamp_ns(),
1481        ) {
1482            Ok(transformed) => transformed,
1483            Err(e) => {
1484                log::error!("Cannot release order {client_order_id} as a limit order: {e}");
1485                return None;
1486            }
1487        };
1488
1489        Some(transformed)
1490    }
1491
1492    fn transform_market_order(&self, order: &OrderAny) -> MarketOrder {
1493        // Market orders reject GTD, so a released GTD order becomes GTC
1494        let time_in_force = match order.time_in_force() {
1495            TimeInForce::Gtd => TimeInForce::Gtc,
1496            time_in_force => time_in_force,
1497        };
1498
1499        MarketOrder::new(
1500            order.trader_id(),
1501            order.strategy_id(),
1502            order.instrument_id(),
1503            order.client_order_id(),
1504            order.order_side(),
1505            order.quantity(),
1506            time_in_force,
1507            UUID4::new(),
1508            self.clock.borrow().timestamp_ns(),
1509            order.is_reduce_only(),
1510            order.is_quote_quantity(),
1511            order.contingency_type(),
1512            order.order_list_id(),
1513            order.linked_order_ids().map(Vec::from),
1514            order.parent_order_id(),
1515            order.exec_algorithm_id(),
1516            order.exec_algorithm_params().cloned(),
1517            order.exec_spawn_id(),
1518            order.tags().map(Vec::from),
1519        )
1520    }
1521
1522    fn submit_released_order(
1523        &self,
1524        order: &OrderAny,
1525        transformed: &OrderAny,
1526        mut command: SubmitOrder,
1527        released_price: Price,
1528        release_type: ReleaseType,
1529    ) {
1530        let client_order_id = order.client_order_id();
1531        command.order_init = OrderInitialized::from(transformed);
1532        let add_result = self.cache.borrow_mut().add_order(
1533            transformed.clone(),
1534            command.position_id,
1535            command.client_id,
1536            true,
1537        );
1538
1539        if let Err(e) = add_result {
1540            log::error!("Failed to add order: {e}");
1541        } else {
1542            let topic = get_event_order_topic(order.strategy_id());
1543            msgbus::publish_order_event(topic, transformed.last_event());
1544        }
1545
1546        // Preserve the release branches' distinct clock-read and event-ID ordering
1547        let (event_id, ts_event, ts_init) = match release_type {
1548            ReleaseType::Limit => (
1549                UUID4::new(),
1550                self.clock.borrow().timestamp_ns(),
1551                self.clock.borrow().timestamp_ns(),
1552            ),
1553            ReleaseType::Market => {
1554                let ts_now = self.clock.borrow().timestamp_ns();
1555                (UUID4::new(), ts_now, ts_now)
1556            }
1557        };
1558
1559        let event = OrderEventAny::Released(OrderReleased::new(
1560            order.trader_id(),
1561            order.strategy_id(),
1562            order.instrument_id(),
1563            client_order_id,
1564            released_price,
1565            event_id,
1566            ts_event,
1567            ts_init,
1568        ));
1569
1570        let transformed = match self.cache.borrow_mut().update_order(&event) {
1571            Ok(order) => order,
1572            Err(e) => {
1573                log::error!("Failed to apply order event: {e}");
1574                return;
1575            }
1576        };
1577
1578        self.send_risk_event(event.clone());
1579
1580        log::info!("Releasing order {client_order_id}");
1581
1582        let strategy_id = match release_type {
1583            ReleaseType::Limit => transformed.strategy_id(),
1584            ReleaseType::Market => order.strategy_id(),
1585        };
1586
1587        let topic = get_event_order_topic(strategy_id);
1588        msgbus::publish_order_event(topic, &event);
1589
1590        if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
1591            self.send_algo_command(command, exec_algorithm_id);
1592        } else {
1593            self.send_risk_command(TradingCommand::SubmitOrder(command));
1594        }
1595    }
1596
1597    fn update_trailing_stop_order(&mut self, order: &mut OrderAny) {
1598        let trigger_instrument_id = order
1599            .trigger_instrument_id()
1600            .unwrap_or_else(|| order.instrument_id());
1601
1602        let Some(matching_core) = self.matching_cores.get(&trigger_instrument_id) else {
1603            log::error!(
1604                "Cannot update trailing-stop order: no matching core for instrument {trigger_instrument_id}"
1605            );
1606            return;
1607        };
1608
1609        let mut bid = matching_core.bid;
1610        let mut ask = matching_core.ask;
1611        let mut last = matching_core.last;
1612        let price_increment = matching_core.price_increment;
1613        let instrument_id = matching_core.instrument_id;
1614        if bid.is_none() || ask.is_none() || last.is_none() {
1615            if let Some(q) = self.cache.borrow().quote(&instrument_id) {
1616                bid.get_or_insert(q.bid_price);
1617                ask.get_or_insert(q.ask_price);
1618            }
1619
1620            if let Some(t) = self.cache.borrow().trade(&instrument_id) {
1621                last.get_or_insert(t.price);
1622            }
1623        }
1624
1625        let was_activated = is_order_activated(order);
1626
1627        if !self.maybe_activate_trailing_stop(order, bid, ask, last) {
1628            return; // Not yet activated
1629        }
1630
1631        if !was_activated {
1632            // Activation transitioned: re-key the held entry from its inert
1633            // pre-activation state regardless of the trailing outcome below.
1634            self.refresh_matching_core_entry(order, trigger_instrument_id);
1635        }
1636
1637        let (new_trigger_px, new_limit_px) = match trailing_stop_calculate(
1638            price_increment,
1639            order.trigger_price(),
1640            order,
1641            bid,
1642            ask,
1643            last,
1644        ) {
1645            Ok(pair) => pair,
1646            Err(e) => {
1647                log::warn!("Cannot calculate trailing-stop update: {e}");
1648                return;
1649            }
1650        };
1651
1652        if new_trigger_px.is_none() && new_limit_px.is_none() {
1653            return;
1654        }
1655
1656        let ts_now = self.clock.borrow().timestamp_ns();
1657        let update = OrderUpdated::new(
1658            order.trader_id(),
1659            order.strategy_id(),
1660            order.instrument_id(),
1661            order.client_order_id(),
1662            order.quantity(),
1663            UUID4::new(),
1664            ts_now,
1665            ts_now,
1666            false,
1667            order.venue_order_id(),
1668            order.account_id(),
1669            new_limit_px,
1670            new_trigger_px,
1671            None,
1672            order.is_quote_quantity(),
1673        );
1674        let wrapped = OrderEventAny::Updated(update);
1675
1676        *order = match self.cache.borrow_mut().update_order(&wrapped) {
1677            Ok(order) => order,
1678            Err(e) => {
1679                log::error!("Failed to apply order event: {e}");
1680                return;
1681            }
1682        };
1683
1684        self.refresh_matching_core_entry(order, trigger_instrument_id);
1685
1686        self.send_risk_event(wrapped);
1687    }
1688
1689    fn refresh_matching_core_entry(
1690        &mut self,
1691        order: &OrderAny,
1692        trigger_instrument_id: InstrumentId,
1693    ) {
1694        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) {
1695            if let Err(e) = matching_core.delete_order(order.client_order_id()) {
1696                log::debug!("Cannot update trailing-stop match info: {e:?}");
1697            }
1698
1699            // A trailing stop with no trigger price yet must stay inert:
1700            // its limit price must never match as a plain limit order.
1701            let (trigger_price, limit_price) = if order.trigger_price().is_some() {
1702                (order.trigger_price(), order.price())
1703            } else {
1704                (None, None)
1705            };
1706
1707            matching_core.add_order(RestingOrder::new_with_trigger_type(
1708                order.client_order_id(),
1709                order.order_side(),
1710                order.order_type(),
1711                order.emulation_trigger(),
1712                trigger_price,
1713                limit_price,
1714                is_order_activated(order),
1715            ));
1716        }
1717    }
1718
1719    /// Returns true if the trailing stop is (or has just become) activated.
1720    ///
1721    /// With no `activation_price` the order activates immediately at the current
1722    /// market price; otherwise it activates only once the market touches the
1723    /// activation price.
1724    fn maybe_activate_trailing_stop(
1725        &self,
1726        order: &mut OrderAny,
1727        bid: Option<Price>,
1728        ask: Option<Price>,
1729        last: Option<Price>,
1730    ) -> bool {
1731        let (is_activated, activation_price, trigger_type, order_side) = match order {
1732            OrderAny::TrailingStopMarket(inner) => (
1733                inner.is_activated,
1734                inner.activation_price,
1735                inner.trigger_type,
1736                inner.order_side(),
1737            ),
1738            OrderAny::TrailingStopLimit(inner) => (
1739                inner.is_activated,
1740                inner.activation_price,
1741                inner.trigger_type,
1742                inner.order_side(),
1743            ),
1744            _ => return true,
1745        };
1746
1747        if is_activated {
1748            return true;
1749        }
1750
1751        if let Some(activation_price) = activation_price {
1752            let hit = match order_side {
1753                OrderSide::Buy => ask.is_some_and(|a| a <= activation_price),
1754                OrderSide::Sell => bid.is_some_and(|b| b >= activation_price),
1755            };
1756
1757            if hit {
1758                Self::set_trailing_stop_activated(order, None);
1759                store_trailing_stop_activation(&mut self.cache.borrow_mut(), order);
1760            }
1761            return hit;
1762        }
1763
1764        let market_price = match trigger_type {
1765            TriggerType::LastPrice => last,
1766            _ => match order_side {
1767                OrderSide::Buy => ask,
1768                OrderSide::Sell => bid,
1769            },
1770        };
1771
1772        let Some(market_price) = market_price else {
1773            log::error!(
1774                "Cannot activate trailing stop {}: no market price available",
1775                order.client_order_id()
1776            );
1777            return false;
1778        };
1779
1780        Self::set_trailing_stop_activated(order, Some(market_price));
1781        store_trailing_stop_activation(&mut self.cache.borrow_mut(), order);
1782        true
1783    }
1784
1785    fn set_trailing_stop_activated(order: &mut OrderAny, activation_price: Option<Price>) {
1786        match order {
1787            OrderAny::TrailingStopMarket(inner) => {
1788                if let Some(price) = activation_price {
1789                    inner.activation_price = Some(price);
1790                }
1791                inner.set_activated();
1792            }
1793            OrderAny::TrailingStopLimit(inner) => {
1794                if let Some(price) = activation_price {
1795                    inner.activation_price = Some(price);
1796                }
1797                inner.set_activated();
1798            }
1799            _ => {}
1800        }
1801    }
1802
1803    fn send_algo_command(&self, command: SubmitOrder, exec_algorithm_id: ExecAlgorithmId) {
1804        let id = command.strategy_id;
1805        log::info!("{id} {CMD}{SEND} {command}");
1806
1807        let endpoint = format!("{exec_algorithm_id}.execute");
1808        msgbus::send_any(endpoint.into(), &TradingCommand::SubmitOrder(command));
1809    }
1810
1811    fn send_risk_command(&self, command: TradingCommand) {
1812        log_cmd_send(&command);
1813        let endpoint = MessagingSwitchboard::risk_engine_queue_execute();
1814        msgbus::send_trading_command(endpoint, command);
1815    }
1816
1817    fn send_exec_command(&self, command: TradingCommand) {
1818        log_cmd_send(&command);
1819        let endpoint = MessagingSwitchboard::exec_engine_queue_execute();
1820        msgbus::send_trading_command(endpoint, command);
1821    }
1822
1823    fn send_risk_event(&self, event: OrderEventAny) {
1824        log_evt_send(&event);
1825        let endpoint = MessagingSwitchboard::risk_engine_process();
1826        msgbus::send_order_event(endpoint, event);
1827    }
1828
1829    // Applies an event the emulator generated, then notifies the portfolio and the strategy
1830    fn apply_order_event(&self, event: &OrderEventAny) {
1831        if let Err(e) = self.cache.borrow_mut().update_order(event) {
1832            log::error!("Failed to apply order event: {e}");
1833            return;
1834        }
1835
1836        self.send_portfolio_order_event(event.clone());
1837        publish_order_event(event);
1838    }
1839
1840    fn send_portfolio_order_event(&self, event: OrderEventAny) {
1841        log_evt_send(&event);
1842        let endpoint = MessagingSwitchboard::portfolio_update_order();
1843        msgbus::send_order_event(endpoint, event);
1844    }
1845
1846    fn send_data_command(&self, command: DataCommand) {
1847        log::info!("{CMD}{SEND} {command:?}");
1848        let endpoint = MessagingSwitchboard::data_engine_queue_execute();
1849        msgbus::send_data_command(endpoint, command);
1850    }
1851}
1852
1853#[derive(Clone, Copy)]
1854enum ReleaseType {
1855    Limit,
1856    Market,
1857}
1858
1859fn publish_order_event(event: &OrderEventAny) {
1860    msgbus::publish_order_event(get_event_order_topic(event.strategy_id()), event);
1861
1862    if let OrderEventAny::Canceled(_) = event {
1863        msgbus::publish_order_event(get_order_canceled_topic(event.instrument_id()), event);
1864    }
1865}
1866
1867fn is_order_activated(order: &OrderAny) -> bool {
1868    match order {
1869        OrderAny::TrailingStopMarket(o) => o.is_activated,
1870        OrderAny::TrailingStopLimit(o) => o.is_activated,
1871        _ => true,
1872    }
1873}
1874
1875fn log_cmd_send(command: &TradingCommand) {
1876    if let Some(id) = command.strategy_id() {
1877        log::info!("{id} {CMD}{SEND} {command}");
1878    } else {
1879        log::info!("{CMD}{SEND} {command}");
1880    }
1881}
1882
1883fn log_evt_send(event: &OrderEventAny) {
1884    let id = event.strategy_id();
1885    log::info!("{id} {EVT}{SEND} {event}");
1886}
1887
1888#[cfg(test)]
1889mod tests {
1890    use std::{cell::RefCell, rc::Rc};
1891
1892    use nautilus_common::{
1893        cache::Cache,
1894        clock::VirtualClock,
1895        messages::data::{DataCommand, SubscribeCommand, UnsubscribeCommand},
1896        msgbus::{
1897            MessagingSwitchboard,
1898            stubs::{
1899                TypedIntoMessageSavingHandler, get_any_saving_handler,
1900                get_typed_into_message_saving_handler,
1901            },
1902        },
1903    };
1904    use nautilus_core::{Params, UUID4, UnixNanos};
1905    use nautilus_model::{
1906        data::{QuoteTick, TradeTick},
1907        enums::{
1908            AggressorSide, OrderSide, OrderType, TimeInForce, TrailingOffsetType, TriggerType,
1909        },
1910        identifiers::{
1911            AccountId, ClientId, ClientOrderId, OrderListId, StrategyId, Symbol, TradeId, TraderId,
1912        },
1913        instruments::{
1914            CryptoPerpetual, Instrument, InstrumentAny, SyntheticInstrument,
1915            stubs::{btcusd_bybit, crypto_perpetual_ethusdt},
1916        },
1917        orders::{OrderList, OrderTestBuilder, stubs::TestOrderEventStubs},
1918        types::{Price, Quantity},
1919    };
1920    use rstest::{fixture, rstest};
1921    use rust_decimal_macros::dec;
1922    use ustr::Ustr;
1923
1924    use super::*;
1925
1926    #[fixture]
1927    fn instrument() -> CryptoPerpetual {
1928        crypto_perpetual_ethusdt()
1929    }
1930
1931    #[expect(clippy::type_complexity)]
1932    fn create_emulator() -> (
1933        Rc<RefCell<dyn Clock>>,
1934        Rc<RefCell<Cache>>,
1935        Rc<RefCell<OrderEmulator>>,
1936    ) {
1937        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
1938        let cache = Rc::new(RefCell::new(Cache::new(None, None)));
1939        let emulator = Rc::new(RefCell::new(OrderEmulator::new(
1940            clock.clone(),
1941            cache.clone(),
1942        )));
1943
1944        OrderEmulator::register_msgbus_handlers(&emulator);
1945
1946        (clock, cache, emulator)
1947    }
1948
1949    fn create_stop_market_order(instrument: &CryptoPerpetual, trigger: TriggerType) -> OrderAny {
1950        OrderTestBuilder::new(OrderType::StopMarket)
1951            .instrument_id(instrument.id())
1952            .side(OrderSide::Buy)
1953            .trigger_price(Price::from("5100.00"))
1954            .quantity(Quantity::from(1))
1955            .emulation_trigger(trigger)
1956            .build()
1957    }
1958
1959    fn create_stop_limit_order(instrument: &CryptoPerpetual, trigger: TriggerType) -> OrderAny {
1960        OrderTestBuilder::new(OrderType::StopLimit)
1961            .instrument_id(instrument.id())
1962            .side(OrderSide::Buy)
1963            .price(Price::from("5100.00"))
1964            .trigger_price(Price::from("5100.00"))
1965            .quantity(Quantity::from(1))
1966            .emulation_trigger(trigger)
1967            .build()
1968    }
1969
1970    fn create_list_stop_market_order(
1971        instrument: &CryptoPerpetual,
1972        client_order_id: &str,
1973        order_list_id: OrderListId,
1974    ) -> OrderAny {
1975        OrderTestBuilder::new(OrderType::StopMarket)
1976            .instrument_id(instrument.id())
1977            .client_order_id(ClientOrderId::from(client_order_id))
1978            .order_list_id(order_list_id)
1979            .side(OrderSide::Buy)
1980            .trigger_price(Price::from("5100.00"))
1981            .quantity(Quantity::from(1))
1982            .emulation_trigger(TriggerType::BidAsk)
1983            .build()
1984    }
1985
1986    fn create_submit_order(instrument: &CryptoPerpetual, order: &OrderAny) -> SubmitOrder {
1987        SubmitOrder::new(
1988            TraderId::from("TRADER-001"),
1989            None,
1990            StrategyId::from("STRATEGY-001"),
1991            instrument.id(),
1992            order.client_order_id(),
1993            order.init_event().clone(),
1994            None,
1995            None,
1996            None,
1997            UUID4::new(),
1998            0.into(),
1999            None, // correlation_id
2000        )
2001    }
2002
2003    fn create_quote_tick(instrument: &CryptoPerpetual, bid: &str, ask: &str) -> QuoteTick {
2004        QuoteTick::new(
2005            instrument.id(),
2006            Price::from(bid),
2007            Price::from(ask),
2008            Quantity::from(10),
2009            Quantity::from(10),
2010            0.into(),
2011            0.into(),
2012        )
2013    }
2014
2015    fn create_trade_tick(instrument: &CryptoPerpetual, price: &str) -> TradeTick {
2016        TradeTick::new(
2017            instrument.id(),
2018            Price::from(price),
2019            Quantity::from(1),
2020            AggressorSide::Buy,
2021            TradeId::from("T-001"),
2022            0.into(),
2023            0.into(),
2024        )
2025    }
2026
2027    fn add_instrument_to_cache(cache: &Rc<RefCell<Cache>>, instrument: &CryptoPerpetual) {
2028        cache
2029            .borrow_mut()
2030            .add_instrument(InstrumentAny::CryptoPerpetual(instrument.clone()))
2031            .unwrap();
2032    }
2033
2034    fn register_risk_event_handler(id: &str) -> TypedIntoMessageSavingHandler<OrderEventAny> {
2035        let (handler, saving_handler) =
2036            get_typed_into_message_saving_handler::<OrderEventAny>(Some(Ustr::from(id)));
2037        msgbus::register_order_event_endpoint(MessagingSwitchboard::risk_engine_process(), handler);
2038        saving_handler
2039    }
2040
2041    fn register_exec_event_handler(
2042        cache: Rc<RefCell<Cache>>,
2043        id: &str,
2044    ) -> TypedIntoMessageSavingHandler<OrderEventAny> {
2045        let messages = Rc::new(RefCell::new(Vec::new()));
2046        let messages_for_handler = messages.clone();
2047        msgbus::register_order_event_endpoint(
2048            MessagingSwitchboard::exec_engine_process(),
2049            TypedIntoHandler::from(move |event: OrderEventAny| {
2050                cache.borrow_mut().update_order(&event).unwrap();
2051                messages_for_handler.borrow_mut().push(event);
2052            }),
2053        );
2054        TypedIntoMessageSavingHandler::new_with_messages(Some(Ustr::from(id)), messages)
2055    }
2056
2057    fn register_portfolio_event_handler(id: &str) -> TypedIntoMessageSavingHandler<OrderEventAny> {
2058        let (handler, saving_handler) =
2059            get_typed_into_message_saving_handler::<OrderEventAny>(Some(Ustr::from(id)));
2060        msgbus::register_order_event_endpoint(
2061            MessagingSwitchboard::portfolio_update_order(),
2062            handler,
2063        );
2064        saving_handler
2065    }
2066
2067    fn register_data_command_handler(id: &str) -> TypedIntoMessageSavingHandler<DataCommand> {
2068        let (handler, saving_handler) =
2069            get_typed_into_message_saving_handler::<DataCommand>(Some(Ustr::from(id)));
2070        msgbus::register_data_command_endpoint(
2071            MessagingSwitchboard::data_engine_queue_execute(),
2072            handler,
2073        );
2074        saving_handler
2075    }
2076
2077    fn subscribe_order_topic(
2078        strategy_id: StrategyId,
2079    ) -> (TypedHandler<OrderEventAny>, Rc<RefCell<Vec<OrderEventAny>>>) {
2080        let events = Rc::new(RefCell::new(Vec::new()));
2081        let handler = TypedHandler::from({
2082            let events = events.clone();
2083            move |event: &OrderEventAny| {
2084                events.borrow_mut().push(event.clone());
2085            }
2086        });
2087        msgbus::subscribe_order_events(
2088            format!("events.order.{strategy_id}").into(),
2089            handler.clone(),
2090            None,
2091        );
2092        (handler, events)
2093    }
2094
2095    fn subscribe_order_cancel_topic(
2096        instrument_id: InstrumentId,
2097    ) -> (TypedHandler<OrderEventAny>, Rc<RefCell<Vec<OrderEventAny>>>) {
2098        let events = Rc::new(RefCell::new(Vec::new()));
2099        let handler = TypedHandler::from({
2100            let events = events.clone();
2101            move |event: &OrderEventAny| {
2102                events.borrow_mut().push(event.clone());
2103            }
2104        });
2105        msgbus::subscribe_order_events(
2106            get_order_canceled_topic(instrument_id).into(),
2107            handler.clone(),
2108            None,
2109        );
2110        (handler, events)
2111    }
2112
2113    #[rstest]
2114    fn test_dispatch_manager_publish_initialized_publishes_order_event(
2115        instrument: CryptoPerpetual,
2116    ) {
2117        let (_clock, _cache, emulator) = create_emulator();
2118        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2119        let strategy_id = order.strategy_id();
2120        let client_order_id = order.client_order_id();
2121        let event = OrderEventAny::Initialized(order.init_event().clone());
2122        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
2123
2124        emulator
2125            .borrow_mut()
2126            .dispatch_manager_action(OrderManagerAction::PublishInitialized(event));
2127        msgbus::unsubscribe_order_events(
2128            format!("events.order.{strategy_id}").into(),
2129            &order_handler,
2130        );
2131        let order_events = order_events.borrow();
2132
2133        assert_eq!(order_events.len(), 1);
2134        assert!(matches!(
2135            &order_events[0],
2136            OrderEventAny::Initialized(event) if event.client_order_id == client_order_id
2137        ));
2138    }
2139
2140    #[rstest]
2141    fn test_dispatch_manager_submit_to_emulator_applies_emulated_event(
2142        instrument: CryptoPerpetual,
2143    ) {
2144        let (_clock, cache, emulator) = create_emulator();
2145        let risk_events = register_risk_event_handler("RiskEngine.process.dispatch_emulated");
2146        add_instrument_to_cache(&cache, &instrument);
2147        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2148        let client_order_id = order.client_order_id();
2149        let command = create_submit_order(&instrument, &order);
2150        cache
2151            .borrow_mut()
2152            .add_order(order, None, None, false)
2153            .unwrap();
2154        emulator
2155            .borrow_mut()
2156            .cache_submit_order_command(command.clone());
2157
2158        emulator
2159            .borrow_mut()
2160            .dispatch_manager_action(OrderManagerAction::SubmitToEmulator(command));
2161        let cache = cache.borrow();
2162        let cached_order = cache.order(&client_order_id).unwrap();
2163        let risk_events = risk_events.get_messages();
2164
2165        assert_eq!(cached_order.status(), OrderStatus::Emulated);
2166        assert_eq!(risk_events.len(), 1);
2167        assert!(matches!(risk_events[0], OrderEventAny::Emulated(_)));
2168    }
2169
2170    #[rstest]
2171    fn test_registered_execute_endpoint_routes_submit_order(instrument: CryptoPerpetual) {
2172        let (_clock, cache, emulator) = create_emulator();
2173        let risk_events = register_risk_event_handler("RiskEngine.process.endpoint_emulated");
2174        let data_commands = register_data_command_handler("DataEngine.queue_execute.endpoint");
2175        add_instrument_to_cache(&cache, &instrument);
2176        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2177        let client_order_id = order.client_order_id();
2178        let command = create_submit_order(&instrument, &order);
2179        cache
2180            .borrow_mut()
2181            .add_order(order, None, None, false)
2182            .unwrap();
2183        emulator
2184            .borrow_mut()
2185            .cache_submit_order_command(command.clone());
2186
2187        msgbus::send_trading_command(
2188            MessagingSwitchboard::order_emulator_execute(),
2189            TradingCommand::SubmitOrder(command),
2190        );
2191        let cache = cache.borrow();
2192        let cached_order = cache.order(&client_order_id).unwrap();
2193        let risk_events = risk_events.get_messages();
2194        let data_commands = data_commands.get_messages();
2195
2196        assert_eq!(cached_order.status(), OrderStatus::Emulated);
2197        assert_eq!(risk_events.len(), 1);
2198        assert!(matches!(risk_events[0], OrderEventAny::Emulated(_)));
2199        assert_eq!(data_commands.len(), 1);
2200        assert!(matches!(
2201            &data_commands[0],
2202            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
2203                if command.instrument_ids == [instrument.id()]
2204        ));
2205    }
2206
2207    #[rstest]
2208    fn test_cancel_all_orders_filters_emulated_orders_by_client(instrument: CryptoPerpetual) {
2209        let (_clock, cache, emulator) = create_emulator();
2210        let _risk_events = register_risk_event_handler("RiskEngine.process.cancel_all_client");
2211        let portfolio_events =
2212            register_portfolio_event_handler("Portfolio.update_order.cancel_all_client");
2213        let _data_commands = register_data_command_handler("DataEngine.queue_execute.cancel_all");
2214        add_instrument_to_cache(&cache, &instrument);
2215
2216        let selected_client = ClientId::from("CLIENT-001");
2217        let other_client = ClientId::from("CLIENT-002");
2218        let make_order = |client_order_id| {
2219            OrderTestBuilder::new(OrderType::StopMarket)
2220                .instrument_id(instrument.id())
2221                .client_order_id(client_order_id)
2222                .side(OrderSide::Buy)
2223                .trigger_price(Price::from("5100.00"))
2224                .quantity(Quantity::from(1))
2225                .emulation_trigger(TriggerType::BidAsk)
2226                .build()
2227        };
2228        let selected_order = make_order(ClientOrderId::from("O-EMULATED-SELECTED"));
2229        let other_order = make_order(ClientOrderId::from("O-EMULATED-OTHER"));
2230        let unclaimed_order = make_order(ClientOrderId::from("O-EMULATED-UNCLAIMED"));
2231        let synthetic_formula = format!("{} * 1.0", instrument.id());
2232        let synthetic = SyntheticInstrument::builder()
2233            .symbol(Symbol::from("ETH-INDEX"))
2234            .price_precision(instrument.price_precision())
2235            .components(vec![instrument.id()])
2236            .formula(&synthetic_formula)
2237            .ts_event(0.into())
2238            .ts_init(0.into())
2239            .build()
2240            .unwrap();
2241        let trigger_instrument_id = synthetic.id;
2242        let cross_trigger_order = OrderTestBuilder::new(OrderType::StopMarket)
2243            .instrument_id(instrument.id())
2244            .client_order_id(ClientOrderId::from("O-EMULATED-CROSS-TRIGGER"))
2245            .side(OrderSide::Buy)
2246            .trigger_price(Price::from("5100.00"))
2247            .quantity(Quantity::from(1))
2248            .emulation_trigger(TriggerType::BidAsk)
2249            .trigger_instrument_id(trigger_instrument_id)
2250            .build();
2251        let mut selected_submit = create_submit_order(&instrument, &selected_order);
2252        selected_submit.client_id = Some(selected_client);
2253        let mut other_submit = create_submit_order(&instrument, &other_order);
2254        other_submit.client_id = Some(other_client);
2255        let unclaimed_submit = create_submit_order(&instrument, &unclaimed_order);
2256        let mut cross_trigger_submit = create_submit_order(&instrument, &cross_trigger_order);
2257        cross_trigger_submit.client_id = Some(selected_client);
2258        {
2259            let mut cache = cache.borrow_mut();
2260            cache.add_synthetic(synthetic).unwrap();
2261            cache
2262                .add_order(selected_order.clone(), None, Some(selected_client), false)
2263                .unwrap();
2264            cache
2265                .add_order(other_order.clone(), None, Some(other_client), false)
2266                .unwrap();
2267            cache
2268                .add_order(unclaimed_order.clone(), None, None, false)
2269                .unwrap();
2270            cache
2271                .add_order(
2272                    cross_trigger_order.clone(),
2273                    None,
2274                    Some(selected_client),
2275                    false,
2276                )
2277                .unwrap();
2278        }
2279        emulator.borrow_mut().handle_submit_order(&selected_submit);
2280        emulator.borrow_mut().handle_submit_order(&other_submit);
2281        emulator.borrow_mut().handle_submit_order(&unclaimed_submit);
2282        emulator
2283            .borrow_mut()
2284            .handle_submit_order(&cross_trigger_submit);
2285
2286        emulator
2287            .borrow_mut()
2288            .execute(TradingCommand::CancelAllOrders(CancelAllOrders::new(
2289                TraderId::from("TRADER-001"),
2290                Some(selected_client),
2291                StrategyId::from("CALLER-001"),
2292                trigger_instrument_id,
2293                None,
2294                UUID4::new(),
2295                0.into(),
2296                None,
2297                None,
2298            )));
2299
2300        assert_eq!(
2301            cache
2302                .borrow()
2303                .order(&cross_trigger_order.client_order_id())
2304                .unwrap()
2305                .status(),
2306            OrderStatus::Emulated
2307        );
2308
2309        emulator
2310            .borrow_mut()
2311            .execute(TradingCommand::CancelAllOrders(CancelAllOrders::new(
2312                TraderId::from("TRADER-001"),
2313                Some(selected_client),
2314                StrategyId::from("CALLER-001"),
2315                instrument.id(),
2316                None,
2317                UUID4::new(),
2318                0.into(),
2319                None,
2320                None,
2321            )));
2322
2323        let cache = cache.borrow();
2324        assert_eq!(
2325            cache
2326                .order(&selected_order.client_order_id())
2327                .unwrap()
2328                .status(),
2329            OrderStatus::Canceled
2330        );
2331        assert_eq!(
2332            cache
2333                .order(&other_order.client_order_id())
2334                .unwrap()
2335                .status(),
2336            OrderStatus::Emulated
2337        );
2338        assert_eq!(
2339            cache
2340                .order(&unclaimed_order.client_order_id())
2341                .unwrap()
2342                .status(),
2343            OrderStatus::Emulated
2344        );
2345        assert_eq!(
2346            cache
2347                .order(&cross_trigger_order.client_order_id())
2348                .unwrap()
2349                .status(),
2350            OrderStatus::Canceled
2351        );
2352        let portfolio_events = portfolio_events.get_messages();
2353        assert_eq!(portfolio_events.len(), 2);
2354        let canceled_ids: AHashSet<_> = portfolio_events
2355            .iter()
2356            .filter_map(|event| match event {
2357                OrderEventAny::Canceled(event) => Some(event.client_order_id),
2358                _ => None,
2359            })
2360            .collect();
2361        assert_eq!(
2362            canceled_ids,
2363            AHashSet::from_iter([
2364                selected_order.client_order_id(),
2365                cross_trigger_order.client_order_id(),
2366            ])
2367        );
2368    }
2369
2370    #[rstest]
2371    fn test_dispatch_manager_submit_to_risk_uses_risk_queue(instrument: CryptoPerpetual) {
2372        let (_clock, _cache, emulator) = create_emulator();
2373        let (handler, messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
2374            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
2375        msgbus::register_trading_command_endpoint(
2376            MessagingSwitchboard::risk_engine_queue_execute(),
2377            handler,
2378        );
2379        let order = create_stop_market_order(&instrument, TriggerType::Default);
2380        let command = create_submit_order(&instrument, &order);
2381        let client_order_id = command.client_order_id;
2382
2383        emulator
2384            .borrow_mut()
2385            .dispatch_manager_action(OrderManagerAction::SubmitToRisk(command));
2386
2387        let messages = messages.get_messages();
2388        assert_eq!(messages.len(), 1);
2389        assert!(matches!(
2390            messages.first(),
2391            Some(TradingCommand::SubmitOrder(command))
2392                if command.client_order_id == client_order_id
2393        ));
2394    }
2395
2396    #[rstest]
2397    fn test_dispatch_manager_submit_to_algorithm_uses_dynamic_endpoint(
2398        instrument: CryptoPerpetual,
2399    ) {
2400        let (_clock, _cache, emulator) = create_emulator();
2401        let exec_algorithm_id = ExecAlgorithmId::from("ALG-001");
2402        let endpoint = format!("{exec_algorithm_id}.execute");
2403        let (handler, messages) =
2404            get_any_saving_handler::<TradingCommand>(Some(Ustr::from("ALG-001.execute")));
2405        msgbus::register_any(endpoint.into(), handler);
2406        let order = create_stop_market_order(&instrument, TriggerType::Default);
2407        let command = create_submit_order(&instrument, &order);
2408        let client_order_id = command.client_order_id;
2409
2410        emulator
2411            .borrow_mut()
2412            .dispatch_manager_action(OrderManagerAction::SubmitToAlgorithm {
2413                command,
2414                exec_algorithm_id,
2415            });
2416
2417        let messages = messages.get_messages();
2418        assert_eq!(messages.len(), 1);
2419        assert!(matches!(
2420            messages.first(),
2421            Some(TradingCommand::SubmitOrder(command))
2422                if command.client_order_id == client_order_id
2423        ));
2424    }
2425
2426    #[rstest]
2427    fn test_dispatch_manager_cancel_local_applies_and_publishes_event(instrument: CryptoPerpetual) {
2428        let (_clock, cache, emulator) = create_emulator();
2429        let portfolio_events =
2430            register_portfolio_event_handler("Portfolio.update_order.dispatch_canceled");
2431        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2432        let client_order_id = order.client_order_id();
2433        let strategy_id = order.strategy_id();
2434        let instrument_id = order.instrument_id();
2435        let command = create_submit_order(&instrument, &order);
2436        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
2437        let (cancel_handler, cancel_events) = subscribe_order_cancel_topic(instrument_id);
2438        cache
2439            .borrow_mut()
2440            .add_order(order.clone(), None, None, false)
2441            .unwrap();
2442        emulator.borrow_mut().cache_submit_order_command(command);
2443
2444        emulator
2445            .borrow_mut()
2446            .dispatch_manager_action(OrderManagerAction::CancelLocal(order));
2447        msgbus::unsubscribe_order_events(get_event_order_topic(strategy_id).into(), &order_handler);
2448        msgbus::unsubscribe_order_events(
2449            get_order_canceled_topic(instrument_id).into(),
2450            &cancel_handler,
2451        );
2452        let cached_order = cache
2453            .borrow()
2454            .order(&client_order_id)
2455            .map(|order| order.clone())
2456            .unwrap();
2457        let portfolio_events = portfolio_events.get_messages();
2458        let order_events = order_events.borrow();
2459        let cancel_events = cancel_events.borrow();
2460        let commands = emulator.borrow().get_submit_order_commands();
2461
2462        assert_eq!(cached_order.status(), OrderStatus::Canceled);
2463        assert_eq!(portfolio_events.len(), 1);
2464        assert_eq!(order_events.len(), 1);
2465        assert_eq!(cancel_events.len(), 1);
2466        assert!(matches!(
2467            &portfolio_events[0],
2468            OrderEventAny::Canceled(event) if event.client_order_id == client_order_id
2469        ));
2470        assert_eq!(order_events[0].client_order_id(), client_order_id);
2471        assert_eq!(cancel_events[0].client_order_id(), client_order_id);
2472        assert!(!commands.contains_key(&client_order_id));
2473    }
2474
2475    #[rstest]
2476    fn test_dispatch_manager_modify_local_quantity_updates_order(instrument: CryptoPerpetual) {
2477        let (_clock, cache, emulator) = create_emulator();
2478        let risk_events = register_risk_event_handler("RiskEngine.process.dispatch_updated");
2479        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2480        let client_order_id = order.client_order_id();
2481        let quantity = Quantity::from(2);
2482        cache
2483            .borrow_mut()
2484            .add_order(order.clone(), None, None, false)
2485            .unwrap();
2486
2487        emulator
2488            .borrow_mut()
2489            .dispatch_manager_action(OrderManagerAction::ModifyLocalQuantity { order, quantity });
2490        let cache = cache.borrow();
2491        let cached_order = cache.order(&client_order_id).unwrap();
2492        let risk_events = risk_events.get_messages();
2493
2494        assert_eq!(cached_order.quantity(), quantity);
2495        assert_eq!(risk_events.len(), 1);
2496        assert!(matches!(
2497            &risk_events[0],
2498            OrderEventAny::Updated(event) if event.client_order_id == client_order_id
2499        ));
2500    }
2501
2502    #[rstest]
2503    fn test_subscribed_quotes_initially_empty() {
2504        let (_clock, _cache, emulator) = create_emulator();
2505
2506        assert!(emulator.borrow().subscribed_quotes().is_empty());
2507    }
2508
2509    #[rstest]
2510    fn test_subscribed_trades_initially_empty() {
2511        let (_clock, _cache, emulator) = create_emulator();
2512
2513        assert!(emulator.borrow().subscribed_trades().is_empty());
2514    }
2515
2516    #[rstest]
2517    fn test_get_submit_order_commands_initially_empty() {
2518        let (_clock, _cache, emulator) = create_emulator();
2519
2520        assert!(emulator.borrow().get_submit_order_commands().is_empty());
2521    }
2522
2523    #[rstest]
2524    fn test_get_matching_core_returns_none_when_not_created(instrument: CryptoPerpetual) {
2525        let (_clock, _cache, emulator) = create_emulator();
2526
2527        assert!(
2528            emulator
2529                .borrow()
2530                .get_matching_core(&instrument.id())
2531                .is_none()
2532        );
2533    }
2534
2535    #[rstest]
2536    fn test_create_matching_core(instrument: CryptoPerpetual) {
2537        let (_clock, _cache, emulator) = create_emulator();
2538
2539        emulator.borrow_mut().create_matching_core(
2540            instrument.id(),
2541            instrument.price_increment,
2542            TriggerType::BidAsk,
2543        );
2544
2545        assert!(
2546            emulator
2547                .borrow()
2548                .get_matching_core(&instrument.id())
2549                .is_some()
2550        );
2551    }
2552
2553    #[rstest]
2554    fn test_on_quote_tick_no_matching_core_does_not_panic(instrument: CryptoPerpetual) {
2555        let (_clock, _cache, emulator) = create_emulator();
2556        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
2557
2558        emulator.borrow_mut().on_quote_tick(quote);
2559    }
2560
2561    #[rstest]
2562    fn test_on_trade_tick_no_matching_core_does_not_panic(instrument: CryptoPerpetual) {
2563        let (_clock, _cache, emulator) = create_emulator();
2564        let trade = create_trade_tick(&instrument, "5065.00");
2565
2566        emulator.borrow_mut().on_trade_tick(trade);
2567    }
2568
2569    #[rstest]
2570    fn test_submit_order_bid_ask_trigger_creates_matching_core(instrument: CryptoPerpetual) {
2571        let (_clock, cache, emulator) = create_emulator();
2572        add_instrument_to_cache(&cache, &instrument);
2573        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2574        let command = create_submit_order(&instrument, &order);
2575        cache
2576            .borrow_mut()
2577            .add_order(order, None, None, false)
2578            .unwrap();
2579
2580        emulator
2581            .borrow_mut()
2582            .cache_submit_order_command(command.clone());
2583        emulator.borrow_mut().handle_submit_order(&command);
2584
2585        assert!(
2586            emulator
2587                .borrow()
2588                .get_matching_core(&instrument.id())
2589                .is_some()
2590        );
2591    }
2592
2593    #[rstest]
2594    fn test_submit_order_bid_ask_trigger_tracks_quote_subscription(instrument: CryptoPerpetual) {
2595        let (_clock, cache, emulator) = create_emulator();
2596        let data_commands = register_data_command_handler("DataEngine.queue_execute.quotes");
2597        add_instrument_to_cache(&cache, &instrument);
2598        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2599        let command = create_submit_order(&instrument, &order);
2600        cache
2601            .borrow_mut()
2602            .add_order(order, None, None, false)
2603            .unwrap();
2604
2605        emulator
2606            .borrow_mut()
2607            .cache_submit_order_command(command.clone());
2608        emulator.borrow_mut().handle_submit_order(&command);
2609
2610        assert_eq!(emulator.borrow().subscribed_quotes(), vec![instrument.id()]);
2611        assert!(emulator.borrow().subscribed_trades().is_empty());
2612
2613        let commands = data_commands.get_messages();
2614        assert_eq!(commands.len(), 1);
2615        assert!(matches!(
2616            &commands[0],
2617            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
2618                if command.instrument_ids == [instrument.id()]
2619        ));
2620
2621        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
2622        msgbus::publish_quote(get_quotes_topic(instrument.id()), &quote);
2623        let core = emulator
2624            .borrow()
2625            .get_matching_core(&instrument.id())
2626            .unwrap();
2627        assert_eq!(core.bid, Some(Price::from("5060.00")));
2628        assert_eq!(core.ask, Some(Price::from("5070.00")));
2629    }
2630
2631    #[rstest]
2632    fn test_submit_order_last_price_trigger_tracks_trade_subscription(instrument: CryptoPerpetual) {
2633        let (_clock, cache, emulator) = create_emulator();
2634        let data_commands = register_data_command_handler("DataEngine.queue_execute.trades");
2635        add_instrument_to_cache(&cache, &instrument);
2636        let order = create_stop_market_order(&instrument, TriggerType::LastPrice);
2637        let command = create_submit_order(&instrument, &order);
2638        cache
2639            .borrow_mut()
2640            .add_order(order, None, None, false)
2641            .unwrap();
2642
2643        emulator
2644            .borrow_mut()
2645            .cache_submit_order_command(command.clone());
2646        emulator.borrow_mut().handle_submit_order(&command);
2647
2648        assert!(emulator.borrow().subscribed_quotes().is_empty());
2649        assert_eq!(emulator.borrow().subscribed_trades(), vec![instrument.id()]);
2650
2651        let commands = data_commands.get_messages();
2652        assert_eq!(commands.len(), 1);
2653        assert!(matches!(
2654            &commands[0],
2655            DataCommand::Subscribe(SubscribeCommand::Trades(command))
2656                if command.instrument_ids == [instrument.id()]
2657        ));
2658
2659        let trade = create_trade_tick(&instrument, "5065.00");
2660        msgbus::publish_trade(get_trades_topic(instrument.id()), &trade);
2661        let core = emulator
2662            .borrow()
2663            .get_matching_core(&instrument.id())
2664            .unwrap();
2665        assert_eq!(core.last, Some(Price::from("5065.00")));
2666    }
2667
2668    #[rstest]
2669    fn test_reset_unsubscribes_market_data_and_clears_state(instrument: CryptoPerpetual) {
2670        let (_clock, cache, emulator) = create_emulator();
2671        let data_commands = register_data_command_handler("DataEngine.queue_execute.reset");
2672        add_instrument_to_cache(&cache, &instrument);
2673        let quote_order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2674        let quote_command = create_submit_order(&instrument, &quote_order);
2675        let trade_order = OrderTestBuilder::new(OrderType::StopMarket)
2676            .instrument_id(instrument.id())
2677            .client_order_id(ClientOrderId::from("O-RESET-TRADE"))
2678            .side(OrderSide::Buy)
2679            .trigger_price(Price::from("5100.00"))
2680            .quantity(Quantity::from(1))
2681            .emulation_trigger(TriggerType::LastPrice)
2682            .build();
2683        let trade_command = create_submit_order(&instrument, &trade_order);
2684        cache
2685            .borrow_mut()
2686            .add_order(quote_order, None, None, false)
2687            .unwrap();
2688        cache
2689            .borrow_mut()
2690            .add_order(trade_order, None, None, false)
2691            .unwrap();
2692        emulator
2693            .borrow_mut()
2694            .cache_submit_order_command(quote_command.clone());
2695        emulator.borrow_mut().handle_submit_order(&quote_command);
2696        emulator
2697            .borrow_mut()
2698            .cache_submit_order_command(trade_command.clone());
2699        emulator.borrow_mut().handle_submit_order(&trade_command);
2700        data_commands.clear();
2701
2702        emulator.borrow_mut().reset();
2703        let commands = data_commands.get_messages();
2704        let emulator_ref = emulator.borrow();
2705
2706        assert!(emulator_ref.subscribed_quotes.is_empty());
2707        assert!(emulator_ref.subscribed_trades.is_empty());
2708        assert!(emulator_ref.subscribed_strategies.is_empty());
2709        assert!(emulator_ref.monitored_positions.is_empty());
2710        assert!(emulator_ref.quote_handlers.is_empty());
2711        assert!(emulator_ref.trade_handlers.is_empty());
2712        assert!(emulator_ref.get_submit_order_commands().is_empty());
2713        assert!(emulator_ref.get_matching_core(&instrument.id()).is_none());
2714        assert!(commands.iter().any(|command| matches!(
2715            command,
2716            DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(command))
2717                if command.instrument_ids == [instrument.id()]
2718        )));
2719        assert!(commands.iter().any(|command| matches!(
2720            command,
2721            DataCommand::Unsubscribe(UnsubscribeCommand::Trades(command))
2722                if command.instrument_ids == [instrument.id()]
2723        )));
2724
2725        drop(emulator_ref);
2726        emulator.borrow_mut().create_matching_core(
2727            instrument.id(),
2728            instrument.price_increment,
2729            TriggerType::BidAsk,
2730        );
2731        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
2732        let trade = create_trade_tick(&instrument, "5065.00");
2733        msgbus::publish_quote(get_quotes_topic(instrument.id()), &quote);
2734        msgbus::publish_trade(get_trades_topic(instrument.id()), &trade);
2735        let core = emulator
2736            .borrow()
2737            .get_matching_core(&instrument.id())
2738            .unwrap();
2739        assert_eq!(core.bid, None);
2740        assert_eq!(core.ask, None);
2741        assert_eq!(core.last, None);
2742    }
2743
2744    #[rstest]
2745    fn test_reset_unsubscribes_in_sorted_instrument_order(instrument: CryptoPerpetual) {
2746        let (_clock, cache, emulator) = create_emulator();
2747        let data_commands = register_data_command_handler("DataEngine.queue_execute.reset_order");
2748
2749        // Registered in an order that is neither sorted nor reverse sorted, and five
2750        // entries leave a 1-in-120 chance that AHash order coincides with the expectation.
2751        let ids = [
2752            "SOLUSDT-PERP.BINANCE",
2753            "ADAUSDT-PERP.BINANCE",
2754            "XRPUSDT-PERP.BINANCE",
2755            "BTCUSDT-PERP.BINANCE",
2756            "DOTUSDT-PERP.BINANCE",
2757        ];
2758
2759        for (i, id) in ids.iter().enumerate() {
2760            let mut variant = instrument.clone();
2761            variant.id = InstrumentId::from(*id);
2762            add_instrument_to_cache(&cache, &variant);
2763
2764            let order = OrderTestBuilder::new(OrderType::StopMarket)
2765                .instrument_id(variant.id)
2766                .client_order_id(ClientOrderId::from(format!("O-RESET-{i}").as_str()))
2767                .side(OrderSide::Buy)
2768                .trigger_price(Price::from("5100.00"))
2769                .quantity(Quantity::from(1))
2770                .emulation_trigger(TriggerType::BidAsk)
2771                .build();
2772            let command = create_submit_order(&variant, &order);
2773            cache
2774                .borrow_mut()
2775                .add_order(order, None, None, false)
2776                .unwrap();
2777            emulator
2778                .borrow_mut()
2779                .cache_submit_order_command(command.clone());
2780            emulator.borrow_mut().handle_submit_order(&command);
2781        }
2782        data_commands.clear();
2783
2784        emulator.borrow_mut().reset();
2785
2786        let unsubscribed: Vec<InstrumentId> = data_commands
2787            .get_messages()
2788            .iter()
2789            .filter_map(|command| match command {
2790                DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(command)) => {
2791                    Some(command.single_instrument_id().unwrap())
2792                }
2793                _ => None,
2794            })
2795            .collect();
2796
2797        let mut expected: Vec<InstrumentId> =
2798            ids.iter().map(|id| InstrumentId::from(*id)).collect();
2799        expected.sort();
2800
2801        assert_eq!(unsubscribed, expected);
2802    }
2803
2804    #[rstest]
2805    fn test_submit_order_list_handles_reentrant_order_events(instrument: CryptoPerpetual) {
2806        let (_clock, cache, emulator) = create_emulator();
2807        let data_commands = register_data_command_handler("DataEngine.queue_execute.list");
2808        add_instrument_to_cache(&cache, &instrument);
2809        let order_list_id = OrderListId::from("OL-EMULATOR-001");
2810        let first_order = create_list_stop_market_order(&instrument, "O-LIST-001", order_list_id);
2811        let second_order = create_list_stop_market_order(&instrument, "O-LIST-002", order_list_id);
2812        let orders = vec![first_order.clone(), second_order.clone()];
2813        let order_list = OrderList::from_orders(&orders, 0.into());
2814        let order_inits = orders
2815            .iter()
2816            .map(|order| order.init_event().clone())
2817            .collect();
2818        cache
2819            .borrow_mut()
2820            .add_order(first_order.clone(), None, None, false)
2821            .unwrap();
2822        cache
2823            .borrow_mut()
2824            .add_order(second_order.clone(), None, None, false)
2825            .unwrap();
2826        let command = SubmitOrderList::new(
2827            TraderId::from("TRADER-001"),
2828            None,
2829            StrategyId::from("STRATEGY-001"),
2830            order_list,
2831            order_inits,
2832            None,
2833            None,
2834            None,
2835            UUID4::new(),
2836            0.into(),
2837            None,
2838        );
2839
2840        emulator
2841            .borrow_mut()
2842            .execute(TradingCommand::SubmitOrderList(command));
2843
2844        let commands = data_commands.get_messages();
2845        let cache = cache.borrow();
2846        let first_status = cache
2847            .order(&first_order.client_order_id())
2848            .unwrap()
2849            .status();
2850        let second_status = cache
2851            .order(&second_order.client_order_id())
2852            .unwrap()
2853            .status();
2854        drop(cache);
2855        let emulator = emulator.borrow();
2856
2857        assert_eq!(first_status, OrderStatus::Emulated);
2858        assert_eq!(second_status, OrderStatus::Emulated);
2859        assert!(emulator.get_matching_core(&instrument.id()).is_some());
2860        assert_eq!(emulator.subscribed_quotes(), vec![instrument.id()]);
2861        assert!(commands.iter().any(|command| matches!(
2862            command,
2863            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
2864                if command.instrument_ids == [instrument.id()]
2865        )));
2866    }
2867
2868    #[rstest]
2869    fn test_submit_order_caches_command(instrument: CryptoPerpetual) {
2870        let (_clock, cache, emulator) = create_emulator();
2871        add_instrument_to_cache(&cache, &instrument);
2872        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2873        let client_order_id = order.client_order_id();
2874        let command = create_submit_order(&instrument, &order);
2875        cache
2876            .borrow_mut()
2877            .add_order(order, None, None, false)
2878            .unwrap();
2879
2880        emulator
2881            .borrow_mut()
2882            .cache_submit_order_command(command.clone());
2883        emulator.borrow_mut().handle_submit_order(&command);
2884
2885        let commands = emulator.borrow().get_submit_order_commands();
2886        assert!(commands.contains_key(&client_order_id));
2887    }
2888
2889    #[rstest]
2890    fn test_handle_submit_order_applies_emulated_event_to_cache(instrument: CryptoPerpetual) {
2891        let (_clock, cache, emulator) = create_emulator();
2892        let risk_events = register_risk_event_handler("RiskEngine.process.emulated");
2893        add_instrument_to_cache(&cache, &instrument);
2894        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2895        let client_order_id = order.client_order_id();
2896        let strategy_id = order.strategy_id();
2897        let command = create_submit_order(&instrument, &order);
2898        cache
2899            .borrow_mut()
2900            .add_order(order, None, None, false)
2901            .unwrap();
2902        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
2903
2904        emulator
2905            .borrow_mut()
2906            .cache_submit_order_command(command.clone());
2907        emulator.borrow_mut().handle_submit_order(&command);
2908        msgbus::unsubscribe_order_events(
2909            format!("events.order.{strategy_id}").into(),
2910            &order_handler,
2911        );
2912        let cache = cache.borrow();
2913        let cached_order = cache.order(&client_order_id).unwrap();
2914        let risk_events = risk_events.get_messages();
2915        let order_events = order_events.borrow();
2916
2917        assert_eq!(cached_order.status(), OrderStatus::Emulated);
2918        assert_eq!(cached_order.event_count(), 2);
2919        assert_eq!(risk_events.len(), 1);
2920        assert!(matches!(risk_events[0], OrderEventAny::Emulated(_)));
2921        assert_eq!(order_events.len(), 1);
2922        assert!(matches!(order_events[0], OrderEventAny::Emulated(_)));
2923    }
2924
2925    #[rstest]
2926    fn test_submit_order_missing_synthetic_trigger_cancels_order(instrument: CryptoPerpetual) {
2927        let (_clock, cache, emulator) = create_emulator();
2928        add_instrument_to_cache(&cache, &instrument);
2929        let synthetic_id = InstrumentId::from("BTC-ETH-INDEX.SYNTH");
2930        let order = OrderTestBuilder::new(OrderType::StopMarket)
2931            .instrument_id(instrument.id())
2932            .side(OrderSide::Buy)
2933            .trigger_price(Price::from("5100.00"))
2934            .quantity(Quantity::from(1))
2935            .emulation_trigger(TriggerType::BidAsk)
2936            .trigger_instrument_id(synthetic_id)
2937            .build();
2938        let client_order_id = order.client_order_id();
2939        let command = create_submit_order(&instrument, &order);
2940        cache
2941            .borrow_mut()
2942            .add_order(order, None, None, false)
2943            .unwrap();
2944
2945        emulator
2946            .borrow_mut()
2947            .cache_submit_order_command(command.clone());
2948        emulator.borrow_mut().handle_submit_order(&command);
2949
2950        let cache = cache.borrow();
2951        let cached_order = cache.order(&client_order_id).unwrap();
2952        let emulator = emulator.borrow();
2953        let commands = emulator.get_submit_order_commands();
2954
2955        assert_eq!(cached_order.status(), OrderStatus::Canceled);
2956        assert!(emulator.get_matching_core(&synthetic_id).is_none());
2957        assert!(emulator.subscribed_quotes().is_empty());
2958        assert!(emulator.subscribed_trades().is_empty());
2959        assert!(!commands.contains_key(&client_order_id));
2960    }
2961
2962    #[rstest]
2963    fn test_submit_order_releases_when_immediately_triggered(instrument: CryptoPerpetual) {
2964        let (_clock, cache, emulator) = create_emulator();
2965        let risk_events =
2966            register_risk_event_handler("RiskEngine.process.submit_immediately_triggered");
2967        add_instrument_to_cache(&cache, &instrument);
2968        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
2969        core.set_bid_raw(Price::from("5099.00"));
2970        core.set_ask_raw(Price::from("5101.00"));
2971        emulator
2972            .borrow_mut()
2973            .matching_cores
2974            .insert(instrument.id(), core);
2975        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
2976        let client_order_id = order.client_order_id();
2977        let command = create_submit_order(&instrument, &order);
2978        cache
2979            .borrow_mut()
2980            .add_order(order, None, None, false)
2981            .unwrap();
2982
2983        emulator.borrow_mut().handle_submit_order(&command);
2984
2985        let cache = cache.borrow();
2986        let cached_order = cache.order(&client_order_id).unwrap();
2987        let emulator = emulator.borrow();
2988        let matching_core = emulator.get_matching_core(&instrument.id()).unwrap();
2989        let risk_events = risk_events.get_messages();
2990        assert_eq!(cached_order.status(), OrderStatus::Released);
2991        assert_eq!(cached_order.order_type(), OrderType::Market);
2992        assert!(!matching_core.order_exists(client_order_id));
2993        assert_eq!(emulator.subscribed_strategy_count(), 1);
2994        assert!(
2995            !emulator
2996                .get_submit_order_commands()
2997                .contains_key(&client_order_id)
2998        );
2999        assert_eq!(risk_events.len(), 1);
3000        assert!(matches!(
3001            &risk_events[0],
3002            OrderEventAny::Released(event) if event.client_order_id == client_order_id
3003        ));
3004    }
3005
3006    #[rstest]
3007    #[case::bid_ask_quote_below(TriggerType::BidAsk, Some(("5098.00", "5099.00")), Some("5101.00"), (Some(false), Some(true)))]
3008    #[case::bid_ask_quote_above(TriggerType::BidAsk, Some(("5101.00", "5102.00")), Some("5099.00"), (Some(true), Some(false)))]
3009    #[case::default_quote(TriggerType::Default, Some(("5098.00", "5099.00")), Some("5101.00"), (Some(false), Some(true)))]
3010    #[case::last_price_trade_below(TriggerType::LastPrice, Some(("5101.00", "5102.00")), Some("5099.00"), (Some(false), Some(true)))]
3011    #[case::last_price_trade_above(TriggerType::LastPrice, Some(("5098.00", "5099.00")), Some("5101.00"), (Some(true), Some(false)))]
3012    #[case::last_price_quote_only(TriggerType::LastPrice, Some(("5098.00", "5099.00")), None, (None, None))]
3013    #[case::last_price_trade_only(
3014        TriggerType::LastPrice,
3015        None,
3016        Some("5101.00"),
3017        (Some(true), Some(false))
3018    )]
3019    #[case::bid_ask_trade_only(TriggerType::BidAsk, None, Some("5101.00"), (None, None))]
3020    #[case::default_trade_only(TriggerType::Default, None, Some("5101.00"), (None, None))]
3021    #[case::bid_ask_no_data(TriggerType::BidAsk, None, None, (None, None))]
3022    #[case::default_no_data(TriggerType::Default, None, None, (None, None))]
3023    #[case::last_price_no_data(TriggerType::LastPrice, None, None, (None, None))]
3024    fn test_submit_order_uses_cached_emulation_source(
3025        instrument: CryptoPerpetual,
3026        #[case] emulation_trigger: TriggerType,
3027        #[case] quote: Option<(&str, &str)>,
3028        #[case] trade: Option<&str>,
3029        #[case] stop_matches: (Option<bool>, Option<bool>),
3030        #[values(
3031            OrderType::Limit,
3032            OrderType::StopMarket,
3033            OrderType::StopLimit,
3034            OrderType::MarketIfTouched,
3035            OrderType::LimitIfTouched
3036        )]
3037        order_type: OrderType,
3038        #[values(OrderSide::Buy, OrderSide::Sell)] side: OrderSide,
3039    ) {
3040        let (_clock, cache, emulator) = create_emulator();
3041        let risk_events = register_risk_event_handler("RiskEngine.process.submit_cached_source");
3042        add_instrument_to_cache(&cache, &instrument);
3043        if let Some((bid, ask)) = quote {
3044            cache
3045                .borrow_mut()
3046                .add_quote(create_quote_tick(&instrument, bid, ask))
3047                .unwrap();
3048        }
3049
3050        if let Some(price) = trade {
3051            cache
3052                .borrow_mut()
3053                .add_trade(create_trade_tick(&instrument, price))
3054                .unwrap();
3055        }
3056
3057        let order = OrderTestBuilder::new(order_type)
3058            .instrument_id(instrument.id())
3059            .side(side)
3060            .price(Price::from("5100.00"))
3061            .trigger_price(Price::from("5100.00"))
3062            .quantity(Quantity::from(3))
3063            .emulation_trigger(emulation_trigger)
3064            .build();
3065        let client_order_id = order.client_order_id();
3066        let command = create_submit_order(&instrument, &order);
3067        cache
3068            .borrow_mut()
3069            .add_order(order, None, None, false)
3070            .unwrap();
3071
3072        emulator.borrow_mut().handle_submit_order(&command);
3073
3074        let stop_matches = match side {
3075            OrderSide::Buy => stop_matches.0,
3076            OrderSide::Sell => stop_matches.1,
3077        };
3078
3079        let should_release = stop_matches.is_some_and(|matches| match order_type {
3080            OrderType::StopMarket | OrderType::StopLimit => matches,
3081            _ => !matches,
3082        });
3083
3084        let expected_status = if should_release {
3085            OrderStatus::Released
3086        } else {
3087            OrderStatus::Emulated
3088        };
3089
3090        let expected_type = if should_release {
3091            match order_type {
3092                OrderType::StopLimit | OrderType::LimitIfTouched => OrderType::Limit,
3093                _ => OrderType::Market,
3094            }
3095        } else {
3096            order_type
3097        };
3098
3099        let cache = cache.borrow();
3100        let cached_order = cache.order(&client_order_id).unwrap();
3101        let emulator = emulator.borrow();
3102        let core = emulator.get_matching_core(&instrument.id()).unwrap();
3103        let events = risk_events.get_messages();
3104        assert_eq!(cached_order.status(), expected_status);
3105        assert_eq!(cached_order.order_type(), expected_type);
3106        assert_eq!(cached_order.quantity(), Quantity::from(3));
3107        assert_eq!(core.order_exists(client_order_id), !should_release);
3108        assert_eq!(
3109            emulator
3110                .get_submit_order_commands()
3111                .contains_key(&client_order_id),
3112            !should_release
3113        );
3114        assert_eq!(events.len(), 1);
3115        assert_eq!(events[0].client_order_id(), client_order_id);
3116        assert_eq!(
3117            matches!(&events[0], OrderEventAny::Released(_)),
3118            should_release
3119        );
3120        assert_eq!(
3121            matches!(&events[0], OrderEventAny::Emulated(_)),
3122            !should_release
3123        );
3124
3125        let expected_price = match (emulation_trigger, side) {
3126            (TriggerType::LastPrice, _) => trade.map(Price::from),
3127            (_, OrderSide::Buy) => quote.map(|(_, ask)| Price::from(ask)),
3128            (_, OrderSide::Sell) => quote.map(|(bid, _)| Price::from(bid)),
3129        };
3130
3131        let released_price = match &events[0] {
3132            OrderEventAny::Released(event) => Some(event.released_price),
3133            _ => None,
3134        };
3135
3136        assert_eq!(released_price, expected_price.filter(|_| should_release));
3137    }
3138
3139    #[rstest]
3140    #[case(TriggerType::Default, "5099.00", "5101.00", None)]
3141    #[case(TriggerType::BidAsk, "5099.00", "5101.00", None)]
3142    #[case(TriggerType::LastPrice, "5098.00", "5098.00", Some("5098.00"))]
3143    fn test_create_matching_core_seeds_cached_prices(
3144        instrument: CryptoPerpetual,
3145        #[case] emulation_trigger: TriggerType,
3146        #[case] expected_bid: &str,
3147        #[case] expected_ask: &str,
3148        #[case] expected_last: Option<&str>,
3149    ) {
3150        let (_clock, cache, emulator) = create_emulator();
3151        cache
3152            .borrow_mut()
3153            .add_quote(create_quote_tick(&instrument, "5099.00", "5101.00"))
3154            .unwrap();
3155        cache
3156            .borrow_mut()
3157            .add_trade(create_trade_tick(&instrument, "5098.00"))
3158            .unwrap();
3159
3160        let core = emulator.borrow_mut().create_matching_core(
3161            instrument.id(),
3162            instrument.price_increment,
3163            emulation_trigger,
3164        );
3165
3166        let emulator = emulator.borrow();
3167        let stored_core = emulator.get_matching_core(&instrument.id()).unwrap();
3168        assert_eq!(core.bid, Some(Price::from(expected_bid)));
3169        assert_eq!(core.ask, Some(Price::from(expected_ask)));
3170        assert_eq!(core.last, expected_last.map(Price::from));
3171        assert_eq!(stored_core.bid, core.bid);
3172        assert_eq!(stored_core.ask, core.ask);
3173        assert_eq!(stored_core.last, core.last);
3174    }
3175
3176    #[rstest]
3177    fn test_submit_order_seeds_cached_trigger_instrument(
3178        instrument: CryptoPerpetual,
3179        #[values(TriggerType::BidAsk, TriggerType::LastPrice)] emulation_trigger: TriggerType,
3180    ) {
3181        let (_clock, cache, emulator) = create_emulator();
3182        let _risk_events =
3183            register_risk_event_handler("RiskEngine.process.cached_trigger_instrument");
3184        let trigger_instrument = btcusd_bybit();
3185        add_instrument_to_cache(&cache, &instrument);
3186        add_instrument_to_cache(&cache, &trigger_instrument);
3187        cache
3188            .borrow_mut()
3189            .add_quote(create_quote_tick(&instrument, "5099.00", "5101.00"))
3190            .unwrap();
3191        cache
3192            .borrow_mut()
3193            .add_trade(create_trade_tick(&instrument, "5101.00"))
3194            .unwrap();
3195        cache
3196            .borrow_mut()
3197            .add_quote(create_quote_tick(&trigger_instrument, "5098.00", "5099.00"))
3198            .unwrap();
3199        cache
3200            .borrow_mut()
3201            .add_trade(create_trade_tick(&trigger_instrument, "5099.00"))
3202            .unwrap();
3203        let order = OrderTestBuilder::new(OrderType::StopMarket)
3204            .instrument_id(instrument.id())
3205            .trigger_instrument_id(trigger_instrument.id())
3206            .side(OrderSide::Buy)
3207            .trigger_price(Price::from("5100.00"))
3208            .quantity(Quantity::from(1))
3209            .emulation_trigger(emulation_trigger)
3210            .build();
3211        let client_order_id = order.client_order_id();
3212        let command = create_submit_order(&instrument, &order);
3213        cache
3214            .borrow_mut()
3215            .add_order(order, None, None, false)
3216            .unwrap();
3217
3218        emulator.borrow_mut().handle_submit_order(&command);
3219
3220        let emulator = emulator.borrow();
3221        let core = emulator
3222            .get_matching_core(&trigger_instrument.id())
3223            .unwrap();
3224
3225        let expected_bid = if emulation_trigger == TriggerType::LastPrice {
3226            "5099.00"
3227        } else {
3228            "5098.00"
3229        };
3230
3231        assert_eq!(core.bid, Some(Price::from(expected_bid)));
3232        assert_eq!(core.ask, Some(Price::from("5099.00")));
3233        assert_eq!(
3234            core.last,
3235            (emulation_trigger == TriggerType::LastPrice).then(|| Price::from("5099.00"))
3236        );
3237        assert_eq!(
3238            cache.borrow().order(&client_order_id).unwrap().status(),
3239            OrderStatus::Emulated
3240        );
3241        assert!(emulator.get_matching_core(&instrument.id()).is_none());
3242    }
3243
3244    #[rstest]
3245    fn test_quote_emulated_trailing_stop_uses_latest_cached_trade(
3246        instrument: CryptoPerpetual,
3247        #[values(TriggerType::Default, TriggerType::BidAsk)] emulation_trigger: TriggerType,
3248        #[values(TriggerType::LastPrice, TriggerType::LastOrBidAsk)] trigger_type: TriggerType,
3249    ) {
3250        let (_clock, cache, emulator) = create_emulator();
3251        let _risk_events = register_risk_event_handler("RiskEngine.process.cached_trailing_trade");
3252        add_instrument_to_cache(&cache, &instrument);
3253        cache
3254            .borrow_mut()
3255            .add_quote(create_quote_tick(&instrument, "5000.00", "5001.00"))
3256            .unwrap();
3257        cache
3258            .borrow_mut()
3259            .add_trade(create_trade_tick(&instrument, "5000.00"))
3260            .unwrap();
3261        let order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
3262            .instrument_id(instrument.id())
3263            .side(OrderSide::Sell)
3264            .quantity(Quantity::from(3))
3265            .trigger_type(trigger_type)
3266            .trailing_offset(dec!(10))
3267            .trailing_offset_type(TrailingOffsetType::Price)
3268            .emulation_trigger(emulation_trigger)
3269            .build();
3270        let client_order_id = order.client_order_id();
3271        let command = create_submit_order(&instrument, &order);
3272        cache
3273            .borrow_mut()
3274            .add_order(order, None, None, false)
3275            .unwrap();
3276
3277        emulator.borrow_mut().handle_submit_order(&command);
3278
3279        assert_eq!(
3280            cache
3281                .borrow()
3282                .order(&client_order_id)
3283                .unwrap()
3284                .trigger_price(),
3285            Some(Price::from("4990.00"))
3286        );
3287
3288        cache
3289            .borrow_mut()
3290            .add_trade(create_trade_tick(&instrument, "5050.00"))
3291            .unwrap();
3292        emulator
3293            .borrow_mut()
3294            .on_quote_tick(create_quote_tick(&instrument, "5010.00", "5011.00"));
3295
3296        let cache = cache.borrow();
3297        let order = cache.order(&client_order_id).unwrap();
3298        assert_eq!(order.trigger_price(), Some(Price::from("5040.00")));
3299        assert_eq!(order.quantity(), Quantity::from(3));
3300        assert_eq!(order.status(), OrderStatus::Emulated);
3301    }
3302
3303    #[rstest]
3304    fn test_submit_order_preserves_existing_matching_core(
3305        instrument: CryptoPerpetual,
3306        #[values(TriggerType::BidAsk, TriggerType::LastPrice)] emulation_trigger: TriggerType,
3307    ) {
3308        let (_clock, cache, emulator) = create_emulator();
3309        let _risk_events = register_risk_event_handler("RiskEngine.process.cached_existing_core");
3310        add_instrument_to_cache(&cache, &instrument);
3311        cache
3312            .borrow_mut()
3313            .add_quote(create_quote_tick(&instrument, "5098.00", "5099.00"))
3314            .unwrap();
3315        cache
3316            .borrow_mut()
3317            .add_trade(create_trade_tick(&instrument, "5099.00"))
3318            .unwrap();
3319
3320        let first = create_stop_market_order(&instrument, emulation_trigger);
3321        let first_id = first.client_order_id();
3322        let first_command = create_submit_order(&instrument, &first);
3323        cache
3324            .borrow_mut()
3325            .add_order(first, None, None, false)
3326            .unwrap();
3327        emulator.borrow_mut().handle_submit_order(&first_command);
3328        cache
3329            .borrow_mut()
3330            .add_quote(create_quote_tick(&instrument, "5101.00", "5102.00"))
3331            .unwrap();
3332        cache
3333            .borrow_mut()
3334            .add_trade(create_trade_tick(&instrument, "5102.00"))
3335            .unwrap();
3336        let second = OrderTestBuilder::new(OrderType::StopMarket)
3337            .instrument_id(instrument.id())
3338            .client_order_id(ClientOrderId::from("O-CACHED-SECOND"))
3339            .side(OrderSide::Buy)
3340            .trigger_price(Price::from("5100.00"))
3341            .quantity(Quantity::from(1))
3342            .emulation_trigger(emulation_trigger)
3343            .build();
3344        let second_id = second.client_order_id();
3345        let second_command = create_submit_order(&instrument, &second);
3346        cache
3347            .borrow_mut()
3348            .add_order(second, None, None, false)
3349            .unwrap();
3350
3351        emulator.borrow_mut().handle_submit_order(&second_command);
3352
3353        let emulator = emulator.borrow();
3354        let core = emulator.get_matching_core(&instrument.id()).unwrap();
3355        assert_eq!(core.ask, Some(Price::from("5099.00")));
3356        assert_eq!(
3357            core.last,
3358            (emulation_trigger == TriggerType::LastPrice).then(|| Price::from("5099.00"))
3359        );
3360        assert!(core.order_exists(first_id));
3361        assert!(core.order_exists(second_id));
3362        assert_eq!(
3363            cache.borrow().order(&first_id).unwrap().status(),
3364            OrderStatus::Emulated
3365        );
3366        assert_eq!(
3367            cache.borrow().order(&second_id).unwrap().status(),
3368            OrderStatus::Emulated
3369        );
3370    }
3371
3372    #[rstest]
3373    #[case(OrderType::StopMarket, ("5099.00", "5101.00"), "5099.00", "5101.00")]
3374    #[case(OrderType::Limit, ("5098.00", "5099.00"), "5101.00", "5099.00")]
3375    fn test_modify_last_price_order_ignores_cached_quote_before_first_trade(
3376        instrument: CryptoPerpetual,
3377        #[case] order_type: OrderType,
3378        #[case] quote: (&str, &str),
3379        #[case] trade: &str,
3380        #[case] next_trade: &str,
3381    ) {
3382        let (_clock, cache, emulator) = create_emulator();
3383        let risk_events = register_risk_event_handler("RiskEngine.process.modify_cached_quote");
3384        add_instrument_to_cache(&cache, &instrument);
3385        cache
3386            .borrow_mut()
3387            .add_quote(create_quote_tick(&instrument, quote.0, quote.1))
3388            .unwrap();
3389        cache
3390            .borrow_mut()
3391            .add_trade(create_trade_tick(&instrument, trade))
3392            .unwrap();
3393        let order = OrderTestBuilder::new(order_type)
3394            .instrument_id(instrument.id())
3395            .side(OrderSide::Buy)
3396            .trigger_price(Price::from("5150.00"))
3397            .price(Price::from("5050.00"))
3398            .quantity(Quantity::from(1))
3399            .emulation_trigger(TriggerType::LastPrice)
3400            .build();
3401        let client_order_id = order.client_order_id();
3402        let command = create_submit_order(&instrument, &order);
3403        cache
3404            .borrow_mut()
3405            .add_order(order.clone(), None, None, false)
3406            .unwrap();
3407        emulator.borrow_mut().handle_submit_order(&command);
3408
3409        let modify = ModifyOrder::new(
3410            order.trader_id(),
3411            None,
3412            order.strategy_id(),
3413            instrument.id(),
3414            client_order_id,
3415            None,
3416            None,
3417            (order_type == OrderType::Limit).then(|| Price::from("5100.00")),
3418            (order_type == OrderType::StopMarket).then(|| Price::from("5100.00")),
3419            UUID4::new(),
3420            0.into(),
3421            None,
3422            None,
3423        );
3424
3425        emulator.borrow_mut().handle_modify_order(&modify);
3426
3427        {
3428            let cache = cache.borrow();
3429            let cached_order = cache.order(&client_order_id).unwrap();
3430            let core = emulator
3431                .borrow()
3432                .get_matching_core(&instrument.id())
3433                .unwrap();
3434            assert_eq!(core.bid, Some(Price::from(trade)));
3435            assert_eq!(core.ask, Some(Price::from(trade)));
3436            assert_eq!(core.last, Some(Price::from(trade)));
3437            assert_eq!(cached_order.status(), OrderStatus::Emulated);
3438            assert_eq!(
3439                if order_type == OrderType::Limit {
3440                    cached_order.price()
3441                } else {
3442                    cached_order.trigger_price()
3443                },
3444                Some(Price::from("5100.00"))
3445            );
3446            assert!(matches!(risk_events.get_messages().as_slice(),
3447                [OrderEventAny::Emulated(event)] if event.client_order_id == client_order_id
3448            ));
3449        }
3450
3451        emulator
3452            .borrow_mut()
3453            .on_trade_tick(create_trade_tick(&instrument, next_trade));
3454
3455        assert_eq!(
3456            cache.borrow().order(&client_order_id).unwrap().status(),
3457            OrderStatus::Released
3458        );
3459        assert!(matches!(risk_events.get_messages().as_slice(),
3460            [OrderEventAny::Emulated(_), OrderEventAny::Released(event)]
3461                if event.client_order_id == client_order_id
3462                    && event.released_price == Price::from(next_trade)
3463        ));
3464    }
3465
3466    #[rstest]
3467    fn test_submit_limit_order_releases_when_immediately_fillable(instrument: CryptoPerpetual) {
3468        let (_clock, cache, emulator) = create_emulator();
3469        let risk_events =
3470            register_risk_event_handler("RiskEngine.process.submit_immediately_fillable");
3471        add_instrument_to_cache(&cache, &instrument);
3472        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
3473        core.set_bid_raw(Price::from("5099.00"));
3474        core.set_ask_raw(Price::from("5101.00"));
3475        emulator
3476            .borrow_mut()
3477            .matching_cores
3478            .insert(instrument.id(), core);
3479        let order = OrderTestBuilder::new(OrderType::Limit)
3480            .instrument_id(instrument.id())
3481            .side(OrderSide::Buy)
3482            .price(Price::from("5101.00"))
3483            .quantity(Quantity::from(1))
3484            .emulation_trigger(TriggerType::BidAsk)
3485            .build();
3486        let client_order_id = order.client_order_id();
3487        let command = create_submit_order(&instrument, &order);
3488        cache
3489            .borrow_mut()
3490            .add_order(order, None, None, false)
3491            .unwrap();
3492
3493        emulator.borrow_mut().handle_submit_order(&command);
3494
3495        let cache = cache.borrow();
3496        let cached_order = cache.order(&client_order_id).unwrap();
3497        let emulator = emulator.borrow();
3498        let matching_core = emulator.get_matching_core(&instrument.id()).unwrap();
3499        let risk_events = risk_events.get_messages();
3500        assert_eq!(cached_order.status(), OrderStatus::Released);
3501        assert_eq!(cached_order.order_type(), OrderType::Market);
3502        assert!(!matching_core.order_exists(client_order_id));
3503        assert!(
3504            !emulator
3505                .get_submit_order_commands()
3506                .contains_key(&client_order_id)
3507        );
3508        assert_eq!(risk_events.len(), 1);
3509        assert!(matches!(
3510            &risk_events[0],
3511            OrderEventAny::Released(event) if event.client_order_id == client_order_id
3512        ));
3513    }
3514
3515    #[rstest]
3516    fn test_modify_order_reindexes_matching_state(instrument: CryptoPerpetual) {
3517        let (_clock, cache, emulator) = create_emulator();
3518        add_instrument_to_cache(&cache, &instrument);
3519        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
3520        let client_order_id = order.client_order_id();
3521        let command = create_submit_order(&instrument, &order);
3522        cache
3523            .borrow_mut()
3524            .add_order(order.clone(), None, None, false)
3525            .unwrap();
3526        emulator.borrow_mut().handle_submit_order(&command);
3527        let exec_events =
3528            register_exec_event_handler(cache.clone(), "ExecEngine.process.modify_reindex");
3529        let portfolio_events = register_portfolio_event_handler("Portfolio.modify_reindex");
3530        let new_trigger = Price::from("5200.00");
3531        let modify = ModifyOrder::new(
3532            order.trader_id(),
3533            None,
3534            order.strategy_id(),
3535            instrument.id(),
3536            client_order_id,
3537            None,
3538            None,
3539            None,
3540            Some(new_trigger),
3541            UUID4::new(),
3542            0.into(),
3543            None,
3544            None,
3545        );
3546
3547        emulator.borrow_mut().handle_modify_order(&modify);
3548
3549        let cache = cache.borrow();
3550        let cached_order = cache.order(&client_order_id).unwrap();
3551        let emulator = emulator.borrow();
3552        let matching_core = emulator.get_matching_core(&instrument.id()).unwrap();
3553        let match_info = matching_core.get_order(client_order_id).unwrap();
3554        let portfolio_events = portfolio_events.get_messages();
3555        assert_eq!(cached_order.trigger_price(), Some(new_trigger));
3556        assert_eq!(match_info.trigger_price, Some(new_trigger));
3557        assert_eq!(matching_core.get_orders().len(), 1);
3558        assert!(exec_events.get_messages().is_empty());
3559        assert_eq!(portfolio_events.len(), 1);
3560        assert!(matches!(
3561            &portfolio_events[0],
3562            OrderEventAny::Updated(event) if event.client_order_id == client_order_id
3563        ));
3564    }
3565
3566    #[rstest]
3567    fn test_modify_order_releases_when_updated_trigger_matches(instrument: CryptoPerpetual) {
3568        let (_clock, cache, emulator) = create_emulator();
3569        let risk_events =
3570            register_risk_event_handler("RiskEngine.process.modify_immediately_triggered");
3571        add_instrument_to_cache(&cache, &instrument);
3572        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
3573        core.set_bid_raw(Price::from("5099.00"));
3574        core.set_ask_raw(Price::from("5101.00"));
3575        emulator
3576            .borrow_mut()
3577            .matching_cores
3578            .insert(instrument.id(), core);
3579        let order = OrderTestBuilder::new(OrderType::StopMarket)
3580            .instrument_id(instrument.id())
3581            .side(OrderSide::Buy)
3582            .trigger_price(Price::from("5200.00"))
3583            .quantity(Quantity::from(1))
3584            .emulation_trigger(TriggerType::BidAsk)
3585            .build();
3586        let client_order_id = order.client_order_id();
3587        let command = create_submit_order(&instrument, &order);
3588        cache
3589            .borrow_mut()
3590            .add_order(order.clone(), None, None, false)
3591            .unwrap();
3592        emulator.borrow_mut().handle_submit_order(&command);
3593        risk_events.clear();
3594        let exec_events = register_exec_event_handler(
3595            cache.clone(),
3596            "ExecEngine.process.modify_immediately_triggered",
3597        );
3598        let portfolio_events =
3599            register_portfolio_event_handler("Portfolio.modify_immediately_triggered");
3600
3601        let modify = ModifyOrder::new(
3602            order.trader_id(),
3603            None,
3604            order.strategy_id(),
3605            instrument.id(),
3606            client_order_id,
3607            None,
3608            None,
3609            None,
3610            Some(Price::from("5100.00")),
3611            UUID4::new(),
3612            0.into(),
3613            None,
3614            None,
3615        );
3616
3617        emulator.borrow_mut().handle_modify_order(&modify);
3618
3619        let cache = cache.borrow();
3620        let cached_order = cache.order(&client_order_id).unwrap();
3621        let emulator = emulator.borrow();
3622        let matching_core = emulator.get_matching_core(&instrument.id()).unwrap();
3623        let portfolio_events = portfolio_events.get_messages();
3624        let risk_events = risk_events.get_messages();
3625        assert_eq!(cached_order.status(), OrderStatus::Released);
3626        assert_eq!(cached_order.order_type(), OrderType::Market);
3627        assert!(!matching_core.order_exists(client_order_id));
3628        assert!(
3629            !emulator
3630                .get_submit_order_commands()
3631                .contains_key(&client_order_id)
3632        );
3633        assert!(exec_events.get_messages().is_empty());
3634        assert_eq!(portfolio_events.len(), 1);
3635        assert!(matches!(
3636            &portfolio_events[0],
3637            OrderEventAny::Updated(event) if event.client_order_id == client_order_id
3638        ));
3639        assert_eq!(risk_events.len(), 1);
3640        assert!(matches!(
3641            &risk_events[0],
3642            OrderEventAny::Released(event) if event.client_order_id == client_order_id
3643        ));
3644    }
3645
3646    #[rstest]
3647    fn test_update_order_applies_updated_event_to_cache(instrument: CryptoPerpetual) {
3648        let (_clock, cache, emulator) = create_emulator();
3649        let risk_events = register_risk_event_handler("RiskEngine.process.updated");
3650        let mut order = create_stop_market_order(&instrument, TriggerType::BidAsk);
3651        let client_order_id = order.client_order_id();
3652        cache
3653            .borrow_mut()
3654            .add_order(order.clone(), None, None, false)
3655            .unwrap();
3656
3657        emulator
3658            .borrow_mut()
3659            .update_order(&mut order, Quantity::from(2));
3660        let cache = cache.borrow();
3661        let cached_order = cache.order(&client_order_id).unwrap();
3662        let risk_events = risk_events.get_messages();
3663
3664        assert_eq!(order.quantity(), Quantity::from(2));
3665        assert_eq!(cached_order.quantity(), Quantity::from(2));
3666        assert_eq!(cached_order.status(), OrderStatus::Initialized);
3667        assert_eq!(risk_events.len(), 1);
3668        assert!(matches!(risk_events[0], OrderEventAny::Updated(_)));
3669    }
3670
3671    #[rstest]
3672    fn test_fill_market_order_applies_released_event_to_cache(instrument: CryptoPerpetual) {
3673        let (_clock, cache, emulator) = create_emulator();
3674        let risk_events = register_risk_event_handler("RiskEngine.process.released");
3675        add_instrument_to_cache(&cache, &instrument);
3676        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
3677        let client_order_id = order.client_order_id();
3678        let strategy_id = order.strategy_id();
3679        let command = create_submit_order(&instrument, &order);
3680        cache
3681            .borrow_mut()
3682            .add_order(order, None, None, false)
3683            .unwrap();
3684
3685        emulator
3686            .borrow_mut()
3687            .cache_submit_order_command(command.clone());
3688        emulator.borrow_mut().handle_submit_order(&command);
3689        risk_events.clear();
3690        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
3691        {
3692            let mut emulator = emulator.borrow_mut();
3693            emulator
3694                .matching_cores
3695                .get_mut(&instrument.id())
3696                .unwrap()
3697                .set_ask_raw(Price::from("5100.00"));
3698            emulator.fill_market_order(client_order_id);
3699        }
3700
3701        msgbus::unsubscribe_order_events(
3702            format!("events.order.{strategy_id}").into(),
3703            &order_handler,
3704        );
3705        let cache = cache.borrow();
3706        let cached_order = cache.order(&client_order_id).unwrap();
3707        let risk_events = risk_events.get_messages();
3708        let order_events = order_events.borrow();
3709
3710        assert_eq!(cached_order.status(), OrderStatus::Released);
3711        assert_eq!(risk_events.len(), 1);
3712        assert!(matches!(risk_events[0], OrderEventAny::Released(_)));
3713        assert_eq!(order_events.len(), 2);
3714        assert!(matches!(order_events[0], OrderEventAny::Initialized(_)));
3715        assert!(matches!(order_events[1], OrderEventAny::Released(_)));
3716    }
3717
3718    #[rstest]
3719    fn test_fill_limit_order_publishes_transformed_initialized_before_released(
3720        instrument: CryptoPerpetual,
3721    ) {
3722        let (_clock, cache, emulator) = create_emulator();
3723        let risk_events = register_risk_event_handler("RiskEngine.process.limit_released");
3724        add_instrument_to_cache(&cache, &instrument);
3725        let order = create_stop_limit_order(&instrument, TriggerType::BidAsk);
3726        let client_order_id = order.client_order_id();
3727        let strategy_id = order.strategy_id();
3728        let command = create_submit_order(&instrument, &order);
3729        cache
3730            .borrow_mut()
3731            .add_order(order, None, None, false)
3732            .unwrap();
3733
3734        emulator
3735            .borrow_mut()
3736            .cache_submit_order_command(command.clone());
3737        emulator.borrow_mut().handle_submit_order(&command);
3738        risk_events.clear();
3739        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
3740        {
3741            let mut emulator = emulator.borrow_mut();
3742            emulator
3743                .matching_cores
3744                .get_mut(&instrument.id())
3745                .unwrap()
3746                .set_ask_raw(Price::from("5100.00"));
3747            emulator.fill_limit_order(client_order_id);
3748        }
3749
3750        msgbus::unsubscribe_order_events(
3751            format!("events.order.{strategy_id}").into(),
3752            &order_handler,
3753        );
3754        let cache = cache.borrow();
3755        let cached_order = cache.order(&client_order_id).unwrap();
3756        let risk_events = risk_events.get_messages();
3757        let order_events = order_events.borrow();
3758
3759        assert_eq!(cached_order.status(), OrderStatus::Released);
3760        assert_eq!(risk_events.len(), 1);
3761        assert!(matches!(risk_events[0], OrderEventAny::Released(_)));
3762        assert_eq!(order_events.len(), 2);
3763        assert!(matches!(order_events[0], OrderEventAny::Initialized(_)));
3764        assert!(matches!(order_events[1], OrderEventAny::Released(_)));
3765    }
3766
3767    fn register_risk_command_handler(id: &str) -> TypedIntoMessageSavingHandler<TradingCommand> {
3768        let (handler, saving_handler) =
3769            get_typed_into_message_saving_handler::<TradingCommand>(Some(Ustr::from(id)));
3770        msgbus::register_trading_command_endpoint(
3771            MessagingSwitchboard::risk_engine_queue_execute(),
3772            handler,
3773        );
3774        saving_handler
3775    }
3776
3777    fn submit_to_emulator(
3778        cache: &Rc<RefCell<Cache>>,
3779        emulator: &Rc<RefCell<OrderEmulator>>,
3780        instrument: &CryptoPerpetual,
3781        order: &OrderAny,
3782    ) {
3783        let command = create_submit_order(instrument, order);
3784        cache
3785            .borrow_mut()
3786            .add_order(order.clone(), None, None, false)
3787            .unwrap();
3788        emulator
3789            .borrow_mut()
3790            .cache_submit_order_command(command.clone());
3791        emulator.borrow_mut().handle_submit_order(&command);
3792    }
3793
3794    #[rstest]
3795    fn test_release_without_market_data_retains_command_and_retries(
3796        instrument: CryptoPerpetual,
3797        #[values(OrderType::StopLimit, OrderType::StopMarket, OrderType::Limit)]
3798        order_type: OrderType,
3799        #[values(OrderSide::Buy, OrderSide::Sell)] side: OrderSide,
3800        #[values(false, true)] use_algorithm: bool,
3801    ) {
3802        let (_clock, cache, emulator) = create_emulator();
3803        let risk_events = register_risk_event_handler("RiskEngine.process.release_retry");
3804        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.release_retry");
3805        let endpoint = Ustr::from("ALG-RETRY.execute");
3806        let (handler, algo_commands) = get_any_saving_handler::<TradingCommand>(Some(endpoint));
3807        msgbus::register_any(endpoint.into(), handler);
3808        add_instrument_to_cache(&cache, &instrument);
3809        let mut builder = OrderTestBuilder::new(order_type);
3810        builder
3811            .instrument_id(instrument.id())
3812            .side(side)
3813            .price(match side {
3814                OrderSide::Buy => Price::from("5102.00"),
3815                OrderSide::Sell => Price::from("5098.00"),
3816            })
3817            .trigger_price(Price::from("5100.00"))
3818            .quantity(Quantity::from(3))
3819            .emulation_trigger(TriggerType::BidAsk);
3820
3821        if use_algorithm {
3822            builder
3823                .exec_algorithm_id(ExecAlgorithmId::from("ALG-RETRY"))
3824                .exec_spawn_id(ClientOrderId::from("O-SPAWN-RETRY"));
3825        }
3826
3827        let order = builder.build();
3828        let client_order_id = order.client_order_id();
3829        let command = create_submit_order(&instrument, &order);
3830        cache
3831            .borrow_mut()
3832            .add_order(order.clone(), None, None, false)
3833            .unwrap();
3834        emulator.borrow_mut().handle_submit_order(&command);
3835        let held_order = cache.borrow().order(&client_order_id).unwrap().clone();
3836        let held_entry = *emulator
3837            .borrow()
3838            .get_matching_core(&instrument.id())
3839            .unwrap()
3840            .get_order(client_order_id)
3841            .unwrap();
3842        risk_events.clear();
3843        let (order_handler, order_events) = subscribe_order_topic(order.strategy_id());
3844
3845        {
3846            let mut emulator = emulator.borrow_mut();
3847            let core = emulator.matching_cores.get_mut(&instrument.id()).unwrap();
3848
3849            // The opposite quote side cannot supply the release price
3850            match side {
3851                OrderSide::Buy => core.set_bid_raw(Price::from("5099.00")),
3852                OrderSide::Sell => core.set_ask_raw(Price::from("5101.00")),
3853            }
3854
3855            match order_type {
3856                OrderType::StopMarket => emulator.fill_market_order(client_order_id),
3857                _ => emulator.fill_limit_order(client_order_id),
3858            }
3859        }
3860
3861        {
3862            let emulator = emulator.borrow();
3863            let cache = cache.borrow();
3864            let cached_order = cache.order(&client_order_id).unwrap();
3865            assert_eq!(cached_order.status(), OrderStatus::Emulated);
3866            assert_eq!(cached_order.order_type(), order_type);
3867            assert_eq!(cached_order.events(), held_order.events());
3868            assert_eq!(
3869                emulator.get_submit_order_commands().get(&client_order_id),
3870                Some(&command),
3871            );
3872            assert_eq!(
3873                emulator
3874                    .get_matching_core(&instrument.id())
3875                    .unwrap()
3876                    .get_order(client_order_id),
3877                Some(&held_entry),
3878            );
3879            assert!(risk_events.get_messages().is_empty());
3880            assert!(order_events.borrow().is_empty());
3881            assert!(risk_commands.get_messages().is_empty());
3882            assert!(algo_commands.get_messages().is_empty());
3883        }
3884
3885        let quote = create_quote_tick(&instrument, "5099.00", "5101.00");
3886        emulator.borrow_mut().on_quote_tick(quote);
3887        emulator.borrow_mut().on_quote_tick(quote);
3888        msgbus::unsubscribe_order_events(
3889            format!("events.order.{}", order.strategy_id()).into(),
3890            &order_handler,
3891        );
3892        let risk_commands = risk_commands.get_messages();
3893        let algo_commands = algo_commands.get_messages();
3894        let (commands, other_commands) = if use_algorithm {
3895            (&algo_commands, &risk_commands)
3896        } else {
3897            (&risk_commands, &algo_commands)
3898        };
3899
3900        let [TradingCommand::SubmitOrder(released_command)] = commands.as_slice() else {
3901            panic!("Expected one released SubmitOrder, was {commands:?}");
3902        };
3903
3904        let emulator = emulator.borrow();
3905        let cache = cache.borrow();
3906        let cached_order = cache.order(&client_order_id).unwrap();
3907        let events = cached_order.events();
3908        let order_events = order_events.borrow();
3909        let mut expected_command = command;
3910        expected_command.order_init = OrderInitialized::from(&*cached_order);
3911
3912        let released_price = match side {
3913            OrderSide::Buy => Price::from("5101.00"),
3914            OrderSide::Sell => Price::from("5099.00"),
3915        };
3916
3917        assert_eq!(cached_order.status(), OrderStatus::Released);
3918        assert_eq!(
3919            cached_order.order_type(),
3920            match order_type {
3921                OrderType::StopLimit => OrderType::Limit,
3922                _ => OrderType::Market,
3923            },
3924        );
3925        assert_eq!(cached_order.emulation_trigger(), None);
3926        assert_eq!(&events[..held_order.event_count()], held_order.events());
3927        assert_eq!(events.len(), held_order.event_count() + 2);
3928        assert_eq!(
3929            order_events.as_slice(),
3930            [
3931                OrderEventAny::Initialized(released_command.order_init.clone()),
3932                cached_order.last_event().clone(),
3933            ],
3934        );
3935        assert!(matches!(
3936            cached_order.last_event(),
3937            OrderEventAny::Released(event) if event.released_price == released_price
3938        ));
3939        assert_eq!(
3940            risk_events.get_messages(),
3941            [cached_order.last_event().clone()]
3942        );
3943        assert_eq!(released_command, &expected_command);
3944        assert!(other_commands.is_empty());
3945        assert!(emulator.get_submit_order_commands().is_empty());
3946        assert_eq!(
3947            emulator
3948                .get_matching_core(&instrument.id())
3949                .unwrap()
3950                .get_order(client_order_id),
3951            None,
3952        );
3953    }
3954
3955    #[rstest]
3956    fn test_release_updates_cache_before_publication_and_defers_reentrant_commands(
3957        instrument: CryptoPerpetual,
3958        #[values(OrderType::StopLimit, OrderType::StopMarket)] order_type: OrderType,
3959    ) {
3960        let (_clock, cache, emulator) = create_emulator();
3961        let risk_commands =
3962            register_risk_command_handler("RiskEngine.queue_execute.release_events");
3963        add_instrument_to_cache(&cache, &instrument);
3964        let order = OrderTestBuilder::new(order_type)
3965            .instrument_id(instrument.id())
3966            .side(OrderSide::Buy)
3967            .price(Price::from("5102.00"))
3968            .trigger_price(Price::from("5100.00"))
3969            .quantity(Quantity::from(3))
3970            .emulation_trigger(TriggerType::BidAsk)
3971            .build();
3972        let sibling = OrderTestBuilder::new(OrderType::StopMarket)
3973            .instrument_id(instrument.id())
3974            .client_order_id(ClientOrderId::from("O-REENTRANT-CANCEL"))
3975            .side(OrderSide::Buy)
3976            .trigger_price(Price::from("5200.00"))
3977            .quantity(Quantity::from(2))
3978            .emulation_trigger(TriggerType::BidAsk)
3979            .build();
3980        submit_to_emulator(&cache, &emulator, &instrument, &order);
3981        submit_to_emulator(&cache, &emulator, &instrument, &sibling);
3982        let client_order_id = order.client_order_id();
3983
3984        let cancel = CancelOrder::new(
3985            sibling.trader_id(),
3986            None,
3987            sibling.strategy_id(),
3988            sibling.instrument_id(),
3989            sibling.client_order_id(),
3990            None,
3991            UUID4::new(),
3992            0.into(),
3993            None,
3994            None,
3995        );
3996        let observed = Rc::new(RefCell::new(Vec::new()));
3997
3998        msgbus::register_trading_command_endpoint(
3999            MessagingSwitchboard::risk_engine_queue_execute(),
4000            TypedIntoHandler::from({
4001                let cache = cache.clone();
4002                let observed = observed.clone();
4003                let exec_handler = risk_commands.handler();
4004                move |command: TradingCommand| {
4005                    let cached_order = cache.borrow().order_owned(&client_order_id).unwrap();
4006                    observed.borrow_mut().push((
4007                        "exec",
4008                        cached_order.status(),
4009                        cached_order.order_type(),
4010                        cached_order.last_event().clone(),
4011                    ));
4012                    exec_handler.handle(command);
4013                }
4014            }),
4015        );
4016
4017        let order_handler = TypedHandler::from({
4018            let cache = cache.clone();
4019            let observed = observed.clone();
4020            move |event: &OrderEventAny| {
4021                if event.client_order_id() != client_order_id {
4022                    return;
4023                }
4024
4025                let cached_order = cache.borrow().order_owned(&client_order_id).unwrap();
4026                observed.borrow_mut().push((
4027                    "order",
4028                    cached_order.status(),
4029                    cached_order.order_type(),
4030                    cached_order.last_event().clone(),
4031                ));
4032
4033                if matches!(event, OrderEventAny::Initialized(_)) {
4034                    msgbus::send_trading_command(
4035                        MessagingSwitchboard::order_emulator_execute(),
4036                        TradingCommand::CancelOrder(cancel.clone()),
4037                    );
4038                }
4039            }
4040        });
4041
4042        msgbus::subscribe_order_events(
4043            format!("events.order.{}", order.strategy_id()).into(),
4044            order_handler.clone(),
4045            None,
4046        );
4047        msgbus::register_order_event_endpoint(
4048            MessagingSwitchboard::risk_engine_process(),
4049            TypedIntoHandler::from({
4050                let cache = cache.clone();
4051                let observed = observed.clone();
4052                move |event: OrderEventAny| {
4053                    if event.client_order_id() != client_order_id {
4054                        return;
4055                    }
4056
4057                    let cached_order = cache.borrow().order_owned(&client_order_id).unwrap();
4058                    observed.borrow_mut().push((
4059                        "risk",
4060                        cached_order.status(),
4061                        cached_order.order_type(),
4062                        cached_order.last_event().clone(),
4063                    ));
4064                }
4065            }),
4066        );
4067
4068        emulator
4069            .borrow_mut()
4070            .on_quote_tick(create_quote_tick(&instrument, "5099.00", "5101.00"));
4071        msgbus::unsubscribe_order_events(
4072            format!("events.order.{}", order.strategy_id()).into(),
4073            &order_handler,
4074        );
4075        let cache = cache.borrow();
4076        let cached_order = cache.order(&client_order_id).unwrap();
4077        let events = cached_order.events();
4078
4079        let released_type = match order_type {
4080            OrderType::StopLimit => OrderType::Limit,
4081            _ => OrderType::Market,
4082        };
4083
4084        assert_eq!(
4085            observed.borrow().as_slice(),
4086            [
4087                (
4088                    "order",
4089                    OrderStatus::Initialized,
4090                    released_type,
4091                    events[events.len() - 2].clone(),
4092                ),
4093                (
4094                    "risk",
4095                    OrderStatus::Released,
4096                    released_type,
4097                    cached_order.last_event().clone(),
4098                ),
4099                (
4100                    "order",
4101                    OrderStatus::Released,
4102                    released_type,
4103                    cached_order.last_event().clone(),
4104                ),
4105                (
4106                    "exec",
4107                    OrderStatus::Released,
4108                    released_type,
4109                    cached_order.last_event().clone(),
4110                ),
4111            ],
4112        );
4113        assert_eq!(
4114            cache.order(&sibling.client_order_id()).unwrap().status(),
4115            OrderStatus::Canceled,
4116        );
4117        assert!(matches!(
4118            risk_commands.get_messages().as_slice(),
4119            [TradingCommand::SubmitOrder(command)] if command.client_order_id == client_order_id
4120        ));
4121        assert!(emulator.borrow().get_submit_order_commands().is_empty());
4122        assert!(
4123            emulator
4124                .borrow()
4125                .get_matching_core(&instrument.id())
4126                .unwrap()
4127                .get_orders()
4128                .is_empty()
4129        );
4130    }
4131
4132    #[rstest]
4133    fn test_release_stops_when_cached_order_rejects_released_event(
4134        instrument: CryptoPerpetual,
4135        #[values(OrderType::StopLimit, OrderType::StopMarket)] order_type: OrderType,
4136    ) {
4137        let (_clock, cache, emulator) = create_emulator();
4138        let risk_events = register_risk_event_handler("RiskEngine.process.release_rejected");
4139        let risk_commands =
4140            register_risk_command_handler("RiskEngine.queue_execute.release_rejected");
4141        add_instrument_to_cache(&cache, &instrument);
4142        let order = OrderTestBuilder::new(order_type)
4143            .instrument_id(instrument.id())
4144            .side(OrderSide::Buy)
4145            .price(Price::from("5102.00"))
4146            .trigger_price(Price::from("5100.00"))
4147            .quantity(Quantity::from(3))
4148            .emulation_trigger(TriggerType::BidAsk)
4149            .build();
4150        submit_to_emulator(&cache, &emulator, &instrument, &order);
4151        let client_order_id = order.client_order_id();
4152        let held_order = cache.borrow().order_owned(&client_order_id).unwrap();
4153        let canceled =
4154            TestOrderEventStubs::canceled(&order, AccountId::from("ACCOUNT-RELEASE"), None);
4155        let observed = Rc::new(RefCell::new(Vec::new()));
4156
4157        let order_handler = TypedHandler::from({
4158            let cache = cache.clone();
4159            let canceled = canceled.clone();
4160            let observed = observed.clone();
4161            move |event: &OrderEventAny| {
4162                observed.borrow_mut().push(event.clone());
4163
4164                if matches!(event, OrderEventAny::Initialized(_)) {
4165                    cache.borrow_mut().update_order(&canceled).unwrap();
4166                }
4167            }
4168        });
4169
4170        msgbus::subscribe_order_events(
4171            format!("events.order.{}", order.strategy_id()).into(),
4172            order_handler.clone(),
4173            None,
4174        );
4175        risk_events.clear();
4176
4177        emulator
4178            .borrow_mut()
4179            .on_quote_tick(create_quote_tick(&instrument, "5099.00", "5101.00"));
4180        msgbus::unsubscribe_order_events(
4181            format!("events.order.{}", order.strategy_id()).into(),
4182            &order_handler,
4183        );
4184
4185        let cache = cache.borrow();
4186        let cached_order = cache.order(&client_order_id).unwrap();
4187        let events = cached_order.events();
4188        let mut expected_events = held_order.events();
4189        expected_events.push(events[held_order.events().len()]);
4190        expected_events.push(&canceled);
4191
4192        assert_eq!(cached_order.status(), OrderStatus::Canceled);
4193        assert_eq!(events, expected_events);
4194        assert!(matches!(
4195            events[events.len() - 2],
4196            OrderEventAny::Initialized(_)
4197        ));
4198        assert_eq!(
4199            observed.borrow().as_slice(),
4200            [events[events.len() - 2].clone()]
4201        );
4202        assert!(risk_events.get_messages().is_empty());
4203        assert!(risk_commands.get_messages().is_empty());
4204        assert!(emulator.borrow().get_submit_order_commands().is_empty());
4205        assert!(
4206            emulator
4207                .borrow()
4208                .get_matching_core(&instrument.id())
4209                .unwrap()
4210                .get_orders()
4211                .is_empty()
4212        );
4213    }
4214
4215    #[rstest]
4216    fn test_fill_limit_order_without_command_retains_order(instrument: CryptoPerpetual) {
4217        let (_clock, cache, emulator) = create_emulator();
4218        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.no_command");
4219        add_instrument_to_cache(&cache, &instrument);
4220        let order = create_stop_limit_order(&instrument, TriggerType::BidAsk);
4221        let client_order_id = order.client_order_id();
4222        submit_to_emulator(&cache, &emulator, &instrument, &order);
4223        let held_order = cache.borrow().order(&client_order_id).unwrap().clone();
4224        let mut emulator = emulator.borrow_mut();
4225        let command = emulator
4226            .manager
4227            .pop_submit_order_command(client_order_id)
4228            .unwrap();
4229        emulator
4230            .matching_cores
4231            .get_mut(&instrument.id())
4232            .unwrap()
4233            .set_ask_raw(Price::from("5101.00"));
4234
4235        emulator.fill_limit_order(client_order_id);
4236
4237        let cache = cache.borrow();
4238        let cached_order = cache.order(&client_order_id).unwrap();
4239        assert_eq!(command.client_order_id, client_order_id);
4240        assert_eq!(cached_order.status(), OrderStatus::Emulated);
4241        assert_eq!(cached_order.order_type(), OrderType::StopLimit);
4242        assert_eq!(cached_order.events(), held_order.events());
4243        assert!(
4244            emulator
4245                .get_matching_core(&instrument.id())
4246                .unwrap()
4247                .order_exists(client_order_id)
4248        );
4249        assert!(risk_commands.get_messages().is_empty());
4250    }
4251
4252    #[rstest]
4253    #[should_panic(expected = "invalid operation `fill_market_order` with no command")]
4254    fn test_fill_market_order_without_command_panics(
4255        instrument: CryptoPerpetual,
4256        #[values(OrderType::StopMarket, OrderType::Limit)] order_type: OrderType,
4257    ) {
4258        let (_clock, cache, emulator) = create_emulator();
4259        add_instrument_to_cache(&cache, &instrument);
4260        let order = OrderTestBuilder::new(order_type)
4261            .instrument_id(instrument.id())
4262            .side(OrderSide::Buy)
4263            .price(Price::from("5102.00"))
4264            .trigger_price(Price::from("5100.00"))
4265            .quantity(Quantity::from(3))
4266            .emulation_trigger(TriggerType::BidAsk)
4267            .build();
4268        let client_order_id = order.client_order_id();
4269        submit_to_emulator(&cache, &emulator, &instrument, &order);
4270        let mut emulator = emulator.borrow_mut();
4271        emulator
4272            .manager
4273            .pop_submit_order_command(client_order_id)
4274            .unwrap();
4275        emulator
4276            .matching_cores
4277            .get_mut(&instrument.id())
4278            .unwrap()
4279            .set_ask_raw(Price::from("5101.00"));
4280
4281        match order_type {
4282            OrderType::Limit => emulator.fill_limit_order(client_order_id),
4283            _ => emulator.fill_market_order(client_order_id),
4284        }
4285    }
4286
4287    #[rstest]
4288    fn test_release_gtd_stop_market_as_gtc_market_order(instrument: CryptoPerpetual) {
4289        let (_clock, cache, emulator) = create_emulator();
4290        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.gtd");
4291        add_instrument_to_cache(&cache, &instrument);
4292        let order = OrderTestBuilder::new(OrderType::StopMarket)
4293            .instrument_id(instrument.id())
4294            .side(OrderSide::Buy)
4295            .trigger_price(Price::from("5100.00"))
4296            .quantity(Quantity::from(1))
4297            .time_in_force(TimeInForce::Gtd)
4298            .expire_time(UnixNanos::from(1_000_000_000_000))
4299            .emulation_trigger(TriggerType::BidAsk)
4300            .build();
4301        let client_order_id = order.client_order_id();
4302        submit_to_emulator(&cache, &emulator, &instrument, &order);
4303
4304        emulator
4305            .borrow_mut()
4306            .on_quote_tick(create_quote_tick(&instrument, "5100.00", "5101.00"));
4307
4308        let cache = cache.borrow();
4309        let cached_order = cache.order(&client_order_id).unwrap();
4310        assert_eq!(cached_order.order_type(), OrderType::Market);
4311        assert_eq!(cached_order.time_in_force(), TimeInForce::Gtc);
4312        assert_eq!(cached_order.status(), OrderStatus::Released);
4313        assert!(matches!(
4314            risk_commands.get_messages().as_slice(),
4315            [TradingCommand::SubmitOrder(command)] if command.client_order_id == client_order_id
4316        ));
4317    }
4318
4319    #[rstest]
4320    fn test_last_price_emulated_stop_triggers_on_trades_not_quotes(
4321        instrument: CryptoPerpetual,
4322        #[values(false, true)] modified: bool,
4323    ) {
4324        let (_clock, cache, emulator) = create_emulator();
4325        let _data_commands = register_data_command_handler("DataEngine.queue_execute.last_price");
4326        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.last_price");
4327        add_instrument_to_cache(&cache, &instrument);
4328
4329        // A quote-triggered order on the same instrument subscribes the emulator to quotes
4330        let quote_order = OrderTestBuilder::new(OrderType::StopMarket)
4331            .instrument_id(instrument.id())
4332            .client_order_id(ClientOrderId::from("O-QUOTE"))
4333            .side(OrderSide::Buy)
4334            .trigger_price(Price::from("5200.00"))
4335            .quantity(Quantity::from(1))
4336            .emulation_trigger(TriggerType::BidAsk)
4337            .build();
4338
4339        // The modified case re-indexes the order through the modify path
4340        let initial_trigger = if modified { "5150.00" } else { "5100.00" };
4341        let last_order = OrderTestBuilder::new(OrderType::StopMarket)
4342            .instrument_id(instrument.id())
4343            .client_order_id(ClientOrderId::from("O-LAST"))
4344            .side(OrderSide::Buy)
4345            .trigger_price(Price::from(initial_trigger))
4346            .quantity(Quantity::from(1))
4347            .emulation_trigger(TriggerType::LastPrice)
4348            .build();
4349        submit_to_emulator(&cache, &emulator, &instrument, &quote_order);
4350        submit_to_emulator(&cache, &emulator, &instrument, &last_order);
4351
4352        if modified {
4353            let modify = ModifyOrder::new(
4354                last_order.trader_id(),
4355                None,
4356                last_order.strategy_id(),
4357                instrument.id(),
4358                last_order.client_order_id(),
4359                None,
4360                None,
4361                None,
4362                Some(Price::from("5100.00")),
4363                UUID4::new(),
4364                0.into(),
4365                None,
4366                None,
4367            );
4368            emulator.borrow_mut().handle_modify_order(&modify);
4369        }
4370
4371        emulator
4372            .borrow_mut()
4373            .on_quote_tick(create_quote_tick(&instrument, "5100.00", "5101.00"));
4374        let commands_after_quote = risk_commands.get_messages().len();
4375        emulator
4376            .borrow_mut()
4377            .on_trade_tick(create_trade_tick(&instrument, "5101.00"));
4378
4379        assert_eq!(commands_after_quote, 0);
4380        assert!(matches!(
4381            risk_commands.get_messages().as_slice(),
4382            [TradingCommand::SubmitOrder(command)]
4383                if command.client_order_id == last_order.client_order_id()
4384        ));
4385    }
4386
4387    #[rstest]
4388    fn test_last_price_emulated_trailing_stop_triggers_on_trades_not_quotes(
4389        instrument: CryptoPerpetual,
4390    ) {
4391        let (_clock, cache, emulator) = create_emulator();
4392        let _data_commands = register_data_command_handler("DataEngine.queue_execute.trailing");
4393        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.trailing");
4394        add_instrument_to_cache(&cache, &instrument);
4395        let quote_order = OrderTestBuilder::new(OrderType::StopMarket)
4396            .instrument_id(instrument.id())
4397            .client_order_id(ClientOrderId::from("O-QUOTE"))
4398            .side(OrderSide::Buy)
4399            .trigger_price(Price::from("5200.00"))
4400            .quantity(Quantity::from(1))
4401            .emulation_trigger(TriggerType::BidAsk)
4402            .build();
4403        let trailing_order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
4404            .instrument_id(instrument.id())
4405            .client_order_id(ClientOrderId::from("O-TRAILING"))
4406            .side(OrderSide::Sell)
4407            .quantity(Quantity::from(1))
4408            .trigger_type(TriggerType::LastPrice)
4409            .trailing_offset(dec!(10))
4410            .trailing_offset_type(TrailingOffsetType::Price)
4411            .emulation_trigger(TriggerType::LastPrice)
4412            .build();
4413        submit_to_emulator(&cache, &emulator, &instrument, &quote_order);
4414        submit_to_emulator(&cache, &emulator, &instrument, &trailing_order);
4415
4416        // The first trade activates the stop and trails its trigger to 4990
4417        emulator
4418            .borrow_mut()
4419            .on_trade_tick(create_trade_tick(&instrument, "5000.00"));
4420        let trigger_price = cache
4421            .borrow()
4422            .order(&trailing_order.client_order_id())
4423            .unwrap()
4424            .trigger_price();
4425        emulator
4426            .borrow_mut()
4427            .on_quote_tick(create_quote_tick(&instrument, "4980.00", "4981.00"));
4428        let commands_after_quote = risk_commands.get_messages().len();
4429        emulator
4430            .borrow_mut()
4431            .on_trade_tick(create_trade_tick(&instrument, "4985.00"));
4432
4433        assert_eq!(trigger_price, Some(Price::from("4990.00")));
4434        assert_eq!(commands_after_quote, 0);
4435        assert!(matches!(
4436            risk_commands.get_messages().as_slice(),
4437            [TradingCommand::SubmitOrder(command)]
4438                if command.client_order_id == trailing_order.client_order_id()
4439        ));
4440    }
4441
4442    #[rstest]
4443    fn test_failed_limit_release_keeps_order_emulated(instrument: CryptoPerpetual) {
4444        let (_clock, cache, emulator) = create_emulator();
4445        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.failed");
4446        add_instrument_to_cache(&cache, &instrument);
4447        let mut order = OrderTestBuilder::new(OrderType::StopLimit)
4448            .instrument_id(instrument.id())
4449            .side(OrderSide::Buy)
4450            .price(Price::from("5100.00"))
4451            .trigger_price(Price::from("5100.00"))
4452            .quantity(Quantity::from(10))
4453            .display_qty(Quantity::from(5))
4454            .emulation_trigger(TriggerType::BidAsk)
4455            .build();
4456        let client_order_id = order.client_order_id();
4457        submit_to_emulator(&cache, &emulator, &instrument, &order);
4458
4459        // A display quantity above the reduced quantity cannot form a limit order on release
4460        emulator
4461            .borrow_mut()
4462            .update_order(&mut order, Quantity::from(3));
4463
4464        emulator
4465            .borrow_mut()
4466            .on_quote_tick(create_quote_tick(&instrument, "5100.00", "5101.00"));
4467
4468        let emulator = emulator.borrow();
4469        let cache = cache.borrow();
4470        assert_eq!(
4471            cache.order(&client_order_id).unwrap().status(),
4472            OrderStatus::Emulated
4473        );
4474        assert!(
4475            emulator
4476                .get_submit_order_commands()
4477                .contains_key(&client_order_id)
4478        );
4479        assert!(
4480            emulator
4481                .get_matching_core(&instrument.id())
4482                .unwrap()
4483                .order_exists(client_order_id)
4484        );
4485        assert!(risk_commands.get_messages().is_empty());
4486    }
4487
4488    #[rstest]
4489    #[case(
4490        OrderType::StopMarket,
4491        OrderType::Market,
4492        TimeInForce::Gtc,
4493        TimeInForce::Gtc
4494    )]
4495    #[case(
4496        OrderType::StopMarket,
4497        OrderType::Market,
4498        TimeInForce::Gtd,
4499        TimeInForce::Gtc
4500    )]
4501    #[case(
4502        OrderType::StopLimit,
4503        OrderType::Limit,
4504        TimeInForce::Gtc,
4505        TimeInForce::Gtc
4506    )]
4507    #[case(
4508        OrderType::StopLimit,
4509        OrderType::Limit,
4510        TimeInForce::Gtd,
4511        TimeInForce::Gtd
4512    )]
4513    fn test_release_order_sends_transformed_command(
4514        instrument: CryptoPerpetual,
4515        #[case] order_type: OrderType,
4516        #[case] released_type: OrderType,
4517        #[case] time_in_force: TimeInForce,
4518        #[case] released_time_in_force: TimeInForce,
4519        #[values(false, true)] use_algorithm: bool,
4520    ) {
4521        let (clock, cache, emulator) = create_emulator();
4522        let release_time = UnixNanos::from(29);
4523        clock
4524            .borrow_mut()
4525            .as_any_mut()
4526            .downcast_mut::<VirtualClock>()
4527            .unwrap()
4528            .advance_time(release_time, true);
4529        let _risk_events = register_risk_event_handler("RiskEngine.process.release_command");
4530        let risk_commands =
4531            register_risk_command_handler("RiskEngine.queue_execute.release_command");
4532        let exec_algorithm_id = ExecAlgorithmId::from("ALG-RELEASE");
4533        let endpoint = Ustr::from("ALG-RELEASE.execute");
4534        let (handler, algo_commands) = get_any_saving_handler::<TradingCommand>(Some(endpoint));
4535        msgbus::register_any(endpoint.into(), handler);
4536        add_instrument_to_cache(&cache, &instrument);
4537        let trigger_instrument = btcusd_bybit();
4538        add_instrument_to_cache(&cache, &trigger_instrument);
4539        let mut builder = OrderTestBuilder::new(order_type);
4540        builder
4541            .instrument_id(instrument.id())
4542            .trigger_instrument_id(trigger_instrument.id())
4543            .side(OrderSide::Buy)
4544            .price(Price::from("5099.00"))
4545            .trigger_price(Price::from("5100.00"))
4546            .quantity(Quantity::from(3))
4547            .ts_init(UnixNanos::from(11))
4548            .time_in_force(time_in_force)
4549            .reduce_only(true)
4550            .quote_quantity(true)
4551            .tags(vec![Ustr::from("release-tag")])
4552            .emulation_trigger(TriggerType::BidAsk);
4553
4554        if order_type == OrderType::StopLimit {
4555            builder.post_only(true).display_qty(Quantity::from(1));
4556        }
4557
4558        if time_in_force == TimeInForce::Gtd {
4559            builder.expire_time(UnixNanos::from(1_000_000_000_000));
4560        }
4561
4562        if use_algorithm {
4563            builder
4564                .exec_algorithm_id(exec_algorithm_id)
4565                .exec_algorithm_params(
4566                    [(Ustr::from("release-algo-param"), Ustr::from("preserved"))].into(),
4567                )
4568                .exec_spawn_id(ClientOrderId::from("O-SPAWN-RELEASE"));
4569        }
4570
4571        let mut order = builder.build();
4572        let client_order_id = order.client_order_id();
4573        let mut command = create_submit_order(&instrument, &order);
4574        command.client_id = Some(ClientId::from("EXECUTION-RELEASE"));
4575        command.position_id = Some(PositionId::from("P-RELEASE"));
4576        command.exec_algorithm_id = order.exec_algorithm_id();
4577        let params = Params::new([("release-param".to_string(), "preserved".into())].into());
4578        command.params = Some(params);
4579        command.ts_init = UnixNanos::from(123);
4580        command.correlation_id = Some(UUID4::new());
4581        command.causation_id = Some(UUID4::new());
4582        let original_command = command.clone();
4583        cache
4584            .borrow_mut()
4585            .add_order(order.clone(), command.position_id, command.client_id, false)
4586            .unwrap();
4587
4588        emulator.borrow_mut().handle_submit_order(&command);
4589        emulator
4590            .borrow_mut()
4591            .update_order(&mut order, Quantity::from(2));
4592        emulator.borrow_mut().on_quote_tick(create_quote_tick(
4593            &trigger_instrument,
4594            "5100.00",
4595            "5101.00",
4596        ));
4597
4598        let risk_commands = risk_commands.get_messages();
4599        let algo_commands = algo_commands.get_messages();
4600        let (commands, other_commands) = if use_algorithm {
4601            (&algo_commands, &risk_commands)
4602        } else {
4603            (&risk_commands, &algo_commands)
4604        };
4605
4606        let [TradingCommand::SubmitOrder(released_command)] = commands.as_slice() else {
4607            panic!("expected one released SubmitOrder, was {commands:?}");
4608        };
4609
4610        let cache = cache.borrow();
4611        let cached_order = cache.order(&client_order_id).unwrap();
4612        let events = cached_order.events();
4613        let mut expected_command = original_command;
4614        let expected_init = &mut expected_command.order_init;
4615        expected_init.order_type = released_type;
4616        expected_init.quantity = Quantity::from(2);
4617        expected_init.time_in_force = released_time_in_force;
4618        expected_init.trigger_price = None;
4619        expected_init.trigger_type = None;
4620        expected_init.emulation_trigger = None;
4621        expected_init.event_id = released_command.order_init.event_id;
4622        expected_init.ts_event = release_time;
4623        expected_init.ts_init = release_time;
4624        if released_type == OrderType::Market {
4625            expected_init.price = None;
4626            expected_init.expire_time = None;
4627            expected_init.trigger_instrument_id = None;
4628        }
4629
4630        assert_eq!(cached_order.status(), OrderStatus::Released);
4631        assert_eq!(
4632            events[events.len() - 2],
4633            &OrderEventAny::Initialized(released_command.order_init.clone()),
4634        );
4635        assert_eq!(released_command, &expected_command);
4636        assert_eq!(
4637            OrderInitialized::from(&*cached_order),
4638            expected_command.order_init
4639        );
4640        assert!(other_commands.is_empty());
4641    }
4642
4643    #[rstest]
4644    fn test_quote_tick_updates_matching_core_prices(instrument: CryptoPerpetual) {
4645        let (_clock, cache, emulator) = create_emulator();
4646        add_instrument_to_cache(&cache, &instrument);
4647        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
4648        let command = create_submit_order(&instrument, &order);
4649        cache
4650            .borrow_mut()
4651            .add_order(order, None, None, false)
4652            .unwrap();
4653        emulator
4654            .borrow_mut()
4655            .cache_submit_order_command(command.clone());
4656        emulator.borrow_mut().handle_submit_order(&command);
4657
4658        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
4659        emulator.borrow_mut().on_quote_tick(quote);
4660
4661        let core = emulator
4662            .borrow()
4663            .get_matching_core(&instrument.id())
4664            .unwrap();
4665        assert_eq!(core.bid, Some(Price::from("5060.00")));
4666        assert_eq!(core.ask, Some(Price::from("5070.00")));
4667    }
4668
4669    #[rstest]
4670    fn test_trade_tick_updates_matching_core_last_price(instrument: CryptoPerpetual) {
4671        let (_clock, cache, emulator) = create_emulator();
4672        add_instrument_to_cache(&cache, &instrument);
4673        let order = create_stop_market_order(&instrument, TriggerType::LastPrice);
4674        let command = create_submit_order(&instrument, &order);
4675        cache
4676            .borrow_mut()
4677            .add_order(order, None, None, false)
4678            .unwrap();
4679        emulator
4680            .borrow_mut()
4681            .cache_submit_order_command(command.clone());
4682        emulator.borrow_mut().handle_submit_order(&command);
4683
4684        let trade = create_trade_tick(&instrument, "5065.00");
4685        emulator.borrow_mut().on_trade_tick(trade);
4686
4687        let core = emulator
4688            .borrow()
4689            .get_matching_core(&instrument.id())
4690            .unwrap();
4691        assert_eq!(core.last, Some(Price::from("5065.00")));
4692    }
4693
4694    #[rstest]
4695    fn test_cancel_order_removes_from_matching_core(instrument: CryptoPerpetual) {
4696        let (_clock, cache, emulator) = create_emulator();
4697        add_instrument_to_cache(&cache, &instrument);
4698        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
4699        let command = create_submit_order(&instrument, &order);
4700        cache
4701            .borrow_mut()
4702            .add_order(order.clone(), None, None, false)
4703            .unwrap();
4704        emulator
4705            .borrow_mut()
4706            .cache_submit_order_command(command.clone());
4707        emulator.borrow_mut().handle_submit_order(&command);
4708
4709        emulator.borrow_mut().cancel_order(&order);
4710
4711        let core = emulator
4712            .borrow()
4713            .get_matching_core(&instrument.id())
4714            .unwrap();
4715        assert!(core.get_orders().is_empty());
4716    }
4717
4718    #[rstest]
4719    fn test_cancel_order_removes_cached_command(instrument: CryptoPerpetual) {
4720        let (_clock, cache, emulator) = create_emulator();
4721        add_instrument_to_cache(&cache, &instrument);
4722        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
4723        let client_order_id = order.client_order_id();
4724        let command = create_submit_order(&instrument, &order);
4725        cache
4726            .borrow_mut()
4727            .add_order(order.clone(), None, None, false)
4728            .unwrap();
4729        emulator
4730            .borrow_mut()
4731            .cache_submit_order_command(command.clone());
4732        emulator.borrow_mut().handle_submit_order(&command);
4733
4734        emulator.borrow_mut().cancel_order(&order);
4735
4736        let commands = emulator.borrow().get_submit_order_commands();
4737        assert!(!commands.contains_key(&client_order_id));
4738    }
4739
4740    #[rstest]
4741    fn test_trailing_stop_waits_for_activation_price(instrument: CryptoPerpetual) {
4742        let (_clock, cache, emulator) = create_emulator();
4743        let risk_events = register_risk_event_handler("RiskEngine.process.trailing_activation");
4744        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4745            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4746        msgbus::register_trading_command_endpoint(
4747            MessagingSwitchboard::risk_engine_queue_execute(),
4748            handler,
4749        );
4750        add_instrument_to_cache(&cache, &instrument);
4751
4752        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
4753        core.set_bid_raw(Price::from("5000.00"));
4754        core.set_ask_raw(Price::from("5001.00"));
4755        emulator
4756            .borrow_mut()
4757            .matching_cores
4758            .insert(instrument.id(), core);
4759
4760        let order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
4761            .instrument_id(instrument.id())
4762            .client_order_id(ClientOrderId::from("O-TRAILING-1"))
4763            .side(OrderSide::Sell)
4764            .quantity(Quantity::from(1))
4765            .activation_price(Price::from("5050.00"))
4766            .trailing_offset(dec!(10))
4767            .trailing_offset_type(TrailingOffsetType::Price)
4768            .emulation_trigger(TriggerType::BidAsk)
4769            .build();
4770        let client_order_id = order.client_order_id();
4771        let command = create_submit_order(&instrument, &order);
4772        cache
4773            .borrow_mut()
4774            .add_order(order, None, None, false)
4775            .unwrap();
4776
4777        emulator
4778            .borrow_mut()
4779            .cache_submit_order_command(command.clone());
4780        emulator.borrow_mut().handle_submit_order(&command);
4781
4782        {
4783            let cache = cache.borrow();
4784            let cached = cache.order(&client_order_id).unwrap();
4785            let is_activated = match &*cached {
4786                OrderAny::TrailingStopMarket(o) => o.is_activated,
4787                _ => panic!("expected trailing stop market order"),
4788            };
4789            assert!(
4790                !is_activated,
4791                "must not activate before the activation price trades"
4792            );
4793            assert!(cached.trigger_price().is_none());
4794        }
4795        assert!(
4796            !risk_events
4797                .get_messages()
4798                .iter()
4799                .any(|e| matches!(e, OrderEventAny::Updated(_))),
4800            "no trailing update before activation"
4801        );
4802
4803        // Market rallies through the 5050 activation price: activates and starts trailing
4804        emulator
4805            .borrow_mut()
4806            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));
4807
4808        {
4809            let cache = cache.borrow();
4810            let cached = cache.order(&client_order_id).unwrap();
4811            let is_activated = match &*cached {
4812                OrderAny::TrailingStopMarket(o) => o.is_activated,
4813                _ => panic!("expected trailing stop market order"),
4814            };
4815            assert!(is_activated);
4816            assert_eq!(cached.trigger_price(), Some(Price::from("5045.00")));
4817        }
4818
4819        assert!(risk_commands.get_messages().is_empty());
4820
4821        // Market falls through the trailed trigger: order releases
4822        emulator
4823            .borrow_mut()
4824            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));
4825
4826        let commands = risk_commands.get_messages();
4827        assert_eq!(commands.len(), 1);
4828        assert!(matches!(
4829            &commands[0],
4830            TradingCommand::SubmitOrder(command) if command.client_order_id == client_order_id
4831        ));
4832    }
4833
4834    #[rstest]
4835    fn test_pending_activation_trailing_stop_limit_not_matched_as_plain_limit(
4836        instrument: CryptoPerpetual,
4837    ) {
4838        let (_clock, cache, emulator) = create_emulator();
4839        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4840            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4841        msgbus::register_trading_command_endpoint(
4842            MessagingSwitchboard::risk_engine_queue_execute(),
4843            handler,
4844        );
4845        add_instrument_to_cache(&cache, &instrument);
4846
4847        // Market is already through the limit price, but below the activation price
4848        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
4849        core.set_bid_raw(Price::from("5000.00"));
4850        core.set_ask_raw(Price::from("5001.00"));
4851        emulator
4852            .borrow_mut()
4853            .matching_cores
4854            .insert(instrument.id(), core);
4855
4856        let order = OrderTestBuilder::new(OrderType::TrailingStopLimit)
4857            .instrument_id(instrument.id())
4858            .client_order_id(ClientOrderId::from("O-TRAILING-LIMIT-1"))
4859            .side(OrderSide::Sell)
4860            .price(Price::from("5010.00"))
4861            .quantity(Quantity::from(1))
4862            .activation_price(Price::from("5050.00"))
4863            .trailing_offset(dec!(10))
4864            .trailing_offset_type(TrailingOffsetType::Price)
4865            .limit_offset(dec!(5))
4866            .emulation_trigger(TriggerType::BidAsk)
4867            .build();
4868        let client_order_id = order.client_order_id();
4869        let command = create_submit_order(&instrument, &order);
4870        cache
4871            .borrow_mut()
4872            .add_order(order, None, None, false)
4873            .unwrap();
4874
4875        emulator
4876            .borrow_mut()
4877            .cache_submit_order_command(command.clone());
4878        emulator.borrow_mut().handle_submit_order(&command);
4879
4880        // The market falls below where the trailed trigger would sit (4990)
4881        // while staying below the 5050 activation price: must not release.
4882        emulator
4883            .borrow_mut()
4884            .on_quote_tick(create_quote_tick(&instrument, "4985.00", "4986.00"));
4885
4886        assert!(risk_commands.get_messages().is_empty());
4887        assert!(
4888            emulator
4889                .borrow()
4890                .get_submit_order_commands()
4891                .contains_key(&client_order_id),
4892            "order must remain held by the emulator before activation"
4893        );
4894    }
4895
4896    #[rstest]
4897    fn test_trailing_stop_with_preset_trigger_activates_and_triggers(instrument: CryptoPerpetual) {
4898        let (_clock, cache, emulator) = create_emulator();
4899        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4900            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4901        msgbus::register_trading_command_endpoint(
4902            MessagingSwitchboard::risk_engine_queue_execute(),
4903            handler,
4904        );
4905        add_instrument_to_cache(&cache, &instrument);
4906
4907        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
4908        core.set_bid_raw(Price::from("5000.00"));
4909        core.set_ask_raw(Price::from("5001.00"));
4910        emulator
4911            .borrow_mut()
4912            .matching_cores
4913            .insert(instrument.id(), core);
4914
4915        // Both prices preset: the 5045 trigger must stay inert until 5050 activates
4916        let order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
4917            .instrument_id(instrument.id())
4918            .client_order_id(ClientOrderId::from("O-TRAILING-2"))
4919            .side(OrderSide::Sell)
4920            .quantity(Quantity::from(1))
4921            .trigger_price(Price::from("5045.00"))
4922            .activation_price(Price::from("5050.00"))
4923            .trailing_offset(dec!(10))
4924            .trailing_offset_type(TrailingOffsetType::Price)
4925            .emulation_trigger(TriggerType::BidAsk)
4926            .build();
4927        let client_order_id = order.client_order_id();
4928        let command = create_submit_order(&instrument, &order);
4929        cache
4930            .borrow_mut()
4931            .add_order(order, None, None, false)
4932            .unwrap();
4933
4934        emulator
4935            .borrow_mut()
4936            .cache_submit_order_command(command.clone());
4937        emulator.borrow_mut().handle_submit_order(&command);
4938
4939        assert!(
4940            risk_commands.get_messages().is_empty(),
4941            "preset trigger must not release before activation"
4942        );
4943
4944        // Rally touches the 5050 activation price; the trailed candidate does
4945        // not improve on the preset trigger, so no price update is emitted.
4946        emulator
4947            .borrow_mut()
4948            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));
4949
4950        assert!(risk_commands.get_messages().is_empty());
4951
4952        // Bid falls through the preset trigger: the order must release
4953        emulator
4954            .borrow_mut()
4955            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));
4956
4957        let commands = risk_commands.get_messages();
4958        assert_eq!(commands.len(), 1);
4959        assert!(matches!(
4960            &commands[0],
4961            TradingCommand::SubmitOrder(command) if command.client_order_id == client_order_id
4962        ));
4963    }
4964
4965    #[rstest]
4966    fn test_trailing_stop_activates_despite_trailing_calculate_error(instrument: CryptoPerpetual) {
4967        let (_clock, cache, emulator) = create_emulator();
4968        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4969            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4970        msgbus::register_trading_command_endpoint(
4971            MessagingSwitchboard::risk_engine_queue_execute(),
4972            handler,
4973        );
4974        add_instrument_to_cache(&cache, &instrument);
4975
4976        // Quote-only data: `LastOrBidAsk` trailing calculation errors without a last
4977        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
4978        core.set_bid_raw(Price::from("5000.00"));
4979        core.set_ask_raw(Price::from("5001.00"));
4980        emulator
4981            .borrow_mut()
4982            .matching_cores
4983            .insert(instrument.id(), core);
4984
4985        let order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
4986            .instrument_id(instrument.id())
4987            .client_order_id(ClientOrderId::from("O-TRAILING-3"))
4988            .side(OrderSide::Sell)
4989            .quantity(Quantity::from(1))
4990            .trigger_price(Price::from("5045.00"))
4991            .trigger_type(TriggerType::LastOrBidAsk)
4992            .activation_price(Price::from("5050.00"))
4993            .trailing_offset(dec!(10))
4994            .trailing_offset_type(TrailingOffsetType::Price)
4995            .emulation_trigger(TriggerType::BidAsk)
4996            .build();
4997        let client_order_id = order.client_order_id();
4998        let command = create_submit_order(&instrument, &order);
4999        cache
5000            .borrow_mut()
5001            .add_order(order, None, None, false)
5002            .unwrap();
5003
5004        emulator
5005            .borrow_mut()
5006            .cache_submit_order_command(command.clone());
5007        emulator.borrow_mut().handle_submit_order(&command);
5008
5009        // Activation touches while the trailing calculation errors: the preset
5010        // trigger must still be honored once the market crosses it.
5011        emulator
5012            .borrow_mut()
5013            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));
5014        emulator
5015            .borrow_mut()
5016            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));
5017
5018        let commands = risk_commands.get_messages();
5019        assert_eq!(commands.len(), 1);
5020        assert!(matches!(
5021            &commands[0],
5022            TradingCommand::SubmitOrder(command) if command.client_order_id == client_order_id
5023        ));
5024    }
5025
5026    #[rstest]
5027    fn test_basis_point_trailing_stop_limit_releases_on_price_increment(
5028        mut instrument: CryptoPerpetual,
5029    ) {
5030        instrument.price_increment = Price::from("0.25");
5031        let (_clock, cache, emulator) = create_emulator();
5032        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5033            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5034        msgbus::register_trading_command_endpoint(
5035            MessagingSwitchboard::risk_engine_queue_execute(),
5036            handler,
5037        );
5038        add_instrument_to_cache(&cache, &instrument);
5039
5040        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
5041        core.set_bid_raw(Price::from("1500.00"));
5042        core.set_ask_raw(Price::from("1500.25"));
5043        emulator
5044            .borrow_mut()
5045            .matching_cores
5046            .insert(instrument.id(), core);
5047
5048        let order = OrderTestBuilder::new(OrderType::TrailingStopLimit)
5049            .instrument_id(instrument.id())
5050            .client_order_id(ClientOrderId::from("O-TRAILING-LIMIT-BPS"))
5051            .side(OrderSide::Sell)
5052            .quantity(Quantity::from(1))
5053            .trigger_type(TriggerType::BidAsk)
5054            .trailing_offset(dec!(33))
5055            .limit_offset(dec!(37))
5056            .trailing_offset_type(TrailingOffsetType::BasisPoints)
5057            .emulation_trigger(TriggerType::BidAsk)
5058            .build();
5059        let client_order_id = order.client_order_id();
5060        let command = create_submit_order(&instrument, &order);
5061        cache
5062            .borrow_mut()
5063            .add_order(order, None, None, false)
5064            .unwrap();
5065
5066        emulator
5067            .borrow_mut()
5068            .cache_submit_order_command(command.clone());
5069        emulator.borrow_mut().handle_submit_order(&command);
5070        emulator
5071            .borrow_mut()
5072            .on_quote_tick(create_quote_tick(&instrument, "1500.00", "1500.25"));
5073
5074        // From the 1500.00 bid, 33 bps is 1495.05 and 37 bps is 1494.45, both between increments
5075        let trailed_trigger = cache
5076            .borrow()
5077            .order(&client_order_id)
5078            .unwrap()
5079            .trigger_price();
5080        let trailed_price = cache.borrow().order(&client_order_id).unwrap().price();
5081
5082        emulator
5083            .borrow_mut()
5084            .on_quote_tick(create_quote_tick(&instrument, "1494.00", "1494.25"));
5085
5086        assert_eq!(trailed_trigger, Some(Price::from("1495.00")));
5087        assert_eq!(trailed_price, Some(Price::from("1494.50")));
5088        let commands = risk_commands.get_messages();
5089
5090        let [TradingCommand::SubmitOrder(released)] = commands.as_slice() else {
5091            panic!("expected one released SubmitOrder, was {commands:?}");
5092        };
5093
5094        assert_eq!(released.client_order_id, client_order_id);
5095        let cache = cache.borrow();
5096        let released_order = cache.order(&client_order_id).unwrap();
5097        assert_eq!(released_order.order_type(), OrderType::Limit);
5098        assert_eq!(released_order.price(), Some(Price::from("1494.50")));
5099    }
5100
5101    #[rstest]
5102    fn test_trailing_stop_limit_without_limit_price_triggered_after_submit(
5103        instrument: CryptoPerpetual,
5104    ) {
5105        let (_clock, cache, emulator) = create_emulator();
5106        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5107            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5108        msgbus::register_trading_command_endpoint(
5109            MessagingSwitchboard::risk_engine_queue_execute(),
5110            handler,
5111        );
5112        add_instrument_to_cache(&cache, &instrument);
5113
5114        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
5115        core.set_bid_raw(Price::from("5000.00"));
5116        core.set_ask_raw(Price::from("5001.00"));
5117        emulator
5118            .borrow_mut()
5119            .matching_cores
5120            .insert(instrument.id(), core);
5121
5122        let order = OrderTestBuilder::new(OrderType::TrailingStopLimit)
5123            .instrument_id(instrument.id())
5124            .client_order_id(ClientOrderId::from("O-TRAILING-LIMIT-1"))
5125            .side(OrderSide::Sell)
5126            .quantity(Quantity::from(1))
5127            .trigger_price(Price::from("5045.00"))
5128            .trigger_type(TriggerType::LastOrBidAsk)
5129            .activation_price(Price::from("5050.00"))
5130            .trailing_offset(dec!(10))
5131            .limit_offset(dec!(5))
5132            .trailing_offset_type(TrailingOffsetType::Price)
5133            .emulation_trigger(TriggerType::BidAsk)
5134            .build();
5135        let client_order_id = order.client_order_id();
5136        let command = create_submit_order(&instrument, &order);
5137        cache
5138            .borrow_mut()
5139            .add_order(order, None, None, false)
5140            .unwrap();
5141
5142        emulator
5143            .borrow_mut()
5144            .cache_submit_order_command(command.clone());
5145        emulator.borrow_mut().handle_submit_order(&command);
5146
5147        // Quote-only data: activation touches but `LastOrBidAsk` cannot price the limit
5148        // without a last, so the crossed trigger must hold the order rather than release it.
5149        emulator
5150            .borrow_mut()
5151            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));
5152        emulator
5153            .borrow_mut()
5154            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));
5155
5156        let held_price = cache.borrow().order(&client_order_id).unwrap().price();
5157        let held_commands = risk_commands.get_messages().len();
5158
5159        // The first trade prices the limit, and the next update releases it
5160        emulator
5161            .borrow_mut()
5162            .on_trade_tick(create_trade_tick(&instrument, "5044.00"));
5163        emulator
5164            .borrow_mut()
5165            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));
5166
5167        assert_eq!(held_price, None);
5168        assert_eq!(held_commands, 0);
5169        let commands = risk_commands.get_messages();
5170
5171        let [TradingCommand::SubmitOrder(released)] = commands.as_slice() else {
5172            panic!("expected one released SubmitOrder, was {commands:?}");
5173        };
5174
5175        assert_eq!(released.client_order_id, client_order_id);
5176        let cache = cache.borrow();
5177        let released_order = cache.order(&client_order_id).unwrap();
5178        assert_eq!(released_order.order_type(), OrderType::Limit);
5179        assert_eq!(released_order.price(), Some(Price::from("5039.00")));
5180    }
5181
5182    #[rstest]
5183    fn test_trailing_stop_limit_without_limit_price_triggered_by_modify(
5184        instrument: CryptoPerpetual,
5185    ) {
5186        let (_clock, cache, emulator) = create_emulator();
5187        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5188            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5189        msgbus::register_trading_command_endpoint(
5190            MessagingSwitchboard::risk_engine_queue_execute(),
5191            handler,
5192        );
5193        add_instrument_to_cache(&cache, &instrument);
5194
5195        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
5196        core.set_bid_raw(Price::from("5000.00"));
5197        core.set_ask_raw(Price::from("5001.00"));
5198        emulator
5199            .borrow_mut()
5200            .matching_cores
5201            .insert(instrument.id(), core);
5202
5203        let order = OrderTestBuilder::new(OrderType::TrailingStopLimit)
5204            .instrument_id(instrument.id())
5205            .client_order_id(ClientOrderId::from("O-TRAILING-LIMIT-2"))
5206            .side(OrderSide::Sell)
5207            .quantity(Quantity::from(1))
5208            .trigger_type(TriggerType::LastOrBidAsk)
5209            .activation_price(Price::from("5050.00"))
5210            .trailing_offset(dec!(10))
5211            .limit_offset(dec!(5))
5212            .trailing_offset_type(TrailingOffsetType::Price)
5213            .emulation_trigger(TriggerType::BidAsk)
5214            .build();
5215        let client_order_id = order.client_order_id();
5216        let command = create_submit_order(&instrument, &order);
5217        cache
5218            .borrow_mut()
5219            .add_order(order.clone(), None, None, false)
5220            .unwrap();
5221
5222        emulator
5223            .borrow_mut()
5224            .cache_submit_order_command(command.clone());
5225        emulator.borrow_mut().handle_submit_order(&command);
5226        let _exec_events =
5227            register_exec_event_handler(cache.clone(), "ExecEngine.process.trailing_modify");
5228        emulator
5229            .borrow_mut()
5230            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));
5231
5232        let modify = ModifyOrder::new(
5233            order.trader_id(),
5234            None,
5235            order.strategy_id(),
5236            instrument.id(),
5237            client_order_id,
5238            None,
5239            None,
5240            None,
5241            Some(Price::from("5060.00")),
5242            UUID4::new(),
5243            0.into(),
5244            None,
5245            None,
5246        );
5247        emulator.borrow_mut().handle_modify_order(&modify);
5248
5249        let held_commands = risk_commands.get_messages().len();
5250        let held_in_core =
5251            emulator.borrow().matching_cores[&instrument.id()].order_exists(client_order_id);
5252
5253        let (held_status, held_trigger, held_price) = {
5254            let cache = cache.borrow();
5255            let held_order = cache.order(&client_order_id).unwrap();
5256            (
5257                held_order.status(),
5258                held_order.trigger_price(),
5259                held_order.price(),
5260            )
5261        };
5262
5263        // The first trade prices the limit, and the next update releases it
5264        emulator
5265            .borrow_mut()
5266            .on_trade_tick(create_trade_tick(&instrument, "5055.00"));
5267        emulator
5268            .borrow_mut()
5269            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));
5270
5271        assert_eq!(held_commands, 0);
5272        assert!(held_in_core);
5273        assert_eq!(held_status, OrderStatus::Emulated);
5274        assert_eq!(held_trigger, Some(Price::from("5060.00")));
5275        assert_eq!(held_price, None);
5276        let commands = risk_commands.get_messages();
5277
5278        let [TradingCommand::SubmitOrder(released)] = commands.as_slice() else {
5279            panic!("expected one released SubmitOrder, was {commands:?}");
5280        };
5281
5282        assert_eq!(released.client_order_id, client_order_id);
5283        let cache = cache.borrow();
5284        let released_order = cache.order(&client_order_id).unwrap();
5285        assert_eq!(released_order.order_type(), OrderType::Limit);
5286        assert_eq!(released_order.price(), Some(Price::from("5050.00")));
5287    }
5288
5289    #[rstest]
5290    fn test_cancel_emulated_oco_leg_cancels_sibling(instrument: CryptoPerpetual) {
5291        let (_clock, cache, emulator) = create_emulator();
5292        add_instrument_to_cache(&cache, &instrument);
5293        let client_order_id_a = ClientOrderId::from("O-OCO-A");
5294        let client_order_id_b = ClientOrderId::from("O-OCO-B");
5295        let order_a = OrderTestBuilder::new(OrderType::StopMarket)
5296            .instrument_id(instrument.id())
5297            .strategy_id(StrategyId::from("STRATEGY-001"))
5298            .client_order_id(client_order_id_a)
5299            .side(OrderSide::Sell)
5300            .trigger_price(Price::from("4900.00"))
5301            .quantity(Quantity::from(1))
5302            .emulation_trigger(TriggerType::BidAsk)
5303            .contingency_type(ContingencyType::Oco)
5304            .linked_order_ids(vec![client_order_id_b])
5305            .build();
5306        let order_b = OrderTestBuilder::new(OrderType::StopMarket)
5307            .instrument_id(instrument.id())
5308            .strategy_id(StrategyId::from("STRATEGY-001"))
5309            .client_order_id(client_order_id_b)
5310            .side(OrderSide::Sell)
5311            .trigger_price(Price::from("4950.00"))
5312            .quantity(Quantity::from(1))
5313            .emulation_trigger(TriggerType::BidAsk)
5314            .contingency_type(ContingencyType::Oco)
5315            .linked_order_ids(vec![client_order_id_a])
5316            .build();
5317        cache
5318            .borrow_mut()
5319            .add_order(order_a.clone(), None, None, false)
5320            .unwrap();
5321        cache
5322            .borrow_mut()
5323            .add_order(order_b.clone(), None, None, false)
5324            .unwrap();
5325
5326        for order in [&order_a, &order_b] {
5327            let command = create_submit_order(&instrument, order);
5328            emulator
5329                .borrow_mut()
5330                .cache_submit_order_command(command.clone());
5331            emulator.borrow_mut().handle_submit_order(&command);
5332        }
5333        assert_eq!(
5334            cache.borrow().order(&client_order_id_a).unwrap().status(),
5335            OrderStatus::Emulated
5336        );
5337        assert_eq!(
5338            cache.borrow().order(&client_order_id_b).unwrap().status(),
5339            OrderStatus::Emulated
5340        );
5341
5342        let cancel = CancelOrder::new(
5343            order_a.trader_id(),
5344            None,
5345            order_a.strategy_id(),
5346            instrument.id(),
5347            client_order_id_a,
5348            None,
5349            UUID4::new(),
5350            0.into(),
5351            None,
5352            None,
5353        );
5354        msgbus::send_trading_command(
5355            MessagingSwitchboard::order_emulator_execute(),
5356            TradingCommand::CancelOrder(cancel),
5357        );
5358
5359        let cache = cache.borrow();
5360        assert_eq!(
5361            cache.order(&client_order_id_a).unwrap().status(),
5362            OrderStatus::Canceled
5363        );
5364        assert_eq!(
5365            cache.order(&client_order_id_b).unwrap().status(),
5366            OrderStatus::Canceled,
5367            "OCO sibling must be canceled through the deferred self-published event"
5368        );
5369    }
5370
5371    #[rstest]
5372    fn test_reentrant_command_from_event_handler_does_not_panic(instrument: CryptoPerpetual) {
5373        let (_clock, cache, emulator) = create_emulator();
5374        add_instrument_to_cache(&cache, &instrument);
5375        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
5376        let client_order_id = order.client_order_id();
5377        let strategy_id = order.strategy_id();
5378        let command = create_submit_order(&instrument, &order);
5379        cache
5380            .borrow_mut()
5381            .add_order(order.clone(), None, None, false)
5382            .unwrap();
5383        emulator
5384            .borrow_mut()
5385            .cache_submit_order_command(command.clone());
5386
5387        // A strategy-style subscriber issuing an emulator command synchronously
5388        // from within its own order-event handler.
5389        let cancel = CancelOrder::new(
5390            order.trader_id(),
5391            None,
5392            strategy_id,
5393            instrument.id(),
5394            client_order_id,
5395            None,
5396            UUID4::new(),
5397            0.into(),
5398            None,
5399            None,
5400        );
5401        let handler = TypedHandler::from(move |event: &OrderEventAny| {
5402            if matches!(event, OrderEventAny::Emulated(_)) {
5403                msgbus::send_trading_command(
5404                    MessagingSwitchboard::order_emulator_execute(),
5405                    TradingCommand::CancelOrder(cancel.clone()),
5406                );
5407            }
5408        });
5409        msgbus::subscribe_order_events(format!("events.order.{strategy_id}").into(), handler, None);
5410
5411        // Must not panic on the reentrant borrow; the command is deferred and drained
5412        msgbus::send_trading_command(
5413            MessagingSwitchboard::order_emulator_execute(),
5414            TradingCommand::SubmitOrder(command),
5415        );
5416
5417        assert!(
5418            cache.borrow().order(&client_order_id).unwrap().is_closed(),
5419            "deferred cancel must be processed after the active call completes"
5420        );
5421    }
5422
5423    #[rstest]
5424    fn test_released_order_preserves_chronological_event_history(instrument: CryptoPerpetual) {
5425        let (_clock, cache, emulator) = create_emulator();
5426        add_instrument_to_cache(&cache, &instrument);
5427        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
5428        core.set_bid_raw(Price::from("5099.00"));
5429        core.set_ask_raw(Price::from("5101.00"));
5430        emulator
5431            .borrow_mut()
5432            .matching_cores
5433            .insert(instrument.id(), core);
5434
5435        let order = create_stop_limit_order(&instrument, TriggerType::BidAsk);
5436        let client_order_id = order.client_order_id();
5437        let original_init = OrderEventAny::Initialized(order.init_event().clone());
5438        let command = create_submit_order(&instrument, &order);
5439        cache
5440            .borrow_mut()
5441            .add_order(order, None, None, false)
5442            .unwrap();
5443
5444        emulator
5445            .borrow_mut()
5446            .cache_submit_order_command(command.clone());
5447        emulator.borrow_mut().handle_submit_order(&command);
5448
5449        let cache = cache.borrow();
5450        let transformed = cache.order(&client_order_id).unwrap();
5451        let events = transformed.events();
5452
5453        assert_eq!(
5454            events.first(),
5455            Some(&&original_init),
5456            "released order history must start with the original OrderInitialized event, was {events:?}",
5457        );
5458    }
5459
5460    #[rstest]
5461    fn test_on_start_cancels_emulated_child_of_closed_positionless_parent(
5462        instrument: CryptoPerpetual,
5463    ) {
5464        let (_clock, cache, emulator) = create_emulator();
5465        add_instrument_to_cache(&cache, &instrument);
5466
5467        let mut parent = OrderTestBuilder::new(OrderType::Limit)
5468            .instrument_id(instrument.id())
5469            .client_order_id(ClientOrderId::from("O-PARENT"))
5470            .side(OrderSide::Buy)
5471            .price(Price::from("5000.00"))
5472            .quantity(Quantity::from(1))
5473            .submit(true)
5474            .build();
5475        parent
5476            .apply(OrderEventAny::Canceled(OrderCanceled::new(
5477                parent.trader_id(),
5478                parent.strategy_id(),
5479                parent.instrument_id(),
5480                parent.client_order_id(),
5481                UUID4::new(),
5482                0.into(),
5483                0.into(),
5484                false,
5485                None,
5486                None,
5487                None,
5488            )))
5489            .unwrap();
5490        assert!(parent.is_closed());
5491        assert!(parent.position_id().is_none());
5492
5493        let mut child = OrderTestBuilder::new(OrderType::StopMarket)
5494            .instrument_id(instrument.id())
5495            .client_order_id(ClientOrderId::from("O-CHILD"))
5496            .side(OrderSide::Sell)
5497            .trigger_price(Price::from("4900.00"))
5498            .quantity(Quantity::from(1))
5499            .emulation_trigger(TriggerType::BidAsk)
5500            .parent_order_id(parent.client_order_id())
5501            .build();
5502        child
5503            .apply(OrderEventAny::Emulated(OrderEmulated::new(
5504                child.trader_id(),
5505                child.strategy_id(),
5506                child.instrument_id(),
5507                child.client_order_id(),
5508                UUID4::new(),
5509                0.into(),
5510                0.into(),
5511            )))
5512            .unwrap();
5513        assert_eq!(child.status(), OrderStatus::Emulated);
5514
5515        cache
5516            .borrow_mut()
5517            .add_order(parent.clone(), None, None, false)
5518            .unwrap();
5519        cache
5520            .borrow_mut()
5521            .add_order(child.clone(), None, None, false)
5522            .unwrap();
5523
5524        emulator.borrow_mut().on_start().unwrap();
5525
5526        assert!(
5527            cache
5528                .borrow()
5529                .order(&child.client_order_id())
5530                .unwrap()
5531                .is_closed(),
5532            "emulated child of a closed position-less parent must be canceled on reactivation"
5533        );
5534    }
5535}