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,6 +39,7 @@
     qNEHVI,
     qLogNEHVI,
     qNegIntPosVar,
+    qLogPF,
 ]
 
 AnySingleObjectiveAcquisitionFunction = Union[

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

@@ -1,7 +1,11 @@
 from typing import List, Literal, Union
 
-from pydantic import model_validator
+from pydantic import Field, model_validator
 
+from bofire.data_models.acquisition_functions.api import (
+    AnySingleObjectiveAcquisitionFunction,
+    qLogNEI,
+)
 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
@@ -11,6 +15,10 @@
 class MultiFidelityStrategy(SoboStrategy):
     type: Literal["MultiFidelityStrategy"] = "MultiFidelityStrategy"
 
+    acquisition_function: AnySingleObjectiveAcquisitionFunction = Field(
+        default_factory=lambda: qLogNEI(),
+    )
+
     fidelity_thresholds: Union[List[float], float] = 0.1
 
     @model_validator(mode="after")

bofire/data_models/strategies/predictives/sobo.py

@@ -1,22 +1,19 @@
-from typing import List, Literal, Optional, Type
+from typing import List, Literal, Optional, Type, Union
 
 import pydantic
 from pydantic import Field, field_validator, model_validator
 
 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
 from bofire.data_models.strategies.predictives.botorch import BotorchStrategy
 
 
 class SoboBaseStrategy(BotorchStrategy):
-    acquisition_function: AnySingleObjectiveAcquisitionFunction = Field(
-        default_factory=lambda: qLogNEI(),
-    )
-
     @classmethod
     def is_feature_implemented(cls, my_type: Type[Feature]) -> bool:
         """Method to check if a specific feature type is implemented for the strategy
@@ -47,23 +44,34 @@ 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
+    acquisition_function: Union[AnySingleObjectiveAcquisitionFunction, qLogPF] = Field(
+        default_factory=lambda: qLogNEI(),
+    )
+
+    @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 AdditiveSoboStrategy(SoboBaseStrategy):
     type: Literal["AdditiveSoboStrategy"] = "AdditiveSoboStrategy"
+    acquisition_function: AnySingleObjectiveAcquisitionFunction = Field(
+        default_factory=lambda: qLogNEI(),
+    )
     use_output_constraints: bool = True
 
     @field_validator("domain")
@@ -96,6 +104,10 @@ def check_adaptable_weights(cls, self):
 class MultiplicativeSoboStrategy(SoboBaseStrategy, _CheckAdaptableWeightsMixin):
     type: Literal["MultiplicativeSoboStrategy"] = "MultiplicativeSoboStrategy"
 
+    acquisition_function: AnySingleObjectiveAcquisitionFunction = Field(
+        default_factory=lambda: qLogNEI(),
+    )
+
     @field_validator("domain")
     def validate_is_multiobjective(cls, v, info):
         if (len(v.outputs.get_by_objective(Objective))) < 2:
@@ -121,6 +133,9 @@ class MultiplicativeAdditiveSoboStrategy(SoboBaseStrategy, _CheckAdaptableWeight
     type: Literal["MultiplicativeAdditiveSoboStrategy"] = (
         "MultiplicativeAdditiveSoboStrategy"
     )
+    acquisition_function: AnySingleObjectiveAcquisitionFunction = Field(
+        default_factory=lambda: qLogNEI(),
+    )
     use_output_constraints: bool = True
     additive_features: List[str] = Field(default_factory=list)
 
@@ -137,5 +152,8 @@ def validate_additive_features(cls, v, values):
 
 class CustomSoboStrategy(SoboBaseStrategy):
     type: Literal["CustomSoboStrategy"] = "CustomSoboStrategy"
+    acquisition_function: AnySingleObjectiveAcquisitionFunction = Field(
+        default_factory=lambda: qLogNEI(),
+    )
     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,
@@ -42,9 +47,6 @@
 from bofire.data_models.strategies.api import (
     MultiplicativeSoboStrategy as MultiplicativeDataModel,
 )
-from bofire.data_models.strategies.predictives.sobo import (
-    SoboBaseStrategy as SoboBaseDataModel,
-)
 from bofire.data_models.strategies.predictives.sobo import SoboStrategy as SoboDataModel
 from bofire.strategies.predictives.botorch import BotorchStrategy
 from bofire.utils.torch_tools import (
@@ -61,7 +63,7 @@
 class SoboStrategy(BotorchStrategy):
     def __init__(
         self,
-        data_model: SoboBaseDataModel,
+        data_model: SoboDataModel,
         **kwargs,
     ):
         super().__init__(data_model=data_model, **kwargs)
@@ -111,7 +113,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 +167,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 +178,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,
@@ -207,7 +213,7 @@ def __init__(
         data_model: AdditiveDataModel,
         **kwargs,
     ):
-        super().__init__(data_model=data_model, **kwargs)
+        super().__init__(data_model=data_model, **kwargs)  # type: ignore
         self.use_output_constraints = data_model.use_output_constraints
 
     def _get_objective_and_constraints(
@@ -310,7 +316,7 @@ def __init__(
         data_model: MultiplicativeDataModel,
         **kwargs,
     ):
-        super().__init__(data_model=data_model, **kwargs)
+        super().__init__(data_model=data_model, **kwargs)  # type: ignore
 
     def _get_objective_and_constraints(
         self,
@@ -372,7 +378,7 @@ def __init__(
         **kwargs,
     ):
         self.additive_features = data_model.additive_features
-        super().__init__(data_model=data_model, **kwargs)
+        super().__init__(data_model=data_model, **kwargs)  # type: ignore
 
     def _get_objective_and_constraints(
         self,
@@ -440,7 +446,7 @@ def __init__(
         data_model: CustomDataModel,
         **kwargs,
     ):
-        super().__init__(data_model=data_model, **kwargs)
+        super().__init__(data_model=data_model, **kwargs)  # type: ignore
         self.use_output_constraints = data_model.use_output_constraints
         if data_model.dump is not None:
             self.loads(data_model.dump)