All Optimizers

Subpackages

mealpy.optimizer module

class mealpy.optimizer.Optimizer(**kwargs)[source]

Bases: object

The base class of all algorithms. All methods in this class will be inherited

Note

  • The function solve() is the most important method, trained the model

  • The parallel (multithreading or multiprocessing) is used in method: generate_population(), update_target_for_population()

  • The general format of:

    • population = [agent_1, agent_2, …, agent_N]

    • agent = [solution, target]

    • target = [fitness value, objective_list]

    • objective_list = [obj_1, obj_2, …, obj_M]

AVAILABLE_MODES = ['process', 'thread', 'swarm']
EPSILON = 1e-09
OPT_INFO: ClassVar[mealpy.utils.opt_info.OptInfo | None] = None
PARALLEL_MODES = ['process', 'thread']
SUPPORTED_ARRAYS = [<class 'list'>, <class 'tuple'>, <class 'numpy.ndarray'>]
SUPPORTED_MODES = ['process', 'thread', 'swarm', 'single']
after_initialization() → None[source]
amend_solution(solution: ndarray) → ndarray[source]

This function is based on optimizer’s strategy. In each optimizer, this function can be overridden

Parameters

solution – The position

Returns

The valid solution based on optimizer’s strategy

before_initialization(starting_solutions: Optional[Union[List, Tuple, ndarray]] = None) → None[source]
Parameters

starting_solutions – The starting solutions (not recommended)

before_main_loop()[source]
check_mode_and_workers(mode, n_workers)[source]
check_problem(problem, seed) → None[source]
check_termination(mode='start', termination=None, epoch=None)[source]
static compare_fitness(fitness_x: Union[float, int], fitness_y: Union[float, int], minmax: str = 'min') → bool[source]
static compare_target(target_x: Target, target_y: Target, minmax: str = 'min') → bool[source]
correct_solution(solution: ndarray) → ndarray[source]

This function is based on optimizer’s strategy and problem-specific condition DO NOT override this function

Parameters

solution – The position

Returns

The correct solution that can be used to calculate target

crossover_arithmetic(dad_pos=None, mom_pos=None)[source]
Parameters

  • dad_pos – position of dad

  • mom_pos – position of mom

Returns

position of 1st and 2nd child

Return type

list

static duplicate_pop(pop: List[Agent]) → List[Agent][source]
evolve(epoch: int) → None[source]
generate_agent(solution: Optional[ndarray] = None) → Agent[source]

Generate new agent with full information

Parameters

solution (np.ndarray) – The solution

generate_empty_agent(solution: Optional[ndarray] = None) → Agent[source]

Generate new agent with solution

Parameters

solution (np.ndarray) – The solution

generate_group_population(pop: List[Agent], n_groups: int, m_agents: int) → List[source]

Generate a list of group population from pop

Parameters

  • pop – The current population

  • n_groups – The n groups

  • m_agents – The m agents in each group

Returns

A list of group population

generate_opposition_solution(agent: Optional[Agent] = None, g_best: Optional[Agent] = None) → ndarray[source]
Parameters

  • agent – The current agent

  • g_best – the global best agent

Returns

The opposite solution

generate_population(pop_size: Optional[int] = None) → List[Agent][source]
Parameters

pop_size (int) – number of solutions

Returns

population or list of solutions/agents

Return type

list

get_attributes() → Dict[source]

Get all attributes in optimizer.

static get_best_agent(pop: List[Agent], minmax: str = 'min') → Tuple[Agent, int][source]
Parameters

  • pop – The population of agents

  • minmax – The type of problem

Returns

The best agent The index of best agent

static get_better_agent(agent_x: Agent, agent_y: Agent, minmax: str = 'min', reverse: bool = False) → Agent[source]
Parameters

  • agent_x – First agent

  • agent_y – Second agent

  • minmax – The problem type

  • reverse – Reverse the minmax

Returns

The better agent based on fitness

get_index_kway_tournament_selection(pop: Optional[List] = None, k_way: float = 0.2, output: int = 2, reverse: bool = False) → List[source]
Parameters

  • pop – The population

  • k_way (float/int) – The percent or number of solutions are randomized pick

  • output (int) – The number of outputs

  • reverse (bool) – set True when finding the worst fitness

Returns

List of the selected indexes

Return type

list

get_index_roulette_wheel_selection(list_fitness: array)[source]

This method can handle min/max problem, and negative or positive fitness value.

Parameters

list_fitness (nd.array) – 1-D numpy array

Returns

Index of selected solution

Return type

int

