Skip to content

Algorithms

deap_er.algorithms

PolicyActionGuard(max_promotes_per_gen=1, max_tune_gen=5, min_exam_size=1, promote_cooldown=0, n_evals=None, nevals_used=0, generation=0, promotes_this_gen=0, last_promote_gen=None) dataclass

Hard caps for :func:~deap_er.algorithms.apply_policy_action.

Tracks per-generation promote counts, promote cooldown, inner tune generation limits, minimum exam size, and an optional evaluation budget. Call :meth:begin_generation at the start of each outer generation so promote limits reset.

Attributes:

Name Type Description
max_promotes_per_gen int

Maximum promote_subtree calls per generation. 0 blocks every promote for that generation.
max_tune_gen int

Maximum inner n_gen accepted by tune_ephemerals.
min_exam_size int

Minimum catalog size required for next_lexicase_cases.
promote_cooldown int

Generations that must pass after a promote before another promote is allowed.
n_evals int | None

Optional evaluation budget. When set, actions whose estimated cost would exceed the remaining budget are rejected.
nevals_used int

Evaluations already charged to this guard.
generation int

Current outer generation index.
promotes_this_gen int

Promotes applied in the current generation.
last_promote_gen int | None

Generation index of the last promote, or None when no promote has run yet.

__post_init__()

Reject invalid cap configuration at construction.

Source code in deap_er/private/operators/policy_action_guard.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def __post_init__(self) -> None:
    """Reject invalid cap configuration at construction."""
    if self.max_promotes_per_gen < 0:
        raise ValueError("max_promotes_per_gen must be at least 0")
    if self.max_tune_gen < 1:
        raise ValueError("max_tune_gen must be at least 1")
    if self.min_exam_size < 1:
        raise ValueError("min_exam_size must be at least 1")
    if self.promote_cooldown < 0:
        raise ValueError("promote_cooldown must be at least 0")
    if self.n_evals is not None and self.n_evals < 0:
        raise ValueError("n_evals must be at least 0")
    if self.nevals_used < 0:
        raise ValueError("nevals_used must be at least 0")

begin_generation(generation=None)

Reset per-generation counters and optionally bump generation.

Parameters:

Name Type Description Default
generation int | None

When given, replaces :attr:generation.

 None
Source code in deap_er/private/operators/policy_action_guard.py 
83
84
85
86
87
88
89
90
91
def begin_generation(self, generation: int | None = None) -> None:
    """Reset per-generation counters and optionally bump ``generation``.

    Args:
        generation: When given, replaces :attr:`generation`.
    """
    if generation is not None:
        self.generation = generation
    self.promotes_this_gen = 0

allows(action, /, **kwargs)

Return whether action may run under the current caps.

Parameters:

Name Type Description Default
action str

Policy action token.

 required
**kwargs Any

Arguments that would be forwarded to :func:~deap_er.algorithms.apply_policy_action.

 {}

Returns:

Type Description
bool

False when a cap would be violated; otherwise True.
Source code in deap_er/private/operators/policy_action_guard.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def allows(self, action: str, /, **kwargs: Any) -> bool:
    """Return whether ``action`` may run under the current caps.

    Args:
        action: Policy action token.
        **kwargs: Arguments that would be forwarded to
            :func:`~deap_er.algorithms.apply_policy_action`.

    Returns:
        ``False`` when a cap would be violated; otherwise ``True``.
    """
    if action == "promote_subtree" and not self._promote_allowed():
        return False
    if action == "tune_ephemerals":
        n_gen = int(kwargs.get("n_gen", 5))
        if n_gen > self.max_tune_gen:
            return False
    if action == "next_lexicase_cases" and not _exam_size_allowed(
        self.min_exam_size,
        kwargs,
    ):
        return False
    if self.n_evals is not None:
        cost = estimate_policy_action_evals(action, **kwargs)
        if self.nevals_used + cost > self.n_evals:
            return False
    return True

note_applied(action, /, *, evals=0)

Record a successfully applied action.

Parameters:

Name Type Description Default
action str

Policy action token that ran.

 required
evals int

Evaluations to charge against :attr:n_evals.

 0
Source code in deap_er/private/operators/policy_action_guard.py 
121
122
123
124
125
126
127
128
129
130
131
def note_applied(self, action: str, /, *, evals: int = 0) -> None:
    """Record a successfully applied action.

    Args:
        action: Policy action token that ran.
        evals: Evaluations to charge against :attr:`n_evals`.
    """
    self.nevals_used += evals
    if action == "promote_subtree":
        self.promotes_this_gen += 1
        self.last_promote_gen = self.generation

PolicyActionResult(applied, rejected, value=None) dataclass

Outcome of :func:apply_policy_action.

Attributes:

Name Type Description
applied bool

True when an underlying callable ran.
rejected bool

True when the token is unknown, required kwargs were missing, or a guard cap rejected the action.
value Any

Return value from the underlying callable when applied is True; otherwise None.

ea_generate_update(toolbox, generations, hof=None, stats=None, verbose=False, logger=None, log_time=False, fronts=None)

Evolve a strategy that generates and updates a population.

Requires generate, update, and evaluate on toolbox. An empty generate batch stops the loop and returns the last evaluated population. update is not called with [].

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with the generate, update, and evaluate operators.

 required
generations int

Number of generations to run.

 required
hof EvoRecords | None

Optional HallOfFame or ParetoFront to update.

 None
stats EvoStats | None

Optional Statistics or MultiStatistics to compile.

 None
verbose bool

If True, print the logbook stream each generation.

 False
logger Logger | None

If given with verbose, the stream is logged.

 None
log_time bool

If True, record per-generation duration.

 False
fronts list[Any] | None

Optional list that receives a ParetoFront snapshot of each generation's population.

 None

Returns:

Type Description
EvoAlgoResult

The final population and the logbook.
Source code in deap_er/private/algorithms/ea_generate_update.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def ea_generate_update(
    toolbox: Toolbox,
    generations: int,
    hof: EvoRecords | None = None,
    stats: EvoStats | None = None,
    verbose: bool = False,
    logger: Logger | None = None,
    log_time: bool = False,
    fronts: list[Any] | None = None,
) -> EvoAlgoResult:
    """Evolve a strategy that generates and updates a population.

    Requires ``generate``, ``update``, and ``evaluate`` on ``toolbox``.
    An empty ``generate`` batch stops the loop and returns the last
    evaluated population. ``update`` is not called with ``[]``.

    Args:
        toolbox: Toolbox with the generate, update, and evaluate operators.
        generations: Number of generations to run.
        hof: Optional HallOfFame or ParetoFront to update.
        stats: Optional Statistics or MultiStatistics to compile.
        verbose: If True, print the logbook stream each generation.
        logger: If given with ``verbose``, the stream is logged.
        log_time: If True, record per-generation ``duration``.
        fronts: Optional list that receives a ParetoFront snapshot
            of each generation's population.

    Returns:
        The final population and the logbook.
    """
    logbook = new_logbook(stats, log_time=log_time)

    population: list[Individual] = []
    for gen in range(1, generations + 1):
        t0 = time.perf_counter()
        next_population = toolbox.generate()
        if not next_population:
            break
        population = next_population
        nevals = evaluate_invalid(toolbox, population)

        toolbox.update(population)
        duration = time.perf_counter() - t0 if log_time else None

        record_generation(
            logbook,
            gen,
            nevals,
            population=population,
            offspring=population,
            hof=hof,
            stats=stats,
            verbose=verbose,
            logger=logger,
            duration=duration,
            fronts=fronts,
        )

    return population, logbook

