Add WFG benchmark test

benchmarks/mo_runner.py

@@ -0,0 +1,49 @@
+import sys
+from kurobako import solver
+from kurobako.solver.optuna import OptunaSolverFactory
+import optuna
+
+# import json
+
+
+optuna.logging.disable_default_handler()
+
+
+def create_study(seed: int) -> optuna.Study:
+
+    n_objectives = 2
+    directions = ["minimize" for _ in range(n_objectives)]
+
+    sampler_name = sys.argv[1]
+
+    # Sampler.
+    sampler_cls = getattr(
+        optuna.samplers,
+        sampler_name,
+        getattr(optuna.integration, sampler_name, None),
+    )
+    if sampler_cls is None:
+        raise ValueError("Unknown sampler: {}.".format(sampler_name))
+
+    # TODO(drumehiron): sampler_kwargs
+    # sampler_kwargs = json.loads(sys.argv[2])
+    # try:
+    #     sampler_kwargs["seed"] = seed
+    #     sampler = sampler_cls(**sampler_kwargs)
+    # except ValueError:
+    #     del sampler_kwargs["seed"]
+    #     sampler = sampler_cls(**sampler_kwargs)
+    sampler = sampler_cls()
+
+    return optuna.create_study(
+        directions=directions,
+        sampler=sampler,
+        pruner=optuna.pruners.NopPruner(),
+    )
+
+
+if __name__ == "__main__":
+
+    factory = OptunaSolverFactory(create_study)
+    runner = solver.SolverRunner(factory)
+    runner.run()

benchmarks/problems/WFGtestSuite/shape_functions.py

@@ -0,0 +1,104 @@
+import abc
+
+import numpy as np
+
+
+class BaseShapeFunction(object, metaclass=abc.ABCMeta):
+    def __init__(self, n_objectives: int) -> None:
+
+        self._n_objectives = n_objectives
+
+    def __call__(self, m: int, x: np.ndarray) -> float:
+        assert 1 <= m <= self.n_objectives
+        assert x.shape == (self.n_objectives - 1,)
+        return self._call(m, x)
+
+    @abc.abstractmethod
+    def _call(self, m: int, x: np.ndarray) -> float:
+        raise NotImplementedError
+
+    @property
+    def n_objectives(self) -> int:
+
+        return self._n_objectives
+
+
+class LinearShapeFunction(BaseShapeFunction):
+    def _call(self, m: int, x: np.ndarray) -> float:
+
+        if m == 1:
+            return x[:-1].prod()
+
+        if m == self.n_objectives:
+            return 1 - x[0]
+
+        return x[: self.n_objectives - m].prod() * (1.0 - x[self.n_objectives - m])
+
+
+class ConvexShapeFunction(BaseShapeFunction):
+    def _call(self, m: int, x: np.ndarray) -> float:
+
+        if m == 1:
+            return (
+                1
+                - np.cos(
+                    x * np.pi / 2,
+                )
+            )[:-1].prod()
+
+        if m == self.n_objectives:
+            return 1 - np.sin(x[0] * np.pi / 2.0)
+
+        return (1.0 - np.cos(x * np.pi / 2.0))[: self.n_objectives - m].prod() * (
+            1.0 - np.sin(x[self.n_objectives - m] * np.pi / 2.0)
+        )
+
+
+class ConcaveShapeFunction(BaseShapeFunction):
+    def _call(self, m: int, x: np.ndarray) -> float:
+
+        if m == 1:
+            return np.sin(x * np.pi / 2.0)[:-1].prod()
+
+        if m == self.n_objectives:
+            return np.cos(x[0] * np.pi / 2.0)
+
+        return np.sin(x * np.pi / 2.0)[: self.n_objectives - m].prod() * np.cos(
+            x[self.n_objectives - m] * np.pi / 2.0
+        )
+
+
+class MixedConvexOrConcaveShapeFunction(BaseShapeFunction):
+    def __init__(self, n_objectives, alpha: float, n_segments: int) -> None:
+        super().__init__(n_objectives)
+        self._alpha = alpha
+        self._n_segments = n_segments
+
+    def _call(self, m: int, x: np.ndarray) -> float:
+        if m == self.n_objectives:
+            two_A_pi = 2 * self._n_segments * np.pi
+            return np.power(
+                1 - x[0] - np.cos(two_A_pi * x[0] + np.pi / 2.0) / two_A_pi, self._alpha
+            )
+
+        raise ValueError("m should be the number of objectives")
+
+
+class DisconnectedShapeFunction(BaseShapeFunction):
+    def __init__(
+        self, n_objectives, alpha: float, beta: float, n_disconnected_regions: int
+    ) -> None:
+        super().__init__(n_objectives)
+        self._alpha = alpha
+        self._beta = beta
+        self._n_disconnected_regions = n_disconnected_regions
+
+    def _call(self, m: int, x: np.ndarray) -> float:
+        if m == self.n_objectives:
+            return (
+                1
+                - np.power(x[0], self._alpha)
+                * np.cos(self._n_disconnected_regions * np.power(x[0], self._beta) * np.pi) ** 2
+            )
+
+        raise ValueError("m should be the number of objectives")

benchmarks/problems/WFGtestSuite/transformation_functions.py