get_levy_flight_step(beta: float = 1.0, multiplier: float = 0.001, size: Optional[Union[List, Tuple, ndarray]] = None, case: int = 0) → Union[float, List, ndarray][source]

Get the Levy-flight step size

Parameters

  • beta (float) –

    Should be in range [0, 2].

    • 0-1: small range –> exploit

    • 1-2: large range –> explore

  • multiplier (float) – default = 0.001

  • size (tuple, list) – size of levy-flight steps, for example: (3, 2), 5, (4, )

  • case (int) –

    Should be one of these value [0, 1, -1].

    • 0: return multiplier * s * self.generator.uniform()

    • 1: return multiplier * s * self.generator.normal(0, 1)

    • -1: return multiplier * s

Returns

The step size of Levy-flight trajectory

Return type

float, list, np.ndarray

get_name() → str[source]

Get name of the optimizer

get_parameters() → Dict[source]

Get parameters of optimizer.

static get_sorted_and_trimmed_population(pop: Optional[List[Agent]] = None, pop_size: Optional[int] = None, minmax: str = 'min') → List[Agent][source]
Parameters

  • pop – The population

  • pop_size – The number of selected agents

  • minmax – The problem type

Returns

The sorted and trimmed population with pop_size size

static get_sorted_population(pop: List[Agent], minmax: str = 'min') → Tuple[List[Agent], List][source]

Get sorted population based on type (minmax) of problem

Parameters

  • pop – The population

  • minmax – The type of the problem

Returns

Sorted population (1st agent is the best, last agent is the worst) Sorted index

static get_special_agents(pop: Optional[List[Agent]] = None, n_best: int = 3, n_worst: int = 3, minmax: str = 'min', return_index: bool = False) → Tuple[List[Agent], Optional[Union[List[Agent], List[int]]], Optional[Union[List[Agent], List[int]]]][source]

Get special agents including sorted population, n1 best agents (or indices), n2 worst agents (or indices).

Parameters

  • pop – The original population.

  • n_best – Top n1 best agents, default n1=3, good level reduction.

  • n_worst – Top n2 worst agents, default n2=3, worst level reduction.

  • minmax – The problem type.

  • return_index – Return the original index instead of the Agent object.

Returns

The sorted_population, n1 best agents (or indices), and n2 worst agents (or indices).

static get_special_fitness(pop: Optional[List[Agent]] = None, minmax: str = 'min') → Tuple[Union[float, ndarray], float, float][source]

Get special target include the total fitness, the best fitness, and the worst fitness

Parameters

  • pop – The population

  • minmax – The problem type

Returns

The total fitness, the best fitness, and the worst fitness

get_target(solution: ndarray, counted: bool = True) → Target[source]

Get target value

Parameters

  • solution – The real-value solution

  • counted – Indicating the number of function evaluations is increasing or not

Returns

The target value

static get_worst_agent(pop: List[Agent], minmax: str = 'min') → Tuple[Agent, int][source]
Parameters

  • pop – The population of agents

  • minmax – The type of problem

Returns

The worst agent The index of worst agent (Optional)

static greedy_selection_population(pop_old: Optional[List[Agent]] = None, pop_new: Optional[List[Agent]] = None, minmax: str = 'min') → List[Agent][source]
Parameters

  • pop_old – The current population

  • pop_new – The next population

  • minmax – The problem type

Returns

The new population with better solutions

improved_ms(pop=None, g_best=None)[source]
initialization() → None[source]
initialize_variables()[source]
static sample_indexes_exclude_list(generator, pop_size: int, exclude_list: Union[list, ndarray], n_samples: int, replace: bool = False)[source]

Randomly select ‘n_samples’ unique indices excluding ‘exclude_list’ using Boolean Masking. Returns a scalar if n_samples == 1, otherwise returns a numpy array.

static sample_indexes_exclude_one(generator, pop_size: int, exclude_idx: int, n_samples: int, replace: bool = False)[source]

Randomly select ‘n_samples’ unique indices excluding ‘exclude_idx’ using the highly optimized Offset Trick. Returns a scalar if n_samples == 1, otherwise returns a numpy array.

set_parameters(parameters: Union[List, Tuple, Dict]) → None[source]

Set the parameters for current optimizer.

if paras is a list of parameter’s name, then it will set the default value in optimizer as current parameters if paras is a dict of parameter’s name and value, then it will override the current parameters

Parameters

parameters – The parameters