ea_generate_update_restarts(toolbox, restart_strategy, hof=None, stats=None, verbose=False, logger=None, log_time=False, log_restarts=True, fronts=None)

Evolve with IPOP or BIPOP CMA restarts until the budget is spent.

Requires generate, update, and evaluate on toolbox. The toolbox operators should be bound to restart_strategy.generate and restart_strategy.update. An empty generate batch stops the loop and returns the last evaluated population.

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with generate, update, and evaluate operators.

 required
restart_strategy RestartStrategy

Restart wrapper that tracks stagnation and relaunches the inner CMA strategy.

 required
hof EvoRecords | None

Optional HallOfFame or ParetoFront to update.

 None
stats EvoStats | None

Optional Statistics or MultiStatistics to compile.

 None
verbose bool

If True, print the logbook stream each generation.

 False
logger Logger | None

If given with verbose, the stream is logged.

 None
log_time bool

If True, record per-generation duration.

 False
log_restarts bool

If True, log restart, regime, lambda, and evals columns.

 True
fronts list[Any] | None

Optional list that receives a ParetoFront snapshot of each generation's population.

 None

Returns:

Type Description
EvoAlgoResult

The final population and the logbook.
Source code in deap_er/private/algorithms/ea_generate_update_restarts.py 
 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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def ea_generate_update_restarts(
    toolbox: Toolbox,
    restart_strategy: RestartStrategy,
    hof: EvoRecords | None = None,
    stats: EvoStats | None = None,
    verbose: bool = False,
    logger: Logger | None = None,
    log_time: bool = False,
    log_restarts: bool = True,
    fronts: list[Any] | None = None,
) -> EvoAlgoResult:
    """Evolve with IPOP or BIPOP CMA restarts until the budget is spent.

    Requires ``generate``, ``update``, and ``evaluate`` on ``toolbox``.
    The toolbox operators should be bound to ``restart_strategy.generate``
    and ``restart_strategy.update``. An empty ``generate`` batch stops
    the loop and returns the last evaluated population.

    Args:
        toolbox: Toolbox with generate, update, and evaluate operators.
        restart_strategy: Restart wrapper that tracks stagnation and
            relaunches the inner CMA strategy.
        hof: Optional HallOfFame or ParetoFront to update.
        stats: Optional Statistics or MultiStatistics to compile.
        verbose: If True, print the logbook stream each generation.
        logger: If given with ``verbose``, the stream is logged.
        log_time: If True, record per-generation ``duration``.
        log_restarts: If True, log ``restart``, ``regime``, ``lambda``,
            and ``evals`` columns.
        fronts: Optional list that receives a ParetoFront snapshot
            of each generation's population.

    Returns:
        The final population and the logbook.
    """
    logbook = new_logbook(stats, log_time=log_time)
    if log_restarts:
        logbook.header.extend(["restart", "regime", "lambda", "evals"])

    population: list[Individual] = []
    gen = 0
    while not restart_strategy.is_done():
        t0 = time.perf_counter()
        next_population = toolbox.generate()
        if not next_population:
            break
        population = next_population
        nevals = evaluate_invalid(toolbox, population)
        restart_strategy.update(population)
        gen += 1
        duration = time.perf_counter() - t0 if log_time else None
        _record_restart_generation(
            logbook,
            gen,
            population,
            nevals=nevals,
            hof=hof,
            stats=stats,
            duration=duration,
            fronts=fronts,
            extra=_restart_log_extra(restart_strategy, log_restarts),
        )
        _log_verbose(logbook, verbose, logger)
        if restart_strategy.is_done():
            break
        if restart_strategy.should_restart():
            restart_strategy.restart()

    return population, logbook

ea_map_elites(toolbox, archive, descriptor_fn, initial, generations, batch_size, cx_prob, mut_prob, stats=None, verbose=False, logger=None, log_time=False, n_evals=None)

Run MAP-Elites with var_or variation on archive elites.

Generation zero evaluates initial and seeds the archive. Later generations sample parents from archive, vary them with var_or, evaluate the offspring, and try to improve cells. When n_evals is set, the generation that meets or exceeds that count is the last one recorded. Generations remain the default stop.

Requires clone, mate, mutate, and evaluate on toolbox. archive stores single-objective fitness only. Accepts :class:~deap_er.records.GridArchive, :class:~deap_er.records.CvtArchive, or :class:~deap_er.records.UnstructuredArchive.

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with the evolution operators.

 required
archive MapElitesArchive

MAP-Elites archive updated in place.

 required
descriptor_fn Callable[[Individual], Sequence[float]]

Maps an evaluated individual to a behavior descriptor.

 required
initial list[Individual]

Individuals evaluated and archived before generation one.

 required
generations int

Number of variation generations after the initial seeding generation.

 required
batch_size int

Offspring produced each variation generation. When cx_prob is positive, the parent pool is at least two individuals so crossover can run.

 required
cx_prob float

Probability of crossover in var_or.

 required
mut_prob float

Probability of mutation in var_or.

 required
stats EvoStats | None

Optional Statistics or MultiStatistics compiled from the offspring each generation. An empty seed or offspring list skips that compile so reducers such as max do not run on no data. Archive metrics still record.

 None
verbose bool

If True, print the logbook stream each generation.

 False
logger Logger | None

If given with verbose, the stream is logged.

 None
log_time bool

If True, record per-generation duration.

 False
n_evals int | None

Optional evaluation budget. The generation that meets or exceeds this count is finished, then the loop stops. None keeps the generation limit only. Counts fitness assignments through evaluate_invalid, including EvalCache hits.

 None

Returns:

Type Description
tuple[MapElitesArchive, Logbook]

The archive and the logbook.

Raises:

Type Description
ValueError

If a variation generation runs while the archive and initial are both empty, or if n_evals is negative.
Source code in deap_er/private/algorithms/ea_map_elites.py 
 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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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
