D. Jabs:

- Added linear, epsilon decay to uniform distribution

PyEvo/mutation/uniform.py

@@ -1,4 +1,5 @@
 from typing import Union
+from abc import ABC, abstractmethod
 
 import numpy as np
 from PyHyperparameterSpace.configuration import HyperparameterConfiguration
@@ -11,14 +12,25 @@
 class UniformMutation(Mutation):
     """
     Class representing a mutation operation that introduces noise to float and integer hyperparameters.
-    The mutation perturbs hyperparameters using random values drawn from a uniform distribution.
+
+    This mutation perturbs hyperparameters using random values drawn from a uniform distribution.
+
+        Args:
+            low (Union[int, float]):
+                The lower bound (:= a) of the uniform distribution U(a, b)
+
+            high (Union[int, float]):
+                The upper bound (:= b) of the uniform distribution U(a, b)
+
+            prob (float):
+                Mutation Probability that the mutation occurs
     """
 
-    def __init__(self, low: Union[int, float], high: Union[int, float], hp_type: str, prob: float):
-        assert hp_type == "int" or hp_type == "float", f"Illegal hp_type {hp_type}. It should be 'int', 'float'!"
-        assert low < high, f"Illegal low {low} or high {high}. It should be that low < high!"
+    def __init__(self, low: Union[int, float], high: Union[int, float], prob: float):
+        assert low < 0, f"Illegal low {low}. The argument should be less than 0 !"
+        assert high > 0, f"Illegal high {high}. The argument should be higher than 0!"
+        assert low < high, f"Illegal low {low} or high {high}. The arguments should satisfy the constraint low < high!"
         super().__init__(prob)
-        self._hp_type = hp_type
         self._low = low
         self._high = high
 
@@ -31,28 +43,189 @@ def _mutate(
             optimizer: str,
             **kwargs,
     ) -> list[HyperparameterConfiguration]:
-        if self._hp_type == "float":
-            hp_type = Float
-            data_type = (float, np.float_)
-            sampling = random.uniform
-            epsilon = 1e-10
-        else:
-            hp_type = Integer
-            data_type = (int, np.int_)
-            sampling = random.random_integers
-            epsilon = 1
-
         for ind in pop:
-            for key, hp in ind.items():
-                if isinstance(cs[key], hp_type) and random.random() <= self._prob:
-                    if isinstance(hp, data_type):
-                        ind[key] += sampling(low=self._low, high=self._high)
-                        if ind[key] < cs[key].lb:
-                            ind[key] = cs[key].lb
-                        elif ind[key] >= cs[key].ub:
-                            ind[key] = cs[key].ub - epsilon
-                    elif isinstance(hp, np.ndarray):
-                        ind[key] += sampling(low=self._low, high=self._high, size=hp.shape)
-                        ind[key][ind[key] < cs[key].lb] = cs[key].lb
-                        ind[key][ind[key] >= cs[key].ub] = cs[key].ub - epsilon
+            for key, value in ind.items():
+                hp = cs[key]
+                if isinstance(hp, Float) and random.random() <= self._prob:
+                    # Case: Hyperparameter is continuous
+                    value = hp.adjust_configuration(
+                        value + random.uniform(low=self._low, high=self._high)
+                    )
+                    ind[key] = value
+                elif isinstance(hp, Integer) and random.random() <= self._prob:
+                    # Case: Hyperparameter is discrete
+                    value = hp.adjust_configuration(
+                        value + random.random_integers(low=self._low, high=self._high)
+                    )
+                    ind[key] = value
         return pop
