Probabiltiy of Feasibility ACQF

bofire/data_models/acquisition_functions/acquisition_function.py

@@ -1,4 +1,4 @@
-from typing import Annotated, Dict, Literal, Optional
+from typing import Annotated, Any, Dict, Literal, Optional
 
 from pydantic import Field, PositiveFloat
 
@@ -7,15 +7,15 @@
 
 
 class AcquisitionFunction(BaseModel):
-    type: str
+    type: Any
 
 
 class SingleObjectiveAcquisitionFunction(AcquisitionFunction):
-    type: str
+    type: Any
 
 
 class MultiObjectiveAcquisitionFunction(AcquisitionFunction):
-    type: str
+    type: Any
 
 
 class qNEI(SingleObjectiveAcquisitionFunction):
@@ -87,3 +87,22 @@ class qNegIntPosVar(SingleObjectiveAcquisitionFunction):
     type: Literal["qNegIntPosVar"] = "qNegIntPosVar"
     n_mc_samples: IntPowerOfTwo = 512
     weights: Optional[Dict[str, PositiveFloat]] = Field(default_factory=lambda: None)
+
+
+class qLogPF(SingleObjectiveAcquisitionFunction):
+    """MC based batch LogProbability of Feasibility acquisition function.
+
+    It is used to select the next batch of experiments to maximize the
+    probability of finding feasible solutions with respect to output
+    constraints in the next batch. It can be only used in the SoboStrategy
+    and is especially useful in combination with the FeasibleExperimentCondition
+    within the StepwiseStrategy.
+
+    Attributes:
+        n_mc_samples: Number of Monte Carlo samples to use to
+            approximate the probability of feasibility.
+
+    """
+
+    type: Literal["qLogPF"] = "qLogPF"
+    n_mc_samples: IntPowerOfTwo = 512

bofire/data_models/acquisition_functions/api.py

@@ -10,6 +10,7 @@
     qLogEI,
     qLogNEHVI,
     qLogNEI,
+    qLogPF,
     qNegIntPosVar,
     qNEHVI,
     qNEI,
@@ -38,16 +39,11 @@
     qNEHVI,
     qLogNEHVI,
     qNegIntPosVar,
+    qLogPF,
 ]
 
 AnySingleObjectiveAcquisitionFunction = Union[
-    qNEI,
-    qEI,
-    qSR,
-    qUCB,
-    qPI,
-    qLogEI,
-    qLogNEI,
+    qNEI, qEI, qSR, qUCB, qPI, qLogEI, qLogNEI, qLogPF
 ]
 
 AnyMultiObjectiveAcquisitionFunction = Union[qEHVI, qLogEHVI, qNEHVI, qLogNEHVI]

bofire/data_models/strategies/api.py

@@ -58,6 +58,7 @@
     AlwaysTrueCondition,
     AnyCondition,
     CombiCondition,
+    FeasibleExperimentCondition,
     NumberOfExperimentsCondition,
 )
 from bofire.data_models.strategies.stepwise.stepwise import Step, StepwiseStrategy

bofire/data_models/strategies/predictives/multi_fidelity.py

@@ -4,11 +4,11 @@
 
 from bofire.data_models.domain.api import Domain, Outputs
 from bofire.data_models.features.api import TaskInput
-from bofire.data_models.strategies.predictives.sobo import SoboStrategy
+from bofire.data_models.strategies.predictives.sobo import SoboStrategy, _ForbidPFMixin
 from bofire.data_models.surrogates.api import BotorchSurrogates, MultiTaskGPSurrogate
 
 
-class MultiFidelityStrategy(SoboStrategy):
+class MultiFidelityStrategy(SoboStrategy, _ForbidPFMixin):
     type: Literal["MultiFidelityStrategy"] = "MultiFidelityStrategy"
 
     fidelity_thresholds: Union[List[float], float] = 0.1

bofire/data_models/strategies/predictives/sobo.py

@@ -6,6 +6,7 @@
 from bofire.data_models.acquisition_functions.api import (
     AnySingleObjectiveAcquisitionFunction,
     qLogNEI,
+    qLogPF,
 )
 from bofire.data_models.features.api import Feature
 from bofire.data_models.objectives.api import ConstrainedObjective, Objective
@@ -47,23 +48,43 @@ def is_objective_implemented(cls, my_type: Type[Objective]) -> bool:
 class SoboStrategy(SoboBaseStrategy):
     type: Literal["SoboStrategy"] = "SoboStrategy"
 
