Merge pull request #152 from hrntsm/feature/moea_d

Add MOEA/D sampler

package/samplers/moead/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 <Hiroaki Natsume>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

package/samplers/moead/README.md

@@ -0,0 +1,74 @@
+---
+author: Hiroaki Natsume
+title: MOEA/D sampler
+description: Sampler using MOEA/D algorithm. MOEA/D stands for "Multi-Objective Evolutionary Algorithm based on Decomposition.
+tags: [sampler, multiobjective]
+optuna_versions: [4.0.0]
+license: MIT License
+---
+
+## Abstract
+
+Sampler using MOEA/D algorithm. MOEA/D stands for "Multi-Objective Evolutionary Algorithm based on Decomposition.
+
+This sampler is specialized for multiobjective optimization. The objective function is internally decomposed into multiple single-objective subproblems to perform optimization.
+
+## Class or Function Names
+
+- MOEADSampler
+
+## Installation
+
+```
+pip install scipy
+```
+
+## Example
+
+```python
+import optuna
+import optunahub
+
+def objective(trial: optuna.Trial) -> tuple[float, float]:
+    x = trial.suggest_float("x", 0, 5)
+    y = trial.suggest_float("y", 0, 3)
+
+    v0 = 4 * x**2 + 4 * y**2
+    v1 = (x - 5) ** 2 + (y - 5) ** 2
+    return v0, v1
+
+
+if __name__ == "__main__":
+    population_size = 100
+    n_trials = 1000
+
+    mod = optunahub.load_module("samplers/moead")
+    sampler = mod.MOEADSampler(
+        population_size=population_size,
+        scalar_aggregation_func="tchebycheff",
+        n_neighbors=population_size // 10,
+    )
+    study = optuna.create_study(sampler=sampler)
+    study.optimize(objective, n_trials=n_trials)
+```
+
+## Others
+
+Comparison between Random, NSGAII and MOEA/D with ZDT1 as the objective function.
+See `compare_2objective.py` in moead directory for details.
+
+### Pareto Front Plot
+
+| MOEA/D                      | NSGAII                       | Random                       |
+| --------------------------- | ---------------------------- | ---------------------------- |
+| ![MOEA/D](images/moead.png) | ![NSGAII](images/nsgaii.png) | ![Random](images/random.png) |
+
+### Compare
+
+![Compare](images/compare_pareto_front.png)
+
+### Reference
+
+Q. Zhang and H. Li,
+"MOEA/D: A Multiobjective Evolutionary Algorithm Based on Decomposition," in IEEE Transactions on Evolutionary Computation, vol. 11, no. 6, pp. 712-731, Dec. 2007,
+doi: 10.1109/TEVC.2007.892759.

package/samplers/moead/__init__.py

@@ -0,0 +1,4 @@
+from .moead import MOEADSampler
+
+
+__all__ = ["MOEADSampler"]

