Base optimizer (`evopt.base_optimizer`)

Warning

This module is part of the internal implementation of evopt and is not intended for direct use by end users. Its API may change without notice.

black-box optimization base implementation.

This module provides an abstract base class for implementing black-box optimization algorithms. It handles common functionality such as parameter handling, parallel evaluation of solutions, logging, and statistics calculation.

The BaseOptimizer class is designed to be extended by specific black-box optimization algorithm implementations like CMA-ES, DE, PSO, etc. At the current stage only CMA-ES is implemented.

Example

Creating a custom optimizer by extending BaseOptimizer:

>>> class MyOptimizer(BaseOptimizer):
...     def setup_opt(self, epoch=None):
...         # Initialize optimization algorithm
...         pass
...
...     def optimize(self):
...         # Run optimization algorithm
...         while not self.check_termination():
...             solutions = self.generate_solutions()
...             errors = self.process_batch(solutions)
...             self.update_algorithm(solutions, errors)
...         return self.best_solution
...
...     def check_termination(self):
...         # Check if optimization should stop
...         return self.current_epoch >= self.n_epochs

class evopt.base_optimizer.BaseOptimizer(parameters: dict, evaluator, batch_size: int, directory_manager: DirectoryManager, sigma_threshold: float = 0.1, rand_seed: int = 1, start_epoch: int | None = None, verbose: bool = True, n_epochs: int | None = None, target_dict: dict | None = None, max_workers: int = 1, cores_per_worker: int = 1, **kwargs)[source]

Bases: ABC

Abstract base class for evolutionary optimization algorithms.

This class provides the foundation for implementing evolutionary optimization algorithms, handling common tasks such as parameter management, parallel solution evaluation, statistics tracking, and result logging. It abstracts away the infrastructure details, allowing subclasses to focus on algorithm-specific implementation.

The class maintains a history of optimization metrics across epochs, supports both serial and parallel evaluation of solutions, and provides detailed logging of optimization progress.

parameters

Parameter definitions with (min, max) bounds.

Type:: dict

target_dict

Target values to optimize towards.

Type:: dict, optional

evaluator

Function that evaluates parameter sets.

Type:: callable

n_epochs

Maximum number of epochs to run.

Type:: int, optional

batch_size

Number of solutions to evaluate per epoch.

Type:: int

dir_manager

Manages output directories and files.

Type:: DirectoryManager

sigma_threshold

Convergence threshold for normalized sigmas.

Type:: float

rand_seed

Random seed for reproducibility.

Type:: int

verbose

Whether to print detailed progress information.

Type:: bool

current_epoch

The current optimization epoch.

Type:: int

max_workers

Maximum number of parallel workers.

Type:: int

Note

This is an abstract class and cannot be instantiated directly. Subclasses must implement the setup_opt, optimize, and check_termination methods.

abstract check_termination() → bool[source]

Check if optimization should terminate.

Determines whether the optimization process should stop based on convergence criteria or maximum epochs.

Returns:: True if optimization should stop, False otherwise.
Return type:: bool

Note

This is an abstract method that must be implemented by subclasses.

Example

>>> class MyOptimizer(BaseOptimizer):
...     def check_termination(self):
...         if self.current_epoch >= self.n_epochs:
...             return True
...         return all(ns < self.sigma_threshold
...                   for ns in self.norm_sigmas.values())

cleanup()[source]

Explicitly clean up resources used by the optimizer.

Releases all resources used by the optimizer, including any process pools. This method should be called when the optimizer is no longer needed.

Example

>>> # After optimization is complete
>>> optimizer.cleanup()

property get_init_params: list

Generate initial parameters within normalized bounds.

Creates random initial parameter values uniformly distributed within the normalized parameter bounds.

Returns:: List of initial normalized parameter values.
Return type:: list

Example

>>> np.random.seed(1)  # For reproducible example
>>> optimizer = MyOptimizer(parameters={'x': (0, 10), 'y': (-5, 5)}, ...)
>>> optimizer.get_init_params
[2.17, -0.45]  # Random values within normalized bounds

property get_init_sigmas: ndarray

Calculate initial standard deviations based on parameter bounds.

Computes initial sigma values as 1/4 of the range between min and max for each parameter, providing a reasonable starting point for exploration.

Returns:: Array of initial standard deviation values for each parameter.
Return type:: np.ndarray

Example

>>> optimizer = MyOptimizer(parameters={'x': (0, 10), 'y': (-5, 5)}, ...)
>>> optimizer.get_init_sigmas
array([2.5, 2.5])  # 1/4 of parameter ranges

property get_norm_bounds: list

Calculate normalized parameter bounds.

Normalizes the parameter bounds by dividing by the initial sigma values. This creates a unified scale for parameters with different ranges.

Returns:: List of (normalized_min, normalized_max) tuples for each parameter.
Return type:: list

Example

>>> optimizer = MyOptimizer(parameters={'x': (0, 10), 'y': (-5, 5)}, ...)
>>> optimizer.get_norm_bounds
[(0.0, 4.0), (-2.0, 2.0)]  # Bounds divided by sigmas

property mean_error: list

Get the historical mean error values.

Returns:: Copy of the mean error history list.
Return type:: list

Example

>>> optimizer.mean_error
[10.5, 8.2, 6.7, 4.3, 2.1]  # Error trajectory across epochs

property mean_params: dict

Get the historical mean parameter values.

Returns:

Dictionary with parameter names as keys and lists of historical: mean values as values. Each value is a copy of the internal list.

Return type:

dict

Example

