Botorch with cardinality constraint via sampling

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 ### Added
 - `py.typed` file to enable the use of type checkers on the user side
+- `ContinuousCardinalityConstraint` is now compatible with `BotorchRecommender`
+- Utilities `inactive_parameter_combinations` and`n_inactive_parameter_combinations` 
+  in both `ContinuousCardinalityConstraint`and `SubspaceContinuous`
+- Attribute `n_threshold_inactive_parameters_generator` added to `BotorchRecommender`
+- Class `_FixedNumericalContinuousParameter`
 
 ### Fixed
 - `CategoricalParameter` and `TaskParameter` no longer incorrectly coerce a single

baybe/constraints/continuous.py

@@ -1,6 +1,9 @@
 """Continuous constraints."""
 
 import math
+from collections.abc import Iterator
+from itertools import combinations
+from math import comb
 
 import numpy as np
 from attrs import define
@@ -46,6 +49,26 @@ class ContinuousCardinalityConstraint(
 ):
     """Class for continuous cardinality constraints."""
 
+    @property
+    def n_inactive_parameter_combinations(self) -> int:
+        """The number of possible inactive parameter combinations."""
+        return sum(
+            comb(len(self.parameters), n_inactive_parameters)
+            for n_inactive_parameters in self._inactive_set_sizes()
+        )
+
+    def _inactive_set_sizes(self) -> range:
+        """Get all possible sizes of inactive parameter sets."""
+        return range(
+            len(self.parameters) - self.max_cardinality,
+            len(self.parameters) - self.min_cardinality + 1,
+        )
+
+    def inactive_parameter_combinations(self) -> Iterator[frozenset[str]]:
+        """Get an iterator over all possible combinations of inactive parameters."""
+        for n_inactive_parameters in self._inactive_set_sizes():
+            yield from combinations(self.parameters, n_inactive_parameters)
+
     def sample_inactive_parameters(self, batch_size: int = 1) -> list[set[str]]:
         """Sample sets of inactive parameters according to the cardinality constraints.
 

baybe/constraints/validation.py

@@ -8,8 +8,14 @@
 from baybe.constraints.discrete import (
     DiscreteDependenciesConstraint,
 )
+from baybe.parameters import NumericalContinuousParameter
 from baybe.parameters.base import Parameter
 
+try:  # For python < 3.11, use the exceptiongroup backport
+    ExceptionGroup
+except NameError:
+    from exceptiongroup import ExceptionGroup
+
 
 def validate_constraints(  # noqa: DOC101, DOC103
     constraints: Collection[Constraint], parameters: Collection[Parameter]
@@ -26,6 +32,8 @@ def validate_constraints(  # noqa: DOC101, DOC103
         ValueError: If any discrete constraint includes a continuous parameter.
         ValueError: If any discrete constraint that is valid only for numerical
             discrete parameters includes non-numerical discrete parameters.
+        ValueError: If any parameter affected by a cardinality constraint does
+            not include zero.
     """
     if sum(isinstance(itm, DiscreteDependenciesConstraint) for itm in constraints) > 1:
         raise ValueError(
@@ -41,6 +49,9 @@ def validate_constraints(  # noqa: DOC101, DOC103
     param_names_discrete = [p.name for p in parameters if p.is_discrete]
     param_names_continuous = [p.name for p in parameters if p.is_continuous]
     param_names_non_numerical = [p.name for p in parameters if not p.is_numerical]
+    params_continuous: list[NumericalContinuousParameter] = [
+        p for p in parameters if isinstance(p, NumericalContinuousParameter)
+    ]
 
     for constraint in constraints:
         if not all(p in param_names_all for p in constraint.parameters):
@@ -78,6 +89,11 @@ def validate_constraints(  # noqa: DOC101, DOC103
                 f"Parameter list of the affected constraint: {constraint.parameters}."
             )
 
+        if isinstance(constraint, ContinuousCardinalityConstraint):
+            validate_cardinality_constraint_parameter_bounds(
+                constraint, params_continuous
+            )
+
 
 def validate_cardinality_constraints_are_nonoverlapping(
     constraints: Collection[ContinuousCardinalityConstraint],
@@ -98,3 +114,39 @@ def validate_cardinality_constraints_are_nonoverlapping(
                 f"cannot share the same parameters. Found the following overlapping "
                 f"parameter sets: {s1}, {s2}."
             )
+
+
+def validate_cardinality_constraint_parameter_bounds(
+    constraint: ContinuousCardinalityConstraint,
+    parameters: Collection[NumericalContinuousParameter],
+) -> None:
+    """Validate that all parameters of a continuous cardinality constraint include zero.
+
+    Args:
+        constraint: A continuous cardinality constraint.
+        parameters: A collection of parameters, including those affected by the
+            constraint.
+
+    Raises:
+        ValueError: If one of the affected parameters does not include zero.
+        ExceptionGroup: If several of the affected parameters do not include zero.
+    """
+    exceptions = []
+    for name in constraint.parameters:
+        # We implicitly assume that the corresponding parameter exists
+        parameter = next(p for p in parameters if p.name == name)
+
+        if not parameter.is_in_range(0.0):
+            exceptions.append(
+                ValueError(
+                    f"The bounds of all parameters affected by a constraint of type "
+                    f"'{ContinuousCardinalityConstraint.__name__}' must include zero, "
+                    f"but the bounds of parameter '{name}' are: "
+                    f"{parameter.bounds.to_tuple()}"
+                )
+            )
+
+    if exceptions:
+        if len(exceptions) == 1:
+            raise exceptions[0]
+        raise ExceptionGroup("invalid parameter bounds", exceptions)

baybe/parameters/numerical.py

@@ -142,3 +142,26 @@ def summary(self) -> dict:  # noqa: D102
             Upper_Bound=self.bounds.upper,
         )
         return param_dict
+
+
+@define(frozen=True, slots=False)
+class _FixedNumericalContinuousParameter(ContinuousParameter):
+    """Parameter class for fixed numerical parameters."""
+
+    is_numeric: ClassVar[bool] = True
+    # See base class.
+
+    value: float = field(converter=float)
+    """The fixed value of the parameter."""
+
+    @property
+    def bounds(self) -> Interval:
+        """The value of the parameter as a degenerate interval."""
+        return Interval(self.value, self.value)
+
+    def is_in_range(self, item: float) -> bool:
+        # See base class.
+        return item == self.value
+
+    def summary(self) -> dict:
+        raise NotImplementedError()

baybe/recommenders/pure/bayesian/botorch.py

@@ -1,13 +1,19 @@
 """Botorch recommender."""
 
+from __future__ import annotations
+
 import math
-from typing import Any, ClassVar
+from collections.abc import Collection
+from typing import TYPE_CHECKING, Any, ClassVar
 
 import pandas as pd
 from attr.converters import optional
 from attrs import define, field
+from attrs.validators import ge, instance_of
 
+from baybe.constraints import ContinuousCardinalityConstraint
 from baybe.exceptions import NoMCAcquisitionFunctionError
+from baybe.parameters.numerical import _FixedNumericalContinuousParameter
 from baybe.recommenders.pure.bayesian.base import BayesianRecommender
 from baybe.searchspace import (
     SearchSpace,
@@ -21,6 +27,9 @@
     sample_numerical_df,
 )
 
+if TYPE_CHECKING:
+    from torch import Tensor
+
 
 @define(kw_only=True)
 class BotorchRecommender(BayesianRecommender):
@@ -58,6 +67,16 @@ class BotorchRecommender(BayesianRecommender):
     """Percentage of discrete search space that is sampled when performing hybrid search
     space optimization. Ignored when ``hybrid_sampler="None"``."""
 
+    n_threshold_inactive_parameters_generator: int = field(
+        default=10, validator=[instance_of(int), ge(1)]
+    )
+    """Threshold used for checking which inactive parameters generator is used when
+    cardinality constraints are present. When the size of the combinatorial list of
+    all possible inactive parameters is larger than the threshold, a fixed number of
+    randomly generated inactive parameter configurations are used and the best
+    optimum among them is recommended; Otherwise, we find the best one by iterating the
+    combinatorial list of all possible inactive parameters """
+
     @sampling_percentage.validator
     def _validate_percentage(  # noqa: DOC101, DOC103
         self, _: Any, value: float
@@ -151,31 +170,174 @@ def _recommend_continuous(
                 f"acquisition functions for batch sizes > 1."
             )
 
+        if len(subspace_continuous.constraints_cardinality):
+            points, _ = self._recommend_continuous_with_cardinality_constraints(
+                subspace_continuous,
+                batch_size,
+            )
+        else:
+            points, _ = self._recommend_continuous_without_cardinality_constraints(
+                subspace_continuous,
+                batch_size,
+            )
+
+        # Return optimized points as dataframe
+        rec = pd.DataFrame(points, columns=subspace_continuous.param_names)
+        return rec
+
+    def _recommend_continuous_with_cardinality_constraints(
+        self,
+        subspace_continuous: SubspaceContinuous,
+        batch_size: int,
+    ) -> tuple[Tensor, Tensor]:
+        """Recommend from a continuous search space with cardinality constraints.
+
+        Args:
+            subspace_continuous: The continuous subspace from which to generate
+                recommendations.
+            batch_size: The size of the recommendation batch.
+
+        Returns:
+            The recommendations.
+            The acquisition values.
+
+        Raises:
+            RuntimeError: If the continuous search space has no cardinality constraint.
+        """
+        import torch
+
+        if not subspace_continuous.constraints_cardinality:
+            raise RuntimeError(
+                f"This method expects a subspace object with constraints of type "
+                f"{ContinuousCardinalityConstraint.__name__}. For a subspace object "
+                f"without constraints of type"
+                f" {ContinuousCardinalityConstraint.__name__}, use method"
+                f"{self._recommend_continuous_without_cardinality_constraints.__name__}."  # noqa
+            )
+
+        acqf_values_all: list[Tensor] = []
+        points_all: list[Tensor] = []
+
+        def append_recommendation_for_inactive_parameters_setting(
+            inactive_parameters: Collection[str],
+        ):
+            """Append the recommendation for each inactive parameter configuration.
+
+            Args:
+                inactive_parameters: A list of inactive parameters.
+            """
+            # Create a new subspace by ensuring all active parameters being
+            # non-zeros.
+            subspace_continuous_without_cardinality_constraints = (
+                subspace_continuous._remove_cardinality_constraints(inactive_parameters)
+            )
+            try:
+                # Optimize the acquisition function
+                (
+                    points_i,
+                    acqf_values_i,
+                ) = self._recommend_continuous_without_cardinality_constraints(
+                    subspace_continuous_without_cardinality_constraints,
+                    batch_size,
+                )
+                # Append recommendation list and acquisition function values
+                points_all.append(points_i.unsqueeze(0))
+                acqf_values_all.append(acqf_values_i.unsqueeze(0))
+
+            # The optimization problem may be infeasible for certain inactive
+            # parameters. The optimize_acqf raises a ValueError when the optimization
+            # problem is infeasible.
+            except ValueError:
+                pass
+
+        # Below we start recommendation
+        if (
+            subspace_continuous.n_inactive_parameter_combinations
+            > self.n_threshold_inactive_parameters_generator
+        ):
+            # When the combinatorial list is too large, randomly set some parameters
+            # inactive.
+            for _ in range(self.n_threshold_inactive_parameters_generator):
+                inactive_params_sample = tuple(
+                    subspace_continuous._sample_inactive_parameters(1)[0]
+                )
+                append_recommendation_for_inactive_parameters_setting(
+                    inactive_params_sample
+                )
+        else:
+            # When the combinatorial list is not too large, iterate the combinatorial
+            # list of all possible inactive parameters.
+            for (
+                inactive_params_generator
+            ) in subspace_continuous.inactive_parameter_combinations():
+                append_recommendation_for_inactive_parameters_setting(
+                    inactive_params_generator
+                )
+
+        # Find the best option
+        points = torch.cat(points_all)[torch.argmax(torch.cat(acqf_values_all)), :]
+        acqf_values = torch.max(torch.cat(acqf_values_all))
+        return points, acqf_values
+
+    def _recommend_continuous_without_cardinality_constraints(
+        self,
+        subspace_continuous: SubspaceContinuous,
+        batch_size: int,
+    ) -> tuple[Tensor, Tensor]:
+        """Recommend from a continuous search space without cardinality constraints.
+
+        Args:
+            subspace_continuous: The continuous subspace from which to generate
+                recommendations.
+            batch_size: The size of the recommendation batch.
+
+        Returns:
+            The recommendations.
+            The acquisition values.
+
+        Raises:
+            RuntimeError: If the continuous search space has any cardinality
+                constraints.
+        """
         import torch
         from botorch.optim import optimize_acqf
 
-        points, _ = optimize_acqf(
+        if subspace_continuous.constraints_cardinality:
+            raise RuntimeError(
+                f"This method expects only subspace object without constraints of type "
+                f"{ContinuousCardinalityConstraint.__name__}. For a subspace object "
+                f"with constraints of type {ContinuousCardinalityConstraint.__name__}, "
+                f"try method {self._recommend_continuous.__name__}."
+            )
+
+        fixed_parameters = {
+            idx: p.value
+            for (idx, p) in enumerate(subspace_continuous.parameters)
+            if isinstance(p, _FixedNumericalContinuousParameter)
+        }
+
+        points, acqf_values = optimize_acqf(
             acq_function=self._botorch_acqf,
             bounds=torch.from_numpy(subspace_continuous.param_bounds_comp),
             q=batch_size,
             num_restarts=5,  # TODO make choice for num_restarts
             raw_samples=10,  # TODO make choice for raw_samples
+            fixed_features=fixed_parameters or None,
             equality_constraints=[
                 c.to_botorch(subspace_continuous.parameters)
                 for c in subspace_continuous.constraints_lin_eq
             ]
-            or None,  # TODO: https://github.com/pytorch/botorch/issues/2042
+            or None,
+            # TODO: https://github.com/pytorch/botorch/issues/2042
             inequality_constraints=[
                 c.to_botorch(subspace_continuous.parameters)
                 for c in subspace_continuous.constraints_lin_ineq
             ]
-            or None,  # TODO: https://github.com/pytorch/botorch/issues/2042
+            or None,
+            # TODO: https://github.com/pytorch/botorch/issues/2042
             sequential=self.sequential_continuous,
         )
-
-        # Return optimized points as dataframe
-        rec = pd.DataFrame(points, columns=subspace_continuous.param_names)
-        return rec
+        return points, acqf_values
 
     def _recommend_hybrid(
         self,