Strategies

deap_er.strategies

StrategyMultiObjective(population, sigma, **kwargs)

Multi-objective Covariance Matrix Adaptation evolution strategy.

Parameters:

Name	Type	Description	Default
population	list[Individual]	Initial parent population.	required
sigma	float	Initial step size for every parent.	required
**kwargs	Any	Optional strategy parameters. See the table below.	{}

.. dropdown:: Table of Kwargs :margin: 0 5 0 0

• offsprings - (int)
  • The number of children to produce at each generation.
  • Default: 1
• survivors - (int)
  • The number of parents to keep for the next generation.
  • Default: len(population)
• ss_dmp - (float)
  • Damping of the step-size.
  • Default: 1.0 + len(population[0]) / 2.0
• th_cum - (float)
  • Time horizon of the cumulative contribution.
  • Default: 2.0 / (len(population[0]) + 2.0)
• tgt_sr - (float)
  • Target success rate.
  • Default: 1.0 / 5.5
• thresh_sr - (float)
  • Threshold success rate.
  • Default: 0.44
• ss_learn_rate - (float)
  • Learning rate of the step-size.
  • Default: tgt_sr / (2.0 + tgt_sr)
• cm_learn_rate - (float)
  • Learning rate of the covariance matrix.
  • Default: 2.0 / (len(population[0]) ** 2 + 6.0)
• low, up - (float or sequence)
  • Optional box bounds on generated individuals.
• bound_mode - (str)
  • clip (default) or resample. Both are constraint-handling approximations; the update treats the repaired point as the sample.
• resample_limit - (int)
  • Failed redraws before clipping one sample. Default: 100

See the class docstring.

Source code in deap_er/private/strategies/cma_multi_objective.py

def __init__(self, population: list[Individual], sigma: float, **kwargs: Any) -> None:
    """See the class docstring."""
    self.parents = population
    self.dim = len(self.parents[0])
    pop_size = len(population)

    self.mu: int
    self.lamb: int
    self.ss_dmp: float
    self.tgt_sr: float
    self.ss_learn_rate: float
    self.th_cum: float
    self.cm_learn_rate: float
    self.thresh_sr: float
    self.low: Any
    self.up: Any
    self.bound_mode: str
    self.resample_limit: int

    self.compute_params(**kwargs)

    self.sigmas = [sigma] * pop_size
    self.big_a = [numpy.identity(self.dim) for _ in range(pop_size)]
    self.inv_cholesky = [numpy.identity(self.dim) for _ in range(pop_size)]
    self.pc = [numpy.zeros(self.dim) for _ in range(pop_size)]
    self.psucc = [self.tgt_sr] * pop_size

compute_params(**kwargs)

Recompute strategy parameters from kwargs.

Called from the constructor. Call again if offsprings or survivors changes during evolution.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Optional strategy parameters. See the class docstring. Parameters given to earlier calls stay pinned unless overridden here.	{}

Source code in deap_er/private/strategies/cma_multi_objective.py

def compute_params(self, **kwargs: Any) -> None:
    """Recompute strategy parameters from ``kwargs``.

    Called from the constructor. Call again if ``offsprings`` or
    ``survivors`` changes during evolution.

    Args:
        **kwargs: Optional strategy parameters. See the class
            docstring. Parameters given to earlier calls stay
            pinned unless overridden here.
    """
    kwargs, self.hyperparams = merge_hyperparams(self, kwargs, MO_HYPERPARAMS)
    self.mu = kwargs.get("survivors", len(self.parents))
    self.lamb = kwargs.get("offsprings", 1)
    self.ss_dmp = kwargs.get("ss_dmp", 1.0 + self.dim / 2.0)
    self.tgt_sr = kwargs.get("tgt_sr", 1.0 / (5.0 + 0.5))
    self.ss_learn_rate = kwargs.get("ss_learn_rate", self.tgt_sr / (2.0 + self.tgt_sr))
    self.th_cum = kwargs.get("th_cum", 2.0 / (self.dim + 2.0))
    self.cm_learn_rate = kwargs.get("cm_learn_rate", 2.0 / (self.dim**2 + 6.0))
    self.thresh_sr = kwargs.get("thresh_sr", 0.44)
    update_bound_attrs(self, kwargs)

reset_state(parents, sigma, **kwargs)

Reset mutable CMA state for a restart.

Hyperparameters and bounds from earlier calls are kept.

Source code in deap_er/private/strategies/cma_multi_objective.py

