Skip to main content

nautilus_trading/strategy/
mod.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16pub mod api;
17#[doc(hidden)]
18pub mod binding;
19pub mod config;
20pub mod core;
21
22pub use core::{StrategyCore, StrategyNative};
23use std::panic::{AssertUnwindSafe, catch_unwind};
24
25use ahash::AHashSet;
26pub use api::{OrderApi, PortfolioApi};
27use binding::StrategyBinding;
28pub use config::{ImportableStrategyConfig, StrategyConfig};
29use nautilus_common::{
30    actor::DataActor,
31    component::Component,
32    enums::ComponentState,
33    logging::{CMD, EVT, RECV, SEND},
34    messages::execution::{
35        BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder, ModifyOrder,
36        QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList, TradingCommand,
37    },
38    msgbus::{self, MessagingSwitchboard},
39    timer::TimeEvent,
40};
41use nautilus_core::{DurationNanos, Params, UUID4};
42use nautilus_execution::order_manager::OrderManagerAction;
43use nautilus_model::{
44    enums::{OrderSide, OrderStatus, OrderType, PositionSide, TimeInForce},
45    events::{
46        OrderAccepted, OrderCancelRejected, OrderCanceled, OrderDenied, OrderEmulated,
47        OrderEventAny, OrderExpired, OrderFillVoided, OrderFilled, OrderInitialized,
48        OrderModifyRejected, OrderPendingCancel, OrderPendingUpdate, OrderRejected, OrderReleased,
49        OrderSubmitted, OrderTriggered, OrderUpdated, PositionChanged, PositionClosed,
50        PositionEvent, PositionOpened,
51    },
52    identifiers::{
53        AccountId, ClientId, ClientOrderId, ExecAlgorithmId, InstrumentId, PositionId, StrategyId,
54        TraderId,
55    },
56    orders::{
57        LIMIT_ORDER_TYPES, Order, OrderAny, OrderCore, OrderError, OrderList, STOP_ORDER_TYPES,
58    },
59    position::Position,
60    types::{Price, Quantity},
61};
62use ustr::Ustr;
63
64/// Describes one child update in a batch modify request.
65pub type BatchModifyOrder = (
66    ClientOrderId,
67    Option<Quantity>,
68    Option<Price>,
69    Option<Price>,
70);
71
72/// Core trait for implementing trading strategies in NautilusTrader.
73///
74/// Strategies are specialized [`DataActor`]s that combine data ingestion capabilities with
75/// order and position management functionality. By implementing this trait,
76/// custom strategies gain access to the full trading execution stack including order
77/// submission, modification, cancellation, and position management.
78///
79/// # Key Capabilities
80///
81/// - All [`DataActor`] capabilities (data subscriptions, event handling, timers).
82/// - Order lifecycle management (submit, modify, cancel).
83/// - Position management (open, close, monitor).
84/// - Access to the trading cache and portfolio.
85/// - Event routing for orders and emulator events.
86///
87/// # Implementation
88///
89/// Use the `nautilus_strategy!` macro to generate the native runtime wiring
90/// and `Strategy` implementations. Normal strategy logic should call facade
91/// methods such as `strategy_id()`, `clock()`, `cache()`, `order()`, and
92/// `portfolio()`. Native runtime code that needs the internal core should use
93/// [`StrategyNative`]. For strategies that override additional trait methods,
94/// pass them in a block:
95///
96/// ```ignore
97/// nautilus_strategy!(MyStrategy, {
98///     fn on_order_rejected(&mut self, event: OrderRejected) {
99///         // custom handling
100///     }
101/// });
102/// ```
103///
104/// Default methods that read or mutate native runtime state carry explicit
105/// [`StrategyNative`] and [`Component`] bounds. Implementations that only need
106/// behavioral callbacks do not own or implement native runtime state.
107pub trait Strategy: DataActor {
108    /// Returns the instrument IDs this strategy intends to claim for external order routing.
109    ///
110    /// Live strategy registration materializes this configuration intent as active cache claims.
111    fn external_order_instrument_ids(&self) -> Option<Vec<InstrumentId>> {
112        None
113    }
114
115    /// Replaces this strategy's active external order claims with `instrument_ids`.
116    ///
117    /// External orders, fills, and materialized reconciliation activity for matching instrument
118    /// IDs are assigned to the strategy. Passing an empty vector releases every claim owned by the
119    /// strategy. Existing cached orders keep their assigned strategy ID.
120    ///
121    /// # Errors
122    ///
123    /// Returns an error if the strategy is not registered, the cache is already borrowed, an
124    /// instrument is repeated, or an instrument is claimed by another strategy.
125    fn set_external_order_instrument_ids(
126        &mut self,
127        instrument_ids: Vec<InstrumentId>,
128    ) -> anyhow::Result<()>
129    where
130        Self: StrategyNative,
131    {
132        let core = StrategyNative::strategy_core_mut(self);
133        let strategy_id = registered_strategy_id(core)?;
134        if !core.actor.is_registered() {
135            anyhow::bail!("Strategy {strategy_id} is not registered with a trader");
136        }
137        let cache = core.cache_rc();
138        cache
139            .try_borrow_mut()
140            .map_err(|e| anyhow::anyhow!("Cannot set external order claims: {e}"))?
141            .set_external_order_claims(strategy_id, &instrument_ids)?;
142        core.config.external_order_instrument_ids = Some(instrument_ids);
143        Ok(())
144    }
145
146    /// Returns the runtime strategy ID, when configured or registered.
147    fn strategy_id(&self) -> Option<StrategyId>
148    where
149        Self: StrategyNative,
150    {
151        StrategyNative::strategy_core(self).strategy_id()
152    }
153
154    /// Returns the user-facing order creation API.
155    fn order(&self) -> OrderApi<'_>
156    where
157        Self: StrategyNative,
158    {
159        StrategyNative::strategy_core(self).order()
160    }
161
162    /// Returns the user-facing portfolio read API.
163    fn portfolio(&self) -> PortfolioApi<'_>
164    where
165        Self: StrategyNative,
166    {
167        StrategyNative::strategy_core(self).portfolio_api()
168    }
169
170    /// Submits an order.
171    ///
172    /// # Errors
173    ///
174    /// Returns an error if the strategy is not registered or order submission fails.
175    fn submit_order(
176        &mut self,
177        order: OrderAny,
178        position_id: Option<PositionId>,
179        client_id: Option<ClientId>,
180        params: Option<Params>,
181    ) -> anyhow::Result<()>
182    where
183        Self: StrategyBinding,
184    {
185        self.binding_submit_order(order, position_id, client_id, params)
186    }
187
188    /// Submits an order list.
189    ///
190    /// # Errors
191    ///
192    /// Returns an error if the strategy is not registered, the order list is invalid,
193    /// or order list submission fails.
194    fn submit_order_list(
195        &mut self,
196        mut orders: Vec<OrderAny>,
197        position_id: Option<PositionId>,
198        client_id: Option<ClientId>,
199        params: Option<Params>,
200    ) -> anyhow::Result<()>
201    where
202        Self: StrategyNative,
203    {
204        if orders.is_empty() {
205            anyhow::bail!("OrderList denied: no orders to submit");
206        }
207
208        for order in &orders {
209            if order.status() != OrderStatus::Initialized {
210                anyhow::bail!(
211                    "Order in list denied: invalid status for {}, expected INITIALIZED",
212                    order.client_order_id()
213                );
214            }
215        }
216
217        let first_venue = orders[0].instrument_id().venue;
218        for order in &orders {
219            if order.instrument_id().venue != first_venue {
220                anyhow::bail!(
221                    "OrderList denied: orders must share the same venue; \
222                     expected {first_venue}, found {} on {}",
223                    order.instrument_id().venue,
224                    order.client_order_id(),
225                );
226            }
227        }
228
229        let should_deny = {
230            let core = StrategyNative::strategy_core_mut(self);
231            let tag = core.market_exit_tag;
232            core.is_exiting
233                && orders.iter().any(|o| {
234                    !o.is_reduce_only() && !o.tags().is_some_and(|tags| tags.contains(&tag))
235                })
236        };
237
238        if should_deny {
239            self.deny_order_list(&orders, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
240            return Ok(());
241        }
242
243        let core = StrategyNative::strategy_core_mut(self);
244
245        let trader_id = registered_trader_id(core)?;
246        let strategy_id = registered_strategy_id(core)?;
247        let ts_init = core.clock_mut().timestamp_ns();
248
249        // TODO: Replace with fluent builder API for order list construction
250        let order_list = if orders.first().is_some_and(|o| o.order_list_id().is_some()) {
251            OrderList::from_orders(&orders, ts_init)
252        } else {
253            core.order_factory().create_list(&mut orders, ts_init)
254        };
255
256        if let Err(e) = order_list.validate() {
257            anyhow::bail!("OrderList denied: {e}");
258        }
259
260        {
261            let cache_rc = core.cache_rc();
262            let mut cache = cache_rc.try_borrow_mut().map_err(|_| {
263                anyhow::anyhow!(
264                    "Cannot submit order list {}: cache is currently borrowed",
265                    order_list.id
266                )
267            })?;
268
269            if cache.order_list_exists(&order_list.id) {
270                anyhow::bail!("OrderList denied: duplicate {}", order_list.id);
271            }
272
273            for order in &orders {
274                if cache.order_exists(&order.client_order_id()) {
275                    anyhow::bail!(
276                        "Order in list denied: duplicate {}",
277                        order.client_order_id()
278                    );
279                }
280            }
281
282            cache.add_order_list(order_list.clone())?;
283            for order in &orders {
284                cache.add_order(order.clone(), position_id, client_id, true)?;
285            }
286        }
287
288        for order in &orders {
289            publish_order_initialized(order);
290        }
291
292        let params = params.filter(|params| !params.is_empty());
293
294        let first_order = orders.first();
295        let order_inits: Vec<_> = orders.iter().map(|o| o.init_event().clone()).collect();
296        let exec_algorithm_id = first_order.and_then(Order::exec_algorithm_id);
297
298        let command = SubmitOrderList::new(
299            trader_id,
300            client_id,
301            strategy_id,
302            order_list,
303            order_inits,
304            exec_algorithm_id,
305            position_id,
306            params,
307            UUID4::new(),
308            ts_init,
309            None, // correlation_id
310        );
311
312        let has_emulated_order = orders
313            .iter()
314            .any(|o| o.emulation_trigger().is_some() || o.is_emulated());
315
316        if has_emulated_order {
317            send_emulator_command(TradingCommand::SubmitOrderList(command));
318        } else if let Some(algo_id) = exec_algorithm_id {
319            let endpoint = format!("{algo_id}.execute");
320            msgbus::send_any(endpoint.into(), &TradingCommand::SubmitOrderList(command));
321        } else {
322            send_risk_command(TradingCommand::SubmitOrderList(command));
323        }
324
325        for order in &orders {
326            self.set_gtd_expiry(order)?;
327        }
328
329        Ok(())
330    }
331
332    /// Modifies an order.
333    ///
334    /// # Errors
335    ///
336    /// Returns an error if the strategy is not registered, no supplied value changes,
337    /// or order modification fails. A supplied quantity counts as an update while
338    /// the order is pending update, even when the quantity is unchanged.
339    fn modify_order(
340        &mut self,
341        client_order_id: ClientOrderId,
342        quantity: Option<Quantity>,
343        price: Option<Price>,
344        trigger_price: Option<Price>,
345        client_id: Option<ClientId>,
346        params: Option<Params>,
347    ) -> anyhow::Result<()>
348    where
349        Self: StrategyNative,
350    {
351        let (trader_id, strategy_id) = {
352            let core = StrategyNative::strategy_core_mut(self);
353            (registered_trader_id(core)?, registered_strategy_id(core)?)
354        };
355
356        let params = params.filter(|params| !params.is_empty());
357
358        // TODO: Snapshot the order from the cache. See `cancel_order` for the rationale.
359        let order = StrategyNative::strategy_core_mut(self)
360            .cache_rc()
361            .borrow()
362            .try_order_owned(&client_order_id)
363            .map_err(|e| anyhow::anyhow!("Cannot modify order: {e}"))?;
364
365        let mut updating = false;
366
367        if quantity.is_some_and(|q| q != order.quantity() || order.is_pending_update()) {
368            updating = true;
369        }
370
371        if let Some(price) = price {
372            if !LIMIT_ORDER_TYPES.contains(&order.order_type()) {
373                anyhow::bail!("{} orders do not have a LIMIT price", order.order_type());
374            }
375
376            if Some(price) != order.price() {
377                updating = true;
378            }
379        }
380
381        if let Some(trigger_price) = trigger_price {
382            if !has_trigger_price(order.order_type()) {
383                anyhow::bail!(
384                    "{} orders do not have a STOP trigger price",
385                    order.order_type()
386                );
387            }
388
389            if Some(trigger_price) != order.trigger_price() {
390                updating = true;
391            }
392        }
393
394        if !updating {
395            anyhow::bail!(
396                "Cannot create command ModifyOrder: quantity, price, and trigger were either None \
397                or the same as existing values"
398            );
399        }
400
401        if order.is_closed() || order.is_pending_cancel() {
402            log::warn!(
403                "Cannot create command ModifyOrder: state is {:?}, {order:?}",
404                order.status()
405            );
406            return Ok(());
407        }
408
409        if !self.mark_order_pending_update(&order)? {
410            return Ok(());
411        }
412
413        let command = ModifyOrder::new(
414            trader_id,
415            client_id,
416            strategy_id,
417            order.instrument_id(),
418            order.client_order_id(),
419            order.venue_order_id(),
420            quantity,
421            price,
422            trigger_price,
423            UUID4::new(),
424            StrategyNative::strategy_core_mut(self)
425                .clock_mut()
426                .timestamp_ns(),
427            params,
428            None, // correlation_id
429        );
430
431        if order.emulation_trigger().is_some() || order.is_emulated() {
432            send_emulator_command(TradingCommand::ModifyOrder(command));
433        } else if let Some(algo_id) = order
434            .exec_algorithm_id()
435            .filter(|_| order.is_active_local())
436        {
437            let endpoint = format!("{algo_id}.execute");
438            msgbus::send_any(endpoint.into(), &TradingCommand::ModifyOrder(command));
439        } else {
440            send_risk_command(TradingCommand::ModifyOrder(command));
441        }
442        Ok(())
443    }
444
445    /// Batch modifies multiple orders for the same instrument.
446    ///
447    /// Each tuple is `(client_order_id, quantity, price, trigger_price)`.
448    ///
449    /// # Errors
450    ///
451    /// Returns an error if the strategy is not registered, the orders span multiple instruments,
452    /// contain emulated/local orders, or a child modify is invalid.
453    fn modify_orders(
454        &mut self,
455        updates: Vec<BatchModifyOrder>,
456        client_id: Option<ClientId>,
457        params: Option<Params>,
458    ) -> anyhow::Result<()>
459    where
460        Self: StrategyNative,
461    {
462        if updates.is_empty() {
463            anyhow::bail!("Cannot batch modify empty order list");
464        }
465
466        let (trader_id, strategy_id, ts_init) = {
467            let core = StrategyNative::strategy_core_mut(self);
468            (
469                registered_trader_id(core)?,
470                registered_strategy_id(core)?,
471                core.clock_mut().timestamp_ns(),
472            )
473        };
474
475        let orders: Vec<OrderAny> = {
476            let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
477            let cache = cache_rc.borrow();
478            updates
479                .iter()
480                .map(|(client_order_id, _, _, _)| {
481                    cache
482                        .try_order_owned(client_order_id)
483                        .map_err(|e| anyhow::anyhow!("Cannot modify order: {e}"))
484                })
485                .collect::<Result<_, _>>()?
486        };
487
488        let instrument_id = orders[0].instrument_id();
489
490        for (order, (_, quantity, price, trigger_price)) in orders.iter().zip(updates.iter()) {
491            if order.instrument_id() != instrument_id {
492                anyhow::bail!(
493                    "Cannot batch modify orders for different instruments: {} vs {}",
494                    instrument_id,
495                    order.instrument_id()
496                );
497            }
498
499            if order.is_emulated() || order.is_active_local() {
500                anyhow::bail!("Cannot include emulated or local orders in batch modify");
501            }
502
503            let mut updating = false;
504
505            if quantity.is_some_and(|q| q != order.quantity()) {
506                updating = true;
507            }
508
509            if let Some(price) = price {
510                if !LIMIT_ORDER_TYPES.contains(&order.order_type()) {
511                    anyhow::bail!("{} orders do not have a LIMIT price", order.order_type());
512                }
513
514                if Some(*price) != order.price() {
515                    updating = true;
516                }
517            }
518
519            if let Some(trigger_price) = trigger_price {
520                if !has_trigger_price(order.order_type()) {
521                    anyhow::bail!(
522                        "{} orders do not have a STOP trigger price",
523                        order.order_type()
524                    );
525                }
526
527                if Some(*trigger_price) != order.trigger_price() {
528                    updating = true;
529                }
530            }
531
532            if !updating {
533                anyhow::bail!(
534                    "Cannot create command BatchModifyOrders: quantity, price, and trigger were \
535                    either None or the same as existing values for {}",
536                    order.client_order_id()
537                );
538            }
539
540            if order.is_closed() || order.is_pending_cancel() {
541                anyhow::bail!(
542                    "Cannot create command BatchModifyOrders: state is {:?}, {order:?}",
543                    order.status()
544                );
545            }
546        }
547
548        let params = params.filter(|params| !params.is_empty());
549        let mut modifies = Vec::with_capacity(orders.len());
550
551        for (order, (_, quantity, price, trigger_price)) in orders.into_iter().zip(updates) {
552            if !self.mark_order_pending_update(&order)? {
553                continue;
554            }
555
556            modifies.push(ModifyOrder::new(
557                trader_id,
558                client_id,
559                strategy_id,
560                instrument_id,
561                order.client_order_id(),
562                order.venue_order_id(),
563                quantity,
564                price,
565                trigger_price,
566                UUID4::new(),
567                ts_init,
568                params.clone(),
569                None, // correlation_id
570            ));
571        }
572
573        if modifies.is_empty() {
574            log::warn!("Cannot send `BatchModifyOrders`, no valid modify commands");
575            return Ok(());
576        }
577
578        let command = BatchModifyOrders::new(
579            trader_id,
580            client_id,
581            strategy_id,
582            instrument_id,
583            modifies,
584            UUID4::new(),
585            ts_init,
586            params,
587            None, // correlation_id
588        );
589
590        send_risk_command(TradingCommand::ModifyOrders(command));
591        Ok(())
592    }
593
594    /// Cancels an order.
595    ///
596    /// # Errors
597    ///
598    /// Returns an error if the strategy is not registered or order cancellation fails.
599    fn cancel_order(
600        &mut self,
601        client_order_id: ClientOrderId,
602        client_id: Option<ClientId>,
603        params: Option<Params>,
604    ) -> anyhow::Result<()>
605    where
606        Self: StrategyNative,
607    {
608        let (trader_id, strategy_id, ts_init) = {
609            let core = StrategyNative::strategy_core_mut(self);
610            (
611                registered_trader_id(core)?,
612                registered_strategy_id(core)?,
613                core.clock_mut().timestamp_ns(),
614            )
615        };
616
617        let params = params.filter(|params| !params.is_empty());
618
619        // TODO: Snapshot the order from the cache. Callers identify it by ID; we own the
620        // snapshot so later calls (which take `&OrderAny` and may re-enter the cache)
621        // run without holding a live cache borrow.
622        let order = StrategyNative::strategy_core_mut(self)
623            .cache_rc()
624            .borrow()
625            .try_order_owned(&client_order_id)
626            .map_err(|e| anyhow::anyhow!("Cannot cancel order: {e}"))?;
627
628        if !self.mark_order_pending_cancel(&order)? {
629            return Ok(());
630        }
631
632        let command = CancelOrder::new(
633            trader_id,
634            client_id,
635            strategy_id,
636            order.instrument_id(),
637            order.client_order_id(),
638            order.venue_order_id(),
639            UUID4::new(),
640            ts_init,
641            params,
642            None, // correlation_id
643        );
644
645        if order.emulation_trigger().is_some() || order.is_emulated() {
646            send_emulator_command(TradingCommand::CancelOrder(command));
647        } else if let Some(algo_id) = order
648            .exec_algorithm_id()
649            .filter(|_| order.is_active_local())
650        {
651            let endpoint = format!("{algo_id}.execute");
652            msgbus::send_any(endpoint.into(), &TradingCommand::CancelOrder(command));
653        } else {
654            send_exec_command(TradingCommand::CancelOrder(command));
655        }
656
657        if StrategyNative::strategy_core(self).config.manage_gtd_expiry
658            && order.time_in_force() == TimeInForce::Gtd
659            && self.has_gtd_expiry_timer(&order.client_order_id())
660        {
661            self.cancel_gtd_expiry(&order.client_order_id());
662        }
663
664        Ok(())
665    }
666
667    /// Batch cancels multiple orders for the same instrument.
668    ///
669    /// # Errors
670    ///
671    /// Returns an error if the strategy is not registered, the orders span multiple instruments,
672    /// or contain emulated/local orders.
673    fn cancel_orders(
674        &mut self,
675        client_order_ids: Vec<ClientOrderId>,
676        client_id: Option<ClientId>,
677        params: Option<Params>,
678    ) -> anyhow::Result<()>
679    where
680        Self: StrategyNative,
681    {
682        if client_order_ids.is_empty() {
683            anyhow::bail!("Cannot batch cancel empty order list");
684        }
685
686        let (trader_id, strategy_id, ts_init) = {
687            let core = StrategyNative::strategy_core_mut(self);
688            (
689                registered_trader_id(core)?,
690                registered_strategy_id(core)?,
691                core.clock_mut().timestamp_ns(),
692            )
693        };
694
695        // TODO: Snapshot all orders from the cache. See `cancel_order` for the rationale.
696        let orders: Vec<OrderAny> = {
697            let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
698            let cache = cache_rc.borrow();
699            client_order_ids
700                .iter()
701                .map(|id| {
702                    cache
703                        .try_order_owned(id)
704                        .map_err(|e| anyhow::anyhow!("Cannot cancel order: {e}"))
705                })
706                .collect::<Result<_, _>>()?
707        };
708
709        let instrument_id = orders[0].instrument_id();
710
711        for order in &orders {
712            if order.instrument_id() != instrument_id {
713                anyhow::bail!(
714                    "Cannot batch cancel orders for different instruments: {} vs {}",
715                    instrument_id,
716                    order.instrument_id()
717                );
718            }
719
720            if order.is_emulated() || order.is_active_local() {
721                anyhow::bail!("Cannot include emulated or local orders in batch cancel");
722            }
723        }
724
725        let mut cancels = Vec::with_capacity(orders.len());
726
727        for order in orders {
728            if !self.mark_order_pending_cancel(&order)? {
729                continue;
730            }
731
732            cancels.push(CancelOrder::new(
733                trader_id,
734                client_id,
735                strategy_id,
736                instrument_id,
737                order.client_order_id(),
738                order.venue_order_id(),
739                UUID4::new(),
740                ts_init,
741                params.clone(),
742                None, // correlation_id
743            ));
744        }
745
746        if cancels.is_empty() {
747            log::warn!("Cannot send `BatchCancelOrders`, no valid cancel commands");
748            return Ok(());
749        }
750
751        let command = BatchCancelOrders::new(
752            trader_id,
753            client_id,
754            strategy_id,
755            instrument_id,
756            cancels,
757            UUID4::new(),
758            ts_init,
759            params,
760            None, // correlation_id
761        );
762
763        send_exec_command(TradingCommand::CancelOrders(command));
764        Ok(())
765    }
766
767    /// Marks an order as pending update locally before the modify command leaves the strategy.
768    ///
769    /// # Errors
770    ///
771    /// Returns an error if applying the pending update event to the cache fails.
772    fn mark_order_pending_update(&mut self, order: &OrderAny) -> anyhow::Result<bool>
773    where
774        Self: StrategyNative,
775    {
776        if order.is_active_local() || order.is_pending_update() {
777            return Ok(true);
778        }
779
780        let strategy_id = order.strategy_id();
781        required_account_id(order, "pending update")?;
782        let event = OrderEventAny::PendingUpdate(self.generate_order_pending_update(order));
783
784        {
785            let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
786            let mut cache = cache_rc.borrow_mut();
787            match cache.update_order(&event) {
788                Ok(_) => {}
789                Err(e)
790                    if matches!(
791                        e.downcast_ref::<OrderError>(),
792                        Some(OrderError::InvalidStateTransition)
793                    ) =>
794                {
795                    log::warn!("InvalidStateTrigger: {e}, did not apply pending update event");
796                    return Ok(false);
797                }
798                Err(e) => return Err(e),
799            }
800        }
801
802        let topic = format!("events.order.{strategy_id}");
803        msgbus::publish_order_event(topic.into(), &event);
804        msgbus::publish_order_event(
805            msgbus::switchboard::get_order_pending_update_topic(order.instrument_id()),
806            &event,
807        );
808
809        Ok(true)
810    }
811
812    /// Marks an order as pending cancel locally before the cancel command leaves the strategy.
813    ///
814    /// # Errors
815    ///
816    /// Returns an error if applying the pending cancel event to the cache fails.
817    fn mark_order_pending_cancel(&mut self, order: &OrderAny) -> anyhow::Result<bool>
818    where
819        Self: StrategyNative,
820    {
821        if order.is_closed() || order.is_pending_cancel() {
822            log::warn!(
823                "Cannot cancel order: state is {:?}, {order:?}",
824                order.status()
825            );
826            return Ok(false);
827        }
828
829        if order.is_active_local() {
830            return Ok(true);
831        }
832
833        let strategy_id = order.strategy_id();
834        required_account_id(order, "pending cancel")?;
835        let event = OrderEventAny::PendingCancel(self.generate_order_pending_cancel(order));
836
837        {
838            let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
839            let mut cache = cache_rc.borrow_mut();
840            match cache.update_order(&event) {
841                Ok(_) => {}
842                Err(e)
843                    if matches!(
844                        e.downcast_ref::<OrderError>(),
845                        Some(OrderError::InvalidStateTransition)
846                    ) =>
847                {
848                    log::warn!("InvalidStateTrigger: {e}, did not apply pending cancel event");
849                    return Ok(false);
850                }
851                Err(e) => return Err(e),
852            }
853            cache.update_order_pending_cancel_local(order);
854        }
855
856        let topic = format!("events.order.{strategy_id}");
857        msgbus::publish_order_event(topic.into(), &event);
858        msgbus::publish_order_event(
859            msgbus::switchboard::get_order_pending_cancel_topic(order.instrument_id()),
860            &event,
861        );
862
863        Ok(true)
864    }
865
866    /// Generates an `OrderPendingUpdate` event for an order.
867    fn generate_order_pending_update(&mut self, order: &OrderAny) -> OrderPendingUpdate
868    where
869        Self: StrategyNative,
870    {
871        let ts_now = StrategyNative::strategy_core_mut(self)
872            .clock_mut()
873            .timestamp_ns();
874
875        OrderPendingUpdate::new(
876            order.trader_id(),
877            order.strategy_id(),
878            order.instrument_id(),
879            order.client_order_id(),
880            order.account_id(),
881            UUID4::new(),
882            ts_now,
883            ts_now,
884            false,
885            order.venue_order_id(),
886        )
887    }
888
889    /// Generates an `OrderPendingCancel` event for an order.
890    fn generate_order_pending_cancel(&mut self, order: &OrderAny) -> OrderPendingCancel
891    where
892        Self: StrategyNative,
893    {
894        let ts_now = StrategyNative::strategy_core_mut(self)
895            .clock_mut()
896            .timestamp_ns();
897
898        OrderPendingCancel::new(
899            order.trader_id(),
900            order.strategy_id(),
901            order.instrument_id(),
902            order.client_order_id(),
903            order.account_id(),
904            UUID4::new(),
905            ts_now,
906            ts_now,
907            false,
908            order.venue_order_id(),
909        )
910    }
911
912    /// Cancels all open orders for the given instrument.
913    ///
914    /// When `strategy_only` is `true`, only orders associated with this strategy are canceled. When
915    /// `false`, one [`CancelAllOrders`] command is sent even when the cache has no matching order.
916    /// The execution engine selects the explicit, venue-routed, or default client and may cancel
917    /// orders associated with other strategies for the same instrument, client, and account. It
918    /// does not broadcast across execution clients.
919    ///
920    /// # Errors
921    ///
922    /// Returns an error if the strategy is not registered or order cancellation fails.
923    fn cancel_all_orders(
924        &mut self,
925        instrument_id: InstrumentId,
926        order_side: Option<OrderSide>,
927        client_id: Option<ClientId>,
928        strategy_only: bool,
929        params: Option<Params>,
930    ) -> anyhow::Result<()>
931    where
932        Self: StrategyNative,
933    {
934        let params = params.filter(|params| !params.is_empty());
935        let core = StrategyNative::strategy_core_mut(self);
936
937        let trader_id = registered_trader_id(core)?;
938        let strategy_id = registered_strategy_id(core)?;
939        let ts_init = core.clock_mut().timestamp_ns();
940
941        if !strategy_only {
942            let command_id = UUID4::new();
943            let command = CancelAllOrders::new(
944                trader_id,
945                client_id,
946                strategy_id,
947                instrument_id,
948                order_side,
949                command_id,
950                ts_init,
951                params,
952                Some(command_id),
953            );
954
955            send_exec_command(TradingCommand::CancelAllOrders(command));
956            return Ok(());
957        }
958
959        let cache = core.cache_ref();
960
961        let mut open_order_ids: Vec<ClientOrderId> = cache
962            .orders_open(
963                None,
964                Some(&instrument_id),
965                Some(&strategy_id),
966                None,
967                order_side,
968            )
969            .into_iter()
970            .map(|order| order.client_order_id())
971            .collect();
972
973        let mut emulated_order_ids: Vec<ClientOrderId> = cache
974            .orders_emulated(
975                None,
976                Some(&instrument_id),
977                Some(&strategy_id),
978                None,
979                order_side,
980            )
981            .into_iter()
982            .map(|order| order.client_order_id())
983            .collect();
984
985        let mut inflight_order_ids: Vec<ClientOrderId> = cache
986            .orders_inflight(
987                None,
988                Some(&instrument_id),
989                Some(&strategy_id),
990                None,
991                order_side,
992            )
993            .into_iter()
994            .map(|order| order.client_order_id())
995            .collect();
996
997        // An order submitted earlier in the same handler stays local until its queued
998        // command is processed, so it is not yet open or in flight
999        let mut local_order_ids: Vec<ClientOrderId> = cache
1000            .orders_active_local(
1001                None,
1002                Some(&instrument_id),
1003                Some(&strategy_id),
1004                None,
1005                order_side,
1006            )
1007            .into_iter()
1008            .map(|order| order.client_order_id())
1009            .collect();
1010
1011        // Sort the algorithm IDs so the per-algo cancel cascade fires msgbus
1012        // events in a deterministic order across runs; the cache returns an
1013        // unordered AHashSet.
1014        let mut exec_algorithm_ids: Vec<_> = cache.exec_algorithm_ids().into_iter().collect();
1015        exec_algorithm_ids.sort();
1016        let mut algo_order_ids: Vec<ClientOrderId> = Vec::new();
1017
1018        for algo_id in &exec_algorithm_ids {
1019            algo_order_ids.extend(
1020                cache
1021                    .orders_for_exec_algorithm(
1022                        algo_id,
1023                        None,
1024                        Some(&instrument_id),
1025                        Some(&strategy_id),
1026                        None,
1027                        order_side,
1028                    )
1029                    .into_iter()
1030                    .map(|order| order.client_order_id()),
1031            );
1032        }
1033
1034        let matches_client = |client_order_id: &ClientOrderId| {
1035            client_id.is_none_or(|client_id| {
1036                cache
1037                    .client_id(client_order_id)
1038                    .is_none_or(|order_client_id| *order_client_id == client_id)
1039            })
1040        };
1041
1042        open_order_ids.retain(&matches_client);
1043        emulated_order_ids.retain(&matches_client);
1044        inflight_order_ids.retain(&matches_client);
1045        local_order_ids.retain(&matches_client);
1046        algo_order_ids.retain(&matches_client);
1047
1048        let open_count = open_order_ids.len();
1049        let emulated_count = emulated_order_ids.len();
1050        let inflight_count = inflight_order_ids.len();
1051        let local_count = local_order_ids.len();
1052        let algo_count = algo_order_ids.len();
1053
1054        let mut cancel_routes: Vec<_> = open_order_ids
1055            .iter()
1056            .chain(&emulated_order_ids)
1057            .chain(&inflight_order_ids)
1058            .chain(&local_order_ids)
1059            .chain(&algo_order_ids)
1060            .map(|client_order_id| {
1061                (
1062                    *client_order_id,
1063                    client_id.or_else(|| cache.client_id(client_order_id).copied()),
1064                )
1065            })
1066            .collect();
1067        cancel_routes.sort_by_key(|(client_order_id, _)| *client_order_id);
1068        cancel_routes.dedup_by_key(|(client_order_id, _)| *client_order_id);
1069
1070        drop(cache);
1071
1072        if open_count == 0
1073            && emulated_count == 0
1074            && inflight_count == 0
1075            && local_count == 0
1076            && algo_count == 0
1077        {
1078            let side_str = order_side.map(|s| format!(" {s}")).unwrap_or_default();
1079            log::info!("No {instrument_id} open, emulated, or inflight{side_str} orders to cancel");
1080            return Ok(());
1081        }
1082
1083        let side_str = order_side.map(|s| format!(" {s}")).unwrap_or_default();
1084
1085        if open_count > 0 {
1086            log::info!(
1087                "Canceling {open_count} open{side_str} {instrument_id} order{}",
1088                if open_count == 1 { "" } else { "s" }
1089            );
1090        }
1091
1092        if emulated_count > 0 {
1093            log::info!(
1094                "Canceling {emulated_count} emulated{side_str} {instrument_id} order{}",
1095                if emulated_count == 1 { "" } else { "s" }
1096            );
1097        }
1098
1099        if inflight_count > 0 {
1100            log::info!(
1101                "Canceling {inflight_count} inflight{side_str} {instrument_id} order{}",
1102                if inflight_count == 1 { "" } else { "s" }
1103            );
1104        }
1105
1106        let mut error = None;
1107
1108        for (client_order_id, client_id) in cancel_routes {
1109            if let Err(e) = self.cancel_order(client_order_id, client_id, params.clone()) {
1110                error = Some(match error {
1111                    Some(previous) => anyhow::anyhow!("{previous}; {e}"),
1112                    None => e,
1113                });
1114            }
1115        }
1116
1117        error.map_or(Ok(()), Err)
1118    }
1119
1120    /// Closes a position by submitting a market order for the opposite side.
1121    ///
1122    /// # Errors
1123    ///
1124    /// Returns an error if the strategy is not registered or position closing fails.
1125    #[expect(clippy::too_many_arguments)]
1126    fn close_position(
1127        &mut self,
1128        position: &Position,
1129        client_id: Option<ClientId>,
1130        tags: Option<Vec<Ustr>>,
1131        time_in_force: Option<TimeInForce>,
1132        reduce_only: Option<bool>,
1133        quote_quantity: Option<bool>,
1134        params: Option<Params>,
1135    ) -> anyhow::Result<()>
1136    where
1137        Self: StrategyNative,
1138    {
1139        let core = StrategyNative::strategy_core_mut(self);
1140
1141        if position.is_closed() {
1142            log::warn!("Cannot close position (already closed): {}", position.id);
1143            return Ok(());
1144        }
1145
1146        let Some(closing_side) = OrderCore::closing_side(position.side) else {
1147            log::warn!("Cannot close flat position: {}", position.id);
1148            return Ok(());
1149        };
1150
1151        let order = core.order_factory().market(
1152            position.instrument_id,
1153            closing_side,
1154            position.quantity,
1155            time_in_force,
1156            reduce_only.or(Some(true)),
1157            quote_quantity,
1158            None,
1159            None,
1160            tags,
1161            None,
1162        );
1163
1164        self.submit_order(order, Some(position.id), client_id, params)
1165    }
1166
1167    /// Closes all open positions for the given instrument.
1168    ///
1169    /// # Errors
1170    ///
1171    /// Returns an error if the strategy is not registered or position closing fails.
1172    #[expect(clippy::too_many_arguments)]
1173    fn close_all_positions(
1174        &mut self,
1175        instrument_id: InstrumentId,
1176        position_side: Option<PositionSide>,
1177        client_id: Option<ClientId>,
1178        tags: Option<Vec<Ustr>>,
1179        time_in_force: Option<TimeInForce>,
1180        reduce_only: Option<bool>,
1181        quote_quantity: Option<bool>,
1182        params: Option<Params>,
1183    ) -> anyhow::Result<()>
1184    where
1185        Self: StrategyNative,
1186    {
1187        let core = StrategyNative::strategy_core_mut(self);
1188        let strategy_id = registered_strategy_id(core)?;
1189        let cache = core.cache_ref();
1190
1191        let positions_open = cache.positions_open(
1192            None,
1193            Some(&instrument_id),
1194            Some(&strategy_id),
1195            None,
1196            position_side,
1197        );
1198
1199        let side_str = position_side.map(|s| format!(" {s}")).unwrap_or_default();
1200
1201        if positions_open.is_empty() {
1202            log::info!("No {instrument_id} open{side_str} positions to close");
1203            return Ok(());
1204        }
1205
1206        let count = positions_open.len();
1207        log::info!(
1208            "Closing {count} open{side_str} position{}",
1209            if count == 1 { "" } else { "s" }
1210        );
1211
1212        let positions_data: Vec<_> = positions_open
1213            .iter()
1214            .map(|p| (p.id, p.instrument_id, p.side, p.quantity, p.is_closed()))
1215            .collect();
1216        drop(positions_open);
1217
1218        drop(cache);
1219
1220        for (pos_id, pos_instrument_id, pos_side, pos_quantity, is_closed) in positions_data {
1221            if is_closed {
1222                continue;
1223            }
1224
1225            let core = StrategyNative::strategy_core_mut(self);
1226            let Some(closing_side) = OrderCore::closing_side(pos_side) else {
1227                continue;
1228            };
1229            let order = core.order_factory().market(
1230                pos_instrument_id,
1231                closing_side,
1232                pos_quantity,
1233                time_in_force,
1234                reduce_only.or(Some(true)),
1235                quote_quantity,
1236                None,
1237                None,
1238                tags.clone(),
1239                None,
1240            );
1241
1242            self.submit_order(order, Some(pos_id), client_id, params.clone())?;
1243        }
1244
1245        Ok(())
1246    }
1247
1248    /// Queries account state from the execution client.
1249    ///
1250    /// Creates a [`QueryAccount`] command and sends it to the execution engine,
1251    /// which will request the current account state from the execution client.
1252    ///
1253    /// # Errors
1254    ///
1255    /// Returns an error if the strategy is not registered.
1256    fn query_account(
1257        &mut self,
1258        account_id: AccountId,
1259        client_id: Option<ClientId>,
1260        params: Option<Params>,
1261    ) -> anyhow::Result<()>
1262    where
1263        Self: StrategyNative,
1264    {
1265        let core = StrategyNative::strategy_core_mut(self);
1266
1267        let trader_id = registered_trader_id(core)?;
1268        let ts_init = core.clock_mut().timestamp_ns();
1269
1270        let command = QueryAccount::new(
1271            trader_id,
1272            client_id,
1273            account_id,
1274            UUID4::new(),
1275            ts_init,
1276            params,
1277            None, // correlation_id
1278        );
1279
1280        send_exec_command(TradingCommand::QueryAccount(command));
1281        Ok(())
1282    }
1283
1284    /// Queries order state from the execution client.
1285    ///
1286    /// Creates a [`QueryOrder`] command and sends it to the execution engine,
1287    /// which will request the current order state from the execution client.
1288    ///
1289    /// # Errors
1290    ///
1291    /// Returns an error if the strategy is not registered.
1292    fn query_order(
1293        &mut self,
1294        order: &OrderAny,
1295        client_id: Option<ClientId>,
1296        params: Option<Params>,
1297    ) -> anyhow::Result<()>
1298    where
1299        Self: StrategyNative,
1300    {
1301        let core = StrategyNative::strategy_core_mut(self);
1302
1303        let trader_id = registered_trader_id(core)?;
1304        let strategy_id = registered_strategy_id(core)?;
1305        let ts_init = core.clock_mut().timestamp_ns();
1306
1307        let command = QueryOrder::new(
1308            trader_id,
1309            client_id,
1310            strategy_id,
1311            order.instrument_id(),
1312            order.client_order_id(),
1313            order.venue_order_id(),
1314            UUID4::new(),
1315            ts_init,
1316            params,
1317            None, // correlation_id
1318        );
1319
1320        send_exec_command(TradingCommand::QueryOrder(command));
1321        Ok(())
1322    }
1323
1324    /// Handles an order event, dispatching to the appropriate handler.
1325    fn handle_order_event(&mut self, event: OrderEventAny)
1326    where
1327        Self: StrategyNative,
1328    {
1329        let state = {
1330            let core = StrategyNative::strategy_core_mut(self);
1331            let id = &core.actor.actor_id;
1332
1333            let is_warning = match &event {
1334                OrderEventAny::Rejected(event) => {
1335                    !event.due_post_only || core.config.log_rejected_due_post_only_as_warning
1336                }
1337                OrderEventAny::Denied(_)
1338                | OrderEventAny::CancelRejected(_)
1339                | OrderEventAny::ModifyRejected(_) => true,
1340                _ => false,
1341            };
1342
1343            if is_warning {
1344                log::warn!("{id} {RECV}{EVT} {event}");
1345            } else if core.actor.config.log_events {
1346                log::info!("{id} {RECV}{EVT} {event}");
1347            }
1348
1349            core.actor.state()
1350        };
1351
1352        let client_order_id = event.client_order_id();
1353        let cached_order_is_closed = {
1354            let core = StrategyNative::strategy_core_mut(self);
1355            core.cache_ref()
1356                .order(&client_order_id)
1357                .map(|order| order.is_closed())
1358        };
1359        let is_terminal = match &event {
1360            OrderEventAny::FillVoided(event) => {
1361                cached_order_is_closed.unwrap_or(!event.is_reopened)
1362            }
1363            OrderEventAny::Filled(_) => cached_order_is_closed.unwrap_or(true),
1364            OrderEventAny::Canceled(_)
1365            | OrderEventAny::Rejected(_)
1366            | OrderEventAny::Expired(_)
1367            | OrderEventAny::Denied(_) => true,
1368            _ => false,
1369        };
1370
1371        // GTD timer cleanup runs regardless of state so timers do not leak when
1372        // terminal events arrive during the post-stop delay.
1373        if is_terminal {
1374            self.cancel_gtd_expiry(&client_order_id);
1375        }
1376
1377        // Events are logged unconditionally so residual events received after stop
1378        // remain observable, but dispatch is gated on the running state.
1379        if state != ComponentState::Running {
1380            return;
1381        }
1382
1383        if matches!(&event, OrderEventAny::CancelRejected(_)) {
1384            let order = StrategyNative::strategy_core_mut(self)
1385                .cache_ref()
1386                .order(&client_order_id)
1387                .map(|order| order.clone());
1388            if let Some(order) = order
1389                && (order.is_open() || order.is_inflight())
1390                && !self.has_gtd_expiry_timer(&client_order_id)
1391                && let Err(e) = self.set_gtd_expiry(&order)
1392            {
1393                log::error!(
1394                    "Failed to restore GTD expiry for cancel-rejected order {client_order_id}: {e}"
1395                );
1396            }
1397        }
1398
1399        if matches!(&event, OrderEventAny::FillVoided(event) if event.is_reopened) {
1400            let order = StrategyNative::strategy_core_mut(self)
1401                .cache_ref()
1402                .order(&client_order_id)
1403                .map(|order| order.clone());
1404            if let Some(order) = order
1405                && order.is_open()
1406                && !self.has_gtd_expiry_timer(&client_order_id)
1407                && let Err(e) = self.set_gtd_expiry(&order)
1408            {
1409                log::error!(
1410                    "Failed to restore GTD expiry for reopened order {client_order_id}: {e}"
1411                );
1412            }
1413        }
1414
1415        let manager_actions = {
1416            let core = StrategyNative::strategy_core_mut(self);
1417            if core.config.manage_contingent_orders {
1418                core.order_manager
1419                    .as_mut()
1420                    .map_or_else(Vec::new, |manager| manager.handle_event(&event))
1421            } else {
1422                Vec::new()
1423            }
1424        };
1425        self.dispatch_manager_actions(manager_actions);
1426
1427        match &event {
1428            OrderEventAny::Initialized(e) => self.on_order_initialized(e.clone()),
1429            OrderEventAny::Denied(e) => self.on_order_denied(*e),
1430            OrderEventAny::Emulated(e) => self.on_order_emulated(*e),
1431            OrderEventAny::Released(e) => self.on_order_released(*e),
1432            OrderEventAny::Submitted(e) => self.on_order_submitted(*e),
1433            OrderEventAny::Rejected(e) => self.on_order_rejected(*e),
1434            OrderEventAny::Accepted(e) => self.on_order_accepted(*e),
1435            OrderEventAny::Canceled(e) => self.on_order_canceled(e),
1436            OrderEventAny::Expired(e) => self.on_order_expired(*e),
1437            OrderEventAny::Triggered(e) => self.on_order_triggered(*e),
1438            OrderEventAny::PendingUpdate(e) => self.on_order_pending_update(*e),
1439            OrderEventAny::PendingCancel(e) => self.on_order_pending_cancel(*e),
1440            OrderEventAny::ModifyRejected(e) => self.on_order_modify_rejected(*e),
1441            OrderEventAny::CancelRejected(e) => self.on_order_cancel_rejected(*e),
1442            OrderEventAny::Updated(e) => self.on_order_updated(*e),
1443            OrderEventAny::Filled(e) => self.on_order_filled(e),
1444            OrderEventAny::FillVoided(e) => self.on_order_fill_voided(e),
1445        }
1446        self.on_order_event(event);
1447    }
1448
1449    fn dispatch_manager_actions(&mut self, actions: Vec<OrderManagerAction>)
1450    where
1451        Self: StrategyNative,
1452    {
1453        for action in actions {
1454            match action {
1455                OrderManagerAction::PublishInitialized(event) => {
1456                    let topic = msgbus::switchboard::get_event_order_topic(event.strategy_id());
1457                    msgbus::publish_order_event(topic, &event);
1458                }
1459                OrderManagerAction::SubmitToEmulator(command) => {
1460                    send_emulator_command(TradingCommand::SubmitOrder(command));
1461                }
1462                OrderManagerAction::SubmitToRisk(command) => {
1463                    send_risk_command(TradingCommand::SubmitOrder(command));
1464                }
1465                OrderManagerAction::SubmitToAlgorithm {
1466                    command,
1467                    exec_algorithm_id,
1468                } => send_algo_command(command, exec_algorithm_id),
1469                OrderManagerAction::CancelLocal(order) => {
1470                    let client_order_id = order.client_order_id();
1471                    if let Err(e) = self.cancel_order(client_order_id, None, None) {
1472                        log::error!(
1473                            "Failed to dispatch contingent cancel for {client_order_id}: {e}"
1474                        );
1475                    }
1476                }
1477                OrderManagerAction::ModifyLocalQuantity { order, quantity } => {
1478                    let client_order_id = order.client_order_id();
1479                    if let Err(e) =
1480                        self.modify_order(client_order_id, Some(quantity), None, None, None, None)
1481                    {
1482                        log::error!(
1483                            "Failed to dispatch contingent modify for {client_order_id}: {e}"
1484                        );
1485                    }
1486                }
1487            }
1488        }
1489    }
1490
1491    /// Handles a position event, dispatching to the appropriate handler.
1492    fn handle_position_event(&mut self, event: PositionEvent)
1493    where
1494        Self: StrategyNative,
1495    {
1496        let state = {
1497            let core = StrategyNative::strategy_core_mut(self);
1498
1499            if core.actor.config.log_events {
1500                let id = &core.actor.actor_id;
1501                log::info!("{id} {RECV}{EVT} {event:?}");
1502            }
1503
1504            core.actor.state()
1505        };
1506
1507        if state != ComponentState::Running {
1508            return;
1509        }
1510
1511        match &event {
1512            PositionEvent::PositionOpened(e) => self.on_position_opened(e.clone()),
1513            PositionEvent::PositionChanged(e) => self.on_position_changed(e.clone()),
1514            PositionEvent::PositionClosed(e) => self.on_position_closed(e.clone()),
1515            PositionEvent::PositionAdjusted(_) => {
1516                return;
1517            }
1518        }
1519        self.on_position_event(event);
1520    }
1521
1522    // -- LIFECYCLE METHODS -----------------------------------------------------------------------
1523
1524    /// Called when the strategy is started.
1525    ///
1526    /// Override this method to implement custom initialization logic.
1527    /// The default implementation reactivates GTD timers if `manage_gtd_expiry` is enabled.
1528    ///
1529    /// # Errors
1530    ///
1531    /// Returns an error if strategy initialization fails.
1532    fn on_start(&mut self) -> anyhow::Result<()>
1533    where
1534        Self: StrategyNative,
1535    {
1536        let core = StrategyNative::strategy_core_mut(self);
1537        let strategy_id = registered_strategy_id(core)?;
1538        log::info!("Starting {strategy_id}");
1539
1540        if core.config.manage_gtd_expiry {
1541            self.reactivate_gtd_timers();
1542        }
1543
1544        Ok(())
1545    }
1546
1547    /// Routes a time event through the framework-managed strategy handlers when called directly.
1548    ///
1549    /// Framework timer dispatch never calls this method. Implement [`DataActor::on_time_event`] for
1550    /// user timer callbacks. Runtime hosts call [`route_time_event`] directly so an override cannot
1551    /// replace framework-managed GTD expiry or market exit routing.
1552    ///
1553    /// # Errors
1554    ///
1555    /// Returns an error if time event handling fails.
1556    fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()>
1557    where
1558        Self: StrategyNative + Component,
1559    {
1560        route_time_event(self, event);
1561        Ok(())
1562    }
1563
1564    // -- EVENT HANDLERS --------------------------------------------------------------------------
1565
1566    /// Called when an order is initialized.
1567    ///
1568    /// Override this method to implement custom logic when an order is first created.
1569    #[allow(unused_variables)]
1570    fn on_order_initialized(&mut self, event: OrderInitialized) {}
1571
1572    /// Called when any order event is received after the specific order handler runs.
1573    ///
1574    /// Override this method to implement custom logic for all order events.
1575    #[allow(unused_variables)]
1576    fn on_order_event(&mut self, event: OrderEventAny) {}
1577
1578    /// Called when an order is denied by the system.
1579    ///
1580    /// Override this method to implement custom logic when an order is denied before submission.
1581    #[allow(unused_variables)]
1582    fn on_order_denied(&mut self, event: OrderDenied) {}
1583
1584    /// Called when an order is emulated.
1585    ///
1586    /// Override this method to implement custom logic when an order is taken over by the emulator.
1587    #[allow(unused_variables)]
1588    fn on_order_emulated(&mut self, event: OrderEmulated) {}
1589
1590    /// Called when an order is released from emulation.
1591    ///
1592    /// Override this method to implement custom logic when an emulated order is released.
1593    #[allow(unused_variables)]
1594    fn on_order_released(&mut self, event: OrderReleased) {}
1595
1596    /// Called when an order is submitted to the venue.
1597    ///
1598    /// Override this method to implement custom logic when an order is submitted.
1599    #[allow(unused_variables)]
1600    fn on_order_submitted(&mut self, event: OrderSubmitted) {}
1601
1602    /// Called when an order is rejected by the venue.
1603    ///
1604    /// Override this method to implement custom logic when an order is rejected.
1605    #[allow(unused_variables)]
1606    fn on_order_rejected(&mut self, event: OrderRejected) {}
1607
1608    /// Called when an order is accepted by the venue.
1609    ///
1610    /// Override this method to implement custom logic when an order is accepted.
1611    #[allow(unused_variables)]
1612    fn on_order_accepted(&mut self, event: OrderAccepted) {}
1613
1614    /// Called when an order expires.
1615    ///
1616    /// Override this method to implement custom logic when an order expires.
1617    #[allow(unused_variables)]
1618    fn on_order_expired(&mut self, event: OrderExpired) {}
1619
1620    /// Called when an order is triggered.
1621    ///
1622    /// Override this method to implement custom logic when a stop or conditional order is triggered.
1623    #[allow(unused_variables)]
1624    fn on_order_triggered(&mut self, event: OrderTriggered) {}
1625
1626    /// Called when an order modification is pending.
1627    ///
1628    /// Override this method to implement custom logic when an order is pending modification.
1629    #[allow(unused_variables)]
1630    fn on_order_pending_update(&mut self, event: OrderPendingUpdate) {}
1631
1632    /// Called when an order cancellation is pending.
1633    ///
1634    /// Override this method to implement custom logic when an order is pending cancellation.
1635    #[allow(unused_variables)]
1636    fn on_order_pending_cancel(&mut self, event: OrderPendingCancel) {}
1637
1638    /// Called when an order modification is rejected.
1639    ///
1640    /// Override this method to implement custom logic when an order modification is rejected.
1641    #[allow(unused_variables)]
1642    fn on_order_modify_rejected(&mut self, event: OrderModifyRejected) {}
1643
1644    /// Called when an order cancellation is rejected.
1645    ///
1646    /// Override this method to implement custom logic when an order cancellation is rejected.
1647    #[allow(unused_variables)]
1648    fn on_order_cancel_rejected(&mut self, event: OrderCancelRejected) {}
1649
1650    /// Called when an order is updated.
1651    ///
1652    /// Override this method to implement custom logic when an order is modified.
1653    #[allow(unused_variables)]
1654    fn on_order_updated(&mut self, event: OrderUpdated) {}
1655
1656    /// Called when an order is canceled.
1657    ///
1658    /// Override this method to implement custom logic when an order is canceled.
1659    #[allow(unused_variables)]
1660    fn on_order_canceled(&mut self, event: &OrderCanceled) {}
1661
1662    /// Called when an order is filled.
1663    ///
1664    /// Override this method to implement custom logic when an order is filled.
1665    #[allow(unused_variables)]
1666    fn on_order_filled(&mut self, event: &OrderFilled) {}
1667
1668    /// Called when an applied order fill is partly or fully voided.
1669    #[allow(unused_variables)]
1670    fn on_order_fill_voided(&mut self, event: &OrderFillVoided) {}
1671
1672    /// Called when a position is opened.
1673    ///
1674    /// Override this method to implement custom logic when a position is opened.
1675    #[allow(unused_variables)]
1676    fn on_position_opened(&mut self, event: PositionOpened) {}
1677
1678    /// Called after a position opened, changed, or closed handler runs.
1679    ///
1680    /// Override this method to implement custom logic for all position events.
1681    #[allow(unused_variables)]
1682    fn on_position_event(&mut self, event: PositionEvent) {}
1683
1684    /// Called when a position is changed (quantity or price updated).
1685    ///
1686    /// Override this method to implement custom logic when a position changes.
1687    #[allow(unused_variables)]
1688    fn on_position_changed(&mut self, event: PositionChanged) {}
1689
1690    /// Called when a position is closed.
1691    ///
1692    /// Override this method to implement custom logic when a position is closed.
1693    #[allow(unused_variables)]
1694    fn on_position_closed(&mut self, event: PositionClosed) {}
1695
1696    /// Called when a market exit has been initiated.
1697    ///
1698    /// Override this method to implement custom logic when a market exit begins.
1699    fn on_market_exit(&mut self) {}
1700
1701    /// Called after a market exit has completed.
1702    ///
1703    /// Override this method to implement custom logic after a market exit completes.
1704    fn post_market_exit(&mut self) {}
1705
1706    /// Returns whether the strategy is currently executing a market exit.
1707    ///
1708    /// Strategies can check this to avoid submitting new orders during exit.
1709    fn is_exiting(&self) -> bool
1710    where
1711        Self: StrategyNative,
1712    {
1713        StrategyNative::strategy_core(self).is_exiting
1714    }
1715
1716    /// Initiates an iterative market exit for the strategy.
1717    ///
1718    /// Will cancel all open orders and close all open positions, and wait for
1719    /// all in-flight orders to resolve and positions to close. The strategy
1720    /// remains running after the exit completes.
1721    ///
1722    /// The `on_market_exit` hook is called when the exit process begins.
1723    /// The `post_market_exit` hook is called when the exit process completes.
1724    ///
1725    /// Uses `market_exit_time_in_force` and `market_exit_reduce_only` from
1726    /// the strategy config for closing market orders.
1727    ///
1728    /// # Errors
1729    ///
1730    /// Returns an error if the market exit cannot be initiated.
1731    fn market_exit(&mut self) -> anyhow::Result<()>
1732    where
1733        Self: StrategyNative,
1734    {
1735        let core = StrategyNative::strategy_core_mut(self);
1736        let strategy_id = registered_strategy_id(core)?;
1737
1738        if core.actor.state() != ComponentState::Running {
1739            log::warn!("{strategy_id} Cannot market exit: strategy is not running");
1740            return Ok(());
1741        }
1742
1743        if core.is_exiting {
1744            log::warn!("{strategy_id} Market exit called when already in progress");
1745            return Ok(());
1746        }
1747
1748        core.is_exiting = true;
1749        core.market_exit_attempts = 0;
1750        let time_in_force = core.config.market_exit_time_in_force;
1751        let reduce_only = core.config.market_exit_reduce_only;
1752
1753        log::info!("{strategy_id} Initiating market exit...");
1754
1755        self.on_market_exit();
1756
1757        let core = StrategyNative::strategy_core_mut(self);
1758        let cache = core.cache_ref();
1759
1760        let mut instruments: AHashSet<InstrumentId> = AHashSet::new();
1761
1762        for client_order_id in
1763            cache.iter_client_order_ids_open(None, None, Some(&strategy_id), None)
1764        {
1765            if let Some(order) = cache.order(&client_order_id) {
1766                instruments.insert(order.instrument_id());
1767            }
1768        }
1769
1770        for client_order_id in
1771            cache.iter_client_order_ids_inflight(None, None, Some(&strategy_id), None)
1772        {
1773            if let Some(order) = cache.order(&client_order_id) {
1774                instruments.insert(order.instrument_id());
1775            }
1776        }
1777
1778        for position_id in cache.iter_position_open_ids(None, None, Some(&strategy_id), None) {
1779            if let Some(position) = cache.position(&position_id) {
1780                instruments.insert(position.instrument_id);
1781            }
1782        }
1783
1784        let market_exit_tag = core.market_exit_tag;
1785        // Sort so the per-instrument cancel_all_orders/close_all_positions
1786        // cascade fires msgbus commands in a deterministic sequence; the
1787        // upstream dedup is AHash-backed.
1788        let mut instruments: Vec<_> = instruments.into_iter().collect();
1789        instruments.sort();
1790        drop(cache);
1791
1792        for instrument_id in instruments {
1793            if let Err(e) = self.cancel_all_orders(instrument_id, None, None, true, None) {
1794                log::error!("Error canceling orders for {instrument_id}: {e}");
1795            }
1796
1797            if let Err(e) = self.close_all_positions(
1798                instrument_id,
1799                None,
1800                None,
1801                Some(vec![market_exit_tag]),
1802                Some(time_in_force),
1803                Some(reduce_only),
1804                None,
1805                None,
1806            ) {
1807                log::error!("Error closing positions for {instrument_id}: {e}");
1808            }
1809        }
1810
1811        let core = StrategyNative::strategy_core_mut(self);
1812        let interval_ms = core.config.market_exit_interval_ms;
1813        let timer_name = core.market_exit_timer_name;
1814
1815        log::info!("{strategy_id} Setting market exit timer at {interval_ms}ms intervals");
1816
1817        let Ok(interval_ns) = DurationNanos::try_from_millis(interval_ms) else {
1818            core.is_exiting = false;
1819            core.market_exit_attempts = 0;
1820            anyhow::bail!("Market exit timer interval exceeds the nanosecond range");
1821        };
1822        let result = core.clock_mut().set_timer_ns(
1823            timer_name.as_str(),
1824            interval_ns,
1825            None,
1826            None,
1827            None,
1828            None,
1829            None,
1830        );
1831
1832        if let Err(e) = result {
1833            // Reset exit state on timer failure (caller handles pending_stop)
1834            core.is_exiting = false;
1835            core.market_exit_attempts = 0;
1836            return Err(e);
1837        }
1838
1839        Ok(())
1840    }
1841
1842    /// Checks if the market exit is complete and finalizes if so.
1843    ///
1844    /// This method is called by the market exit timer.
1845    fn check_market_exit(&mut self, _event: TimeEvent)
1846    where
1847        Self: StrategyNative + Component,
1848    {
1849        // Guard against stale timer events after cancel_market_exit
1850        if !self.is_exiting() {
1851            return;
1852        }
1853
1854        let core = StrategyNative::strategy_core_mut(self);
1855        let Some(strategy_id) = core.strategy_id() else {
1856            log::error!("Cannot check market exit: strategy_id is not set");
1857            return;
1858        };
1859
1860        core.market_exit_attempts += 1;
1861        let attempts = core.market_exit_attempts;
1862        let max_attempts = core.config.market_exit_max_attempts;
1863
1864        log::debug!(
1865            "{strategy_id} Market exit check triggered (attempt {attempts}/{max_attempts})"
1866        );
1867
1868        if attempts >= max_attempts {
1869            let cache = core.cache_ref();
1870            let open_orders_count =
1871                cache.orders_open_count(None, None, Some(&strategy_id), None, None);
1872            let inflight_orders_count =
1873                cache.orders_inflight_count(None, None, Some(&strategy_id), None, None);
1874            let open_positions_count =
1875                cache.positions_open_count(None, None, Some(&strategy_id), None, None);
1876
1877            drop(cache);
1878
1879            log::warn!(
1880                "{strategy_id} Market exit max attempts ({max_attempts}) reached, \
1881                completing with open orders: {open_orders_count}, \
1882                inflight orders: {inflight_orders_count}, \
1883                open positions: {open_positions_count}"
1884            );
1885
1886            self.finalize_market_exit();
1887            return;
1888        }
1889
1890        let cache = core.cache_ref();
1891        let has_open_orders = !cache
1892            .orders_open(None, None, Some(&strategy_id), None, None)
1893            .is_empty();
1894        let has_inflight_orders = !cache
1895            .orders_inflight(None, None, Some(&strategy_id), None, None)
1896            .is_empty();
1897
1898        if has_open_orders || has_inflight_orders {
1899            return;
1900        }
1901
1902        let positions_data: Vec<_> = cache
1903            .positions_open(None, None, Some(&strategy_id), None, None)
1904            .iter()
1905            .map(|p| (p.id, p.instrument_id, p.side, p.quantity, p.is_closed()))
1906            .collect();
1907
1908        if !positions_data.is_empty() {
1909            // If there are open positions but no orders, re-send close orders
1910            drop(cache);
1911
1912            for (pos_id, instrument_id, side, quantity, is_closed) in positions_data {
1913                if is_closed {
1914                    continue;
1915                }
1916
1917                let core = StrategyNative::strategy_core_mut(self);
1918                let time_in_force = core.config.market_exit_time_in_force;
1919                let reduce_only = core.config.market_exit_reduce_only;
1920                let market_exit_tag = core.market_exit_tag;
1921                let Some(closing_side) = OrderCore::closing_side(side) else {
1922                    continue;
1923                };
1924                let order = core.order_factory().market(
1925                    instrument_id,
1926                    closing_side,
1927                    quantity,
1928                    Some(time_in_force),
1929                    Some(reduce_only),
1930                    None,
1931                    None,
1932                    None,
1933                    Some(vec![market_exit_tag]),
1934                    None,
1935                );
1936
1937                if let Err(e) = self.submit_order(order, Some(pos_id), None, None) {
1938                    log::error!("Error re-submitting close order for position {pos_id}: {e}");
1939                }
1940            }
1941            return;
1942        }
1943
1944        drop(cache);
1945        self.finalize_market_exit();
1946    }
1947
1948    /// Finalizes the market exit process.
1949    ///
1950    /// Cancels the market exit timer, resets state, calls the `post_market_exit` hook,
1951    /// and stops the strategy if a stop was pending.
1952    fn finalize_market_exit(&mut self)
1953    where
1954        Self: StrategyNative + Component,
1955    {
1956        let (actor_id, should_stop) = {
1957            let core = StrategyNative::strategy_core_mut(self);
1958            let actor_id = core.actor_id();
1959            let should_stop = core.pending_stop;
1960            (actor_id, should_stop)
1961        };
1962
1963        self.cancel_market_exit();
1964
1965        let hook_result = catch_unwind(AssertUnwindSafe(|| {
1966            self.post_market_exit();
1967        }));
1968
1969        if let Err(e) = hook_result {
1970            log::error!("{actor_id} Error in post_market_exit: {e:?}");
1971        }
1972
1973        if should_stop {
1974            log::info!("{actor_id} Market exit complete, stopping strategy");
1975
1976            if let Err(e) = Component::stop(self) {
1977                log::error!("{actor_id} Failed to stop: {e}");
1978            }
1979        }
1980
1981        let core = StrategyNative::strategy_core_mut(self);
1982        debug_assert!(
1983            !(core.pending_stop
1984                && !core.is_exiting
1985                && core.actor.state() == ComponentState::Running),
1986            "INVARIANT: stuck state after finalize_market_exit"
1987        );
1988    }
1989
1990    /// Cancels an active market exit without calling hooks.
1991    ///
1992    /// Used when `stop()` is called during an active market exit to avoid state leaks.
1993    fn cancel_market_exit(&mut self)
1994    where
1995        Self: StrategyNative,
1996    {
1997        let core = StrategyNative::strategy_core_mut(self);
1998        let timer_name = core.market_exit_timer_name;
1999
2000        if core
2001            .clock_mut()
2002            .timer_names()
2003            .contains(&timer_name.as_str())
2004        {
2005            core.clock_mut().cancel_timer(timer_name.as_str());
2006        }
2007
2008        core.is_exiting = false;
2009        core.pending_stop = false;
2010        core.market_exit_attempts = 0;
2011    }
2012
2013    /// Stops the strategy with optional managed stop behavior.
2014    ///
2015    /// If `manage_stop` is enabled in the config, the strategy will first complete
2016    /// any active market exit (or initiate one) before stopping. If `manage_stop`
2017    /// is disabled, the strategy stops immediately, cleaning up any active market
2018    /// exit state.
2019    ///
2020    /// # Returns
2021    ///
2022    /// Returns `true` if the strategy should proceed with stopping, `false` if
2023    /// the stop is being deferred until market exit completes.
2024    fn stop(&mut self) -> bool
2025    where
2026        Self: StrategyNative,
2027    {
2028        let (manage_stop, is_exiting, should_initiate_exit) = {
2029            let core = StrategyNative::strategy_core_mut(self);
2030            let actor_id = core.actor_id();
2031            let manage_stop = core.config.manage_stop;
2032            let state = core.actor.state();
2033            let pending_stop = core.pending_stop;
2034            let is_exiting = core.is_exiting;
2035
2036            if manage_stop {
2037                if state != ComponentState::Running {
2038                    return true; // Proceed with stop
2039                }
2040
2041                if pending_stop {
2042                    return false; // Already waiting for market exit
2043                }
2044
2045                core.pending_stop = true;
2046                let should_initiate_exit = !is_exiting;
2047
2048                if should_initiate_exit {
2049                    log::info!("{actor_id} Initiating market exit before stop");
2050                }
2051
2052                (manage_stop, is_exiting, should_initiate_exit)
2053            } else {
2054                (manage_stop, is_exiting, false)
2055            }
2056        };
2057
2058        if manage_stop {
2059            if should_initiate_exit && let Err(e) = self.market_exit() {
2060                log::warn!("Market exit failed during stop: {e}, proceeding with stop");
2061                StrategyNative::strategy_core_mut(self).pending_stop = false;
2062                return true;
2063            }
2064            debug_assert!(
2065                self.is_exiting(),
2066                "INVARIANT: deferring stop but not exiting"
2067            );
2068            return false; // Defer stop until market exit completes
2069        }
2070
2071        // manage_stop is false - clean up any active market exit
2072        if is_exiting {
2073            self.cancel_market_exit();
2074        }
2075
2076        true // Proceed with stop
2077    }
2078
2079    /// Denies an order by generating an `OrderDenied` event.
2080    ///
2081    /// This method creates an `OrderDenied` event, applies it to the order,
2082    /// and updates the cache.
2083    fn deny_order(&mut self, order: &OrderAny, reason: Ustr)
2084    where
2085        Self: StrategyNative,
2086    {
2087        let core = StrategyNative::strategy_core_mut(self);
2088        let Some(trader_id) = core.trader_id() else {
2089            log::error!(
2090                "Cannot deny order {}: trader_id is not set",
2091                order.client_order_id()
2092            );
2093            return;
2094        };
2095        let Some(strategy_id) = core.strategy_id() else {
2096            log::error!(
2097                "Cannot deny order {}: strategy_id is not set",
2098                order.client_order_id()
2099            );
2100            return;
2101        };
2102        let ts_now = core.clock_mut().timestamp_ns();
2103
2104        let event = OrderDenied::new(
2105            trader_id,
2106            strategy_id,
2107            order.instrument_id(),
2108            order.client_order_id(),
2109            reason,
2110            UUID4::new(),
2111            ts_now,
2112            ts_now,
2113        );
2114
2115        log::warn!(
2116            "{strategy_id} Order {} denied: {reason}",
2117            order.client_order_id()
2118        );
2119
2120        let publish_initialized = {
2121            let cache_rc = core.cache_rc();
2122            let mut cache = cache_rc.borrow_mut();
2123            if cache.order_exists(&order.client_order_id()) {
2124                false
2125            } else {
2126                match cache.add_order(order.clone(), None, None, true) {
2127                    Ok(()) => true,
2128                    Err(e) => {
2129                        log::warn!("Failed to add denied order to cache: {e}");
2130                        false
2131                    }
2132                }
2133            }
2134        };
2135
2136        if publish_initialized {
2137            publish_order_initialized(order);
2138        }
2139
2140        let event = OrderEventAny::Denied(event);
2141        let applied = {
2142            let cache_rc = core.cache_rc();
2143            let mut cache = cache_rc.borrow_mut();
2144            if let Err(e) = cache.update_order(&event) {
2145                log::warn!("Failed to apply OrderDenied event: {e}");
2146                false
2147            } else {
2148                true
2149            }
2150        };
2151
2152        if applied {
2153            let topic = format!("events.order.{strategy_id}");
2154            msgbus::publish_order_event(topic.into(), &event);
2155        }
2156    }
2157
2158    /// Denies all orders in an order list.
2159    ///
2160    /// This method denies each non-closed order in the list.
2161    fn deny_order_list(&mut self, orders: &[OrderAny], reason: Ustr)
2162    where
2163        Self: StrategyNative,
2164    {
2165        for order in orders {
2166            if !order.is_closed() {
2167                self.deny_order(order, reason);
2168            }
2169        }
2170    }
2171
2172    // -- GTD EXPIRY MANAGEMENT -------------------------------------------------------------------
2173
2174    /// Sets a GTD expiry timer for an order.
2175    ///
2176    /// Creates a timer that will automatically cancel the order when it expires.
2177    ///
2178    /// # Errors
2179    ///
2180    /// Returns an error if timer creation fails.
2181    fn set_gtd_expiry(&mut self, order: &OrderAny) -> anyhow::Result<()>
2182    where
2183        Self: StrategyNative,
2184    {
2185        let core = StrategyNative::strategy_core_mut(self);
2186
2187        if !core.config.manage_gtd_expiry || order.time_in_force() != TimeInForce::Gtd {
2188            return Ok(());
2189        }
2190
2191        let Some(expire_time) = order.expire_time() else {
2192            return Ok(());
2193        };
2194
2195        let client_order_id = order.client_order_id();
2196        let timer_name = format!("GTD-EXPIRY:{client_order_id}");
2197
2198        let current_time_ns = {
2199            let clock = core.clock_mut();
2200            clock.timestamp_ns()
2201        };
2202
2203        if current_time_ns >= expire_time.as_u64() {
2204            log::info!("GTD order {client_order_id} already expired, canceling immediately");
2205            return self.cancel_order(order.client_order_id(), None, None);
2206        }
2207
2208        {
2209            let mut clock = core.clock_mut();
2210            clock.set_time_alert_ns(&timer_name, expire_time, None, None)?;
2211        }
2212
2213        core.gtd_timers
2214            .insert(client_order_id, Ustr::from(&timer_name));
2215
2216        log::debug!("Set GTD expiry timer for {client_order_id} at {expire_time}");
2217        Ok(())
2218    }
2219
2220    /// Cancels a GTD expiry timer for an order.
2221    fn cancel_gtd_expiry(&mut self, client_order_id: &ClientOrderId)
2222    where
2223        Self: StrategyNative,
2224    {
2225        let core = StrategyNative::strategy_core_mut(self);
2226
2227        if let Some(timer_name) = core.gtd_timers.remove(client_order_id) {
2228            core.clock_mut().cancel_timer(timer_name.as_str());
2229            log::debug!("Canceled GTD expiry timer for {client_order_id}");
2230        }
2231    }
2232
2233    /// Checks if a GTD expiry timer exists for an order.
2234    fn has_gtd_expiry_timer(&mut self, client_order_id: &ClientOrderId) -> bool
2235    where
2236        Self: StrategyNative,
2237    {
2238        let core = StrategyNative::strategy_core_mut(self);
2239        core.gtd_timers.contains_key(client_order_id)
2240    }
2241
2242    /// Handles GTD order expiry by canceling the order.
2243    ///
2244    /// This method is called when a GTD expiry timer fires.
2245    fn expire_gtd_order(&mut self, event: TimeEvent)
2246    where
2247        Self: StrategyNative,
2248    {
2249        let timer_name = event.name;
2250        let Some(client_order_id) = timer_name
2251            .strip_prefix("GTD-EXPIRY:")
2252            .and_then(|value| ClientOrderId::new_checked(value).ok())
2253        else {
2254            log::error!("Invalid GTD timer name format: {timer_name}");
2255            return;
2256        };
2257
2258        let core = StrategyNative::strategy_core_mut(self);
2259        if core.gtd_timers.get(&client_order_id) != Some(&timer_name) {
2260            return;
2261        }
2262        core.gtd_timers.remove(&client_order_id);
2263
2264        let order = core.cache_ref().order(&client_order_id).map(|o| o.clone());
2265        let Some(order) = order else {
2266            log::warn!("GTD order {client_order_id} not found in cache");
2267            return;
2268        };
2269
2270        log::info!("GTD order {client_order_id} expired");
2271
2272        if let Err(e) = self.cancel_order(order.client_order_id(), None, None) {
2273            log::error!("Failed to cancel expired GTD order {client_order_id}: {e}");
2274        }
2275    }
2276
2277    /// Reactivates GTD timers for open orders on strategy start.
2278    ///
2279    /// Queries the cache for all open GTD orders and creates timers for those
2280    /// that haven't expired yet. Orders that have already expired are canceled immediately.
2281    fn reactivate_gtd_timers(&mut self)
2282    where
2283        Self: StrategyNative,
2284    {
2285        let core = StrategyNative::strategy_core_mut(self);
2286        let Some(strategy_id) = core.strategy_id() else {
2287            log::error!("Cannot reactivate GTD timers: strategy_id is not set");
2288            return;
2289        };
2290        let current_time_ns = core.clock_mut().timestamp_ns();
2291
2292        let gtd_orders: Vec<OrderAny> = core
2293            .cache_ref()
2294            .orders_open(None, None, Some(&strategy_id), None, None)
2295            .into_iter()
2296            .filter(|o| o.time_in_force() == TimeInForce::Gtd)
2297            .map(|o| o.clone())
2298            .collect();
2299
2300        for order in gtd_orders {
2301            let Some(expire_time) = order.expire_time() else {
2302                continue;
2303            };
2304
2305            let expire_time_ns = expire_time.as_u64();
2306            let client_order_id = order.client_order_id();
2307
2308            if current_time_ns >= expire_time_ns {
2309                log::info!("GTD order {client_order_id} already expired, canceling immediately");
2310                if let Err(e) = self.cancel_order(order.client_order_id(), None, None) {
2311                    log::error!("Failed to cancel expired GTD order {client_order_id}: {e}");
2312                }
2313            } else if let Err(e) = self.set_gtd_expiry(&order) {
2314                log::error!("Failed to set GTD expiry timer for {client_order_id}: {e}");
2315            }
2316        }
2317    }
2318}
2319
2320/// Routes a time event through framework-managed strategy handlers.
2321///
2322/// Runtime hosts call this before the user-facing [`DataActor::on_time_event`] callback so custom
2323/// [`Strategy::on_time_event`] implementations cannot replace managed GTD expiry or market exit
2324/// routing.
2325pub fn route_time_event<T>(strategy: &mut T, event: &TimeEvent)
2326where
2327    T: Strategy + StrategyNative + Component + ?Sized,
2328{
2329    let (gtd_order_id, is_market_exit) = {
2330        let core = StrategyNative::strategy_core(strategy);
2331        let gtd_order_id = event
2332            .name
2333            .strip_prefix("GTD-EXPIRY:")
2334            .and_then(|value| ClientOrderId::new_checked(value).ok())
2335            .filter(|client_order_id| core.gtd_timers.get(client_order_id) == Some(&event.name));
2336        let is_market_exit = event.name == core.market_exit_timer_name;
2337        (gtd_order_id, is_market_exit)
2338    };
2339
2340    if gtd_order_id.is_none() && !is_market_exit {
2341        return;
2342    }
2343
2344    let core = StrategyNative::strategy_core_mut(strategy);
2345    if core.managed_time_event_last_id == Some(event.event_id) {
2346        return;
2347    }
2348    core.managed_time_event_last_id = Some(event.event_id);
2349
2350    if gtd_order_id.is_some() {
2351        strategy.expire_gtd_order(event.clone());
2352    } else {
2353        strategy.check_market_exit(event.clone());
2354    }
2355}
2356
2357pub(super) fn submit_order_native<T>(
2358    strategy: &mut T,
2359    order: &OrderAny,
2360    position_id: Option<PositionId>,
2361    client_id: Option<ClientId>,
2362    params: Option<Params>,
2363) -> anyhow::Result<()>
2364where
2365    T: Strategy + StrategyNative + ?Sized,
2366{
2367    let core = StrategyNative::strategy_core_mut(strategy);
2368
2369    let trader_id = registered_trader_id(core)?;
2370    let strategy_id = registered_strategy_id(core)?;
2371    let ts_init = core.clock_mut().timestamp_ns();
2372
2373    if order.status() != OrderStatus::Initialized {
2374        anyhow::bail!(
2375            "Order denied: invalid status for {}, expected INITIALIZED",
2376            order.client_order_id()
2377        );
2378    }
2379
2380    let market_exit_tag = core.market_exit_tag;
2381    let is_market_exit_order = order
2382        .tags()
2383        .is_some_and(|tags| tags.contains(&market_exit_tag));
2384    let should_deny_for_market_exit =
2385        core.is_exiting && !order.is_reduce_only() && !is_market_exit_order;
2386
2387    if should_deny_for_market_exit {
2388        strategy.deny_order(order, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
2389        return Ok(());
2390    }
2391
2392    let core = StrategyNative::strategy_core_mut(strategy);
2393    let params = params.filter(|params| !params.is_empty());
2394
2395    {
2396        let cache_rc = core.cache_rc();
2397        let mut cache = cache_rc.try_borrow_mut().map_err(|_| {
2398            anyhow::anyhow!(
2399                "Cannot submit order {}: cache is currently borrowed",
2400                order.client_order_id()
2401            )
2402        })?;
2403        cache.add_order(order.clone(), position_id, client_id, true)?;
2404    }
2405
2406    publish_order_initialized(order);
2407
2408    let command = SubmitOrder::new(
2409        trader_id,
2410        client_id,
2411        strategy_id,
2412        order.instrument_id(),
2413        order.client_order_id(),
2414        order.init_event().clone(),
2415        order.exec_algorithm_id(),
2416        position_id,
2417        params,
2418        UUID4::new(),
2419        ts_init,
2420        None, // correlation_id
2421    );
2422
2423    if order.emulation_trigger().is_some() {
2424        send_emulator_command(TradingCommand::SubmitOrder(command));
2425    } else if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
2426        send_algo_command(command, exec_algorithm_id);
2427    } else {
2428        send_risk_command(TradingCommand::SubmitOrder(command));
2429    }
2430
2431    strategy.set_gtd_expiry(order)?;
2432    Ok(())
2433}
2434
2435fn publish_order_initialized(order: &OrderAny) {
2436    let topic = msgbus::switchboard::get_event_order_topic(order.strategy_id());
2437    let event = OrderEventAny::Initialized(order.init_event().clone());
2438    msgbus::publish_order_event(topic, &event);
2439}
2440
2441fn send_emulator_command(command: TradingCommand) {
2442    log_cmd_send(&command);
2443    let endpoint = MessagingSwitchboard::order_emulator_execute();
2444    msgbus::send_trading_command(endpoint, command);
2445}
2446
2447fn send_algo_command(command: SubmitOrder, exec_algorithm_id: ExecAlgorithmId) {
2448    let id = command.strategy_id;
2449    log::info!("{id} {CMD}{SEND} {command}");
2450
2451    let endpoint = format!("{exec_algorithm_id}.execute");
2452    msgbus::send_any(endpoint.into(), &TradingCommand::SubmitOrder(command));
2453}
2454
2455fn send_risk_command(command: TradingCommand) {
2456    log_cmd_send(&command);
2457    let endpoint = MessagingSwitchboard::risk_engine_queue_execute();
2458    msgbus::send_trading_command(endpoint, command);
2459}
2460
2461fn send_exec_command(command: TradingCommand) {
2462    log_cmd_send(&command);
2463    let endpoint = MessagingSwitchboard::exec_engine_queue_execute();
2464    msgbus::send_trading_command(endpoint, command);
2465}
2466
2467fn log_cmd_send(command: &TradingCommand) {
2468    if let Some(id) = command.strategy_id() {
2469        log::info!("{id} {CMD}{SEND} {command}");
2470    } else {
2471        log::info!("{CMD}{SEND} {command}");
2472    }
2473}
2474
2475fn registered_trader_id(core: &StrategyCore) -> anyhow::Result<TraderId> {
2476    core.trader_id()
2477        .ok_or_else(|| anyhow::anyhow!("Strategy not registered: trader_id is not set"))
2478}
2479
2480fn registered_strategy_id(core: &StrategyCore) -> anyhow::Result<StrategyId> {
2481    core.strategy_id()
2482        .ok_or_else(|| anyhow::anyhow!("Strategy not registered: strategy_id is not set"))
2483}
2484
2485// Trailing stops carry a venue-maintained trigger that a modify can reset, matching the risk
2486// engine's modify projection.
2487fn has_trigger_price(order_type: OrderType) -> bool {
2488    STOP_ORDER_TYPES.contains(&order_type)
2489        || matches!(
2490            order_type,
2491            OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
2492        )
2493}
2494
2495fn required_account_id(order: &OrderAny, operation: &str) -> anyhow::Result<AccountId> {
2496    order.account_id().ok_or_else(|| {
2497        anyhow::anyhow!(
2498            "Cannot generate {operation} event for {}: account_id is not set",
2499            order.client_order_id()
2500        )
2501    })
2502}
2503
2504#[cfg(test)]
2505mod tests {
2506    use std::{cell::RefCell, rc::Rc, sync::Mutex};
2507
2508    use nautilus_common::{
2509        actor::{
2510            DataActor,
2511            registry::{deregister_actor, try_get_actor_unchecked},
2512        },
2513        cache::{Cache, ORDER_NOT_FOUND},
2514        clock::{Clock, VirtualClock},
2515        component::{Component, deregister_component, register_component_actor},
2516        enums::ComponentState,
2517        logging::{
2518            arm_shutdown_on_error, disarm_shutdown_on_error, init_logging,
2519            take_shutdown_on_error_trigger,
2520        },
2521        msgbus::{
2522            self, MessagingSwitchboard, TypedHandler, TypedIntoHandler,
2523            stubs::{
2524                TypedIntoMessageSavingHandler, TypedMessageSavingHandler, get_any_saving_handler,
2525                get_typed_into_message_saving_handler, get_typed_message_saving_handler,
2526            },
2527        },
2528        timer::{TimeEvent, TimeEventCallback},
2529    };
2530    use nautilus_core::{DurationNanos, UnixNanos};
2531    use nautilus_execution::engine::ExecutionEngine;
2532    use nautilus_model::{
2533        enums::{
2534            ContingencyType, LiquiditySide, OrderSide, OrderStatus, OrderType,
2535            PositionAdjustmentType, PositionSide, TrailingOffsetType, TriggerType,
2536        },
2537        events::{
2538            OrderAccepted, OrderCanceled, OrderFilled, OrderRejected, PositionAdjusted,
2539            order::spec::{
2540                OrderAcceptedSpec, OrderCancelRejectedSpec, OrderCanceledSpec, OrderEmulatedSpec,
2541                OrderExpiredSpec, OrderFillVoidedSpec, OrderFilledSpec, OrderModifyRejectedSpec,
2542                OrderRejectedSpec, OrderUpdatedSpec,
2543            },
2544        },
2545        identifiers::{
2546            AccountId, ActorId, ClientOrderId, InstrumentId, OrderListId, PositionId, StrategyId,
2547            TradeId, TraderId, VenueOrderId,
2548        },
2549        orderbook::own::OwnOrderBook,
2550        orders::{LimitOrder, MarketOrder, OrderTestBuilder, stubs::TestOrderEventStubs},
2551        stubs::TestDefault,
2552        types::{Currency, Money, Price},
2553    };
2554    use nautilus_portfolio::portfolio::Portfolio;
2555    use rstest::rstest;
2556    use rust_decimal::Decimal;
2557    use serde_json::Value;
2558
2559    use super::*;
2560    use crate::nautilus_strategy;
2561
2562    #[derive(Debug)]
2563    struct TestStrategy {
2564        core: StrategyCore,
2565        on_order_rejected_called: bool,
2566        on_order_event_called: bool,
2567        on_order_accepted_called: bool,
2568        on_order_canceled_called: bool,
2569        on_order_filled_called: bool,
2570        on_order_fill_voided_called: bool,
2571        on_order_expired_called: bool,
2572        order_event_timeline: Rc<RefCell<Vec<&'static str>>>,
2573        on_position_event_called: bool,
2574        on_position_opened_called: bool,
2575        on_position_changed_called: bool,
2576        on_position_closed_called: bool,
2577    }
2578
2579    #[derive(Debug)]
2580    struct ModifyRejectionStrategy {
2581        core: StrategyCore,
2582        rejections: Vec<(OrderModifyRejected, OrderAny)>,
2583    }
2584
2585    #[derive(Debug)]
2586    struct CoreFreeStrategy {
2587        started: bool,
2588    }
2589
2590    #[derive(Debug)]
2591    struct InitializedModifyStrategy {
2592        core: StrategyCore,
2593        modified_quantity: Quantity,
2594    }
2595
2596    #[derive(Debug)]
2597    struct TimerOverrideStrategy {
2598        core: StrategyCore,
2599        gtd_expiries: usize,
2600        market_exit_checks: usize,
2601    }
2602
2603    impl DataActor for ModifyRejectionStrategy {}
2604
2605    nautilus_strategy!(ModifyRejectionStrategy, {
2606        fn on_order_modify_rejected(&mut self, event: OrderModifyRejected) {
2607            let order = self.cache().order(&event.client_order_id).unwrap();
2608            self.rejections.push((event, order));
2609        }
2610    });
2611
2612    impl DataActor for CoreFreeStrategy {
2613        fn on_start(&mut self) -> anyhow::Result<()> {
2614            self.started = true;
2615            Ok(())
2616        }
2617    }
2618
2619    impl Strategy for CoreFreeStrategy {}
2620
2621    impl DataActor for InitializedModifyStrategy {}
2622
2623    nautilus_strategy!(InitializedModifyStrategy, {
2624        fn on_order_initialized(&mut self, event: OrderInitialized) {
2625            self.modify_order(
2626                event.client_order_id,
2627                Some(self.modified_quantity),
2628                None,
2629                None,
2630                None,
2631                None,
2632            )
2633            .unwrap();
2634        }
2635    });
2636
2637    impl DataActor for TimerOverrideStrategy {
2638        fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
2639            Strategy::on_time_event(self, event)
2640        }
2641    }
2642
2643    nautilus_strategy!(TimerOverrideStrategy, {
2644        fn check_market_exit(&mut self, _event: TimeEvent) {
2645            self.market_exit_checks += 1;
2646        }
2647
2648        fn expire_gtd_order(&mut self, _event: TimeEvent) {
2649            self.gtd_expiries += 1;
2650        }
2651    });
2652
2653    impl TestStrategy {
2654        fn new(config: StrategyConfig) -> Self {
2655            Self {
2656                core: StrategyCore::new(config),
2657                on_order_rejected_called: false,
2658                on_order_event_called: false,
2659                on_order_accepted_called: false,
2660                on_order_canceled_called: false,
2661                on_order_filled_called: false,
2662                on_order_fill_voided_called: false,
2663                on_order_expired_called: false,
2664                order_event_timeline: Rc::new(RefCell::new(Vec::new())),
2665                on_position_event_called: false,
2666                on_position_opened_called: false,
2667                on_position_changed_called: false,
2668                on_position_closed_called: false,
2669            }
2670        }
2671    }
2672
2673    impl DataActor for TestStrategy {}
2674
2675    nautilus_strategy!(TestStrategy, {
2676        fn on_order_canceled(&mut self, _event: &OrderCanceled) {
2677            self.on_order_canceled_called = true;
2678        }
2679
2680        fn on_order_filled(&mut self, _event: &OrderFilled) {
2681            self.on_order_filled_called = true;
2682            self.order_event_timeline.borrow_mut().push("specific");
2683        }
2684
2685        fn on_order_fill_voided(&mut self, _event: &OrderFillVoided) {
2686            self.on_order_fill_voided_called = true;
2687        }
2688
2689        fn on_order_rejected(&mut self, _event: OrderRejected) {
2690            self.on_order_rejected_called = true;
2691        }
2692
2693        fn on_order_event(&mut self, _event: OrderEventAny) {
2694            self.on_order_event_called = true;
2695            self.order_event_timeline.borrow_mut().push("aggregate");
2696        }
2697
2698        fn on_order_accepted(&mut self, _event: OrderAccepted) {
2699            self.on_order_accepted_called = true;
2700        }
2701
2702        fn on_order_expired(&mut self, _event: OrderExpired) {
2703            self.on_order_expired_called = true;
2704        }
2705
2706        fn on_position_opened(&mut self, _event: PositionOpened) {
2707            self.on_position_opened_called = true;
2708        }
2709
2710        fn on_position_event(&mut self, _event: PositionEvent) {
2711            self.on_position_event_called = true;
2712        }
2713
2714        fn on_position_changed(&mut self, _event: PositionChanged) {
2715            self.on_position_changed_called = true;
2716        }
2717
2718        fn on_position_closed(&mut self, _event: PositionClosed) {
2719            self.on_position_closed_called = true;
2720        }
2721    });
2722
2723    fn create_test_strategy() -> TestStrategy {
2724        let config = StrategyConfig {
2725            strategy_id: Some(StrategyId::from("TEST-001")),
2726            order_id_tag: Some("001".to_string()),
2727            ..Default::default()
2728        };
2729        TestStrategy::new(config)
2730    }
2731
2732    fn create_gtd_managed_strategy() -> TestStrategy {
2733        TestStrategy::new(StrategyConfig {
2734            strategy_id: Some(StrategyId::from("TEST-001")),
2735            order_id_tag: Some("001".to_string()),
2736            manage_gtd_expiry: true,
2737            ..Default::default()
2738        })
2739    }
2740
2741    fn register_strategy(strategy: &mut TestStrategy) {
2742        let _clock = register_strategy_with_clock(strategy);
2743    }
2744
2745    fn register_strategy_with_clock(strategy: &mut TestStrategy) -> Rc<RefCell<VirtualClock>> {
2746        let trader_id = TraderId::from("TRADER-001");
2747        let clock = Rc::new(RefCell::new(VirtualClock::new()));
2748        let cache = Rc::new(RefCell::new(Cache::default()));
2749        let portfolio = Rc::new(RefCell::new(Portfolio::new(
2750            clock.clone(),
2751            cache.clone(),
2752            None,
2753        )));
2754
2755        strategy
2756            .core
2757            .register(trader_id, clock.clone(), cache, portfolio)
2758            .unwrap();
2759        strategy.initialize().unwrap();
2760        clock
2761    }
2762
2763    fn register_gtd_strategy(strategy: &mut TestStrategy) -> Rc<RefCell<VirtualClock>> {
2764        let clock = register_strategy_with_clock(strategy);
2765        clock
2766            .borrow_mut()
2767            .register_default_handler(TimeEventCallback::from(|_event: TimeEvent| {}));
2768        clock
2769    }
2770
2771    fn start_strategy(strategy: &mut TestStrategy) {
2772        strategy.start().unwrap();
2773    }
2774
2775    fn stop_strategy(strategy: &mut TestStrategy) {
2776        Component::stop(strategy).unwrap();
2777    }
2778
2779    fn make_filled(client_order_id: ClientOrderId) -> OrderEventAny {
2780        OrderEventAny::Filled(
2781            OrderFilledSpec::builder()
2782                .trader_id(TraderId::from("TRADER-001"))
2783                .strategy_id(StrategyId::from("TEST-001"))
2784                .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2785                .client_order_id(client_order_id)
2786                .venue_order_id(VenueOrderId::test_default())
2787                .account_id(AccountId::from("ACC-001"))
2788                .trade_id(TradeId::test_default())
2789                .last_qty(Quantity::default())
2790                .last_px(Price::default())
2791                .currency(Currency::from("USD"))
2792                .liquidity_side(LiquiditySide::Taker)
2793                .event_id(UUID4::default())
2794                .build(),
2795        )
2796    }
2797
2798    fn make_fill_voided(client_order_id: ClientOrderId, is_reopened: bool) -> OrderEventAny {
2799        OrderEventAny::FillVoided(
2800            OrderFillVoidedSpec::builder()
2801                .trader_id(TraderId::from("TRADER-001"))
2802                .strategy_id(StrategyId::from("TEST-001"))
2803                .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2804                .client_order_id(client_order_id)
2805                .venue_order_id(VenueOrderId::test_default())
2806                .account_id(AccountId::from("ACC-001"))
2807                .is_reopened(is_reopened)
2808                .build(),
2809        )
2810    }
2811
2812    fn make_terminal_fill_voided(client_order_id: ClientOrderId) -> OrderEventAny {
2813        make_fill_voided(client_order_id, false)
2814    }
2815
2816    fn make_canceled(client_order_id: ClientOrderId) -> OrderEventAny {
2817        OrderEventAny::Canceled(
2818            OrderCanceledSpec::builder()
2819                .trader_id(TraderId::from("TRADER-001"))
2820                .strategy_id(StrategyId::from("TEST-001"))
2821                .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2822                .client_order_id(client_order_id)
2823                .account_id(AccountId::from("ACC-001"))
2824                .event_id(UUID4::default())
2825                .build(),
2826        )
2827    }
2828
2829    fn make_rejected(client_order_id: ClientOrderId) -> OrderEventAny {
2830        OrderEventAny::Rejected(
2831            OrderRejectedSpec::builder()
2832                .trader_id(TraderId::from("TRADER-001"))
2833                .strategy_id(StrategyId::from("TEST-001"))
2834                .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2835                .client_order_id(client_order_id)
2836                .account_id(AccountId::from("ACC-001"))
2837                .reason("Test rejection".into())
2838                .event_id(UUID4::default())
2839                .build(),
2840        )
2841    }
2842
2843    fn make_cancel_rejected(client_order_id: ClientOrderId) -> OrderEventAny {
2844        OrderEventAny::CancelRejected(
2845            OrderCancelRejectedSpec::builder()
2846                .trader_id(TraderId::from("TRADER-001"))
2847                .strategy_id(StrategyId::from("TEST-001"))
2848                .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2849                .client_order_id(client_order_id)
2850                .reason("Test rejection".into())
2851                .venue_order_id(VenueOrderId::from(client_order_id.as_str()))
2852                .account_id(AccountId::from("ACC-001"))
2853                .build(),
2854        )
2855    }
2856
2857    fn make_expired(client_order_id: ClientOrderId) -> OrderEventAny {
2858        OrderEventAny::Expired(
2859            OrderExpiredSpec::builder()
2860                .trader_id(TraderId::from("TRADER-001"))
2861                .strategy_id(StrategyId::from("TEST-001"))
2862                .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2863                .client_order_id(client_order_id)
2864                .account_id(AccountId::from("ACC-001"))
2865                .event_id(UUID4::default())
2866                .build(),
2867        )
2868    }
2869
2870    fn make_accepted(client_order_id: ClientOrderId) -> OrderEventAny {
2871        OrderEventAny::Accepted(
2872            OrderAcceptedSpec::builder()
2873                .trader_id(TraderId::from("TRADER-001"))
2874                .strategy_id(StrategyId::from("TEST-001"))
2875                .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2876                .client_order_id(client_order_id)
2877                .venue_order_id(VenueOrderId::test_default())
2878                .account_id(AccountId::from("ACC-001"))
2879                .event_id(UUID4::default())
2880                .build(),
2881        )
2882    }
2883
2884    fn make_accepted_market_order(client_order_id: &str) -> OrderAny {
2885        let mut order = OrderAny::Market(MarketOrder::new(
2886            TraderId::from("TRADER-001"),
2887            StrategyId::from("TEST-001"),
2888            InstrumentId::from("BTCUSDT.BINANCE"),
2889            ClientOrderId::from(client_order_id),
2890            OrderSide::Buy,
2891            Quantity::from(100_000),
2892            TimeInForce::Gtc,
2893            UUID4::new(),
2894            UnixNanos::default(),
2895            false,
2896            false,
2897            None,
2898            None,
2899            None,
2900            None,
2901            None,
2902            None,
2903            None,
2904            None,
2905        ));
2906        let account_id = AccountId::from("ACC-001");
2907        order
2908            .apply(TestOrderEventStubs::submitted(&order, account_id))
2909            .unwrap();
2910        order
2911            .apply(TestOrderEventStubs::accepted(
2912                &order,
2913                account_id,
2914                // Derived per order: a venue order ID has a single owning client order, so
2915                // batch tests holding two accepted orders cannot share the default.
2916                VenueOrderId::from(client_order_id),
2917            ))
2918            .unwrap();
2919        order
2920    }
2921
2922    fn make_accepted_limit_order(client_order_id: &str) -> OrderAny {
2923        let mut order = OrderAny::Limit(LimitOrder::new(
2924            TraderId::from("TRADER-001"),
2925            StrategyId::from("TEST-001"),
2926            InstrumentId::from("BTCUSDT.BINANCE"),
2927            ClientOrderId::from(client_order_id),
2928            OrderSide::Buy,
2929            Quantity::from("1.0"),
2930            Price::from("50000.0"),
2931            TimeInForce::Gtc,
2932            None,
2933            false,
2934            false,
2935            false,
2936            None,
2937            None,
2938            None,
2939            None,
2940            None,
2941            None,
2942            None,
2943            None,
2944            None,
2945            None,
2946            None,
2947            UUID4::new(),
2948            UnixNanos::default(),
2949        ));
2950        let account_id = AccountId::from("ACC-001");
2951        order
2952            .apply(TestOrderEventStubs::submitted(&order, account_id))
2953            .unwrap();
2954        order
2955            .apply(TestOrderEventStubs::accepted(
2956                &order,
2957                account_id,
2958                // Derived per order, as above.
2959                VenueOrderId::from(client_order_id),
2960            ))
2961            .unwrap();
2962        order
2963    }
2964
2965    fn make_submitted_gtd_limit_order(client_order_id: &str, expire_time: UnixNanos) -> OrderAny {
2966        let mut order = OrderTestBuilder::new(OrderType::Limit)
2967            .trader_id(TraderId::from("TRADER-001"))
2968            .strategy_id(StrategyId::from("TEST-001"))
2969            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2970            .client_order_id(ClientOrderId::from(client_order_id))
2971            .side(OrderSide::Buy)
2972            .quantity(Quantity::from("1.0"))
2973            .price(Price::from("50000.0"))
2974            .time_in_force(TimeInForce::Gtd)
2975            .expire_time(expire_time)
2976            .build();
2977        let account_id = AccountId::from("ACC-001");
2978        order
2979            .apply(TestOrderEventStubs::submitted(&order, account_id))
2980            .unwrap();
2981        order
2982    }
2983
2984    fn make_accepted_gtd_limit_order(client_order_id: &str, expire_time: UnixNanos) -> OrderAny {
2985        let mut order = make_submitted_gtd_limit_order(client_order_id, expire_time);
2986        let account_id = AccountId::from("ACC-001");
2987        order
2988            .apply(TestOrderEventStubs::accepted(
2989                &order,
2990                account_id,
2991                VenueOrderId::from(client_order_id),
2992            ))
2993            .unwrap();
2994        order
2995    }
2996
2997    fn make_initialized_market_order(client_order_id: &str) -> OrderAny {
2998        OrderAny::Market(MarketOrder::new(
2999            TraderId::from("TRADER-001"),
3000            StrategyId::from("TEST-001"),
3001            InstrumentId::from("BTCUSDT.BINANCE"),
3002            ClientOrderId::from(client_order_id),
3003            OrderSide::Buy,
3004            Quantity::from(100_000),
3005            TimeInForce::Gtc,
3006            UUID4::new(),
3007            UnixNanos::default(),
3008            false,
3009            false,
3010            None,
3011            None,
3012            None,
3013            None,
3014            None,
3015            None,
3016            None,
3017            None,
3018        ))
3019    }
3020
3021    fn make_initialized_trailing_stop_order(
3022        order_type: OrderType,
3023        client_order_id: &str,
3024    ) -> OrderAny {
3025        let mut builder = OrderTestBuilder::new(order_type);
3026        builder
3027            .trader_id(TraderId::from("TRADER-001"))
3028            .strategy_id(StrategyId::from("TEST-001"))
3029            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
3030            .client_order_id(ClientOrderId::from(client_order_id))
3031            .side(OrderSide::Sell)
3032            .quantity(Quantity::from(100_000))
3033            .trigger_price(Price::from("49000.0"))
3034            .trailing_offset(Decimal::from(100))
3035            .trailing_offset_type(TrailingOffsetType::Price);
3036
3037        if order_type == OrderType::TrailingStopLimit {
3038            builder
3039                .price(Price::from("48900.0"))
3040                .limit_offset(Decimal::from(100));
3041        }
3042
3043        builder.build()
3044    }
3045
3046    fn make_initialized_algorithm_order(client_order_id: &str) -> OrderAny {
3047        OrderAny::Market(MarketOrder::new(
3048            TraderId::from("TRADER-001"),
3049            StrategyId::from("TEST-001"),
3050            InstrumentId::from("BTCUSDT.BINANCE"),
3051            ClientOrderId::from(client_order_id),
3052            OrderSide::Buy,
3053            Quantity::from(100_000),
3054            TimeInForce::Gtc,
3055            UUID4::new(),
3056            UnixNanos::default(),
3057            false,
3058            false,
3059            None,
3060            None,
3061            None,
3062            None,
3063            Some(ExecAlgorithmId::from("TWAP")),
3064            None,
3065            Some(ClientOrderId::from(client_order_id)),
3066            None,
3067        ))
3068    }
3069
3070    fn add_order_to_cache(strategy: &TestStrategy, order: &OrderAny) {
3071        let cache_rc = strategy.core.cache_rc();
3072        let mut cache = cache_rc.borrow_mut();
3073        cache.add_order(order.clone(), None, None, true).unwrap();
3074    }
3075
3076    fn make_submit_command(order: &OrderAny) -> SubmitOrder {
3077        SubmitOrder::new(
3078            order.trader_id(),
3079            None,
3080            order.strategy_id(),
3081            order.instrument_id(),
3082            order.client_order_id(),
3083            order.init_event().clone(),
3084            order.exec_algorithm_id(),
3085            None,
3086            None,
3087            UUID4::new(),
3088            UnixNanos::default(),
3089            None, // correlation_id
3090        )
3091    }
3092
3093    fn add_order_to_cache_and_own_book(strategy: &TestStrategy, order: &OrderAny) {
3094        let cache_rc = strategy.core.cache_rc();
3095        let mut cache = cache_rc.borrow_mut();
3096        cache.add_order(order.clone(), None, None, true).unwrap();
3097        cache
3098            .add_own_order_book(OwnOrderBook::new(order.instrument_id()))
3099            .unwrap();
3100        cache.update_own_order_book(order);
3101    }
3102
3103    fn make_position_opened() -> PositionEvent {
3104        PositionEvent::PositionOpened(PositionOpened {
3105            trader_id: TraderId::from("TRADER-001"),
3106            strategy_id: StrategyId::from("TEST-001"),
3107            instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3108            position_id: PositionId::test_default(),
3109            account_id: AccountId::from("ACC-001"),
3110            opening_order_id: ClientOrderId::from("O-001"),
3111            entry: OrderSide::Buy,
3112            side: PositionSide::Long,
3113            signed_qty: 1.0,
3114            quantity: Quantity::default(),
3115            last_qty: Quantity::default(),
3116            last_px: Price::default(),
3117            currency: Currency::from("USD"),
3118            avg_px_open: 0.0,
3119            realized_pnl: None,
3120            event_id: UUID4::default(),
3121            ts_event: UnixNanos::default(),
3122            ts_init: UnixNanos::default(),
3123        })
3124    }
3125
3126    fn make_position_changed() -> PositionEvent {
3127        let currency = Currency::from("USD");
3128        PositionEvent::PositionChanged(PositionChanged {
3129            trader_id: TraderId::from("TRADER-001"),
3130            strategy_id: StrategyId::from("TEST-001"),
3131            instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3132            position_id: PositionId::test_default(),
3133            account_id: AccountId::from("ACC-001"),
3134            opening_order_id: ClientOrderId::from("O-001"),
3135            entry: OrderSide::Buy,
3136            side: PositionSide::Long,
3137            signed_qty: 2.0,
3138            quantity: Quantity::default(),
3139            peak_quantity: Quantity::default(),
3140            last_qty: Quantity::default(),
3141            last_px: Price::default(),
3142            currency,
3143            avg_px_open: 0.0,
3144            avg_px_close: None,
3145            realized_return: 0.0,
3146            realized_pnl: None,
3147            unrealized_pnl: Money::zero(currency),
3148            event_id: UUID4::default(),
3149            ts_opened: UnixNanos::default(),
3150            ts_event: UnixNanos::default(),
3151            ts_init: UnixNanos::default(),
3152        })
3153    }
3154
3155    fn make_position_closed() -> PositionEvent {
3156        let currency = Currency::from("USD");
3157        PositionEvent::PositionClosed(PositionClosed {
3158            trader_id: TraderId::from("TRADER-001"),
3159            strategy_id: StrategyId::from("TEST-001"),
3160            instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3161            position_id: PositionId::test_default(),
3162            account_id: AccountId::from("ACC-001"),
3163            opening_order_id: ClientOrderId::from("O-001"),
3164            closing_order_id: Some(ClientOrderId::from("O-002")),
3165            entry: OrderSide::Buy,
3166            side: PositionSide::Flat,
3167            signed_qty: 0.0,
3168            quantity: Quantity::default(),
3169            peak_quantity: Quantity::default(),
3170            last_qty: Quantity::default(),
3171            last_px: Price::default(),
3172            currency,
3173            avg_px_open: 0.0,
3174            avg_px_close: None,
3175            realized_return: 0.0,
3176            realized_pnl: None,
3177            unrealized_pnl: Money::zero(currency),
3178            duration: DurationNanos::default(),
3179            event_id: UUID4::default(),
3180            ts_opened: UnixNanos::default(),
3181            ts_closed: None,
3182            ts_event: UnixNanos::default(),
3183            ts_init: UnixNanos::default(),
3184        })
3185    }
3186
3187    fn make_position_adjusted() -> PositionEvent {
3188        PositionEvent::PositionAdjusted(PositionAdjusted {
3189            trader_id: TraderId::from("TRADER-001"),
3190            strategy_id: StrategyId::from("TEST-001"),
3191            instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3192            position_id: PositionId::test_default(),
3193            account_id: AccountId::from("ACC-001"),
3194            adjustment_type: PositionAdjustmentType::Funding,
3195            quantity_change: None,
3196            pnl_change: None,
3197            reason: None,
3198            event_id: UUID4::default(),
3199            ts_event: UnixNanos::default(),
3200            ts_init: UnixNanos::default(),
3201        })
3202    }
3203
3204    #[rstest]
3205    fn test_strategy_creation() {
3206        let strategy = create_test_strategy();
3207        assert_eq!(strategy.strategy_id(), Some(StrategyId::from("TEST-001")));
3208        assert!(!strategy.on_order_rejected_called);
3209        assert!(!strategy.on_position_opened_called);
3210    }
3211
3212    #[rstest]
3213    fn test_strategy_registration() {
3214        let mut strategy = create_test_strategy();
3215        register_strategy(&mut strategy);
3216
3217        assert!(strategy.is_registered());
3218        let _ = strategy.order().generate_client_order_id();
3219        let _ = strategy.portfolio().is_initialized();
3220    }
3221
3222    #[rstest]
3223    fn test_set_external_order_instrument_ids_replaces_claims_atomically() {
3224        let mut strategy = create_test_strategy();
3225        register_strategy(&mut strategy);
3226        let cache = strategy.core.cache_rc();
3227        let strategy_id = StrategyId::from("TEST-001");
3228        let other_strategy_id = StrategyId::from("OTHER-001");
3229        let audusd = InstrumentId::from("AUDUSD.SIM");
3230        let eurusd = InstrumentId::from("EURUSD.SIM");
3231        let gbpusd = InstrumentId::from("GBPUSD.SIM");
3232
3233        strategy
3234            .set_external_order_instrument_ids(vec![audusd, eurusd])
3235            .unwrap();
3236        cache
3237            .borrow_mut()
3238            .set_external_order_claims(other_strategy_id, &[gbpusd])
3239            .unwrap();
3240
3241        let result = strategy.set_external_order_instrument_ids(vec![eurusd, gbpusd]);
3242
3243        assert_eq!(
3244            result.unwrap_err().to_string(),
3245            "External order claim for GBPUSD.SIM already exists for OTHER-001"
3246        );
3247        assert_eq!(
3248            cache.borrow().external_order_claim(&audusd),
3249            Some(strategy_id)
3250        );
3251        assert_eq!(
3252            cache.borrow().external_order_claim(&eurusd),
3253            Some(strategy_id)
3254        );
3255        assert_eq!(
3256            cache.borrow().external_order_claim(&gbpusd),
3257            Some(other_strategy_id)
3258        );
3259        assert_eq!(
3260            strategy.core.config.external_order_instrument_ids,
3261            Some(vec![audusd, eurusd])
3262        );
3263
3264        strategy
3265            .set_external_order_instrument_ids(vec![eurusd])
3266            .unwrap();
3267
3268        assert_eq!(cache.borrow().external_order_claim(&audusd), None);
3269        assert_eq!(
3270            cache.borrow().external_order_claim(&eurusd),
3271            Some(strategy_id)
3272        );
3273        assert_eq!(
3274            cache.borrow().external_order_claim(&gbpusd),
3275            Some(other_strategy_id)
3276        );
3277        assert_eq!(
3278            strategy.core.config.external_order_instrument_ids,
3279            Some(vec![eurusd])
3280        );
3281    }
3282
3283    #[rstest]
3284    fn test_set_external_order_instrument_ids_rejects_unregistered_strategy() {
3285        let mut strategy = create_test_strategy();
3286
3287        let error = strategy
3288            .set_external_order_instrument_ids(vec![InstrumentId::from("AUDUSD.SIM")])
3289            .unwrap_err();
3290
3291        assert_eq!(
3292            error.to_string(),
3293            "Strategy TEST-001 is not registered with a trader"
3294        );
3295        assert!(strategy.core.config.external_order_instrument_ids.is_none());
3296    }
3297
3298    #[rstest]
3299    fn test_strategy_native_methods_are_available_on_strategy_type() {
3300        let mut strategy = create_test_strategy();
3301        register_strategy(&mut strategy);
3302
3303        drop(strategy.order_factory());
3304
3305        assert!(Rc::ptr_eq(
3306            &strategy.order_factory_rc(),
3307            strategy.core.order_factory.as_ref().unwrap()
3308        ));
3309        assert!(Rc::ptr_eq(
3310            &strategy.portfolio_rc(),
3311            strategy.core.portfolio.as_ref().unwrap()
3312        ));
3313    }
3314
3315    #[rstest]
3316    #[case(false, false, false, Some(log::Level::Warn))]
3317    #[case(false, false, true, Some(log::Level::Warn))]
3318    #[case(false, true, false, Some(log::Level::Warn))]
3319    #[case(false, true, true, Some(log::Level::Warn))]
3320    #[case(true, false, false, None)]
3321    #[case(true, false, true, Some(log::Level::Info))]
3322    #[case(true, true, false, Some(log::Level::Warn))]
3323    #[case(true, true, true, Some(log::Level::Warn))]
3324    fn test_handle_order_event_logs_and_dispatches_rejection(
3325        #[case] due_post_only: bool,
3326        #[case] log_rejected_due_post_only_as_warning: bool,
3327        #[case] log_events: bool,
3328        #[case] expected_level: Option<log::Level>,
3329    ) {
3330        static LOG_CAPTURE: OrderEventLogCapture = OrderEventLogCapture(Mutex::new(Vec::new()));
3331
3332        let mut strategy = create_test_strategy();
3333        strategy.core.config.log_rejected_due_post_only_as_warning =
3334            log_rejected_due_post_only_as_warning;
3335        strategy.core.actor.config.log_events = log_events;
3336        register_strategy(&mut strategy);
3337        start_strategy(&mut strategy);
3338
3339        let event = OrderEventAny::Rejected(
3340            OrderRejectedSpec::builder()
3341                .strategy_id(StrategyId::from("TEST-001"))
3342                .client_order_id(ClientOrderId::from("O-001"))
3343                .due_post_only(due_post_only)
3344                .build(),
3345        );
3346
3347        log::set_logger(&LOG_CAPTURE).unwrap();
3348        log::set_max_level(log::LevelFilter::Info);
3349        let expected_records: Vec<_> = expected_level
3350            .map(|level| (level, format!("TEST-001 {RECV}{EVT} {event}")))
3351            .into_iter()
3352            .collect();
3353
3354        strategy.handle_order_event(event);
3355
3356        assert_eq!(*LOG_CAPTURE.0.lock().unwrap(), expected_records);
3357        assert!(strategy.on_order_rejected_called);
3358        assert!(strategy.on_order_event_called);
3359    }
3360
3361    struct OrderEventLogCapture(Mutex<Vec<(log::Level, String)>>);
3362
3363    impl log::Log for OrderEventLogCapture {
3364        fn enabled(&self, metadata: &log::Metadata<'_>) -> bool {
3365            metadata.target() == "nautilus_trading::strategy"
3366        }
3367
3368        fn log(&self, record: &log::Record<'_>) {
3369            if self.enabled(record.metadata()) {
3370                self.0
3371                    .lock()
3372                    .unwrap()
3373                    .push((record.level(), record.args().to_string()));
3374            }
3375        }
3376
3377        fn flush(&self) {}
3378    }
3379
3380    #[rstest]
3381    fn test_dispatch_manager_actions_routes_every_action() {
3382        let mut strategy = create_test_strategy();
3383        register_strategy(&mut strategy);
3384        let strategy_id = StrategyId::from("TEST-001");
3385        let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
3386        let initialized_order = make_initialized_market_order("O-MANAGER-INIT");
3387        let emulator_order = OrderTestBuilder::new(OrderType::StopMarket)
3388            .trader_id(TraderId::from("TRADER-001"))
3389            .strategy_id(strategy_id)
3390            .instrument_id(instrument_id)
3391            .client_order_id(ClientOrderId::from("O-MANAGER-EMULATOR"))
3392            .side(OrderSide::Buy)
3393            .trigger_price(Price::from("51000.0"))
3394            .quantity(Quantity::from(100_000))
3395            .emulation_trigger(TriggerType::BidAsk)
3396            .build();
3397        let risk_order = make_initialized_market_order("O-MANAGER-RISK");
3398        let algorithm_order = make_initialized_algorithm_order("O-MANAGER-ALGORITHM");
3399        let cancel_order = OrderTestBuilder::new(OrderType::Limit)
3400            .trader_id(TraderId::from("TRADER-001"))
3401            .strategy_id(strategy_id)
3402            .instrument_id(instrument_id)
3403            .client_order_id(ClientOrderId::from("O-MANAGER-CANCEL"))
3404            .side(OrderSide::Buy)
3405            .price(Price::from("50000.0"))
3406            .quantity(Quantity::from(100_000))
3407            .submit(true)
3408            .build();
3409        let missing_cancel_order = OrderTestBuilder::new(OrderType::Limit)
3410            .trader_id(TraderId::from("TRADER-001"))
3411            .strategy_id(strategy_id)
3412            .instrument_id(instrument_id)
3413            .client_order_id(ClientOrderId::from("O-MANAGER-MISSING-CANCEL"))
3414            .side(OrderSide::Buy)
3415            .price(Price::from("50000.0"))
3416            .quantity(Quantity::from(100_000))
3417            .submit(true)
3418            .build();
3419        let modify_order = OrderTestBuilder::new(OrderType::Limit)
3420            .trader_id(TraderId::from("TRADER-001"))
3421            .strategy_id(strategy_id)
3422            .instrument_id(instrument_id)
3423            .client_order_id(ClientOrderId::from("O-MANAGER-MODIFY"))
3424            .side(OrderSide::Buy)
3425            .price(Price::from("50000.0"))
3426            .quantity(Quantity::from(100_000))
3427            .submit(true)
3428            .build();
3429        add_order_to_cache(&strategy, &cancel_order);
3430        add_order_to_cache(&strategy, &modify_order);
3431
3432        let (order_handler, order_events) =
3433            get_typed_message_saving_handler(Some(Ustr::from("manager-action-order-events")));
3434        let order_topic = format!("events.order.{strategy_id}");
3435        msgbus::subscribe_order_events(order_topic.clone().into(), order_handler.clone(), None);
3436        let (emulator_handler, emulator_messages): (
3437            _,
3438            TypedIntoMessageSavingHandler<TradingCommand>,
3439        ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
3440        msgbus::register_trading_command_endpoint(
3441            MessagingSwitchboard::order_emulator_execute(),
3442            emulator_handler,
3443        );
3444        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3445            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3446        msgbus::register_trading_command_endpoint(
3447            MessagingSwitchboard::risk_engine_queue_execute(),
3448            risk_handler,
3449        );
3450        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3451            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
3452        msgbus::register_trading_command_endpoint(
3453            MessagingSwitchboard::exec_engine_queue_execute(),
3454            exec_handler,
3455        );
3456        let (algorithm_handler, algorithm_messages) =
3457            get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
3458        msgbus::register_any("TWAP.execute".into(), algorithm_handler);
3459
3460        strategy.dispatch_manager_actions(vec![
3461            OrderManagerAction::CancelLocal(missing_cancel_order),
3462            OrderManagerAction::PublishInitialized(OrderEventAny::Initialized(
3463                initialized_order.init_event().clone(),
3464            )),
3465            OrderManagerAction::SubmitToEmulator(make_submit_command(&emulator_order)),
3466            OrderManagerAction::SubmitToRisk(make_submit_command(&risk_order)),
3467            OrderManagerAction::SubmitToAlgorithm {
3468                command: make_submit_command(&algorithm_order),
3469                exec_algorithm_id: ExecAlgorithmId::from("TWAP"),
3470            },
3471            OrderManagerAction::CancelLocal(cancel_order.clone()),
3472            OrderManagerAction::ModifyLocalQuantity {
3473                order: modify_order.clone(),
3474                quantity: Quantity::from(50_000),
3475            },
3476            OrderManagerAction::ModifyLocalQuantity {
3477                order: modify_order.clone(),
3478                quantity: Quantity::from(100_000),
3479            },
3480        ]);
3481        msgbus::unsubscribe_order_events(order_topic.into(), &order_handler);
3482
3483        let order_events = order_events.get_messages();
3484        assert_eq!(order_events.len(), 3);
3485        assert!(matches!(
3486            &order_events[0],
3487            OrderEventAny::Initialized(event)
3488                if event.client_order_id == initialized_order.client_order_id()
3489        ));
3490        assert!(matches!(
3491            &order_events[1],
3492            OrderEventAny::PendingCancel(event)
3493                if event.client_order_id == cancel_order.client_order_id()
3494        ));
3495        assert!(matches!(
3496            &order_events[2],
3497            OrderEventAny::PendingUpdate(event)
3498                if event.client_order_id == modify_order.client_order_id()
3499        ));
3500        assert!(matches!(
3501            emulator_messages.get_messages().as_slice(),
3502            [TradingCommand::SubmitOrder(command)]
3503                if command.client_order_id == emulator_order.client_order_id()
3504        ));
3505        assert!(matches!(
3506            risk_messages.get_messages().as_slice(),
3507            [
3508                TradingCommand::SubmitOrder(submit),
3509                TradingCommand::ModifyOrder(first_modify),
3510                TradingCommand::ModifyOrder(second_modify),
3511            ]
3512
3513                if submit.client_order_id == risk_order.client_order_id()
3514                    && first_modify.client_order_id == modify_order.client_order_id()
3515                    && first_modify.quantity == Some(Quantity::from(50_000))
3516                    && second_modify.client_order_id == modify_order.client_order_id()
3517                    && second_modify.quantity == Some(Quantity::from(100_000))
3518        ));
3519        assert!(matches!(
3520            algorithm_messages.get_messages().as_slice(),
3521            [TradingCommand::SubmitOrder(command)]
3522                if command.client_order_id == algorithm_order.client_order_id()
3523        ));
3524        assert!(matches!(
3525            exec_messages.get_messages().as_slice(),
3526            [TradingCommand::CancelOrder(command)]
3527                if command.client_order_id == cancel_order.client_order_id()
3528        ));
3529    }
3530
3531    #[rstest]
3532    #[case::disabled(false)]
3533    #[case::enabled(true)]
3534    fn test_contingent_manager_dispatch_precedes_user_handlers_and_is_idempotent(
3535        #[case] manage_contingent_orders: bool,
3536    ) {
3537        let mut strategy = TestStrategy::new(StrategyConfig {
3538            strategy_id: Some(StrategyId::from("TEST-001")),
3539            order_id_tag: Some("001".to_string()),
3540            manage_contingent_orders,
3541            ..Default::default()
3542        });
3543        register_strategy(&mut strategy);
3544        start_strategy(&mut strategy);
3545        let timeline = strategy.order_event_timeline.clone();
3546        let endpoint_timeline = timeline.clone();
3547        let exec_handler = TypedIntoHandler::from(move |command: TradingCommand| {
3548            assert!(matches!(command, TradingCommand::CancelOrder(_)));
3549            endpoint_timeline.borrow_mut().push("endpoint");
3550        });
3551        msgbus::register_trading_command_endpoint(
3552            MessagingSwitchboard::exec_engine_queue_execute(),
3553            exec_handler,
3554        );
3555        let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
3556        let parent_id = ClientOrderId::from("O-CONTINGENT-PARENT");
3557        let sibling_id = ClientOrderId::from("O-CONTINGENT-SIBLING");
3558        let parent = OrderTestBuilder::new(OrderType::Limit)
3559            .trader_id(TraderId::from("TRADER-001"))
3560            .strategy_id(StrategyId::from("TEST-001"))
3561            .instrument_id(instrument_id)
3562            .client_order_id(parent_id)
3563            .side(OrderSide::Buy)
3564            .price(Price::from("50000.0"))
3565            .quantity(Quantity::from(100_000))
3566            .contingency_type(ContingencyType::Oco)
3567            .linked_order_ids(vec![parent_id, sibling_id])
3568            .submit(true)
3569            .build();
3570        let sibling = OrderTestBuilder::new(OrderType::Limit)
3571            .trader_id(TraderId::from("TRADER-001"))
3572            .strategy_id(StrategyId::from("TEST-001"))
3573            .instrument_id(instrument_id)
3574            .client_order_id(sibling_id)
3575            .side(OrderSide::Sell)
3576            .price(Price::from("51000.0"))
3577            .quantity(Quantity::from(100_000))
3578            .submit(true)
3579            .build();
3580        add_order_to_cache(&strategy, &parent);
3581        add_order_to_cache(&strategy, &sibling);
3582        let event = OrderEventAny::Filled(
3583            OrderFilledSpec::builder()
3584                .trader_id(parent.trader_id())
3585                .strategy_id(parent.strategy_id())
3586                .instrument_id(parent.instrument_id())
3587                .client_order_id(parent.client_order_id())
3588                .venue_order_id(VenueOrderId::from("V-CONTINGENT-PARENT"))
3589                .account_id(AccountId::from("ACCOUNT-001"))
3590                .trade_id(TradeId::from("T-CONTINGENT-PARENT"))
3591                .order_side(parent.order_side())
3592                .order_type(parent.order_type())
3593                .last_qty(parent.quantity())
3594                .last_px(Price::from("50000.0"))
3595                .liquidity_side(LiquiditySide::Taker)
3596                .build(),
3597        );
3598        strategy
3599            .core
3600            .cache_rc()
3601            .borrow_mut()
3602            .update_order(&event)
3603            .unwrap();
3604
3605        strategy.handle_order_event(event.clone());
3606        strategy.handle_order_event(event);
3607
3608        let timeline = timeline.borrow();
3609        let sibling_status = strategy
3610            .core
3611            .cache_ref()
3612            .order(&sibling_id)
3613            .unwrap()
3614            .status();
3615
3616        if manage_contingent_orders {
3617            assert_eq!(
3618                timeline.as_slice(),
3619                ["endpoint", "specific", "aggregate", "specific", "aggregate"]
3620            );
3621            assert_eq!(sibling_status, OrderStatus::PendingCancel);
3622        } else {
3623            assert_eq!(
3624                timeline.as_slice(),
3625                ["specific", "aggregate", "specific", "aggregate"]
3626            );
3627            assert_eq!(sibling_status, OrderStatus::Submitted);
3628        }
3629    }
3630
3631    #[rstest]
3632    fn test_handle_order_fill_voided_dispatches_to_specific_handler() {
3633        let mut strategy = create_test_strategy();
3634        register_strategy(&mut strategy);
3635        start_strategy(&mut strategy);
3636
3637        strategy.handle_order_event(make_fill_voided(ClientOrderId::from("O-001"), false));
3638
3639        assert!(strategy.on_order_fill_voided_called);
3640        assert!(strategy.on_order_event_called);
3641    }
3642
3643    #[rstest]
3644    #[case::opened(make_position_opened())]
3645    #[case::changed(make_position_changed())]
3646    #[case::closed(make_position_closed())]
3647    fn test_handle_position_event_dispatches_to_handler(#[case] event: PositionEvent) {
3648        let mut strategy = create_test_strategy();
3649        register_strategy(&mut strategy);
3650        start_strategy(&mut strategy);
3651
3652        let expected_opened = matches!(event, PositionEvent::PositionOpened(_));
3653        let expected_changed = matches!(event, PositionEvent::PositionChanged(_));
3654        let expected_closed = matches!(event, PositionEvent::PositionClosed(_));
3655
3656        strategy.handle_position_event(event);
3657
3658        assert_eq!(strategy.on_position_opened_called, expected_opened);
3659        assert_eq!(strategy.on_position_changed_called, expected_changed);
3660        assert_eq!(strategy.on_position_closed_called, expected_closed);
3661        assert!(strategy.on_position_event_called);
3662    }
3663
3664    #[rstest]
3665    fn test_handle_position_event_skips_dispatch_when_stopped() {
3666        let mut strategy = create_test_strategy();
3667        register_strategy(&mut strategy);
3668        start_strategy(&mut strategy);
3669        stop_strategy(&mut strategy);
3670        assert_eq!(strategy.state(), ComponentState::Stopped);
3671
3672        strategy.handle_position_event(make_position_opened());
3673
3674        assert!(!strategy.on_position_event_called);
3675        assert!(!strategy.on_position_opened_called);
3676    }
3677
3678    #[rstest]
3679    fn test_handle_position_event_skips_dispatch_for_adjusted() {
3680        let mut strategy = create_test_strategy();
3681        register_strategy(&mut strategy);
3682        start_strategy(&mut strategy);
3683
3684        strategy.handle_position_event(make_position_adjusted());
3685
3686        assert!(!strategy.on_position_event_called);
3687        assert!(!strategy.on_position_opened_called);
3688        assert!(!strategy.on_position_changed_called);
3689        assert!(!strategy.on_position_closed_called);
3690    }
3691
3692    #[rstest]
3693    fn test_strategy_default_handlers_do_not_panic() {
3694        let mut strategy = create_test_strategy();
3695
3696        strategy.on_order_initialized(OrderInitialized::default());
3697        strategy.on_order_event(OrderEventAny::Accepted(OrderAccepted::default()));
3698        strategy.on_order_denied(OrderDenied::default());
3699        strategy.on_order_emulated(OrderEmulated::default());
3700        strategy.on_order_released(OrderReleased::default());
3701        strategy.on_order_submitted(OrderSubmitted::default());
3702        strategy.on_order_rejected(OrderRejected::default());
3703        strategy.on_order_canceled(&OrderCanceled::default());
3704        strategy.on_order_expired(OrderExpired::default());
3705        strategy.on_order_triggered(OrderTriggered::default());
3706        strategy.on_order_pending_update(OrderPendingUpdate::default());
3707        strategy.on_order_pending_cancel(OrderPendingCancel::default());
3708        strategy.on_order_modify_rejected(OrderModifyRejected::default());
3709        strategy.on_order_cancel_rejected(OrderCancelRejected::default());
3710        strategy.on_order_updated(OrderUpdated::default());
3711        strategy.on_order_filled(&OrderFilledSpec::builder().build());
3712        strategy.on_order_fill_voided(&OrderFillVoidedSpec::builder().build());
3713        strategy.on_position_event(make_position_opened());
3714    }
3715
3716    #[rstest]
3717    fn test_submit_order_publishes_order_initialized_after_cache_insert_before_send() {
3718        let mut strategy = create_test_strategy();
3719        register_strategy(&mut strategy);
3720
3721        let order = make_initialized_market_order("O-20250208-INIT-001");
3722        let client_order_id = order.client_order_id();
3723        let cache_rc = strategy.core.cache_rc();
3724        let timeline = Rc::new(RefCell::new(Vec::new()));
3725        let event_messages = Rc::new(RefCell::new(Vec::new()));
3726
3727        let event_handler = {
3728            let event_messages = event_messages.clone();
3729            let timeline = timeline.clone();
3730            TypedHandler::from_with_id("events.order.initialized", move |event: &OrderEventAny| {
3731                assert!(cache_rc.borrow().order_exists(&client_order_id));
3732                assert!(matches!(event, OrderEventAny::Initialized(_)));
3733                event_messages.borrow_mut().push(event.clone());
3734                timeline.borrow_mut().push("init");
3735            })
3736        };
3737        let risk_handler = {
3738            let timeline = timeline.clone();
3739            TypedIntoHandler::from_with_id(
3740                "RiskEngine.queue_execute",
3741                move |command: TradingCommand| {
3742                    assert!(matches!(command, TradingCommand::SubmitOrder(_)));
3743                    timeline.borrow_mut().push("command");
3744                },
3745            )
3746        };
3747        msgbus::register_trading_command_endpoint(
3748            MessagingSwitchboard::risk_engine_queue_execute(),
3749            risk_handler,
3750        );
3751
3752        let topic = format!("events.order.{}", order.strategy_id());
3753        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3754
3755        strategy
3756            .submit_order(order.clone(), None, None, None)
3757            .unwrap();
3758
3759        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3760
3761        let event_messages = event_messages.borrow();
3762        assert_eq!(event_messages.len(), 1);
3763        assert_eq!(
3764            event_messages[0],
3765            OrderEventAny::Initialized(order.init_event().clone())
3766        );
3767        assert_eq!(timeline.borrow().as_slice(), &["init", "command"]);
3768    }
3769
3770    #[rstest]
3771    fn test_submit_order_routes_emulated_order_to_order_emulator() {
3772        let mut strategy = create_test_strategy();
3773        register_strategy(&mut strategy);
3774        let (emulator_handler, emulator_messages): (
3775            _,
3776            TypedIntoMessageSavingHandler<TradingCommand>,
3777        ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
3778        msgbus::register_trading_command_endpoint(
3779            MessagingSwitchboard::order_emulator_execute(),
3780            emulator_handler,
3781        );
3782        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3783            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3784        msgbus::register_trading_command_endpoint(
3785            MessagingSwitchboard::risk_engine_queue_execute(),
3786            risk_handler,
3787        );
3788        let order = OrderTestBuilder::new(OrderType::StopMarket)
3789            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
3790            .client_order_id(ClientOrderId::from("O-20250208-EMULATED-001"))
3791            .side(OrderSide::Buy)
3792            .trigger_price(Price::from("51000.0"))
3793            .quantity(Quantity::from(100_000))
3794            .emulation_trigger(TriggerType::BidAsk)
3795            .build();
3796        let client_order_id = order.client_order_id();
3797
3798        strategy.submit_order(order, None, None, None).unwrap();
3799
3800        let emulator_messages = emulator_messages.get_messages();
3801        assert_eq!(emulator_messages.len(), 1);
3802        assert!(matches!(
3803            emulator_messages.first(),
3804            Some(TradingCommand::SubmitOrder(command))
3805                if command.client_order_id == client_order_id
3806        ));
3807        assert!(risk_messages.get_messages().is_empty());
3808    }
3809
3810    #[rstest]
3811    fn test_submit_order_errors_when_strategy_not_registered() {
3812        let mut strategy = create_test_strategy();
3813        let order = make_initialized_market_order("O-20250208-UNREGISTERED-001");
3814
3815        let err = strategy
3816            .submit_order(order, None, None, None)
3817            .unwrap_err()
3818            .to_string();
3819
3820        assert_eq!(err, "Strategy not registered: trader_id is not set");
3821    }
3822
3823    #[rstest]
3824    fn test_submit_order_uses_stored_strategy_id_when_actor_id_diverges() {
3825        let mut strategy = create_test_strategy();
3826        register_strategy(&mut strategy);
3827        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3828            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3829        msgbus::register_trading_command_endpoint(
3830            MessagingSwitchboard::risk_engine_queue_execute(),
3831            risk_handler,
3832        );
3833
3834        // A hyphenless actor ID has no strategy ID form, so any re-parse of it panics
3835        strategy.core.actor.actor_id = ActorId::from("Strategy");
3836        let order = make_initialized_market_order("O-20250208-DIVERGED-001");
3837
3838        strategy.submit_order(order, None, None, None).unwrap();
3839
3840        let risk_messages = risk_messages.get_messages();
3841        assert_eq!(risk_messages.len(), 1);
3842        let Some(TradingCommand::SubmitOrder(command)) = risk_messages.first() else {
3843            panic!("Expected a SubmitOrder command, was {risk_messages:?}");
3844        };
3845        assert_eq!(command.strategy_id, StrategyId::from("TEST-001"));
3846        assert_eq!(
3847            command.client_order_id,
3848            ClientOrderId::from("O-20250208-DIVERGED-001")
3849        );
3850    }
3851
3852    #[rstest]
3853    fn test_required_account_id_errors_when_missing_for_strategy_event() {
3854        let order = make_initialized_market_order("O-20250208-NO-ACCOUNT-001");
3855
3856        let err = required_account_id(&order, "pending cancel")
3857            .unwrap_err()
3858            .to_string();
3859
3860        assert_eq!(
3861            err,
3862            "Cannot generate pending cancel event for O-20250208-NO-ACCOUNT-001: \
3863             account_id is not set"
3864        );
3865    }
3866
3867    #[rstest]
3868    fn test_submit_order_rejects_non_initialized_without_events() {
3869        let mut strategy = create_test_strategy();
3870        register_strategy(&mut strategy);
3871
3872        let order = make_accepted_market_order("O-20250208-ACCEPTED-001");
3873        let topic = format!("events.order.{}", order.strategy_id());
3874        let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
3875            get_typed_message_saving_handler(Some(Ustr::from("events.order.invalid")));
3876
3877        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3878        let result = strategy.submit_order(order, None, None, None);
3879
3880        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3881
3882        assert!(result.is_err());
3883        assert!(
3884            result
3885                .unwrap_err()
3886                .to_string()
3887                .contains("expected INITIALIZED")
3888        );
3889        assert!(event_messages.get_messages().is_empty());
3890    }
3891
3892    #[rstest]
3893    fn test_submit_order_returns_error_when_cache_already_borrowed() {
3894        let mut strategy = create_test_strategy();
3895        register_strategy(&mut strategy);
3896
3897        let order = make_initialized_market_order("O-20250208-BORROWED-001");
3898        let cache_rc = strategy.core.cache_rc();
3899        let _cache = cache_rc.borrow();
3900
3901        let result = catch_unwind(AssertUnwindSafe(|| {
3902            strategy.submit_order(order, None, None, None)
3903        }));
3904
3905        let err = result
3906            .expect("submit_order should not panic")
3907            .unwrap_err()
3908            .to_string();
3909
3910        assert_eq!(
3911            err,
3912            "Cannot submit order O-20250208-BORROWED-001: cache is currently borrowed"
3913        );
3914    }
3915
3916    #[rstest]
3917    fn test_submit_order_list_publishes_order_initialized_after_cache_insert_before_send() {
3918        let mut strategy = create_test_strategy();
3919        register_strategy(&mut strategy);
3920
3921        let order_list_id = OrderListId::from("OL-20250208-LIST-INIT");
3922        let mut orders = vec![
3923            make_initialized_market_order("O-20250208-LIST-INIT-001"),
3924            make_initialized_market_order("O-20250208-LIST-INIT-002"),
3925        ];
3926
3927        for order in &mut orders {
3928            order.set_order_list_id(order_list_id);
3929        }
3930
3931        let client_order_id1 = orders[0].client_order_id();
3932        let client_order_id2 = orders[1].client_order_id();
3933        let cache_rc = strategy.core.cache_rc();
3934        let timeline = Rc::new(RefCell::new(Vec::new()));
3935        let event_messages = Rc::new(RefCell::new(Vec::new()));
3936
3937        let event_handler = {
3938            let event_messages = event_messages.clone();
3939            let timeline = timeline.clone();
3940            TypedHandler::from_with_id(
3941                "events.order.list_initialized",
3942                move |event: &OrderEventAny| {
3943                    match event {
3944                        OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
3945                            let cache = cache_rc.borrow();
3946                            assert!(cache.order_exists(&client_order_id1));
3947                            assert!(cache.order_exists(&client_order_id2));
3948                            assert!(cache.order_list_exists(&order_list_id));
3949                            let order_list = cache.order_list(&order_list_id).unwrap();
3950                            assert_eq!(
3951                                order_list.client_order_ids.as_slice(),
3952                                &[client_order_id1, client_order_id2]
3953                            );
3954                            timeline.borrow_mut().push("init1");
3955                        }
3956                        OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
3957                            assert!(cache_rc.borrow().order_exists(&client_order_id2));
3958                            timeline.borrow_mut().push("init2");
3959                        }
3960                        _ => panic!("unexpected order event {event:?}"),
3961                    }
3962                    event_messages.borrow_mut().push(event.clone());
3963                },
3964            )
3965        };
3966        let risk_handler = {
3967            let timeline = timeline.clone();
3968            TypedIntoHandler::from_with_id(
3969                "RiskEngine.queue_execute",
3970                move |command: TradingCommand| {
3971                    assert!(matches!(command, TradingCommand::SubmitOrderList(_)));
3972                    timeline.borrow_mut().push("command");
3973                },
3974            )
3975        };
3976        msgbus::register_trading_command_endpoint(
3977            MessagingSwitchboard::risk_engine_queue_execute(),
3978            risk_handler,
3979        );
3980
3981        let topic = format!("events.order.{}", orders[0].strategy_id());
3982        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3983
3984        strategy
3985            .submit_order_list(orders.clone(), None, None, None)
3986            .unwrap();
3987
3988        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3989
3990        let event_messages = event_messages.borrow();
3991        assert_eq!(event_messages.len(), 2);
3992        assert_eq!(
3993            event_messages[0],
3994            OrderEventAny::Initialized(orders[0].init_event().clone())
3995        );
3996        assert_eq!(
3997            event_messages[1],
3998            OrderEventAny::Initialized(orders[1].init_event().clone())
3999        );
4000        assert_eq!(timeline.borrow().as_slice(), &["init1", "init2", "command"]);
4001    }
4002
4003    #[rstest]
4004    fn test_submit_order_list_returns_error_when_cache_already_borrowed() {
4005        let mut strategy = create_test_strategy();
4006        register_strategy(&mut strategy);
4007
4008        let order_list_id = OrderListId::from("OL-20250208-BORROWED");
4009        let mut orders = vec![
4010            make_initialized_market_order("O-20250208-LIST-BORROWED-001"),
4011            make_initialized_market_order("O-20250208-LIST-BORROWED-002"),
4012        ];
4013
4014        for order in &mut orders {
4015            order.set_order_list_id(order_list_id);
4016        }
4017
4018        let cache_rc = strategy.core.cache_rc();
4019        let _cache = cache_rc.borrow();
4020
4021        let result = catch_unwind(AssertUnwindSafe(|| {
4022            strategy.submit_order_list(orders, None, None, None)
4023        }));
4024
4025        let err = result
4026            .expect("submit_order_list should not panic")
4027            .unwrap_err()
4028            .to_string();
4029
4030        assert_eq!(
4031            err,
4032            "Cannot submit order list OL-20250208-BORROWED: cache is currently borrowed"
4033        );
4034    }
4035
4036    #[rstest]
4037    fn test_submit_order_list_create_list_branch_publishes_init_after_cache_insert() {
4038        let mut strategy = create_test_strategy();
4039        register_strategy(&mut strategy);
4040
4041        let orders = vec![
4042            make_initialized_market_order("O-20250208-LIST-CREATE-001"),
4043            make_initialized_market_order("O-20250208-LIST-CREATE-002"),
4044        ];
4045
4046        let client_order_id1 = orders[0].client_order_id();
4047        let client_order_id2 = orders[1].client_order_id();
4048        let cache_rc = strategy.core.cache_rc();
4049        let timeline = Rc::new(RefCell::new(Vec::new()));
4050        let event_messages = Rc::new(RefCell::new(Vec::new()));
4051
4052        let event_handler = {
4053            let event_messages = event_messages.clone();
4054            let timeline = timeline.clone();
4055            TypedHandler::from_with_id(
4056                "events.order.list_create_initialized",
4057                move |event: &OrderEventAny| {
4058                    match event {
4059                        OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
4060                            let cache = cache_rc.borrow();
4061                            let cached_order1 = cache.order(&client_order_id1).unwrap();
4062                            let cached_order2 = cache.order(&client_order_id2).unwrap();
4063                            let order_list_id = cached_order1.order_list_id().unwrap();
4064                            assert_eq!(cached_order2.order_list_id(), Some(order_list_id));
4065                            assert_eq!(e.order_list_id, Some(order_list_id));
4066                            assert!(cache.order_list_exists(&order_list_id));
4067                            let order_list = cache.order_list(&order_list_id).unwrap();
4068                            assert_eq!(
4069                                order_list.client_order_ids.as_slice(),
4070                                &[client_order_id1, client_order_id2]
4071                            );
4072                            timeline.borrow_mut().push("init1");
4073                        }
4074                        OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
4075                            let cache = cache_rc.borrow();
4076                            let cached_order = cache.order(&client_order_id2).unwrap();
4077                            assert_eq!(e.order_list_id, cached_order.order_list_id());
4078                            timeline.borrow_mut().push("init2");
4079                        }
4080                        _ => panic!("unexpected order event {event:?}"),
4081                    }
4082                    event_messages.borrow_mut().push(event.clone());
4083                },
4084            )
4085        };
4086        let risk_handler = {
4087            let timeline = timeline.clone();
4088            TypedIntoHandler::from_with_id(
4089                "RiskEngine.queue_execute",
4090                move |command: TradingCommand| {
4091                    let TradingCommand::SubmitOrderList(command) = command else {
4092                        panic!("expected SubmitOrderList command");
4093                    };
4094                    assert!(
4095                        command
4096                            .order_inits
4097                            .iter()
4098                            .all(|init| init.order_list_id == Some(command.order_list.id))
4099                    );
4100                    timeline.borrow_mut().push("command");
4101                },
4102            )
4103        };
4104        msgbus::register_trading_command_endpoint(
4105            MessagingSwitchboard::risk_engine_queue_execute(),
4106            risk_handler,
4107        );
4108
4109        let topic = format!("events.order.{}", orders[0].strategy_id());
4110        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4111
4112        strategy
4113            .submit_order_list(orders, None, None, None)
4114            .unwrap();
4115
4116        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4117
4118        let cache = strategy.cache();
4119        let cached_order1 = cache.order(&client_order_id1).unwrap();
4120        let cached_order2 = cache.order(&client_order_id2).unwrap();
4121        let order_list_id = cached_order1.order_list_id().unwrap();
4122        assert_eq!(cached_order2.order_list_id(), Some(order_list_id));
4123
4124        let event_messages = event_messages.borrow();
4125        assert_eq!(event_messages.len(), 2);
4126        let OrderEventAny::Initialized(init1) = &event_messages[0] else {
4127            panic!("expected first OrderInitialized event");
4128        };
4129        let OrderEventAny::Initialized(init2) = &event_messages[1] else {
4130            panic!("expected second OrderInitialized event");
4131        };
4132        assert_eq!(init1.order_list_id, Some(order_list_id));
4133        assert_eq!(init2.order_list_id, Some(order_list_id));
4134        assert_eq!(timeline.borrow().as_slice(), &["init1", "init2", "command"]);
4135
4136        let order_list = cache.order_list(&order_list_id).unwrap();
4137        assert_eq!(
4138            order_list.client_order_ids.as_slice(),
4139            &[client_order_id1, client_order_id2]
4140        );
4141    }
4142
4143    #[rstest]
4144    fn test_submit_order_list_routes_optional_params_to_risk() {
4145        let mut strategy = create_test_strategy();
4146        register_strategy(&mut strategy);
4147
4148        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4149            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4150        msgbus::register_trading_command_endpoint(
4151            MessagingSwitchboard::risk_engine_queue_execute(),
4152            risk_handler,
4153        );
4154
4155        let no_params_orders = vec![
4156            make_initialized_market_order("O-20250208-LIST-001"),
4157            make_initialized_market_order("O-20250208-LIST-002"),
4158        ];
4159        strategy
4160            .submit_order_list(no_params_orders, None, None, None)
4161            .unwrap();
4162
4163        let mut params = Params::new();
4164        params.insert(
4165            "routing_hint".to_string(),
4166            Value::String("prefer_batch".to_string()),
4167        );
4168        let param_orders = vec![
4169            make_initialized_market_order("O-20250208-LIST-003"),
4170            make_initialized_market_order("O-20250208-LIST-004"),
4171        ];
4172        strategy
4173            .submit_order_list(param_orders, None, None, Some(params.clone()))
4174            .unwrap();
4175
4176        let risk_messages = risk_messages.get_messages();
4177        assert_eq!(risk_messages.len(), 2);
4178        let Some(TradingCommand::SubmitOrderList(no_params_command)) = risk_messages.first() else {
4179            panic!("expected SubmitOrderList command");
4180        };
4181        let Some(TradingCommand::SubmitOrderList(param_command)) = risk_messages.get(1) else {
4182            panic!("expected SubmitOrderList command");
4183        };
4184        assert!(no_params_command.params.is_none());
4185        assert_eq!(param_command.params.as_ref(), Some(&params));
4186    }
4187
4188    #[rstest]
4189    fn test_modify_order_routes_non_emulated_orders_to_risk() {
4190        let mut strategy = create_test_strategy();
4191        register_strategy(&mut strategy);
4192
4193        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4194            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4195        msgbus::register_trading_command_endpoint(
4196            MessagingSwitchboard::risk_engine_queue_execute(),
4197            risk_handler,
4198        );
4199
4200        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4201            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
4202        msgbus::register_trading_command_endpoint(
4203            MessagingSwitchboard::exec_engine_queue_execute(),
4204            exec_handler,
4205        );
4206
4207        let order = OrderAny::Market(MarketOrder::new(
4208            TraderId::from("TRADER-001"),
4209            StrategyId::from("TEST-001"),
4210            InstrumentId::from("BTCUSDT.BINANCE"),
4211            ClientOrderId::from("O-20250208-0003"),
4212            OrderSide::Buy,
4213            Quantity::from(100_000),
4214            TimeInForce::Gtc,
4215            UUID4::new(),
4216            UnixNanos::default(),
4217            false,
4218            false,
4219            None,
4220            None,
4221            None,
4222            None,
4223            None,
4224            None,
4225            None,
4226            None,
4227        ));
4228        add_order_to_cache(&strategy, &order);
4229
4230        strategy
4231            .modify_order(
4232                order.client_order_id(),
4233                Some(Quantity::from(200_000)),
4234                None,
4235                None,
4236                None,
4237                None,
4238            )
4239            .unwrap();
4240
4241        let risk_messages = risk_messages.get_messages();
4242        let exec_messages = exec_messages.get_messages();
4243
4244        assert_eq!(risk_messages.len(), 1);
4245        assert!(matches!(
4246            risk_messages.first(),
4247            Some(TradingCommand::ModifyOrder(_))
4248        ));
4249        assert!(exec_messages.is_empty());
4250    }
4251
4252    #[rstest]
4253    #[case::trailing_stop_market(OrderType::TrailingStopMarket)]
4254    #[case::trailing_stop_limit(OrderType::TrailingStopLimit)]
4255    fn test_modify_order_changes_trailing_stop_trigger(#[case] order_type: OrderType) {
4256        let mut strategy = create_test_strategy();
4257        register_strategy(&mut strategy);
4258
4259        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4260            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4261        msgbus::register_trading_command_endpoint(
4262            MessagingSwitchboard::risk_engine_queue_execute(),
4263            risk_handler,
4264        );
4265
4266        let order = make_initialized_trailing_stop_order(order_type, "O-TRAILING-MODIFY");
4267        add_order_to_cache(&strategy, &order);
4268
4269        strategy
4270            .modify_order(
4271                order.client_order_id(),
4272                None,
4273                None,
4274                Some(Price::from("49500.0")),
4275                None,
4276                None,
4277            )
4278            .unwrap();
4279
4280        let risk_messages = risk_messages.get_messages();
4281
4282        let [TradingCommand::ModifyOrder(command)] = risk_messages.as_slice() else {
4283            panic!("expected one ModifyOrder command, was {risk_messages:?}");
4284        };
4285
4286        assert_eq!(command.client_order_id, order.client_order_id());
4287        assert_eq!(command.quantity, None);
4288        assert_eq!(command.price, None);
4289        assert_eq!(command.trigger_price, Some(Price::from("49500.0")));
4290    }
4291
4292    #[rstest]
4293    #[case::trailing_stop_market(OrderType::TrailingStopMarket)]
4294    #[case::trailing_stop_limit(OrderType::TrailingStopLimit)]
4295    fn test_modify_orders_changes_trailing_stop_trigger(#[case] order_type: OrderType) {
4296        let mut strategy = create_test_strategy();
4297        register_strategy(&mut strategy);
4298
4299        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4300            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4301        msgbus::register_trading_command_endpoint(
4302            MessagingSwitchboard::risk_engine_queue_execute(),
4303            risk_handler,
4304        );
4305
4306        let mut order = make_initialized_trailing_stop_order(order_type, "O-TRAILING-BATCH-MODIFY");
4307        let account_id = AccountId::from("ACC-001");
4308        order
4309            .apply(TestOrderEventStubs::submitted(&order, account_id))
4310            .unwrap();
4311        order
4312            .apply(TestOrderEventStubs::accepted(
4313                &order,
4314                account_id,
4315                VenueOrderId::from("V-TRAILING-BATCH-MODIFY"),
4316            ))
4317            .unwrap();
4318        add_order_to_cache(&strategy, &order);
4319
4320        strategy
4321            .modify_orders(
4322                vec![(
4323                    order.client_order_id(),
4324                    None,
4325                    None,
4326                    Some(Price::from("49500.0")),
4327                )],
4328                None,
4329                None,
4330            )
4331            .unwrap();
4332
4333        let risk_messages = risk_messages.get_messages();
4334
4335        let [TradingCommand::ModifyOrders(command)] = risk_messages.as_slice() else {
4336            panic!("expected one BatchModifyOrders command, was {risk_messages:?}");
4337        };
4338
4339        let [modify] = command.modifies.as_slice() else {
4340            panic!("expected one ModifyOrder, was {:?}", command.modifies);
4341        };
4342
4343        assert_eq!(modify.client_order_id, order.client_order_id());
4344        assert_eq!(
4345            modify.venue_order_id,
4346            Some(VenueOrderId::from("V-TRAILING-BATCH-MODIFY"))
4347        );
4348        assert_eq!(modify.quantity, None);
4349        assert_eq!(modify.price, None);
4350        assert_eq!(modify.trigger_price, Some(Price::from("49500.0")));
4351    }
4352
4353    #[rstest]
4354    #[case::market(OrderType::Market)]
4355    #[case::limit(OrderType::Limit)]
4356    #[case::market_to_limit(OrderType::MarketToLimit)]
4357    fn test_modify_order_rejects_trigger_for_order_without_trigger(#[case] order_type: OrderType) {
4358        let mut strategy = create_test_strategy();
4359        register_strategy(&mut strategy);
4360
4361        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4362            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4363        msgbus::register_trading_command_endpoint(
4364            MessagingSwitchboard::risk_engine_queue_execute(),
4365            risk_handler,
4366        );
4367
4368        let mut builder = OrderTestBuilder::new(order_type);
4369        builder
4370            .trader_id(TraderId::from("TRADER-001"))
4371            .strategy_id(StrategyId::from("TEST-001"))
4372            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4373            .client_order_id(ClientOrderId::from("O-NO-TRIGGER-MODIFY"))
4374            .quantity(Quantity::from(100_000));
4375
4376        if order_type == OrderType::Limit {
4377            builder.price(Price::from("50000.0"));
4378        }
4379
4380        let order = builder.build();
4381        add_order_to_cache(&strategy, &order);
4382
4383        let result = strategy.modify_order(
4384            order.client_order_id(),
4385            None,
4386            None,
4387            Some(Price::from("49500.0")),
4388            None,
4389            None,
4390        );
4391
4392        assert_eq!(
4393            result.unwrap_err().to_string(),
4394            format!("{order_type} orders do not have a STOP trigger price")
4395        );
4396        assert!(risk_messages.get_messages().is_empty());
4397    }
4398
4399    #[rstest]
4400    fn test_modify_order_routes_active_local_algorithm_order_to_algorithm() {
4401        let mut strategy = create_test_strategy();
4402        register_strategy(&mut strategy);
4403
4404        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4405            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4406        msgbus::register_trading_command_endpoint(
4407            MessagingSwitchboard::risk_engine_queue_execute(),
4408            risk_handler,
4409        );
4410        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4411            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
4412        msgbus::register_trading_command_endpoint(
4413            MessagingSwitchboard::exec_engine_queue_execute(),
4414            exec_handler,
4415        );
4416        let (algo_handler, algo_messages) =
4417            get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4418        msgbus::register_any("TWAP.execute".into(), algo_handler);
4419
4420        let order = make_initialized_algorithm_order("O-20250208-ALGO-MODIFY-001");
4421        add_order_to_cache(&strategy, &order);
4422
4423        strategy
4424            .modify_order(
4425                order.client_order_id(),
4426                Some(Quantity::from(200_000)),
4427                None,
4428                None,
4429                None,
4430                None,
4431            )
4432            .unwrap();
4433
4434        let algo_messages = algo_messages.get_messages();
4435        assert_eq!(algo_messages.len(), 1);
4436        assert!(matches!(
4437            algo_messages.first(),
4438            Some(TradingCommand::ModifyOrder(command))
4439                if command.client_order_id == order.client_order_id()
4440        ));
4441        assert!(risk_messages.get_messages().is_empty());
4442        assert!(exec_messages.get_messages().is_empty());
4443    }
4444
4445    #[rstest]
4446    fn test_modify_order_routes_accepted_algorithm_order_to_risk() {
4447        let mut strategy = create_test_strategy();
4448        register_strategy(&mut strategy);
4449
4450        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4451            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4452        msgbus::register_trading_command_endpoint(
4453            MessagingSwitchboard::risk_engine_queue_execute(),
4454            risk_handler,
4455        );
4456        let (algo_handler, algo_messages) =
4457            get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4458        msgbus::register_any("TWAP.execute".into(), algo_handler);
4459
4460        let mut order = make_initialized_algorithm_order("O-20250208-ALGO-MODIFY-002");
4461        let account_id = AccountId::from("ACC-001");
4462        order
4463            .apply(TestOrderEventStubs::submitted(&order, account_id))
4464            .unwrap();
4465        order
4466            .apply(TestOrderEventStubs::accepted(
4467                &order,
4468                account_id,
4469                VenueOrderId::from("O-20250208-ALGO-MODIFY-002"),
4470            ))
4471            .unwrap();
4472        add_order_to_cache(&strategy, &order);
4473
4474        strategy
4475            .modify_order(
4476                order.client_order_id(),
4477                Some(Quantity::from(200_000)),
4478                None,
4479                None,
4480                None,
4481                None,
4482            )
4483            .unwrap();
4484
4485        let risk_messages = risk_messages.get_messages();
4486        assert_eq!(risk_messages.len(), 1);
4487        assert!(matches!(
4488            risk_messages.first(),
4489            Some(TradingCommand::ModifyOrder(command))
4490                if command.client_order_id == order.client_order_id()
4491        ));
4492        assert!(algo_messages.get_messages().is_empty());
4493        assert_eq!(
4494            strategy
4495                .cache()
4496                .order(&order.client_order_id())
4497                .unwrap()
4498                .status(),
4499            OrderStatus::PendingUpdate
4500        );
4501    }
4502
4503    #[rstest]
4504    fn test_modify_order_routes_emulated_order_to_order_emulator() {
4505        let mut strategy = create_test_strategy();
4506        register_strategy(&mut strategy);
4507
4508        let (emulator_handler, emulator_messages): (
4509            _,
4510            TypedIntoMessageSavingHandler<TradingCommand>,
4511        ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4512        msgbus::register_trading_command_endpoint(
4513            MessagingSwitchboard::order_emulator_execute(),
4514            emulator_handler,
4515        );
4516        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4517            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4518        msgbus::register_trading_command_endpoint(
4519            MessagingSwitchboard::risk_engine_queue_execute(),
4520            risk_handler,
4521        );
4522        let mut order = OrderTestBuilder::new(OrderType::StopMarket)
4523            .trader_id(TraderId::from("TRADER-001"))
4524            .strategy_id(StrategyId::from("TEST-001"))
4525            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4526            .client_order_id(ClientOrderId::from("O-20250208-EMULATED-MODIFY-001"))
4527            .side(OrderSide::Buy)
4528            .trigger_price(Price::from("51000.0"))
4529            .quantity(Quantity::from(100_000))
4530            .emulation_trigger(TriggerType::BidAsk)
4531            .build();
4532        order
4533            .apply(OrderEventAny::Emulated(
4534                OrderEmulatedSpec::builder()
4535                    .trader_id(order.trader_id())
4536                    .strategy_id(order.strategy_id())
4537                    .instrument_id(order.instrument_id())
4538                    .client_order_id(order.client_order_id())
4539                    .build(),
4540            ))
4541            .unwrap();
4542        add_order_to_cache(&strategy, &order);
4543
4544        strategy
4545            .modify_order(
4546                order.client_order_id(),
4547                None,
4548                None,
4549                Some(Price::from("52000.0")),
4550                None,
4551                None,
4552            )
4553            .unwrap();
4554
4555        let emulator_messages = emulator_messages.get_messages();
4556        assert_eq!(emulator_messages.len(), 1);
4557        assert!(matches!(
4558            emulator_messages.first(),
4559            Some(TradingCommand::ModifyOrder(command))
4560                if command.client_order_id == order.client_order_id()
4561        ));
4562        assert!(risk_messages.get_messages().is_empty());
4563    }
4564
4565    #[rstest]
4566    fn test_modify_order_routes_initialized_trigger_order_to_emulator_reentrantly() {
4567        let strategy_id = StrategyId::from("REENTRANT-001");
4568        let modified_quantity = Quantity::from(200_000);
4569        let mut strategy = InitializedModifyStrategy {
4570            core: StrategyCore::new(StrategyConfig {
4571                strategy_id: Some(strategy_id),
4572                ..Default::default()
4573            }),
4574            modified_quantity,
4575        };
4576        let trader_id = TraderId::from("TRADER-001");
4577        let clock = Rc::new(RefCell::new(VirtualClock::new()));
4578        let cache = Rc::new(RefCell::new(Cache::default()));
4579        let portfolio = Rc::new(RefCell::new(Portfolio::new(
4580            clock.clone(),
4581            cache.clone(),
4582            None,
4583        )));
4584        strategy
4585            .core
4586            .register(trader_id, clock, cache, portfolio)
4587            .unwrap();
4588        strategy.initialize().unwrap();
4589        strategy.start().unwrap();
4590
4591        let actor_id = strategy.actor_id().inner();
4592        register_component_actor(strategy);
4593        let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
4594            let mut strategy =
4595                try_get_actor_unchecked::<InitializedModifyStrategy>(&actor_id).unwrap();
4596            strategy.handle_order_event(event.clone());
4597        });
4598        let topic = format!("events.order.{strategy_id}");
4599        msgbus::subscribe_order_events(topic.clone().into(), order_handler.clone(), None);
4600
4601        let (emulator_handler, emulator_messages): (
4602            _,
4603            TypedIntoMessageSavingHandler<TradingCommand>,
4604        ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4605        msgbus::register_trading_command_endpoint(
4606            MessagingSwitchboard::order_emulator_execute(),
4607            emulator_handler,
4608        );
4609        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4610            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4611        msgbus::register_trading_command_endpoint(
4612            MessagingSwitchboard::risk_engine_queue_execute(),
4613            risk_handler,
4614        );
4615        let (algo_handler, algo_messages) =
4616            get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4617        msgbus::register_any("TWAP.execute".into(), algo_handler);
4618
4619        let order = OrderTestBuilder::new(OrderType::StopMarket)
4620            .trader_id(trader_id)
4621            .strategy_id(strategy_id)
4622            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4623            .client_order_id(ClientOrderId::from("O-REENTRANT-001"))
4624            .side(OrderSide::Buy)
4625            .trigger_price(Price::from("51000.0"))
4626            .quantity(Quantity::from(100_000))
4627            .emulation_trigger(TriggerType::BidAsk)
4628            .build();
4629        let client_order_id = order.client_order_id();
4630
4631        let mut strategy = try_get_actor_unchecked::<InitializedModifyStrategy>(&actor_id).unwrap();
4632        strategy.submit_order(order, None, None, None).unwrap();
4633        drop(strategy);
4634
4635        msgbus::unsubscribe_order_events(topic.into(), &order_handler);
4636        deregister_component(&strategy_id.inner());
4637        deregister_actor(&actor_id);
4638
4639        let emulator_messages = emulator_messages.get_messages();
4640        assert_eq!(emulator_messages.len(), 2);
4641        assert!(matches!(
4642            emulator_messages.first(),
4643            Some(TradingCommand::ModifyOrder(command))
4644                if command.client_order_id == client_order_id
4645                    && command.quantity == Some(modified_quantity)
4646        ));
4647        assert!(
4648            risk_messages
4649                .get_messages()
4650                .iter()
4651                .all(|command| !matches!(command, TradingCommand::ModifyOrder(_)))
4652        );
4653        assert!(
4654            algo_messages
4655                .get_messages()
4656                .iter()
4657                .all(|command| !matches!(command, TradingCommand::ModifyOrder(_)))
4658        );
4659        assert!(matches!(
4660            emulator_messages.get(1),
4661            Some(TradingCommand::SubmitOrder(command))
4662                if command.client_order_id == client_order_id
4663        ));
4664    }
4665
4666    #[rstest]
4667    fn test_modify_order_prefers_emulator_over_algorithm_for_emulated_algorithm_order() {
4668        let mut strategy = create_test_strategy();
4669        register_strategy(&mut strategy);
4670
4671        let (emulator_handler, emulator_messages): (
4672            _,
4673            TypedIntoMessageSavingHandler<TradingCommand>,
4674        ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4675        msgbus::register_trading_command_endpoint(
4676            MessagingSwitchboard::order_emulator_execute(),
4677            emulator_handler,
4678        );
4679        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4680            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4681        msgbus::register_trading_command_endpoint(
4682            MessagingSwitchboard::risk_engine_queue_execute(),
4683            risk_handler,
4684        );
4685        let (algo_handler, algo_messages) =
4686            get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4687        msgbus::register_any("TWAP.execute".into(), algo_handler);
4688
4689        let mut order = OrderTestBuilder::new(OrderType::StopMarket)
4690            .trader_id(TraderId::from("TRADER-001"))
4691            .strategy_id(StrategyId::from("TEST-001"))
4692            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4693            .client_order_id(ClientOrderId::from("O-20250208-EMULATED-ALGO-MODIFY-001"))
4694            .side(OrderSide::Buy)
4695            .trigger_price(Price::from("51000.0"))
4696            .quantity(Quantity::from(100_000))
4697            .emulation_trigger(TriggerType::BidAsk)
4698            .exec_algorithm_id(ExecAlgorithmId::from("TWAP"))
4699            .exec_spawn_id(ClientOrderId::from("O-20250208-EMULATED-ALGO-MODIFY-001"))
4700            .build();
4701        order
4702            .apply(OrderEventAny::Emulated(
4703                OrderEmulatedSpec::builder()
4704                    .trader_id(order.trader_id())
4705                    .strategy_id(order.strategy_id())
4706                    .instrument_id(order.instrument_id())
4707                    .client_order_id(order.client_order_id())
4708                    .build(),
4709            ))
4710            .unwrap();
4711
4712        // An `Emulated` order satisfies both `is_emulated` and `is_active_local`, so this
4713        // order matches the emulator branch and the algorithm branch at once.
4714        assert!(order.is_emulated());
4715        assert!(order.is_active_local());
4716        assert!(order.exec_algorithm_id().is_some());
4717
4718        add_order_to_cache(&strategy, &order);
4719
4720        strategy
4721            .modify_order(
4722                order.client_order_id(),
4723                None,
4724                None,
4725                Some(Price::from("52000.0")),
4726                None,
4727                None,
4728            )
4729            .unwrap();
4730
4731        let emulator_messages = emulator_messages.get_messages();
4732        assert_eq!(emulator_messages.len(), 1);
4733        assert!(matches!(
4734            emulator_messages.first(),
4735            Some(TradingCommand::ModifyOrder(command))
4736                if command.client_order_id == order.client_order_id()
4737        ));
4738        assert!(algo_messages.get_messages().is_empty());
4739        assert!(risk_messages.get_messages().is_empty());
4740    }
4741
4742    #[rstest]
4743    fn test_modify_order_marks_order_pending_update_locally_before_send() {
4744        let mut strategy = create_test_strategy();
4745        register_strategy(&mut strategy);
4746
4747        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4748            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4749        msgbus::register_trading_command_endpoint(
4750            MessagingSwitchboard::risk_engine_queue_execute(),
4751            risk_handler,
4752        );
4753
4754        let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
4755            get_typed_message_saving_handler(Some(Ustr::from("events.order.pending_update")));
4756        let order = make_accepted_limit_order("O-20250208-UPDATE-001");
4757        let topic = format!("events.order.{}", order.strategy_id());
4758        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4759        add_order_to_cache(&strategy, &order);
4760
4761        strategy
4762            .modify_order(
4763                order.client_order_id(),
4764                None,
4765                Some(Price::from("51000.0")),
4766                None,
4767                None,
4768                None,
4769            )
4770            .unwrap();
4771
4772        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4773
4774        let cache = strategy.cache();
4775        let cached_order = cache.order(&order.client_order_id()).unwrap();
4776        assert_eq!(cached_order.status(), OrderStatus::PendingUpdate);
4777
4778        let risk_messages = risk_messages.get_messages();
4779        assert_eq!(risk_messages.len(), 1);
4780        assert!(matches!(
4781            risk_messages.first(),
4782            Some(TradingCommand::ModifyOrder(_))
4783        ));
4784
4785        let event_messages = event_messages.get_messages();
4786        assert_eq!(event_messages.len(), 1);
4787        assert!(matches!(
4788            event_messages.first(),
4789            Some(OrderEventAny::PendingUpdate(_))
4790        ));
4791    }
4792
4793    #[rstest]
4794    fn test_late_modify_rejection_reaches_running_strategy_before_acceptance() {
4795        let mut strategy = ModifyRejectionStrategy {
4796            core: StrategyCore::new(StrategyConfig {
4797                strategy_id: Some(StrategyId::from("TEST-001")),
4798                ..Default::default()
4799            }),
4800            rejections: Vec::new(),
4801        };
4802
4803        let clock = Rc::new(RefCell::new(VirtualClock::new()));
4804        let cache = Rc::new(RefCell::new(Cache::default()));
4805
4806        let portfolio = Rc::new(RefCell::new(Portfolio::new(
4807            clock.clone(),
4808            cache.clone(),
4809            None,
4810        )));
4811        strategy
4812            .core
4813            .register(
4814                TraderId::from("TRADER-001"),
4815                clock.clone(),
4816                cache.clone(),
4817                portfolio,
4818            )
4819            .unwrap();
4820        strategy.initialize().unwrap();
4821        strategy.start().unwrap();
4822        let mut engine = ExecutionEngine::new(clock, cache.clone(), None);
4823        let order = OrderTestBuilder::new(OrderType::Limit)
4824            .trader_id(TraderId::from("TRADER-001"))
4825            .strategy_id(StrategyId::from("TEST-001"))
4826            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4827            .client_order_id(ClientOrderId::from("O-LATE-MODIFY-001"))
4828            .side(OrderSide::Buy)
4829            .quantity(Quantity::from(10))
4830            .price(Price::from("50000.00"))
4831            .build();
4832        cache
4833            .borrow_mut()
4834            .add_order(order.clone(), None, None, false)
4835            .unwrap();
4836        let account_id = AccountId::from("ACC-001");
4837        engine.process(&TestOrderEventStubs::submitted(&order, account_id));
4838        assert_eq!(
4839            cache
4840                .borrow()
4841                .order(&order.client_order_id())
4842                .unwrap()
4843                .status(),
4844            OrderStatus::Submitted
4845        );
4846
4847        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4848            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4849        msgbus::register_trading_command_endpoint(
4850            MessagingSwitchboard::risk_engine_queue_execute(),
4851            risk_handler,
4852        );
4853
4854        for (quantity, price) in [(30, "52000.00"), (20, "51000.00")] {
4855            strategy
4856                .modify_order(
4857                    order.client_order_id(),
4858                    Some(Quantity::from(quantity)),
4859                    Some(Price::from(price)),
4860                    None,
4861                    None,
4862                    None,
4863                )
4864                .unwrap();
4865        }
4866
4867        let commands = risk_messages.get_messages();
4868        assert_eq!(commands.len(), 2);
4869
4870        for (command, quantity, price) in [
4871            (&commands[0], 30, "52000.00"),
4872            (&commands[1], 20, "51000.00"),
4873        ] {
4874            let TradingCommand::ModifyOrder(command) = command else {
4875                panic!("Expected ModifyOrder command");
4876            };
4877
4878            assert_eq!(command.client_order_id, order.client_order_id());
4879            assert_eq!(command.venue_order_id, None);
4880            assert_eq!(command.quantity, Some(Quantity::from(quantity)));
4881            assert_eq!(command.price, Some(Price::from(price)));
4882        }
4883
4884        assert_eq!(
4885            cache
4886                .borrow()
4887                .order(&order.client_order_id())
4888                .unwrap()
4889                .status(),
4890            OrderStatus::PendingUpdate
4891        );
4892        let strategy = Rc::new(RefCell::new(strategy));
4893
4894        let handler = TypedHandler::from({
4895            let strategy = strategy.clone();
4896            move |event: &OrderEventAny| strategy.borrow_mut().handle_order_event(event.clone())
4897        });
4898
4899        let topic = msgbus::switchboard::get_event_order_topic(order.strategy_id());
4900        msgbus::subscribe_order_events(topic.into(), handler.clone(), None);
4901        engine.process(&OrderEventAny::Updated(
4902            OrderUpdatedSpec::builder()
4903                .trader_id(order.trader_id())
4904                .strategy_id(order.strategy_id())
4905                .instrument_id(order.instrument_id())
4906                .client_order_id(order.client_order_id())
4907                .venue_order_id(VenueOrderId::from("V-AMENDED-001"))
4908                .account_id(account_id)
4909                .quantity(Quantity::from(20))
4910                .price(Price::from("51000.00"))
4911                .build(),
4912        ));
4913        assert_eq!(
4914            cache
4915                .borrow()
4916                .order(&order.client_order_id())
4917                .unwrap()
4918                .status(),
4919            OrderStatus::Submitted
4920        );
4921        let rejection = OrderModifyRejectedSpec::builder()
4922            .trader_id(order.trader_id())
4923            .strategy_id(order.strategy_id())
4924            .instrument_id(order.instrument_id())
4925            .client_order_id(order.client_order_id())
4926            .account_id(account_id)
4927            .reason(Ustr::from("First amend rejected"))
4928            .event_id(UUID4::new())
4929            .ts_event(UnixNanos::from(123))
4930            .ts_init(UnixNanos::from(456))
4931            .build();
4932        engine.process(&OrderEventAny::ModifyRejected(rejection));
4933        msgbus::unsubscribe_order_events(topic.into(), &handler);
4934
4935        let strategy = strategy.borrow();
4936        assert_eq!(strategy.state(), ComponentState::Running);
4937        assert_eq!(strategy.rejections.len(), 1);
4938        let (received, cached) = &strategy.rejections[0];
4939        assert_eq!(*received, rejection);
4940        assert_eq!(cached.status(), OrderStatus::Submitted);
4941        assert_eq!(cached.previous_status(), Some(OrderStatus::Submitted));
4942        assert_eq!(cached.quantity(), Quantity::from(20));
4943        assert_eq!(cached.price(), Some(Price::from("51000.00")));
4944        assert_eq!(cached.filled_qty(), Quantity::from(0));
4945        assert_eq!(cached.leaves_qty(), Quantity::from(20));
4946        assert_eq!(
4947            cached.venue_order_id(),
4948            Some(VenueOrderId::from("V-AMENDED-001"))
4949        );
4950        assert_eq!(cached.events().len(), 5);
4951        assert_eq!(
4952            cached.last_event(),
4953            &OrderEventAny::ModifyRejected(rejection)
4954        );
4955    }
4956
4957    #[rstest]
4958    fn test_modify_orders_marks_orders_pending_update_locally_before_send() {
4959        let mut strategy = create_test_strategy();
4960        register_strategy(&mut strategy);
4961
4962        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4963            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4964        msgbus::register_trading_command_endpoint(
4965            MessagingSwitchboard::risk_engine_queue_execute(),
4966            risk_handler,
4967        );
4968
4969        let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
4970            get_typed_message_saving_handler(Some(Ustr::from("events.order.batch_pending_update")));
4971        let order1 = make_accepted_limit_order("O-20250208-BATCH-UPDATE-001");
4972        let order2 = make_accepted_limit_order("O-20250208-BATCH-UPDATE-002");
4973        let topic = format!("events.order.{}", order1.strategy_id());
4974        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4975        add_order_to_cache(&strategy, &order1);
4976        add_order_to_cache(&strategy, &order2);
4977
4978        strategy
4979            .modify_orders(
4980                vec![
4981                    (
4982                        order1.client_order_id(),
4983                        None,
4984                        Some(Price::from("51000.0")),
4985                        None,
4986                    ),
4987                    (
4988                        order2.client_order_id(),
4989                        Some(Quantity::from("2.0")),
4990                        None,
4991                        None,
4992                    ),
4993                ],
4994                None,
4995                None,
4996            )
4997            .unwrap();
4998
4999        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
5000
5001        let cache = strategy.cache();
5002        let cached_order1 = cache.order(&order1.client_order_id()).unwrap();
5003        let cached_order2 = cache.order(&order2.client_order_id()).unwrap();
5004        assert_eq!(cached_order1.status(), OrderStatus::PendingUpdate);
5005        assert_eq!(cached_order2.status(), OrderStatus::PendingUpdate);
5006
5007        let risk_messages = risk_messages.get_messages();
5008        assert_eq!(risk_messages.len(), 1);
5009        let Some(TradingCommand::ModifyOrders(command)) = risk_messages.first() else {
5010            panic!("expected BatchModifyOrders command");
5011        };
5012        assert_eq!(command.modifies.len(), 2);
5013        assert_eq!(
5014            command
5015                .modifies
5016                .iter()
5017                .map(|modify| modify.client_order_id)
5018                .collect::<Vec<_>>(),
5019            vec![order1.client_order_id(), order2.client_order_id()]
5020        );
5021
5022        let event_messages = event_messages.get_messages();
5023        assert_eq!(event_messages.len(), 2);
5024        assert!(
5025            event_messages
5026                .iter()
5027                .all(|event| matches!(event, OrderEventAny::PendingUpdate(_)))
5028        );
5029    }
5030
5031    #[rstest]
5032    fn test_cancel_order_marks_order_pending_cancel_locally_before_send() {
5033        let mut strategy = create_test_strategy();
5034        register_strategy(&mut strategy);
5035
5036        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5037            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5038        msgbus::register_trading_command_endpoint(
5039            MessagingSwitchboard::exec_engine_queue_execute(),
5040            exec_handler,
5041        );
5042
5043        let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
5044            get_typed_message_saving_handler(Some(Ustr::from("events.order.pending_cancel")));
5045        let order = make_accepted_market_order("O-20250208-CANCEL-001");
5046        let topic = format!("events.order.{}", order.strategy_id());
5047        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
5048        add_order_to_cache(&strategy, &order);
5049
5050        strategy
5051            .cancel_order(order.client_order_id(), None, None)
5052            .unwrap();
5053
5054        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
5055
5056        let cache = strategy.cache();
5057        let cached_order = cache.order(&order.client_order_id()).unwrap();
5058        assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
5059        let cache = strategy.core.cache_ref();
5060        assert!(cache.is_order_pending_cancel_local(&order.client_order_id()));
5061
5062        let exec_messages = exec_messages.get_messages();
5063        assert_eq!(exec_messages.len(), 1);
5064        assert!(matches!(
5065            exec_messages.first(),
5066            Some(TradingCommand::CancelOrder(_))
5067        ));
5068
5069        let event_messages = event_messages.get_messages();
5070        assert_eq!(event_messages.len(), 1);
5071        assert!(matches!(
5072            event_messages.first(),
5073            Some(OrderEventAny::PendingCancel(_))
5074        ));
5075    }
5076
5077    #[rstest]
5078    fn test_cancel_all_orders_strategy_only_sends_only_caller_strategy_cancels() {
5079        let mut strategy = create_test_strategy();
5080        register_strategy(&mut strategy);
5081
5082        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5083            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5084        msgbus::register_trading_command_endpoint(
5085            MessagingSwitchboard::exec_engine_queue_execute(),
5086            exec_handler,
5087        );
5088
5089        let order = make_accepted_market_order("O-20250208-CANCEL-ALL-001");
5090        let mut sibling_order = OrderTestBuilder::new(OrderType::Market)
5091            .trader_id(TraderId::from("TRADER-001"))
5092            .strategy_id(StrategyId::from("SIBLING-001"))
5093            .instrument_id(order.instrument_id())
5094            .client_order_id(ClientOrderId::from("O-20250208-CANCEL-ALL-002"))
5095            .side(OrderSide::Buy)
5096            .quantity(Quantity::from(100_000))
5097            .build();
5098        let account_id = AccountId::from("ACC-001");
5099        sibling_order
5100            .apply(TestOrderEventStubs::submitted(&sibling_order, account_id))
5101            .unwrap();
5102        sibling_order
5103            .apply(TestOrderEventStubs::accepted(
5104                &sibling_order,
5105                account_id,
5106                VenueOrderId::from("2"),
5107            ))
5108            .unwrap();
5109        add_order_to_cache(&strategy, &order);
5110        add_order_to_cache(&strategy, &sibling_order);
5111        strategy.core.cache_rc().borrow_mut().build_index();
5112
5113        strategy
5114            .cancel_all_orders(order.instrument_id(), None, None, true, None)
5115            .unwrap();
5116
5117        let messages = exec_messages.get_messages();
5118        let cache = strategy.cache();
5119        let cached_order = cache.order(&order.client_order_id()).unwrap();
5120        let cached_sibling = cache.order(&sibling_order.client_order_id()).unwrap();
5121        assert_eq!(messages.len(), 1);
5122        assert!(matches!(
5123            messages.first(),
5124            Some(TradingCommand::CancelOrder(command))
5125                if command.client_order_id == order.client_order_id()
5126        ));
5127        assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
5128        assert_eq!(cached_sibling.status(), OrderStatus::Accepted);
5129    }
5130
5131    #[rstest]
5132    fn test_cancel_all_orders_strategy_only_deduplicates_emulated_inflight_order() {
5133        let mut strategy = create_test_strategy();
5134        register_strategy(&mut strategy);
5135
5136        let (emulator_handler, emulator_messages): (
5137            _,
5138            TypedIntoMessageSavingHandler<TradingCommand>,
5139        ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
5140        msgbus::register_trading_command_endpoint(
5141            MessagingSwitchboard::order_emulator_execute(),
5142            emulator_handler,
5143        );
5144        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5145            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5146        msgbus::register_trading_command_endpoint(
5147            MessagingSwitchboard::exec_engine_queue_execute(),
5148            exec_handler,
5149        );
5150
5151        let client_order_id = ClientOrderId::from("O-20250208-CANCEL-ALL-EMULATED-001");
5152        let order = OrderTestBuilder::new(OrderType::StopMarket)
5153            .trader_id(TraderId::from("TRADER-001"))
5154            .strategy_id(StrategyId::from("TEST-001"))
5155            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
5156            .client_order_id(client_order_id)
5157            .side(OrderSide::Buy)
5158            .trigger_price(Price::from("51000.0"))
5159            .quantity(Quantity::from(100_000))
5160            .emulation_trigger(TriggerType::BidAsk)
5161            .exec_algorithm_id(ExecAlgorithmId::from("ALGO-001"))
5162            .exec_spawn_id(client_order_id)
5163            .build();
5164        add_order_to_cache(&strategy, &order);
5165        strategy.core.cache_rc().borrow_mut().build_index();
5166
5167        strategy
5168            .cancel_all_orders(order.instrument_id(), None, None, true, None)
5169            .unwrap();
5170
5171        let emulator_messages = emulator_messages.get_messages();
5172        assert_eq!(emulator_messages.len(), 1);
5173        assert!(matches!(
5174            emulator_messages.first(),
5175            Some(TradingCommand::CancelOrder(command))
5176                if command.client_order_id == order.client_order_id()
5177        ));
5178        assert!(exec_messages.get_messages().is_empty());
5179    }
5180
5181    #[rstest]
5182    fn test_cancel_all_orders_without_strategy_only_sends_cancel_all_command() {
5183        let mut strategy = create_test_strategy();
5184        register_strategy(&mut strategy);
5185
5186        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5187            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5188        msgbus::register_trading_command_endpoint(
5189            MessagingSwitchboard::exec_engine_queue_execute(),
5190            exec_handler,
5191        );
5192
5193        let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
5194
5195        strategy
5196            .cancel_all_orders(instrument_id, None, None, false, None)
5197            .unwrap();
5198
5199        let messages = exec_messages.get_messages();
5200        assert_eq!(messages.len(), 1);
5201        let Some(TradingCommand::CancelAllOrders(command)) = messages.first() else {
5202            panic!("Expected a CancelAllOrders command, was {messages:?}");
5203        };
5204        assert_eq!(command.strategy_id, StrategyId::from("TEST-001"));
5205        assert_eq!(command.instrument_id, instrument_id);
5206        assert_eq!(command.client_id, None);
5207        assert_eq!(command.order_side, None);
5208        assert_eq!(command.correlation_id, Some(command.command_id));
5209        assert_eq!(command.causation_id, None);
5210    }
5211
5212    #[rstest]
5213    fn test_cancel_all_orders_without_strategy_only_delegates_local_routing_with_params() {
5214        let mut strategy = create_test_strategy();
5215        register_strategy(&mut strategy);
5216
5217        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5218            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5219        msgbus::register_trading_command_endpoint(
5220            MessagingSwitchboard::exec_engine_queue_execute(),
5221            exec_handler,
5222        );
5223        let (emulator_handler, emulator_messages): (
5224            _,
5225            TypedIntoMessageSavingHandler<TradingCommand>,
5226        ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
5227        msgbus::register_trading_command_endpoint(
5228            MessagingSwitchboard::order_emulator_execute(),
5229            emulator_handler,
5230        );
5231        let exec_algorithm_id = ExecAlgorithmId::from("ALGO-001");
5232        let algorithm_endpoint = format!("{exec_algorithm_id}.execute");
5233        let (algorithm_handler, algorithm_messages) =
5234            get_any_saving_handler::<TradingCommand>(Some(Ustr::from("ALGO-001.execute")));
5235        msgbus::register_any(algorithm_endpoint.into(), algorithm_handler);
5236
5237        let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
5238        let sibling_strategy_id = StrategyId::from("SIBLING-001");
5239        let selected_client = ClientId::from("CLIENT-001");
5240        let account_id = AccountId::from("ACC-001");
5241
5242        let mut open_order = OrderTestBuilder::new(OrderType::Limit)
5243            .strategy_id(sibling_strategy_id)
5244            .instrument_id(instrument_id)
5245            .client_order_id(ClientOrderId::from("O-BROAD-OPEN-001"))
5246            .side(OrderSide::Buy)
5247            .price(Price::from("49000.0"))
5248            .quantity(Quantity::from(100_000))
5249            .build();
5250        open_order
5251            .apply(TestOrderEventStubs::submitted(&open_order, account_id))
5252            .unwrap();
5253        open_order
5254            .apply(TestOrderEventStubs::accepted(
5255                &open_order,
5256                account_id,
5257                VenueOrderId::from("V-BROAD-OPEN-001"),
5258            ))
5259            .unwrap();
5260        let emulated_order = OrderTestBuilder::new(OrderType::StopMarket)
5261            .strategy_id(sibling_strategy_id)
5262            .instrument_id(instrument_id)
5263            .client_order_id(ClientOrderId::from("O-BROAD-EMULATED-001"))
5264            .side(OrderSide::Buy)
5265            .trigger_price(Price::from("51000.0"))
5266            .quantity(Quantity::from(100_000))
5267            .emulation_trigger(TriggerType::BidAsk)
5268            .build();
5269        let algorithm_order_id = ClientOrderId::from("O-BROAD-ALGO-001");
5270        let algorithm_order = OrderTestBuilder::new(OrderType::Market)
5271            .strategy_id(sibling_strategy_id)
5272            .instrument_id(instrument_id)
5273            .client_order_id(algorithm_order_id)
5274            .side(OrderSide::Buy)
5275            .quantity(Quantity::from(100_000))
5276            .exec_algorithm_id(exec_algorithm_id)
5277            .exec_spawn_id(algorithm_order_id)
5278            .build();
5279
5280        {
5281            let cache_rc = strategy.core.cache_rc();
5282            let mut cache = cache_rc.borrow_mut();
5283            for order in [&open_order, &emulated_order, &algorithm_order] {
5284                cache
5285                    .add_order(order.clone(), None, Some(selected_client), true)
5286                    .unwrap();
5287            }
5288            cache.build_index();
5289        }
5290
5291        let mut params = Params::new();
5292        params.insert(
5293            "routing_hint".to_string(),
5294            Value::String("broad_cancel".to_string()),
5295        );
5296        strategy
5297            .cancel_all_orders(
5298                instrument_id,
5299                Some(OrderSide::Buy),
5300                Some(selected_client),
5301                false,
5302                Some(params.clone()),
5303            )
5304            .unwrap();
5305
5306        let exec_messages = exec_messages.get_messages();
5307        let Some(TradingCommand::CancelAllOrders(exec_command)) = exec_messages.first() else {
5308            panic!("Expected an execution CancelAllOrders command, was {exec_messages:?}");
5309        };
5310
5311        assert_eq!(exec_messages.len(), 1);
5312        assert!(emulator_messages.get_messages().is_empty());
5313        assert!(algorithm_messages.get_messages().is_empty());
5314        assert_eq!(exec_command.client_id, Some(selected_client));
5315        assert_eq!(exec_command.strategy_id, StrategyId::from("TEST-001"));
5316        assert_eq!(exec_command.instrument_id, instrument_id);
5317        assert_eq!(exec_command.order_side, Some(OrderSide::Buy));
5318        assert_eq!(exec_command.params.as_ref(), Some(&params));
5319        assert_eq!(exec_command.correlation_id, Some(exec_command.command_id));
5320        assert_eq!(exec_command.causation_id, None);
5321    }
5322
5323    #[rstest]
5324    fn test_cancel_all_orders_strategy_only_filters_by_client() {
5325        let mut strategy = create_test_strategy();
5326        register_strategy(&mut strategy);
5327
5328        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5329            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5330        msgbus::register_trading_command_endpoint(
5331            MessagingSwitchboard::exec_engine_queue_execute(),
5332            exec_handler,
5333        );
5334
5335        let selected_client = ClientId::from("CLIENT-001");
5336        let other_client = ClientId::from("CLIENT-002");
5337        let selected_order = make_accepted_market_order("O-20250208-CANCEL-CLIENT-001");
5338        let other_order = make_accepted_market_order("O-20250208-CANCEL-CLIENT-002");
5339        let cache_rc = strategy.core.cache_rc();
5340        {
5341            let mut cache = cache_rc.borrow_mut();
5342            cache
5343                .add_order(selected_order.clone(), None, Some(selected_client), true)
5344                .unwrap();
5345            cache
5346                .add_order(other_order.clone(), None, Some(other_client), true)
5347                .unwrap();
5348            cache.build_index();
5349        }
5350
5351        strategy
5352            .cancel_all_orders(
5353                selected_order.instrument_id(),
5354                None,
5355                Some(selected_client),
5356                true,
5357                None,
5358            )
5359            .unwrap();
5360
5361        let messages = exec_messages.get_messages();
5362        let cache = strategy.cache();
5363        let cached_selected = cache.order(&selected_order.client_order_id()).unwrap();
5364        let cached_other = cache.order(&other_order.client_order_id()).unwrap();
5365        assert_eq!(messages.len(), 1);
5366        assert!(matches!(
5367            messages.first(),
5368            Some(TradingCommand::CancelOrder(command))
5369                if command.client_id == Some(selected_client)
5370                    && command.client_order_id == selected_order.client_order_id()
5371        ));
5372        assert_eq!(cached_selected.status(), OrderStatus::PendingCancel);
5373        assert_eq!(cached_other.status(), OrderStatus::Accepted);
5374    }
5375
5376    #[rstest]
5377    fn test_cancel_all_orders_strategy_only_filters_by_side_and_preserves_params() {
5378        let mut strategy = create_test_strategy();
5379        register_strategy(&mut strategy);
5380
5381        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5382            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5383        msgbus::register_trading_command_endpoint(
5384            MessagingSwitchboard::exec_engine_queue_execute(),
5385            exec_handler,
5386        );
5387
5388        let buy_order = make_accepted_market_order("O-20250208-CANCEL-SIDE-001");
5389        let mut sell_order = OrderTestBuilder::new(OrderType::Market)
5390            .trader_id(TraderId::from("TRADER-001"))
5391            .strategy_id(StrategyId::from("TEST-001"))
5392            .instrument_id(buy_order.instrument_id())
5393            .client_order_id(ClientOrderId::from("O-20250208-CANCEL-SIDE-002"))
5394            .side(OrderSide::Sell)
5395            .quantity(Quantity::from(100_000))
5396            .build();
5397        let account_id = AccountId::from("ACC-001");
5398        sell_order
5399            .apply(TestOrderEventStubs::submitted(&sell_order, account_id))
5400            .unwrap();
5401        sell_order
5402            .apply(TestOrderEventStubs::accepted(
5403                &sell_order,
5404                account_id,
5405                VenueOrderId::from("O-20250208-CANCEL-SIDE-002"),
5406            ))
5407            .unwrap();
5408        add_order_to_cache(&strategy, &buy_order);
5409        add_order_to_cache(&strategy, &sell_order);
5410        strategy.core.cache_rc().borrow_mut().build_index();
5411
5412        let mut params = Params::new();
5413        params.insert(
5414            "routing_hint".to_string(),
5415            Value::String("strategy_only".to_string()),
5416        );
5417        strategy
5418            .cancel_all_orders(
5419                buy_order.instrument_id(),
5420                Some(OrderSide::Buy),
5421                None,
5422                true,
5423                Some(params.clone()),
5424            )
5425            .unwrap();
5426
5427        let messages = exec_messages.get_messages();
5428        let Some(TradingCommand::CancelOrder(command)) = messages.first() else {
5429            panic!("Expected a CancelOrder command, was {messages:?}");
5430        };
5431        let cache = strategy.cache();
5432        let cached_buy = cache.order(&buy_order.client_order_id()).unwrap();
5433        let cached_sell = cache.order(&sell_order.client_order_id()).unwrap();
5434        assert_eq!(messages.len(), 1);
5435        assert_eq!(command.trader_id, TraderId::from("TRADER-001"));
5436        assert_eq!(command.client_id, None);
5437        assert_eq!(command.strategy_id, StrategyId::from("TEST-001"));
5438        assert_eq!(command.instrument_id, buy_order.instrument_id());
5439        assert_eq!(command.client_order_id, buy_order.client_order_id());
5440        assert_eq!(command.venue_order_id, buy_order.venue_order_id());
5441        assert_eq!(command.params.as_ref(), Some(&params));
5442        assert_eq!(cached_buy.status(), OrderStatus::PendingCancel);
5443        assert_eq!(cached_sell.status(), OrderStatus::Accepted);
5444    }
5445
5446    #[rstest]
5447    fn test_cancel_all_orders_strategy_only_continues_after_error_and_returns_error() {
5448        let mut strategy = create_test_strategy();
5449        register_strategy(&mut strategy);
5450
5451        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5452            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5453        msgbus::register_trading_command_endpoint(
5454            MessagingSwitchboard::exec_engine_queue_execute(),
5455            exec_handler,
5456        );
5457
5458        let mut failing_order = make_accepted_market_order("O-20250208-CANCEL-ERROR-001");
5459        let OrderAny::Market(order) = &mut failing_order else {
5460            panic!("Expected a MarketOrder");
5461        };
5462        order.account_id = None;
5463        let succeeding_order = make_accepted_market_order("O-20250208-CANCEL-ERROR-002");
5464        add_order_to_cache(&strategy, &failing_order);
5465        add_order_to_cache(&strategy, &succeeding_order);
5466        strategy.core.cache_rc().borrow_mut().build_index();
5467
5468        let error = strategy
5469            .cancel_all_orders(failing_order.instrument_id(), None, None, true, None)
5470            .unwrap_err()
5471            .to_string();
5472
5473        let messages = exec_messages.get_messages();
5474        let cache = strategy.cache();
5475        let cached_failing = cache.order(&failing_order.client_order_id()).unwrap();
5476        let cached_succeeding = cache.order(&succeeding_order.client_order_id()).unwrap();
5477        assert_eq!(
5478            error,
5479            "Cannot generate pending cancel event for O-20250208-CANCEL-ERROR-001: \
5480             account_id is not set"
5481        );
5482        assert_eq!(messages.len(), 1);
5483        assert!(matches!(
5484            messages.first(),
5485            Some(TradingCommand::CancelOrder(command))
5486                if command.client_order_id == succeeding_order.client_order_id()
5487        ));
5488        assert_eq!(cached_failing.status(), OrderStatus::Accepted);
5489        assert_eq!(cached_succeeding.status(), OrderStatus::PendingCancel);
5490    }
5491
5492    #[rstest]
5493    fn test_cancel_all_orders_strategy_only_cancels_order_submitted_in_same_handler() {
5494        let mut strategy = create_test_strategy();
5495        register_strategy(&mut strategy);
5496
5497        let (risk_handler, _risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5498            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5499        msgbus::register_trading_command_endpoint(
5500            MessagingSwitchboard::risk_engine_queue_execute(),
5501            risk_handler,
5502        );
5503        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5504            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5505        msgbus::register_trading_command_endpoint(
5506            MessagingSwitchboard::exec_engine_queue_execute(),
5507            exec_handler,
5508        );
5509
5510        let order = OrderTestBuilder::new(OrderType::Limit)
5511            .trader_id(TraderId::from("TRADER-001"))
5512            .strategy_id(StrategyId::from("TEST-001"))
5513            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
5514            .client_order_id(ClientOrderId::from("O-20250208-CANCEL-ALL-LOCAL-001"))
5515            .side(OrderSide::Buy)
5516            .price(Price::from("50000.0"))
5517            .quantity(Quantity::from(100_000))
5518            .build();
5519        let sibling_order = OrderTestBuilder::new(OrderType::Limit)
5520            .trader_id(TraderId::from("TRADER-001"))
5521            .strategy_id(StrategyId::from("SIBLING-001"))
5522            .instrument_id(order.instrument_id())
5523            .client_order_id(ClientOrderId::from("O-20250208-CANCEL-ALL-LOCAL-002"))
5524            .side(OrderSide::Buy)
5525            .price(Price::from("50000.0"))
5526            .quantity(Quantity::from(100_000))
5527            .build();
5528        add_order_to_cache(&strategy, &sibling_order);
5529
5530        // The submit command is queued, so the order is still INITIALIZED when canceled
5531        strategy
5532            .submit_order(order.clone(), None, None, None)
5533            .unwrap();
5534        strategy
5535            .cancel_all_orders(order.instrument_id(), None, None, true, None)
5536            .unwrap();
5537
5538        let messages = exec_messages.get_messages();
5539        assert_eq!(messages.len(), 1);
5540        assert!(matches!(
5541            messages.first(),
5542            Some(TradingCommand::CancelOrder(command))
5543                if command.client_order_id == order.client_order_id()
5544        ));
5545    }
5546
5547    #[rstest]
5548    fn test_cancel_orders_marks_orders_pending_cancel_locally_before_send() {
5549        let mut strategy = create_test_strategy();
5550        register_strategy(&mut strategy);
5551
5552        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5553            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5554        msgbus::register_trading_command_endpoint(
5555            MessagingSwitchboard::exec_engine_queue_execute(),
5556            exec_handler,
5557        );
5558
5559        let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
5560            get_typed_message_saving_handler(Some(Ustr::from("events.order.batch_pending_cancel")));
5561        let order1 = make_accepted_market_order("O-20250208-CANCEL-001");
5562        let order2 = make_accepted_market_order("O-20250208-CANCEL-002");
5563        let topic = format!("events.order.{}", order1.strategy_id());
5564        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
5565        add_order_to_cache(&strategy, &order1);
5566        add_order_to_cache(&strategy, &order2);
5567
5568        strategy
5569            .cancel_orders(
5570                vec![order1.client_order_id(), order2.client_order_id()],
5571                None,
5572                None,
5573            )
5574            .unwrap();
5575
5576        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
5577
5578        let cache = strategy.cache();
5579        let cached_order1 = cache.order(&order1.client_order_id()).unwrap();
5580        let cached_order2 = cache.order(&order2.client_order_id()).unwrap();
5581        assert_eq!(cached_order1.status(), OrderStatus::PendingCancel);
5582        assert_eq!(cached_order2.status(), OrderStatus::PendingCancel);
5583        let cache = strategy.core.cache_ref();
5584        assert!(cache.is_order_pending_cancel_local(&order1.client_order_id()));
5585        assert!(cache.is_order_pending_cancel_local(&order2.client_order_id()));
5586
5587        let exec_messages = exec_messages.get_messages();
5588        assert_eq!(exec_messages.len(), 1);
5589        let Some(TradingCommand::CancelOrders(command)) = exec_messages.first() else {
5590            panic!("expected BatchCancelOrders command");
5591        };
5592        assert_eq!(command.cancels.len(), 2);
5593
5594        let event_messages = event_messages.get_messages();
5595        assert_eq!(event_messages.len(), 2);
5596        assert!(
5597            event_messages
5598                .iter()
5599                .all(|event| matches!(event, OrderEventAny::PendingCancel(_)))
5600        );
5601    }
5602
5603    #[rstest]
5604    fn test_cancel_order_updates_own_book_status_before_send() {
5605        let mut strategy = create_test_strategy();
5606        register_strategy(&mut strategy);
5607
5608        let (exec_handler, _exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5609            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5610        msgbus::register_trading_command_endpoint(
5611            MessagingSwitchboard::exec_engine_queue_execute(),
5612            exec_handler,
5613        );
5614
5615        let order = make_accepted_limit_order("O-20250208-CANCEL-OWN-BOOK-001");
5616        add_order_to_cache_and_own_book(&strategy, &order);
5617
5618        strategy
5619            .cancel_order(order.client_order_id(), None, None)
5620            .unwrap();
5621
5622        let mut accepted = AHashSet::new();
5623        accepted.insert(OrderStatus::Accepted);
5624        let mut pending_cancel = AHashSet::new();
5625        pending_cancel.insert(OrderStatus::PendingCancel);
5626
5627        let cache = strategy.cache();
5628        let own_book = cache.own_order_book(&order.instrument_id()).unwrap();
5629        assert!(own_book.bids_as_map(Some(&accepted), None, None).is_empty());
5630        let pending_bids = own_book.bids_as_map(Some(&pending_cancel), None, None);
5631        assert_eq!(pending_bids.values().map(Vec::len).sum::<usize>(), 1);
5632    }
5633
5634    #[rstest]
5635    fn test_cancel_order_returns_error_when_not_in_cache() {
5636        let mut strategy = create_test_strategy();
5637        register_strategy(&mut strategy);
5638
5639        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5640            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5641        msgbus::register_trading_command_endpoint(
5642            MessagingSwitchboard::exec_engine_queue_execute(),
5643            exec_handler,
5644        );
5645
5646        let missing_id = ClientOrderId::from("O-MISSING");
5647        let err = strategy
5648            .cancel_order(missing_id, None, None)
5649            .expect_err("expected cancel_order to fail when order is not in cache");
5650
5651        assert_eq!(
5652            err.to_string(),
5653            format!("Cannot cancel order: {ORDER_NOT_FOUND}: {missing_id}")
5654        );
5655        assert!(exec_messages.get_messages().is_empty());
5656    }
5657
5658    #[rstest]
5659    fn test_modify_order_returns_error_when_not_in_cache() {
5660        let mut strategy = create_test_strategy();
5661        register_strategy(&mut strategy);
5662
5663        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5664            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5665        msgbus::register_trading_command_endpoint(
5666            MessagingSwitchboard::risk_engine_queue_execute(),
5667            risk_handler,
5668        );
5669
5670        let missing_id = ClientOrderId::from("O-MISSING");
5671        let err = strategy
5672            .modify_order(missing_id, Some(Quantity::from(1)), None, None, None, None)
5673            .expect_err("expected modify_order to fail when order is not in cache");
5674
5675        assert_eq!(
5676            err.to_string(),
5677            format!("Cannot modify order: {ORDER_NOT_FOUND}: {missing_id}")
5678        );
5679        assert!(risk_messages.get_messages().is_empty());
5680    }
5681
5682    #[rstest]
5683    fn test_modify_orders_returns_error_when_any_id_missing() {
5684        let mut strategy = create_test_strategy();
5685        register_strategy(&mut strategy);
5686
5687        let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5688            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5689        msgbus::register_trading_command_endpoint(
5690            MessagingSwitchboard::risk_engine_queue_execute(),
5691            risk_handler,
5692        );
5693
5694        let order = make_accepted_limit_order("O-PRESENT");
5695        add_order_to_cache(&strategy, &order);
5696
5697        let missing_id = ClientOrderId::from("O-MISSING");
5698        let err = strategy
5699            .modify_orders(
5700                vec![
5701                    (
5702                        order.client_order_id(),
5703                        None,
5704                        Some(Price::from("51000.0")),
5705                        None,
5706                    ),
5707                    (missing_id, Some(Quantity::from("2.0")), None, None),
5708                ],
5709                None,
5710                None,
5711            )
5712            .expect_err("expected modify_orders to fail when any id is missing");
5713
5714        assert_eq!(
5715            err.to_string(),
5716            format!("Cannot modify order: {ORDER_NOT_FOUND}: {missing_id}")
5717        );
5718        assert!(risk_messages.get_messages().is_empty());
5719    }
5720
5721    #[rstest]
5722    fn test_cancel_orders_returns_error_when_any_id_missing() {
5723        let mut strategy = create_test_strategy();
5724        register_strategy(&mut strategy);
5725
5726        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5727            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5728        msgbus::register_trading_command_endpoint(
5729            MessagingSwitchboard::exec_engine_queue_execute(),
5730            exec_handler,
5731        );
5732
5733        let order = make_accepted_limit_order("O-PRESENT");
5734        add_order_to_cache(&strategy, &order);
5735
5736        let missing_id = ClientOrderId::from("O-MISSING");
5737        let err = strategy
5738            .cancel_orders(vec![order.client_order_id(), missing_id], None, None)
5739            .expect_err("expected cancel_orders to fail when any id is missing");
5740
5741        assert_eq!(
5742            err.to_string(),
5743            format!("Cannot cancel order: {ORDER_NOT_FOUND}: {missing_id}")
5744        );
5745        assert!(exec_messages.get_messages().is_empty());
5746    }
5747
5748    // -- GTD EXPIRY TESTS ----------------------------------------------------------------------------
5749
5750    #[rstest]
5751    fn test_has_gtd_expiry_timer_when_timer_not_set() {
5752        let mut strategy = create_test_strategy();
5753        let client_order_id = ClientOrderId::from("O-001");
5754
5755        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5756    }
5757
5758    #[rstest]
5759    fn test_has_gtd_expiry_timer_when_timer_set() {
5760        let mut strategy = create_test_strategy();
5761        let client_order_id = ClientOrderId::from("O-001");
5762
5763        strategy
5764            .core
5765            .gtd_timers
5766            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5767
5768        assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5769    }
5770
5771    #[rstest]
5772    fn test_cancel_gtd_expiry_removes_timer() {
5773        let mut strategy = create_test_strategy();
5774        register_strategy(&mut strategy);
5775
5776        let client_order_id = ClientOrderId::from("O-001");
5777        strategy
5778            .core
5779            .gtd_timers
5780            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5781
5782        strategy.cancel_gtd_expiry(&client_order_id);
5783
5784        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5785    }
5786
5787    #[rstest]
5788    fn test_cancel_gtd_expiry_when_timer_not_set() {
5789        let mut strategy = create_test_strategy();
5790        register_strategy(&mut strategy);
5791
5792        let client_order_id = ClientOrderId::from("O-001");
5793
5794        strategy.cancel_gtd_expiry(&client_order_id);
5795
5796        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5797    }
5798
5799    #[rstest]
5800    #[case::matching_order(Ustr::from("GTD-EXPIRY:O-PRESENT"))]
5801    #[case::empty_order_id(Ustr::from("GTD-EXPIRY:"))]
5802    fn test_route_time_event_ignores_unregistered_gtd_timer(#[case] timer_name: Ustr) {
5803        let mut strategy = create_test_strategy();
5804        register_strategy(&mut strategy);
5805
5806        let order = make_accepted_limit_order("O-PRESENT");
5807        let client_order_id = order.client_order_id();
5808        add_order_to_cache(&strategy, &order);
5809        let event = TimeEvent::new(
5810            timer_name,
5811            UUID4::new(),
5812            UnixNanos::default(),
5813            UnixNanos::default(),
5814        );
5815
5816        route_time_event(&mut strategy, &event);
5817
5818        let cache = strategy.core.cache_ref();
5819        let cached_order = cache.order(&client_order_id).unwrap();
5820        assert_eq!(cached_order.status(), OrderStatus::Accepted);
5821        assert!(!strategy.core.gtd_timers.contains_key(&client_order_id));
5822    }
5823
5824    #[rstest]
5825    #[case::filled(make_filled)]
5826    #[case::canceled(make_canceled)]
5827    #[case::rejected(make_rejected)]
5828    #[case::expired(make_expired)]
5829    #[case::fill_voided(make_terminal_fill_voided)]
5830    fn test_handle_order_event_cancels_gtd_timer_for_terminal_event(
5831        #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
5832    ) {
5833        let mut strategy = create_test_strategy();
5834        register_strategy(&mut strategy);
5835        start_strategy(&mut strategy);
5836
5837        let client_order_id = ClientOrderId::from("O-001");
5838        strategy
5839            .core
5840            .gtd_timers
5841            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5842
5843        strategy.handle_order_event(make_event(client_order_id));
5844
5845        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5846    }
5847
5848    #[rstest]
5849    #[case::partial_fill(make_filled)]
5850    #[case::non_reopened_fill_void(make_terminal_fill_voided)]
5851    fn test_handle_order_event_keeps_gtd_timer_when_cached_order_remains_open(
5852        #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
5853    ) {
5854        let mut strategy = create_test_strategy();
5855        register_strategy(&mut strategy);
5856        start_strategy(&mut strategy);
5857
5858        let client_order_id = ClientOrderId::from("O-001");
5859        let order = make_accepted_limit_order(client_order_id.as_str());
5860        add_order_to_cache(&strategy, &order);
5861        strategy
5862            .core
5863            .gtd_timers
5864            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5865
5866        strategy.handle_order_event(make_event(client_order_id));
5867
5868        assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5869    }
5870
5871    #[rstest]
5872    #[case::filled(make_filled)]
5873    #[case::canceled(make_canceled)]
5874    #[case::rejected(make_rejected)]
5875    #[case::expired(make_expired)]
5876    #[case::fill_voided(make_terminal_fill_voided)]
5877    fn test_handle_order_event_cancels_gtd_timer_when_stopped(
5878        #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
5879    ) {
5880        let mut strategy = create_test_strategy();
5881        register_strategy(&mut strategy);
5882        start_strategy(&mut strategy);
5883
5884        let client_order_id = ClientOrderId::from("O-001");
5885        strategy
5886            .core
5887            .gtd_timers
5888            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5889
5890        stop_strategy(&mut strategy);
5891        assert_eq!(strategy.state(), ComponentState::Stopped);
5892
5893        strategy.handle_order_event(make_event(client_order_id));
5894
5895        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5896    }
5897
5898    #[rstest]
5899    fn test_handle_order_event_skips_gtd_cancel_for_non_terminal() {
5900        let mut strategy = create_test_strategy();
5901        register_strategy(&mut strategy);
5902        start_strategy(&mut strategy);
5903
5904        let client_order_id = ClientOrderId::from("O-001");
5905        strategy
5906            .core
5907            .gtd_timers
5908            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5909
5910        strategy.handle_order_event(make_accepted(client_order_id));
5911
5912        assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5913    }
5914
5915    #[rstest]
5916    fn test_handle_reopened_fill_void_keeps_gtd_timer() {
5917        let mut strategy = create_test_strategy();
5918        register_strategy(&mut strategy);
5919        start_strategy(&mut strategy);
5920
5921        let client_order_id = ClientOrderId::from("O-001");
5922        strategy
5923            .core
5924            .gtd_timers
5925            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5926
5927        strategy.handle_order_event(make_fill_voided(client_order_id, true));
5928
5929        assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5930    }
5931
5932    #[rstest]
5933    fn test_handle_cancel_rejected_restores_future_gtd_timer() {
5934        let mut strategy = create_gtd_managed_strategy();
5935        let clock = register_gtd_strategy(&mut strategy);
5936        start_strategy(&mut strategy);
5937        let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
5938
5939        let order = make_accepted_gtd_limit_order("O-001", expire_time);
5940        let client_order_id = order.client_order_id();
5941        add_order_to_cache(&strategy, &order);
5942
5943        strategy.handle_order_event(make_cancel_rejected(client_order_id));
5944
5945        assert_eq!(
5946            strategy.core.gtd_timers.get(&client_order_id),
5947            Some(&Ustr::from("GTD-EXPIRY:O-001"))
5948        );
5949    }
5950
5951    #[rstest]
5952    fn test_cancel_rejected_restores_timer_after_cancel_order_removes_it() {
5953        let mut strategy = create_gtd_managed_strategy();
5954        let clock = register_gtd_strategy(&mut strategy);
5955        start_strategy(&mut strategy);
5956        let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
5957        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5958            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5959        msgbus::register_trading_command_endpoint(
5960            MessagingSwitchboard::exec_engine_queue_execute(),
5961            exec_handler,
5962        );
5963
5964        let order = make_accepted_gtd_limit_order("O-001", expire_time);
5965        let client_order_id = order.client_order_id();
5966        add_order_to_cache(&strategy, &order);
5967        strategy
5968            .core
5969            .gtd_timers
5970            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5971
5972        strategy.cancel_order(client_order_id, None, None).unwrap();
5973        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5974
5975        let event = make_cancel_rejected(client_order_id);
5976        strategy
5977            .core
5978            .cache_rc()
5979            .borrow_mut()
5980            .update_order(&event)
5981            .unwrap();
5982        assert_eq!(
5983            strategy
5984                .core
5985                .cache_ref()
5986                .order(&client_order_id)
5987                .unwrap()
5988                .status(),
5989            OrderStatus::Accepted
5990        );
5991
5992        strategy.handle_order_event(event);
5993
5994        assert_eq!(
5995            strategy.core.gtd_timers.get(&client_order_id),
5996            Some(&Ustr::from("GTD-EXPIRY:O-001"))
5997        );
5998        assert_eq!(exec_messages.get_messages().len(), 1);
5999    }
6000
6001    #[rstest]
6002    fn test_cancel_rejected_while_submitted_restores_timer_through_acceptance() {
6003        let mut strategy = create_gtd_managed_strategy();
6004        let clock = register_gtd_strategy(&mut strategy);
6005        start_strategy(&mut strategy);
6006        let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
6007        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6008            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6009        msgbus::register_trading_command_endpoint(
6010            MessagingSwitchboard::exec_engine_queue_execute(),
6011            exec_handler,
6012        );
6013
6014        let order = make_submitted_gtd_limit_order("O-001", expire_time);
6015        let client_order_id = order.client_order_id();
6016        add_order_to_cache(&strategy, &order);
6017        strategy.set_gtd_expiry(&order).unwrap();
6018
6019        strategy.cancel_order(client_order_id, None, None).unwrap();
6020        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6021
6022        let cancel_rejected = make_cancel_rejected(client_order_id);
6023        strategy
6024            .core
6025            .cache_rc()
6026            .borrow_mut()
6027            .update_order(&cancel_rejected)
6028            .unwrap();
6029        assert_eq!(
6030            strategy
6031                .core
6032                .cache_ref()
6033                .order(&client_order_id)
6034                .unwrap()
6035                .status(),
6036            OrderStatus::Submitted
6037        );
6038        strategy.handle_order_event(cancel_rejected);
6039        assert_eq!(
6040            strategy.core.gtd_timers.get(&client_order_id),
6041            Some(&Ustr::from("GTD-EXPIRY:O-001"))
6042        );
6043        assert_eq!(strategy.core.gtd_timers.len(), 1);
6044
6045        let accepted = make_accepted(client_order_id);
6046        strategy
6047            .core
6048            .cache_rc()
6049            .borrow_mut()
6050            .update_order(&accepted)
6051            .unwrap();
6052        strategy.handle_order_event(accepted);
6053
6054        assert_eq!(
6055            strategy.core.gtd_timers.get(&client_order_id),
6056            Some(&Ustr::from("GTD-EXPIRY:O-001"))
6057        );
6058        assert_eq!(strategy.core.gtd_timers.len(), 1);
6059        assert_eq!(exec_messages.get_messages().len(), 1);
6060    }
6061
6062    #[rstest]
6063    #[case::future_expiry(false)]
6064    #[case::elapsed_expiry(true)]
6065    fn test_cancel_rejected_after_acceptance_restores_gtd_expiry(#[case] expired: bool) {
6066        let mut strategy = create_gtd_managed_strategy();
6067        let clock = register_gtd_strategy(&mut strategy);
6068        start_strategy(&mut strategy);
6069        let expire_time = UnixNanos::from(1_000_000_000);
6070        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6071            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6072        msgbus::register_trading_command_endpoint(
6073            MessagingSwitchboard::exec_engine_queue_execute(),
6074            exec_handler,
6075        );
6076        let order = make_submitted_gtd_limit_order("O-001", expire_time);
6077        let client_order_id = order.client_order_id();
6078        add_order_to_cache(&strategy, &order);
6079        strategy.set_gtd_expiry(&order).unwrap();
6080
6081        strategy.cancel_order(client_order_id, None, None).unwrap();
6082        let accepted = make_accepted(client_order_id);
6083        strategy
6084            .core
6085            .cache_rc()
6086            .borrow_mut()
6087            .update_order(&accepted)
6088            .unwrap();
6089        strategy.handle_order_event(accepted);
6090        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6091        assert_eq!(exec_messages.get_messages().len(), 1);
6092
6093        if expired {
6094            clock.borrow_mut().set_time(expire_time);
6095        }
6096        let cancel_rejected = make_cancel_rejected(client_order_id);
6097        strategy
6098            .core
6099            .cache_rc()
6100            .borrow_mut()
6101            .update_order(&cancel_rejected)
6102            .unwrap();
6103        strategy.handle_order_event(cancel_rejected);
6104
6105        if !expired {
6106            assert_eq!(
6107                strategy.core.gtd_timers.get(&client_order_id),
6108                Some(&Ustr::from("GTD-EXPIRY:O-001"))
6109            );
6110            assert_eq!(
6111                strategy
6112                    .core
6113                    .cache_ref()
6114                    .order(&client_order_id)
6115                    .unwrap()
6116                    .status(),
6117                OrderStatus::Accepted
6118            );
6119            assert_eq!(exec_messages.get_messages().len(), 1);
6120            let events = clock.borrow_mut().advance_time(expire_time, true);
6121            assert_eq!(events.len(), 1);
6122            route_time_event(&mut strategy, &events[0]);
6123        }
6124
6125        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6126        assert_eq!(
6127            strategy
6128                .core
6129                .cache_ref()
6130                .order(&client_order_id)
6131                .unwrap()
6132                .status(),
6133            OrderStatus::PendingCancel
6134        );
6135        let commands = exec_messages.get_messages();
6136        assert_eq!(commands.len(), 2);
6137        for command in commands {
6138            assert!(matches!(
6139                command,
6140                TradingCommand::CancelOrder(command) if command.client_order_id == client_order_id
6141            ));
6142        }
6143    }
6144
6145    #[rstest]
6146    fn test_cancel_rejected_while_submitted_retries_already_expired_gtd_order() {
6147        let mut strategy = create_gtd_managed_strategy();
6148        let clock = register_gtd_strategy(&mut strategy);
6149        start_strategy(&mut strategy);
6150        clock.borrow_mut().set_time(UnixNanos::from(1_000_000_000));
6151        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6152            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6153        msgbus::register_trading_command_endpoint(
6154            MessagingSwitchboard::exec_engine_queue_execute(),
6155            exec_handler,
6156        );
6157
6158        let order = make_submitted_gtd_limit_order("O-001", UnixNanos::from(1));
6159        let client_order_id = order.client_order_id();
6160        add_order_to_cache(&strategy, &order);
6161
6162        strategy.handle_order_event(make_cancel_rejected(client_order_id));
6163
6164        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6165        assert!(matches!(
6166            exec_messages.get_messages().as_slice(),
6167            [TradingCommand::CancelOrder(command)]
6168                if command.client_order_id == client_order_id
6169        ));
6170        assert_eq!(
6171            strategy
6172                .core
6173                .cache_ref()
6174                .order(&client_order_id)
6175                .unwrap()
6176                .status(),
6177            OrderStatus::PendingCancel
6178        );
6179    }
6180
6181    #[rstest]
6182    fn test_cancel_rejected_while_submitted_preserves_existing_gtd_timer() {
6183        let mut strategy = create_gtd_managed_strategy();
6184        let clock = register_gtd_strategy(&mut strategy);
6185        start_strategy(&mut strategy);
6186        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6187            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6188        msgbus::register_trading_command_endpoint(
6189            MessagingSwitchboard::exec_engine_queue_execute(),
6190            exec_handler,
6191        );
6192
6193        let order =
6194            make_submitted_gtd_limit_order("O-001", UnixNanos::from(2_000_000_000_000_000_000));
6195        let client_order_id = order.client_order_id();
6196        add_order_to_cache(&strategy, &order);
6197        let timer_name = Ustr::from("GTD-EXPIRY:O-001");
6198        strategy.core.gtd_timers.insert(client_order_id, timer_name);
6199
6200        strategy.handle_order_event(make_cancel_rejected(client_order_id));
6201
6202        assert_eq!(
6203            strategy.core.gtd_timers.get(&client_order_id),
6204            Some(&timer_name)
6205        );
6206        assert_eq!(strategy.core.gtd_timers.len(), 1);
6207        // The map entry is seeded without a clock alert, so an unintended
6208        // re-arm is observable only as a new alert on the clock.
6209        assert!(!clock.borrow().timer_names().contains(&timer_name.as_str()));
6210        assert!(exec_messages.get_messages().is_empty());
6211    }
6212
6213    #[rstest]
6214    fn test_handle_cancel_rejected_retries_already_expired_gtd_order() {
6215        let mut strategy = create_gtd_managed_strategy();
6216        let clock = register_gtd_strategy(&mut strategy);
6217        start_strategy(&mut strategy);
6218        clock.borrow_mut().set_time(UnixNanos::from(1_000_000_000));
6219        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6220            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6221        msgbus::register_trading_command_endpoint(
6222            MessagingSwitchboard::exec_engine_queue_execute(),
6223            exec_handler,
6224        );
6225
6226        let order = make_accepted_gtd_limit_order("O-001", UnixNanos::from(1));
6227        let client_order_id = order.client_order_id();
6228        add_order_to_cache(&strategy, &order);
6229
6230        strategy.handle_order_event(make_cancel_rejected(client_order_id));
6231
6232        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6233        assert!(matches!(
6234            exec_messages.get_messages().as_slice(),
6235            [TradingCommand::CancelOrder(command)]
6236                if command.client_order_id == client_order_id
6237        ));
6238        assert_eq!(
6239            strategy
6240                .core
6241                .cache_ref()
6242                .order(&client_order_id)
6243                .unwrap()
6244                .status(),
6245            OrderStatus::PendingCancel
6246        );
6247    }
6248
6249    #[rstest]
6250    fn test_handle_cancel_rejected_preserves_existing_gtd_timer() {
6251        let mut strategy = create_gtd_managed_strategy();
6252        let _clock = register_gtd_strategy(&mut strategy);
6253        start_strategy(&mut strategy);
6254        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6255            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6256        msgbus::register_trading_command_endpoint(
6257            MessagingSwitchboard::exec_engine_queue_execute(),
6258            exec_handler,
6259        );
6260
6261        let order =
6262            make_accepted_gtd_limit_order("O-001", UnixNanos::from(2_000_000_000_000_000_000));
6263        let client_order_id = order.client_order_id();
6264        add_order_to_cache(&strategy, &order);
6265        strategy
6266            .core
6267            .gtd_timers
6268            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
6269
6270        strategy.handle_order_event(make_cancel_rejected(client_order_id));
6271
6272        assert_eq!(
6273            strategy.core.gtd_timers.get(&client_order_id),
6274            Some(&Ustr::from("GTD-EXPIRY:O-001"))
6275        );
6276        assert!(exec_messages.get_messages().is_empty());
6277    }
6278
6279    #[rstest]
6280    fn test_handle_cancel_rejected_preserves_existing_timer_for_expired_order() {
6281        let mut strategy = create_gtd_managed_strategy();
6282        let clock = register_gtd_strategy(&mut strategy);
6283        start_strategy(&mut strategy);
6284        clock.borrow_mut().set_time(UnixNanos::from(1_000_000_000));
6285        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6286            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6287        msgbus::register_trading_command_endpoint(
6288            MessagingSwitchboard::exec_engine_queue_execute(),
6289            exec_handler,
6290        );
6291
6292        let order = make_accepted_gtd_limit_order("O-001", UnixNanos::from(1));
6293        let client_order_id = order.client_order_id();
6294        add_order_to_cache(&strategy, &order);
6295        strategy
6296            .core
6297            .gtd_timers
6298            .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
6299
6300        strategy.handle_order_event(make_cancel_rejected(client_order_id));
6301
6302        assert_eq!(
6303            strategy.core.gtd_timers.get(&client_order_id),
6304            Some(&Ustr::from("GTD-EXPIRY:O-001"))
6305        );
6306        assert!(exec_messages.get_messages().is_empty());
6307    }
6308
6309    #[rstest]
6310    fn test_restored_gtd_timer_event_cancels_order() {
6311        let mut strategy = create_gtd_managed_strategy();
6312        let clock = register_gtd_strategy(&mut strategy);
6313        start_strategy(&mut strategy);
6314        let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
6315        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6316            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6317        msgbus::register_trading_command_endpoint(
6318            MessagingSwitchboard::exec_engine_queue_execute(),
6319            exec_handler,
6320        );
6321
6322        let order = make_accepted_gtd_limit_order("O-001", expire_time);
6323        let client_order_id = order.client_order_id();
6324        add_order_to_cache(&strategy, &order);
6325
6326        strategy.handle_order_event(make_cancel_rejected(client_order_id));
6327        assert!(strategy.has_gtd_expiry_timer(&client_order_id));
6328
6329        let events = clock.borrow_mut().advance_time(expire_time, true);
6330        for event in &events {
6331            route_time_event(&mut strategy, event);
6332        }
6333
6334        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6335        assert!(matches!(
6336            exec_messages.get_messages().as_slice(),
6337            [TradingCommand::CancelOrder(command)]
6338                if command.client_order_id == client_order_id
6339        ));
6340    }
6341
6342    enum IneligibleRestoreCase {
6343        MissingOrder,
6344        ClosedOrder,
6345        NonGtdOrder,
6346        ManagementDisabled,
6347        Stopped,
6348        Emulated,
6349    }
6350
6351    fn assert_cancel_rejected_does_not_restore(case: &IneligibleRestoreCase) {
6352        let mut strategy = match case {
6353            IneligibleRestoreCase::ManagementDisabled => create_test_strategy(),
6354            _ => create_gtd_managed_strategy(),
6355        };
6356        let _clock = register_gtd_strategy(&mut strategy);
6357        start_strategy(&mut strategy);
6358        let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6359            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6360        msgbus::register_trading_command_endpoint(
6361            MessagingSwitchboard::exec_engine_queue_execute(),
6362            exec_handler,
6363        );
6364
6365        let client_order_id = ClientOrderId::from("O-001");
6366        let order = match case {
6367            IneligibleRestoreCase::MissingOrder => None,
6368            IneligibleRestoreCase::NonGtdOrder => {
6369                Some(make_accepted_limit_order(client_order_id.as_str()))
6370            }
6371            IneligibleRestoreCase::ClosedOrder => {
6372                let mut order = make_accepted_gtd_limit_order(
6373                    client_order_id.as_str(),
6374                    UnixNanos::from(2_000_000_000_000_000_000),
6375                );
6376                order.apply(make_canceled(client_order_id)).unwrap();
6377                Some(order)
6378            }
6379            IneligibleRestoreCase::Emulated => {
6380                let mut order = make_submitted_gtd_limit_order(
6381                    client_order_id.as_str(),
6382                    UnixNanos::from(2_000_000_000_000_000_000),
6383                );
6384                order.set_emulation_trigger(Some(TriggerType::BidAsk));
6385                Some(order)
6386            }
6387            IneligibleRestoreCase::ManagementDisabled | IneligibleRestoreCase::Stopped => {
6388                Some(make_accepted_gtd_limit_order(
6389                    client_order_id.as_str(),
6390                    UnixNanos::from(2_000_000_000_000_000_000),
6391                ))
6392            }
6393        };
6394
6395        if let Some(order) = order {
6396            add_order_to_cache(&strategy, &order);
6397        }
6398
6399        if matches!(case, IneligibleRestoreCase::Stopped) {
6400            stop_strategy(&mut strategy);
6401        }
6402
6403        strategy.handle_order_event(make_cancel_rejected(client_order_id));
6404
6405        assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6406        assert!(exec_messages.get_messages().is_empty());
6407    }
6408
6409    #[rstest]
6410    fn test_handle_cancel_rejected_does_not_restore_missing_order() {
6411        assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::MissingOrder);
6412    }
6413
6414    #[rstest]
6415    fn test_handle_cancel_rejected_does_not_restore_closed_order() {
6416        assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::ClosedOrder);
6417    }
6418
6419    #[rstest]
6420    fn test_handle_cancel_rejected_does_not_restore_non_gtd_order() {
6421        assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::NonGtdOrder);
6422    }
6423
6424    #[rstest]
6425    fn test_handle_cancel_rejected_does_not_restore_when_disabled() {
6426        assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::ManagementDisabled);
6427    }
6428
6429    #[rstest]
6430    fn test_handle_cancel_rejected_does_not_restore_when_stopped() {
6431        assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::Stopped);
6432    }
6433
6434    #[rstest]
6435    fn test_handle_cancel_rejected_does_not_restore_emulated_order() {
6436        assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::Emulated);
6437    }
6438
6439    #[rstest]
6440    fn test_handle_order_event_skips_dispatch_when_stopped() {
6441        let mut strategy = create_test_strategy();
6442        register_strategy(&mut strategy);
6443        start_strategy(&mut strategy);
6444        stop_strategy(&mut strategy);
6445        assert_eq!(strategy.state(), ComponentState::Stopped);
6446
6447        strategy.handle_order_event(make_rejected(ClientOrderId::from("O-001")));
6448
6449        assert!(!strategy.on_order_event_called);
6450        assert!(!strategy.on_order_rejected_called);
6451    }
6452
6453    #[rstest]
6454    fn test_on_start_calls_reactivate_gtd_timers_when_enabled() {
6455        let config = StrategyConfig {
6456            strategy_id: Some(StrategyId::from("TEST-001")),
6457            order_id_tag: Some("001".to_string()),
6458            manage_gtd_expiry: true,
6459            ..Default::default()
6460        };
6461        let mut strategy = TestStrategy::new(config);
6462        register_strategy(&mut strategy);
6463
6464        let result = Strategy::on_start(&mut strategy);
6465        assert!(result.is_ok());
6466    }
6467
6468    #[rstest]
6469    fn test_on_start_does_not_panic_when_gtd_disabled() {
6470        let config = StrategyConfig {
6471            strategy_id: Some(StrategyId::from("TEST-001")),
6472            order_id_tag: Some("001".to_string()),
6473            manage_gtd_expiry: false,
6474            ..Default::default()
6475        };
6476        let mut strategy = TestStrategy::new(config);
6477        register_strategy(&mut strategy);
6478
6479        let result = Strategy::on_start(&mut strategy);
6480        assert!(result.is_ok());
6481    }
6482
6483    #[rstest]
6484    fn test_on_start_errors_when_strategy_id_is_not_set() {
6485        let mut strategy = TestStrategy::new(StrategyConfig::default());
6486
6487        let err = Strategy::on_start(&mut strategy).unwrap_err().to_string();
6488
6489        assert_eq!(err, "Strategy not registered: strategy_id is not set");
6490    }
6491
6492    // -- QUERY TESTS ---------------------------------------------------------------------------------
6493
6494    #[rstest]
6495    fn test_query_account_when_registered() {
6496        let mut strategy = create_test_strategy();
6497        register_strategy(&mut strategy);
6498
6499        let account_id = AccountId::from("ACC-001");
6500
6501        let result = strategy.query_account(account_id, None, None);
6502
6503        assert!(result.is_ok());
6504    }
6505
6506    #[rstest]
6507    fn test_query_account_with_client_id() {
6508        let mut strategy = create_test_strategy();
6509        register_strategy(&mut strategy);
6510
6511        let account_id = AccountId::from("ACC-001");
6512        let client_id = ClientId::from("BINANCE");
6513
6514        let result = strategy.query_account(account_id, Some(client_id), None);
6515
6516        assert!(result.is_ok());
6517    }
6518
6519    #[rstest]
6520    fn test_query_order_when_registered() {
6521        let mut strategy = create_test_strategy();
6522        register_strategy(&mut strategy);
6523
6524        let order = OrderAny::Market(MarketOrder::test_default());
6525
6526        let result = strategy.query_order(&order, None, None);
6527
6528        assert!(result.is_ok());
6529    }
6530
6531    #[rstest]
6532    fn test_query_order_with_client_id() {
6533        let mut strategy = create_test_strategy();
6534        register_strategy(&mut strategy);
6535
6536        let order = OrderAny::Market(MarketOrder::test_default());
6537        let client_id = ClientId::from("BINANCE");
6538
6539        let result = strategy.query_order(&order, Some(client_id), None);
6540
6541        assert!(result.is_ok());
6542    }
6543
6544    #[rstest]
6545    fn test_is_exiting_returns_false_by_default() {
6546        let strategy = create_test_strategy();
6547        assert!(!strategy.is_exiting());
6548    }
6549
6550    #[rstest]
6551    fn test_is_exiting_returns_true_when_set_manually() {
6552        let mut strategy = create_test_strategy();
6553        register_strategy(&mut strategy);
6554
6555        // Manually set the exiting state (as market_exit would do)
6556        strategy.core.is_exiting = true;
6557
6558        assert!(strategy.is_exiting());
6559    }
6560
6561    #[rstest]
6562    fn test_market_exit_sets_is_exiting_flag() {
6563        // Test the state changes that market_exit would make
6564        let mut strategy = create_test_strategy();
6565        register_strategy(&mut strategy);
6566
6567        assert!(!strategy.core.is_exiting);
6568
6569        // Simulate what market_exit does to the state
6570        strategy.core.is_exiting = true;
6571        strategy.core.market_exit_attempts = 0;
6572
6573        assert!(strategy.core.is_exiting);
6574        assert_eq!(strategy.core.market_exit_attempts, 0);
6575    }
6576
6577    #[rstest]
6578    fn test_market_exit_uses_config_time_in_force_and_reduce_only() {
6579        let config = StrategyConfig {
6580            strategy_id: Some(StrategyId::from("TEST-001")),
6581            order_id_tag: Some("001".to_string()),
6582            market_exit_time_in_force: TimeInForce::Ioc,
6583            market_exit_reduce_only: false,
6584            ..Default::default()
6585        };
6586        let strategy = TestStrategy::new(config);
6587
6588        assert_eq!(
6589            strategy.core.config.market_exit_time_in_force,
6590            TimeInForce::Ioc
6591        );
6592        assert!(!strategy.core.config.market_exit_reduce_only);
6593    }
6594
6595    #[rstest]
6596    fn test_market_exit_resets_attempt_counter() {
6597        let mut strategy = create_test_strategy();
6598        register_strategy(&mut strategy);
6599
6600        // Manually set attempts to simulate prior exit
6601        strategy.core.market_exit_attempts = 50;
6602
6603        // Reset via the reset method
6604        strategy.core.reset_market_exit_state();
6605
6606        assert_eq!(strategy.core.market_exit_attempts, 0);
6607    }
6608
6609    #[rstest]
6610    fn test_market_exit_second_call_returns_early_when_exiting() {
6611        let mut strategy = create_test_strategy();
6612        register_strategy(&mut strategy);
6613
6614        // First set exiting to true to simulate an in-progress exit
6615        strategy.core.is_exiting = true;
6616
6617        // Second call should return Ok and not change state
6618        let result = strategy.market_exit();
6619        assert!(result.is_ok());
6620        assert!(strategy.core.is_exiting);
6621    }
6622
6623    #[rstest]
6624    fn test_finalize_market_exit_resets_state() {
6625        let mut strategy = create_test_strategy();
6626        register_strategy(&mut strategy);
6627
6628        // Set up exiting state
6629        strategy.core.is_exiting = true;
6630        strategy.core.pending_stop = true;
6631        strategy.core.market_exit_attempts = 50;
6632
6633        strategy.finalize_market_exit();
6634
6635        assert!(!strategy.core.is_exiting);
6636        assert!(!strategy.core.pending_stop);
6637        assert_eq!(strategy.core.market_exit_attempts, 0);
6638    }
6639
6640    #[rstest]
6641    fn test_market_exit_config_defaults() {
6642        let config = StrategyConfig::default();
6643
6644        assert!(!config.manage_stop);
6645        assert_eq!(config.market_exit_interval_ms, 100);
6646        assert_eq!(config.market_exit_max_attempts, 100);
6647    }
6648
6649    #[rstest]
6650    fn test_market_exit_with_custom_config() {
6651        let config = StrategyConfig {
6652            strategy_id: Some(StrategyId::from("TEST-001")),
6653            manage_stop: true,
6654            market_exit_interval_ms: 50,
6655            market_exit_max_attempts: 200,
6656            ..Default::default()
6657        };
6658        let strategy = TestStrategy::new(config);
6659
6660        assert!(strategy.core.config.manage_stop);
6661        assert_eq!(strategy.core.config.market_exit_interval_ms, 50);
6662        assert_eq!(strategy.core.config.market_exit_max_attempts, 200);
6663    }
6664
6665    #[derive(Debug)]
6666    struct MarketExitHookTrackingStrategy {
6667        core: StrategyCore,
6668        on_market_exit_called: bool,
6669        post_market_exit_called: bool,
6670    }
6671
6672    impl MarketExitHookTrackingStrategy {
6673        fn new(config: StrategyConfig) -> Self {
6674            Self {
6675                core: StrategyCore::new(config),
6676                on_market_exit_called: false,
6677                post_market_exit_called: false,
6678            }
6679        }
6680    }
6681
6682    impl DataActor for MarketExitHookTrackingStrategy {}
6683
6684    nautilus_strategy!(MarketExitHookTrackingStrategy, {
6685        fn on_market_exit(&mut self) {
6686            self.on_market_exit_called = true;
6687        }
6688
6689        fn post_market_exit(&mut self) {
6690            self.post_market_exit_called = true;
6691        }
6692    });
6693
6694    #[rstest]
6695    fn test_market_exit_calls_on_market_exit_hook() {
6696        let config = StrategyConfig {
6697            strategy_id: Some(StrategyId::from("TEST-001")),
6698            order_id_tag: Some("001".to_string()),
6699            ..Default::default()
6700        };
6701        let mut strategy = MarketExitHookTrackingStrategy::new(config);
6702
6703        let trader_id = TraderId::from("TRADER-001");
6704        let clock = Rc::new(RefCell::new(VirtualClock::new()));
6705        let cache = Rc::new(RefCell::new(Cache::default()));
6706        let portfolio = Rc::new(RefCell::new(Portfolio::new(
6707            clock.clone(),
6708            cache.clone(),
6709            None,
6710        )));
6711        strategy
6712            .core
6713            .register(trader_id, clock, cache, portfolio)
6714            .unwrap();
6715        strategy.initialize().unwrap();
6716        strategy.start().unwrap();
6717
6718        let _ = strategy.market_exit();
6719
6720        assert!(strategy.on_market_exit_called);
6721    }
6722
6723    #[rstest]
6724    fn test_finalize_market_exit_calls_post_market_exit_hook() {
6725        let config = StrategyConfig {
6726            strategy_id: Some(StrategyId::from("TEST-001")),
6727            order_id_tag: Some("001".to_string()),
6728            ..Default::default()
6729        };
6730        let mut strategy = MarketExitHookTrackingStrategy::new(config);
6731
6732        let trader_id = TraderId::from("TRADER-001");
6733        let clock = Rc::new(RefCell::new(VirtualClock::new()));
6734        let cache = Rc::new(RefCell::new(Cache::default()));
6735        let portfolio = Rc::new(RefCell::new(Portfolio::new(
6736            clock.clone(),
6737            cache.clone(),
6738            None,
6739        )));
6740        strategy
6741            .core
6742            .register(trader_id, clock, cache, portfolio)
6743            .unwrap();
6744
6745        strategy.core.is_exiting = true;
6746        strategy.finalize_market_exit();
6747
6748        assert!(strategy.post_market_exit_called);
6749    }
6750
6751    #[derive(Debug)]
6752    struct FailingPostExitStrategy {
6753        core: StrategyCore,
6754    }
6755
6756    impl FailingPostExitStrategy {
6757        fn new(config: StrategyConfig) -> Self {
6758            Self {
6759                core: StrategyCore::new(config),
6760            }
6761        }
6762    }
6763
6764    impl DataActor for FailingPostExitStrategy {}
6765
6766    nautilus_strategy!(FailingPostExitStrategy, {
6767        fn post_market_exit(&mut self) {
6768            panic!("Simulated error in post_market_exit");
6769        }
6770    });
6771
6772    #[rstest]
6773    fn test_finalize_market_exit_handles_hook_panic() {
6774        let config = StrategyConfig {
6775            strategy_id: Some(StrategyId::from("TEST-001")),
6776            order_id_tag: Some("001".to_string()),
6777            ..Default::default()
6778        };
6779        let mut strategy = FailingPostExitStrategy::new(config);
6780
6781        let trader_id = TraderId::from("TRADER-001");
6782        let clock = Rc::new(RefCell::new(VirtualClock::new()));
6783        let cache = Rc::new(RefCell::new(Cache::default()));
6784        let portfolio = Rc::new(RefCell::new(Portfolio::new(
6785            clock.clone(),
6786            cache.clone(),
6787            None,
6788        )));
6789        strategy
6790            .core
6791            .register(trader_id, clock, cache, portfolio)
6792            .unwrap();
6793
6794        strategy.core.is_exiting = true;
6795        strategy.core.pending_stop = true;
6796
6797        // This should not panic - it should catch the panic in post_market_exit
6798        strategy.finalize_market_exit();
6799
6800        // State should still be reset
6801        assert!(!strategy.core.is_exiting);
6802        assert!(!strategy.core.pending_stop);
6803    }
6804
6805    #[rstest]
6806    fn test_check_market_exit_increments_attempts_before_finalizing() {
6807        let mut strategy = create_test_strategy();
6808        register_strategy(&mut strategy);
6809
6810        strategy.core.is_exiting = true;
6811        assert_eq!(strategy.core.market_exit_attempts, 0);
6812
6813        let event = TimeEvent::new(
6814            Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
6815            UUID4::new(),
6816            UnixNanos::default(),
6817            UnixNanos::default(),
6818        );
6819        strategy.check_market_exit(event);
6820
6821        // With no orders/positions, check_market_exit will finalize immediately
6822        // which resets attempts to 0. This is correct behavior.
6823        // The attempt WAS incremented to 1 during the check, then reset on finalize.
6824        assert!(!strategy.core.is_exiting);
6825        assert_eq!(strategy.core.market_exit_attempts, 0);
6826    }
6827
6828    #[rstest]
6829    fn test_route_time_event_is_idempotent_when_callback_forwards() {
6830        let mut strategy = TestStrategy::new(StrategyConfig {
6831            strategy_id: Some(StrategyId::from("TEST-001")),
6832            ..Default::default()
6833        });
6834        register_strategy(&mut strategy);
6835
6836        let order = make_accepted_limit_order("O-PRESENT");
6837        add_order_to_cache(&strategy, &order);
6838        strategy.core.cache_rc().borrow_mut().build_index();
6839        strategy.core.is_exiting = true;
6840        let event = TimeEvent::new(
6841            strategy.core.market_exit_timer_name,
6842            UUID4::new(),
6843            UnixNanos::default(),
6844            UnixNanos::default(),
6845        );
6846
6847        route_time_event(&mut strategy, &event);
6848        Strategy::on_time_event(&mut strategy, &event).unwrap();
6849
6850        assert!(strategy.core.is_exiting);
6851        assert_eq!(strategy.core.market_exit_attempts, 1);
6852        assert_eq!(
6853            strategy.core.managed_time_event_last_id,
6854            Some(event.event_id)
6855        );
6856    }
6857
6858    #[rstest]
6859    fn test_route_time_event_deduplicates_overridden_managed_handlers() {
6860        let mut strategy = TimerOverrideStrategy {
6861            core: StrategyCore::new(StrategyConfig {
6862                strategy_id: Some(StrategyId::from("TEST-001")),
6863                ..Default::default()
6864            }),
6865            gtd_expiries: 0,
6866            market_exit_checks: 0,
6867        };
6868        let market_event = TimeEvent::new(
6869            strategy.core.market_exit_timer_name,
6870            UUID4::new(),
6871            UnixNanos::default(),
6872            UnixNanos::default(),
6873        );
6874
6875        route_time_event(&mut strategy, &market_event);
6876        DataActor::on_time_event(&mut strategy, &market_event).unwrap();
6877
6878        let client_order_id = ClientOrderId::from("O-001");
6879        let gtd_timer_name = Ustr::from("GTD-EXPIRY:O-001");
6880        strategy
6881            .core
6882            .gtd_timers
6883            .insert(client_order_id, gtd_timer_name);
6884        let gtd_event = TimeEvent::new(
6885            gtd_timer_name,
6886            UUID4::new(),
6887            UnixNanos::default(),
6888            UnixNanos::default(),
6889        );
6890
6891        route_time_event(&mut strategy, &gtd_event);
6892        DataActor::on_time_event(&mut strategy, &gtd_event).unwrap();
6893
6894        assert_eq!(strategy.market_exit_checks, 1);
6895        assert_eq!(strategy.gtd_expiries, 1);
6896        assert!(strategy.core.gtd_timers.contains_key(&client_order_id));
6897        assert_eq!(
6898            strategy.core.managed_time_event_last_id,
6899            Some(gtd_event.event_id)
6900        );
6901    }
6902
6903    #[rstest]
6904    fn test_route_time_event_ignores_unowned_market_exit_timer() {
6905        let mut strategy = TestStrategy::new(StrategyConfig {
6906            strategy_id: Some(StrategyId::from("TEST-001")),
6907            ..Default::default()
6908        });
6909        register_strategy(&mut strategy);
6910
6911        let order = make_accepted_limit_order("O-PRESENT");
6912        add_order_to_cache(&strategy, &order);
6913        strategy.core.cache_rc().borrow_mut().build_index();
6914        strategy.core.is_exiting = true;
6915        let event = TimeEvent::new(
6916            Ustr::from("MARKET_EXIT_CHECK:OTHER-001"),
6917            UUID4::new(),
6918            UnixNanos::default(),
6919            UnixNanos::default(),
6920        );
6921
6922        route_time_event(&mut strategy, &event);
6923
6924        assert!(strategy.core.is_exiting);
6925        assert_eq!(strategy.core.market_exit_attempts, 0);
6926        assert_eq!(strategy.core.managed_time_event_last_id, None);
6927    }
6928
6929    #[rstest]
6930    fn test_check_market_exit_finalizes_when_max_attempts_reached() {
6931        let config = StrategyConfig {
6932            strategy_id: Some(StrategyId::from("TEST-001")),
6933            order_id_tag: Some("001".to_string()),
6934            market_exit_max_attempts: 3,
6935            ..Default::default()
6936        };
6937        let mut strategy = TestStrategy::new(config);
6938        register_strategy(&mut strategy);
6939
6940        strategy.core.is_exiting = true;
6941        strategy.core.market_exit_attempts = 2; // One below max
6942
6943        let event = TimeEvent::new(
6944            Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
6945            UUID4::new(),
6946            UnixNanos::default(),
6947            UnixNanos::default(),
6948        );
6949        strategy.check_market_exit(event);
6950
6951        // Should have finalized since attempts >= max_attempts
6952        assert!(!strategy.core.is_exiting);
6953        assert_eq!(strategy.core.market_exit_attempts, 0);
6954    }
6955
6956    #[rstest]
6957    fn test_check_market_exit_finalizes_when_no_orders_or_positions() {
6958        let mut strategy = create_test_strategy();
6959        register_strategy(&mut strategy);
6960
6961        strategy.core.is_exiting = true;
6962
6963        let event = TimeEvent::new(
6964            Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
6965            UUID4::new(),
6966            UnixNanos::default(),
6967            UnixNanos::default(),
6968        );
6969        strategy.check_market_exit(event);
6970
6971        // Should have finalized since there are no orders or positions
6972        assert!(!strategy.core.is_exiting);
6973    }
6974
6975    #[rstest]
6976    fn test_market_exit_timer_name_format() {
6977        let config = StrategyConfig {
6978            strategy_id: Some(StrategyId::from("MY-STRATEGY-001")),
6979            ..Default::default()
6980        };
6981        let strategy = TestStrategy::new(config);
6982
6983        assert_eq!(
6984            strategy.core.market_exit_timer_name,
6985            "MARKET_EXIT_CHECK:MY-STRATEGY-001"
6986        );
6987    }
6988
6989    #[rstest]
6990    fn test_reset_market_exit_state() {
6991        let mut strategy = create_test_strategy();
6992
6993        strategy.core.is_exiting = true;
6994        strategy.core.pending_stop = true;
6995        strategy.core.market_exit_attempts = 50;
6996
6997        strategy.core.reset_market_exit_state();
6998
6999        assert!(!strategy.core.is_exiting);
7000        assert!(!strategy.core.pending_stop);
7001        assert_eq!(strategy.core.market_exit_attempts, 0);
7002    }
7003
7004    #[rstest]
7005    fn test_cancel_market_exit_resets_state_without_hooks() {
7006        let config = StrategyConfig {
7007            strategy_id: Some(StrategyId::from("TEST-001")),
7008            order_id_tag: Some("001".to_string()),
7009            ..Default::default()
7010        };
7011        let mut strategy = MarketExitHookTrackingStrategy::new(config);
7012
7013        let trader_id = TraderId::from("TRADER-001");
7014        let clock = Rc::new(RefCell::new(VirtualClock::new()));
7015        let cache = Rc::new(RefCell::new(Cache::default()));
7016        let portfolio = Rc::new(RefCell::new(Portfolio::new(
7017            clock.clone(),
7018            cache.clone(),
7019            None,
7020        )));
7021        strategy
7022            .core
7023            .register(trader_id, clock, cache, portfolio)
7024            .unwrap();
7025
7026        // Set up exiting state
7027        strategy.core.is_exiting = true;
7028        strategy.core.pending_stop = true;
7029        strategy.core.market_exit_attempts = 50;
7030
7031        // Call cancel_market_exit
7032        strategy.cancel_market_exit();
7033
7034        // State should be reset
7035        assert!(!strategy.core.is_exiting);
7036        assert!(!strategy.core.pending_stop);
7037        assert_eq!(strategy.core.market_exit_attempts, 0);
7038
7039        // Hooks should NOT have been called
7040        assert!(!strategy.on_market_exit_called);
7041        assert!(!strategy.post_market_exit_called);
7042    }
7043
7044    #[rstest]
7045    fn test_market_exit_returns_early_when_not_running() {
7046        let mut strategy = create_test_strategy();
7047        register_strategy(&mut strategy);
7048
7049        // State is not Running (default is PreInitialized)
7050        assert!(!strategy.is_running());
7051
7052        let result = strategy.market_exit();
7053
7054        // Should return Ok but not set is_exiting
7055        assert!(result.is_ok());
7056        assert!(!strategy.core.is_exiting);
7057    }
7058
7059    #[rstest]
7060    fn test_stop_with_manage_stop_false_cleans_up_active_exit() {
7061        let config = StrategyConfig {
7062            strategy_id: Some(StrategyId::from("TEST-001")),
7063            order_id_tag: Some("001".to_string()),
7064            manage_stop: false,
7065            ..Default::default()
7066        };
7067        let mut strategy = TestStrategy::new(config);
7068        register_strategy(&mut strategy);
7069
7070        // Simulate an active market exit
7071        strategy.core.is_exiting = true;
7072        strategy.core.market_exit_attempts = 5;
7073
7074        // Call stop
7075        let should_proceed = Strategy::stop(&mut strategy);
7076
7077        // Should clean up state and allow stop to proceed
7078        assert!(should_proceed);
7079        assert!(!strategy.core.is_exiting);
7080        assert_eq!(strategy.core.market_exit_attempts, 0);
7081    }
7082
7083    #[rstest]
7084    fn test_stop_with_manage_stop_true_defers_when_running() {
7085        let config = StrategyConfig {
7086            strategy_id: Some(StrategyId::from("TEST-001")),
7087            order_id_tag: Some("001".to_string()),
7088            manage_stop: true,
7089            ..Default::default()
7090        };
7091        let mut strategy = TestStrategy::new(config);
7092
7093        // Custom setup with a default callback so timer scheduling succeeds
7094        let trader_id = TraderId::from("TRADER-001");
7095        let clock = Rc::new(RefCell::new(VirtualClock::new()));
7096        clock
7097            .borrow_mut()
7098            .register_default_handler(TimeEventCallback::from(|_event: TimeEvent| {}));
7099        let cache = Rc::new(RefCell::new(Cache::default()));
7100        let portfolio = Rc::new(RefCell::new(Portfolio::new(
7101            clock.clone(),
7102            cache.clone(),
7103            None,
7104        )));
7105        strategy
7106            .core
7107            .register(trader_id, clock, cache, portfolio)
7108            .unwrap();
7109        strategy.initialize().unwrap();
7110        strategy.start().unwrap();
7111
7112        let should_proceed = Strategy::stop(&mut strategy);
7113
7114        // Should set pending_stop and defer
7115        assert!(!should_proceed);
7116        assert!(strategy.core.pending_stop);
7117    }
7118
7119    #[rstest]
7120    fn test_stop_with_manage_stop_true_returns_early_if_pending() {
7121        let config = StrategyConfig {
7122            strategy_id: Some(StrategyId::from("TEST-001")),
7123            order_id_tag: Some("001".to_string()),
7124            manage_stop: true,
7125            ..Default::default()
7126        };
7127        let mut strategy = TestStrategy::new(config);
7128        register_strategy(&mut strategy);
7129        start_strategy(&mut strategy);
7130        strategy.core.pending_stop = true;
7131
7132        // Call stop again
7133        let should_proceed = Strategy::stop(&mut strategy);
7134
7135        // Should return early without changing state
7136        assert!(!should_proceed);
7137        assert!(strategy.core.pending_stop);
7138    }
7139
7140    #[rstest]
7141    fn test_stop_with_manage_stop_true_proceeds_when_not_running() {
7142        let config = StrategyConfig {
7143            strategy_id: Some(StrategyId::from("TEST-001")),
7144            order_id_tag: Some("001".to_string()),
7145            manage_stop: true,
7146            ..Default::default()
7147        };
7148        let mut strategy = TestStrategy::new(config);
7149        register_strategy(&mut strategy);
7150
7151        // State is not Running (default)
7152        assert!(!strategy.is_running());
7153
7154        let should_proceed = Strategy::stop(&mut strategy);
7155
7156        // Should proceed with stop
7157        assert!(should_proceed);
7158    }
7159
7160    #[rstest]
7161    fn test_finalize_market_exit_stops_strategy_when_pending() {
7162        let config = StrategyConfig {
7163            strategy_id: Some(StrategyId::from("TEST-001")),
7164            order_id_tag: Some("001".to_string()),
7165            ..Default::default()
7166        };
7167        let mut strategy = TestStrategy::new(config);
7168        register_strategy(&mut strategy);
7169        start_strategy(&mut strategy);
7170
7171        // Simulate a market exit with pending stop
7172        strategy.core.is_exiting = true;
7173        strategy.core.pending_stop = true;
7174
7175        strategy.finalize_market_exit();
7176
7177        // Should have transitioned to Stopped
7178        assert_eq!(strategy.state(), ComponentState::Stopped);
7179        assert!(!strategy.core.is_exiting);
7180        assert!(!strategy.core.pending_stop);
7181    }
7182
7183    #[rstest]
7184    fn test_finalize_market_exit_stays_running_when_not_pending() {
7185        let config = StrategyConfig {
7186            strategy_id: Some(StrategyId::from("TEST-001")),
7187            order_id_tag: Some("001".to_string()),
7188            ..Default::default()
7189        };
7190        let mut strategy = TestStrategy::new(config);
7191        register_strategy(&mut strategy);
7192        start_strategy(&mut strategy);
7193
7194        // Simulate a market exit without pending stop
7195        strategy.core.is_exiting = true;
7196        strategy.core.pending_stop = false;
7197
7198        strategy.finalize_market_exit();
7199
7200        // Should stay Running
7201        assert_eq!(strategy.state(), ComponentState::Running);
7202        assert!(!strategy.core.is_exiting);
7203    }
7204
7205    #[rstest]
7206    fn test_submit_order_denied_during_market_exit_when_not_reduce_only() {
7207        let mut strategy = create_test_strategy();
7208        register_strategy(&mut strategy);
7209        start_strategy(&mut strategy);
7210        strategy.core.is_exiting = true;
7211
7212        let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
7213            get_typed_message_saving_handler(Some(Ustr::from("events.order.denied")));
7214        let order = OrderAny::Market(MarketOrder::new(
7215            TraderId::from("TRADER-001"),
7216            StrategyId::from("TEST-001"),
7217            InstrumentId::from("BTCUSDT.BINANCE"),
7218            ClientOrderId::from("O-20250208-0001"),
7219            OrderSide::Buy,
7220            Quantity::from(100_000),
7221            TimeInForce::Gtc,
7222            UUID4::new(),
7223            UnixNanos::default(),
7224            false, // not reduce_only
7225            false,
7226            None,
7227            None,
7228            None,
7229            None,
7230            None,
7231            None,
7232            None,
7233            None,
7234        ));
7235        let topic = format!("events.order.{}", order.strategy_id());
7236        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
7237        let client_order_id = order.client_order_id();
7238        let result = strategy.submit_order(order.clone(), None, None, None);
7239
7240        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
7241
7242        assert!(result.is_ok());
7243        let cache = strategy.cache();
7244        let cached_order = cache.order(&client_order_id).unwrap();
7245        assert_eq!(cached_order.status(), OrderStatus::Denied);
7246
7247        let event_messages = event_messages.get_messages();
7248        assert_eq!(event_messages.len(), 2);
7249        assert_eq!(
7250            event_messages[0],
7251            OrderEventAny::Initialized(order.init_event().clone())
7252        );
7253        let OrderEventAny::Denied(denied) = &event_messages[1] else {
7254            panic!("expected OrderDenied event");
7255        };
7256        assert_eq!(denied.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
7257    }
7258
7259    #[rstest]
7260    fn test_submit_order_list_denied_during_market_exit_publishes_init_then_denied_events() {
7261        let mut strategy = create_test_strategy();
7262        register_strategy(&mut strategy);
7263        start_strategy(&mut strategy);
7264        strategy.core.is_exiting = true;
7265
7266        let orders = vec![
7267            make_initialized_market_order("O-20250208-LIST-DENY-001"),
7268            make_initialized_market_order("O-20250208-LIST-DENY-002"),
7269        ];
7270        let client_order_id1 = orders[0].client_order_id();
7271        let client_order_id2 = orders[1].client_order_id();
7272        let cache_rc = strategy.core.cache_rc();
7273        let timeline = Rc::new(RefCell::new(Vec::new()));
7274        let event_messages = Rc::new(RefCell::new(Vec::new()));
7275
7276        let event_handler = {
7277            let event_messages = event_messages.clone();
7278            let timeline = timeline.clone();
7279            TypedHandler::from_with_id("events.order.list_denied", move |event: &OrderEventAny| {
7280                match event {
7281                    OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
7282                        assert!(cache_rc.borrow().order_exists(&client_order_id1));
7283                        timeline.borrow_mut().push("init1");
7284                    }
7285                    OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
7286                        assert!(cache_rc.borrow().order_exists(&client_order_id2));
7287                        timeline.borrow_mut().push("init2");
7288                    }
7289                    OrderEventAny::Denied(e) if e.client_order_id == client_order_id1 => {
7290                        assert_eq!(e.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
7291                        let cache = cache_rc.borrow();
7292                        let cached_order = cache.order(&client_order_id1).unwrap();
7293                        assert_eq!(cached_order.status(), OrderStatus::Denied);
7294                        timeline.borrow_mut().push("denied1");
7295                    }
7296                    OrderEventAny::Denied(e) if e.client_order_id == client_order_id2 => {
7297                        assert_eq!(e.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
7298                        let cache = cache_rc.borrow();
7299                        let cached_order = cache.order(&client_order_id2).unwrap();
7300                        assert_eq!(cached_order.status(), OrderStatus::Denied);
7301                        timeline.borrow_mut().push("denied2");
7302                    }
7303                    _ => panic!("unexpected order event {event:?}"),
7304                }
7305                event_messages.borrow_mut().push(event.clone());
7306            })
7307        };
7308        let risk_handler = {
7309            let timeline = timeline.clone();
7310            TypedIntoHandler::from_with_id(
7311                "RiskEngine.queue_execute",
7312                move |_command: TradingCommand| {
7313                    timeline.borrow_mut().push("command");
7314                },
7315            )
7316        };
7317        msgbus::register_trading_command_endpoint(
7318            MessagingSwitchboard::risk_engine_queue_execute(),
7319            risk_handler,
7320        );
7321
7322        let topic = format!("events.order.{}", orders[0].strategy_id());
7323        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
7324        let result = strategy.submit_order_list(orders.clone(), None, None, None);
7325
7326        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
7327
7328        assert!(result.is_ok());
7329
7330        let cache = strategy.cache();
7331        let cached_order1 = cache.order(&client_order_id1).unwrap();
7332        let cached_order2 = cache.order(&client_order_id2).unwrap();
7333        assert_eq!(cached_order1.status(), OrderStatus::Denied);
7334        assert_eq!(cached_order2.status(), OrderStatus::Denied);
7335
7336        let event_messages = event_messages.borrow();
7337        assert_eq!(event_messages.len(), 4);
7338        assert_eq!(
7339            event_messages[0],
7340            OrderEventAny::Initialized(orders[0].init_event().clone())
7341        );
7342        assert!(matches!(
7343            &event_messages[1],
7344            OrderEventAny::Denied(e)
7345                if e.client_order_id == client_order_id1
7346                    && e.reason == Ustr::from("MARKET_EXIT_IN_PROGRESS")
7347        ));
7348        assert_eq!(
7349            event_messages[2],
7350            OrderEventAny::Initialized(orders[1].init_event().clone())
7351        );
7352        assert!(matches!(
7353            &event_messages[3],
7354            OrderEventAny::Denied(e)
7355                if e.client_order_id == client_order_id2
7356                    && e.reason == Ustr::from("MARKET_EXIT_IN_PROGRESS")
7357        ));
7358        assert_eq!(
7359            timeline.borrow().as_slice(),
7360            &["init1", "denied1", "init2", "denied2"]
7361        );
7362    }
7363
7364    #[rstest]
7365    fn test_submit_order_list_market_exit_rejects_non_initialized_without_events() {
7366        let mut strategy = create_test_strategy();
7367        register_strategy(&mut strategy);
7368        start_strategy(&mut strategy);
7369        strategy.core.is_exiting = true;
7370
7371        let order = make_accepted_market_order("O-20250208-LIST-DENY-ACCEPTED");
7372        let topic = format!("events.order.{}", order.strategy_id());
7373        let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
7374            get_typed_message_saving_handler(Some(Ustr::from("events.order.list_invalid")));
7375
7376        msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
7377        let result = strategy.submit_order_list(vec![order], None, None, None);
7378
7379        msgbus::unsubscribe_order_events(topic.into(), &event_handler);
7380
7381        assert!(result.is_err());
7382        assert!(
7383            result
7384                .unwrap_err()
7385                .to_string()
7386                .contains("expected INITIALIZED")
7387        );
7388        assert!(event_messages.get_messages().is_empty());
7389    }
7390
7391    #[rstest]
7392    fn test_submit_order_list_rejects_mixed_venues_with_friendly_error() {
7393        let mut strategy = create_test_strategy();
7394        register_strategy(&mut strategy);
7395        start_strategy(&mut strategy);
7396
7397        let binance_order = make_initialized_market_order("O-MIXED-VENUE-001");
7398        let bybit_order = OrderAny::Market(MarketOrder::new(
7399            TraderId::from("TRADER-001"),
7400            StrategyId::from("TEST-001"),
7401            InstrumentId::from("BTCUSDT.BYBIT"),
7402            ClientOrderId::from("O-MIXED-VENUE-002"),
7403            OrderSide::Buy,
7404            Quantity::from(100_000),
7405            TimeInForce::Gtc,
7406            UUID4::new(),
7407            UnixNanos::default(),
7408            false,
7409            false,
7410            None,
7411            None,
7412            None,
7413            None,
7414            None,
7415            None,
7416            None,
7417            None,
7418        ));
7419
7420        let result = strategy.submit_order_list(vec![binance_order, bybit_order], None, None, None);
7421
7422        let err = result.unwrap_err();
7423        let msg = err.to_string();
7424        assert!(
7425            msg.contains("OrderList denied: orders must share the same venue"),
7426            "unexpected error: {msg}",
7427        );
7428        assert!(msg.contains("BINANCE"), "expected BINANCE in error: {msg}");
7429        assert!(msg.contains("BYBIT"), "expected BYBIT in error: {msg}");
7430    }
7431
7432    #[rstest]
7433    fn test_submit_order_allowed_during_market_exit_when_reduce_only() {
7434        let mut strategy = create_test_strategy();
7435        register_strategy(&mut strategy);
7436        start_strategy(&mut strategy);
7437        strategy.core.is_exiting = true;
7438
7439        let order = OrderAny::Market(MarketOrder::new(
7440            TraderId::from("TRADER-001"),
7441            StrategyId::from("TEST-001"),
7442            InstrumentId::from("BTCUSDT.BINANCE"),
7443            ClientOrderId::from("O-20250208-0001"),
7444            OrderSide::Buy,
7445            Quantity::from(100_000),
7446            TimeInForce::Gtc,
7447            UUID4::new(),
7448            UnixNanos::default(),
7449            true, // reduce_only
7450            false,
7451            None,
7452            None,
7453            None,
7454            None,
7455            None,
7456            None,
7457            None,
7458            None,
7459        ));
7460        let client_order_id = order.client_order_id();
7461        let result = strategy.submit_order(order, None, None, None);
7462
7463        assert!(result.is_ok());
7464        let cache = strategy.cache();
7465        let cached_order = cache.order(&client_order_id).unwrap();
7466        assert_ne!(cached_order.status(), OrderStatus::Denied);
7467    }
7468
7469    #[rstest]
7470    fn test_submit_order_allowed_during_market_exit_when_tagged() {
7471        let mut strategy = create_test_strategy();
7472        register_strategy(&mut strategy);
7473        start_strategy(&mut strategy);
7474        strategy.core.is_exiting = true;
7475
7476        let order = OrderAny::Market(MarketOrder::new(
7477            TraderId::from("TRADER-001"),
7478            StrategyId::from("TEST-001"),
7479            InstrumentId::from("BTCUSDT.BINANCE"),
7480            ClientOrderId::from("O-20250208-0002"),
7481            OrderSide::Buy,
7482            Quantity::from(100_000),
7483            TimeInForce::Gtc,
7484            UUID4::new(),
7485            UnixNanos::default(),
7486            false, // not reduce_only
7487            false,
7488            None,
7489            None,
7490            None,
7491            None,
7492            None,
7493            None,
7494            None,
7495            Some(vec![Ustr::from("MARKET_EXIT")]),
7496        ));
7497        let client_order_id = order.client_order_id();
7498        let result = strategy.submit_order(order, None, None, None);
7499
7500        assert!(result.is_ok());
7501        let cache = strategy.cache();
7502        let cached_order = cache.order(&client_order_id).unwrap();
7503        assert_ne!(cached_order.status(), OrderStatus::Denied);
7504    }
7505
7506    #[derive(Debug)]
7507    struct MacroTestSimple {
7508        core: StrategyCore,
7509    }
7510
7511    nautilus_strategy!(MacroTestSimple);
7512
7513    impl DataActor for MacroTestSimple {}
7514
7515    #[derive(Debug)]
7516    struct MacroTestWithHooks {
7517        core: StrategyCore,
7518    }
7519
7520    nautilus_strategy!(MacroTestWithHooks, {
7521        fn on_order_rejected(&mut self, _event: OrderRejected) {}
7522    });
7523
7524    impl DataActor for MacroTestWithHooks {}
7525
7526    #[derive(Debug)]
7527    struct MacroTestCustomField {
7528        inner: StrategyCore,
7529    }
7530
7531    nautilus_strategy!(MacroTestCustomField, inner, {
7532        fn external_order_instrument_ids(&self) -> Option<Vec<InstrumentId>> {
7533            None
7534        }
7535    });
7536
7537    impl DataActor for MacroTestCustomField {}
7538
7539    #[rstest]
7540    fn test_strategy_behavior_does_not_require_native_core_access() {
7541        fn assert_strategy<T: Strategy + DataActor>() {}
7542
7543        assert_strategy::<CoreFreeStrategy>();
7544
7545        let mut strategy = CoreFreeStrategy { started: false };
7546        DataActor::on_start(&mut strategy).unwrap();
7547
7548        assert!(strategy.started);
7549    }
7550
7551    #[rstest]
7552    fn test_nautilus_strategy_macro_forms() {
7553        let config = StrategyConfig {
7554            strategy_id: Some(StrategyId::from("MACRO-001")),
7555            order_id_tag: Some("001".to_string()),
7556            ..Default::default()
7557        };
7558
7559        let simple = MacroTestSimple {
7560            core: StrategyCore::new(config.clone()),
7561        };
7562        assert_eq!(simple.strategy_id(), config.strategy_id);
7563        assert_eq!(simple.config().order_id_tag, config.order_id_tag);
7564        assert_eq!(simple.actor_id(), ActorId::from("MACRO-001"));
7565
7566        let hooks = MacroTestWithHooks {
7567            core: StrategyCore::new(config.clone()),
7568        };
7569        assert_eq!(hooks.strategy_id(), config.strategy_id);
7570        assert_eq!(hooks.config().order_id_tag, config.order_id_tag);
7571        assert_eq!(hooks.actor_id(), ActorId::from("MACRO-001"));
7572
7573        let custom = MacroTestCustomField {
7574            inner: StrategyCore::new(config.clone()),
7575        };
7576        assert_eq!(custom.strategy_id(), config.strategy_id);
7577        assert_eq!(custom.config().order_id_tag, config.order_id_tag);
7578        assert_eq!(custom.actor_id(), ActorId::from("MACRO-001"));
7579        assert!(custom.external_order_instrument_ids().is_none());
7580    }
7581
7582    #[rstest]
7583    fn test_cancel_all_orders_returns_all_errors_without_logging() {
7584        let _guard = init_logging(
7585            TraderId::from("TRADER-001"),
7586            UUID4::new(),
7587            Default::default(),
7588            Default::default(),
7589        )
7590        .unwrap();
7591        let mut strategy = create_test_strategy();
7592        register_strategy(&mut strategy);
7593        let (handler, messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
7594            get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
7595        msgbus::register_trading_command_endpoint(
7596            MessagingSwitchboard::exec_engine_queue_execute(),
7597            handler,
7598        );
7599        let mut first = make_accepted_market_order("O-CANCEL-001");
7600        let mut second = make_accepted_market_order("O-CANCEL-002");
7601        for order in [&mut first, &mut second] {
7602            let OrderAny::Market(order) = order else {
7603                unreachable!()
7604            };
7605
7606            order.account_id = None;
7607        }
7608
7609        let third = make_accepted_market_order("O-CANCEL-003");
7610        for order in [&first, &second, &third] {
7611            add_order_to_cache(&strategy, order);
7612        }
7613
7614        strategy.core.cache_rc().borrow_mut().build_index();
7615        arm_shutdown_on_error(true);
7616        let result = strategy.cancel_all_orders(first.instrument_id(), None, None, true, None);
7617        let trigger = take_shutdown_on_error_trigger();
7618        disarm_shutdown_on_error();
7619
7620        assert_eq!(
7621            result.unwrap_err().to_string(),
7622            "Cannot generate pending cancel event for O-CANCEL-001: account_id is not set; Cannot generate pending cancel event for O-CANCEL-002: account_id is not set"
7623        );
7624        assert_eq!(trigger, None);
7625        let messages = messages.get_messages();
7626        assert_eq!(messages.len(), 1);
7627        assert!(
7628            matches!(&messages[0], TradingCommand::CancelOrder(command) if command.client_order_id == third.client_order_id())
7629        );
7630        let cache = strategy.cache();
7631        assert_eq!(
7632            cache.order(&first.client_order_id()).unwrap().status(),
7633            OrderStatus::Accepted
7634        );
7635        assert_eq!(
7636            cache.order(&second.client_order_id()).unwrap().status(),
7637            OrderStatus::Accepted
7638        );
7639        assert_eq!(
7640            cache.order(&third.client_order_id()).unwrap().status(),
7641            OrderStatus::PendingCancel
7642        );
7643    }
7644}