>>> optimizer.mean_params
{'x': [5.2, 5.1, 5.05, 5.01], 'y': [0.5, 0.3, 0.1, 0.05]}

property mean_targets: dict

Get the historical mean target values.

Returns:

Dictionary with target names as keys and lists of historical: mean observed values as values. Each value is a copy of the internal list.

Return type:

dict

Example

>>> optimizer.mean_targets
{'stress': [250, 240, 230, 225], 'weight': [120, 118, 117, 116.5]}

property norm_sigmas: dict

Get the historical normalized sigma values.

Normalized sigmas represent the standard deviation divided by initial sigma, serving as a measure of convergence.

Returns:

Dictionary with parameter names as keys and lists of historical: normalized sigma values as values. Each value is a copy of the internal list.

Return type:

dict

Example

>>> optimizer.norm_sigmas
{'x': [1.0, 0.7, 0.5, 0.2], 'y': [1.0, 0.8, 0.5, 0.3]}

abstract optimize()[source]

Run the optimization process.

Executes the optimization algorithm until termination criteria are met.

Returns:: A representation of the best solution found.

Note

This is an abstract method that must be implemented by subclasses.

Example

>>> class MyCmaesOptimizer(BaseOptimizer):
...     def optimize(self):
...         self.setup_opt(epoch=self.start_epoch)
...         while not self.check_termination():
...             solutions = self.es.ask()
...             errors = self.process_batch(solutions)
...             self.es.tell(solutions, errors)
...             self.current_epoch += 1
...         return {p: v[-1] for p, v in self.mean_params.items()}

print_epoch(mean_error: float, sigma_error: float, mean_params: ndarray, sigma_params: ndarray, norm_sigmas: ndarray) → None[source]

Print statistics for the current epoch.

Displays aggregated statistics for the current epoch, including mean error, error standard deviation, and parameter statistics. Only prints if verbose mode is enabled.

Parameters:

mean_error – Mean error across all solutions in the epoch.
sigma_error – Standard deviation of errors in the epoch.
mean_params – Array of mean parameter values.
sigma_params – Array of parameter standard deviations.
norm_sigmas – Array of normalized sigma values.

print_solution(sol_id: int, params: ndarray, error: float) → None[source]

Print information about a single evaluated solution.

Displays details of a solution evaluation, including the solution ID, parameter values, and error. Only prints if verbose mode is enabled.

Parameters:

sol_id – Solution identifier within the current epoch.
params – Array of parameter values used for this solution.
error – Error value for the solution (lower is better).

process_batch(solutions: list) → list[source]

Process a batch of solutions in parallel or serial mode.

This method handles the evaluation of multiple parameter sets (solutions) using the provided evaluator function. It supports both serial and parallel processing based on max_workers setting.

Parameters:: solutions – List of parameter value arrays to evaluate.
Returns:: List of error values for each solution.
Return type:: list
Raises:: Exception – If solution evaluation fails.

Example

>>> # Generate solutions with some algorithm
>>> solutions = [[1.2, 3.4], [2.3, 1.4], [3.1, 2.8]]
>>> errors = optimizer.process_batch(solutions)
>>> print(errors)
[12.5, 8.7, 15.2]  # Error value for each solution

rescale_params(params: ndarray) → ndarray[source]

Rescale normalized parameters to their original scale.

Converts normalized parameter values (used internally by optimization algorithms) back to their original scale by multiplying by the initial sigma values.

Parameters:: params – Normalized parameter values.
Returns:: Parameter values in their original scale.
Return type:: np.ndarray

Example

>>> normalized_params = np.array([1.0, -0.5])
>>> optimizer.rescale_params(normalized_params)
array([2.5, -1.25])  # Assuming init_sigmas = [2.5, 2.5]

abstract setup_opt(epoch: int | None = None)[source]

Set up the optimization algorithm.

Initializes or reinitializes the optimization algorithm, potentially starting from a specific epoch (for resuming interrupted optimizations).

Parameters:: epoch – Starting epoch number. If None, starts from 0.

Note

This is an abstract method that must be implemented by subclasses.

Example

>>> class MyCmaesOptimizer(BaseOptimizer):
...     def setup_opt(self, epoch=None):
...         self.es = cma.CMAEvolutionStrategy(
...             self.init_params,
...             0.5,
...             {'seed': self.rand_seed}
...         )
...         if epoch is not None:
...             # Load state from checkpoint if available
...             self.es = self.dir_manager.load_checkpoint(epoch)
...             self.current_epoch = epoch

property sigma_error: list

Get the historical error standard deviations.

Returns:: Copy of the error standard deviation history list.
Return type:: list

Example

>>> optimizer.sigma_error
[5.2, 4.3, 3.1, 2.5, 1.2]  # Error variance trajectory

property sigma_params: dict

Get the historical parameter standard deviations.

Returns:

Dictionary with parameter names as keys and lists of historical: standard deviation values as values. Each value is a copy of the internal list.

Return type:

dict

Example

>>> optimizer.sigma_params
{'x': [2.0, 1.5, 1.0, 0.5], 'y': [1.0, 0.7, 0.4, 0.2]}

property sigma_targets: dict

Get the historical target standard deviations.

Returns:

Dictionary with target names as keys and lists of historical: standard deviation values as values. Each value is a copy of the internal list.

Return type:

dict

Example

>>> optimizer.sigma_targets
{'stress': [25, 20, 15, 10], 'weight': [8, 6, 4, 3]}

Base optimizer (evopt.base_optimizer)

Base optimizer (`evopt.base_optimizer`)