-    @field_validator("domain")
-    @classmethod
-    def validate_is_singleobjective(cls, v, values):
-        if len(v.outputs) == 1:
-            return v
+    @model_validator(mode="after")
+    def validate_is_singleobjective(self):
         if (
-            len(v.outputs.get_by_objective(excludes=ConstrainedObjective))
-            - len(v.outputs.get_by_objective(includes=None, excludes=Objective))
-        ) > 1:
+            len(self.domain.outputs.get_by_objective(excludes=ConstrainedObjective))
+            - len(
+                self.domain.outputs.get_by_objective(includes=None, excludes=Objective)  # type: ignore
+            )
+        ) > 1 and not isinstance(self.acquisition_function, qLogPF):
             raise ValueError(
                 "SOBO strategy can only deal with one no-constraint objective.",
             )
-        return v
+        if isinstance(self.acquisition_function, qLogPF):
+            if len(self.domain.outputs.get_by_objective(ConstrainedObjective)) == 0:
+                raise ValueError(
+                    "At least one constrained objective is required for qLogPF.",
+                )
+        return self
+
 
+class _ForbidPFMixin:
+    """
+    Mixin to forbid the use of qLogPF acquisition function in single-objective strategies
+    that are not the SoboStrategy.
+    """
+
+    @field_validator("acquisition_function")
+    def validate_acquisition_function(cls, acquisition_function):
+        if isinstance(acquisition_function, qLogPF):
+            raise ValueError(
+                "qLogPF acquisition function is only allowed in the ´SoboStrategy´.",
+            )
+        return acquisition_function
 
-class AdditiveSoboStrategy(SoboBaseStrategy):
+
+class AdditiveSoboStrategy(SoboBaseStrategy, _ForbidPFMixin):
     type: Literal["AdditiveSoboStrategy"] = "AdditiveSoboStrategy"
+
     use_output_constraints: bool = True
 
     @field_validator("domain")
@@ -93,7 +114,9 @@ def check_adaptable_weights(cls, self):
         return self
 
 
-class MultiplicativeSoboStrategy(SoboBaseStrategy, _CheckAdaptableWeightsMixin):
+class MultiplicativeSoboStrategy(
+    SoboBaseStrategy, _CheckAdaptableWeightsMixin, _ForbidPFMixin
+):
     type: Literal["MultiplicativeSoboStrategy"] = "MultiplicativeSoboStrategy"
 
     @field_validator("domain")
@@ -105,7 +128,9 @@ def validate_is_multiobjective(cls, v, info):
         return v
 
 
