nautilus_execution/order_emulator/
emulator.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2025 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::{
17    cell::RefCell,
18    collections::{HashMap, HashSet},
19    fmt::Debug,
20    rc::Rc,
21};
22
23use nautilus_common::{
24    cache::Cache,
25    clock::Clock,
26    logging::{CMD, EVT, RECV},
27    messages::execution::{
28        CancelAllOrders, CancelOrder, ModifyOrder, SubmitOrder, SubmitOrderList, TradingCommand,
29    },
30    msgbus::{self, handler::ShareableMessageHandler},
31};
32use nautilus_core::UUID4;
33use nautilus_model::{
34    data::{OrderBookDeltas, QuoteTick, TradeTick},
35    enums::{ContingencyType, OrderSide, OrderStatus, OrderType, TriggerType},
36    events::{OrderCanceled, OrderEmulated, OrderEventAny, OrderReleased, OrderUpdated},
37    identifiers::{ClientOrderId, InstrumentId, PositionId, StrategyId},
38    instruments::Instrument,
39    orders::{LimitOrder, MarketOrder, Order, OrderAny, PassiveOrderAny},
40    types::{Price, Quantity},
41};
42
43use crate::{
44    matching_core::OrderMatchingCore, order_manager::manager::OrderManager,
45    trailing::trailing_stop_calculate,
46};
47
48pub struct OrderEmulator {
49    clock: Rc<RefCell<dyn Clock>>,
50    cache: Rc<RefCell<Cache>>,
51    manager: OrderManager,
52    matching_cores: HashMap<InstrumentId, OrderMatchingCore>,
53    subscribed_quotes: HashSet<InstrumentId>,
54    subscribed_trades: HashSet<InstrumentId>,
55    subscribed_strategies: HashSet<StrategyId>,
56    monitored_positions: HashSet<PositionId>,
57    on_event_handler: Option<ShareableMessageHandler>,
58}
59
60impl Debug for OrderEmulator {
61    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62        f.debug_struct(stringify!(OrderEmulator))
63            .field("cores", &self.matching_cores.len())
64            .field("subscribed_quotes", &self.subscribed_quotes.len())
65            .finish()
66    }
67}
68
69impl OrderEmulator {
70    pub fn new(clock: Rc<RefCell<dyn Clock>>, cache: Rc<RefCell<Cache>>) -> Self {
71        // TODO: Impl Actor Trait
72        // self.register_base(portfolio, msgbus, cache, clock);
73
74        let active_local = true;
75        let manager = OrderManager::new(clock.clone(), cache.clone(), active_local);
76
77        Self {
78            clock,
79            cache,
80            manager,
81            matching_cores: HashMap::new(),
82            subscribed_quotes: HashSet::new(),
83            subscribed_trades: HashSet::new(),
84            subscribed_strategies: HashSet::new(),
85            monitored_positions: HashSet::new(),
86            on_event_handler: None,
87        }
88    }
89
90    pub fn set_on_event_handler(&mut self, handler: ShareableMessageHandler) {
91        self.on_event_handler = Some(handler);
92    }
93
94    // TODO: WIP
95    // pub fn set_submit_order_handler(&mut self, handler: SubmitOrderHandlerAny) {
96    //     self.manager.set_submit_order_handler(handler);
97    // }
98    //
99    // pub fn set_cancel_order_handler(&mut self, handler: CancelOrderHandlerAny) {
100    //     self.manager.set_cancel_order_handler(handler);
101    // }
102    //
103    // pub fn set_modify_order_handler(&mut self, handler: ModifyOrderHandlerAny) {
104    //     self.manager.set_modify_order_handler(handler);
105    // }
106
107    #[must_use]
108    pub fn subscribed_quotes(&self) -> Vec<InstrumentId> {
109        let mut quotes: Vec<InstrumentId> = self.subscribed_quotes.iter().copied().collect();
110        quotes.sort();
111        quotes
112    }
113
114    #[must_use]
115    pub fn subscribed_trades(&self) -> Vec<InstrumentId> {
116        let mut trades: Vec<_> = self.subscribed_trades.iter().copied().collect();
117        trades.sort();
118        trades
119    }
120
121    #[must_use]
122    pub fn get_submit_order_commands(&self) -> HashMap<ClientOrderId, SubmitOrder> {
123        self.manager.get_submit_order_commands()
124    }
125
126    #[must_use]
127    pub fn get_matching_core(&self, instrument_id: &InstrumentId) -> Option<OrderMatchingCore> {
128        self.matching_cores.get(instrument_id).cloned()
129    }
130
131    /// Reactivates emulated orders from cache on start.
132    ///
133    /// # Errors
134    ///
135    /// Returns an error if no emulated orders are found or processing fails.
136    ///
137    /// # Panics
138    ///
139    /// Panics if a cached client ID cannot be unwrapped.
140    pub fn on_start(&mut self) -> anyhow::Result<()> {
141        let emulated_orders: Vec<OrderAny> = self
142            .cache
143            .borrow()
144            .orders_emulated(None, None, None, None)
145            .into_iter()
146            .cloned()
147            .collect();
148
149        if emulated_orders.is_empty() {
150            log::error!("No emulated orders to reactivate");
151            return Ok(());
152        }
153
154        for order in emulated_orders {
155            if !matches!(
156                order.status(),
157                OrderStatus::Initialized | OrderStatus::Emulated
158            ) {
159                continue; // No longer emulated
160            }
161
162            if let Some(parent_order_id) = &order.parent_order_id() {
163                let parent_order = if let Some(order) = self.cache.borrow().order(parent_order_id) {
164                    order.clone()
165                } else {
166                    log::error!("Cannot handle order: parent {parent_order_id} not found");
167                    continue;
168                };
169
170                let is_position_closed = parent_order
171                    .position_id()
172                    .is_some_and(|id| self.cache.borrow().is_position_closed(&id));
173                if parent_order.is_closed() && is_position_closed {
174                    self.manager.cancel_order(&order);
175                    continue; // Parent already closed
176                }
177
178                if parent_order.contingency_type() == Some(ContingencyType::Oto)
179                    && (parent_order.is_active_local()
180                        || parent_order.filled_qty() == Quantity::zero(0))
181                {
182                    continue; // Process contingency order later once parent triggered
183                }
184            }
185
186            let position_id = self
187                .cache
188                .borrow()
189                .position_id(&order.client_order_id())
190                .copied();
191            let client_id = self
192                .cache
193                .borrow()
194                .client_id(&order.client_order_id())
195                .copied();
196
197            let command = SubmitOrder::new(
198                order.trader_id(),
199                client_id.unwrap(),
200                order.strategy_id(),
201                order.instrument_id(),
202                order.client_order_id(),
203                order.venue_order_id().unwrap(),
204                order.clone(),
205                order.exec_algorithm_id(),
206                position_id,
207                UUID4::new(),
208                self.clock.borrow().timestamp_ns(),
209            )?;
210
211            self.handle_submit_order(command);
212        }
213
214        Ok(())
215    }
216
217    pub fn on_event(&mut self, event: OrderEventAny) {
218        log::info!("{RECV}{EVT} {event}");
219
220        self.manager.handle_event(event.clone());
221
222        if let Some(order) = self.cache.borrow().order(&event.client_order_id())
223            && order.is_closed()
224            && let Some(matching_core) = self.matching_cores.get_mut(&order.instrument_id())
225            && let Err(e) = matching_core.delete_order(&PassiveOrderAny::from(order.clone()))
226        {
227            log::error!("Error deleting order: {e}");
228        }
229        // else: Order not in cache yet
230    }
231
232    pub const fn on_stop(&self) {}
233
234    pub fn on_reset(&mut self) {
235        self.manager.reset();
236        self.matching_cores.clear();
237    }
238
239    pub const fn on_dispose(&self) {}
240
241    pub fn execute(&mut self, command: TradingCommand) {
242        log::info!("{RECV}{CMD} {command}");
243
244        match command {
245            TradingCommand::SubmitOrder(command) => self.handle_submit_order(command),
246            TradingCommand::SubmitOrderList(command) => self.handle_submit_order_list(command),
247            TradingCommand::ModifyOrder(command) => self.handle_modify_order(command),
248            TradingCommand::CancelOrder(command) => self.handle_cancel_order(command),
249            TradingCommand::CancelAllOrders(command) => self.handle_cancel_all_orders(command),
250            _ => log::error!("Cannot handle command: unrecognized {command:?}"),
251        }
252    }
253
254    fn create_matching_core(
255        &mut self,
256        instrument_id: InstrumentId,
257        price_increment: Price,
258    ) -> OrderMatchingCore {
259        let matching_core =
260            OrderMatchingCore::new(instrument_id, price_increment, None, None, None);
261        self.matching_cores
262            .insert(instrument_id, matching_core.clone());
263        log::info!("Creating matching core for {instrument_id:?}");
264        matching_core
265    }
266
267    /// # Panics
268    ///
269    /// Panics if the emulation trigger type is `NoTrigger`.
270    pub fn handle_submit_order(&mut self, command: SubmitOrder) {
271        let mut order = command.order.clone();
272        let emulation_trigger = order.emulation_trigger();
273
274        assert_ne!(
275            emulation_trigger,
276            Some(TriggerType::NoTrigger),
277            "command.order.emulation_trigger must not be TriggerType::NoTrigger"
278        );
279        assert!(
280            self.manager
281                .get_submit_order_commands()
282                .contains_key(&order.client_order_id()),
283            "command.order.client_order_id must be in submit_order_commands"
284        );
285
286        if !matches!(
287            emulation_trigger,
288            Some(TriggerType::Default | TriggerType::BidAsk | TriggerType::LastPrice)
289        ) {
290            log::error!("Cannot emulate order: `TriggerType` {emulation_trigger:?} not supported");
291            self.manager.cancel_order(&order);
292            return;
293        }
294
295        self.check_monitoring(command.strategy_id, command.position_id);
296
297        // Get or create matching core
298        let trigger_instrument_id = order
299            .trigger_instrument_id()
300            .unwrap_or_else(|| order.instrument_id());
301
302        let matching_core = self.matching_cores.get(&trigger_instrument_id).cloned();
303
304        let mut matching_core = if let Some(core) = matching_core {
305            core
306        } else {
307            // Handle synthetic instruments
308            let (instrument_id, price_increment) = if trigger_instrument_id.is_synthetic() {
309                let synthetic = self
310                    .cache
311                    .borrow()
312                    .synthetic(&trigger_instrument_id)
313                    .cloned();
314                if let Some(synthetic) = synthetic {
315                    (synthetic.id, synthetic.price_increment)
316                } else {
317                    log::error!(
318                        "Cannot emulate order: no synthetic instrument {trigger_instrument_id} for trigger"
319                    );
320                    self.manager.cancel_order(&order);
321                    return;
322                }
323            } else {
324                let instrument = self
325                    .cache
326                    .borrow()
327                    .instrument(&trigger_instrument_id)
328                    .cloned();
329                if let Some(instrument) = instrument {
330                    (instrument.id(), instrument.price_increment())
331                } else {
332                    log::error!(
333                        "Cannot emulate order: no instrument {trigger_instrument_id} for trigger"
334                    );
335                    self.manager.cancel_order(&order);
336                    return;
337                }
338            };
339
340            self.create_matching_core(instrument_id, price_increment)
341        };
342
343        // Update trailing stop
344        if matches!(
345            order.order_type(),
346            OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
347        ) {
348            self.update_trailing_stop_order(&mut order);
349            if order.trigger_price().is_none() {
350                log::error!(
351                    "Cannot handle trailing stop order with no trigger_price and no market updates"
352                );
353
354                self.manager.cancel_order(&order);
355                return;
356            }
357        }
358
359        // Cache command
360        self.manager.cache_submit_order_command(command);
361
362        // Check if immediately marketable
363        matching_core.match_order(&PassiveOrderAny::from(order.clone()), true);
364
365        // Handle data subscriptions
366        match emulation_trigger.unwrap() {
367            TriggerType::Default | TriggerType::BidAsk => {
368                if !self.subscribed_quotes.contains(&trigger_instrument_id) {
369                    if !trigger_instrument_id.is_synthetic() {
370                        // TODO: Impl Actor Trait
371                        // self.subscribe_order_book_deltas(&trigger_instrument_id);
372                    }
373                    // TODO: Impl Actor Trait
374                    // self.subscribe_quote_ticks(&trigger_instrument_id)?;
375                    self.subscribed_quotes.insert(trigger_instrument_id);
376                }
377            }
378            TriggerType::LastPrice => {
379                if !self.subscribed_trades.contains(&trigger_instrument_id) {
380                    // TODO: Impl Actor Trait
381                    // self.subscribe_trade_ticks(&trigger_instrument_id)?;
382                    self.subscribed_trades.insert(trigger_instrument_id);
383                }
384            }
385            _ => {
386                log::error!("Invalid TriggerType: {emulation_trigger:?}");
387                return;
388            }
389        }
390
391        // Check if order was already released
392        if !self
393            .manager
394            .get_submit_order_commands()
395            .contains_key(&order.client_order_id())
396        {
397            return; // Already released
398        }
399
400        // Hold in matching core
401        if let Err(e) = matching_core.add_order(PassiveOrderAny::from(order.clone())) {
402            log::error!("Cannot add order: {e:?}");
403            return;
404        }
405
406        // Generate emulated event if needed
407        if order.status() == OrderStatus::Initialized {
408            let event = OrderEmulated::new(
409                order.trader_id(),
410                order.strategy_id(),
411                order.instrument_id(),
412                order.client_order_id(),
413                UUID4::new(),
414                self.clock.borrow().timestamp_ns(),
415                self.clock.borrow().timestamp_ns(),
416            );
417
418            if let Err(e) = order.apply(OrderEventAny::Emulated(event)) {
419                log::error!("Cannot apply order event: {e:?}");
420                return;
421            }
422
423            if let Err(e) = self.cache.borrow_mut().update_order(&order) {
424                log::error!("Cannot update order: {e:?}");
425                return;
426            }
427
428            self.manager.send_risk_event(OrderEventAny::Emulated(event));
429
430            msgbus::publish(
431                format!("events.order.{}", order.strategy_id()).into(),
432                &OrderEventAny::Emulated(event),
433            );
434        }
435
436        // Since we are cloning the matching core, we need to insert it back into the original hashmap
437        self.matching_cores
438            .insert(trigger_instrument_id, matching_core);
439
440        log::info!("Emulating {order}");
441    }
442
443    fn handle_submit_order_list(&mut self, command: SubmitOrderList) {
444        self.check_monitoring(command.strategy_id, command.position_id);
445
446        for order in &command.order_list.orders {
447            if let Some(parent_order_id) = order.parent_order_id() {
448                let cache = self.cache.borrow();
449                let parent_order = if let Some(parent_order) = cache.order(&parent_order_id) {
450                    parent_order
451                } else {
452                    log::error!("Parent order for {} not found", order.client_order_id());
453                    continue;
454                };
455
456                if parent_order.contingency_type() == Some(ContingencyType::Oto) {
457                    continue; // Process contingency order later once parent triggered
458                }
459            }
460
461            if let Err(e) = self.manager.create_new_submit_order(
462                order,
463                command.position_id,
464                Some(command.client_id),
465            ) {
466                log::error!("Error creating new submit order: {e}");
467            }
468        }
469    }
470
471    fn handle_modify_order(&mut self, command: ModifyOrder) {
472        if let Some(order) = self.cache.borrow().order(&command.client_order_id) {
473            let price = match command.price {
474                Some(price) => Some(price),
475                None => order.price(),
476            };
477
478            let trigger_price = match command.trigger_price {
479                Some(trigger_price) => Some(trigger_price),
480                None => order.trigger_price(),
481            };
482
483            // Generate event
484            let ts_now = self.clock.borrow().timestamp_ns();
485            let event = OrderUpdated::new(
486                order.trader_id(),
487                order.strategy_id(),
488                order.instrument_id(),
489                order.client_order_id(),
490                command.quantity.unwrap_or(order.quantity()),
491                UUID4::new(),
492                ts_now,
493                ts_now,
494                false,
495                order.venue_order_id(),
496                order.account_id(),
497                price,
498                trigger_price,
499            );
500
501            self.manager.send_exec_event(OrderEventAny::Updated(event));
502
503            let trigger_instrument_id = order
504                .trigger_instrument_id()
505                .unwrap_or_else(|| order.instrument_id());
506
507            if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) {
508                matching_core.match_order(&PassiveOrderAny::from(order.clone()), false);
509            } else {
510                log::error!(
511                    "Cannot handle `ModifyOrder`: no matching core for trigger instrument {trigger_instrument_id}"
512                );
513            }
514        } else {
515            log::error!("Cannot modify order: {} not found", command.client_order_id);
516        }
517    }
518
519    pub fn handle_cancel_order(&mut self, command: CancelOrder) {
520        let order = if let Some(order) = self.cache.borrow().order(&command.client_order_id) {
521            order.clone()
522        } else {
523            log::error!("Cannot cancel order: {} not found", command.client_order_id);
524            return;
525        };
526
527        let trigger_instrument_id = order
528            .trigger_instrument_id()
529            .unwrap_or_else(|| order.instrument_id());
530
531        let matching_core = if let Some(core) = self.matching_cores.get(&trigger_instrument_id) {
532            core
533        } else {
534            self.manager.cancel_order(&order);
535            return;
536        };
537
538        if !matching_core.order_exists(order.client_order_id())
539            && order.is_open()
540            && !order.is_pending_cancel()
541        {
542            // Order not held in the emulator
543            self.manager
544                .send_exec_command(TradingCommand::CancelOrder(command));
545        } else {
546            self.manager.cancel_order(&order);
547        }
548    }
549
550    fn handle_cancel_all_orders(&mut self, command: CancelAllOrders) {
551        let matching_core = match self.matching_cores.get(&command.instrument_id) {
552            Some(core) => core,
553            None => return, // No orders to cancel
554        };
555
556        let orders_to_cancel = match command.order_side {
557            OrderSide::NoOrderSide => {
558                // Get both bid and ask orders
559                let mut all_orders = Vec::new();
560                all_orders.extend(matching_core.get_orders_bid().iter().cloned());
561                all_orders.extend(matching_core.get_orders_ask().iter().cloned());
562                all_orders
563            }
564            OrderSide::Buy => matching_core.get_orders_bid().to_vec(),
565            OrderSide::Sell => matching_core.get_orders_ask().to_vec(),
566        };
567
568        // Process all orders in a single iteration
569        for order in orders_to_cancel {
570            self.manager.cancel_order(&OrderAny::from(order));
571        }
572    }
573
574    // --------------------------------------------------------------------------------------------
575
576    pub fn update_order(&mut self, order: &mut OrderAny, new_quantity: Quantity) {
577        log::info!(
578            "Updating order {} quantity to {new_quantity}",
579            order.client_order_id(),
580        );
581
582        // Generate event
583        let ts_now = self.clock.borrow().timestamp_ns();
584        let event = OrderUpdated::new(
585            order.trader_id(),
586            order.strategy_id(),
587            order.instrument_id(),
588            order.client_order_id(),
589            new_quantity,
590            UUID4::new(),
591            ts_now,
592            ts_now,
593            false,
594            None,
595            order.account_id(),
596            None,
597            None,
598        );
599
600        if let Err(e) = order.apply(OrderEventAny::Updated(event)) {
601            log::error!("Cannot apply order event: {e:?}");
602            return;
603        }
604        if let Err(e) = self.cache.borrow_mut().update_order(order) {
605            log::error!("Cannot update order: {e:?}");
606            return;
607        }
608
609        self.manager.send_risk_event(OrderEventAny::Updated(event));
610    }
611
612    // -----------------------------------------------------------------------------------------------
613
614    pub fn on_order_book_deltas(&mut self, deltas: OrderBookDeltas) {
615        log::debug!("Processing {deltas:?}");
616
617        let instrument_id = &deltas.instrument_id;
618        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
619            if let Some(book) = self.cache.borrow().order_book(instrument_id) {
620                let best_bid = book.best_bid_price();
621                let best_ask = book.best_ask_price();
622
623                if let Some(best_bid) = best_bid {
624                    matching_core.set_bid_raw(best_bid);
625                }
626
627                if let Some(best_ask) = best_ask {
628                    matching_core.set_ask_raw(best_ask);
629                }
630            } else {
631                log::error!(
632                    "Cannot handle `OrderBookDeltas`: no book being maintained for {}",
633                    deltas.instrument_id
634                );
635            }
636
637            self.iterate_orders(instrument_id);
638        } else {
639            log::error!(
640                "Cannot handle `OrderBookDeltas`: no matching core for instrument {}",
641                deltas.instrument_id
642            );
643        }
644    }
645
646    pub fn on_quote_tick(&mut self, quote: QuoteTick) {
647        log::debug!("Processing {quote}:?");
648
649        let instrument_id = &quote.instrument_id;
650        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
651            matching_core.set_bid_raw(quote.bid_price);
652            matching_core.set_ask_raw(quote.ask_price);
653
654            self.iterate_orders(instrument_id);
655        } else {
656            log::error!(
657                "Cannot handle `QuoteTick`: no matching core for instrument {}",
658                quote.instrument_id
659            );
660        }
661    }
662
663    pub fn on_trade_tick(&mut self, trade: TradeTick) {
664        log::debug!("Processing {trade:?}");
665
666        let instrument_id = &trade.instrument_id;
667        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
668            matching_core.set_last_raw(trade.price);
669            if !self.subscribed_quotes.contains(instrument_id) {
670                matching_core.set_bid_raw(trade.price);
671                matching_core.set_ask_raw(trade.price);
672            }
673
674            self.iterate_orders(instrument_id);
675        } else {
676            log::error!(
677                "Cannot handle `TradeTick`: no matching core for instrument {}",
678                trade.instrument_id
679            );
680        }
681    }
682
683    fn iterate_orders(&mut self, instrument_id: &InstrumentId) {
684        let orders = if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
685            matching_core.iterate();
686
687            matching_core.get_orders()
688        } else {
689            log::error!("Cannot iterate orders: no matching core for instrument {instrument_id}");
690            return;
691        };
692
693        for order in orders {
694            if order.is_closed() {
695                continue;
696            }
697
698            let mut order: OrderAny = order.clone().into();
699            if matches!(
700                order.order_type(),
701                OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
702            ) {
703                self.update_trailing_stop_order(&mut order);
704            }
705        }
706    }
707
708    pub fn cancel_order(&mut self, order: &OrderAny) {
709        log::info!("Canceling order {}", order.client_order_id());
710
711        let mut order = order.clone();
712        order.set_emulation_trigger(Some(TriggerType::NoTrigger));
713
714        let trigger_instrument_id = order
715            .trigger_instrument_id()
716            .unwrap_or(order.instrument_id());
717
718        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id)
719            && let Err(e) = matching_core.delete_order(&PassiveOrderAny::from(order.clone()))
720        {
721            log::error!("Cannot delete order: {e:?}");
722        }
723
724        self.cache
725            .borrow_mut()
726            .update_order_pending_cancel_local(&order);
727
728        // Generate event
729        let ts_now = self.clock.borrow().timestamp_ns();
730        let event = OrderCanceled::new(
731            order.trader_id(),
732            order.strategy_id(),
733            order.instrument_id(),
734            order.client_order_id(),
735            UUID4::new(),
736            ts_now,
737            ts_now,
738            false,
739            order.venue_order_id(),
740            order.account_id(),
741        );
742
743        self.manager.send_exec_event(OrderEventAny::Canceled(event));
744    }
745
746    fn check_monitoring(&mut self, strategy_id: StrategyId, position_id: Option<PositionId>) {
747        if !self.subscribed_strategies.contains(&strategy_id) {
748            // Subscribe to all strategy events
749            if let Some(handler) = &self.on_event_handler {
750                msgbus::subscribe_str(format!("events.order.{strategy_id}"), handler.clone(), None);
751                msgbus::subscribe_str(
752                    format!("events.position.{strategy_id}"),
753                    handler.clone(),
754                    None,
755                );
756                self.subscribed_strategies.insert(strategy_id);
757                log::info!("Subscribed to strategy {strategy_id} order and position events");
758            }
759        }
760
761        if let Some(position_id) = position_id
762            && !self.monitored_positions.contains(&position_id)
763        {
764            self.monitored_positions.insert(position_id);
765        }
766    }
767
768    /// # Panics
769    ///
770    /// Panics if the order type is invalid for a stop order.
771    pub fn trigger_stop_order(&mut self, order: &mut OrderAny) {
772        match order.order_type() {
773            OrderType::StopLimit | OrderType::LimitIfTouched | OrderType::TrailingStopLimit => {
774                self.fill_limit_order(order);
775            }
776            OrderType::Market | OrderType::MarketIfTouched | OrderType::TrailingStopMarket => {
777                self.fill_market_order(order);
778            }
779            _ => panic!("invalid `OrderType`, was {}", order.order_type()),
780        }
781    }
782
783    /// # Panics
784    ///
785    /// Panics if a limit order has no price.
786    pub fn fill_limit_order(&mut self, order: &mut OrderAny) {
787        if matches!(order.order_type(), OrderType::Limit) {
788            self.fill_market_order(order);
789            return;
790        }
791
792        // Fetch command
793        let mut command = match self
794            .manager
795            .pop_submit_order_command(order.client_order_id())
796        {
797            Some(command) => command,
798            None => return, // Order already released
799        };
800
801        let trigger_instrument_id = order
802            .trigger_instrument_id()
803            .unwrap_or(order.instrument_id());
804
805        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) {
806            if let Err(e) = matching_core.delete_order(&PassiveOrderAny::from(order.clone())) {
807                log::error!("Error deleting order: {e:?}");
808            }
809
810            let emulation_trigger = TriggerType::NoTrigger;
811
812            // Transform order
813            let mut transformed = if let Ok(transformed) = LimitOrder::new_checked(
814                order.trader_id(),
815                order.strategy_id(),
816                order.instrument_id(),
817                order.client_order_id(),
818                order.order_side(),
819                order.quantity(),
820                order.price().unwrap(),
821                order.time_in_force(),
822                order.expire_time(),
823                order.is_post_only(),
824                order.is_reduce_only(),
825                order.is_quote_quantity(),
826                order.display_qty(),
827                Some(emulation_trigger),
828                Some(trigger_instrument_id),
829                order.contingency_type(),
830                order.order_list_id(),
831                order.linked_order_ids().map(Vec::from),
832                order.parent_order_id(),
833                order.exec_algorithm_id(),
834                order.exec_algorithm_params().cloned(),
835                order.exec_spawn_id(),
836                order.tags().map(Vec::from),
837                UUID4::new(),
838                self.clock.borrow().timestamp_ns(),
839            ) {
840                transformed
841            } else {
842                log::error!("Cannot create limit order");
843                return;
844            };
845            transformed.liquidity_side = order.liquidity_side();
846
847            // TODO: fix
848            // let triggered_price = order.trigger_price();
849            // if triggered_price.is_some() {
850            //     transformed.trigger_price() = (triggered_price.unwrap());
851            // }
852
853            let original_events = order.events();
854
855            for event in original_events {
856                transformed.events.insert(0, event.clone());
857            }
858
859            if let Err(e) = self.cache.borrow_mut().add_order(
860                OrderAny::Limit(transformed.clone()),
861                command.position_id,
862                Some(command.client_id),
863                true,
864            ) {
865                log::error!("Failed to add order: {e}");
866            }
867
868            // Replace commands order with transformed order
869            command.order = OrderAny::Limit(transformed.clone());
870
871            msgbus::publish(
872                format!("events.order.{}", order.strategy_id()).into(),
873                transformed.last_event(),
874            );
875
876            // Determine triggered price
877            let released_price = match order.order_side() {
878                OrderSide::Buy => matching_core.ask,
879                OrderSide::Sell => matching_core.bid,
880                _ => panic!("invalid `OrderSide`"),
881            };
882
883            // Generate event
884            let event = OrderReleased::new(
885                order.trader_id(),
886                order.strategy_id(),
887                order.instrument_id(),
888                order.client_order_id(),
889                released_price.unwrap(),
890                UUID4::new(),
891                self.clock.borrow().timestamp_ns(),
892                self.clock.borrow().timestamp_ns(),
893            );
894
895            if let Err(e) = transformed.apply(OrderEventAny::Released(event)) {
896                log::error!("Failed to apply order event: {e}");
897            }
898
899            if let Err(e) = self
900                .cache
901                .borrow_mut()
902                .update_order(&OrderAny::Limit(transformed.clone()))
903            {
904                log::error!("Failed to update order: {e}");
905            }
906
907            self.manager.send_risk_event(OrderEventAny::Released(event));
908
909            log::info!("Releasing order {}", order.client_order_id());
910
911            // Publish event
912            msgbus::publish(
913                format!("events.order.{}", transformed.strategy_id()).into(),
914                &OrderEventAny::Released(event),
915            );
916
917            if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
918                self.manager.send_algo_command(command, exec_algorithm_id);
919            } else {
920                self.manager
921                    .send_exec_command(TradingCommand::SubmitOrder(command));
922            }
923        } else {
924            log::error!(
925                "Cannot fill limit order: no matching core for instrument {trigger_instrument_id}"
926            );
927        }
928    }
929
930    /// # Panics
931    ///
932    /// Panics if a market order command is missing.
933    pub fn fill_market_order(&mut self, order: &mut OrderAny) {
934        // Fetch command
935        let mut command = match self
936            .manager
937            .pop_submit_order_command(order.client_order_id())
938        {
939            Some(command) => command,
940            None => panic!("invalid operation `_fill_market_order` with no command"),
941        };
942
943        let trigger_instrument_id = order
944            .trigger_instrument_id()
945            .unwrap_or(order.instrument_id());
946
947        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) {
948            if let Err(e) = matching_core.delete_order(&PassiveOrderAny::from(order.clone())) {
949                log::error!("Cannot delete order: {e:?}");
950            }
951
952            order.set_emulation_trigger(Some(TriggerType::NoTrigger));
953
954            // Transform order
955            let mut transformed = MarketOrder::new(
956                order.trader_id(),
957                order.strategy_id(),
958                order.instrument_id(),
959                order.client_order_id(),
960                order.order_side(),
961                order.quantity(),
962                order.time_in_force(),
963                UUID4::new(),
964                self.clock.borrow().timestamp_ns(),
965                order.is_reduce_only(),
966                order.is_quote_quantity(),
967                order.contingency_type(),
968                order.order_list_id(),
969                order.linked_order_ids().map(Vec::from),
970                order.parent_order_id(),
971                order.exec_algorithm_id(),
972                order.exec_algorithm_params().cloned(),
973                order.exec_spawn_id(),
974                order.tags().map(Vec::from),
975            );
976
977            let original_events = order.events();
978
979            for event in original_events {
980                transformed.events.insert(0, event.clone());
981            }
982
983            if let Err(e) = self.cache.borrow_mut().add_order(
984                OrderAny::Market(transformed.clone()),
985                command.position_id,
986                Some(command.client_id),
987                true,
988            ) {
989                log::error!("Failed to add order: {e}");
990            }
991
992            // Replace commands order with transformed order
993            command.order = OrderAny::Market(transformed.clone());
994
995            msgbus::publish(
996                format!("events.order.{}", order.strategy_id()).into(),
997                transformed.last_event(),
998            );
999
1000            // Determine triggered price
1001            // TODO: fix unwraps
1002            let released_price = match order.order_side() {
1003                OrderSide::Buy => matching_core.ask,
1004                OrderSide::Sell => matching_core.bid,
1005                _ => panic!("invalid `OrderSide`"),
1006            };
1007
1008            // Generate event
1009            let ts_now = self.clock.borrow().timestamp_ns();
1010            let event = OrderReleased::new(
1011                order.trader_id(),
1012                order.strategy_id(),
1013                order.instrument_id(),
1014                order.client_order_id(),
1015                released_price.unwrap(),
1016                UUID4::new(),
1017                ts_now,
1018                ts_now,
1019            );
1020
1021            if let Err(e) = transformed.apply(OrderEventAny::Released(event)) {
1022                log::error!("Failed to apply order event: {e}");
1023            }
1024
1025            if let Err(e) = self
1026                .cache
1027                .borrow_mut()
1028                .update_order(&OrderAny::Market(transformed))
1029            {
1030                log::error!("Failed to update order: {e}");
1031            }
1032            self.manager.send_risk_event(OrderEventAny::Released(event));
1033
1034            log::info!("Releasing order {}", order.client_order_id());
1035
1036            // Publish event
1037            msgbus::publish(
1038                format!("events.order.{}", order.strategy_id()).into(),
1039                &OrderEventAny::Released(event),
1040            );
1041
1042            if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
1043                self.manager.send_algo_command(command, exec_algorithm_id);
1044            } else {
1045                self.manager
1046                    .send_exec_command(TradingCommand::SubmitOrder(command));
1047            }
1048        } else {
1049            log::error!(
1050                "Cannot fill limit order: no matching core for instrument {trigger_instrument_id}"
1051            );
1052        }
1053    }
1054
1055    #[allow(clippy::too_many_lines)]
1056    fn update_trailing_stop_order(&mut self, order: &mut OrderAny) {
1057        let Some(matching_core) = self.matching_cores.get(&order.instrument_id()) else {
1058            log::error!(
1059                "Cannot update trailing-stop order: no matching core for instrument {}",
1060                order.instrument_id()
1061            );
1062            return;
1063        };
1064
1065        let mut bid = matching_core.bid;
1066        let mut ask = matching_core.ask;
1067        let mut last = matching_core.last;
1068
1069        if bid.is_none() || ask.is_none() || last.is_none() {
1070            if let Some(q) = self.cache.borrow().quote(&matching_core.instrument_id) {
1071                bid.get_or_insert(q.bid_price);
1072                ask.get_or_insert(q.ask_price);
1073            }
1074            if let Some(t) = self.cache.borrow().trade(&matching_core.instrument_id) {
1075                last.get_or_insert(t.price);
1076            }
1077        }
1078
1079        let (new_trigger_px, new_limit_px) = match trailing_stop_calculate(
1080            matching_core.price_increment,
1081            order.trigger_price(),
1082            order.activation_price(),
1083            order,
1084            bid,
1085            ask,
1086            last,
1087        ) {
1088            Ok(pair) => pair,
1089            Err(e) => {
1090                log::warn!("Cannot calculate trailing-stop update: {e}");
1091                return;
1092            }
1093        };
1094
1095        if new_trigger_px.is_none() && new_limit_px.is_none() {
1096            return;
1097        }
1098
1099        let ts_now = self.clock.borrow().timestamp_ns();
1100        let update = OrderUpdated::new(
1101            order.trader_id(),
1102            order.strategy_id(),
1103            order.instrument_id(),
1104            order.client_order_id(),
1105            order.quantity(),
1106            UUID4::new(),
1107            ts_now,
1108            ts_now,
1109            false,
1110            order.venue_order_id(),
1111            order.account_id(),
1112            new_limit_px,
1113            new_trigger_px,
1114        );
1115        let wrapped = OrderEventAny::Updated(update);
1116        if let Err(e) = order.apply(wrapped.clone()) {
1117            log::error!("Failed to apply order event: {e}");
1118            return;
1119        }
1120        if let Err(e) = self.cache.borrow_mut().update_order(order) {
1121            log::error!("Failed to update order in cache: {e}");
1122            return;
1123        }
1124        self.manager.send_risk_event(wrapped);
1125    }
1126}