Add tests for BayesianOptimizerController and fix verbosity

Author: peanutfun

climada/util/calibrate/bayesian_optimizer.py

@@ -29,7 +29,7 @@
 import numpy as np
 import matplotlib.pyplot as plt
 import matplotlib.axes as maxes
-from bayes_opt import BayesianOptimization, Events, UtilityFunction
+from bayes_opt import BayesianOptimization, Events, UtilityFunction, ScreenLogger
 from bayes_opt.target_space import TargetSpace
 
 from .base import Input, Output, Optimizer, OutputEvaluator
@@ -237,8 +237,8 @@ class BayesianOptimizerController(object):
     """
 
     # Init attributes
-    init_points: int = 0
-    n_iter: int = 0
+    init_points: int
+    n_iter: int
     min_improvement: float = 1e-3
     min_improvement_count: int = 2
     kappa: float = 2.576
@@ -256,9 +256,13 @@ class BayesianOptimizerController(object):
 
     def __post_init__(self):
         """Set the decay factor for :py:attr:`kappa`."""
-        self.kappa_decay = np.exp(
-            (np.log(self.kappa_min) - np.log(self.kappa)) / self.n_iter
-        )
+        if self.init_points < 0 or self.n_iter < 0:
+            raise ValueError("'init_points' and 'n_iter' must be 0 or positive")
+        self.kappa_decay = self._calc_kappa_decay()
+
+    def _calc_kappa_decay(self):
+        """Compute the decay factor for :py:attr:`kappa`."""
+        return np.exp((np.log(self.kappa_min) - np.log(self.kappa)) / self.n_iter)
 
     @classmethod
     def from_input(cls, inp: Input, sampling_base: float = 4, **kwargs):
@@ -291,10 +295,6 @@ def _previous_max(self):
             return -np.inf
         return self._improvements[-1].target
 
-    def is_converged(self) -> bool:
-        """Check if convergence criteria are met"""
-        return True
-
     def optimizer_params(self) -> dict[str, Union[int, float, str, UtilityFunction]]:
         """Return parameters for the optimizer"""
         return {
@@ -365,7 +365,7 @@ def _maybe_stop_early(self, instance):
             instance.dispatch(Events.OPTIMIZATION_END)
             raise StopEarly()
 
-    def update(self, event, instance):
+    def update(self, event: str, instance: BayesianOptimization):
         """Update the step tracker of this instance.
 
         For step events, check if the latest guess is the new maximum. Also check if the
@@ -441,7 +441,8 @@ class BayesianOptimizer(Optimizer):
     input : Input
         The input data for this optimizer. See the Notes below for input requirements.
     verbose : int, optional
-        Verbosity of the optimizer output. Defaults to 1.
+        Verbosity of the optimizer output. Defaults to 0. The output is *not* affected
+        by the CLIMADA logging settings.
     random_state : int, optional
         Seed for initializing the random number generator. Defaults to 1.
     allow_duplicate_points : bool, optional
@@ -472,14 +473,12 @@ class BayesianOptimizer(Optimizer):
         The optimizer instance of this class.
     """
 
-    verbose: InitVar[int] = 1
+    verbose: int = 0
     random_state: InitVar[int] = 1
     allow_duplicate_points: InitVar[bool] = True
     bayes_opt_kwds: InitVar[Optional[Mapping[str, Any]]] = None
 
-    def __post_init__(
-        self, verbose, random_state, allow_duplicate_points, bayes_opt_kwds
-    ):
+    def __post_init__(self, random_state, allow_duplicate_points, bayes_opt_kwds):
         """Create optimizer"""
         if bayes_opt_kwds is None:
             bayes_opt_kwds = {}
@@ -491,7 +490,6 @@ def __post_init__(
             f=self._opt_func,
             pbounds=self.input.bounds,
             constraint=self.input.constraints,
-            verbose=verbose,
             random_state=random_state,
             allow_duplicate_points=allow_duplicate_points,
             **bayes_opt_kwds,
@@ -501,10 +499,7 @@ def _target_func(self, data: pd.DataFrame, predicted: pd.DataFrame) -> Number:
         """Invert the cost function because BayesianOptimization maximizes the target"""
         return -self.input.cost_func(data, predicted)
 
-    def run(
-        self,
-        controller: BayesianOptimizerController,
-    ) -> BayesianOptimizerOutput:
+    def run(self, controller: BayesianOptimizerController) -> BayesianOptimizerOutput:
         """Execute the optimization
 
         ``BayesianOptimization`` *maximizes* a target function. Therefore, this class
