ENH: Add exit_tag support for trades and positions to indicate closur…

CHANGELOG.md

@@ -5,6 +5,10 @@ These were the major changes contributing to each release:
 
 ### 0.x.x
 
+* Enhancement: `Position.close()`/`Trade.close()` accept a `tag=` kwarg; SL/TP-triggered
+  closes are auto-tagged `"sl"`/`"tp"`; new `Trade.exit_tag` property and `stats._trades`
+  `'ExitTag'` column expose why a trade was closed (#1352)
+
 ### 0.6.6
 (2026-07-22)
 

backtesting/_stats.py

@@ -74,6 +74,7 @@ def compute_stats(
         })
         trades_df['Duration'] = trades_df['ExitTime'] - trades_df['EntryTime']
         trades_df['Tag'] = [t.tag for t in trades]
+        trades_df['ExitTag'] = [t.exit_tag for t in trades]
 
         # Add indicator values
         if len(trades_df) and strategy_instance:

backtesting/backtesting.py

@@ -367,12 +367,15 @@ def is_short(self) -> bool:
         """True if the position is short (position size is negative)."""
         return self.size < 0
 
-    def close(self, portion: float = 1.):
+    def close(self, portion: float = 1., *, tag=None):
         """
         Close portion of position by closing `portion` of each active trade. See `Trade.close`.
+
+        If `tag` is given, it is passed through to each `Trade.close()` call,
+        ending up as `Trade.exit_tag` of every closed trade.
         """
         for trade in self.__broker.trades:
-            trade.close(portion)
+            trade.close(portion, tag=tag)
 
     def __repr__(self):
         return f'<Position: {self.size} ({len(self.__broker.trades)} trades)>'
