Refine exit process for Strategy

Author: faysou

nautilus_trader/trading/config.py

@@ -62,6 +62,9 @@ class StrategyConfig(NautilusConfig, kw_only=True, frozen=True):
         If commands should be logged by the strategy.
     log_rejected_due_post_only_as_warning : bool, default True
         If order rejected events where `due_post_only` is True should be logged as warnings.
+    inflight_check_interval_ms : int, default 100
+        The interval in milliseconds to check for in-flight orders and open positions
+        during a market exit.
 
     """
 
@@ -76,6 +79,7 @@ class StrategyConfig(NautilusConfig, kw_only=True, frozen=True):
     log_events: bool = True
     log_commands: bool = True
     log_rejected_due_post_only_as_warning: bool = True
+    inflight_check_interval_ms: int = 100
 
 
 class ImportableStrategyConfig(NautilusConfig, frozen=True):

nautilus_trader/trading/controller.py

@@ -29,6 +29,7 @@
 from nautilus_trader.trading.config import StrategyFactory
 from nautilus_trader.trading.messages import CreateActor
 from nautilus_trader.trading.messages import CreateStrategy
+from nautilus_trader.trading.messages import MarketExitStrategy
 from nautilus_trader.trading.messages import RemoveActor
 from nautilus_trader.trading.messages import RemoveStrategy
 from nautilus_trader.trading.messages import StartActor
@@ -93,6 +94,8 @@ def execute(self, command: Command) -> None:
             self.start_strategy_from_id(command.strategy_id)
         elif isinstance(command, StopStrategy):
             self.stop_strategy_from_id(command.strategy_id)
+        elif isinstance(command, MarketExitStrategy):
+            self.market_exit_strategy_from_id(command.strategy_id)
         elif isinstance(command, RemoveStrategy):
             self.remove_strategy_from_id(command.strategy_id)
 
@@ -210,6 +213,25 @@ def stop_strategy(self, strategy: Strategy) -> None:
         """
         self._trader.stop_strategy(strategy.id)
 
+    def market_exit_strategy(self, strategy: Strategy) -> None:
+        """
+        Market exit the given `strategy`.
+
+        Will log a warning if the strategy is not ``RUNNING``.
+
+        Parameters
+        ----------
+        strategy : Strategy
+            The strategy to market exit.
+
+        Raises
+        ------
+        ValueError
+            If `strategy` is not already registered with the trader.
+
+        """
+        self._trader.market_exit_strategy(strategy.id)
+
     def remove_actor(self, actor: Actor) -> None:
         """
         Remove the given `actor`.
@@ -383,6 +405,25 @@ def stop_strategy_from_id(self, strategy_id: StrategyId) -> None:
         """
         self._trader.stop_strategy(strategy_id)
 
+    def market_exit_strategy_from_id(self, strategy_id: StrategyId) -> None:
+        """
+        Market exit the strategy corresponding to `strategy_id`.
+
+        Will log a warning if the strategy is not ``RUNNING``.
+
+        Parameters
+        ----------
+        strategy_id : StrategyId
+            The ID of the strategy to market exit.
+
+        Raises
+        ------
+        ValueError
+            If `strategy` is not already registered with the trader.
+
+        """
+        self._trader.market_exit_strategy(strategy_id)
+
     def remove_actor_from_id(self, actor_id: ComponentId) -> None:
         """
         Remove the actor corresponding to `actor_id`.

nautilus_trader/trading/messages.py

@@ -235,3 +235,29 @@ def __init__(
         super().__init__(command_id or UUID4(), ts_init)
 
         self.strategy_id = strategy_id
+
+
+class MarketExitStrategy(Command):
+    """
+    Represents a command to exit the market for a strategy.
+
+    Parameters
+    ----------
+    strategy_id : StrategyId
+        The ID of the strategy to exit the market for.
+    command_id : UUID4
+        The command ID.
+    ts_init : int
+        UNIX timestamp (nanoseconds) when the object was initialized.
+
+    """
+
+    def __init__(
+        self,
+        strategy_id: StrategyId,
+        command_id: UUID4 | None = None,
+        ts_init: int = 0,
+    ) -> None:
+        super().__init__(command_id or UUID4(), ts_init)
+
+        self.strategy_id = strategy_id

nautilus_trader/trading/strategy.pxd

@@ -67,6 +67,7 @@ cdef class Strategy(Actor):
     cdef bint _log_events
     cdef bint _log_commands
     cdef bint _log_rejected_due_post_only_as_warning
+    cdef bint _is_exiting
 
     cdef readonly OrderFactory order_factory
     """The order factory for the strategy.\n\n:returns: `OrderFactory`"""
@@ -97,6 +98,9 @@ cdef class Strategy(Actor):
     )
     cpdef void change_id(self, StrategyId strategy_id)
     cpdef void change_order_id_tag(self, str order_id_tag)
+    cpdef void on_market_exit(self)
+    cpdef void after_market_exit(self)
+    cpdef void market_exit(self)
 
 # -- ABSTRACT METHODS -----------------------------------------------------------------------------
 
@@ -170,6 +174,7 @@ cdef class Strategy(Actor):
     cdef str _get_gtd_expiry_timer_name(self, ClientOrderId client_order_id)
     cdef void _set_gtd_expiry(self, Order order)
     cpdef void _expire_gtd_order(self, TimeEvent event)
+    cpdef void _check_market_exit(self, TimeEvent event)
 
 # -- EVENTS ---------------------------------------------------------------------------------------
 

nautilus_trader/trading/strategy.pyx

@@ -25,14 +25,18 @@ attempts to operate without a managing `Trader` instance.
 
 """
 