solve(problem: Optional[Union[Dict, Problem]] = None, mode: str = 'single', n_workers: Optional[int] = None, termination: Optional[Union[Dict, Termination]] = None, starting_solutions: Optional[Union[List, Tuple, ndarray]] = None, seed: Optional[int] = None) → Agent[source]
Parameters

  • problem – an instance of Problem class or a dictionary

  • mode –

    Parallel: ‘process’, ‘thread’; Sequential: ‘swarm’, ‘single’.

    • ’process’: The parallel mode with multiple cores run the tasks

    • ’thread’: The parallel mode with multiple threads run the tasks

    • ’swarm’: The sequential mode that no effect on updating phase of other agents

    • ’single’: The sequential mode that effect on updating phase of other agents, this is default mode

  • n_workers – The number of workers (cores or threads) to do the tasks (effect only on parallel mode)

  • termination – The termination dictionary or an instance of Termination class

  • starting_solutions – List or 2D matrix (numpy array) of starting positions with length equal pop_size parameter

  • seed – seed for random number generation needed to be explicitly set to int value

Returns

g_best, the best found agent, that hold the best solution and the best target. Access by: .g_best.solution, .g_best.target

Return type

g_best

track_optimize_process() → None[source]

Save some historical data after training process finished

track_optimize_step(pop: Optional[List[Agent]] = None, epoch: Optional[int] = None, runtime: Optional[float] = None) → None[source]

Save some historical data and print out the detailed information of training process in each epoch

Parameters

  • pop – the current population

  • epoch – current iteration

  • runtime – the runtime for current iteration

update_global_best_agent(pop: List[Agent], save: bool = True) → Union[List, Tuple][source]

Update global best and current best solutions in history object. Also update global worst and current worst solutions in history object.

Parameters

  • pop (list) – The population of pop_size individuals

  • save (bool) – True if you want to add new current/global best to history, False if you just want to update current/global best

Returns

Sorted population and the global best solution

Return type

list

update_target_for_population(pop: Optional[List[Agent]] = None) → List[Agent][source]

Update target for the input population

Parameters

pop – the population of agents

Returns

population with updated target value

Return type

list

mealpy.tuner module

class mealpy.tuner.ParameterGrid(param_grid)[source]

Bases: object

Please check out this class from the scikit-learn library.

It represents a grid of parameters with a discrete number of values for each parameter. This class is useful for iterating over parameter value combinations using the Python built-in function iter, and the generated parameter combinations’ order is deterministic.

Parameters

param_grid (dict of str to sequence, or sequence of such) –

The parameter grid to explore, as a dictionary mapping estimator parameters to sequences of allowed values.

An empty dict signifies default parameters.

A sequence of dicts signifies a sequence of grids to search, and is useful to avoid exploring parameter combinations that make no sense or have no effect. See the examples below.

Examples

>>> from mealpy.tuner import ParameterGrid
>>> param_grid = {'a': [1, 2], 'b': [True, False]}
>>> list(ParameterGrid(param_grid)) == ([{'a': 1, 'b': True}, {'a': 1, 'b': False}, {'a': 2, 'b': True}, {'a': 2, 'b': False}])
True

>>> grid = [{'kernel': ['linear']}, {'kernel': ['rbf'], 'gamma': [1, 10]}]
>>> list(ParameterGrid(grid)) == [{'kernel': 'linear'}, {'kernel': 'rbf', 'gamma': 1}, {'kernel': 'rbf', 'gamma': 10}]
True
>>> ParameterGrid(grid)[1] == {'kernel': 'rbf', 'gamma': 1}
True
class mealpy.tuner.Tuner(algorithm: Optional[Union[str, Optimizer]] = None, param_grid: Optional[Union[Dict, List]] = None, **kwargs: object)[source]

Bases: object

Hyperparameter tuning utility for optimization algorithms.

Note

This class provides an automated mechanism to tune hyper-parameters for optimization algorithms, serving as a specialized alternative to scikit-learn’s GridSearchCV. It supports concurrent execution and result exportation in formats like CSV, JSON, and Pandas DataFrame.

Parameters

  • algorithm (Optimizer or str) – The optimization algorithm instance to tune.

  • param_grid (dict or list of dict) – Dictionary with parameter names (str) as keys and lists of parameter settings to try as values, or a list of such dictionaries.

Note

The core workflow revolves around calling the execute() method to perform the tuning process, followed by resolve() to extract or re-run the optimal configuration.

Examples