@@ -555,6 +558,7 @@ def __init__(self, broker: '_Broker', size: int, entry_price: float, entry_bar,
         self.__sl_order: Optional[Order] = None
         self.__tp_order: Optional[Order] = None
         self.__tag = tag
+        self.__exit_tag = None
         self._commissions = 0
 
     def __repr__(self):
@@ -570,12 +574,20 @@ def _replace(self, **kwargs):
     def _copy(self, **kwargs):
         return copy(self)._replace(**kwargs)
 
-    def close(self, portion: float = 1.):
-        """Place new `Order` to close `portion` of the trade at next market price."""
+    def close(self, portion: float = 1., *, tag=None):
+        """
+        Place new `Order` to close `portion` of the trade at next market price.
+
+        If `tag` is given, it is used as the tag of the closing `Order` and ends up as
+        `Trade.exit_tag` of this trade once closed, so it can be used to record *why*
+        the trade was closed. If not given, the closing order (and hence `Trade.exit_tag`)
+        falls back to this trade's opening `Trade.tag`, preserving prior behavior.
+        """
         assert 0 < portion <= 1, "portion must be a fraction between 0 and 1"
         # Ensure size is an int to avoid rounding errors on 32-bit OS
         size = copysign(max(1, int(round(abs(self.__size) * portion))), -self.__size)
-        order = Order(self.__broker, size, parent_trade=self, tag=self.__tag)
+        order = Order(self.__broker, size, parent_trade=self,
+                      tag=(self.__tag if tag is None else tag))
         self.__broker.orders.insert(0, order)
 
     # Fields getters
@@ -621,6 +633,21 @@ def tag(self):
         """
         return self.__tag
 
+    @property
+    def exit_tag(self):
+        """
+        A tag value indicating why/how the trade was closed, or `None` while the
+        trade is still active.
+
+        Unlike `Trade.tag` (the *opening* tag, fixed for the life of the trade),
+        `exit_tag` reflects the *closing* order: `"sl"` or `"tp"` when the trade
+        was closed automatically by its stop-loss or take-profit order, the
+        `tag=` passed to `Trade.close()` / `Position.close()` when closed
+        explicitly, or `None` for a plain close with no tag given (including
+        trades closed as a side effect of an opposing order filling).
+        """
+        return self.__exit_tag
+
     @property
     def _sl_order(self):
         return self.__sl_order
@@ -934,9 +961,19 @@ def _process_orders(self):
                 # If order.size is "greater" than trade.size, this order is a trade.close()
                 # order and part of the trade was already closed beforehand
                 size = copysign(min(abs(_prev_size), abs(order.size)), order.size)
+                # Determine why the trade is being closed, for `Trade.exit_tag`.
+                # SL/TP orders are auto-tagged "sl"/"tp" unless their tag was
+                # explicitly set to something other than the trade's opening tag.
+                if order is trade._sl_order:
+                    exit_tag = 'sl' if order.tag == trade.tag else order.tag
+                elif order is trade._tp_order:
+                    exit_tag = 'tp' if order.tag == trade.tag else order.tag
+                else:
+                    # It's a trade.close()/position.close() order
+                    exit_tag = order.tag
                 # If this trade isn't already closed (e.g. on multiple `trade.close(.5)` calls)
                 if trade in self.trades:
-                    self._reduce_trade(trade, price, size, time_index)
+                    self._reduce_trade(trade, price, size, time_index, exit_tag)
                     assert order.size != -_prev_size or trade not in self.trades
                     if order is trade._sl_order:
                         # Set SL back on the order for stats._trades["SL"]
@@ -1052,7 +1089,8 @@ def _process_orders(self):
         if reprocess_orders:
             self._process_orders()
 
-    def _reduce_trade(self, trade: Trade, price: float, size: float, time_index: int):
+    def _reduce_trade(self, trade: Trade, price: float, size: float, time_index: int,
+                      exit_tag=None):
         assert trade.size * size < 0
         assert abs(trade.size) >= abs(size)
         self._trades_cache_clear()
@@ -1073,17 +1111,17 @@ def _reduce_trade(self, trade: Trade, price: float, size: float, time_index: int
             close_trade = trade._copy(size=-size, sl_order=None, tp_order=None)
             self.trades.append(close_trade)
 
-        self._close_trade(close_trade, price, time_index)
+        self._close_trade(close_trade, price, time_index, exit_tag)
 
-    def _close_trade(self, trade: Trade, price: float, time_index: int):
+    def _close_trade(self, trade: Trade, price: float, time_index: int, exit_tag=None):
         self._trades_cache_clear()
         self.trades.remove(trade)
         if trade._sl_order:
             self.orders.remove(trade._sl_order)
         if trade._tp_order:
             self.orders.remove(trade._tp_order)
 
-        closed_trade = trade._replace(exit_price=price, exit_bar=time_index)
+        closed_trade = trade._replace(exit_price=price, exit_bar=time_index, exit_tag=exit_tag)
         self.closed_trades.append(closed_trade)
         # Apply commission one more time at trade exit
         commission = self._commission(trade.size, price)

backtesting/test/_test.py

@@ -387,7 +387,7 @@ def almost_equal(a, b):
             sorted(stats['_trades'].columns),
             sorted(['Size', 'EntryBar', 'ExitBar', 'EntryPrice', 'ExitPrice',
                     'SL', 'TP', 'PnL', 'ReturnPct', 'EntryTime', 'ExitTime',
-                    'Duration', 'Tag', 'Commission',
+                    'Duration', 'Tag', 'ExitTag', 'Commission',
                     *indicator_columns]))
 
     def test_compute_stats_bordercase(self):
@@ -441,6 +441,18 @@ def next(self):
         with self.assertWarns(UserWarning):
             self.assertEqual(Backtest(GOOG, S).run()._trades.iloc[0].ExitPrice, 705.58)
 
+    def test_trade_sl_hit_sets_exit_tag(self):
+        the_day = pd.Timestamp("2012-10-17 00:00:00")
+
+        class S(_S):
+            def next(self):
+                if self.data.index[-1] == the_day:
+                    self.buy(sl=720)
+
+        trades = Backtest(GOOG, S).run()._trades
+        self.assertEqual(trades.iloc[0].ExitPrice, 720)
+        self.assertEqual(trades.iloc[0].ExitTag, 'sl')
+
     def test_stop_price_between_sl_tp(self):
         class S(_S):
             def next(self):
@@ -591,6 +603,55 @@ def coroutine(self):
         stats = self._Backtest(coroutine).run()
         self.assertEqual(list(stats._trades.Tag), [1, 1, 2])
 
+    def test_trade_close_exit_tag(self):
+        def coroutine(self):
+            yield self.buy(size=1)
+            self.trades[-1].close(tag='my reason')
+            yield
+
+        stats = self._Backtest(coroutine).run()
+        self.assertEqual(list(stats._trades.ExitTag), ['my reason'])
+
+    def test_position_close_exit_tag(self):
+        def coroutine(self):
+            yield self.buy(size=1)
+            self.position.close(tag='my reason')
+            yield
+
+        stats = self._Backtest(coroutine).run()
+        self.assertEqual(list(stats._trades.ExitTag), ['my reason'])
+
+    def test_close_without_tag_falls_back_to_opening_tag(self):
+        # Backward compatibility: a close with no tag= given reuses the trade's
+        # opening tag for the closing order (and hence for `Trade.exit_tag`), same
+        # as before `tag=` existed.
+        def coroutine(self):
+            yield self.buy(size=1, tag='open-reason')
+            self.position.close()
+            yield
+
+        stats = self._Backtest(coroutine).run()
+        self.assertEqual(list(stats._trades.Tag), ['open-reason'])
+        self.assertEqual(list(stats._trades.ExitTag), ['open-reason'])
+
+    def test_position_close_partial_exit_tag(self):
+        def coroutine(self):
+            yield self.buy(size=10)
+            assert self.trades
+            self.position.close(portion=.5, tag='half')
+            yield
+            assert len(self.trades) == 1
+            assert self.trades[0].size == 5
+            assert self.trades[0].exit_tag is None
+            yield
+
+        with self.assertWarnsRegex(UserWarning, 'finalize_trades'):
+            stats = self._Backtest(coroutine, finalize_trades=False).run()
+        trades = stats._trades
+        self.assertEqual(len(trades), 1)
+        self.assertEqual(trades.iloc[0].Size, 5)
+        self.assertEqual(trades.iloc[0].ExitTag, 'half')
+
 
 class TestOptimize(TestCase):
     def test_optimize(self):
@@ -1266,6 +1327,23 @@ def next(self):
         trades = Backtest(SHORT_DATA, S).run()._trades
         self.assertEqual(trades['ExitBar'].iloc[0], 3)
         self.assertEqual(trades['ExitPrice'].iloc[0], 105)
+        self.assertEqual(trades['ExitTag'].iloc[0], 'tp')
+
+    def test_exit_tag_on_opposing_order_close(self):
+        # A trade closed as a side effect of an opposing order filling (FIFO close),
+        # with no explicit `trade.close()`/`position.close()` call anywhere, gets no
+        # exit tag (`None`) by default.
+        class S(_S):
+            def next(self):
+                i = len(self.data.index)
+                if i == 3:
+                    self.buy(size=1)
+                elif i == 4:
+                    self.sell(size=1)
+
+        trades = Backtest(SHORT_DATA, S).run()._trades
+        self.assertEqual(len(trades), 1)
+        self.assertIsNone(trades['ExitTag'].iloc[0])
 
     def test_optimize_datetime_index_with_timezone(self):
         data: pd.DataFrame = GOOG.iloc[:100]