@@ -0,0 +1,197 @@
+import abc
+from typing import Callable
+
+import numpy as np
+
+
+class BaseTransformations(object, metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def __call__(self, *args, **kwargs):
+
+        raise NotImplementedError
+
+
+class BaseBiasTransformation(BaseTransformations, metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def __call__(self, y: np.ndarray) -> float:
+
+        raise NotImplementedError
+
+
+class BaseShiftTransformation(BaseTransformations, metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def __call__(self, y: float) -> float:
+
+        raise NotImplementedError
+
+
+class BaseReductionTransformation(BaseTransformations, metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def __call__(self, y: np.ndarray) -> float:
+
+        raise NotImplementedError
+
+
+class PolynomialBiasTransformation(BaseBiasTransformation):
+    def __init__(self, alpha: float) -> None:
+
+        assert alpha > 0 and alpha != 1.0
+        self._alpha = alpha
+
+    def __call__(self, y: float) -> float:
+
+        return np.power(y, self._alpha)
+
+
+class FlatRegionBiasTransformation(BaseBiasTransformation):
+    def __init__(self, a: float, b: float, c: float) -> None:
+
+        assert 0 <= a <= 1
+        assert 0 <= b <= 1
+        assert 0 <= c <= 1
+        assert b < c
+        assert not (b == 0) or (a == 0 and c != 1)
+        assert not (c == 1) or (a == 1 and b != 0)
+
+        self._a = a
+        self._b = b
+        self._c = c
+
+    def __call__(self, y: float) -> float:
+
+        a = self._a
+        b = self._b
+        c = self._c
+        return (
+            a
+            + min(0, np.floor(y - b)) * a * (b - y) / b
+            - min(0, np.floor(c - y)) * (1.0 - a) * (y - c) / (1.0 - c)
+        )
+
+
+class ParameterDependentBiasTransformation(BaseReductionTransformation):
+    def __init__(
+        self,
+        w: np.ndarray,
+        input_converter: Callable[[np.ndarray], np.ndarray],
+        a: float,
+        b: float,
+        c: float,
+        i: int,
+    ) -> None:
+
+        assert 0 < a < 1
+        assert 0 < b < c
+
+        self._w = w
+        self._input_converter = input_converter
+        self._a = a
+        self._b = b
+        self._c = c
+        self._i = i
+
+    def __call__(self, y: np.ndarray) -> float:
+
+        w = self._w
+        a = self._a
+        b = self._b
+        c = self._c
+        i = self._i
+
+        u = (self._input_converter(y) * w).sum() / w.sum()
+        v = a - (1.0 - 2 * u) * np.fabs(np.floor(0.5 - u) + a)
+        return np.power(y[i], b + (c - b) * v)
+
+
+class LinearShiftTransformation(BaseShiftTransformation):
+    def __init__(self, a: float) -> None:
+
+        assert 0 < a < 1
+
+        self._a = a
+
+    def __call__(self, y: float) -> float:
+
+        return np.fabs(y - self._a) / np.fabs(np.floor(self._a - y) + self._a)
+
+
+class DeceptiveShiftTransformation(BaseShiftTransformation):
+    def __init__(self, a: float, b: float, c: float) -> None:
+
+        assert 0 < a < 1
+        assert 0 < b < 1
+        assert 0 < c < 1
+        assert a - b > 0
+        assert a + b < 1
+
+        self._a = a
+        self._b = b
+        self._c = c
+
+    def __call__(self, y: float) -> float:
+
+        a = self._a
+        b = self._b
+        c = self._c
+
+        q1 = np.floor(y - a + b) * (1.0 - c + (a - b) / b)
+        q2 = np.floor(a + b - y) * (1.0 - c + (1.0 - a - b) / b)
+        return 1.0 + (np.fabs(y - a) - b) * (q1 / (a - b) + q2 / (1.0 - a - b) + 1.0 / b)
+
+
+class MultiModalShiftTransformation(BaseShiftTransformation):
+    def __init__(self, a: int, b: float, c: float) -> None:
+
+        assert a > 0
+        assert b >= 0
+        assert (4 * a + 2) * np.pi >= 4 * b
+        assert 0 < c < 1
+
+        self._a = a
+        self._b = b
+        self._c = c
+
+    def __call__(self, y: float) -> float:
+
+        a = self._a
+        b = self._b
+        c = self._c
+
+        q1 = np.fabs(y - c) / (2 * (np.floor(c - y) + c))
+        q2 = (4 * a + 2) * np.pi * (0.5 - q1)
+        return (1.0 + np.cos(q2) + 4 * b * (q1**2)) / (b + 2)
+
+
+class WeightedSumReductionTransformation(BaseReductionTransformation):
+    def __init__(self, w: np.ndarray, input_converter: Callable[[np.ndarray], np.ndarray]) -> None:
+
+        assert all(w > 0)
+
+        self._w = w
+        self._input_converter = input_converter
+
+    def __call__(self, y: np.ndarray) -> float:
+
+        y = self._input_converter(y)
+        return (y * self._w).sum() / self._w.sum()
+
+
+class NonSeparableReductionTransformation(BaseReductionTransformation):
+    def __init__(self, a: int, input_converter: Callable[[np.ndarray], np.ndarray]) -> None:
+
+        assert a > 0
+
+        self._a = a
+        self._input_converter = input_converter
+
+    def __call__(self, y: np.ndarray) -> float:
+
+        a = float(self._a)
+        y = self._input_converter(y)
+        n = y.shape[0]
+        indices = [(j + k + 1) % n for k in np.arange(n) for j in np.arange(n)]
+
+        q = y.sum() + np.fabs(y[indices].reshape((n, n)) - y)[:, 0 : int(a) - 1].sum()
+        r = n * np.ceil(a / 2) * (1.0 + 2 * a - 2 * np.ceil(a / 2)) / a
+
+        return q / r