Moving Output Objective

bofire/data_models/domain/features.py

@@ -739,15 +739,21 @@ def __call__(
         """
         desis = pd.concat(
             [
-                feat(experiments[f"{feat.key}_pred" if predictions else feat.key])  # type: ignore
+                feat(
+                    experiments[f"{feat.key}_pred" if predictions else feat.key],
+                    experiments[f"{feat.key}_pred" if predictions else feat.key],
+                )  # type: ignore
                 for feat in self.features
                 if feat.objective is not None
                 and not isinstance(feat, CategoricalOutput)
             ]
             + [
                 (
                     pd.Series(
-                        data=feat(experiments.filter(regex=f"{feat.key}(.*)_prob")),
+                        data=feat(
+                            experiments.filter(regex=f"{feat.key}(.*)_prob"),  # type: ignore
+                            experiments.filter(regex=f"{feat.key}(.*)_prob"),  # type: ignore
+                        ),
                         name=f"{feat.key}_pred",
                     )  # type: ignore
                     if predictions

bofire/data_models/features/categorical.py

@@ -356,14 +356,14 @@ def validate_objective_categories(self):
             raise ValueError("categories must match to objective categories")
         return self
 
-    def __call__(self, values: pd.Series) -> pd.Series:
+    def __call__(self, values: pd.Series, values_adapt: pd.Series) -> pd.Series:
         if self.objective is None:
             return pd.Series(
                 data=[np.nan for _ in range(len(values))],
                 index=values.index,
                 name=values.name,
             )
-        return self.objective(values)  # type: ignore
+        return self.objective(values, values_adapt)  # type: ignore
 
     def validate_experimental(self, values: pd.Series) -> pd.Series:
         values = values.map(str)

bofire/data_models/features/continuous.py

@@ -178,14 +178,14 @@ class ContinuousOutput(Output):
         default_factory=lambda: MaximizeObjective(w=1.0)
     )
 
-    def __call__(self, values: pd.Series) -> pd.Series:
+    def __call__(self, values: pd.Series, values_adapt: pd.Series) -> pd.Series:
         if self.objective is None:
             return pd.Series(
                 data=[np.nan for _ in range(len(values))],
                 index=values.index,
                 name=values.name,
             )
-        return self.objective(values)  # type: ignore
+        return self.objective(values, values_adapt)  # type: ignore
 
     def validate_experimental(self, values: pd.Series) -> pd.Series:
         try:

bofire/data_models/objectives/api.py

@@ -1,8 +1,6 @@
 from typing import Union
 
-from bofire.data_models.objectives.categorical import (
-    ConstrainedCategoricalObjective,
-)
+from bofire.data_models.objectives.categorical import ConstrainedCategoricalObjective
 from bofire.data_models.objectives.identity import (
     IdentityObjective,
     MaximizeObjective,
@@ -12,6 +10,7 @@
 from bofire.data_models.objectives.sigmoid import (
     MaximizeSigmoidObjective,
     MinimizeSigmoidObjective,
+    MovingMaximizeSigmoidObjective,
     SigmoidObjective,
 )
 from bofire.data_models.objectives.target import (
@@ -31,6 +30,7 @@
 
 AnyConstraintObjective = Union[
     MaximizeSigmoidObjective,
+    MovingMaximizeSigmoidObjective,
     MinimizeSigmoidObjective,
     TargetObjective,
 ]
@@ -45,4 +45,5 @@
     TargetObjective,
     CloseToTargetObjective,
     ConstrainedCategoricalObjective,
+    MovingMaximizeSigmoidObjective,
 ]

bofire/data_models/objectives/categorical.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Literal, Union
+from typing import Dict, List, Literal, Optional, Union
 
 import numpy as np
 import pandas as pd
@@ -61,12 +61,16 @@ def from_dict_label(self) -> Dict:
         return dict(zip(d.values(), d.keys()))
 
     def __call__(
-        self, x: Union[pd.Series, np.ndarray]
+        self,
+        x: Union[pd.Series, np.ndarray],
+        x_adapt: Optional[Union[pd.Series, np.ndarray]] = None,
     ) -> Union[pd.Series, np.ndarray, float]:
         """The call function returning a probabilistic reward for x.
 
         Args:
             x (np.ndarray): A matrix of x values
+            x_adapt (Optional[np.ndarray], optional): An array of x values which are used to
+                update the objective parameters on the fly. Defaults to None.
 
         Returns:
             np.ndarray: A reward calculated as inner product of probabilities and feasible objectives.

bofire/data_models/objectives/identity.py

@@ -1,4 +1,4 @@
-from typing import Literal, Tuple, Union
+from typing import Literal, Optional, Tuple, Union
 
 import numpy as np
 import pandas as pd
@@ -48,11 +48,17 @@ def validate_lower_upper(cls, bounds):
             )
         return bounds
 
