1use 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 pub fn cache_submit_order_command(&mut self, command: SubmitOrder) {
165 self.manager.cache_submit_order_command(command);
166 }
167
168 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 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 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; }
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; }
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; }
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, UUID4::new(),
376 self.clock.borrow().timestamp_ns(),
377 None, );
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 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 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 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 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 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 }
677
678 self.check_monitoring(strategy_id, position_id);
679
680 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 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 if !self
725 .manager
726 .get_submit_order_commands()
727 .contains_key(&order.client_order_id())
728 {
729 return; }
731
732 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 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; }
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 self.apply_order_event(&event);
856
857 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 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 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 = "e.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 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 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 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 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 pub fn fill_limit_order(&mut self, client_order_id: ClientOrderId) {
1305 self.release_order(client_order_id, ReleaseType::Limit);
1306 }
1307
1308 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 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 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 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 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 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; }
1630
1631 if !was_activated {
1632 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 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 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 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, )
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()), "e);
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, "e_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("e_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()), "e);
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 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 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 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 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, "e_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, "e_order);
4414 submit_to_emulator(&cache, &emulator, &instrument, &trailing_order);
4415
4416 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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}