-class MultiplicativeAdditiveSoboStrategy(SoboBaseStrategy, _CheckAdaptableWeightsMixin):
+class MultiplicativeAdditiveSoboStrategy(
+    SoboBaseStrategy, _CheckAdaptableWeightsMixin, _ForbidPFMixin
+):
     """
     Mixed, weighted multiplicative (primary, strict) and additive (secondary, non-strict) objectives.
 
@@ -121,6 +146,7 @@ class MultiplicativeAdditiveSoboStrategy(SoboBaseStrategy, _CheckAdaptableWeight
     type: Literal["MultiplicativeAdditiveSoboStrategy"] = (
         "MultiplicativeAdditiveSoboStrategy"
     )
+
     use_output_constraints: bool = True
     additive_features: List[str] = Field(default_factory=list)
 
@@ -135,7 +161,8 @@ def validate_additive_features(cls, v, values):
         return v
 
 
-class CustomSoboStrategy(SoboBaseStrategy):
+class CustomSoboStrategy(SoboBaseStrategy, _ForbidPFMixin):
     type: Literal["CustomSoboStrategy"] = "CustomSoboStrategy"
+
     use_output_constraints: bool = True
     dump: Optional[str] = None

bofire/data_models/strategies/stepwise/conditions.py

@@ -1,11 +1,13 @@
 from abc import abstractmethod
-from typing import Annotated, List, Literal, Optional, Union
+from typing import Annotated, Any, List, Literal, Optional, Union
 
 import pandas as pd
-from pydantic import Field, field_validator
+from pydantic import Field, PositiveInt, field_validator
 
 from bofire.data_models.base import BaseModel
+from bofire.data_models.constraints.api import IntrapointConstraint
 from bofire.data_models.domain.api import Domain
+from bofire.data_models.objectives.api import ConstrainedObjective
 
 
 class EvaluateableCondition:
@@ -15,11 +17,68 @@ def evaluate(self, domain: Domain, experiments: Optional[pd.DataFrame]) -> bool:
 
 
 class Condition(BaseModel):
-    type: str
+    type: Any
 
 
 class SingleCondition(BaseModel):
-    type: str
+    type: Any
+
+
+class FeasibleExperimentCondition(SingleCondition, EvaluateableCondition):
+    """Condition to check if a certain number of feasible experiments are available.
+
+    For this purpose, the condition checks if there are any kind of ConstrainedObjective's
+    in the domain. If, yes it checks if there is a certain number of feasible experiments.
+    The condition is fulfilled if the number of feasible experiments is smaller than
+    the number of required feasible experiments. It is not fulfilled when there are no
+    ConstrainedObjective's in the domain.
+    This condition can be used in scenarios where there is a large amount of output constraints
+    and one wants to make sure that they are fulfilled before optimizing the actual objective(s).
+    To do this, it is best to combine this condition with the SoboStrategy and qLogPF
+    as acquisition function.
+
+    Attributes:
+        n_required_feasible_experiments: Number of required feasible experiments.
+        threshold: Threshold for the feasibility calculation. Default is 0.9.
+    """
+
+    type: Literal["FeasibleExperimentCondition"] = "FeasibleExperimentCondition"
+    n_required_feasible_experiments: PositiveInt = 1
+    threshold: Annotated[float, Field(ge=0, le=1)] = 0.9
+
+    def evaluate(self, domain: Domain, experiments: Optional[pd.DataFrame]) -> bool:
+        constrained_outputs = domain.outputs.get_by_objective(ConstrainedObjective)
+        if len(constrained_outputs) == 0:
+            return False
+
+        if experiments is None:
+            return True
+
+        valid_experiments = (
+            constrained_outputs.preprocess_experiments_all_valid_outputs(experiments)
+        )
+        relevant_constraints = domain.constraints.get(IntrapointConstraint)
+        if len(relevant_constraints) > 0:
+            valid_experiments = valid_experiments[
+                relevant_constraints.is_fulfilled(valid_experiments)
+            ]
+
+        # TODO: have a is fulfilled for input features --> work for future PR
+        feasibilities = pd.concat(
+            [
+                feat(
+                    valid_experiments[feat.key],
+                    valid_experiments[feat.key],  # type: ignore
+                )
+                for feat in constrained_outputs
+            ],
+            axis=1,
+        ).product(axis=1)
+
+        return bool(
+            feasibilities[feasibilities >= self.threshold].sum()
+            < self.n_required_feasible_experiments
+        )
 
 
 class NumberOfExperimentsCondition(SingleCondition, EvaluateableCondition):
@@ -72,4 +131,9 @@ def evaluate(self, domain: Domain, experiments: Optional[pd.DataFrame]) -> bool:
         return False
 
 
-AnyCondition = Union[NumberOfExperimentsCondition, CombiCondition, AlwaysTrueCondition]
+AnyCondition = Union[
+    NumberOfExperimentsCondition,
+    CombiCondition,
+    AlwaysTrueCondition,
+    FeasibleExperimentCondition,
+]

bofire/strategies/predictives/sobo.py

@@ -22,12 +22,17 @@
 import torch
 from botorch.acquisition import get_acquisition_function
 from botorch.acquisition.acquisition import AcquisitionFunction
-from botorch.acquisition.objective import ConstrainedMCObjective, GenericMCObjective
+from botorch.acquisition.objective import (
+    ConstrainedMCObjective,
+    GenericMCObjective,
+    IdentityMCObjective,
+)
 from botorch.models.gpytorch import GPyTorchModel
 
 from bofire.data_models.acquisition_functions.api import (
     AnySingleObjectiveAcquisitionFunction,
     qLogNEI,
+    qLogPF,
     qNEI,
     qPI,
     qSR,
@@ -111,7 +116,7 @@ def _get_acqfs(self, n) -> List[AcquisitionFunction]:
     def _get_objective_and_constraints(
         self,
     ) -> Tuple[
-        Union[GenericMCObjective, ConstrainedMCObjective],
+        Union[GenericMCObjective, ConstrainedMCObjective, IdentityMCObjective],
         Union[List[Callable[[torch.Tensor], torch.Tensor]], None],
         Union[List, float],
     ]:
@@ -165,7 +170,9 @@ def _get_objective_and_constraints(
 
         # return regular objective
         return (
-            GenericMCObjective(objective=objective_callable),
+            GenericMCObjective(objective=objective_callable)
+            if not isinstance(self.acquisition_function, qLogPF)
+            else IdentityMCObjective(),
             constraint_callables,
             etas,
         )
@@ -174,7 +181,9 @@ def _get_objective_and_constraints(
     def make(
         cls,
         domain: Domain,
-        acquisition_function: AnySingleObjectiveAcquisitionFunction | None = None,
+        acquisition_function: AnySingleObjectiveAcquisitionFunction
+        | qLogPF
+        | None = None,
         acquisition_optimizer: AnyAcqfOptimizer | None = None,
         surrogate_specs: BotorchSurrogates | None = None,
         outlier_detection_specs: OutlierDetections | None = None,