177
178
179
180
181
182
183
184
def ea_map_elites(
    toolbox: Toolbox,
    archive: MapElitesArchive,
    descriptor_fn: Callable[[Individual], Sequence[float]],
    initial: list[Individual],
    generations: int,
    batch_size: int,
    cx_prob: float,
    mut_prob: float,
    stats: EvoStats | None = None,
    verbose: bool = False,
    logger: Logger | None = None,
    log_time: bool = False,
    n_evals: int | None = None,
) -> tuple[MapElitesArchive, Logbook]:
    """Run MAP-Elites with ``var_or`` variation on archive elites.

    Generation zero evaluates ``initial`` and seeds the archive. Later
    generations sample parents from ``archive``, vary them with
    ``var_or``, evaluate the offspring, and try to improve cells.
    When ``n_evals`` is set, the generation that meets or exceeds that
    count is the last one recorded. Generations remain the default stop.

    Requires ``clone``, ``mate``, ``mutate``, and ``evaluate`` on
    ``toolbox``. ``archive`` stores single-objective fitness only.
    Accepts :class:`~deap_er.records.GridArchive`,
    :class:`~deap_er.records.CvtArchive`, or
    :class:`~deap_er.records.UnstructuredArchive`.

    Args:
        toolbox: Toolbox with the evolution operators.
        archive: MAP-Elites archive updated in place.
        descriptor_fn: Maps an evaluated individual to a behavior
            descriptor.
        initial: Individuals evaluated and archived before generation
            one.
        generations: Number of variation generations after the initial
            seeding generation.
        batch_size: Offspring produced each variation generation. When
            ``cx_prob`` is positive, the parent pool is at least two
            individuals so crossover can run.
        cx_prob: Probability of crossover in ``var_or``.
        mut_prob: Probability of mutation in ``var_or``.
        stats: Optional Statistics or MultiStatistics compiled from the
            offspring each generation. An empty seed or offspring list
            skips that compile so reducers such as ``max`` do not run
            on no data. Archive metrics still record.
        verbose: If True, print the logbook stream each generation.
        logger: If given with ``verbose``, the stream is logged.
        log_time: If True, record per-generation ``duration``.
        n_evals: Optional evaluation budget. The generation that
            meets or exceeds this count is finished, then the loop
            stops. ``None`` keeps the generation limit only. Counts
            fitness assignments through ``evaluate_invalid``,
            including ``EvalCache`` hits.

    Returns:
        The archive and the logbook.

    Raises:
        ValueError: If a variation generation runs while the archive and
            ``initial`` are both empty, or if ``n_evals`` is negative.
    """
    check_n_evals(n_evals)
    logbook = new_logbook(stats, log_time=log_time)
    logbook.header = (
        ["gen", "nevals", "coverage", "num_elites", "qd_score"]
        + (["duration"] if log_time else [])
        + (stats.fields if stats else [])
    )

    t0 = time.perf_counter()
    nevals, used = consume_evals(toolbox, initial, n_evals, 0)
    for individual in initial:
        archive.add(individual, descriptor_fn(individual))
    duration = time.perf_counter() - t0 if log_time else None
    _record_map_elites_generation(
        logbook,
        0,
        nevals,
        archive,
        initial,
        stats,
        verbose,
        logger,
        duration,
    )
    if budget_spent(n_evals, used):
        return archive, logbook

    for gen in range(1, generations + 1):
        t0 = time.perf_counter()
        parents = _parent_pool(archive, initial, batch_size, cx_prob)
        offspring = var_or(toolbox, parents, batch_size, cx_prob, mut_prob)
        nevals, used = consume_evals(toolbox, offspring, n_evals, used)
        for individual in offspring:
            archive.add(individual, descriptor_fn(individual))
        duration = time.perf_counter() - t0 if log_time else None
        _record_map_elites_generation(
            logbook,
            gen,
            nevals,
            archive,
            offspring,
            stats,
            verbose,
            logger,
            duration,
        )
        if budget_spent(n_evals, used):
            break

    return archive, logbook

ea_mu_comma_lambda(toolbox, population, generations, offsprings, survivors, cx_prob, mut_prob, hof=None, stats=None, verbose=False, logger=None, log_time=False, fronts=None, n_evals=None)

Evolve a population with mu-comma-lambda selection.

Requires mate, mutate, select, and evaluate on toolbox. Survivors are selected from the offspring only. When n_evals is set, the generation that meets or exceeds that count is the last one recorded. Generations remain the default stop.

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with the evolution operators.

 required
population list[Individual]

Individuals to evolve. Replaced in place.

 required
generations int

Number of generations to run.

 required
offsprings int

Number of offspring to produce each generation.

 required
survivors int

Number of individuals to keep after selection.

 required
cx_prob float

Probability of mating two individuals.

 required
mut_prob float

Probability of mutating an individual.

 required
hof EvoRecords | None

Optional HallOfFame or ParetoFront to update.

 None
stats EvoStats | None

Optional Statistics or MultiStatistics to compile.

 None
verbose bool

If True, print the logbook stream each generation.

 False
logger Logger | None

If given with verbose, the stream is logged.

 None
log_time bool

If True, record per-generation duration.

 False
fronts list[Any] | None

Optional list that receives a ParetoFront snapshot of each generation's population.

 None
n_evals int | None

Optional evaluation budget. The generation that meets or exceeds this count is finished, then the loop stops. None keeps the generation limit only. Counts fitness assignments through evaluate_invalid, including EvalCache hits.

 None

Returns:

Type Description
EvoAlgoResult

The final population and the logbook.

Raises:

Type Description
ValueError

If survivors is greater than offsprings, or if n_evals is negative.
Source code in deap_er/private/algorithms/ea_mu_comma_lambda.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 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
115
116
117
118
119
120
121
122
123
124
125
126
127
def ea_mu_comma_lambda(
    toolbox: Toolbox,  # NOSONAR python:S107  n_evals matches sibling ea_* drivers
    population: list[Individual],
    generations: int,
    offsprings: int,
    survivors: int,
    cx_prob: float,
    mut_prob: float,
    hof: EvoRecords | None = None,
    stats: EvoStats | None = None,
    verbose: bool = False,
    logger: Logger | None = None,
    log_time: bool = False,
    fronts: list[Any] | None = None,
    n_evals: int | None = None,
) -> EvoAlgoResult:
    """Evolve a population with mu-comma-lambda selection.

    Requires ``mate``, ``mutate``, ``select``, and ``evaluate`` on
    ``toolbox``. Survivors are selected from the offspring only.
    When ``n_evals`` is set, the generation that meets or exceeds
    that count is the last one recorded. Generations remain the
    default stop.

    Args:
        toolbox: Toolbox with the evolution operators.
        population: Individuals to evolve. Replaced in place.
        generations: Number of generations to run.
        offsprings: Number of offspring to produce each generation.
        survivors: Number of individuals to keep after selection.
        cx_prob: Probability of mating two individuals.
        mut_prob: Probability of mutating an individual.
        hof: Optional HallOfFame or ParetoFront to update.
        stats: Optional Statistics or MultiStatistics to compile.
        verbose: If True, print the logbook stream each generation.
        logger: If given with ``verbose``, the stream is logged.
        log_time: If True, record per-generation ``duration``.
        fronts: Optional list that receives a ParetoFront snapshot
            of each generation's population.
        n_evals: Optional evaluation budget. The generation that
            meets or exceeds this count is finished, then the loop
            stops. ``None`` keeps the generation limit only. Counts
            fitness assignments through ``evaluate_invalid``,
            including ``EvalCache`` hits.

    Returns:
        The final population and the logbook.

    Raises:
        ValueError: If ``survivors`` is greater than ``offsprings``,
            or if ``n_evals`` is negative.
    """
    check_n_evals(n_evals)
    if survivors > offsprings:
        raise ValueError(
            "The number of survivors must be less than or equal to the number of offsprings."
        )

    logbook = new_logbook(stats, log_time=log_time)
    t0 = time.perf_counter()
    nevals, used = consume_evals(toolbox, population, n_evals, 0)
    duration = time.perf_counter() - t0 if log_time else None
    record_generation(
        logbook,
        0,
        nevals,
        population=population,
        offspring=population,
        hof=hof,
        stats=stats,
        verbose=verbose,
        logger=logger,
        duration=duration,
        fronts=fronts,
    )
    if budget_spent(n_evals, used):
        return population, logbook

    for gen in range(1, generations + 1):
        t0 = time.perf_counter()
        offspring = var_or(toolbox, population, offsprings, cx_prob, mut_prob)

        nevals, used = consume_evals(toolbox, offspring, n_evals, used)

        population[:] = toolbox.select(offspring, survivors)
        duration = time.perf_counter() - t0 if log_time else None

        record_generation(
            logbook,
            gen,
            nevals,
            population=population,
            offspring=offspring,
            hof=hof,
            stats=stats,
            verbose=verbose,
            logger=logger,
            duration=duration,
            fronts=fronts,
        )
        if budget_spent(n_evals, used):
            break

    return population, logbook