-    def __call__(self, x: Union[pd.Series, np.ndarray]) -> Union[pd.Series, np.ndarray]:
+    def __call__(
+        self,
+        x: Union[pd.Series, np.ndarray],
+        x_adapt: Optional[Union[pd.Series, np.ndarray]] = None,
+    ) -> Union[pd.Series, np.ndarray]:
         """The call function returning a reward for passed x values
 
         Args:
             x (np.ndarray): An array of x values
+            x_adapt (Optional[np.ndarray], optional): An array of x values which are used to
+                update the objective parameters on the fly. Defaults to None.
 
         Returns:
             np.ndarray: The identity as reward, might be normalized to the passed lower and upper bounds
@@ -81,11 +87,17 @@ class MinimizeObjective(IdentityObjective):
 
     type: Literal["MinimizeObjective"] = "MinimizeObjective"
 
-    def __call__(self, x: Union[pd.Series, np.ndarray]) -> Union[pd.Series, np.ndarray]:
+    def __call__(
+        self,
+        x: Union[pd.Series, np.ndarray],
+        x_adapt: Optional[Union[pd.Series, np.ndarray]] = None,
+    ) -> Union[pd.Series, np.ndarray]:
         """The call function returning a reward for passed x values
 
         Args:
             x (np.ndarray): An array of x values
+            x_adapt (Optional[np.ndarray], optional): An array of x values which are used to
+                update the objective parameters on the fly. Defaults to None.
 
         Returns:
             np.ndarray: The negative identity as reward, might be normalized to the passed lower and upper bounds

bofire/data_models/objectives/objective.py

@@ -1,5 +1,5 @@
 from abc import abstractmethod
-from typing import Union
+from typing import Optional, Union
 
 import numpy as np
 import pandas as pd
@@ -15,11 +15,17 @@ class Objective(BaseModel):
     type: str
 
     @abstractmethod
-    def __call__(self, x: Union[pd.Series, np.ndarray]) -> Union[pd.Series, np.ndarray]:
+    def __call__(
+        self,
+        x: Union[pd.Series, np.ndarray],
+        x_adapt: Optional[Union[pd.Series, np.ndarray]] = None,
+    ) -> Union[pd.Series, np.ndarray]:
         """Abstract method to define the call function for the class Objective
 
         Args:
-            x (np.ndarray): An array of x values
+            x (np.ndarray): An array of x values for which the objective should be evaluated.
+            x_adapt (Optional[np.ndarray], optional): An array of x values which are used to
+                update the objective parameters on the fly. Defaults to None.
 
         Returns:
             np.ndarray: The desirability of the passed x values

bofire/data_models/objectives/sigmoid.py

@@ -1,4 +1,4 @@
-from typing import Literal, Union
+from typing import Literal, Optional, Union
 
 import numpy as np
 import pandas as pd
@@ -37,18 +37,63 @@ class MaximizeSigmoidObjective(SigmoidObjective):
 
     type: Literal["MaximizeSigmoidObjective"] = "MaximizeSigmoidObjective"
 
-    def __call__(self, x: Union[pd.Series, np.ndarray]) -> Union[pd.Series, np.ndarray]:
+    def __call__(
+        self,
+        x: Union[pd.Series, np.ndarray],
+        x_adapt: Optional[Union[pd.Series, np.ndarray]] = None,
+    ) -> Union[pd.Series, np.ndarray]:
         """The call function returning a sigmoid shaped reward for passed x values.
 
         Args:
             x (np.ndarray): An array of x values
+            x_adapt (np.ndarray): An array of x values which are used to update the objective parameters on the fly.
 
         Returns:
             np.ndarray: A reward calculated with a sigmoid function. The stepness and the tipping point can be modified via passed arguments.
         """
         return 1 / (1 + np.exp(-1 * self.steepness * (x - self.tp)))
 
 
+class MovingMaximizeSigmoidObjective(SigmoidObjective):
+    """Class for a maximizing sigmoid objective with a moving turning point that depends on so far observed x values.
+
+    Attributes:
+        w (float): float between zero and one for weighting the objective when used in a weighting based strategy.
+        steepness (float): Steepness of the sigmoid function. Has to be greater than zero.
+        tp (float): Relative turning point of the sigmoid function. The actual turning point is calculated by adding
+            the maximum of the observed x values to the relative turning point.
+    """
+
+    type: Literal["MovingMaximizeSigmoidObjective"] = "MovingMaximizeSigmoidObjective"
+
+    def get_adjusted_tp(self, x: Union[pd.Series, np.ndarray]) -> float:
+        """Get the adjusted turning point for the sigmoid function.
+
+        Args:
+            x (np.ndarray): An array of x values
+
+        Returns:
+            float: The adjusted turning point for the sigmoid function.
+        """
+        return x.max() + self.tp
+
+    def __call__(
+        self, x: Union[pd.Series, np.ndarray], x_adapt: Union[pd.Series, np.ndarray]
+    ) -> Union[pd.Series, np.ndarray]:
+        """The call function returning a sigmoid shaped reward for passed x values.
+
+        Args:
+            x (np.ndarray): An array of x values
+            x_adapt (np.ndarray): An array of x values which are used to update the objective parameters on the fly.
+
+        Returns:
+            np.ndarray: A reward calculated with a sigmoid function. The stepness and the tipping point can be modified via passed arguments.
+        """
+        return 1 / (
+            1 + np.exp(-1 * self.steepness * (x - self.get_adjusted_tp(x_adapt)))
+        )
+
+
 class MinimizeSigmoidObjective(SigmoidObjective):
     """Class for a minimizing a sigmoid objective
 