+
+
+class DecayUniformMutation(UniformMutation, ABC):
+    """
+    A mutation class that introduces noise to float and integer hyperparameters
+    with decayed perturbation values over time.
+
+
+        Args:
+            min_low (Union[int, float]):
+                The minimum lower bound (:= a_min) before decaying is starting
+
+            max_low (Union[int, float]):
+                The maximum lower bound (:= a_max) after decaying is finished
+
+            min_high (Union[int, float]):
+                The minimum upper bound (:= b_min) after decaying is finished
+
+            max_high (Union[int, float]):
+                The minimum upper bound (:= b_max) before decaying is starting
+
+            prob (float):
+                Mutation Probability that the mutation occurs
+    """
+    def __init__(
+            self,
+            min_low: Union[int, float],
+            max_low: Union[int, float],
+            min_high: Union[int, float],
+            max_high: Union[int, float],
+            prob: float
+    ):
+        assert min_low < 0, f"Illegal min_low {min_low}. The argument should be less than 0!"
+        assert max_low < 0, f"Illegal max_low {max_low}. The argument should be less than 0!"
+        assert min_high > 0, f"Illegal min_high {min_high}. The argument should be higher than 0!"
+        assert max_high > 0, f"Illegal max_high {max_high}. The argument should be higher than 0!"
+        assert min_low < max_low < min_high < max_high, f"Illegal min_low {min_low}, max_low {max_low}, min_high {min_high} or max_high {max_high}. The arguments should satisfy the constraint min_low < max_low < min_high < max_high!"
+        super().__init__(min_low, max_high, prob)
+        self._min_low = min_low
+        self._max_low = max_low
+        self._min_high = min_high
+        self._max_high = max_high
+
+    @abstractmethod
+    def _update(self):
+        """
+        Updates the current lower (_low) and upper (_high) bound after the decay method.
+        """
+        pass
+
+    def _mutate(
+            self,
+            random: np.random.RandomState,
+            cs: HyperparameterConfigurationSpace,
+            pop: list[HyperparameterConfiguration],
+            fitness: list[float],
+            optimizer: str,
+            **kwargs,
+    ) -> list[HyperparameterConfiguration]:
+        new_pop = super()._mutate(random, cs, pop, fitness, optimizer, **kwargs)
+        self._update()
+        return new_pop
+
+
+class LinearDecayUniformMutation(DecayUniformMutation):
+    """
+    A mutation class that introduces noise to float and integer hyperparameters
+    with decayed perturbation values over time.
+
+    Linear Decaying of the lower and upper bound is defined as follows:
+
+        momentum_low := (low_min - low_max) / (#steps)
+        momentum_high := (high_max - high_min) / (#steps)
+
+        low_t+1 := low_t - momentum_low
+        high_t+1 := high_t - momentum_high
+
+        Args:
+            min_low (Union[int, float]):
+                The minimum lower bound (:= a_min) before decaying is starting
+
+            max_low (Union[int, float]):
+                The maximum lower bound (:= a_max) after decaying is finished
+
+            min_high (Union[int, float]):
+                The minimum upper bound (:= b_min) after decaying is finished
+
+            max_high (Union[int, float]):
+                The minimum upper bound (:= b_max) before decaying is starting
+
+            prob (float):
+                Mutation Probability that the mutation occurs
+            
+            steps (float):
+                Number of times to call the mutation before reaching the minimum standard deviation
+    """
+
+    def __init__(
+            self,
+            min_low: Union[int, float],
+            max_low: Union[int, float],
+            min_high: Union[int, float],
+            max_high: Union[int, float],
+            prob: float,
+            steps: int,
+    ):
+        assert steps > 0, f"Illegal steps {steps}. The argument should be higher than 0!"
+        super().__init__(min_low, max_low, min_high, max_high, prob)
+        self._steps = steps
+
+        self._low_momentum = (self._min_low - self._max_low) / self._steps
+        self._high_momentum = (self._max_high - self._min_high) / self._steps
+
+    def _update(self):
+        self._low = min(self._low - self._low_momentum, self._max_low)
+        self._high = max(self._high - self._high_momentum, self._min_high)
+
+
+class EpsilonDecayUniformMutation(DecayUniformMutation):
+    """
+    A mutation class that introduces noise to float and integer hyperparameters
+    with decayed perturbation values over time.
+
+    Epsilon Decaying of the lower and upper bound is defined as follows:
+
+        decay_low := sqrt_#steps(low_max / low_min)
+        decay_high := sqrt_#steps(high_min / high_max)
+
+        low_t+1 := low_t * decay_low
+        high_t+1 := high_t * decay_high
+
+        Args:
+            min_low (Union[int, float]):
+                The minimum lower bound (:= a_min) before decaying is starting
+
+            max_low (Union[int, float]):
+                The maximum lower bound (:= a_max) after decaying is finished
+
+            min_high (Union[int, float]):
+                The minimum upper bound (:= b_min) after decaying is finished
+
+            max_high (Union[int, float]):
+                The minimum upper bound (:= b_max) before decaying is starting
+
+            prob (float):
+                Mutation Probability that the mutation occurs
+
+            steps (float):
+                Number of times to call the mutation before reaching the minimum standard deviation
+    """
+    def __init__(
+            self,
+            min_low: Union[int, float],
+            max_low: Union[int, float],
+            min_high: Union[int, float],
+            max_high: Union[int, float],
+            prob: float,
+            steps: int,
+    ):
+        assert steps > 0, f"Illegal steps {steps}. The argument should be higher than 0!"
+
+        super().__init__(min_low, max_low, min_high, max_high, prob)
+        self._steps = steps
+
+        self._low_decay = (self._max_low / self._min_low) ** (1 / self._steps)
+        self._high_decay = (self._min_high / self._max_high) ** (1 / self._steps)
+
+    def _update(self):
+        self._low = min(self._low * self._low_decay, self._max_low)
+        self._high = max(self._high * self._high_decay, self._min_high)

