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