package/samplers/moead/_child_generation_strategy.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Optional
+from typing import Sequence
+
+import numpy as np
+from optuna import Study
+from optuna._transform import _SearchSpaceTransform
+from optuna.distributions import BaseDistribution
+from optuna.distributions import FloatDistribution
+from optuna.distributions import IntDistribution
+from optuna.samplers._lazy_random_state import LazyRandomState
+from optuna.samplers.nsgaii._crossover import _is_contained
+from optuna.samplers.nsgaii._crossover import _try_crossover
+from optuna.samplers.nsgaii._crossovers._base import BaseCrossover
+from optuna.trial import FrozenTrial
+
+
+_NUMERICAL_DISTRIBUTIONS = (
+    FloatDistribution,
+    IntDistribution,
+)
+
+
+class MOEAdChildGenerationStrategy:
+    """Generate a child parameter from the given parent population by MOEA/D algorithm.
+    Args:
+        study:
+            Target study object.
+        search_space:
+            A dictionary containing the parameter names and parameter's distributions.
+        parent_population:
+            A list of trials that are selected as parent population.
+    Returns:
+        A dictionary containing the parameter names and parameter's values.
+    """
+
+    def __init__(
+        self,
+        *,
+        mutation_prob: float | None = None,
+        crossover: BaseCrossover,
+        crossover_prob: float,
+        swapping_prob: float,
+        rng: LazyRandomState,
+    ) -> None:
+        if not (mutation_prob is None or 0.0 <= mutation_prob <= 1.0):
+            raise ValueError(
+                "`mutation_prob` must be None or a float value within the range [0.0, 1.0]."
+            )
+
+        if not (0.0 <= crossover_prob <= 1.0):
+            raise ValueError("`crossover_prob` must be a float value within the range [0.0, 1.0].")
+
+        if not (0.0 <= swapping_prob <= 1.0):
+            raise ValueError("`swapping_prob` must be a float value within the range [0.0, 1.0].")
+
+        if not isinstance(crossover, BaseCrossover):
+            raise ValueError(
+                f"'{crossover}' is not a valid crossover."
+                " For valid crossovers see"
+                " https://optuna.readthedocs.io/en/stable/reference/samplers.html."
+            )
+
+        self._crossover_prob = crossover_prob
+        self._mutation_prob = mutation_prob
+        self._swapping_prob = swapping_prob
+        self._crossover = crossover
+        self._rng = rng
+        self._subproblem_id = 0
+
+    def __call__(
+        self,
+        study: Study,
+        search_space: dict[str, BaseDistribution],
+        parent_population: list[FrozenTrial],
+        neighbors: dict[int, list[int]],
+    ) -> dict[str, Any]:
+        """Generate a child parameter from the given parent population by NSGA-II algorithm.
+        Args:
+            study:
+                Target study object.
+            search_space:
+                A dictionary containing the parameter names and parameter's distributions.
+            parent_population:
+                A list of trials that are selected as parent population.
+            neighbors:
+                A dictionary containing the subproblem id and its neighboring subproblems.
+        Returns:
+            A dictionary containing the parameter names and parameter's values.
+        """
+        subproblem_parent_population = [
+            parent_population[i] for i in neighbors[self._subproblem_id]
+        ]
+
+        # We choose a child based on the specified crossover method.
+        if self._rng.rng.rand() < self._crossover_prob:
+            child_params = self._perform_crossover(
+                self._crossover,
+                study,
+                subproblem_parent_population,
+                search_space,
+                self._rng.rng,
+                self._swapping_prob,
+            )
+        else:
+            parent_population_size = len(parent_population)
+            parent_params = parent_population[self._rng.rng.choice(parent_population_size)].params
+            child_params = {name: parent_params[name] for name in search_space.keys()}
+
+        n_params = len(child_params)
+        if self._mutation_prob is None:
+            mutation_prob = 1.0 / max(1.0, n_params)
+        else:
+            mutation_prob = self._mutation_prob
+
+        params = {}
+        for param_name in child_params.keys():
+            if self._rng.rng.rand() >= mutation_prob:
+                params[param_name] = child_params[param_name]
+
+        self._subproblem_id += 1
+        if self._subproblem_id >= len(neighbors):
+            self._subproblem_id = 0
+        return params
+
+    def _perform_crossover(
+        self,
+        crossover: BaseCrossover,
+        study: Study,
+        parent_population: Sequence[FrozenTrial],
+        search_space: Dict[str, BaseDistribution],
+        rng: np.random.RandomState,
+        swapping_prob: float,
+    ) -> Dict[str, Any]:
+        numerical_search_space: Dict[str, BaseDistribution] = {}
+        categorical_search_space: Dict[str, BaseDistribution] = {}
+        for key, value in search_space.items():
+            if isinstance(value, _NUMERICAL_DISTRIBUTIONS):
+                numerical_search_space[key] = value
+            else:
+                categorical_search_space[key] = value
+
+        numerical_transform: Optional[_SearchSpaceTransform] = None
+        if len(numerical_search_space) != 0:
+            numerical_transform = _SearchSpaceTransform(numerical_search_space)
+
+        while True:  # Repeat while parameters lie outside search space boundaries.
+            parents = self._select_parents(crossover, parent_population, rng)
+            child_params = _try_crossover(
+                parents,
+                crossover,
+                study,
+                rng,
+                swapping_prob,
+                categorical_search_space,
+                numerical_search_space,
+                numerical_transform,
+            )
+
+            if _is_contained(child_params, search_space):
+                break
+
+        return child_params
+
+    def _select_parents(
+        self,
+        crossover: BaseCrossover,
+        parent_population: Sequence[FrozenTrial],
+        rng: np.random.RandomState,
+    ) -> List[FrozenTrial]:
+        parents: List[FrozenTrial] = rng.choice(
+            np.array(parent_population), crossover.n_parents, replace=False
+        ).tolist()
+        return parents