bounds_transformer could bypass global_bounds due to the test logic within _trim function in domain_reduction.py (#441)

* Update trim bounds in domain_reduction.py

Previously, when the new upper limit was less than the original lower limit, the new_bounds could bypass the global_bounds.

* Update test_seq_domain_red.py

Added test cases to catch an error when both bounds of new_bounds exceeded the global_bounds

* Update domain_reduction.py

_trim function now avoids an error when both bounds for a given parameter in new_bounds exceed the global_bounds

* Update domain_reduction.py comments

* fixed English in domain_reduction.py

* use numpy to sort bounds,  boundary exceeded warn.

* simple sort test added

* domain_red windows target_space to global_bounds

Added windowing function to improve the convergence of optimizers that use domain_reduction. Improved comments and documentation.

* target_space.max respects bounds; SDRT warnings

* Remove unused function.

This function was used to prototype a solution. It should not have been pushed and can be removed.

* Updated target_space.py docstrings

* Update tests/test_target_space.py

Co-authored-by: till-m <[email protected]>

* Added pbound warnings, updated various tests.

* updated line spacing for consistency and style

* added pbound test condition

---------

Co-authored-by: till-m <[email protected]>

Author: perezed00

bayes_opt/domain_reduction.py

@@ -2,6 +2,7 @@
 
 import numpy as np
 from .target_space import TargetSpace
+from warnings import warn
 
 
 class DomainTransformer():
@@ -36,7 +37,10 @@ def __init__(
         self.minimum_window_value = minimum_window
 
     def initialize(self, target_space: TargetSpace) -> None:
-        """Initialize all of the parameters"""
+        """Initialize all of the parameters.
+        """
+    
+        # Set the original bounds
         self.original_bounds = np.copy(target_space.bounds)
         self.bounds = [self.original_bounds]
 
@@ -47,6 +51,7 @@ def initialize(self, target_space: TargetSpace) -> None:
         else:
             self.minimum_window = [self.minimum_window_value] * len(target_space.bounds)
 
+        # Set initial values
         self.previous_optimal = np.mean(target_space.bounds, axis=1)
         self.current_optimal = np.mean(target_space.bounds, axis=1)
         self.r = target_space.bounds[:, 1] - target_space.bounds[:, 0]
@@ -72,14 +77,13 @@ def initialize(self, target_space: TargetSpace) -> None:
         self._window_bounds_compatibility(self.original_bounds)
 
     def _update(self, target_space: TargetSpace) -> None:
-
+        """ Updates contraction rate, window size, and window center.
+        """
         # setting the previous
         self.previous_optimal = self.current_optimal
         self.previous_d = self.current_d
-
-        self.current_optimal = target_space.params[
-            np.argmax(target_space.target)
-        ]
+        
+        self.current_optimal = target_space.params_to_array(target_space.max()['params'])
 
         self.current_d = 2.0 * (self.current_optimal -
                                 self.previous_optimal) / self.r
@@ -97,32 +101,84 @@ def _update(self, target_space: TargetSpace) -> None:
         self.r = self.contraction_rate * self.r
 
     def _trim(self, new_bounds: np.array, global_bounds: np.array) -> np.array:
-        for i, variable in enumerate(new_bounds):
-            if variable[0] < global_bounds[i, 0]:
-                variable[0] = global_bounds[i, 0]
-            if variable[1] > global_bounds[i, 1]:
-                variable[1] = global_bounds[i, 1]
-        for i, entry in enumerate(new_bounds):
-            if entry[0] > entry[1]:
-                new_bounds[i, 0] = entry[1]
-                new_bounds[i, 1] = entry[0]
-            window_width = abs(entry[0] - entry[1])
-            if window_width < self.minimum_window[i]:
-                dw = (self.minimum_window[i] - window_width) / 2.0
-                left_expansion_space = abs(global_bounds[i, 0] - entry[0]) # should be non-positive
-                right_expansion_space = abs(global_bounds[i, 1] - entry[1]) # should be non-negative
-                # conservative
-                dw_l = min(dw, left_expansion_space)
-                dw_r = min(dw, right_expansion_space)
-                # this crawls towards the edge
-                ddw_r = dw_r + max(dw - dw_l, 0)
-                ddw_l = dw_l + max(dw - dw_r, 0)
-                new_bounds[i, 0] -= ddw_l
-                new_bounds[i, 1] += ddw_r
+        """
+        Adjust the new_bounds and verify that they adhere to global_bounds and minimum_window.
+
+        Parameters:
+        -----------
+        new_bounds : np.array
+            The proposed new_bounds that (may) need adjustment.
+
+        global_bounds : np.array
+            The maximum allowable bounds for each parameter.
+
+        Returns:
+        --------
+        new_bounds : np.array
+            The adjusted bounds after enforcing constraints.
+        """
+
+        #sort bounds
+        new_bounds = np.sort(new_bounds)
+
+        # Validate each parameter's bounds against the global_bounds
+        for i, pbounds in enumerate(new_bounds):
+            # If the one of the bounds is outside the global bounds, reset the bound to the global bound
+            # This is expected to happen when the window is near the global bounds, no warning is issued
+            if (pbounds[0] < global_bounds[i, 0]):
+                pbounds[0] = global_bounds[i, 0]
+
+            if (pbounds[1] > global_bounds[i, 1]):
+                pbounds[1] = global_bounds[i, 1]
+
+            # If a lower bound is greater than the associated global upper bound, reset it to the global lower bound
+            if (pbounds[0] > global_bounds[i, 1]):
+                pbounds[0] = global_bounds[i, 0]
+                warn("\nDomain Reduction Warning:\n"+
+                    "A parameter's lower bound is greater than the global upper bound."+
+                    "The offensive boundary has been reset."+
+                    "Be cautious of subsequent reductions.", stacklevel=2)
+
+            # If an upper bound is less than the associated global lower bound, reset it to the global upper bound
+            if (pbounds[1] < global_bounds[i, 0]):
+                pbounds[1] = global_bounds[i, 1]
+                warn("\nDomain Reduction Warning:\n"+
+                    "A parameter's lower bound is greater than the global upper bound."+
+                    "The offensive boundary has been reset."+
+                    "Be cautious of subsequent reductions.", stacklevel=2)
+
+        # Adjust new_bounds to ensure they respect the minimum window width for each parameter
+        for i, pbounds in enumerate(new_bounds):
+            current_window_width = abs(pbounds[0] - pbounds[1])
+
+            # If the window width is less than the minimum allowable width, adjust it
+            # Note that when minimum_window < width of the global bounds one side always has more space than required
+            if current_window_width < self.minimum_window[i]:
+                width_deficit = (self.minimum_window[i] - current_window_width) / 2.0
+                available_left_space = abs(global_bounds[i, 0] - pbounds[0])
+                available_right_space = abs(global_bounds[i, 1] - pbounds[1])
+                
+                # determine how much to expand on the left and right
+                expand_left = min(width_deficit, available_left_space)
+                expand_right = min(width_deficit, available_right_space)
+                
+                # calculate the deficit on each side
+                expand_left_deficit = width_deficit - expand_left
+                expand_right_deficit = width_deficit - expand_right
+
+                # shift the deficit to the side with more space
+                adjust_left = expand_left + max(expand_right_deficit, 0)
+                adjust_right = expand_right + max(expand_left_deficit, 0)
+                
+                # adjust the bounds
+                pbounds[0] -= adjust_left
+                pbounds[1] += adjust_right
+
         return new_bounds
 
     def _window_bounds_compatibility(self, global_bounds: np.array) -> bool:
-        """Checks if global bounds are compatible with the minimum window sizes."""
+        """Checks if global bounds are compatible with the minimum window sizes.
+        """
         for i, entry in enumerate(global_bounds):
             global_window_width = abs(entry[1] - entry[0])
             if global_window_width < self.minimum_window[i]:
@@ -133,7 +189,8 @@ def _create_bounds(self, parameters: dict, bounds: np.array) -> dict:
         return {param: bounds[i, :] for i, param in enumerate(parameters)}
 
     def transform(self, target_space: TargetSpace) -> dict:
-
+        """Reduces the bounds of the target space.
+        """
         self._update(target_space)
 
         new_bounds = np.array(
@@ -143,6 +200,6 @@ def transform(self, target_space: TargetSpace) -> dict:
             ]
         ).T
 
-        self._trim(new_bounds, self.original_bounds)
+        new_bounds = self._trim(new_bounds, self.original_bounds)
         self.bounds.append(new_bounds)
         return self._create_bounds(target_space.keys, new_bounds)