ea_mu_plus_lambda(toolbox, population, generations, offsprings, survivors, cx_prob, mut_prob, hof=None, stats=None, verbose=False, logger=None, log_time=False, fronts=None, n_evals=None)

Evolve a population with mu-plus-lambda selection.

Requires mate, mutate, select, and evaluate on toolbox. Survivors are selected from the union of parents and offspring. When n_evals is set, the generation that meets or exceeds that count is the last one recorded. Generations remain the default stop.

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with the evolution operators.

 required
population list[Individual]

Individuals to evolve. Replaced in place.

 required
generations int

Number of generations to run.

 required
offsprings int

Number of offspring to produce each generation.

 required
survivors int

Number of individuals to keep after selection.

 required
cx_prob float

Probability of mating two individuals.

 required
mut_prob float

Probability of mutating an individual.

 required
hof EvoRecords | None

Optional HallOfFame or ParetoFront to update.

 None
stats EvoStats | None

Optional Statistics or MultiStatistics to compile.

 None
verbose bool

If True, print the logbook stream each generation.

 False
logger Logger | None

If given with verbose, the stream is logged.

 None
log_time bool

If True, record per-generation duration.

 False
fronts list[Any] | None

Optional list that receives a ParetoFront snapshot of each generation's population.

 None
n_evals int | None

Optional evaluation budget. The generation that meets or exceeds this count is finished, then the loop stops. None keeps the generation limit only. Counts fitness assignments through evaluate_invalid, including EvalCache hits.

 None

Returns:

Type Description
EvoAlgoResult

The final population and the logbook.

Raises:

Type Description
ValueError

If n_evals is negative.
Source code in deap_er/private/algorithms/ea_mu_plus_lambda.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 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
115
116
117
118
119
120
121
def ea_mu_plus_lambda(
    toolbox: Toolbox,  # NOSONAR python:S107  n_evals matches sibling ea_* drivers
    population: list[Individual],
    generations: int,
    offsprings: int,
    survivors: int,
    cx_prob: float,
    mut_prob: float,
    hof: EvoRecords | None = None,
    stats: EvoStats | None = None,
    verbose: bool = False,
    logger: Logger | None = None,
    log_time: bool = False,
    fronts: list[Any] | None = None,
    n_evals: int | None = None,
) -> EvoAlgoResult:
    """Evolve a population with mu-plus-lambda selection.

    Requires ``mate``, ``mutate``, ``select``, and ``evaluate`` on
    ``toolbox``. Survivors are selected from the union of parents
    and offspring. When ``n_evals`` is set, the generation that
    meets or exceeds that count is the last one recorded.
    Generations remain the default stop.

    Args:
        toolbox: Toolbox with the evolution operators.
        population: Individuals to evolve. Replaced in place.
        generations: Number of generations to run.
        offsprings: Number of offspring to produce each generation.
        survivors: Number of individuals to keep after selection.
        cx_prob: Probability of mating two individuals.
        mut_prob: Probability of mutating an individual.
        hof: Optional HallOfFame or ParetoFront to update.
        stats: Optional Statistics or MultiStatistics to compile.
        verbose: If True, print the logbook stream each generation.
        logger: If given with ``verbose``, the stream is logged.
        log_time: If True, record per-generation ``duration``.
        fronts: Optional list that receives a ParetoFront snapshot
            of each generation's population.
        n_evals: Optional evaluation budget. The generation that
            meets or exceeds this count is finished, then the loop
            stops. ``None`` keeps the generation limit only. Counts
            fitness assignments through ``evaluate_invalid``,
            including ``EvalCache`` hits.

    Returns:
        The final population and the logbook.

    Raises:
        ValueError: If ``n_evals`` is negative.
    """
    check_n_evals(n_evals)
    logbook = new_logbook(stats, log_time=log_time)
    t0 = time.perf_counter()
    nevals, used = consume_evals(toolbox, population, n_evals, 0)
    duration = time.perf_counter() - t0 if log_time else None
    record_generation(
        logbook,
        0,
        nevals,
        population=population,
        offspring=population,
        hof=hof,
        stats=stats,
        verbose=verbose,
        logger=logger,
        duration=duration,
        fronts=fronts,
    )
    if budget_spent(n_evals, used):
        return population, logbook

    for gen in range(1, generations + 1):
        t0 = time.perf_counter()
        offspring = var_or(toolbox, population, offsprings, cx_prob, mut_prob)

        nevals, used = consume_evals(toolbox, offspring, n_evals, used)

        population[:] = toolbox.select(population + offspring, survivors)
        duration = time.perf_counter() - t0 if log_time else None

        record_generation(
            logbook,
            gen,
            nevals,
            population=population,
            offspring=offspring,
            hof=hof,
            stats=stats,
            verbose=verbose,
            logger=logger,
            duration=duration,
            fronts=fronts,
        )
        if budget_spent(n_evals, used):
            break

    return population, logbook

ea_policy(toolbox, population, decide, generations, cx_prob, mut_prob, *, exams=None, cases=None, n_cases=None, guard=None, elite_count=8, action_kwargs=None, observe_kwargs=None, hof=None, stats=None, verbose=False, logger=None, log_time=False, fronts=None, n_evals=None)

Evolve a population with one policy step each generation.

Same survivor rule as ea_simple: evaluate invalids, then each generation observe → decide → apply_policy_action, select, vary, and evaluate. Fitness stays on the toolbox. When cases or a pool of exams is available, selection is lexicase on that subset; otherwise toolbox.select is used.

