Time Dependent Bayesian Optimization

TimeDependentBayesianGenerator

Bases: BayesianGenerator, ABC

Time-dependent Bayesian generator for Bayesian Optimization.

Attributes:

Name	Type	Description
name	str	The name of the generator.
target_prediction_time	Optional[PositiveFloat]	The target prediction time.
added_time	PositiveFloat	Time added to the current time to get the target prediction time.
gp_constructor	TimeDependentModelConstructor	Constructor used to generate the model.
forgetting_time	Optional[PositiveFloat]	Time period to forget historical data in seconds.

Methods:

Name	Description
validate_gp_constructor	Validate the Gaussian Process (GP) constructor.
get_training_data	Window data based on the forgetting time.
generate	Generate candidates for Bayesian Optimization.
get_input_data	Convert input data to a torch tensor.
get_acquisition	Get the acquisition function for Bayesian Optimization.

Source code in xopt/generators/bayesian/time_dependent.py

class TimeDependentBayesianGenerator(BayesianGenerator, ABC):
    """
    Time-dependent Bayesian generator for Bayesian Optimization.

    Attributes
    ----------
    name : str
        The name of the generator.
    target_prediction_time : Optional[PositiveFloat]
        The target prediction time.
    added_time : PositiveFloat
        Time added to the current time to get the target prediction time.
    gp_constructor : TimeDependentModelConstructor
        Constructor used to generate the model.
    forgetting_time : Optional[PositiveFloat]
        Time period to forget historical data in seconds.

    Methods
    -------
    validate_gp_constructor(cls, value)
        Validate the Gaussian Process (GP) constructor.
    get_training_data(self, data: pd.DataFrame) -> pd.DataFrame
        Window data based on the forgetting time.
    generate(self, n_candidates: int) -> List[dict]
        Generate candidates for Bayesian Optimization.
    get_input_data(self, data: pd.DataFrame) -> torch.Tensor
        Convert input data to a torch tensor.
    get_acquisition(self, model) -> FixedFeatureAcquisitionFunction
        Get the acquisition function for Bayesian Optimization.
    """

    name = "time_dependent_bayesian_generator"
    target_prediction_time: Optional[PositiveFloat] = Field(None)
    added_time: PositiveFloat = Field(
        0.1,
        description="time added to current time to get target prediction time",
    )

    gp_constructor: TimeDependentModelConstructor = Field(
        TimeDependentModelConstructor(),
        description="constructor used to generate model",
    )
    forgetting_time: Optional[PositiveFloat] = Field(
        None, description="time period to forget historical data in seconds"
    )

    @field_validator("vocs", mode="after")
    def validate_vocs(cls, v, info: ValidationInfo):
        if v.n_objectives != 1:
            raise ValueError("this generator only supports vocs with 1 objective")
        return v

    @field_validator("gp_constructor", mode="before")
    def validate_gp_constructor(
        cls, value: Optional[TimeDependentModelConstructor]
    ) -> TimeDependentModelConstructor:
        """
        Validate the Gaussian Process (GP) constructor.

        Parameters
        ----------
        value : Optional[TimeDependentModelConstructor]
            The GP constructor to validate.

        Returns
        -------
        TimeDependentModelConstructor
            The validated GP constructor.

        Raises
        ------
        ValueError
            If the GP constructor is not found.
        """
        constructor_dict = {"time_dependent": TimeDependentModelConstructor}
        if value is None:
            value = TimeDependentModelConstructor()
        elif isinstance(value, TimeDependentModelConstructor):
            value = value
        elif isinstance(value, str):
            if value in constructor_dict:
                value = constructor_dict[value]()
            else:
                raise ValueError(f"{value} not found")
        elif isinstance(value, dict):
            name = value.pop("name")
            if name in constructor_dict:
                value = constructor_dict[name](**value)
            else:
                raise ValueError(f"Constructor '{name}' not found")

        return value

    def get_training_data(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        Window data based on the forgetting time.

        Parameters
        ----------
        data : pd.DataFrame
            The input data.

        Returns
        -------
        pd.DataFrame
            The windowed data.
        """
        new_data = copy(data)
        if self.forgetting_time is not None:
            new_data = new_data[data["time"] > time.time() - self.forgetting_time]

        return new_data

    def generate(self, n_candidates: int) -> List[dict]:
        """
        Generate candidates for Bayesian Optimization.

        Parameters
        ----------
        n_candidates : int
            The number of candidates to generate.

        Returns
        -------
        List[dict]
            The generated candidates.
        """
        self.target_prediction_time = time.time() + self.added_time
        output = super().generate(n_candidates)

        if time.time() > self.target_prediction_time:
            warnings.warn(
                "target prediction time is in the past! Increase "
                "added time for accurate results",
                RuntimeWarning,
            )
        while time.time() < self.target_prediction_time:
            time.sleep(0.001)

        return output

    def get_input_data(self, data: pd.DataFrame) -> torch.Tensor:
        """
        Convert input data to a torch tensor.

        Parameters
        ----------
        data : pd.DataFrame
            The input data in the form of a pandas DataFrame.

        Returns
        -------
        torch.Tensor
            A torch tensor containing the input data.

        Notes
        -----
        This method takes a pandas DataFrame as input data and converts it into a
        torch tensor. It specifically selects columns corresponding to the model's
        input names (variables), and the resulting tensor is configured with the data
        type and device settings from the generator.
        """
        return torch.tensor(
            data[self.model_input_names + ["time"]].to_numpy(), **self.tkwargs
        )

    def get_acquisition(
        self, model: torch.nn.Module
    ) -> FixedFeatureAcquisitionFunction:
        """
        Get the acquisition function for Bayesian Optimization.

        Parameters
        ----------
        model : torch.nn.Module
            The model used for Bayesian Optimization.

        Returns
        -------
        FixedFeatureAcquisitionFunction
            The acquisition function with fixed features.
        """
        acq = super().get_acquisition(model)

        # identify which column has the `time` attribute
        column = [-1]
        value = torch.tensor(self.target_prediction_time, **self.tkwargs).unsqueeze(0)
        fixed_acq = FixedFeatureAcquisitionFunction(
            acq, self.vocs.n_variables + 1, column, value
        )

        return fixed_acq

model_input_names property

variable names corresponding to trained model

__init__(**kwargs)

Initialize the generator.

Source code in xopt/generator.py

def __init__(self, **kwargs):
    """
    Initialize the generator.
    """
    super().__init__(**kwargs)
    logger.info(f"Initialized generator {self.name}")

add_data(new_data)

Add new data to the generator for Bayesian Optimization.

Parameters:

Name	Type	Description	Default
new_data	DataFrame	The new data to be added to the generator.	required

Notes

This method appends the new data to the existing data in the generator.

Source code in xopt/generators/bayesian/bayesian_generator.py

def add_data(self, new_data: pd.DataFrame):
    """
    Add new data to the generator for Bayesian Optimization.

    Parameters
    ----------
    new_data : pd.DataFrame
        The new data to be added to the generator.

    Notes
    -----
    This method appends the new data to the existing data in the generator.
    """
    self.data = pd.concat([self.data, new_data], axis=0, ignore_index=True)

generate(n_candidates)

Generate candidates for Bayesian Optimization.

Parameters:

Name	Type	Description	Default
n_candidates	int	The number of candidates to generate.	required

Returns:

Type	Description
List[dict]	The generated candidates.

Source code in xopt/generators/bayesian/time_dependent.py

def generate(self, n_candidates: int) -> List[dict]:
    """
    Generate candidates for Bayesian Optimization.

    Parameters
    ----------
    n_candidates : int
        The number of candidates to generate.

    Returns
    -------
    List[dict]
        The generated candidates.
    """
    self.target_prediction_time = time.time() + self.added_time
    output = super().generate(n_candidates)

    if time.time() > self.target_prediction_time:
        warnings.warn(
            "target prediction time is in the past! Increase "
            "added time for accurate results",
            RuntimeWarning,
        )
    while time.time() < self.target_prediction_time:
        time.sleep(0.001)

    return output

get_acquisition(model)

Get the acquisition function for Bayesian Optimization.

Parameters:

Name	Type	Description	Default
model	Module	The model used for Bayesian Optimization.	required

Returns:

Type	Description
FixedFeatureAcquisitionFunction	The acquisition function with fixed features.

Source code in xopt/generators/bayesian/time_dependent.py

def get_acquisition(
    self, model: torch.nn.Module
) -> FixedFeatureAcquisitionFunction:
    """
    Get the acquisition function for Bayesian Optimization.

    Parameters
    ----------
    model : torch.nn.Module
        The model used for Bayesian Optimization.

    Returns
    -------
    FixedFeatureAcquisitionFunction
        The acquisition function with fixed features.
    """
    acq = super().get_acquisition(model)

    # identify which column has the `time` attribute
    column = [-1]
    value = torch.tensor(self.target_prediction_time, **self.tkwargs).unsqueeze(0)
    fixed_acq = FixedFeatureAcquisitionFunction(
        acq, self.vocs.n_variables + 1, column, value
    )

    return fixed_acq

get_input_data(data)

Convert input data to a torch tensor.

Parameters:

Name	Type	Description	Default
data	DataFrame	The input data in the form of a pandas DataFrame.	required

Returns:

Type	Description
Tensor	A torch tensor containing the input data.

Notes

This method takes a pandas DataFrame as input data and converts it into a torch tensor. It specifically selects columns corresponding to the model's input names (variables), and the resulting tensor is configured with the data type and device settings from the generator.

Source code in xopt/generators/bayesian/time_dependent.py

def get_input_data(self, data: pd.DataFrame) -> torch.Tensor:
    """
    Convert input data to a torch tensor.

    Parameters
    ----------
    data : pd.DataFrame
        The input data in the form of a pandas DataFrame.

    Returns
    -------
    torch.Tensor
        A torch tensor containing the input data.

    Notes
    -----
    This method takes a pandas DataFrame as input data and converts it into a
    torch tensor. It specifically selects columns corresponding to the model's
    input names (variables), and the resulting tensor is configured with the data
    type and device settings from the generator.
    """
    return torch.tensor(
        data[self.model_input_names + ["time"]].to_numpy(), **self.tkwargs
    )

get_optimum()

select the best point(s) given by the model using the Posterior mean

Source code in xopt/generators/bayesian/bayesian_generator.py

def get_optimum(self):
    """select the best point(s) given by the
    model using the Posterior mean"""
    acq = qUpperConfidenceBound(
        model=self.model, beta=0.0, objective=self._get_objective()
    )
    if len(self.vocs.constraints):
        acq = ConstrainedMCAcquisitionFunction(
            self.model,
            acq,
            self._get_constraint_callables(),
            sampler=self._get_sampler(self.model),
        )
    bounds = self._get_bounds()

    if self.fixed_features is not None:
        acq = self._apply_fixed_features(acq)

        indices = []
        for idx, name in enumerate(self.vocs.variable_names):
            if name not in self.fixed_features:
                indices += [idx]

        bounds = bounds[:, indices]

    bounds = bounds.to(**self.tkwargs)
    acq = acq.to(**self.tkwargs)

    # use default initial conditions for a global search
    result = self.numerical_optimizer.optimize(acq, bounds, 1)

    return self._process_candidates(result)

get_training_data(data)

Window data based on the forgetting time.

Parameters:

Name	Type	Description	Default
data	DataFrame	The input data.	required

Returns:

Type	Description
DataFrame	The windowed data.

Source code in xopt/generators/bayesian/time_dependent.py

def get_training_data(self, data: pd.DataFrame) -> pd.DataFrame:
    """
    Window data based on the forgetting time.

    Parameters
    ----------
    data : pd.DataFrame
        The input data.

    Returns
    -------
    pd.DataFrame
        The windowed data.
    """
    new_data = copy(data)
    if self.forgetting_time is not None:
        new_data = new_data[data["time"] > time.time() - self.forgetting_time]

    return new_data

model_dump(*args, **kwargs)

overwrite model dump to remove faux class attrs

Source code in xopt/generator.py

def model_dump(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
    """overwrite model dump to remove faux class attrs"""

    res = super().model_dump(*args, **kwargs)

    res.pop("supports_batch_generation", None)
    res.pop("supports_multi_objective", None)

    return res

propose_candidates(model, n_candidates=1)

Propose candidates using Bayesian Optimization.

Parameters:

Name	Type	Description	Default
model	Module	The trained Bayesian model.	required
n_candidates	int	The number of candidates to propose (default is 1).	1

Returns:

Type	Description
Tensor	A tensor containing the proposed candidates.

Notes

This method proposes candidates for Bayesian Optimization by numerically optimizing the acquisition function using the trained model. It updates the state of the Turbo controller if used and calculates the optimization bounds.

Source code in xopt/generators/bayesian/bayesian_generator.py

def propose_candidates(self, model: Module, n_candidates: int = 1) -> Tensor:
    """
    Propose candidates using Bayesian Optimization.

    Parameters
    ----------
    model : Module
        The trained Bayesian model.
    n_candidates : int, optional
        The number of candidates to propose (default is 1).

    Returns
    -------
    Tensor
        A tensor containing the proposed candidates.

    Notes
    -----
    This method proposes candidates for Bayesian Optimization by numerically
    optimizing the acquisition function using the trained model. It updates the
    state of the Turbo controller if used and calculates the optimization bounds.
    """
    # update TurBO state if used with the last `n_candidates` points
    if self.turbo_controller is not None:
        self.turbo_controller.update_state(self, n_candidates)

    # calculate optimization bounds
    bounds = self._get_optimization_bounds()

    # get acquisition function
    acq_funct = self.get_acquisition(model)

    # get initial candidates to start acquisition function optimization
    initial_points = self._get_initial_conditions(n_candidates)

    # get candidates -- grid optimizer does not support batch_initial_conditions
    if isinstance(self.numerical_optimizer, GridOptimizer):
        candidates = self.numerical_optimizer.optimize(
            acq_funct, bounds, n_candidates
        )
    else:
        candidates = self.numerical_optimizer.optimize(
            acq_funct, bounds, n_candidates, batch_initial_conditions=initial_points
        )
    return candidates

train_model(data=None, update_internal=True)

Train a Bayesian model for Bayesian Optimization.

Parameters:

Name	Type	Description	Default
data	DataFrame	The data to be used for training the model. If not provided, the internal data of the generator is used.	None
update_internal	bool	Flag to indicate whether to update the internal model of the generator with the trained model (default is True).	True

Returns:

Type	Description
Module	The trained Bayesian model.

Raises:

Type	Description
ValueError	If no data is available to build the model.

Notes

This method trains a Bayesian model using the provided data or the internal data of the generator. It updates the internal model with the trained model if the 'update_internal' flag is set to True.

Source code in xopt/generators/bayesian/bayesian_generator.py

def train_model(
    self, data: pd.DataFrame | None = None, update_internal: bool = True
) -> Module:
    """
    Train a Bayesian model for Bayesian Optimization.

    Parameters
    ----------
    data : pd.DataFrame, optional
        The data to be used for training the model. If not provided, the internal
        data of the generator is used.
    update_internal : bool, optional
        Flag to indicate whether to update the internal model of the generator
        with the trained model (default is True).

    Returns
    -------
    Module
        The trained Bayesian model.

    Raises
    ------
    ValueError
        If no data is available to build the model.

    Notes
    -----
    This method trains a Bayesian model using the provided data or the internal
    data of the generator. It updates the internal model with the trained model
    if the 'update_internal' flag is set to True.
    """
    if data is None:
        data = self.get_training_data(self.data)
        if data is None:
            raise ValueError("no data available to build model")

    if data.empty:
        raise ValueError("no data available to build model")

    # get input bounds
    variable_bounds = {
        name: ele.domain for name, ele in self.vocs.variables.items()
    }

    # if turbo restrict points is true then set the bounds to the trust region
    # bounds
    if self.turbo_controller is not None:
        if self.turbo_controller.restrict_model_data:
            variable_bounds = dict(
                zip(
                    self.vocs.variable_names,
                    self.turbo_controller.get_trust_region(self).numpy().T,
                )
            )

    # add fixed feature bounds if requested
    if self.fixed_features is not None:
        # get bounds for each fixed_feature (vocs bounds take precedent)
        for key in self.fixed_features:
            if key not in variable_bounds:
                if key not in data:
                    raise KeyError(
                        "generator data needs to contain fixed feature "
                        f"column name `{key}`"
                    )
                f_data = data[key]
                bounds = [f_data.min(), f_data.max()]
                if bounds[1] - bounds[0] < 1e-8:
                    bounds[1] = bounds[0] + 1e-8
                variable_bounds[key] = bounds

    _model = self.gp_constructor.build_model(
        self.model_input_names,
        self.vocs.output_names,
        data,
        {name: variable_bounds[name] for name in self.model_input_names},
        **self.tkwargs,
    )

    if update_internal:
        self.model = _model

    return _model

validate_gp_constructor(value)

Validate the Gaussian Process (GP) constructor.

Parameters:

Name	Type	Description	Default
value	Optional[TimeDependentModelConstructor]	The GP constructor to validate.	required

Returns:

Type	Description
TimeDependentModelConstructor	The validated GP constructor.

Raises:

Type	Description
ValueError	If the GP constructor is not found.

Source code in xopt/generators/bayesian/time_dependent.py

@field_validator("gp_constructor", mode="before")
def validate_gp_constructor(
    cls, value: Optional[TimeDependentModelConstructor]
) -> TimeDependentModelConstructor:
    """
    Validate the Gaussian Process (GP) constructor.

    Parameters
    ----------
    value : Optional[TimeDependentModelConstructor]
        The GP constructor to validate.

    Returns
    -------
    TimeDependentModelConstructor
        The validated GP constructor.

    Raises
    ------
    ValueError
        If the GP constructor is not found.
    """
    constructor_dict = {"time_dependent": TimeDependentModelConstructor}
    if value is None:
        value = TimeDependentModelConstructor()
    elif isinstance(value, TimeDependentModelConstructor):
        value = value
    elif isinstance(value, str):
        if value in constructor_dict:
            value = constructor_dict[value]()
        else:
            raise ValueError(f"{value} not found")
    elif isinstance(value, dict):
        name = value.pop("name")
        if name in constructor_dict:
            value = constructor_dict[name](**value)
        else:
            raise ValueError(f"Constructor '{name}' not found")

    return value

validate_turbo_controller(value, info) classmethod

note default behavior is no use of turbo

Source code in xopt/generators/bayesian/bayesian_generator.py

@field_validator("turbo_controller", mode="before")
@classmethod
def validate_turbo_controller(cls, value: Any, info: ValidationInfo):
    """note default behavior is no use of turbo"""
    if value is None:
        return value

    compatible_turbo_controllers = [
        turbo_controller
        for turbo_controller in cls.get_compatible_turbo_controllers()
        if turbo_controller is not None
    ]

    if len(compatible_turbo_controllers) == 0:
        raise ValueError("no turbo controllers are compatible with this generator")
    else:
        return validate_turbo_controller_base(
            value, compatible_turbo_controllers, info
        )

visualize_model(**kwargs)

Display GP model predictions for the selected output(s).

The GP models are displayed with respect to the named variables. If None are given, the list of variables in vocs is used. Feasible samples are indicated with a filled orange "o", infeasible samples with a hollow red "o". Feasibility is calculated with respect to all constraints unless the selected output is a constraint itself, in which case only that one is considered.

Parameters:

Name	Type	Description	Default
**kwargs		Supported keyword arguments: - output_names : List[str] Outputs for which the GP models are displayed. Defaults to all outputs in vocs. - variable_names : List[str] The variables with respect to which the GP models are displayed (maximum of 2). Defaults to vocs.variable_names. - idx : int Index of the last sample to use. This also selects the point of reference in higher dimensions unless an explicit reference_point is given. - reference_point : dict Reference point determining the value of variables in vocs.variable_names, but not in variable_names (slice plots in higher dimensions). Defaults to last used sample. - show_samples : bool, optional Whether samples are shown. - show_prior_mean : bool, optional Whether the prior mean is shown. - show_feasibility : bool, optional Whether the feasibility region is shown. - show_acquisition : bool, optional Whether the acquisition function is computed and shown (only if acquisition function is not None). - n_grid : int, optional Number of grid points per dimension used to display the model predictions. - axes : Axes, optional Axes object used for plotting. - exponentiate : bool, optional Flag to exponentiate acquisition function before plotting.	{}

Returns:

Name	Type	Description
result	tuple	The matplotlib figure and axes objects.

Source code in xopt/generators/bayesian/bayesian_generator.py

def visualize_model(self, **kwargs):
    """Display GP model predictions for the selected output(s).

    The GP models are displayed with respect to the named variables. If None are given, the list of variables in
    vocs is used. Feasible samples are indicated with a filled orange "o", infeasible samples with a hollow
    red "o". Feasibility is calculated with respect to all constraints unless the selected output is a
    constraint itself, in which case only that one is considered.

    Parameters
    ----------
    **kwargs: dict, optional
        Supported keyword arguments:
        - output_names : List[str]
            Outputs for which the GP models are displayed. Defaults to all outputs in vocs.
        - variable_names : List[str]
            The variables with respect to which the GP models are displayed (maximum of 2).
            Defaults to vocs.variable_names.
        - idx : int
            Index of the last sample to use. This also selects the point of reference in
            higher dimensions unless an explicit reference_point is given.
        - reference_point : dict
            Reference point determining the value of variables in vocs.variable_names, but not in variable_names
            (slice plots in higher dimensions). Defaults to last used sample.
        - show_samples : bool, optional
            Whether samples are shown.
        - show_prior_mean : bool, optional
            Whether the prior mean is shown.
        - show_feasibility : bool, optional
            Whether the feasibility region is shown.
        - show_acquisition : bool, optional
            Whether the acquisition function is computed and shown (only if acquisition function is not None).
        - n_grid : int, optional
            Number of grid points per dimension used to display the model predictions.
        - axes : Axes, optional
            Axes object used for plotting.
        - exponentiate : bool, optional
            Flag to exponentiate acquisition function before plotting.

    Returns
    -------
    result : tuple
        The matplotlib figure and axes objects.
    """
    return visualize_generator_model(self, **kwargs)

yaml(**kwargs)

serialize first then dump to yaml string

Source code in xopt/pydantic.py

def yaml(self, **kwargs):
    """serialize first then dump to yaml string"""
    output = json.loads(
        self.to_json(
            **kwargs,
        )
    )
    return yaml.dump(output)