Iteration with Single-Step Progression to Improve Flexibility and Adaptability in Algorithmic Strategies

backtesting/backtesting.py

@@ -1662,3 +1662,87 @@ def plot(self, *, results: pd.Series = None, filename=None, plot_width=None,
             reverse_indicators=reverse_indicators,
             show_legend=show_legend,
             open_browser=open_browser)
+
+    def initialize(self, **kwargs) -> None:
+        """
+        Initialize the backtest with the given parameters. Keyword arguments are interpreted as strategy parameters.
+
+        :param kwargs: Strategy parameters
+        """
+        data = _Data(self._data.copy(deep=False))
+        broker: _Broker = self._broker(data=data)
+        strategy: Strategy = self._strategy(broker, data, kwargs)
+
+        strategy.init()
+        data._update()  # Strategy.init might have changed/added to data.df
+
+        # Indicators used in Strategy.next()
+        indicator_attrs = {attr: indicator
+                        for attr, indicator in strategy.__dict__.items()
+                        if isinstance(indicator, _Indicator)}.items()
+
+        # Skip first few candles where indicators are still "warming up"
+        # +1 to have at least two entries available
+        start = 1 + max((np.isnan(indicator.astype(float)).argmin(axis=-1).max()
+                        for _, indicator in indicator_attrs), default=0)
+
+        self._step_data = data
+        self._step_broker = broker
+        self._step_strategy = strategy
+        self._step_time = start
+        self._step_indicator_attrs = indicator_attrs
+
+    def next(self, done:bool|None=None, **kwargs) -> None|pd.Series:
+        """
+        Move the backtest one time step forward and return the results for the current step.
+
+        :return: Results and statistics for the current time step
+        :rtype: pd.Series
+        """
+
+        # Disable "invalid value encountered in ..." warnings. Comparison
+        # np.nan >= 3 is not invalid; it's False.
+        # with np.errstate(invalid='ignore'):
+
+        if self._step_time < len(self._data):
+            # Prepare data and indicators for `next` call
+            self._step_data._set_length(self._step_time + 1)
+            for attr, indicator in self._step_indicator_attrs:
+                # Slice indicator on the last dimension (case of 2d indicator)
+                setattr(self._step_strategy, attr, indicator[..., :self._step_time + 1])
+
+            # Handle orders processing and broker stuff
+            try:
+                self._step_broker.next()
+            except _OutOfMoneyError:
+                pass
+
+            # Next tick, a moment before bar close
+            # passing kwargs to be used in the strategy class
+            self._step_strategy.next(**kwargs)
+            self._step_time += 1
+
+        if done==True:
+            # Close any remaining open trades so they produce some stats
+            for trade in self._step_broker.trades:
+                trade.close()
+
+            # Re-run broker one last time to handle orders placed in the last strategy
+            # iteration. Use the same OHLC values as in the last broker iteration.
+            if self._step_time < len(self._data):
+                try_(self._step_broker.next, exception=_OutOfMoneyError)
+
+            # Set data back to full length
+            # for future `indicator._opts['data'].index` calls to work
+            self._step_data._set_length(len(self._data))
+
+
+            equity = pd.Series(self._step_broker._equity).bfill().fillna(self._step_broker._cash).values
+            results = compute_stats(
+                trades=self._step_broker.closed_trades,
+                equity=equity,
+                ohlc_data=self._data,
+                risk_free_rate=0.0,
+                strategy_instance=self._step_strategy,
+            )
+            return results

backtesting/test/_iteration.py

@@ -0,0 +1,35 @@
+from backtesting import Backtest
+from backtesting import Strategy
+from backtesting.test import GOOG, SMA
+from backtesting.lib import crossover
+import types
+import random
+random.seed(0)
+
+class TestStrategy(Strategy):
+    def init(self):
+        print("Init", self.equity)
+
+    def next(self, action=None):
+        # uncomment if you want to test run()
+        # if not action:
+        #     action = random.randint(0, 1)
+        if action!=None:
+            if action == 0:
+                self.buy()
+            elif action == 1:
+                self.position.close()
+
+
+bt = Backtest(GOOG, TestStrategy, cash=10_000, commission=.002)
+
+# stats = bt.run()
+
+bt.initialize()
+while True:
+    action = random.randint(0, 1)
+    stats = bt.next(action=action)
+    if not isinstance(stats, types.NoneType):
+        break
+print(stats)
+bt.plot(results=stats, open_browser=True)