def reset_state(
    self,
    parents: list[Individual],
    sigma: float,
    **kwargs: Any,
) -> None:
    """Reset mutable CMA state for a restart.

    Hyperparameters and bounds from earlier calls are kept.
    """
    self.compute_params(**kwargs)
    self.parents = parents[: self.mu]
    pop_size = len(self.parents)
    self.sigmas = [sigma] * pop_size
    self.big_a = [numpy.identity(self.dim) for _ in range(pop_size)]
    self.inv_cholesky = [numpy.identity(self.dim) for _ in range(pop_size)]
    self.pc = [numpy.zeros(self.dim) for _ in range(pop_size)]
    self.psucc = [self.tgt_sr] * pop_size

update(population)

Select new parents and update each parent's CMA parameters.

Offspring are merged with the current parents, then reduced to survivors by non-dominated sorting of candidates with valid fitness. Step-size and covariance are updated per successful parent.

Parameters:

Name	Type	Description	Default
population	list[Individual]	Evaluated individuals from generate.	required

Source code in deap_er/private/strategies/cma_multi_objective.py

def update(self, population: list[Individual]) -> None:
    """Select new parents and update each parent's CMA parameters.

    Offspring are merged with the current parents, then reduced to
    ``survivors`` by non-dominated sorting of candidates with
    valid fitness. Step-size and covariance are updated per
    successful parent.

    Args:
        population: Evaluated individuals from ``generate``.
    """
    candidates = [ind for ind in population + self.parents if ind.fitness.is_valid()]
    chosen, not_chosen = select(self, candidates)
    last_steps, sigmas, inv_cholesky, big_a, pc, psucc = copy_offspring_state(self, chosen)
    update_chosen_offspring(self, chosen, last_steps, sigmas, inv_cholesky, big_a, pc, psucc)
    decay_rejected_offspring(self, not_chosen, chosen)
    commit_parent_params(self, chosen, sigmas, inv_cholesky, big_a, pc, psucc)
    self.parents = chosen

generate(ind_init)

Sample offsprings individuals from the current parents.

When offsprings equals the parent count and that many parents exist, each parent produces one child. Otherwise parents are drawn from the first non-dominated front, or from every parent if any parent fitness is invalid.

Parameters:

Name	Type	Description	Default
ind_init	Callable[..., Individual]	Callable that turns a sampled vector into an individual.	required

Returns:

Type	Description
list[Individual]	Newly sampled individuals.

Source code in deap_er/private/strategies/cma_multi_objective.py

def generate(self, ind_init: Callable[..., Individual]) -> list[Individual]:
    """Sample ``offsprings`` individuals from the current parents.

    When ``offsprings`` equals the parent count and that many
    parents exist, each parent produces one child. Otherwise
    parents are drawn from the first non-dominated front, or from
    every parent if any parent fitness is invalid.

    Args:
        ind_init: Callable that turns a sampled vector into an
            individual.

    Returns:
        Newly sampled individuals.
    """
    arz = rng.standard_normal((self.lamb, self.dim))
    for i, p in enumerate(self.parents):
        p.ps_ = "p", i
    if not self.parents:
        return []
    one_each = self.lamb == self.mu and len(self.parents) >= self.lamb
    if self.bound_mode == "resample" and (self.low is not None or self.up is not None):
        return resample_offspring(self, ind_init, arz, one_each)
    return clip_offspring(self, ind_init, arz, one_each)

StrategyOnePlusLambda(parent, sigma, **kwargs)

One-plus-lambda Covariance Matrix Adaptation evolution strategy.

Parameters:

Name	Type	Description	Default
parent	Individual	Starting individual. Must have a fitness attribute.	required
sigma	float	Initial standard deviation of the distribution.	required
**kwargs	Any	Optional strategy parameters. See the table below.	{}

Raises:

Type	Description
TypeError	If parent has no fitness attribute.

.. dropdown:: Table of Kwargs :margin: 0 5 0 0

• offsprings - (int)
  • The number of children to produce at each generation.
  • Default: 1
• ss_dmp - (float)
  • Damping of the step-size.
  • Default: 1.0 + len(parent) / (2.0 * offsprings)
• th_cum - (float)
  • Time horizon of the cumulative contribution.
  • Default: 2.0 / (len(parent) + 2.0)
• tgt_sr - (float)
  • Target success rate.
  • Default: 1.0 / (5 + sqrt(offsprings) / 2.0)
• thresh_sr - (float)
  • Threshold success rate.
  • Default: 0.44
• ss_learn_rate - (float)
  • Learning rate of the step-size.
  • Default: tgt_sr * offsprings / (2.0 + tgt_sr * offsprings)
• cm_learn_rate - (float)
  • Learning rate of the covariance matrix.
  • Default: 2.0 / (len(parent) ** 2 + 6.0)
• low, up - (float or sequence)
  • Optional box bounds on generated individuals.
