✨: Add support for parallel optimization

Author: tmechsner

backtesting/backtesting.py

@@ -26,8 +26,6 @@
 from math import copysign
 from numpy.random import default_rng
 
-from .samplers import LatinHypercubeSampler
-
 try:
     from tqdm.auto import tqdm as _tqdm
     _tqdm = partial(_tqdm, leave=False)
@@ -1257,6 +1255,7 @@ def optimize(self, *,
                  return_optimization: bool = False,
                  random_state: Optional[int] = None,
                  n_initial_points: Optional[int] = None,
+                 n_workers: int = 1,
                  **kwargs) -> Union[pd.Series,
                                     Tuple[pd.Series, pd.Series],
                                     Tuple[pd.Series, pd.Series, dict]]:
@@ -1546,7 +1545,8 @@ def _optimize_openbox() -> Union[pd.Series,
                                        Tuple[pd.Series, pd.Series],
                                        Tuple[pd.Series, pd.Series, dict]]:
             try:
-                from openbox import space as sp, Optimizer
+                from openbox import space as sp, Optimizer, ParallelOptimizer
+                from .samplers import LatinHypercubeSampler
             except ImportError:
                 raise ImportError("Need package 'openbox' for method='openbox'. "
                                   "pip install openbox") from None
@@ -1609,20 +1609,31 @@ def eval_run(config: sp.Configuration):
                 logging.warning(f'Only {len(valid_initial_configs)}/{len(initial_configs)} valid configurations are generated for initial design strategy "Latin Hypercube". ')
                 num_random_config = n_initial_points - len(valid_initial_configs)
                 valid_initial_configs += Advisor.sample_random_configs(space, num_random_config, excluded_configs=valid_initial_configs)
-            opt = Optimizer(
-                eval_run,
-                space,
-                num_constraints=1,
-                num_objectives=1,
-                surrogate_type='auto',
-                acq_type='auto',
-                acq_optimizer_type='auto',
-                max_runs=max_tries,
-                task_id='soc',
-                random_state=random_state,
-                initial_configurations=valid_initial_configs,
-                initial_runs=initial_runs,
-            )
+
+            params = {
+                "objective_function": eval_run,
+                "config_space": space,
+                "num_constraints": 1,
+                "num_objectives": 1,
+                "surrogate_type": 'auto',
+                "acq_type": 'auto',
+                "acq_optimizer_type": 'auto',
+                "max_runs": max_tries,
+                "task_id": 'soc',
+                "random_state": random_state,
+                "initial_configurations": valid_initial_configs,
+                "initial_runs": initial_runs,
+            }
+
+            if n_workers == 1:
+                opt = Optimizer(**params)
+            else:
+                opt = ParallelOptimizer(
+                    parallel_strategy="async",
+                    batch_size=n_workers,
+                    batch_strategy="default",
+                    **params
+                )
             history = opt.run()
             optimal_configurations = history.get_incumbents()
             if not len(optimal_configurations):

backtesting/test/_test.py

@@ -9,7 +9,7 @@
 from glob import glob
 from runpy import run_path
 from tempfile import NamedTemporaryFile, gettempdir
-from time import time
+from time import time, sleep
 from unittest import TestCase
 from unittest.mock import patch
 
@@ -87,9 +87,9 @@ def test_run_invalid_param(self):
 
     def test_run_speed(self):
         bt = Backtest(GOOG, SmaCross)
-        start = time.process_time()
+        start = time()
         bt.run()
-        end = time.process_time()
+        end = time()
         self.assertLess(end - start, .3)
 
     def test_data_missing_columns(self):
@@ -581,6 +581,19 @@ def test_method_openbox(self):
             random_state=2)
         self.assertIsInstance(res, pd.Series)
 
+    def test_method_openbox_parallel(self):
+        bt = Backtest(GOOG.iloc[:100], SmaCross)
+        res = bt.optimize(
+            fast=range(2, 20), slow=np.arange(2, 20, dtype=object),
+            constraint=lambda p: p.fast < p.slow,
+            max_tries=30,
+            method='openbox',
+            return_optimization=False,
+            return_heatmap=False,
+            random_state=2,
+            n_workers=2)
+        self.assertIsInstance(res, pd.Series)
+
     def test_timing(self):
         bt = Backtest(GOOG.iloc[:100], SmaCross)
 
@@ -877,7 +890,7 @@ def test_plot_heatmaps(self):
 
             # Preview
             plot_heatmaps(heatmap, filename=f)
-            time.sleep(5)
+            sleep(5)
 
     def test_random_ohlc_data(self):
         generator = random_ohlc_data(GOOG, frac=1)