+import pandas as pd
+
 from nautilus_trader.trading.config import ImportableStrategyConfig
 from nautilus_trader.trading.config import StrategyConfig
+from nautilus_trader.trading.messages import MarketExitStrategy
 
 from libc.stdint cimport uint64_t
 
 from nautilus_trader.cache.base cimport CacheFacade
 from nautilus_trader.cache.cache cimport Cache
 from nautilus_trader.common.actor cimport Actor
+from nautilus_trader.common.component cimport CMD
 from nautilus_trader.common.component cimport EVT
 from nautilus_trader.common.component cimport RECV
 from nautilus_trader.common.component cimport Clock
@@ -163,6 +167,7 @@ cdef class Strategy(Actor):
         self.external_order_claims = self._parse_external_order_claims(config.external_order_claims)
         self.manage_contingent_orders = config.manage_contingent_orders
         self.manage_gtd_expiry = config.manage_gtd_expiry
+        self._is_exiting = False
 
         # Public components
         self.clock = self._clock
@@ -243,6 +248,29 @@ cdef class Strategy(Actor):
             "occur here, such as resetting indicators and other state"
         )
 
+    cpdef void on_market_exit(self):
+        """
+        Actions to be performed when a market exit has been initiated.
+
+        Warnings
+        --------
+        Override this method in a subclass to implement custom market exit logic.
+
+        """
+        # Optionally override in subclass
+
+    cpdef void after_market_exit(self):
+        """
+        Actions to be performed after a market exit has been completed.
+
+        Warnings
+        --------
+        Override this method in a subclass to implement custom logic after
+        market exit.
+
+        """
+        # Optionally override in subclass
+
 # -- REGISTRATION ---------------------------------------------------------------------------------
 
     cpdef void register(
@@ -1664,7 +1692,77 @@ cdef class Strategy(Actor):
         self._log.info(f"Expiring GTD order {order.client_order_id}", LogColor.BLUE)
         self.cancel_order(order)
 
-    # -- HANDLERS -------------------------------------------------------------------------------------
+    cpdef void market_exit(self):
+        """
+        Initiate an iterative market exit for the strategy.
+
+        Will cancel all open orders and close all open positions, and wait for
+        all in-flight orders to resolve and positions to close before stopping
+         the strategy.
+        """
+        if self._is_exiting:
+            return
+
+        self._is_exiting = True
+
+        self._log.info("Initiating market exit...", LogColor.BLUE)
+        self.on_market_exit()
+
+        # Get all instruments the strategy has open orders or positions for
+        cdef list open_orders = self.cache.orders_open(None, None, self.id)
+        cdef list open_positions = self.cache.positions_open(None, None, self.id)
+
+        cdef set instruments = set()
+        cdef Order order
+        for order in open_orders:
+            instruments.add(order.instrument_id)
+
+        cdef Position position
+        for position in open_positions:
+            instruments.add(position.instrument_id)
+
+        cdef InstrumentId instrument_id
+        for instrument_id in instruments:
+            self.cancel_all_orders(instrument_id)
+            self.close_all_positions(instrument_id)
+
+        # Start iterative check
+        self._log.warning(f"Setting market exit timer for {self.id}")
+        self._clock.set_timer(
+            f"MARKET-EXIT-CHECK:{self.id}",
+            pd.Timedelta(milliseconds=self.config.inflight_check_interval_ms),
+            None,
+            None,
+            self._check_market_exit,
+            True,
+            False,
+        )
+
+    cpdef void _check_market_exit(self, TimeEvent event):
+        if self.state != ComponentState.RUNNING:
+            return
+
+        self._log.warning(f"Timer triggered: {event.name}")
+        cdef list open_orders = self.cache.orders_open(None, None, self.id)
+        cdef list inflight_orders = self.cache.orders_inflight(None, None, self.id)
+
+        if open_orders or inflight_orders:
+            return
+
+        cdef list open_positions = self.cache.positions_open(None, None, self.id)
+        if open_positions:
+            # If there are open positions but no orders, we should re-send close orders
+            for position in open_positions:
+                self.close_position(position)
+
+            return
+
+        # All clear
+        if f"MARKET-EXIT-CHECK:{self.id}" in self._clock.timer_names:
+            self._clock.cancel_timer(name=f"MARKET-EXIT-CHECK:{self.id}")
+
+        self.after_market_exit()
+        self.stop()
 
     cpdef void handle_event(self, Event event):
         """

nautilus_trader/trading/trader.py

@@ -635,6 +635,34 @@ def stop_strategy(self, strategy_id: StrategyId) -> None:
 
         strategy.stop()
 
+    def market_exit_strategy(self, strategy_id: StrategyId) -> None:
+        """
+        Market exit the strategy with the given `strategy_id`.
+
+        Parameters
+        ----------
+        strategy_id : StrategyId
+            The strategy ID to market exit.
+
+        Raises
+        ------
+        ValueError
+            If a strategy with the given `strategy_id` is not found.
+
+        """
+        PyCondition.not_none(strategy_id, "strategy_id")
+
+        strategy = self._strategies.get(strategy_id)
+
+        if strategy is None:
+            raise ValueError(f"Cannot market exit strategy, {strategy_id} not found.")
+
+        if not strategy.is_running:
+            self._log.warning(f"Strategy {strategy_id} not running")
+            return
+
+        strategy.market_exit()
+
     def remove_actor(self, actor_id: ComponentId) -> None:
         """
         Remove the actor with the given `actor_id`.