• bound_mode - (str)
  • clip (default) or resample. Both are constraint-handling approximations; the update treats the repaired point as the sample.
• resample_limit - (int)
  • Failed redraws before clipping one sample. Default: 100

See the class docstring.

Source code in deap_er/private/strategies/cma_one_plus_lambda.py

def __init__(self, parent: Individual, sigma: float, **kwargs: Any) -> None:
    """See the class docstring."""
    if not hasattr(parent, "fitness"):
        raise TypeError("The parent must have a fitness attribute.")

    self.parent = parent
    self.sigma = sigma

    self.dim = len(self.parent)
    self.big_c = numpy.identity(self.dim)
    self.big_a = numpy.identity(self.dim)
    self.pc = numpy.zeros(self.dim)

    self.lamb: int
    self.thresh_sr: float
    self.ss_dmp: float
    self.tgt_sr: float
    self.ss_learn_rate: float
    self.th_cum: float
    self.cm_learn_rate: float
    self.psucc: float
    self.low: Any
    self.up: Any
    self.bound_mode: str
    self.resample_limit: int

    self.compute_params(**kwargs)

compute_params(**kwargs)

Recompute strategy parameters from kwargs.

Called from the constructor. Call again if offsprings changes during evolution.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Optional strategy parameters. See the class docstring. Parameters given to earlier calls stay pinned unless overridden here.	{}

Source code in deap_er/private/strategies/cma_one_plus_lambda.py

def compute_params(self, **kwargs: Any) -> None:
    """Recompute strategy parameters from ``kwargs``.

    Called from the constructor. Call again if ``offsprings``
    changes during evolution.

    Args:
        **kwargs: Optional strategy parameters. See the class
            docstring. Parameters given to earlier calls stay
            pinned unless overridden here.
    """
    kwargs, self.hyperparams = merge_hyperparams(self, kwargs, ONE_PLUS_LAMBDA_HYPERPARAMS)
    self.lamb = int(kwargs.get("offsprings", 1))
    self.thresh_sr = float(kwargs.get("thresh_sr", 0.44))

    default = 1.0 + self.dim / (2.0 * self.lamb)
    self.ss_dmp = float(kwargs.get("ss_dmp", default))

    default = 1.0 / (5 + sqrt(self.lamb) / 2.0)
    self.tgt_sr = float(kwargs.get("tgt_sr", default))

    default = self.tgt_sr * self.lamb / (2 + self.tgt_sr * self.lamb)
    self.ss_learn_rate = float(kwargs.get("ss_learn_rate", default))

    default = 2.0 / (self.dim + 2.0)
    self.th_cum = float(kwargs.get("th_cum", default))

    default = 2.0 / (self.dim**2 + 6.0)
    self.cm_learn_rate = float(kwargs.get("cm_learn_rate", default))

    self.psucc = self.tgt_sr
    update_bound_attrs(self, kwargs)

reset_state(parent, sigma, **kwargs)

Reset mutable CMA state for a restart.

Hyperparameters and bounds from earlier calls are kept.

Raises:

Type	Description
TypeError	If parent has no fitness attribute.

Source code in deap_er/private/strategies/cma_one_plus_lambda.py

def reset_state(self, parent: Individual, sigma: float, **kwargs: Any) -> None:
    """Reset mutable CMA state for a restart.

    Hyperparameters and bounds from earlier calls are kept.

    Raises:
        TypeError: If ``parent`` has no fitness attribute.
    """
    if not hasattr(parent, "fitness"):
        raise TypeError("The parent must have a fitness attribute.")
    self.parent = parent
    self.sigma = sigma
    self.big_c = numpy.identity(self.dim)
    self.big_a = numpy.identity(self.dim)
    self.pc = numpy.zeros(self.dim)
    self.compute_params(**kwargs)
    self.psucc = self.tgt_sr

generate(ind_init)

Sample offsprings individuals around the current parent.

Parameters:

Name	Type	Description	Default
ind_init	Callable[..., Individual]	Callable that turns a sampled vector into an individual.	required

Returns:

Type	Description
list[Individual]	Newly sampled individuals.

Source code in deap_er/private/strategies/cma_one_plus_lambda.py

def generate(self, ind_init: Callable[..., Individual]) -> list[Individual]:
    """Sample ``offsprings`` individuals around the current parent.

    Args:
        ind_init: Callable that turns a sampled vector into an
            individual.

    Returns:
        Newly sampled individuals.
    """
    return sample_offspring(
        self.parent,
        self.sigma,
        self.big_a,
        self.lamb,
        self.dim,
        ind_init,
        low=self.low,
        up=self.up,
        bound_mode=self.bound_mode,
        resample_limit=self.resample_limit,
    )