@@ -524,14 +519,29 @@ def run(
             Optimization output. :py:attr:`BayesianOptimizerOutput.p_space` stores data
             on the sampled parameter space.
         """
+        # Register the controller
         for event in (Events.OPTIMIZATION_STEP, Events.OPTIMIZATION_END):
             self.optimizer.subscribe(event, controller)
 
+        # Register the logger
+        if self.verbose > 0:
+            log = ScreenLogger(
+                verbose=self.verbose, is_constrained=self.optimizer.is_constrained
+            )
+            for event in (
+                Events.OPTIMIZATION_START,
+                Events.OPTIMIZATION_STEP,
+                Events.OPTIMIZATION_END,
+            ):
+                self.optimizer.subscribe(event, log)
+
+        # Run the optimization
         while controller.iterations < controller.max_iterations:
             try:
                 LOGGER.info(f"Optimization iteration: {controller.iterations}")
                 self.optimizer.maximize(**controller.optimizer_params())
             except StopEarly:
+                # Start a new iteration
                 continue
             except StopIteration:
                 # Exit the loop

climada/util/calibrate/test/test_bayesian_optimizer.py

@@ -21,10 +21,18 @@
 import unittest
 from unittest.mock import patch, MagicMock
 
+import numpy as np
 import numpy.testing as npt
 import pandas as pd
+from bayes_opt import BayesianOptimization, UtilityFunction, Events
+from scipy.optimize import NonlinearConstraint
 
 from climada.util.calibrate import Input, BayesianOptimizer, BayesianOptimizerController
+from climada.util.calibrate.bayesian_optimizer import (
+    Improvement,
+    StopEarly,
+    BayesianOptimizerOutput,
+)
 
 from .test_base import hazard, exposure
 
@@ -44,12 +52,26 @@ def input():
 class TestBayesianOptimizerController(unittest.TestCase):
     """Tests for the controller of the BayesianOptimizer"""
 
+    def setUp(self):
+        """Create a optimization instance"""
+        self.bayes_opt = BayesianOptimization(
+            f=lambda x: -(x**2),
+            pbounds={"x": (-10, 10)},
+            constraint=NonlinearConstraint(fun=lambda x: x, lb=-0.5, ub=np.inf),
+            verbose=0,
+            allow_duplicate_points=True,
+        )
+
+    def _make_step(self, x, controller):
+        self.bayes_opt.probe({"x": x}, lazy=False)
+        controller.update(Events.OPTIMIZATION_STEP, self.bayes_opt)
+
     def test_kappa_decay(self):
         """Check correct values for kappa_decay"""
-        contr = BayesianOptimizerController(kappa=3, kappa_min=3, n_iter=10)
+        contr = BayesianOptimizerController(0, kappa=3, kappa_min=3, n_iter=10)
         self.assertAlmostEqual(contr.kappa_decay, 1.0)
 
-        contr = BayesianOptimizerController(kappa=3, kappa_min=1, n_iter=10)
+        contr = BayesianOptimizerController(0, kappa=3, kappa_min=1, n_iter=10)
         self.assertAlmostEqual(contr.kappa * (contr.kappa_decay**10), 1.0)
 
     def test_from_input(self):
@@ -62,6 +84,142 @@ def test_from_input(self):
         self.assertEqual(contr.init_points, 3**2)
         self.assertEqual(contr.n_iter, 3**2)
 
+    def test_optimizer_params(self):
+        """Test BayesianOptimizerController.optimizer_params"""
+        contr = BayesianOptimizerController(
+            1, 2, kappa=3, utility_func_kwargs={"xi": 1.11, "kind": "ei"}
+        )
+        result = contr.optimizer_params()
+
+        self.assertDictContainsSubset(
+            {
+                "init_points": 1,
+                "n_iter": 2,
+            },
+            result,
+        )
+        util_func = result["acquisition_function"]
+        self.assertEqual(util_func.kappa, 3)
+        self.assertEqual(util_func._kappa_decay, contr._calc_kappa_decay())
+        self.assertEqual(util_func.xi, 1.11)
+        self.assertEqual(util_func.kind, "ei")
+
+    def test_update_step(self):
+        """Test the update for STEP events"""
+        contr = BayesianOptimizerController(3, 2)
+
+        # Regular step
+        self._make_step(3.0, contr)
+        self.assertEqual(contr.steps, 1)
+        best = Improvement(
+            iteration=0, sample=0, random=True, target=-9.0, improvement=np.inf
+        )
+        self.assertEqual(len(contr._improvements), 1)
+        self.assertTupleEqual(contr._improvements[-1], best)
+
+        # Step that has no effect due to constraints
+        self._make_step(-2.0, contr)
+        self.assertEqual(contr.steps, 2)
+        self.assertEqual(len(contr._improvements), 1)
+        self.assertTupleEqual(contr._improvements[-1], best)
+
+        # Step that is not new max
+        self._make_step(4.0, contr)
+        self.assertEqual(contr.steps, 3)
+        self.assertEqual(len(contr._improvements), 1)
+        self.assertTupleEqual(contr._improvements[-1], best)
+
+        # Two minimal increments, therefore we should see a StopEarly
+        self._make_step(2.999, contr)
+        self.assertEqual(contr.steps, 4)
+        self.assertEqual(len(contr._improvements), 2)
+
+        with self.assertRaises(StopEarly):
+            self._make_step(2.998, contr)
+        self.assertEqual(contr.steps, 5)
+        self.assertEqual(len(contr._improvements), 3)
+
+    def test_update_end(self):
+        """Test the update for END events"""
+        contr = BayesianOptimizerController(1, 1)
+
+        # One step with improvement, then stop
+        self._make_step(3.0, contr)
+        contr.update(Events.OPTIMIZATION_END, self.bayes_opt)
+        self.assertEqual(contr._last_it_improved, 0)
+        self.assertEqual(contr._last_it_end, 1)
+
+        # One step with no more improvement
+        self._make_step(4.0, contr)
+        with self.assertRaises(StopIteration):
+            contr.update(Events.OPTIMIZATION_END, self.bayes_opt)
+
+    def test_improvements(self):
+        """Test ouput of BayesianOptimizerController.improvements"""
+        contr = BayesianOptimizerController(1, 1)
+        self._make_step(3.0, contr)
+        self._make_step(2.0, contr)
+        contr.update(Events.OPTIMIZATION_END, self.bayes_opt)
+        self._make_step(2.5, contr)  # Not better
+        self._make_step(1.0, contr)
+        contr.update(Events.OPTIMIZATION_END, self.bayes_opt)
+        self._make_step(-0.9, contr)  # Constrained
+
+        df = contr.improvements()
+        pd.testing.assert_frame_equal(
+            df,
+            pd.DataFrame.from_dict(
+                data={
+                    "iteration": [0, 0, 1],
+                    "sample": [0, 1, 3],
+                    "random": [True, False, False],
+                    "target": [-9.0, -4.0, -1.0],
+                    "improvement": [np.inf, 9.0 / 4.0 - 1, 3.0],
+                }
+            ).set_index("sample"),
+        )
+
+
+class TestBayesianOptimizerOutput(unittest.TestCase):
+    """Tests for the output class of BayesianOptimizer"""
+
+    def test_p_space_to_dataframe(self):
+        """"""
+        bayes_opt = BayesianOptimization(
+            f=lambda x: -(x**2),
+            pbounds={"x": (-10, 10)},
+            constraint=NonlinearConstraint(fun=lambda x: x, lb=-0.5, ub=np.inf),
+            verbose=0,
+            allow_duplicate_points=True,
+        )
+        bayes_opt.probe({"x": 2.0}, lazy=False)
+        bayes_opt.probe({"x": 1.0}, lazy=False)
+        bayes_opt.probe({"x": -0.9}, lazy=False)
+
+        output = BayesianOptimizerOutput(
+            params=bayes_opt.max["params"],
+            target=bayes_opt.max["target"],
+            p_space=bayes_opt.space,
+        )
+        self.assertDictEqual(output.params, {"x": 1.0})
+        self.assertEqual(output.target, -1.0)
+
+        idx = pd.MultiIndex.from_tuples(
+            [
+                ("Parameters", "x"),
+                ("Calibration", "Cost Function"),
+                ("Calibration", "Constraints Function"),
+                ("Calibration", "Allowed"),
+            ]
+        )
+        df = pd.DataFrame(data=None, columns=idx)
+        df["Parameters", "x"] = [2.0, 1.0, -0.9]
+        df["Calibration", "Cost Function"] = [4.0, 1.0, 0.9**2]
+        df["Calibration", "Constraints Function"] = df["Parameters", "x"]
+        df["Calibration", "Allowed"] = [True, True, False]
+        df.index.rename("Iteration", inplace=True)
+        pd.testing.assert_frame_equal(output.p_space_to_dataframe(), df)
+
 
 class TestBayesianOptimizer(unittest.TestCase):
     """Tests for the optimizer based on bayes_opt.BayesianOptimization"""
@@ -121,4 +279,10 @@ def test_target_func(self, _):
 # Execute Tests
 if __name__ == "__main__":
     TESTS = unittest.TestLoader().loadTestsFromTestCase(TestBayesianOptimizer)
+    TESTS.addTests(
+        unittest.TestLoader().loadTestsFromTestCase(TestBayesianOptimizerOutput)
+    )
+    TESTS.addTests(
+        unittest.TestLoader().loadTestsFromTestCase(TestBayesianOptimizerController)
+    )
     unittest.TextTestRunner(verbosity=2).run(TESTS)