tests/mutation/test_uniform.py

@@ -5,7 +5,7 @@
 from PyHyperparameterSpace.hp.continuous import Float
 from PyHyperparameterSpace.space import HyperparameterConfigurationSpace
 
-from PyEvo.mutation.uniform import UniformMutation
+from PyEvo.mutation.uniform import UniformMutation, LinearDecayUniformMutation, EpsilonDecayUniformMutation
 
 
 class TestUniformMutation(unittest.TestCase):
@@ -35,26 +35,96 @@ def setUp(self):
         self.fitness = [-92, -17, 22, 56, -96, -20, 76, 29, -48, -56]
         self.optimizer_min = "min"
         self.optimizer_max = "max"
-        self.mutator_float = UniformMutation(low=-0.5, high=0.5, hp_type="float", prob=1.0)
-        self.mutator_int = UniformMutation(low=-1, high=2, hp_type="int", prob=1.0)
+        self.mutator = UniformMutation(low=-0.5, high=0.5, prob=1.0)
 
-    def test_mutate_float(self):
+    def test_mutate(self):
         """
-        Tests the method mutate() for float hyperparameters.
+        Tests the method mutate().
         """
-        new_pop = self.mutator_float.mutate(self.random, self.cs, self.pop, self.fitness, self.optimizer_min)
+        new_pop = self.mutator.mutate(self.random, self.cs, self.pop, self.fitness, self.optimizer_min)
 
         self.assertEqual(len(self.pop), len(new_pop))
         self.assertNotEqual(self.pop, new_pop)
 