update(population)

Update parent, step-size, and covariance from population.

The parent is replaced when a better offspring exists. Success rate drives the step-size; a successful replacement also updates the covariance. An unevaluated parent (no fitness values, as after reset_state / a restart) adopts the best offspring without counting a fake success or adapting sigma or the covariance.

Parameters:

Name	Type	Description	Default
population	list[Individual]	Evaluated individuals from generate.	required

Source code in deap_er/private/strategies/cma_one_plus_lambda.py

def update(self, population: list[Individual]) -> None:
    """Update parent, step-size, and covariance from ``population``.

    The parent is replaced when a better offspring exists. Success
    rate drives the step-size; a successful replacement also
    updates the covariance. An unevaluated parent (no fitness
    values, as after ``reset_state`` / a restart) adopts the best
    offspring without counting a fake success or adapting
    sigma or the covariance.

    Args:
        population: Evaluated individuals from ``generate``.
    """
    if not population:
        return
    population.sort(key=lambda ind: ind.fitness, reverse=True)
    if not self.parent.fitness.is_valid():
        self.parent = copy.deepcopy(population[0])
        return
    lambda_succ = sum(self.parent.fitness <= ind.fitness for ind in population)
    psucc = float(lambda_succ) / self.lamb
    self.psucc = (1 - self.ss_learn_rate) * self.psucc + self.ss_learn_rate * psucc

    if self.parent.fitness <= population[0].fitness:
        x_step = (population[0] - numpy.array(self.parent)) / self.sigma
        self.parent = copy.deepcopy(population[0])
        if self.psucc < self.thresh_sr:
            temp_1 = sqrt(self.th_cum * (2 - self.th_cum))
            self.pc = (1 - self.th_cum) * self.pc + temp_1 * x_step
            temp_1 = numpy.outer(self.pc, self.pc)
            self.big_c = (1 - self.cm_learn_rate) * self.big_c + self.cm_learn_rate * temp_1
        else:
            self.pc = (1 - self.th_cum) * self.pc
            temp_1 = numpy.outer(self.pc, self.pc)
            temp_2 = temp_1 + self.th_cum * (2 - self.th_cum) * self.big_c
            self.big_c = (1 - self.cm_learn_rate) * self.big_c + self.cm_learn_rate * temp_2

    # Kept inline rather than shared with the multi-objective strategy:
    # the two groupings of this expression differ in the last ulp.
    temp_1 = self.psucc - self.tgt_sr
    self.sigma *= exp(1.0 / self.ss_dmp * temp_1 / (1.0 - self.tgt_sr))
    self.big_a = numpy.linalg.cholesky(self.big_c)

StrategySeparable(centroid, sigma, **kwargs)

Bases: CmaCore

Separable CMA-ES with a diagonal covariance (Ros and Hansen, 2008).

Learns one variance per gene. Memory and the generate/update step are O(n). There is no learned correlation. Default rank_one and rank_mu are the Strategy defaults scaled by (n + 2) / 3. Step-size uses the same cumulative step-size adaptation as Strategy. Box bounds and the rest of the keyword surface match Strategy, except cm_init is a length-n variance vector (default ones), not an n-by-n matrix.

Parameters:

Name	Type	Description	Default
centroid	Iterable[float]	Starting point of the search distribution.	required
sigma	float	Initial standard deviation of the distribution.	required
**kwargs	Any	Optional strategy parameters. Shared names follow Strategy (offsprings, survivors, weights, cm_cum, ss_cum, ss_dmp, rank_one, rank_mu, low, up, bound_mode, resample_limit). cm_init must be a length-n variance vector or a scalar broadcast to n.	{}

Raises:

Type	Description
RuntimeError	If weights is not superlinear, linear, or equal.
ValueError	If cm_init is not a length-n vector, any variance is not positive, survivors is not in [1, offsprings], or box-bound kwargs are invalid.

See the class docstring.

Source code in deap_er/private/strategies/cma_separable.py

def __init__(self, centroid: Iterable[float], sigma: float, **kwargs: Any) -> None:
    """See the class docstring."""
    init_cma_state(self, centroid, sigma)
    self.compute_params(**kwargs)

compute_params(**kwargs)

Recompute λ, rates, and the diagonal cm_init vector.

Hyperparameters from earlier calls stay pinned, as in Strategy.compute_params.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Same names as Strategy.compute_params, except cm_init is a length-n variance vector.	{}

Raises:

Type	Description
RuntimeError	If weights is unknown.
ValueError	If cm_init is the wrong shape or not strictly positive, or survivors is not in [1, offsprings].