>>> from opfunu.cec_based.cec2017 import F52017
>>> from mealpy import FloatVar, BBO, Tuner
>>>
>>> f1 = F52017(30, f_bias=0)
>>>
>>> p1 = {
>>>     "bounds": FloatVar(lb=f1.lb, ub=f1.ub),
>>>     "obj_func": f1.evaluate,
>>>     "minmax": "min",
>>>     "name": "F5",
>>>     "log_to": "console",
>>> }
>>>
>>> paras_bbo_grid = {
>>>     "epoch": [10, 20, 30],
>>>     "pop_size": [30, 50, 100],
>>>     "n_elites": [2, 3, 4, 5],
>>>     "p_m": [0.01, 0.02, 0.05]
>>> }
>>> term = {
>>>     "max_epoch": 200,
>>>     "max_time": 20,
>>>     "max_fe": 10000
>>> }
>>> if __name__ == "__main__":
>>>     model = BBO.OriginalBBO()
>>>     tuner = Tuner(model, paras_bbo_grid)
>>>     tuner.execute(problem=p1, termination=term, n_trials=5, n_jobs=4, mode="thread", n_workers=6, verbose=True)
>>>
>>>     print(tuner.best_row)
>>>     print(tuner.best_score)
>>>     print(tuner.best_params)
>>>     print(type(tuner.best_params))
>>>
>>>     print(tuner.best_algorithm)
>>>     tuner.export_results(save_path="history/results", save_as="csv")
>>>     tuner.export_figures()
>>>
>>>     g_best = tuner.resolve(mode="thread", n_workers=4, termination=term)
>>>     print(g_best.solution, g_best.target.fitness)
>>>     print(tuner.algorithm.problem.get_name())
>>>     print(tuner.best_algorithm.get_name())
property best_algorithm
property best_params
property best_row
property best_score
execute(problem: Optional[Union[Dict, Problem]] = None, termination: Optional[Union[Dict, Termination]] = None, n_trials: int = 2, n_jobs: Optional[int] = None, mode: str = 'single', n_workers: int = 2, verbose: bool = True) → None[source]

Execute the Tuner utility to evaluate hyperparameter configurations.

Parameters

  • problem (dict or Problem, optional) – An instance of the Problem class or a problem configuration dictionary.

  • termination (dict or Termination, optional) – An instance of the Termination class or a termination configuration dictionary.

  • n_trials (int, default=2) – The number of independent trials to run on the problem for each parameter set.

  • n_jobs (int, optional) – Number of processes to speed up the task by running multiple trials concurrently. If <= 1 or None, executes sequentially. If >= 2, executes in parallel.

  • mode ({'single', 'swarm', 'thread', 'process'}, default='single') – The execution mode applied to the current problem.

  • n_workers (int, default=2) – Number of processes or threads to use if mode is ‘thread’ or ‘process’.

  • verbose (bool, default=True) – Switch for verbose logging of the tuning process.

Raises

TypeError – If the problem type is neither a dictionary nor an instance of the Problem class.

export_figures(save_path=None, file_name='tuning_epoch_fit.csv', color=None, x_label=None, y_label=None, exts=('.png', '.pdf'), verbose=False)[source]

Export results to various file type

Parameters

  • save_path (str) – The path to the folder, default None

  • file_name (str) – The file name (with file type, e.g. dataframe, json, csv; default: “tuning_epoch_fit.csv”) that hold results

Raises

TypeError – Raises TypeError if export type is not supported

export_results(save_path=None, file_name='tuning_best_fit.csv')[source]

Export results to various file type

Parameters

  • save_path (str) – The path to the folder, default None

  • file_name (str) – The file name (with file type, e.g. dataframe, json, csv; default: “tuning_best_fit.csv”) that hold results

Raises

TypeError – Raises TypeError if export type is not supported

resolve(mode: str = 'single', starting_solutions: Optional[Union[List, Tuple, ndarray]] = None, n_workers: Optional[int] = None, termination: Optional[Union[Dict, Termination]] = None) → Agent[source]

Resolve the problem using the optimal hyperparameters found during tuning.

Parameters

  • mode ({'single', 'swarm', 'thread', 'process'}, default='single') – The execution mode for the solver: * ‘process’: Parallel mode using multiple CPU cores. * ‘thread’: Parallel mode using multiple threads. * ‘swarm’: Sequential mode that has no effect on the updating phase of other agents. * ‘single’: Sequential mode that affects the updating phase of other agents.

  • starting_solutions (list, tuple, or np.ndarray, optional) – A 1D list or 2D matrix of starting positions with a length equal to the algorithm’s pop_size parameter.

  • n_workers (int, optional) – The number of parallel workers (cores or threads) to perform the tasks. Effective only when mode is ‘thread’ or ‘process’.

  • termination (dict or Termination, optional) – A termination configuration dictionary or an instance of the Termination class.

Returns

g_best – The best agent found after resolving the problem.

Return type

Agent

mealpy.multitask module