Requires mate, mutate, and evaluate (or evaluate_batch) on toolbox. select is required only when no case subset is in play. Policy-action evaluations count toward n_evals and the generation nevals. When a policy step meets or exceeds that budget, the generation is recorded without variation so the population is not replaced with unevaluated offspring.

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with the evolution operators.

 required
population list[Individual]

Individuals to evolve. Replaced in place.

 required
decide Callable[[PolicyObservation], str]

Maps one PolicyObservation to an action token.

 required
generations int

Number of generations to run.

 required
cx_prob float

Probability of mating two individuals.

 required
mut_prob float

Probability of mutating an individual.

 required
exams CaseExamPool | None

Optional exam pool. Train / held-out scores become observations. next_lexicase_cases reads this pool.

 None
cases Sequence[int] | None

Initial lexicase case indices. Defaults to the first train exam when exams is set.

 None
n_cases int | None

Catalog size for CaseExam.as_cases. Defaults to the length of a valid fitness vector.

 None
guard PolicyActionGuard | None

Optional action guard. begin_generation is called at the start of each outer generation.

 None
elite_count int

Elites used to build the observation.

 8
action_kwargs dict[str, Any] | None

Extra kwargs forwarded to apply_policy_action (for example mut_prob or prim_set).

 None
observe_kwargs dict[str, Any] | None

Extra kwargs forwarded to policy_observe (archive, rows_seen, promoted_library_size).

 None
hof EvoRecords | None

Optional HallOfFame or ParetoFront to update.

 None
stats EvoStats | None

Optional Statistics or MultiStatistics to compile.

 None
verbose bool

If True, print the logbook stream each generation.

 False
logger Logger | None

If given with verbose, the stream is logged.

 None
log_time bool

If True, record per-generation duration.

 False
fronts list[Any] | None

Optional list that receives a ParetoFront snapshot of each generation's population.

 None
n_evals int | None

Optional evaluation budget. Policy-action evaluations count toward the total. When a policy step meets or exceeds this count, that generation is recorded without variation and the loop stops. None keeps the generation limit only.

 None

Returns:

Type Description
EvoAlgoResult

The final population and the logbook.

Raises:

Type Description
ValueError

If n_evals is negative.
Source code in deap_er/private/algorithms/ea_policy.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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
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
208
209
210
211
212
213
def ea_policy(
    toolbox: Toolbox,  # NOSONAR python:S107  n_evals matches sibling ea_* drivers
    population: list[Individual],
    decide: Callable[[PolicyObservation], str],
    generations: int,
    cx_prob: float,
    mut_prob: float,
    *,
    exams: CaseExamPool | None = None,
    cases: Sequence[int] | None = None,
    n_cases: int | None = None,
    guard: PolicyActionGuard | None = None,
    elite_count: int = 8,
    action_kwargs: dict[str, Any] | None = None,
    observe_kwargs: dict[str, Any] | None = None,
    hof: EvoRecords | None = None,
    stats: EvoStats | None = None,
    verbose: bool = False,
    logger: Logger | None = None,
    log_time: bool = False,
    fronts: list[Any] | None = None,
    n_evals: int | None = None,
) -> EvoAlgoResult:
    """Evolve a population with one policy step each generation.

    Same survivor rule as ``ea_simple``: evaluate invalids, then each
    generation observe → decide → ``apply_policy_action``, select,
    vary, and evaluate. Fitness stays on the toolbox. When ``cases``
    or a pool of exams is available, selection is lexicase on that
    subset; otherwise ``toolbox.select`` is used.

    Requires ``mate``, ``mutate``, and ``evaluate`` (or
    ``evaluate_batch``) on ``toolbox``. ``select`` is required only
    when no case subset is in play. Policy-action evaluations count
    toward ``n_evals`` and the generation ``nevals``. When a policy
    step meets or exceeds that budget, the generation is recorded
    without variation so the population is not replaced with
    unevaluated offspring.

    Args:
        toolbox: Toolbox with the evolution operators.
        population: Individuals to evolve. Replaced in place.
        decide: Maps one ``PolicyObservation`` to an action token.
        generations: Number of generations to run.
        cx_prob: Probability of mating two individuals.
        mut_prob: Probability of mutating an individual.
        exams: Optional exam pool. Train / held-out scores become
            observations. ``next_lexicase_cases`` reads this pool.
        cases: Initial lexicase case indices. Defaults to the first
            train exam when ``exams`` is set.
        n_cases: Catalog size for ``CaseExam.as_cases``. Defaults to
            the length of a valid fitness vector.
        guard: Optional action guard. ``begin_generation`` is called
            at the start of each outer generation.
        elite_count: Elites used to build the observation.
        action_kwargs: Extra kwargs forwarded to
            ``apply_policy_action`` (for example ``mut_prob`` or
            ``prim_set``).
        observe_kwargs: Extra kwargs forwarded to ``policy_observe``
            (``archive``, ``rows_seen``, ``promoted_library_size``).
        hof: Optional HallOfFame or ParetoFront to update.
        stats: Optional Statistics or MultiStatistics to compile.
        verbose: If True, print the logbook stream each generation.
        logger: If given with ``verbose``, the stream is logged.
        log_time: If True, record per-generation ``duration``.
        fronts: Optional list that receives a ParetoFront snapshot
            of each generation's population.
        n_evals: Optional evaluation budget. Policy-action
            evaluations count toward the total. When a policy
            step meets or exceeds this count, that generation is
            recorded without variation and the loop stops.
            ``None`` keeps the generation limit only.

    Returns:
        The final population and the logbook.

    Raises:
        ValueError: If ``n_evals`` is negative.
    """
    check_n_evals(n_evals)
    logbook = new_logbook(stats, log_time=log_time, extra_fields=("action",))
    extras = dict(action_kwargs) if action_kwargs else {}
    observe = dict(observe_kwargs) if observe_kwargs else {}
    active_cases = list(cases) if cases is not None else None
    rejected = False
    action = POLICY_ACTION_SKIP_PROMOTE
    train_score = 0.0
    held_out_score: float | None = None

    t0 = time.perf_counter()
    nevals, used = consume_evals(toolbox, population, n_evals, 0)
    _sync_guard_evals(guard, used)
    if active_cases is None:
        active_cases = initial_policy_cases(population, exams, n_cases)
    extra = policy_row_extra(action, exams, train_score, held_out_score)
    _record(
        logbook,
        0,
        nevals,
        population,
        population,
        hof,
        stats,
        verbose,
        logger,
        t0,
        log_time,
        fronts,
        extra,
    )
    if budget_spent(n_evals, used):
        return population, logbook

    for gen in range(1, generations + 1):
        t0 = time.perf_counter()
        if guard is not None:
            guard.begin_generation(gen)
        action, rejected, train_score, held_out_score, applied_cases, action_evals = (
            step_policy_generation(
                decide,
                population,
                used=used,
                rejected=rejected,
                exams=exams,
                guard=guard,
                elite_count=elite_count,
                extras=extras,
                observe=observe,
                toolbox=toolbox,
            )
        )
        used += action_evals
        _sync_guard_evals(guard, used)
        if applied_cases is not None:
            active_cases = applied_cases
        extra = policy_row_extra(action, exams, train_score, held_out_score)
        if budget_spent(n_evals, used):
            _record(
                logbook,
                gen,
                action_evals,
                population,
                population,
                hof,
                stats,
                verbose,
                logger,
                t0,
                log_time,
                fronts,
                extra,
            )
            break
        offspring = select_policy_offspring(toolbox, population, active_cases)
        offspring = var_and(toolbox, offspring, cx_prob, mut_prob)
        nevals, used = consume_evals(toolbox, offspring, n_evals, used)
        _sync_guard_evals(guard, used)
        population[:] = offspring
        _record(
            logbook,
            gen,
            nevals + action_evals,
            population,
            offspring,
            hof,
            stats,
            verbose,
            logger,
            t0,
            log_time,
            fronts,
            extra,
        )
        if budget_spent(n_evals, used):
            break

    return population, logbook