Source code in deap_er/private/strategies/cma_separable.py

def compute_params(self, **kwargs: Any) -> None:
    """Recompute λ, rates, and the diagonal ``cm_init`` vector.

    Hyperparameters from earlier calls stay pinned, as in
    ``Strategy.compute_params``.

    Args:
        **kwargs: Same names as ``Strategy.compute_params``, except
            ``cm_init`` is a length-``n`` variance vector.

    Raises:
        RuntimeError: If ``weights`` is unknown.
        ValueError: If ``cm_init`` is the wrong shape or not
            strictly positive, or ``survivors`` is not in
            ``[1, offsprings]``.
    """
    params, pinned = merge_hyperparams(self, kwargs, CMA_HYPERPARAMS)
    apply_cma_hyperparams(
        self,
        params,
        rank_scale=(self.dim + 2.0) / 3.0,
        cap_survivors="survivors" not in kwargs,
    )
    self.hyperparams = pinned
    if not hasattr(self, "big_c") or "cm_init" in kwargs:
        self.big_c = _variance_vector(kwargs.get("cm_init", numpy.ones(self.dim)), self.dim)
        self.diag_d = numpy.sqrt(self.big_c)
        self.cond = float(numpy.max(self.big_c) / numpy.min(self.big_c))
    update_bound_attrs(self, kwargs)

reset_state(centroid, sigma, **kwargs)

Reset mutable CMA state for a restart.

Hyperparameters and bounds from earlier calls are kept.

Source code in deap_er/private/strategies/cma_separable.py

def reset_state(
    self,
    centroid: Iterable[float],
    sigma: float,
    **kwargs: Any,
) -> None:
    """Reset mutable CMA state for a restart.

    Hyperparameters and bounds from earlier calls are kept.
    """
    reset_cma_state(self, centroid, sigma)
    self.compute_params(cm_init=numpy.ones(self.dim), **kwargs)

generate(ind_init)

Draw lamb axis-aligned samples and apply box bounds.

Parameters:

Name	Type	Description	Default
ind_init	Callable[..., Individual]	Builds an individual from a length-n vector.	required

Returns:

Type	Description
list[Individual]	The sampled population.

Source code in deap_er/private/strategies/cma_separable.py

def generate(self, ind_init: Callable[..., Individual]) -> list[Individual]:
    """Draw ``lamb`` axis-aligned samples and apply box bounds.

    Args:
        ind_init: Builds an individual from a length-``n`` vector.

    Returns:
        The sampled population.
    """
    return generate_cma_offspring(self, self.diag_d, ind_init)

update(population)

Update centroid, step-size, and diagonal covariance.

Individuals are ranked by fitness. The best survivors members drive the update.

Parameters:

Name	Type	Description	Default
population	list[Individual]	Evaluated individuals from generate.	required

Raises:

Type	Description
FloatingPointError	If the update gives a non-finite state, as Strategy.update does.

Source code in deap_er/private/strategies/cma_separable.py

def update(self, population: list[Individual]) -> None:
    """Update centroid, step-size, and diagonal covariance.

    Individuals are ranked by fitness. The best ``survivors``
    members drive the update.

    Args:
        population: Evaluated individuals from ``generate``.

    Raises:
        FloatingPointError: If the update gives a non-finite state,
            as ``Strategy.update`` does.
    """
    snapshot = snapshot_cma_state(self)
    with numpy.errstate(divide="ignore", over="ignore", invalid="ignore"):
        old_centroid, c_diff, samples = shift_cma_centroid(self, population)
        hsig = update_cma_paths(self, c_diff, c_diff / self.diag_d)
        steps = (samples - old_centroid) / self.sigma
        decay = (1 - hsig) * self.rank_one * self.cm_cum * (2 - self.cm_cum)
        keep = 1 - self.rank_one - self.rank_mu + decay
        self.big_c = (
            keep * self.big_c
            + self.rank_one * self.pc**2
            + self.rank_mu * numpy.dot(self.weights, steps**2)
        )
        self.big_c = numpy.maximum(self.big_c, numpy.finfo(float).tiny)
        adapt_cma_sigma(self)
    check_cma_state(self, snapshot)
    self.diag_d = numpy.sqrt(self.big_c)
    self.cond = float(numpy.max(self.big_c) / numpy.min(self.big_c))
    self.update_count += 1

Strategy(centroid, sigma, **kwargs)

Bases: CmaCore

Standard Covariance Matrix Adaptation evolution strategy.

Hansen's chiN and diag(D) are stored as chi_n and diag_d. chi_n is the expected norm of an N-dimensional standard normal vector. diag_d is the diagonal of D, the square-root eigenvalues of the covariance C.

