propulate.surrogate

Module Contents

Classes

Surrogate	Initialize a new surrogate model.
StaticSurrogate	Initialize a static surrogate with a synthetic id, a margin, and empty arrays for baseline and current run.
DynamicSurrogate	Initialize a dynamic surrogate with the configuration space limits.

Attributes

T

propulate.surrogate.T

class propulate.surrogate.Surrogate

Initialize a new surrogate model.

Propulate passes down a surrogate factory, as defined by the user, to each Propulator instance. That means __init__ can be overwritten to take additional arguments.

Raises:

NotImplementedError – Not implemented (abstract base class method needs to be overridden).

abstract start_run(ind: propulate.population.Individual) → None

Signal that a new run is about to start.

This is called before the first yield from the loss_fn. It is assumed that the individual is freshly created. Keep in mind that there might be (private) keys that are not related to limits, like the surrogate key _s; key names related to limits could be provided to the Surrogate constructor if necessary.

Parameters:

ind (propulate.population.Individual) – The individual to be evaluated.

Raises:

NotImplementedError – Not implemented (abstract base class method needs to be overridden).

abstract update(loss: float) → None

Update the surrogate model with the final loss.

Indicative that the current run has finished.

Parameters:

loss (float) – The final loss of the current run.

Raises:

NotImplementedError – Not implemented (abstract base class method needs to be overridden).

abstract cancel(loss: float) → bool

Evaluate surrogate to check if the current run should be cancelled.

This will be called after every yield from the loss_fn.

Parameters:

loss (float) – The loss of the most recent step.

Returns:

If the surrogate model determines that the current run will not result in a lower loss than previous runs.

Return type:

bool

Raises:

NotImplementedError – Not implemented (abstract base class method needs to be overridden).

abstract merge(data: Any) → None

Merge the results of another surrogate model into itself.

Used to synchronize surrogate models from different Propulators.

Implementation of merge has to be commutative! Otherwise, the different surrogate models will diverge.

Parameters:

data (Any) – All relevant information to update its model to the same state as the origin of the data.

Raises:

NotImplementedError – Not implemented (abstract base class method needs to be overridden).

abstract data() → Any

Return all relevant information about the surrogate model for merging with another surrogate.

It most likely only needs to return the most recent loss from update().

Returns:

All relevant information to convey the current state of the surrogate model.

Return type:

T

Raises:

NotImplementedError – Not implemented (abstract base class method needs to be overridden).

class propulate.surrogate.StaticSurrogate(margin: float = 0.8)

Bases: Surrogate

Initialize a static surrogate with a synthetic id, a margin, and empty arrays for baseline and current run.

Parameters:

margin (float, optional) – A margin on top of incoming losses for comparison with the baseline. The default is 0.8.

start_run(ind: propulate.population.Individual) → None

Reset the internal index and the current run array.

Parameters:

ind (propulate.population.Individual) – The individual containing the current configuration.

update(loss: float) → None

Replace the baseline with the current run if the final loss is better, or if there is no prior run.

Parameters:

loss (float) – The (unused) final loss of the current run.

cancel(loss: float) → bool

Cancel the current run if the loss is outside the margin of the baseline.

Parameters:

loss (float) – The next interim loss of the current run.

Returns:

True if the current run is cancelled, False otherwise.

Return type:

bool

merge(data: numpy.ndarray) → None

Replace the baseline with the incoming run if the final loss is better, or if there is no prior run.

Parameters:

data (np.ndarray) – The loss series of the incoming run.

data() → numpy.ndarray

Return the loss series of the best known run so far.

Returns:

The loss series of the best known run so far.

Return type:

np.ndarray

class propulate.surrogate.DynamicSurrogate(limits: Dict[str, Tuple[float, float] | Tuple[int, int] | Tuple[str, Ellipsis]])

Bases: Surrogate

Initialize a dynamic surrogate with the configuration space limits.

Set the global and local kernels for Gaussian process regression. All other needed attributes are initialized as empty arrays or None.

Parameters:

limits (Union[Dict[str, Tuple[float, float]], Dict[str, Tuple[int, int]], Dict[str, Tuple[str, ...]]) – The hyperparameter configuration space’s limits.

start_run(ind: propulate.population.Individual) → None

Encode the config. given as individual and create the local GP’s mean function from the global GP’s prediction.

Parameters:

ind (propulate.population.Individual) – The individual containing the current configuration.

update(loss: float) → None

Update the model with the current run’s final loss and retrain the global Gaussian Process with the new data.

Parameters:

loss (float) – The final loss of the current run.

cancel(loss: float) → bool

Cancel the run if the local GP predicts a final loss higher than the best known loss with an allowed margin.

Parameters:

loss (float) – The next interim loss of the current run.

Returns:

True if the run should be cancelled, False otherwise.

Return type:

bool

merge(data: Tuple[numpy.ndarray, float]) → None

Merge a configuration and loss tuple into the history arrays and retrain the global GP with the new data.

Parameters:

data (Tuple[np.ndarray, float]) – The configurations and final losses of the merged run.

data() → Tuple[numpy.ndarray, float]

Return the latest configuration and loss from the history arrays.

Returns:

The latest configuration and loss.

Return type:

Tuple[np.ndarray, float]