ea_simple(toolbox, population, generations, cx_prob, mut_prob, hof=None, stats=None, verbose=False, logger=None, log_time=False, fronts=None, n_evals=None)

Evolve a population with crossover and mutation on every generation.

Requires mate, mutate, select, and evaluate on toolbox. Survivors are the offspring of the current generation. When n_evals is set, the generation that meets or exceeds that count is the last one recorded. Generations remain the default stop.

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with the evolution operators.

 required
population list[Individual]

Individuals to evolve. Replaced in place.

 required
generations int

Number of generations to run.

 required
cx_prob float

Probability of mating two individuals.

 required
mut_prob float

Probability of mutating an individual.

 required
hof EvoRecords | None

Optional HallOfFame or ParetoFront to update.

 None
stats EvoStats | None

Optional Statistics or MultiStatistics to compile.

 None
verbose bool

If True, print the logbook stream each generation.

 False
logger Logger | None

If given with verbose, the stream is logged.

 None
log_time bool

If True, record per-generation duration.

 False
fronts list[Any] | None

Optional list that receives a ParetoFront snapshot of each generation's population.

 None
n_evals int | None

Optional evaluation budget. The generation that meets or exceeds this count is finished, then the loop stops. None keeps the generation limit only. Counts fitness assignments through evaluate_invalid, including EvalCache hits.

 None

Returns:

Type Description
EvoAlgoResult

The final population and the logbook.

Raises:

Type Description
ValueError

If n_evals is negative.
Source code in deap_er/private/algorithms/ea_simple.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 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
115
116
117
def ea_simple(
    toolbox: Toolbox,
    population: list[Individual],
    generations: int,
    cx_prob: float,
    mut_prob: float,
    hof: EvoRecords | None = None,
    stats: EvoStats | None = None,
    verbose: bool = False,
    logger: Logger | None = None,
    log_time: bool = False,
    fronts: list[Any] | None = None,
    n_evals: int | None = None,
) -> EvoAlgoResult:
    """Evolve a population with crossover and mutation on every generation.

    Requires ``mate``, ``mutate``, ``select``, and ``evaluate`` on
    ``toolbox``. Survivors are the offspring of the current generation.
    When ``n_evals`` is set, the generation that meets or exceeds that
    count is the last one recorded. Generations remain the default stop.

    Args:
        toolbox: Toolbox with the evolution operators.
        population: Individuals to evolve. Replaced in place.
        generations: Number of generations to run.
        cx_prob: Probability of mating two individuals.
        mut_prob: Probability of mutating an individual.
        hof: Optional HallOfFame or ParetoFront to update.
        stats: Optional Statistics or MultiStatistics to compile.
        verbose: If True, print the logbook stream each generation.
        logger: If given with ``verbose``, the stream is logged.
        log_time: If True, record per-generation ``duration``.
        fronts: Optional list that receives a ParetoFront snapshot
            of each generation's population.
        n_evals: Optional evaluation budget. The generation that
            meets or exceeds this count is finished, then the loop
            stops. ``None`` keeps the generation limit only. Counts
            fitness assignments through ``evaluate_invalid``,
            including ``EvalCache`` hits.

    Returns:
        The final population and the logbook.

    Raises:
        ValueError: If ``n_evals`` is negative.
    """
    check_n_evals(n_evals)
    logbook = new_logbook(stats, log_time=log_time)
    t0 = time.perf_counter()
    nevals, used = consume_evals(toolbox, population, n_evals, 0)
    duration = time.perf_counter() - t0 if log_time else None
    record_generation(
        logbook,
        0,
        nevals,
        population=population,
        offspring=population,
        hof=hof,
        stats=stats,
        verbose=verbose,
        logger=logger,
        duration=duration,
        fronts=fronts,
    )
    if budget_spent(n_evals, used):
        return population, logbook

    for gen in range(1, generations + 1):
        t0 = time.perf_counter()
        offspring = toolbox.select(population, len(population))
        offspring = var_and(toolbox, offspring, cx_prob, mut_prob)

        nevals, used = consume_evals(toolbox, offspring, n_evals, used)

        population[:] = offspring
        duration = time.perf_counter() - t0 if log_time else None

        record_generation(
            logbook,
            gen,
            nevals,
            population=population,
            offspring=offspring,
            hof=hof,
            stats=stats,
            verbose=verbose,
            logger=logger,
            duration=duration,
            fronts=fronts,
        )
        if budget_spent(n_evals, used):
            break

    return population, logbook

evaluate_invalid(toolbox, individuals)

Evaluate the individuals whose fitness is invalid.

This is the helper ea_* drivers and apply_policy_action already use. When the toolbox has an evaluate_batch operator, the whole batch of invalid individuals is handed to it in one call and map is not used. Otherwise each individual goes through map and evaluate as usual.

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with the evaluate and map operators.

 required
individuals Sequence[Any]

Individuals to scan for invalid fitness.

 required

Returns:

Type Description
int

The number of individuals that were evaluated.
Source code in deap_er/private/algorithms/loop.py 
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
def evaluate_invalid(toolbox: Toolbox, individuals: Sequence[Any]) -> int:
    """Evaluate the individuals whose fitness is invalid.

    This is the helper ``ea_*`` drivers and ``apply_policy_action``
    already use. When the toolbox has an ``evaluate_batch`` operator,
    the whole batch of invalid individuals is handed to it in one
    call and ``map`` is not used. Otherwise each individual goes
    through ``map`` and ``evaluate`` as usual.

    Args:
        toolbox: Toolbox with the evaluate and map operators.
        individuals: Individuals to scan for invalid fitness.

    Returns:
        The number of individuals that were evaluated.
    """
    invalids = [ind for ind in individuals if not ind.fitness.is_valid()]
    evaluate_batch = getattr(toolbox, "evaluate_batch", None)
    if evaluate_batch is not None:
        fitness = evaluate_batch(invalids)
    else:
        fitness = toolbox.map(toolbox.evaluate, invalids)
    for ind, fit in zip(invalids, fitness, strict=False):
        ind.fitness.values = fit
    return len(invalids)

apply_policy_action(action, /, **kwargs)

Map a discrete policy action token onto existing toolbox callables.

