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