Parameters:

Name	Type	Description	Default
centroid	Iterable[float]	Starting point of the search distribution.	required
sigma	float	Initial standard deviation of the distribution.	required
**kwargs	Any	Optional strategy parameters. See the table below.	{}

.. dropdown:: Table of Kwargs :margin: 0 5 0 0

• offsprings - (int)
  • The number of children to produce at each generation.
  • Default: int(4 + 3 * log(len(centroid)))
• survivors - (int)
  • The number of children to keep as parents for the next generation. Must be between 1 and offsprings.
  • Default: int(offsprings / 2)
• weights - (str)
  • Recombination weights. One of superlinear, linear, or equal.
  • Default: 'superlinear'
• cm_init - (numpy.ndarray)
  • The initial covariance matrix of the distribution.
  • Default: numpy.identity(len(centroid))
• cm_cum - (float)
  • Cumulation constant of the covariance matrix.
  • Default: 4 / (len(centroid) + 4)
• ss_cum - (float)
  • Cumulation constant of the step-size.
  • Default: (mueff + 2) / (len(centroid) + mueff + 3)
• ss_dmp - (float)
  • Damping of the step-size.
  • Default: 1 + 2 * max(0, sqrt((mueff - 1) / (len(centroid) + 1)) - 1) + ss_cum
• rank_one - (float)
  • Learning rate for rank-one update.
  • Default: 2 / ((len(centroid) + 1.3) ** 2 + mueff)
• rank_mu - (float)
  • Learning rate for rank-mu update.
  • Default: 2 * (mueff - 2 + 1 / mueff) / ((len(centroid) + 2) ** 2 + mueff)
• low, up - (float or sequence)
  • Optional box bounds on generated individuals.
• bound_mode - (str)
  • clip (default) or resample. Both are constraint-handling approximations. clip evaluates the clipped point, but the update learns from the unclipped draw and clips the new centroid into the box.
• resample_limit - (int)
  • Failed redraws before clipping one sample. Default: 100

See the class docstring.

Source code in deap_er/private/strategies/cma_standard.py

def __init__(self, centroid: Iterable[float], sigma: float, **kwargs: Any) -> None:
    """See the class docstring."""
    init_cma_state(self, centroid, sigma)
    self.compute_params(**kwargs)

compute_params(**kwargs)

Recompute strategy parameters from kwargs.

Called from the constructor. Call again if offsprings changes during evolution. Hyperparameters given to earlier calls stay pinned unless kwargs overrides them; the rest are re-derived from the new offsprings. A pinned survivors larger than the new offsprings is capped.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Optional strategy parameters. See the class docstring.	{}

Raises:

Type	Description
ValueError	If survivors is not in [1, offsprings].
RuntimeError	If weights is not superlinear, linear, or equal.
FloatingPointError	If cm_init has no finite eigendecomposition.

Source code in deap_er/private/strategies/cma_standard.py

def compute_params(self, **kwargs: Any) -> None:
    """Recompute strategy parameters from ``kwargs``.

    Called from the constructor. Call again if ``offsprings``
    changes during evolution. Hyperparameters given to earlier
    calls stay pinned unless ``kwargs`` overrides them; the rest
    are re-derived from the new ``offsprings``. A pinned
    ``survivors`` larger than the new ``offsprings`` is capped.

    Args:
        **kwargs: Optional strategy parameters. See the class
            docstring.

    Raises:
        ValueError: If ``survivors`` is not in ``[1, offsprings]``.
        RuntimeError: If ``weights`` is not ``superlinear``,
            ``linear``, or ``equal``.
        FloatingPointError: If ``cm_init`` has no finite
            eigendecomposition.
    """
    params, pinned = merge_hyperparams(self, kwargs, CMA_HYPERPARAMS)
    apply_cma_hyperparams(self, params, cap_survivors="survivors" not in kwargs)
    self.hyperparams = pinned
    if not hasattr(self, "big_c") or "cm_init" in kwargs:
        self.big_c = kwargs.get("cm_init", numpy.identity(self.dim))
        self._decompose()
    update_bound_attrs(self, kwargs)

reset_state(centroid, sigma, **kwargs)

Reset mutable CMA state for a restart.

The centroid, step size, paths, and covariance are reset. Hyperparameters and bounds from earlier calls are kept; see compute_params.

Source code in deap_er/private/strategies/cma_standard.py

def reset_state(
    self,
    centroid: Iterable[float],
    sigma: float,
    **kwargs: Any,
) -> None:
    """Reset mutable CMA state for a restart.

    The centroid, step size, paths, and covariance are reset.
    Hyperparameters and bounds from earlier calls are kept; see
    ``compute_params``.
    """
    reset_cma_state(self, centroid, sigma)
    self.compute_params(cm_init=numpy.identity(self.dim), **kwargs)