Push emits action names, not trees. This helper is schema plus thin dispatch only: fitness assignment and rescore ownership stay on the caller. Skip tokens are intentional no-ops. Unknown tokens, missing required kwargs, and guard cap violations are rejected without raising.

Parameters:

Name Type Description Default
action str

One of :data:SUPPORTED_POLICY_ACTIONS.

 required
**kwargs Any

Arguments forwarded to the underlying callable for the chosen action. See that function's docstring. Optional guard (:class:~deap_er.operators.PolicyActionGuard) enforces action caps from :func:~deap_er.operators.guard_policy_action.

 {}

Returns:

Name Type Description
A PolicyActionResult

class:PolicyActionResult describing whether the action
PolicyActionResult

ran, was skipped, or was rejected.
Source code in deap_er/private/algorithms/policy_action.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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def apply_policy_action(action: str, /, **kwargs: Any) -> PolicyActionResult:
    """Map a discrete policy action token onto existing toolbox callables.

    Push emits action names, not trees. This helper is schema plus
    thin dispatch only: fitness assignment and rescore ownership stay
    on the caller. Skip tokens are intentional no-ops. Unknown tokens,
    missing required kwargs, and guard cap violations are rejected
    without raising.

    Args:
        action: One of :data:`SUPPORTED_POLICY_ACTIONS`.
        **kwargs: Arguments forwarded to the underlying callable for
            the chosen action. See that function's docstring. Optional
            ``guard`` (:class:`~deap_er.operators.PolicyActionGuard`)
            enforces action caps from
            :func:`~deap_er.operators.guard_policy_action`.

    Returns:
        A :class:`PolicyActionResult` describing whether the action
        ran, was skipped, or was rejected.
    """
    guard = kwargs.pop("guard", None)
    if action in SKIP_POLICY_ACTIONS:
        return PolicyActionResult(applied=False, rejected=False)
    planned = estimate_policy_action_evals(action, **kwargs) if guard is not None else 0
    if guard is not None and not guard_policy_action(action, guard, **kwargs):
        return PolicyActionResult(applied=False, rejected=True)
    if action == "next_lexicase_cases":
        result = _dispatch_next_lexicase_cases(**kwargs)
    elif action == "tune_ephemerals":
        result = _dispatch_tune_ephemerals(**kwargs)
    elif action == "promote_subtree":
        result = _dispatch_promote_subtree(**kwargs)
    elif action == "evaluate_invalid":
        result = _dispatch_evaluate_invalid(**kwargs)
    elif action == "interpret_tapes":
        result = _dispatch_interpret_tapes(**kwargs)
    elif action == "step_islands":
        result = _dispatch_step_islands(**kwargs)
    else:
        return PolicyActionResult(applied=False, rejected=True)
    if guard is not None and result.applied:
        evals = _applied_eval_cost(action, result, planned)
        guard.note_applied(action, evals=evals)
    return result

estimate_policy_action_evals(action, /, **kwargs)

Estimate how many evaluations an action would spend.

Parameters:

Name Type Description Default
action str

Policy action token.

 required
**kwargs Any

Arguments that would be forwarded to :func:~deap_er.algorithms.apply_policy_action.

 {}

Returns:

Type Description
int

A conservative evaluation count used for budget checks.
Source code in deap_er/private/operators/policy_action_guard.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def estimate_policy_action_evals(action: str, /, **kwargs: Any) -> int:
    """Estimate how many evaluations an action would spend.

    Args:
        action: Policy action token.
        **kwargs: Arguments that would be forwarded to
            :func:`~deap_er.algorithms.apply_policy_action`.

    Returns:
        A conservative evaluation count used for budget checks.
    """
    if action == "tune_ephemerals":
        return _tune_eval_cost(kwargs)
    if action == "evaluate_invalid":
        return _evaluate_invalid_cost(kwargs)
    if action == "step_islands":
        return _step_islands_eval_cost(kwargs)
    return 0

guard_policy_action(action, guard, /, **kwargs)

Return whether action is allowed under guard.

Parameters:

Name Type Description Default
action str

Policy action token.

 required
guard PolicyActionGuard

Guard state and caps.

 required
**kwargs Any

Arguments that would be forwarded to :func:~deap_er.algorithms.apply_policy_action.

 {}

Returns:

Type Description
bool

False when a cap would be violated; otherwise True.
Source code in deap_er/private/operators/policy_action_guard.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
def guard_policy_action(action: str, guard: PolicyActionGuard, /, **kwargs: Any) -> bool:
    """Return whether ``action`` is allowed under ``guard``.

    Args:
        action: Policy action token.
        guard: Guard state and caps.
        **kwargs: Arguments that would be forwarded to
            :func:`~deap_er.algorithms.apply_policy_action`.

    Returns:
        ``False`` when a cap would be violated; otherwise ``True``.
    """
    return guard.allows(action, **kwargs)

step_islands(demes, migrate=None, *, eval_keys=None)

Run one generation on unlike demes, then optionally migrate.

Each deme is a (toolbox, population) pair. The toolbox must provide vary, select, and evaluate (or evaluate_batch). The step is evaluate invalids, vary, evaluate the offspring, then replace the population with select(offspring, len(population)). Populations are modified in place. A MAP-Elites island is a custom vary / select pair that closes over an archive; this function does not call ea_map_elites and does not merge archives.

migrate, if given, receives the list of populations after every deme has stepped. Use mig_ring for a ring; this function does not pick a topology.

Migrants keep their fitness when eval_keys is omitted or every key is equal. Distinct keys mean the destination's cases or matrix differ: immigrant fitness is cleared, including clones created by a replacement migration.

Parameters:

Name Type Description Default
demes Sequence[tuple[Toolbox, list[Individual]]]

(toolbox, population) pairs to step.

 required
migrate Callable[[list[list[Individual]]], None] | None

Optional callable migrate(populations).

 None
eval_keys Sequence[Hashable] | None

Per-deme identity of the evaluation data. Length must match demes when given.

 None

Raises:

Type Description
ValueError