-    def test_mutate_integer(self):
+
+class TestLinearDecayUniformMutation(unittest.TestCase):
+    """
+    Tests the class LinearDecayUniformMutation.
+    """
+
+    def setUp(self):
+        self.random = np.random.RandomState(0)
+        self.cs = HyperparameterConfigurationSpace(
+            values={
+                "max_episodes": Constant("max_episodes", default=3),
+                "max_episode_length": Constant("max_episode_length", default=1000),
+                "hidden1_shape": Constant("hidden1_shape", default=64),
+                "hidden2_shape": Constant("hidden2_shape", default=32),
+                "fc1.weight": Float("fc1.weight", bounds=(-1.0, 1.0), shape=(64, 8)),
+                "fc1.bias": Float("fc1.bias", bounds=(-1.0, 1.0), shape=(64,)),
+                "fc2.weight": Float("fc2.weight", bounds=(-1.0, 1.0), shape=(32, 64)),
+                "fc2.bias": Float("fc2.bias", bounds=(-1.0, 1.0), shape=(32,)),
+                "fc3.weight": Float("fc3.weight", bounds=(-1.0, 1.0), shape=(4, 32)),
+                "fc3.bias": Float("fc3.bias", bounds=(-1.0, 1.0), shape=(4,)),
+                "fc4.bias": Float("fc4.bias", bounds=(-1.0, 1.0), shape=(1,))
+            },
+            seed=0,
+        )
+        self.pop = self.cs.sample_configuration(10)
+        self.fitness = [-92, -17, 22, 56, -96, -20, 76, 29, -48, -56]
+        self.optimizer_min = "min"
+        self.optimizer_max = "max"
+        self.mutator = LinearDecayUniformMutation(min_low=-0.5, max_low=-0.1, min_high=0.1, max_high=0.6, prob=1.0,
+                                                  steps=5)
+
+    def test_mutate(self):
         """
-        Tests the method mutate() for integer hyperparameters.
+        Tests the method mutate().
         """
-        new_pop = self.mutator_int.mutate(self.random, self.cs, self.pop, self.fitness, self.optimizer_min)
+        new_pop = self.mutator.mutate(self.random, self.cs, self.pop, self.fitness, self.optimizer_min)
 
         self.assertEqual(len(self.pop), len(new_pop))
-        self.assertEqual(self.pop, new_pop)
+        self.assertNotEqual(self.pop, new_pop)
+
+
+class TestEpsilonDecayUniformMutation(unittest.TestCase):
+    """
+    Tests the class EpsilonDecayUniformMutation.
+    """
+
+    def setUp(self):
+        self.random = np.random.RandomState(0)
+        self.cs = HyperparameterConfigurationSpace(
+            values={
+                "max_episodes": Constant("max_episodes", default=3),
+                "max_episode_length": Constant("max_episode_length", default=1000),
+                "hidden1_shape": Constant("hidden1_shape", default=64),
+                "hidden2_shape": Constant("hidden2_shape", default=32),
+                "fc1.weight": Float("fc1.weight", bounds=(-1.0, 1.0), shape=(64, 8)),
+                "fc1.bias": Float("fc1.bias", bounds=(-1.0, 1.0), shape=(64,)),
+                "fc2.weight": Float("fc2.weight", bounds=(-1.0, 1.0), shape=(32, 64)),
+                "fc2.bias": Float("fc2.bias", bounds=(-1.0, 1.0), shape=(32,)),
+                "fc3.weight": Float("fc3.weight", bounds=(-1.0, 1.0), shape=(4, 32)),
+                "fc3.bias": Float("fc3.bias", bounds=(-1.0, 1.0), shape=(4,)),
+                "fc4.bias": Float("fc4.bias", bounds=(-1.0, 1.0), shape=(1,))
+            },
+            seed=0,
+        )
+        self.pop = self.cs.sample_configuration(10)
+        self.fitness = [-92, -17, 22, 56, -96, -20, 76, 29, -48, -56]
+        self.optimizer_min = "min"
+        self.optimizer_max = "max"
+        self.mutator = EpsilonDecayUniformMutation(min_low=-0.5, max_low=-0.05, min_high=0.05, max_high=0.6, prob=1.0,
+                                                   steps=5)
+
+    def test_mutate(self):
+        """
+        Tests the method mutate().
+        """
+        new_pop = self.mutator.mutate(self.random, self.cs, self.pop, self.fitness, self.optimizer_min)
+
+        self.assertEqual(len(self.pop), len(new_pop))
+        self.assertNotEqual(self.pop, new_pop)
 
 
 if __name__ == '__main__':