class mealpy.multitask.Multitask(algorithms: Optional[Union[List, Tuple]] = None, problems: Optional[Union[List, Tuple]] = None, terminations: Optional[Union[List, Tuple]] = None, modes: Optional[Union[List, Tuple]] = None, n_workers: Optional[int] = None, **kwargs: object)[source]

Bases: object

Multitask utility class.

This feature enables the execution of multiple algorithms across multiple problems and trials. Additionally, it allows for exporting results in various formats such as Pandas DataFrame, JSON, and CSV.

Parameters

  • algorithms (list, tuple) – List of algorithms to run

  • problems (list, tuple) – List of problems to run

  • terminations (list, tuple) – List of terminations to apply on algorithm/problem

  • modes (list, tuple) – List of modes to apply on algorithm/problem

  • n_workers (int) – Number of workers (threads or processes) to apply on algorithm/problem. Only effect when mode is thread or process.

Examples

>>> ## Import libraries
>>> from opfunu.cec_based.cec2017 import F52017, F102017, F292017
>>> from mealpy import FloatVar
>>> from mealpy import BBO, DE
>>> from mealpy import Multitask

>>> ## Define your own problems
>>> f1 = F52017(30, f_bias=0)
>>> f2 = F102017(30, f_bias=0)
>>> f3 = F292017(30, f_bias=0)
>>> p1 = {
>>>     "bounds": FloatVar(lb=f1.lb, ub=f1.ub),
>>>     "obj_func": f1.evaluate,
>>>     "minmax": "min",
>>>     "name": "F5",
>>>     "log_to": "console",
>>> }
>>> p2 = {
>>>     "bounds": FloatVar(lb=f2.lb, ub=f2.ub),
>>>     "obj_func": f2.evaluate,
>>>     "minmax": "min",
>>>     "name": "F10",
>>>     "log_to": "console",
>>> }
>>> p3 = {
>>>     "bounds": FloatVar(lb=f3.lb, ub=f3.ub),
>>>     "obj_func": f3.evaluate,
>>>     "minmax": "min",
>>>     "name": "F29",
>>>     "log_to": "console",
>>> }

>>> ## Define optimizers
>>> optimizer1 = BBO.DevBBO(epoch=10000, pop_size=50)
>>> optimizer2 = BBO.OriginalBBO(epoch=10000, pop_size=50)
>>> optimizer3 = DE.OriginalDE(epoch=10000, pop_size=50)
>>> optimizer4 = DE.SAP_DE(epoch=10000, pop_size=50)

>>> ## Define termination if needed
>>> term = {
>>>     "max_fe": 30000
>>> }

>>> ## Define and run Multitask
>>> if __name__ == "__main__":
>>>     multitask = Multitask(algorithms=(optimizer1, optimizer2, optimizer3, optimizer4), problems=(p1, p2, p3), terminations=(term, ), modes=("thread", ), n_workers=4)
>>>     # default modes = "single", default termination = epoch (as defined in problem dictionary)
>>>     multitask.execute(n_trials=5, n_jobs=None, save_path="history", save_as="csv", save_convergence=True, verbose=False)
>>>     # multitask.execute(n_trials=5, save_path="history", save_as="csv", save_convergence=True, verbose=False)
check_input(name=None, values=None, kind=None)[source]
execute(n_trials: int = 2, n_jobs: Optional[int] = None, save_path: str = 'history', save_as: str = 'csv', save_convergence: bool = False, verbose: bool = False) → None[source]

Execute the multitask utility to run multiple algorithms across multiple problems.

This method automatically handles the execution of predefined algorithms on predefined problems for a specified number of trials. It supports parallel execution and exports both the best fitness results and convergence history.

Parameters

  • n_trials (int, default=2) – The number of independent repetitions for each algorithm-problem pair.

  • n_jobs (int, optional) – Number of CPU processes used to speed up computation. If <= 1 or None, executes sequentially. If >= 2, executes in parallel.

  • save_path (str, default='history') – The directory path where the execution results and convergence logs will be saved.

  • save_as ({'csv', 'json', 'dataframe'}, default='csv') – The format of the exported result files.

  • save_convergence (bool, default=False) – If True, saves the convergence (fitness history) during generations for each trial.

  • verbose (bool, default=False) – If True, prints detailed logging information during the execution process.

Return type

None

Raises

ValueError – If the save_as format is not supported (not in ‘csv’, ‘json’, ‘dataframe’).

static export_to_csv(result: DataFrame, save_path: str)[source]
static export_to_dataframe(result: DataFrame, save_path: str)[source]
static export_to_json(result: DataFrame, save_path: str)[source]