If a toolbox is missing vary or select, or if eval_keys does not match the number of demes.
Source code in deap_er/private/algorithms/step_islands.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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
def step_islands(
    demes: Sequence[tuple[Toolbox, list[Individual]]],
    migrate: Callable[[list[list[Individual]]], None] | None = None,
    *,
    eval_keys: Sequence[Hashable] | None = None,
) -> None:
    """Run one generation on unlike demes, then optionally migrate.

    Each deme is a ``(toolbox, population)`` pair. The toolbox must
    provide ``vary``, ``select``, and ``evaluate`` (or
    ``evaluate_batch``). The step is evaluate invalids, vary,
    evaluate the offspring, then replace the population with
    ``select(offspring, len(population))``. Populations are modified
    in place. A MAP-Elites island is a custom ``vary`` / ``select``
    pair that closes over an archive; this function does not call
    ``ea_map_elites`` and does not merge archives.

    ``migrate``, if given, receives the list of populations after
    every deme has stepped. Use ``mig_ring`` for a ring; this function
    does not pick a topology.

    Migrants keep their fitness when ``eval_keys`` is omitted or every
    key is equal. Distinct keys mean the destination's cases or matrix
    differ: immigrant fitness is cleared, including clones created by
    a replacement migration.

    Args:
        demes: ``(toolbox, population)`` pairs to step.
        migrate: Optional callable ``migrate(populations)``.
        eval_keys: Per-deme identity of the evaluation data. Length
            must match ``demes`` when given.

    Raises:
        ValueError: If a toolbox is missing ``vary`` or ``select``,
            or if ``eval_keys`` does not match the number of demes.
    """
    if eval_keys is not None and len(eval_keys) != len(demes):
        raise ValueError("eval_keys must have one entry per deme.")

    populations: list[list[Individual]] = []
    for toolbox, population in demes:
        _require_operator(toolbox, "vary")
        _require_operator(toolbox, "select")
        evaluate_invalid(toolbox, population)
        offspring = toolbox.vary(population)
        evaluate_invalid(toolbox, offspring)
        population[:] = toolbox.select(offspring, len(population))
        populations.append(population)

    if migrate is None:
        return

    if eval_keys is None or len(set(eval_keys)) <= 1:
        migrate(populations)
        return

    owner = {id(ind): key for key, pop in zip(eval_keys, populations, strict=True) for ind in pop}
    migrate(populations)
    for key, pop in zip(eval_keys, populations, strict=True):
        for ind in pop:
            if owner.get(id(ind)) != key:
                del ind.fitness.values

var_and(toolbox, population, cx_prob, mut_prob)

Clone a population, then apply crossover and mutation independently.

Each of cx_prob and mut_prob must be in [0, 1]. The result is a new list; fitnesses of varied individuals are cleared.

Requires clone, mate, and mutate on toolbox.

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with the variation operators.

 required
population list[Individual]

Individuals to vary.

 required
cx_prob float

Probability of mating each consecutive pair.

 required
mut_prob float

Probability of mutating each individual.

 required

Returns:

Type Description
list[Individual]

A new list of varied individuals.

Raises:

Type Description
ValueError

If either probability is outside [0, 1].
Source code in deap_er/private/algorithms/variation.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def var_and(
    toolbox: Toolbox, population: list[Individual], cx_prob: float, mut_prob: float
) -> list[Individual]:
    """Clone a population, then apply crossover and mutation independently.

    Each of ``cx_prob`` and ``mut_prob`` must be in ``[0, 1]``. The
    result is a new list; fitnesses of varied individuals are cleared.

    Requires ``clone``, ``mate``, and ``mutate`` on ``toolbox``.

    Args:
        toolbox: Toolbox with the variation operators.
        population: Individuals to vary.
        cx_prob: Probability of mating each consecutive pair.
        mut_prob: Probability of mutating each individual.

    Returns:
        A new list of varied individuals.

    Raises:
        ValueError: If either probability is outside ``[0, 1]``.
    """
    err = "The {0} probability must be in the range of [0, 1]."
    if not (0 <= cx_prob <= 1):
        raise ValueError(err.format("crossover"))
    if not (0 <= mut_prob <= 1):
        raise ValueError(err.format("mutation"))

    offspring = [toolbox.clone(ind) for ind in population]

    for i in range(1, len(offspring), 2):
        if rng.random() < cx_prob:
            offspring[i - 1], offspring[i] = toolbox.mate(offspring[i - 1], offspring[i])
            del offspring[i - 1].fitness.values, offspring[i].fitness.values

    for i in range(len(offspring)):
        if rng.random() < mut_prob:
            (offspring[i],) = toolbox.mutate(offspring[i])  # don't remove the comma!
            del offspring[i].fitness.values

    return offspring

var_or(toolbox, population, offsprings, cx_prob, mut_prob)

Build offspring by applying crossover or mutation or copy.

Each of cx_prob and mut_prob must be in [0, 1], and their sum must also be in [0, 1]. The remaining probability copies an unmodified parent. The result is a new list; fitnesses of varied individuals are cleared. A crossover draw from a one-individual pool clones that parent twice and mates the clones.

Requires clone, mate, and mutate on toolbox.

Parameters:

Name Type Description Default
toolbox Toolbox

Toolbox with the variation operators.

 required
population list[Individual]

Individuals to sample from.

 required
offsprings int

Number of individuals to produce.

 required
cx_prob float

Probability of producing a child by crossover.

 required
mut_prob float

Probability of producing a child by mutation.

 required

Returns:

Type Description
list[Individual]

A new list of offspring.

Raises:

Type Description
ValueError

If either probability is outside [0, 1], or if cx_prob + mut_prob is greater than 1.
Source code in deap_er/private/algorithms/variation.py 
 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
115
116
117
118
119
120
121
122
123
124
125
126
def var_or(
    toolbox: Toolbox,
    population: list[Individual],
    offsprings: int,
    cx_prob: float,
    mut_prob: float,
) -> list[Individual]:
    """Build offspring by applying crossover *or* mutation *or* copy.

    Each of ``cx_prob`` and ``mut_prob`` must be in ``[0, 1]``, and
    their sum must also be in ``[0, 1]``. The remaining probability
    copies an unmodified parent. The result is a new list; fitnesses
    of varied individuals are cleared. A crossover draw from a
    one-individual pool clones that parent twice and mates the clones.

    Requires ``clone``, ``mate``, and ``mutate`` on ``toolbox``.

    Args:
        toolbox: Toolbox with the variation operators.
        population: Individuals to sample from.
        offsprings: Number of individuals to produce.
        cx_prob: Probability of producing a child by crossover.
        mut_prob: Probability of producing a child by mutation.

    Returns:
        A new list of offspring.

    Raises:
        ValueError: If either probability is outside ``[0, 1]``, or if
            ``cx_prob + mut_prob`` is greater than 1.
    """
    err = "The {0} probability must be in the range of [0, 1]."
    if not (0 <= cx_prob <= 1):
        raise ValueError(err.format("crossover"))
    if not (0 <= mut_prob <= 1):
        raise ValueError(err.format("mutation"))

    evolve_prob = cx_prob + mut_prob
    if evolve_prob > 1.0:
        raise ValueError(
            "The sum of the crossover and the mutation "
            "probabilities must be in the range of [0, 1]."
        )

    offspring = []
    for _ in range(offsprings):
        op_choice = rng.random()
        if op_choice < cx_prob:
            if len(population) >= 2:
                pair = rng.sample(population, 2)
            else:
                parent = rng.choice(population)
                pair = (parent, parent)
            ind1, ind2 = map(toolbox.clone, pair)
            ind1, ind2 = toolbox.mate(ind1, ind2)
            del ind1.fitness.values
            offspring.append(ind1)
        elif op_choice < evolve_prob:
            ind = toolbox.clone(rng.choice(population))
            (ind,) = toolbox.mutate(ind)  # don't remove the comma!
            del ind.fitness.values
            offspring.append(ind)
        else:
            offspring.append(toolbox.clone(rng.choice(population)))

    return offspring