generate(ind_init)

Sample offsprings individuals from the current distribution.

Parameters:

Name	Type	Description	Default
ind_init	Callable[..., Individual]	Callable that turns a sampled vector into an individual.	required

Returns:

Type	Description
list[Individual]	Newly sampled individuals.

Source code in deap_er/private/strategies/cma_standard.py

def generate(self, ind_init: Callable[..., Individual]) -> list[Individual]:
    """Sample ``offsprings`` individuals from the current distribution.

    Args:
        ind_init: Callable that turns a sampled vector into an
            individual.

    Returns:
        Newly sampled individuals.
    """
    return generate_cma_offspring(self, self.big_bd, ind_init)

update(population)

Update centroid, step-size, and covariance from population.

Individuals are ranked by fitness. The best survivors members drive the update.

Parameters:

Name	Type	Description	Default
population	list[Individual]	Evaluated individuals from generate.	required

Raises:

Type	Description
FloatingPointError	If the update gives a non-finite state, which happens once sigma collapses to zero, or if the new covariance cannot be decomposed. The strategy is left as it was before the call.

Source code in deap_er/private/strategies/cma_standard.py

def update(self, population: list[Individual]) -> None:
    """Update centroid, step-size, and covariance from ``population``.

    Individuals are ranked by fitness. The best ``survivors``
    members drive the update.

    Args:
        population: Evaluated individuals from ``generate``.

    Raises:
        FloatingPointError: If the update gives a non-finite state,
            which happens once ``sigma`` collapses to zero, or if
            the new covariance cannot be decomposed. The strategy
            is left as it was before the call.
    """
    snapshot = snapshot_cma_state(self)
    with numpy.errstate(divide="ignore", over="ignore", invalid="ignore"):
        old_centroid, c_diff, samples = shift_cma_centroid(self, population)
        y_mean = numpy.dot(self.big_b, (1.0 / self.diag_d) * numpy.dot(self.big_b.T, c_diff))
        hsig = update_cma_paths(self, c_diff, y_mean)
        steps = (samples - old_centroid) / self.sigma
        temp_0 = (1 - hsig) * self.rank_one * self.cm_cum * (2 - self.cm_cum)
        temp_1 = 1 - self.rank_one - self.rank_mu + temp_0
        temp_2 = numpy.outer(self.pc, self.pc)
        temp_3 = numpy.dot((self.weights * steps.T), steps)
        self.big_c = temp_1 * self.big_c + self.rank_one * temp_2 + self.rank_mu * temp_3
        adapt_cma_sigma(self)
    check_cma_state(self, snapshot)
    try:
        self._decompose()
    except FloatingPointError:
        restore_cma_state(self, snapshot)
        raise
    self.update_count += 1

RestartStrategy(strategy, *, mode='bipop', budget, target_f=None, sigma_large=2.0, lambda_factor=2.0, max_large_restarts=9, max_restarts=None, stagnation_window=20, tol_fun=1e-12, condition_limit=100000000000000.0, restart_centroid='random', stagnation_key=None)

Wrap a standard, separable, (1+λ), or MO CMA strategy with IPOP or BIPOP restarts.

See constructor keyword arguments for configuration. target_f is expressed in raw objective space for single-objective runs. The first run uses sigma_large as its initial step size. A run whose step size collapses (the inner update raises FloatingPointError) ends and restarts.

stagnation_key must return a higher-is-better scalar. It is required for multi-objective fitness because there is no default scalarization.

See the class docstring.

Source code in deap_er/private/strategies/restart.py

def __init__(
    self,
    strategy: StrategyLike,
    *,
    mode: Literal["ipop", "bipop"] = "bipop",
    budget: int,
    target_f: float | None = None,
    sigma_large: float = 2.0,
    lambda_factor: float = 2.0,
    max_large_restarts: int = 9,
    max_restarts: int | None = None,
    stagnation_window: int = 20,
    tol_fun: float = 1e-12,
    condition_limit: float = 1e14,
    restart_centroid: str | Callable[[int], numpy.ndarray] = "random",
    stagnation_key: Callable[[Individual], float] | None = None,
) -> None:
    """See the class docstring."""
    self.strategy = strategy
    self.budget = budget
    self.target_f = target_f
    self.max_restarts = max_restarts
    self.restart_centroid = restart_centroid
    self.stagnation_key = stagnation_key

    self.dim = strategy_dim(strategy)
    lambda_default = int(getattr(strategy, "lamb", default_lambda(self.dim)))
    self._schedule = RestartSchedule(
        mode, lambda_default, lambda_factor, max_large_restarts, sigma_large
    )
    self._mu_default = int(getattr(strategy, "mu", 1))
    self._restart_count = 0
    self._run_evals = 0
    self._small_run_cap: int | None = None
    self._evals_used = 0
    self._done = False
    self._ind_init: Callable[..., Individual] | None = None
    self._initial_center = strategy_center(strategy)
    self._best: Individual | None = None
    self._best_w: float = float("-inf")
    self._fitness_weights: tuple[float, ...] | None = None
    self._tracker = RunTracker(
        self.dim,
        lambda_default,
        sigma_large,
        stagnation_window=stagnation_window,
        tol_fun=tol_fun,
        condition_limit=condition_limit,
    )
    set_strategy_sigma(self.strategy, sigma_large)
    self._begin_run(lambda_default, sigma_large)

mode property

Restart scheme, ipop or bipop.

sigma_large property

First-run and large-regime step size.

lambda_factor property

Population growth factor per large restart.

max_large_restarts property

Cap on the large-restart exponent.

evals_used property

Total function evaluations consumed so far.

restart_count property

Number of restarts completed.

regime property

Active restart regime, or None before the first restart.

best_fitness property

Best raw objective seen across all runs for single-objective runs.

Independent of stagnation_key, which only drives stagnation.

remaining_budget()

Function evaluations left before the hard budget is reached.

Source code in deap_er/private/strategies/restart.py

def remaining_budget(self) -> int:
    """Function evaluations left before the hard budget is reached."""
    return max(0, self.budget - self._evals_used)

generate(ind_init)

Sample offspring from the inner strategy within the eval budget.

Source code in deap_er/private/strategies/restart.py

def generate(self, ind_init: Callable[..., Individual]) -> list[Individual]:
    """Sample offspring from the inner strategy within the eval budget."""
    self._ind_init = ind_init
    remaining = self.remaining_budget()
    if remaining <= 0:
        return []
    requested = self.strategy.lamb
    batch = min(requested, remaining)
    if batch != requested:
        resize_offsprings(self.strategy, batch)
    return self.strategy.generate(ind_init)

update(population)

Update the inner strategy and check per-run termination.

Source code in deap_er/private/strategies/restart.py

def update(self, population: list[Individual]) -> None:
    """Update the inner strategy and check per-run termination."""
    if not population:
        self._done = True
        return
    if self._fitness_weights is None:
        self._fitness_weights = population[0].fitness.weights
    require_stagnation_key(self._fitness_weights, self.stagnation_key)
    self._tracker.terminate |= update_or_flag_collapse(self.strategy, population)
    self._run_evals += len(population)
    self._evals_used += len(population)
    if len(self._fitness_weights) == 1:
        self._best_w = max(self._best_w, *(float(ind.fitness.wvalues[0]) for ind in population))
    for ind in population:
        if self._best is None or scalar_fitness(ind, self.stagnation_key) > scalar_fitness(
            self._best, self.stagnation_key
        ):
            self._best = ind
    cond, sigma, largest = strategy_diagnostics(self.strategy)
    self._tracker.observe(
        population, self.stagnation_key, condition=cond, sigma=sigma, largest_eig=largest
    )
    if target_met(self.target_f, self._fitness_weights, self._best_w):
        self._done = True
    if self._small_run_cap is not None and self._run_evals >= self._small_run_cap:
        self._tracker.terminate = True
    if self._evals_used >= self.budget:
        self._done = True

should_restart()

Return whether the current run ended and a restart is due.

Source code in deap_er/private/strategies/restart.py

def should_restart(self) -> bool:
    """Return whether the current run ended and a restart is due."""
    if self._done:
        return False
    if not self._tracker.terminate:
        return False
    if self._evals_used >= self.budget:
        return False
    if self.max_restarts is not None and self._restart_count >= self.max_restarts:
        self._done = True
        return False
    return True

restart()

Finish the current run and launch the next restart.

Source code in deap_er/private/strategies/restart.py

def restart(self) -> None:
    """Finish the current run and launch the next restart."""
    self._schedule.account_run(self._run_evals)
    self._restart_count += 1
    lamb, sigma, self._small_run_cap = self._schedule.next_run(
        self._restart_count, self._evals_used, self.budget
    )
    self._apply_restart(lamb, sigma)
    self._run_evals = 0
    self._begin_run(lamb, sigma)

is_done()

Return whether the global budget or target has been reached.

Source code in deap_er/private/strategies/restart.py

def is_done(self) -> bool:
    """Return whether the global budget or target has been reached."""
    return self._done or self._evals_used >= self.budget