Skip to content

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 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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.

 {}
Source code in deap_er/private/strategies/cma_multi_objective.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
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.
    """
    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.

Source code in deap_er/private/strategies/cma_multi_objective.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def reset_state(
    self,
    parents: list[Individual],
    sigma: float,
    **kwargs: Any,
) -> None:
    """Reset mutable CMA state for a restart."""
    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 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
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 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
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 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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.

 {}
Source code in deap_er/private/strategies/cma_one_plus_lambda.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
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.
    """
    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.

Source code in deap_er/private/strategies/cma_one_plus_lambda.py 
132
133
134
135
136
137
138
139
140
def reset_state(self, parent: Individual, sigma: float, **kwargs: Any) -> None:
    """Reset mutable CMA state for a restart."""
    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 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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 hasattr(self.parent, "fitness"):
        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, or box-bound kwargs are invalid.

See the class docstring.

Source code in deap_er/private/strategies/cma_separable.py 
65
66
67
68
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.

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 missing, the wrong shape, or not strictly positive.
Source code in deap_er/private/strategies/cma_separable.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def compute_params(self, **kwargs: Any) -> None:
    """Recompute λ, rates, and the diagonal ``cm_init`` vector.

    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 missing, the wrong shape, or
            not strictly positive.
    """
    apply_cma_hyperparams(self, kwargs, rank_scale=(self.dim + 2.0) / 3.0)
    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.

Source code in deap_er/private/strategies/cma_separable.py 
89
90
91
92
93
94
95
96
97
def reset_state(
    self,
    centroid: Iterable[float],
    sigma: float,
    **kwargs: Any,
) -> None:
    """Reset mutable CMA state for a restart."""
    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 
 99
100
101
102
103
104
105
106
107
108
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
Source code in deap_er/private/strategies/cma_separable.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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``.
    """
    old_centroid, c_diff = shift_cma_centroid(self, population)
    hsig = update_cma_paths(self, c_diff, c_diff / self.diag_d)
    ar_tmp = numpy.asarray(population[0 : self.mu]) - old_centroid
    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, ar_tmp**2) / self.sigma**2
    )
    self.big_c = numpy.maximum(self.big_c, numpy.finfo(float).tiny)
    adapt_cma_sigma(self)
    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.
    • 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; 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_standard.py 
94
95
96
97
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.

Parameters:

Name Type Description Default
**kwargs Any

Optional strategy parameters. See the class docstring.

 {}

Raises:

Type Description
RuntimeError

If weights is not superlinear, linear, or equal.
Source code in deap_er/private/strategies/cma_standard.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
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.

    Raises:
        RuntimeError: If ``weights`` is not ``superlinear``,
            ``linear``, or ``equal``.
    """
    apply_cma_hyperparams(self, kwargs)
    if not hasattr(self, "big_c") or "cm_init" in kwargs:
        self.big_c = kwargs.get("cm_init", numpy.identity(self.dim))
        self.diag_d, self.big_b = numpy.linalg.eigh(self.big_c)
        indx = numpy.argsort(self.diag_d)
        self.cond = self.diag_d[indx[-1]] / self.diag_d[indx[0]]
        self.diag_d = self.diag_d[indx] ** 0.5
        self.big_b = self.big_b[:, indx]
        self.big_bd = self.big_b * self.diag_d
    update_bound_attrs(self, kwargs)

reset_state(centroid, sigma, **kwargs)

Reset mutable CMA state for a restart.

Source code in deap_er/private/strategies/cma_standard.py 
124
125
126
127
128
129
130
131
132
def reset_state(
    self,
    centroid: Iterable[float],
    sigma: float,
    **kwargs: Any,
) -> None:
    """Reset mutable CMA state for a restart."""
    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 
134
135
136
137
138
139
140
141
142
143
144
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
Source code in deap_er/private/strategies/cma_standard.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
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``.
    """
    old_centroid, c_diff = 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)
    ar_tmp = population[0 : self.mu] - old_centroid
    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 * ar_tmp.T), ar_tmp)
    self.big_c = (
        temp_1 * self.big_c + self.rank_one * temp_2 + self.rank_mu * temp_3 / self.sigma**2
    )
    adapt_cma_sigma(self)
    self.diag_d, self.big_b = numpy.linalg.eigh(self.big_c)
    indx = numpy.argsort(self.diag_d)

    self.cond = self.diag_d[indx[-1]] / self.diag_d[indx[0]]

    self.diag_d = self.diag_d[indx] ** 0.5
    self.big_b = self.big_b[:, indx]
    self.big_bd = self.big_b * self.diag_d

    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.

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 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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.mode = mode
    self.budget = budget
    self.target_f = target_f
    self.sigma_large = sigma_large
    self.lambda_factor = lambda_factor
    self.max_large_restarts = max_large_restarts
    self.max_restarts = max_restarts
    self.restart_centroid = restart_centroid
    self.stagnation_key = stagnation_key

    self.dim = strategy_dim(strategy)
    self._lambda_default = int(getattr(strategy, "lamb", default_lambda(self.dim)))
    self._lambda_large = self._lambda_default
    self._irestart_large = 0
    self._budget_large = 0
    self._budget_small = 0
    self._run_count = 0
    self._restart_count = 0
    self._regime: Literal["large", "small"] | None = None
    self._run_evals = 0
    self._last_large_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._fitness_weights: tuple[float, ...] | None = None
    self._tracker = RunTracker(
        self.dim,
        self._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(self._lambda_default, sigma_large)

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.

remaining_budget()

Function evaluations left before the hard budget is reached.

Source code in deap_er/private/strategies/restart.py 
144
145
146
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 
148
149
150
151
152
153
154
155
156
157
158
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 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
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.strategy.update(population)
    self._run_evals += len(population)
    self._evals_used += len(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._tracker.best_ever):
        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 
191
192
193
194
195
196
197
198
199
200
201
202
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 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def restart(self) -> None:
    """Finish the current run and launch the next restart."""
    self._account_run_budget()
    self._run_count += 1
    self._restart_count += 1
    if self.mode == "ipop":
        lamb, sigma, self._irestart_large = next_ipop_params(
            self._lambda_default,
            self.lambda_factor,
            self._irestart_large,
            self.max_large_restarts,
            self.sigma_large,
        )
        self._regime = "large"
    else:
        lamb, sigma, self._regime, self._irestart_large, self._lambda_large = next_bipop_params(
            lambda_default=self._lambda_default,
            lambda_factor=self.lambda_factor,
            lambda_large=self._lambda_large,
            irestart_large=self._irestart_large,
            max_large_restarts=self.max_large_restarts,
            sigma_large=self.sigma_large,
            restart_count=self._restart_count,
            evals_used=self._evals_used,
            budget=self.budget,
            budget_large=self._budget_large,
            budget_small=self._budget_small,
        )
    self._small_run_cap = (
        max(1, self._last_large_run_evals // 2) if self._regime == "small" else None
    )
    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 
239
240
241
def is_done(self) -> bool:
    """Return whether the global budget or target has been reached."""
    return self._done or self._evals_used >= self.budget