@@ -60,11 +105,16 @@ class MinimizeSigmoidObjective(SigmoidObjective):
 
     type: Literal["MinimizeSigmoidObjective"] = "MinimizeSigmoidObjective"
 
-    def __call__(self, x: Union[pd.Series, np.ndarray]) -> Union[pd.Series, np.ndarray]:
+    def __call__(
+        self,
+        x: Union[pd.Series, np.ndarray],
+        x_adapt: Optional[Union[pd.Series, np.ndarray]] = None,
+    ) -> Union[pd.Series, np.ndarray]:
         """The call function returning a sigmoid shaped reward for passed x values.
 
         Args:
             x (np.ndarray): An array of x values
+            x_adapt (np.ndarray): An array of x values which are used to update the objective parameters on the fly.
 
         Returns:
             np.ndarray: A reward calculated with a sigmoid function. The stepness and the tipping point can be modified via passed arguments.

bofire/data_models/objectives/target.py

@@ -1,4 +1,4 @@
-from typing import Literal, Union
+from typing import Literal, Optional, Union
 
 import numpy as np
 import pandas as pd
@@ -27,7 +27,11 @@ class CloseToTargetObjective(Objective):
     target_value: float
     exponent: float
 
-    def __call__(self, x: Union[pd.Series, np.ndarray]) -> Union[pd.Series, np.ndarray]:
+    def __call__(
+        self,
+        x: Union[pd.Series, np.ndarray],
+        x_adapt: Optional[Union[pd.Series, np.ndarray]] = None,
+    ) -> Union[pd.Series, np.ndarray]:
         return -1 * (np.abs(x - self.target_value) ** self.exponent)
 
 
@@ -48,11 +52,17 @@ class TargetObjective(Objective, ConstrainedObjective):
     tolerance: TGe0
     steepness: TGt0
 
-    def __call__(self, x: Union[pd.Series, np.ndarray]) -> Union[pd.Series, np.ndarray]:
+    def __call__(
+        self,
+        x: Union[pd.Series, np.ndarray],
+        x_adapt: Optional[Union[pd.Series, np.ndarray]] = None,
+    ) -> Union[pd.Series, np.ndarray]:
         """The call function returning a reward for passed x values.
 
         Args:
             x (np.array): An array of x values
+            x_adapt (Optional[np.ndarray], optional): An array of x values which are used to
+                update the objective parameters on the fly. Defaults to None.
 
         Returns:
             np.array: An array of reward values calculated by the product of two sigmoidal shaped functions resulting in a maximum at the target value.

bofire/plot/objective.py

@@ -13,6 +13,7 @@ def plot_objective_plotly(
     lower: float,
     upper: float,
     values: Optional[pd.Series] = None,
+    adapt_values: Optional[pd.Series] = None,
     layout_options: Optional[Dict] = None,
 ):
     """Plot the assigned objective.
@@ -22,6 +23,8 @@ def plot_objective_plotly(
         lower (float): lower bound for the plot
         upper (float): upper bound for the plot
         values (Optional[pd.Series], optional): If provided, scatter also the historical data in the plot. Defaults to None.
+        adapt_values (Optional[pd.Series], optional): If provided, adapt the objective function to the passed values.
+            Defaults to None.
         layout_options: (Dict, optional): Options that are passed to plotlys `update_layout`.
     """
     if feature.objective is None:
@@ -30,14 +33,14 @@ def plot_objective_plotly(
         )
 
     x = pd.Series(np.linspace(lower, upper, 5000))
-    reward = feature.objective.__call__(x)
+    reward = feature.objective.__call__(x, x_adapt=adapt_values)  # type: ignore
 
     fig1 = px.line(x=x, y=reward, title=feature.key)
 
     if values is not None:
         fig2 = px.scatter(
             x=values,
-            y=feature.objective.__call__(values),
+            y=feature.objective.__call__(values, x_adapt=adapt_values),  # type: ignore
         )
         fig = go.Figure(data=fig1.data + fig2.data)  # type: ignore
     else: