Operators

deap_er.operators

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

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

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

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

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

SelAGE2WithMemory()

AGE-MOEA-II selection that remembers normalization anchors.

Instances can be registered into a Toolbox.

See the class docstring.

Source code in deap_er/private/operators/sel_age_moea_2.py

def __init__(self) -> None:
    """See the class docstring."""
    self.best_point = numpy.array([])
    self.worst_point = numpy.array([])
    self.extreme_points: ndarray | None = None
    self.curvature = 1.0

__call__(individuals, sel_count)

Select individuals for the next generation.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_age_moea_2.py

def __call__(self, individuals: list[Individual], sel_count: int) -> list[Individual]:
    """Select individuals for the next generation.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.

    Returns:
        The selected individuals.
    """
    best = self.best_point.reshape(-1) if self.best_point.size else None
    worst = self.worst_point.reshape(-1) if self.worst_point.size else None
    return sel_age_moea_2(
        individuals,
        sel_count,
        best_point=best,
        worst_point=worst,
        extreme_points=self.extreme_points,
        _memory=self,
    )

SelMOEADWithMemory(weights, *, scalarization='tchebycheff', theta=5.0)

MOEA/D selection that remembers the ideal point across generations.

Instances can be registered into a Toolbox.

Parameters:

Name	Type	Description	Default
weights	ndarray	Decomposition weight vectors.	required
scalarization	ScalarizationName | ScalarizationFn	"tchebycheff", "pbi", or a callable.	'tchebycheff'
theta	float	PBI penalty parameter.	5.0

See the class docstring.

Source code in deap_er/private/operators/sel_moead.py

def __init__(
    self,
    weights: ndarray,
    *,
    scalarization: ScalarizationName | ScalarizationFn = "tchebycheff",
    theta: float = 5.0,
) -> None:
    """See the class docstring."""
    self.weights = weights
    self.scalarization = scalarization
    self.theta = theta
    self.ideal_point = numpy.full((1, weights.shape[1]), numpy.inf)

__call__(individuals, sel_count)

Select individuals for the next generation.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_moead.py

def __call__(self, individuals: list[Individual], sel_count: int) -> list[Individual]:
    """Select individuals for the next generation.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.

    Returns:
        The selected individuals.
    """
    return sel_moead(
        individuals,
        sel_count,
        self.weights,
        scalarization=self.scalarization,
        theta=self.theta,
        ideal_point=self.ideal_point,
        _memory=self,
    )

SelNSGA3WithMemory(ref_points)

NSGA-III selection that remembers ideal, nadir, and extreme points.

Instances can be registered into a Toolbox.

Parameters:

Name	Type	Description	Default
ref_points	ndarray	Reference points for selection.	required

See the class docstring.

Source code in deap_er/private/operators/sel_nsga_3.py

def __init__(self, ref_points: ndarray) -> None:
    """See the class docstring."""
    self.ref_points = ref_points
    self.best_point = numpy.full((1, ref_points.shape[1]), numpy.inf)
    self.worst_point = numpy.full((1, ref_points.shape[1]), -numpy.inf)
    self.extreme_points: ndarray | None = None

__call__(individuals, sel_count)

Select individuals for the next generation.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_nsga_3.py

def __call__(self, individuals: list[Individual], sel_count: int) -> list[Individual]:
    """Select individuals for the next generation.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.

    Returns:
        The selected individuals.
    """
    chosen = sel_nsga_3(
        individuals,
        sel_count,
        self.ref_points,
        self.best_point,
        self.worst_point,
        self.extreme_points,
        self,
    )
    return chosen

batch_case_matrix(matrix, subset, fit_weights, batch_size, reduction)

Collapse shuffled case batches into one column per batch.

Parameters:

Name	Type	Description	Default
matrix	ndarray	(n_individuals, n_cases) case matrix.	required
subset	list[int]	Case indices eligible for batching.	required
fit_weights	tuple[float, ...]	Per-case maximize/minimize signs from fitness.	required
batch_size	int	Maximum cases per batch.	required
reduction	CaseReduction	Maps a (n_individuals, batch_width) block to (n_individuals,) batch scores.	required

Returns:

Type	Description
ndarray	Reduced matrix with shape (n_individuals, n_batches) and
tuple[float, ...]	the weight copied from the first case in each batch.

Source code in deap_er/private/operators/case_batch_reduce.py

def batch_case_matrix(
    matrix: numpy.ndarray,
    subset: list[int],
    fit_weights: tuple[float, ...],
    batch_size: int,
    reduction: CaseReduction,
) -> tuple[numpy.ndarray, tuple[float, ...]]:
    """Collapse shuffled case batches into one column per batch.

    Args:
        matrix: ``(n_individuals, n_cases)`` case matrix.
        subset: Case indices eligible for batching.
        fit_weights: Per-case maximize/minimize signs from fitness.
        batch_size: Maximum cases per batch.
        reduction: Maps a ``(n_individuals, batch_width)`` block to
            ``(n_individuals,)`` batch scores.

    Returns:
        Reduced matrix with shape ``(n_individuals, n_batches)`` and
        the weight copied from the first case in each batch.
    """
    batches = partition_case_batches(subset, batch_size)
    if not batches:
        return numpy.empty((matrix.shape[0], 0), dtype=numpy.float64), ()
    reduced = numpy.empty((matrix.shape[0], len(batches)), dtype=numpy.float64)
    weights: list[float] = []
    for col, batch in enumerate(batches):
        reduced[:, col] = reduction(matrix[:, batch])
        weights.append(fit_weights[batch[0]])
    return reduced, tuple(weights)

partition_case_batches(subset, batch_size)

Shuffle subset and split it into consecutive batches.

Parameters:

Name	Type	Description	Default
subset	Sequence[int]	Fitness-case indices to batch.	required
batch_size	int	Maximum cases per batch. Must be at least 1.	required

Returns:

Type	Description
list[list[int]]	Batches in shuffled order. The last batch may be shorter.

Raises:

Type	Description
ValueError	If batch_size is not a positive integer.

Source code in deap_er/private/operators/case_batch_reduce.py

def partition_case_batches(subset: Sequence[int], batch_size: int) -> list[list[int]]:
    """Shuffle ``subset`` and split it into consecutive batches.

    Args:
        subset: Fitness-case indices to batch.
        batch_size: Maximum cases per batch. Must be at least ``1``.

    Returns:
        Batches in shuffled order. The last batch may be shorter.

    Raises:
        ValueError: If ``batch_size`` is not a positive integer.
    """
    if isinstance(batch_size, bool) or not isinstance(batch_size, Integral):
        raise ValueError("batch_size must be a positive int")
    size = int(batch_size)
    if size < 1:
        raise ValueError("batch_size must be at least 1")
    order = list(subset)
    rng.shuffle(order)
    return [order[i : i + size] for i in range(0, len(order), size)]

reduce_case_mean(block)

Mean of case columns along axis 1.

Parameters:

Name	Type	Description	Default
block	ndarray	(n_individuals, n_cases) values.	required

Returns:

Type	Description
ndarray	One scalar per individual.

Source code in deap_er/private/operators/case_batch_reduce.py

def reduce_case_mean(block: numpy.ndarray) -> numpy.ndarray:
    """Mean of case columns along axis 1.

    Args:
        block: ``(n_individuals, n_cases)`` values.

    Returns:
        One scalar per individual.
    """
    return numpy.mean(block, axis=1)

reduce_case_mse(block)

Mean squared case values along axis 1.

Parameters:

Name	Type	Description	Default
block	ndarray	(n_individuals, n_cases) values.	required

Returns:

Type	Description
ndarray	One scalar per individual.

Source code in deap_er/private/operators/case_batch_reduce.py

def reduce_case_mse(block: numpy.ndarray) -> numpy.ndarray:
    """Mean squared case values along axis 1.

    Args:
        block: ``(n_individuals, n_cases)`` values.

    Returns:
        One scalar per individual.
    """
    return numpy.mean(block * block, axis=1)

guard_case_exams(exams, elites, *, matrix=None, trust_matrix=False, solved=None, held_out=None, min_cases=1, mode='unsolved', informed=True)

Repair empty exams and all-solved collapse in place.

An empty exam is replaced by last_good, then held_out, then an informed resample of min_cases when informed is true. A collapsed exam (difficulty 0 under mode) unions held_out and, if still collapsed and informed, bumps the subset size through sample_informed_cases. Chronological splits stay on the caller.

Parameters:

Name	Type	Description	Default
exams	ExamLike	Pool or sequence of exams to repair.	required
elites	list[Individual]	Evaluated individuals that supply the case pack.	required
matrix	ndarray | None	Optional (n_elites, n_cases) pack.	None
trust_matrix	bool	When True, matrix is accepted on shape alone.	False
solved	CaseSolved | None	Optional solve predicate. See :func:score_case_exams.	None
held_out	CaseExam | None	Caller-marked exam injected on collapse. A pool's held_out is used when this argument is omitted.	None
min_cases	int	Minimum catalog size after a repair. Combined with a pool's min_cases by taking the larger floor.	1
mode	DifficultyMode	Difficulty used to detect collapse. hamming keeps a specialist subset that unsolved would treat as solved.	'unsolved'
informed	bool	When True, empty or still-collapsed exams may be filled with sample_informed_cases.	True

Returns:

Type	Description
list[CaseExam]	The repaired exam list (the pool's live list when a pool is given).

Raises:

Type	Description
ValueError	If elites is empty, n_cases is 0, or min_cases is less than 1.

Source code in deap_er/private/operators/case_exam_guard.py

def guard_case_exams(
    exams: ExamLike,
    elites: list[Individual],
    *,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
    solved: CaseSolved | None = None,
    held_out: CaseExam | None = None,
    min_cases: int = 1,
    mode: DifficultyMode = "unsolved",
    informed: bool = True,
) -> list[CaseExam]:
    """Repair empty exams and all-solved collapse in place.

    An empty exam is replaced by ``last_good``, then ``held_out``, then
    an informed resample of ``min_cases`` when ``informed`` is true. A
    collapsed exam (difficulty ``0`` under ``mode``) unions
    ``held_out`` and, if still collapsed and ``informed``, bumps the
    subset size through ``sample_informed_cases``. Chronological splits
    stay on the caller.

    Args:
        exams: Pool or sequence of exams to repair.
        elites: Evaluated individuals that supply the case pack.
        matrix: Optional ``(n_elites, n_cases)`` pack.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape alone.
        solved: Optional solve predicate. See :func:`score_case_exams`.
        held_out: Caller-marked exam injected on collapse. A pool's
            ``held_out`` is used when this argument is omitted.
        min_cases: Minimum catalog size after a repair. Combined with
            a pool's ``min_cases`` by taking the larger floor.
        mode: Difficulty used to detect collapse. ``hamming`` keeps a
            specialist subset that ``unsolved`` would treat as solved.
        informed: When ``True``, empty or still-collapsed exams may be
            filled with ``sample_informed_cases``.

    Returns:
        The repaired exam list (the pool's live list when a pool is given).

    Raises:
        ValueError: If ``elites`` is empty, ``n_cases`` is 0, or
            ``min_cases`` is less than 1.
    """
    if min_cases < 1:
        raise ValueError("min_cases must be at least 1")
    n_cases, solve = elite_solve_matrix(elites, matrix, trust_matrix, solved)
    if n_cases == 0:
        raise ValueError("every individual must have a valid fitness of the same length")
    items, pool = bound_case_exams(exams, n_cases)
    extra = held_out
    if extra is None and pool is not None:
        extra = pool.held_out
    floor = min_cases
    if pool is not None:
        floor = max(floor, pool.min_cases)
    last_good = pool.last_good if pool is not None else None
    for exam in items:
        _repair_case_exam(
            exam,
            elites,
            n_cases,
            solve,
            extra,
            last_good,
            floor,
            matrix,
            trust_matrix,
            solved,
            mode,
            informed,
        )
        if exam.as_cases(n_cases):
            last_good = exam.copy()
            if pool is not None:
                pool.last_good = last_good
    return items

next_lexicase_cases(exams, elites, *, matrix=None, trust_matrix=False, solved=None, case_count=None, informed=True, mut_prob=0.2, mode='unsolved', held_out=None, min_cases=1, length=None)

Vary exams and return the next sel_lexicase(..., cases=) subset.

Scores exams on elites, mutates ranges or mask runs, then guards empty and collapsed exams. The mutated or guarded winner is the cases= list. informed only enables sample_informed_cases inside that guard — it does not overwrite a healthy winner. The default path and informed=False both keep variation.

Parameters:

Name	Type	Description	Default
exams	ExamLike	Pool or sequence of exams.	required
elites	list[Individual]	Evaluated individuals that supply the case pack.	required
matrix	ndarray | None	Optional (n_elites, n_cases) pack.	None
trust_matrix	bool	When True, matrix is accepted on shape alone.	False
solved	CaseSolved | None	Optional solve predicate. See :func:score_case_exams.	None
case_count	int | None	Minimum catalog size forwarded to the guard when a repair is needed. Defaults to min_cases.	None
informed	bool	When True, empty or still-collapsed exams may be filled with sample_informed_cases. A healthy winner is never resampled.	True
mut_prob	float	Per-range or per-run mutation probability.	0.2
mode	DifficultyMode	Difficulty used to pick the exam that feeds lexicase and to detect collapse in the guard.	'unsolved'
held_out	CaseExam | None	Caller-marked exam injected on collapse.	None
min_cases	int	Minimum catalog size after a guard repair.	1
length	int | None	Bound for range mutation. Defaults to each exam's series span or n_cases.	None

Returns:

Type	Description
list[int]	Case indices for the next lexicase call.

Raises:

Type	Description
ValueError	If elites or exams is empty, or n_cases is 0.

Source code in deap_er/private/operators/case_exam_step.py

def next_lexicase_cases(
    exams: ExamLike,
    elites: list[Individual],
    *,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
    solved: CaseSolved | None = None,
    case_count: int | None = None,
    informed: bool = True,
    mut_prob: float = 0.2,
    mode: DifficultyMode = "unsolved",
    held_out: CaseExam | None = None,
    min_cases: int = 1,
    length: int | None = None,
) -> list[int]:
    """Vary exams and return the next ``sel_lexicase(..., cases=)`` subset.

    Scores exams on ``elites``, mutates ranges or mask runs, then guards
    empty and collapsed exams. The mutated or guarded winner is the
    ``cases=`` list. ``informed`` only enables
    ``sample_informed_cases`` inside that guard — it does not overwrite
    a healthy winner. The default path and ``informed=False`` both keep
    variation.

    Args:
        exams: Pool or sequence of exams.
        elites: Evaluated individuals that supply the case pack.
        matrix: Optional ``(n_elites, n_cases)`` pack.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape alone.
        solved: Optional solve predicate. See :func:`score_case_exams`.
        case_count: Minimum catalog size forwarded to the guard when a
            repair is needed. Defaults to ``min_cases``.
        informed: When ``True``, empty or still-collapsed exams may be
            filled with ``sample_informed_cases``. A healthy winner is
            never resampled.
        mut_prob: Per-range or per-run mutation probability.
        mode: Difficulty used to pick the exam that feeds lexicase and
            to detect collapse in the guard.
        held_out: Caller-marked exam injected on collapse.
        min_cases: Minimum catalog size after a guard repair.
        length: Bound for range mutation. Defaults to each exam's
            series span or ``n_cases``.

    Returns:
        Case indices for the next lexicase call.

    Raises:
        ValueError: If ``elites`` or ``exams`` is empty, or ``n_cases`` is 0.
    """
    n_cases, _solve = elite_solve_matrix(elites, matrix, trust_matrix, solved)
    items, pool = bound_case_exams(exams, n_cases)
    if not items:
        raise ValueError("exams must be non-empty")
    for exam in items:
        span = exam.mutation_bound(n_cases) if length is None else length
        _vary_exam(exam, span, mut_prob)
    source: ExamLike = pool if pool is not None else items
    floor = min_cases
    if case_count is not None:
        floor = max(floor, int(case_count))
    repaired = guard_case_exams(
        source,
        elites,
        matrix=matrix,
        trust_matrix=trust_matrix,
        solved=solved,
        held_out=held_out,
        min_cases=floor,
        mode=mode,
        informed=informed,
    )
    scores = score_case_exams(
        repaired,
        elites,
        matrix=matrix,
        trust_matrix=trust_matrix,
        solved=solved,
        mode=mode,
    )
    winner = _argmax_ties(scores)
    return repaired[winner].as_cases(n_cases)

score_case_exams(exams, elites, *, matrix=None, trust_matrix=False, solved=None, mode='unsolved')

Score case subsets on elites by how many cases they still fool.

A case is solved when its value is within 1e-12 of zero, matching sample_informed_cases. unsolved counts selected cases that no elite solves. hamming counts unsolved elite-case pairs.

Parameters:

Name	Type	Description	Default
exams	ExamLike	Exams, a :class:~deap_er.records.CaseExamPool, raw masks, or a flat list of catalog indices (one exam).	required
elites	list[Individual]	Evaluated individuals that supply the case pack.	required
matrix	ndarray | None	Optional (n_elites, n_cases) pack. Ignored when solved is not the default zero test.	None
trust_matrix	bool	When True, matrix is accepted on shape alone.	False
solved	CaseSolved | None	Optional (individual, case) -> bool predicate.	None
mode	DifficultyMode	unsolved or hamming.	'unsolved'

Returns:

Type	Description
list[int]	One difficulty score per exam.

Raises:

Type	Description
ValueError	If elites is empty, fitness lengths differ, or mode is unknown.

Source code in deap_er/private/operators/case_exams.py

def score_case_exams(
    exams: ExamLike,
    elites: list[Individual],
    *,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
    solved: CaseSolved | None = None,
    mode: DifficultyMode = "unsolved",
) -> list[int]:
    """Score case subsets on elites by how many cases they still fool.

    A case is solved when its value is within ``1e-12`` of zero, matching
    ``sample_informed_cases``. ``unsolved`` counts selected cases that no
    elite solves. ``hamming`` counts unsolved elite-case pairs.

    Args:
        exams: Exams, a :class:`~deap_er.records.CaseExamPool`, raw
            masks, or a flat list of catalog indices (one exam).
        elites: Evaluated individuals that supply the case pack.
        matrix: Optional ``(n_elites, n_cases)`` pack. Ignored when
            ``solved`` is not the default zero test.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape alone.
        solved: Optional ``(individual, case) -> bool`` predicate.
        mode: ``unsolved`` or ``hamming``.

    Returns:
        One difficulty score per exam.

    Raises:
        ValueError: If ``elites`` is empty, fitness lengths differ, or
            ``mode`` is unknown.
    """
    n_cases, solve = elite_solve_matrix(elites, matrix, trust_matrix, solved)
    items, _pool = bound_case_exams(exams, n_cases)
    return [exam_difficulty(solve, exam.as_cases(n_cases), mode) for exam in items]

constraint_dominates(ind1, ind2, *, feasible=None, violation=None)

Return whether ind1 constrained-dominates ind2.

Deb's NSGA-II rule (2002, §III-A): a feasible individual beats an infeasible one; two feasibles use ordinary Pareto dominance on fitness; two infeasibles prefer the smaller constraint violation. Fitness values are not rewritten.

When only violation is given, <= 0 is feasible. When only feasible is given, infeasibles do not dominate each other. When both are given, the flag decides feasibility and the violation is used only among infeasibles.

Parameters:

Name	Type	Description	Default
ind1	Individual	Candidate that may dominate.	required
ind2	Individual	Candidate that may be dominated.	required
feasible	Callable[[Individual], bool] | None	Predicate that reports whether an individual is feasible. Optional if violation is given.	None
violation	Callable[[Individual], float] | None	Function returning a scalar constraint violation. Optional if feasible is given.	None

Returns:

Type	Description
bool	True if ind1 constrained-dominates ind2.

Raises:

Type	Description
TypeError	If both callables are omitted, a given argument is not callable, or violation does not return a real scalar.
ValueError	If violation returns a non-finite number.

Source code in deap_er/private/operators/constraint_dominates.py

def constraint_dominates(
    ind1: Individual,
    ind2: Individual,
    *,
    feasible: Callable[[Individual], bool] | None = None,
    violation: Callable[[Individual], float] | None = None,
) -> bool:
    """Return whether ``ind1`` constrained-dominates ``ind2``.

    Deb's NSGA-II rule (2002, §III-A): a feasible individual beats an
    infeasible one; two feasibles use ordinary Pareto dominance on
    ``fitness``; two infeasibles prefer the smaller constraint
    violation. Fitness values are not rewritten.

    When only ``violation`` is given, ``<= 0`` is feasible. When only
    ``feasible`` is given, infeasibles do not dominate each other.
    When both are given, the flag decides feasibility and the
    violation is used only among infeasibles.

    Args:
        ind1: Candidate that may dominate.
        ind2: Candidate that may be dominated.
        feasible: Predicate that reports whether an individual is
            feasible. Optional if ``violation`` is given.
        violation: Function returning a scalar constraint violation.
            Optional if ``feasible`` is given.

    Returns:
        True if ``ind1`` constrained-dominates ``ind2``.

    Raises:
        TypeError: If both callables are omitted, a given argument is
            not callable, or ``violation`` does not return a real
            scalar.
        ValueError: If ``violation`` returns a non-finite number.
    """
    _require_constraint_callables(feasible, violation)
    left = _is_feasible(ind1, feasible, violation)
    right = _is_feasible(ind2, feasible, violation)
    if left != right:
        return left
    if left:
        return bool(ind1.fitness.dominates(ind2.fitness))
    if violation is None:
        return False
    return _as_violation(violation(ind1)) < _as_violation(violation(ind2))

cx_heterogeneous(ind1, ind2, crossovers)

Mate two mixed-encoding individuals with per-gene or per-slice operators.

Both individuals are modified in place. crossovers is either one callable per gene or (slice, callable) pairs (tuple or list), not a mix of the two. A single pair may be passed without wrapping it in another sequence.

In the per-gene form each callable receives the pair of gene values and must return the two replacements. In the per-slice form each callable is an existing cx_* operator: it receives the extracted units, may mutate them in place, and should return the two replacements. Open slices such as slice(3, None) resolve against the individual length. Uncovered genes are left unchanged.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required
crossovers	Sequence[Any]	Per-gene (v1, v2) -> (v1', v2') callables, or (slice, cx_*) pairs for contiguous blocks. A bare pair is one slice unit.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Raises:

Type	Description
ValueError	If the individuals have different lengths, the per-gene list length does not match, the two shapes are mixed, a slice is not contiguous or is inverted, slices overlap, a callable does not return a pair, or a slice operator changes the unit length.

Source code in deap_er/private/operators/cx_hetero.py

def cx_heterogeneous(
    ind1: Individual,
    ind2: Individual,
    crossovers: Sequence[Any],
) -> Mates:
    """Mate two mixed-encoding individuals with per-gene or per-slice operators.

    Both individuals are modified in place. ``crossovers`` is either one
    callable per gene or ``(slice, callable)`` pairs (tuple or list), not
    a mix of the two. A single pair may be passed without wrapping it
    in another sequence.

    In the per-gene form each callable receives the pair of gene values
    and must return the two replacements. In the per-slice form each
    callable is an existing ``cx_*`` operator: it receives the extracted
    units, may mutate them in place, and should return the two
    replacements. Open slices such as ``slice(3, None)`` resolve against
    the individual length. Uncovered genes are left unchanged.

    Args:
        ind1: The first individual.
        ind2: The second individual.
        crossovers: Per-gene ``(v1, v2) -> (v1', v2')`` callables, or
            ``(slice, cx_*)`` pairs for contiguous blocks. A bare pair
            is one slice unit.

    Returns:
        The two individuals after crossover.

    Raises:
        ValueError: If the individuals have different lengths, the
            per-gene list length does not match, the two shapes are
            mixed, a slice is not contiguous or is inverted, slices
            overlap, a callable does not return a pair, or a slice
            operator changes the unit length.
    """
    size = _require_same_length(ind1, ind2)
    specs = _slice_specs(crossovers)
    if specs is not None:
        _apply_slices(ind1, ind2, specs, size)
    else:
        if len(crossovers) != size:
            raise ValueError(
                "crossovers must have the same length as the individual: "
                f"{len(crossovers)} != {size}"
            )
        _apply_genes(ind1, ind2, crossovers)
    return ind1, ind2

cx_ordered(ind1, ind2)

Execute an ordered crossover on two individuals.

Both individuals are modified in place. Alleles may be any hashable values that form a shared permutation.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Raises:

Type	Description
ValueError	If the individuals are not permutations of the same allele set.

Source code in deap_er/private/operators/cx_permutation.py

def cx_ordered(ind1: Individual, ind2: Individual) -> Mates:
    """Execute an ordered crossover on two individuals.

    Both individuals are modified in place. Alleles may be any
    hashable values that form a shared permutation.

    Args:
        ind1: The first individual.
        ind2: The second individual.

    Returns:
        The two individuals after crossover.

    Raises:
        ValueError: If the individuals are not permutations of the
            same allele set.
    """
    size = min(len(ind1), len(ind2))
    _allele_maps(ind1, ind2, size)
    if size < 2:
        return ind1, ind2
    a, b = rng.sample(list(range(size)), 2)
    if a > b:
        a, b = b, a

    holes1 = {ind2[i] for i in range(a, b + 1)}
    holes2 = {ind1[i] for i in range(a, b + 1)}

    temp1, temp2 = ind1, ind2
    k1, k2 = b + 1, b + 1

    for i in range(size):
        src1 = temp1[(i + b + 1) % size]
        if src1 not in holes1:
            ind1[k1 % size] = src1
            k1 += 1

        src2 = temp2[(i + b + 1) % size]
        if src2 not in holes2:
            ind2[k2 % size] = src2
            k2 += 1

    for i in range(a, b + 1):
        slicer(ind1, ind2, i, i + 1)

    return ind1, ind2

cx_partially_matched(ind1, ind2)

Execute a partially matched crossover on two individuals.

Both individuals are modified in place. Alleles may be any hashable values that form a shared permutation.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Raises:

Type	Description
ValueError	If the individuals are not permutations of the same allele set.

Source code in deap_er/private/operators/cx_permutation.py

def cx_partially_matched(ind1: Individual, ind2: Individual) -> Mates:
    """Execute a partially matched crossover on two individuals.

    Both individuals are modified in place. Alleles may be any
    hashable values that form a shared permutation.

    Args:
        ind1: The first individual.
        ind2: The second individual.

    Returns:
        The two individuals after crossover.

    Raises:
        ValueError: If the individuals are not permutations of the
            same allele set.
    """
    size = min(len(ind1), len(ind2))
    p1, p2 = _allele_maps(ind1, ind2, size)
    if size < 2:
        return ind1, ind2

    cxp1 = rng.randint(0, size)
    cxp2 = rng.randint(0, size - 1)

    if cxp2 >= cxp1:
        cxp2 += 1
    else:
        cxp1, cxp2 = cxp2, cxp1

    for i in range(cxp1, cxp2):
        match(ind1, ind2, p1, p2, i)

    return ind1, ind2

cx_uniform_partially_matched(ind1, ind2, cx_prob)

Execute a uniform partially matched crossover on two individuals.

Both individuals are modified in place. Alleles may be any hashable values that form a shared permutation.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required
cx_prob	float	Probability of swapping any two traits.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Raises:

Type	Description
ValueError	If the individuals are not permutations of the same allele set.

Source code in deap_er/private/operators/cx_permutation.py

def cx_uniform_partially_matched(ind1: Individual, ind2: Individual, cx_prob: float) -> Mates:
    """Execute a uniform partially matched crossover on two individuals.

    Both individuals are modified in place. Alleles may be any
    hashable values that form a shared permutation.

    Args:
        ind1: The first individual.
        ind2: The second individual.
        cx_prob: Probability of swapping any two traits.

    Returns:
        The two individuals after crossover.

    Raises:
        ValueError: If the individuals are not permutations of the
            same allele set.
    """
    size = min(len(ind1), len(ind2))
    p1, p2 = _allele_maps(ind1, ind2, size)

    for i in range(size):
        if rng.random() < cx_prob:
            match(ind1, ind2, p1, p2, i)

    return ind1, ind2

cx_es_two_point(ind1, ind2)

Execute a two-point crossover on two individuals and their strategies.

Both individuals and their strategy vectors are modified in place.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_point.py

def cx_es_two_point(ind1: Individual, ind2: Individual) -> Mates:
    """Execute a two-point crossover on two individuals and their strategies.

    Both individuals and their ``strategy`` vectors are modified in place.

    Args:
        ind1: The first individual.
        ind2: The second individual.

    Returns:
        The two individuals after crossover.
    """
    ind1, ind2 = two_point(ind1, ind2, strategy=True)
    return ind1, ind2

cx_es_two_point_copy(ind1, ind2)

Execute a two-point crossover on copies of two individuals and their strategies.

Use this instead of cx_es_two_point when the individuals are based on numpy arrays, to avoid incorrect mating behavior due to the specifics of the numpy array datatype.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_point.py

def cx_es_two_point_copy(ind1: Individual, ind2: Individual) -> Mates:
    """Execute a two-point crossover on copies of two individuals and their strategies.

    Use this instead of ``cx_es_two_point`` when the individuals are
    based on numpy arrays, to avoid incorrect mating behavior due to
    the specifics of the numpy array datatype.

    Args:
        ind1: The first individual.
        ind2: The second individual.

    Returns:
        The two individuals after crossover.
    """
    ind1, ind2 = two_point(ind1, ind2, copy=True, strategy=True)
    return ind1, ind2

cx_messy_one_point(ind1, ind2)

Execute a messy one-point crossover on two individuals.

Cut points are chosen independently, so the individuals may change length. Both individuals are modified in place.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_point.py

def cx_messy_one_point(ind1: Individual, ind2: Individual) -> Mates:
    """Execute a messy one-point crossover on two individuals.

    Cut points are chosen independently, so the individuals may change
    length. Both individuals are modified in place.

    Args:
        ind1: The first individual.
        ind2: The second individual.

    Returns:
        The two individuals after crossover.
    """
    cxp1 = rng.randint(0, len(ind1))
    cxp2 = rng.randint(0, len(ind2))
    tail1 = _segment(ind1, slice(cxp1, None))
    tail2 = _segment(ind2, slice(cxp2, None))
    ind1[cxp1:], ind2[cxp2:] = tail2, tail1
    return ind1, ind2

cx_one_point(ind1, ind2)

Execute a one-point crossover on two individuals.

Both individuals are modified in place.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_point.py

def cx_one_point(ind1: Individual, ind2: Individual) -> Mates:
    """Execute a one-point crossover on two individuals.

    Both individuals are modified in place.

    Args:
        ind1: The first individual.
        ind2: The second individual.

    Returns:
        The two individuals after crossover.
    """
    size = min(len(ind1), len(ind2))
    if size < 2:
        return ind1, ind2
    cxp = rng.randint(1, size - 1)
    ind1, ind2 = slicer(ind1, ind2, cxp)
    return ind1, ind2

cx_two_point(ind1, ind2)

Execute a two-point crossover on two individuals.

Both individuals are modified in place.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_point.py

def cx_two_point(ind1: Individual, ind2: Individual) -> Mates:
    """Execute a two-point crossover on two individuals.

    Both individuals are modified in place.

    Args:
        ind1: The first individual.
        ind2: The second individual.

    Returns:
        The two individuals after crossover.
    """
    ind1, ind2 = two_point(ind1, ind2)
    return ind1, ind2

cx_two_point_copy(ind1, ind2)

Execute a two-point crossover on copies of two individuals.

Use this instead of cx_two_point when the individuals are based on numpy arrays, to avoid incorrect mating behavior due to the specifics of the numpy array datatype.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_point.py

def cx_two_point_copy(ind1: Individual, ind2: Individual) -> Mates:
    """Execute a two-point crossover on copies of two individuals.

    Use this instead of ``cx_two_point`` when the individuals are
    based on numpy arrays, to avoid incorrect mating behavior due
    to the specifics of the numpy array datatype.

    Args:
        ind1: The first individual.
        ind2: The second individual.

    Returns:
        The two individuals after crossover.
    """
    ind1, ind2 = two_point(ind1, ind2, copy=True)
    return ind1, ind2

cx_blend(ind1, ind2, alpha)

Execute a blend crossover on two individuals.

Both individuals are modified in place.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required
alpha	float	Extent of the interval in which the new values can be drawn for each attribute on both sides of the parents' attributes.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_real.py

def cx_blend(ind1: Individual, ind2: Individual, alpha: float) -> Mates:
    """Execute a blend crossover on two individuals.

    Both individuals are modified in place.

    Args:
        ind1: The first individual.
        ind2: The second individual.
        alpha: Extent of the interval in which the new values can be
            drawn for each attribute on both sides of the parents'
            attributes.

    Returns:
        The two individuals after crossover.
    """
    for i, (x1, x2) in enumerate(zip(ind1, ind2, strict=False)):
        gamma = (1.0 + 2.0 * alpha) * rng.random() - alpha
        ind1[i] = (1.0 - gamma) * x1 + gamma * x2
        ind2[i] = gamma * x1 + (1.0 - gamma) * x2

    return ind1, ind2

cx_blend_bounded(ind1, ind2, alpha, low, up)

Execute a bounded blend crossover on two individuals.

Both individuals are modified in place. Each child gene is clamped to [low, up] after the blend draw.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required
alpha	float	Extent of the interval in which the new values can be drawn for each attribute on both sides of the parents' attributes.	required
low	NumOrSeq	Lower bound of the search space.	required
up	NumOrSeq	Upper bound of the search space.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Raises:

Type	Description
ValueError	If a bound sequence is shorter than the shorter individual.

Source code in deap_er/private/operators/cx_real.py

def cx_blend_bounded(
    ind1: Individual, ind2: Individual, alpha: float, low: NumOrSeq, up: NumOrSeq
) -> Mates:
    """Execute a bounded blend crossover on two individuals.

    Both individuals are modified in place. Each child gene is
    clamped to ``[low, up]`` after the blend draw.

    Args:
        ind1: The first individual.
        ind2: The second individual.
        alpha: Extent of the interval in which the new values can be
            drawn for each attribute on both sides of the parents'
            attributes.
        low: Lower bound of the search space.
        up: Upper bound of the search space.

    Returns:
        The two individuals after crossover.

    Raises:
        ValueError: If a bound sequence is shorter than the shorter
            individual.
    """
    size = min(len(ind1), len(ind2))
    low = broadcast_param("low", low, size, _SHORTER_INDIVIDUAL)
    up = broadcast_param("up", up, size, _SHORTER_INDIVIDUAL)

    for i, xl, xu in zip(list(range(size)), low, up, strict=False):
        if xu <= xl:
            continue
        x1, x2 = ind1[i], ind2[i]
        gamma = (1.0 + 2.0 * alpha) * rng.random() - alpha
        ind1[i] = min(max((1.0 - gamma) * x1 + gamma * x2, xl), xu)
        ind2[i] = min(max(gamma * x1 + (1.0 - gamma) * x2, xl), xu)

    return ind1, ind2

cx_es_blend(ind1, ind2, alpha)

Execute a blend crossover on two individuals and their strategies.

Both individuals and their strategy vectors are modified in place.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required
alpha	float	Extent of the interval in which the new values can be drawn for each attribute on both sides of the parents' attributes.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_real.py

def cx_es_blend(ind1: Individual, ind2: Individual, alpha: float) -> Mates:
    """Execute a blend crossover on two individuals and their strategies.

    Both individuals and their ``strategy`` vectors are modified in place.

    Args:
        ind1: The first individual.
        ind2: The second individual.
        alpha: Extent of the interval in which the new values can be
            drawn for each attribute on both sides of the parents'
            attributes.

    Returns:
        The two individuals after crossover.
    """
    zipper = zip(ind1, ind1.strategy, ind2, ind2.strategy, strict=False)
    for i, (x1, s1, x2, s2) in enumerate(zipper):
        gamma = (1.0 + 2.0 * alpha) * rng.random() - alpha
        ind1[i] = (1.0 - gamma) * x1 + gamma * x2
        ind2[i] = gamma * x1 + (1.0 - gamma) * x2

        gamma = (1.0 + 2.0 * alpha) * rng.random() - alpha
        ind1.strategy[i] = (1.0 - gamma) * s1 + gamma * s2
        ind2.strategy[i] = gamma * s1 + (1.0 - gamma) * s2

    return ind1, ind2

cx_simulated_binary(ind1, ind2, eta)

Execute a simulated binary crossover on two individuals.

Both individuals are modified in place.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required
eta	float	Crowding degree of the crossover. Higher values produce children more similar to their parents; smaller values produce children more divergent from their parents.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_real.py

def cx_simulated_binary(ind1: Individual, ind2: Individual, eta: float) -> Mates:
    """Execute a simulated binary crossover on two individuals.

    Both individuals are modified in place.

    Args:
        ind1: The first individual.
        ind2: The second individual.
        eta: Crowding degree of the crossover. Higher values produce
            children more similar to their parents; smaller values
            produce children more divergent from their parents.

    Returns:
        The two individuals after crossover.
    """
    for i, (x1, x2) in enumerate(zip(ind1, ind2, strict=False)):
        rand = rng.random()

        beta = 2.0 * rand if rand <= 0.5 else 1.0 / (2.0 * (1.0 - rand))

        beta **= 1.0 / (eta + 1.0)
        ind1[i] = 0.5 * (((1 + beta) * x1) + ((1 - beta) * x2))
        ind2[i] = 0.5 * (((1 - beta) * x1) + ((1 + beta) * x2))

    return ind1, ind2

cx_simulated_binary_bounded(ind1, ind2, eta, low, up)

Execute a bounded simulated binary crossover on two individuals.

Both individuals are modified in place.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required
eta	float	Crowding degree of the crossover. Higher values produce children more similar to their parents; smaller values produce children more divergent from their parents.	required
low	NumOrSeq	Lower bound of the search space.	required
up	NumOrSeq	Upper bound of the search space.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Raises:

Type	Description
ValueError	If eta is not greater than 0, or if a bound sequence is shorter than the shorter individual.

Source code in deap_er/private/operators/cx_real.py

def cx_simulated_binary_bounded(
    ind1: Individual, ind2: Individual, eta: float, low: NumOrSeq, up: NumOrSeq
) -> Mates:
    """Execute a bounded simulated binary crossover on two individuals.

    Both individuals are modified in place.

    Args:
        ind1: The first individual.
        ind2: The second individual.
        eta: Crowding degree of the crossover. Higher values produce
            children more similar to their parents; smaller values
            produce children more divergent from their parents.
        low: Lower bound of the search space.
        up: Upper bound of the search space.

    Returns:
        The two individuals after crossover.

    Raises:
        ValueError: If ``eta`` is not greater than 0, or if a bound
            sequence is shorter than the shorter individual.
    """
    require_positive_eta(eta)

    def calc_c(diff: float, side: float) -> float:
        """Map a gap to the bound into one bounded SBX child value.

        Args:
            diff: Distance from the nearer parent to the active bound.
            side: ``-1`` for the lower child, ``+1`` for the upper child.

        Returns:
            One child coordinate for the current parent pair.
        """
        beta = 1.0 + (2.0 * diff / (x2 - x1))
        alpha = 2.0 - beta ** -(eta + 1)
        if rand <= 1.0 / alpha:
            beta_q = (rand * alpha) ** (1.0 / (eta + 1))
        else:
            beta_q = (1.0 / (2.0 - rand * alpha)) ** (1.0 / (eta + 1))
        c = 0.5 * (x1 + x2 + side * beta_q * (x2 - x1))
        return float(c)

    size = min(len(ind1), len(ind2))
    low = broadcast_param("low", low, size, _SHORTER_INDIVIDUAL)
    up = broadcast_param("up", up, size, _SHORTER_INDIVIDUAL)

    for i, xl, xu in zip(list(range(size)), low, up, strict=False):
        if xu <= xl:
            continue
        if rng.random() <= 0.5 and abs(ind1[i] - ind2[i]) > 1e-14:
            x1 = min(max(min(ind1[i], ind2[i]), xl), xu)
            x2 = min(max(max(ind1[i], ind2[i]), xl), xu)
            if abs(x1 - x2) <= 1e-14:
                continue
            rand = rng.random()

            c1 = calc_c(x1 - xl, -1.0)
            c1 = min(max(c1, xl), xu)

            c2 = calc_c(xu - x2, 1.0)
            c2 = min(max(c2, xl), xu)

            if rng.random() <= 0.5:
                c1, c2 = c2, c1
            ind1[i] = c1
            ind2[i] = c2

    return ind1, ind2

cx_uniform(ind1, ind2, cx_prob)

Execute a uniform crossover on two individuals.

Both individuals are modified in place.

Parameters:

Name	Type	Description	Default
ind1	Individual	The first individual.	required
ind2	Individual	The second individual.	required
cx_prob	float	Probability of swapping any two traits.	required

Returns:

Type	Description
Mates	The two individuals after crossover.

Source code in deap_er/private/operators/cx_real.py

def cx_uniform(ind1: Individual, ind2: Individual, cx_prob: float) -> Mates:
    """Execute a uniform crossover on two individuals.

    Both individuals are modified in place.

    Args:
        ind1: The first individual.
        ind2: The second individual.
        cx_prob: Probability of swapping any two traits.

    Returns:
        The two individuals after crossover.
    """
    size = min(len(ind1), len(ind2))
    for i in range(size):
        if rng.random() < cx_prob:
            slicer(ind1, ind2, i, i + 1)
    return ind1, ind2

next_downsample_cases(individuals, case_count, generation, *, mode='random', cohort=None, cohorts=None, held_out=None, matrix=None, trust_matrix=False)

Return the next cases= list for lexicase down-sampling.

Chronological meaning stays on the caller. This helper only picks catalog indices for one generation.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Evaluated population supplying fitness length.	required
case_count	int	Target subset size. Values above the catalog are capped. case_count <= 0 returns [].	required
generation	int	Generation index used to rotate cohorts or held-out windows.	required
mode	DownsampleMode	random, informed, cohort, or held_out.	'random'
cohort	Sequence[int] | None	Fixed case indices for mode="cohort". Must supply at least case_count distinct valid indices.	None
cohorts	Sequence[Sequence[int]] | None	Rotating cohort lists for mode="cohort". When both cohort and cohorts are set, cohort wins. Each cohort must supply at least case_count distinct valid indices when selected.	None
held_out	CaseExam | None	Caller-marked exam for mode="held_out".	None
matrix	ndarray | None	Optional (n_individuals, n_cases) pack for informed mode.	None
trust_matrix	bool	When True, matrix is accepted on shape alone. Defaults to False.	False

Returns:

Type	Description
list[int]	Distinct fitness-case indices for the next lexicase call.

Raises:

Type	Description
ValueError	If individuals is empty, case_count is not an integer, or a mode-specific argument is missing.

Source code in deap_er/private/operators/downsample_schedule.py

def next_downsample_cases(
    individuals: list[Individual],
    case_count: int,
    generation: int,
    *,
    mode: DownsampleMode = "random",
    cohort: Sequence[int] | None = None,
    cohorts: Sequence[Sequence[int]] | None = None,
    held_out: CaseExam | None = None,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
) -> list[int]:
    """Return the next ``cases=`` list for lexicase down-sampling.

    Chronological meaning stays on the caller. This helper only picks
    catalog indices for one generation.

    Args:
        individuals: Evaluated population supplying fitness length.
        case_count: Target subset size. Values above the catalog are
            capped. ``case_count <= 0`` returns ``[]``.
        generation: Generation index used to rotate cohorts or
            held-out windows.
        mode: ``random``, ``informed``, ``cohort``, or ``held_out``.
        cohort: Fixed case indices for ``mode="cohort"``. Must supply at
            least ``case_count`` distinct valid indices.
        cohorts: Rotating cohort lists for ``mode="cohort"``. When
            both ``cohort`` and ``cohorts`` are set, ``cohort`` wins.
            Each cohort must supply at least ``case_count`` distinct
            valid indices when selected.
        held_out: Caller-marked exam for ``mode="held_out"``.
        matrix: Optional ``(n_individuals, n_cases)`` pack for
            informed mode.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape
            alone. Defaults to ``False``.

    Returns:
        Distinct fitness-case indices for the next lexicase call.

    Raises:
        ValueError: If ``individuals`` is empty, ``case_count`` is not
            an integer, or a mode-specific argument is missing.
    """
    if isinstance(case_count, bool) or not isinstance(case_count, Integral):
        raise ValueError("case_count must be an int")
    count = int(case_count)
    if not individuals:
        raise ValueError("individuals must be non-empty")
    if count <= 0:
        return []
    n_cases = len(individuals[0].fitness.values)
    if n_cases == 0:
        raise ValueError("every individual must have a valid fitness of the same length")
    size = min(count, n_cases)
    if mode == "informed":
        return sample_informed_cases(
            individuals,
            size,
            matrix=matrix,
            trust_matrix=trust_matrix,
        )
    if mode == "cohort":
        return _cohort_cases(cohort, cohorts, generation, size, n_cases)
    if mode == "held_out":
        return _held_out_cases(held_out, generation, size, n_cases)
    return _random_cases(size, n_cases)

island_eval_keys(exams, *, n_cases, matrix=None, matrices=None)

Return per-deme keys for step_islands(..., eval_keys=).

Keys compare equal when the exam subset and optional matrix identity match, so migrants keep fitness only across demes that evaluate on the same cases and packed matrix. Catalog exams — masks or ranges that fit in n_cases — are canonicalized to a painted boolean mask so equivalent subsets share a key even when stored differently. Series exams keep normalized ranges.

Parameters:

Name	Type	Description	Default
exams	Sequence[CaseExam]	One exam per deme.	required
n_cases	int	Fitness-case count from the current pack.	required
matrix	object | None	Optional matrix shared by every deme.	None
matrices	Sequence[object] | None	Optional per-deme matrices. Must match exams in length. Ignored when matrix is given.	None

Returns:

Type	Description
tuple[tuple[Any, ...], ...]	A tuple of hashable keys, one per exam.

Raises:

Type	Description
ValueError	If both matrix and matrices are given, or if matrices length does not match exams.

Source code in deap_er/private/operators/island_eval_keys.py

def island_eval_keys(
    exams: Sequence[CaseExam],
    *,
    n_cases: int,
    matrix: object | None = None,
    matrices: Sequence[object] | None = None,
) -> tuple[tuple[Any, ...], ...]:
    """Return per-deme keys for ``step_islands(..., eval_keys=)``.

    Keys compare equal when the exam subset and optional matrix
    identity match, so migrants keep fitness only across demes that
    evaluate on the same cases and packed matrix. Catalog exams —
    masks or ranges that fit in ``n_cases`` — are canonicalized to a
    painted boolean mask so equivalent subsets share a key even when
    stored differently. Series exams keep normalized ranges.

    Args:
        exams: One exam per deme.
        n_cases: Fitness-case count from the current pack.
        matrix: Optional matrix shared by every deme.
        matrices: Optional per-deme matrices. Must match ``exams`` in
            length. Ignored when ``matrix`` is given.

    Returns:
        A tuple of hashable keys, one per exam.

    Raises:
        ValueError: If both ``matrix`` and ``matrices`` are given, or
            if ``matrices`` length does not match ``exams``.
    """
    if matrix is not None and matrices is not None:
        raise ValueError("pass either matrix or matrices, not both")
    if matrices is not None and len(matrices) != len(exams):
        raise ValueError("matrices must have one entry per exam")
    matrix_parts: list[tuple[int | None, int] | None]
    if matrix is not None:
        shared = _matrix_part(matrix)
        matrix_parts = [shared] * len(exams)
    elif matrices is not None:
        matrix_parts = [_matrix_part(item) for item in matrices]
    else:
        matrix_parts = [None] * len(exams)
    return tuple((_exam_part(exam, n_cases), matrix_parts[idx]) for idx, exam in enumerate(exams))

mig_fully_connected(populations, mig_count, selection, replacement=None)

Move emigrants along every directed island edge.

For each ordered pair of distinct demes (src, dst), selection picks mig_count emigrants from src and writes them into dst using the same vacancy and cloning rules as mig_ring. Emigrants are selected once per source, then cloned along each outgoing edge so a destination update on another deme does not change who leaves. Destinations claim distinct vacancy indices across incoming edges so a later src does not overwrite an earlier immigrant in the same slot. Edges run in (dst, src) order; when replacement is omitted that order can still matter for which home individuals are displaced. Cost is O(n_demes² · mig_count) selection calls. Deme lengths are unchanged. Populations are modified in place.

Parameters:

Name	Type	Description	Default
populations	list[list[Individual]]	Populations to migrate between.	required
mig_count	int	Number of individuals to migrate along each edge.	required
selection	Callable[..., Any]	Callable that selects emigrants from a population.	required
replacement	Callable[..., Any] | None	Callable that selects destination vacancies in the receiving deme. If omitted, the receiver's own emigrant slots are the vacancies.	None

Source code in deap_er/private/operators/mig_ring.py

def mig_fully_connected(
    populations: list[list[Individual]],
    mig_count: int,
    selection: Callable[..., Any],
    replacement: Callable[..., Any] | None = None,
) -> None:
    """Move emigrants along every directed island edge.

    For each ordered pair of distinct demes ``(src, dst)``, ``selection``
    picks ``mig_count`` emigrants from ``src`` and writes them into
    ``dst`` using the same vacancy and cloning rules as ``mig_ring``.
    Emigrants are selected once per source, then cloned along each
    outgoing edge so a destination update on another deme does not
    change who leaves. Destinations claim distinct vacancy indices
    across incoming edges so a later ``src`` does not overwrite an
    earlier immigrant in the same slot. Edges run in ``(dst, src)``
    order; when ``replacement`` is omitted that order can still
    matter for which home individuals are displaced. Cost is
    ``O(n_demes² · mig_count)`` selection calls. Deme lengths are
    unchanged. Populations are modified in place.

    Args:
        populations: Populations to migrate between.
        mig_count: Number of individuals to migrate along each edge.
        selection: Callable that selects emigrants from a population.
        replacement: Callable that selects destination vacancies in the
            receiving deme. If omitted, the receiver's own emigrant
            slots are the vacancies.
    """
    nbr_demes = len(populations)
    emigrants = [selection(populations[from_deme], mig_count) for from_deme in range(nbr_demes)]
    claimed: list[set[int]] = [set() for _ in range(nbr_demes)]
    for to_deme in range(nbr_demes):
        for from_deme in range(nbr_demes):
            if from_deme != to_deme:
                used = _mig_edge_emigrants(
                    populations,
                    to_deme,
                    emigrants[from_deme],
                    selection,
                    replacement,
                    claimed[to_deme],
                )
                claimed[to_deme].update(used)

mig_random(populations, mig_count, selection, replacement=None)

Move emigrants to a random destination deme per source.

Each source population sends mig_count emigrants to one destination chosen uniformly among the other demes. When only one deme exists, this is a no-op. Otherwise the placement rules match mig_ring. Populations are modified in place.

Parameters:

Name	Type	Description	Default
populations	list[list[Individual]]	Populations to migrate between.	required
mig_count	int	Number of individuals to migrate from each population.	required
selection	Callable[..., Any]	Callable that selects emigrants from a population.	required
replacement	Callable[..., Any] | None	Callable that selects which destination individuals are replaced. If omitted, the destination's own emigrant slots are the vacancies.	None

Source code in deap_er/private/operators/mig_ring.py

def mig_random(
    populations: list[list[Individual]],
    mig_count: int,
    selection: Callable[..., Any],
    replacement: Callable[..., Any] | None = None,
) -> None:
    """Move emigrants to a random destination deme per source.

    Each source population sends ``mig_count`` emigrants to one
    destination chosen uniformly among the other demes. When only one
    deme exists, this is a no-op. Otherwise the placement rules match
    ``mig_ring``. Populations are modified in place.

    Args:
        populations: Populations to migrate between.
        mig_count: Number of individuals to migrate from each population.
        selection: Callable that selects emigrants from a population.
        replacement: Callable that selects which destination individuals
            are replaced. If omitted, the destination's own emigrant
            slots are the vacancies.
    """
    nbr_demes = len(populations)
    if nbr_demes < 2:
        return
    choices = list(range(nbr_demes))
    mig_indices = [int(rng.choice([j for j in choices if j != i])) for i in choices]
    mig_ring(populations, mig_count, selection, replacement, mig_indices=mig_indices)

mig_ring(populations, mig_count, selection, replacement=None, mig_indices=None)

Move emigrants between populations along a ring (or custom map).

From each population, selection picks mig_count emigrants. Those individuals replace members of the destination population. When a source sends more emigrants than the destination has vacancies, or a deme is smaller than mig_count, only as many individuals as both sides can hold are moved. Deme lengths are unchanged. When replacement is omitted, an emigrant whose home vacancy is not filled is cloned so the same object is not left in two demes. A duplicate emigrant already present in the destination is cloned so two dest slots do not share one object. Populations are modified in place.

Parameters:

Name	Type	Description	Default
populations	list[list[Individual]]	Populations to migrate between.	required
mig_count	int	Number of individuals to migrate from each population.	required
selection	Callable[..., Any]	Callable that selects emigrants from a population.	required
replacement	Callable[..., Any] | None	Callable that selects which destination individuals are replaced. If omitted, the destination's own emigrants are the vacancies.	None
mig_indices	list[int] | None	Destination index for each source population. If omitted, each population sends to the next and the last wraps to the first.	None

Source code in deap_er/private/operators/mig_ring.py

def mig_ring(
    populations: list[list[Individual]],
    mig_count: int,
    selection: Callable[..., Any],
    replacement: Callable[..., Any] | None = None,
    mig_indices: list[int] | None = None,
) -> None:
    """Move emigrants between populations along a ring (or custom map).

    From each population, ``selection`` picks ``mig_count`` emigrants.
    Those individuals replace members of the destination population.
    When a source sends more emigrants than the destination has
    vacancies, or a deme is smaller than ``mig_count``, only as
    many individuals as both sides can hold are moved. Deme
    lengths are unchanged. When ``replacement`` is omitted, an
    emigrant whose home vacancy is not filled is cloned so the
    same object is not left in two demes. A duplicate emigrant
    already present in the destination is cloned so two dest
    slots do not share one object. Populations are modified
    in place.

    Args:
        populations: Populations to migrate between.
        mig_count: Number of individuals to migrate from each population.
        selection: Callable that selects emigrants from a population.
        replacement: Callable that selects which destination individuals
            are replaced. If omitted, the destination's own emigrants
            are the vacancies.
        mig_indices: Destination index for each source population. If
            omitted, each population sends to the next and the last
            wraps to the first.
    """
    nbr_demes = len(populations)
    if mig_indices is None:
        mig_indices = list(range(1, nbr_demes)) + [0]

    emigrants: list[list[Individual]] = [[] for _ in range(nbr_demes)]
    vacancies: list[list[int]] = [[] for _ in range(nbr_demes)]

    for from_deme in range(nbr_demes):
        selected = selection(populations[from_deme], mig_count)
        if replacement is None:
            emigrants[from_deme].extend(selected)
            dest_slots = selected
        else:
            emigrants[from_deme].extend(clone_individual(ind) for ind in selected)
            dest_slots = replacement(populations[from_deme], mig_count)
        vacancies[from_deme] = _claim_vacancies(populations[from_deme], dest_slots)

    incoming_filled = _incoming_filled(emigrants, vacancies, mig_indices)
    _place_emigrants(
        populations,
        emigrants,
        vacancies,
        mig_indices,
        incoming_filled,
        replacement,
    )

mut_case_mask(mask, *, mut_prob)

Flip contiguous True/False runs of a boolean mask in place.

Each run is flipped independently with probability mut_prob. mut_prob <= 0 is a no-op.

Parameters:

Name	Type	Description	Default
mask	ndarray	One-dimensional mutable boolean array.	required
mut_prob	float	Probability of flipping each contiguous run.	required

Returns:

Type	Description
tuple[ndarray]	A one-element tuple containing mask.

Source code in deap_er/private/operators/mut_case_exam.py

def mut_case_mask(mask: numpy.ndarray, *, mut_prob: float) -> tuple[numpy.ndarray]:
    """Flip contiguous True/False runs of a boolean mask in place.

    Each run is flipped independently with probability ``mut_prob``.
    ``mut_prob <= 0`` is a no-op.

    Args:
        mask: One-dimensional mutable boolean array.
        mut_prob: Probability of flipping each contiguous run.

    Returns:
        A one-element tuple containing ``mask``.
    """
    if mut_prob <= 0:
        return (mask,)
    runs = _mask_runs(mask)
    if not runs:
        return (mask,)
    draws = rng.take_floats(len(runs))
    for (start, stop), draw in zip(runs, draws, strict=True):
        if draw < mut_prob:
            slc = mask[start:stop]
            slc[:] = ~slc
    return (mask,)

mut_case_ranges(ranges, *, length, mut_prob)

Jitter half-open case ranges in place.

Each interval is mutated independently with probability mut_prob. Endpoints stay inside [0, length] and are swapped if they cross. mut_prob <= 0 is a no-op.

Parameters:

Name	Type	Description	Default
ranges	list[tuple[int, int]]	Mutable list of (start, stop) pairs.	required
length	int	Exclusive upper bound for stop.	required
mut_prob	float	Probability of mutating each interval.	required

Returns:

Type	Description
tuple[list[tuple[int, int]]]	A one-element tuple containing ranges.

Raises:

Type	Description
ValueError	If length is negative.

Source code in deap_er/private/operators/mut_case_exam.py

def mut_case_ranges(
    ranges: list[tuple[int, int]],
    *,
    length: int,
    mut_prob: float,
) -> tuple[list[tuple[int, int]]]:
    """Jitter half-open case ranges in place.

    Each interval is mutated independently with probability ``mut_prob``.
    Endpoints stay inside ``[0, length]`` and are swapped if they cross.
    ``mut_prob <= 0`` is a no-op.

    Args:
        ranges: Mutable list of ``(start, stop)`` pairs.
        length: Exclusive upper bound for ``stop``.
        mut_prob: Probability of mutating each interval.

    Returns:
        A one-element tuple containing ``ranges``.

    Raises:
        ValueError: If ``length`` is negative.
    """
    if length < 0:
        raise ValueError("length must be non-negative")
    if mut_prob <= 0 or not ranges:
        return (ranges,)
    draws = rng.take_floats(len(ranges))
    for i, (start, stop) in enumerate(ranges):
        if draws[i] < mut_prob:
            ranges[i] = _jitter_interval(int(start), int(stop), length)
    return (ranges,)

mut_de(individual, a, b, c, scale, cx_prob, *, low=None, up=None)

Write a DE/rand/1/bin trial onto individual.

Genes selected by the binomial mask (rate cx_prob, at least one gene forced) become a[i] + scale * (b[i] - c[i]). The individual is modified in place. The caller supplies donors and keeps the trial when it is better. Optional low / up clamp written genes only.

Parameters:

Name	Type	Description	Default
individual	Individual	Trial vector to overwrite. Clone the parent first.	required
a	Individual	Base donor.	required
b	Individual	First difference donor.	required
c	Individual	Second difference donor.	required
scale	float	Difference weight F.	required
cx_prob	float	Per-gene crossover rate CR.	required
low	NumOrSeq | None	Lower bound of the search space. Optional.	None
up	NumOrSeq | None	Upper bound of the search space. Optional.	None

Returns:

Type	Description
Mutant	A one-element tuple containing the trial individual.

Raises:

Type	Description
ValueError	If a donor is shorter than the individual, if only one of low / up is set, or if a bound sequence is shorter than the individual.

Source code in deap_er/private/operators/mut_de.py

def mut_de(
    individual: Individual,
    a: Individual,
    b: Individual,
    c: Individual,
    scale: float,
    cx_prob: float,
    *,
    low: NumOrSeq | None = None,
    up: NumOrSeq | None = None,
) -> Mutant:
    """Write a DE/rand/1/bin trial onto *individual*.

    Genes selected by the binomial mask (rate ``cx_prob``, at least
    one gene forced) become ``a[i] + scale * (b[i] - c[i])``. The
    individual is modified in place. The caller supplies donors and
    keeps the trial when it is better. Optional ``low`` / ``up``
    clamp written genes only.

    Args:
        individual: Trial vector to overwrite. Clone the parent first.
        a: Base donor.
        b: First difference donor.
        c: Second difference donor.
        scale: Difference weight ``F``.
        cx_prob: Per-gene crossover rate ``CR``.
        low: Lower bound of the search space. Optional.
        up: Upper bound of the search space. Optional.

    Returns:
        A one-element tuple containing the trial individual.

    Raises:
        ValueError: If a donor is shorter than the individual, if
            only one of ``low`` / ``up`` is set, or if a bound
            sequence is shorter than the individual.
    """
    size = len(individual)
    if size == 0:
        return (individual,)
    if min(len(a), len(b), len(c)) < size:
        raise ValueError(f"{_DONOR_SHORT}: {min(len(a), len(b), len(c))} < {size}")
    if (low is None) ^ (up is None):
        raise ValueError(_BOUNDS_PAIR)

    lows = ups = None
    if low is not None and up is not None:
        lows = broadcast_param("low", low, size)
        ups = broadcast_param("up", up, size)

    index = rng.randrange(size)
    draws = rng.take_floats(size)
    for i, draw in enumerate(draws):
        if i != index and draw >= cx_prob:
            continue
        gene = a[i] + scale * (b[i] - c[i])
        if lows is not None and ups is not None:
            xl, xu = lows[i], ups[i]
            if xu > xl:
                gene = min(max(gene, xl), xu)
        individual[i] = gene

    return (individual,)

mut_gaussian_bounded(individual, mu, sigma, low, up, mut_prob)

Apply a Gaussian mutation and clamp each mutated gene into a box.

The individual is modified in place. The draw is the same N(mu, sigma) add as mut_gaussian. mu, sigma, low, and up may be scalars or per-gene sequences. An empty interval (up <= low) is skipped.

Parameters:

Name	Type	Description	Default
individual	Individual	Individual to mutate.	required
mu	NumOrSeq	Mean of the Gaussian mutation.	required
sigma	NumOrSeq	Standard deviation of the Gaussian mutation.	required
low	NumOrSeq	Lower bound of the search space.	required
up	NumOrSeq	Upper bound of the search space.	required
mut_prob	float	Probability of mutating each attribute.	required

Returns:

Type	Description
Mutant	A one-element tuple containing the mutated individual.

Raises:

Type	Description
ValueError	If mu, sigma, low, or up is a sequence shorter than the individual.

Source code in deap_er/private/operators/mut_gaussian_bounded.py

def mut_gaussian_bounded(
    individual: Individual,
    mu: NumOrSeq,
    sigma: NumOrSeq,
    low: NumOrSeq,
    up: NumOrSeq,
    mut_prob: float,
) -> Mutant:
    """Apply a Gaussian mutation and clamp each mutated gene into a box.

    The individual is modified in place. The draw is the same
    ``N(mu, sigma)`` add as ``mut_gaussian``. ``mu``, ``sigma``,
    ``low``, and ``up`` may be scalars or per-gene sequences. An empty
    interval (``up <= low``) is skipped.

    Args:
        individual: Individual to mutate.
        mu: Mean of the Gaussian mutation.
        sigma: Standard deviation of the Gaussian mutation.
        low: Lower bound of the search space.
        up: Upper bound of the search space.
        mut_prob: Probability of mutating each attribute.

    Returns:
        A one-element tuple containing the mutated individual.

    Raises:
        ValueError: If ``mu``, ``sigma``, ``low``, or ``up`` is a
            sequence shorter than the individual.
    """
    size = len(individual)
    mu = broadcast_param("mu", mu, size)
    sigma = broadcast_param("sigma", sigma, size)
    low = broadcast_param("low", low, size)
    up = broadcast_param("up", up, size)

    idx = list(range(size))
    for i, m, s, xl, xu in zip(idx, mu, sigma, low, up, strict=False):
        if rng.random() < mut_prob:
            if xu <= xl:
                continue
            individual[i] = min(max(individual[i] + rng.gauss(m, s), xl), xu)

    return (individual,)

mut_heterogeneous(individual, mutators, mut_prob)

Mutate each gene with its own callable.

The individual is modified in place. mutators[i] receives the current value of gene i and must return the replacement.

Parameters:

Name	Type	Description	Default
individual	Individual	Individual to mutate.	required
mutators	Sequence[Callable[[Any], Any]]	One gene mutator per attribute.	required
mut_prob	float	Probability of mutating each attribute.	required

Returns:

Type	Description
Mutant	A one-element tuple containing the mutated individual.

Raises:

Type	Description
ValueError	If mutators and individual have different lengths.

Source code in deap_er/private/operators/mut_hetero.py

def mut_heterogeneous(
    individual: Individual,
    mutators: Sequence[Callable[[Any], Any]],
    mut_prob: float,
) -> Mutant:
    """Mutate each gene with its own callable.

    The individual is modified in place. ``mutators[i]`` receives the
    current value of gene ``i`` and must return the replacement.

    Args:
        individual: Individual to mutate.
        mutators: One gene mutator per attribute.
        mut_prob: Probability of mutating each attribute.

    Returns:
        A one-element tuple containing the mutated individual.

    Raises:
        ValueError: If ``mutators`` and ``individual`` have different
            lengths.
    """
    if len(mutators) != len(individual):
        raise ValueError(
            "mutators must have the same length as the individual: "
            f"{len(mutators)} != {len(individual)}"
        )
    for i, mutate_gene in enumerate(mutators):
        if rng.random() < mut_prob:
            individual[i] = mutate_gene(individual[i])
    return (individual,)

iso_line_bit(parent, donor, iso, sigma)

Move toward donor along the line, then apply isotropic flips.

t maps to a Bernoulli draw toward donor. sigma is the probability of flipping the chosen bit afterward.

Parameters:

Name	Type	Description	Default
parent	bool	Current gene value.	required
donor	bool	Elite donor value.	required
iso	float	Line extension for t ~ Uniform(-iso, 1 + iso).	required
sigma	float	Flip probability after the line draw.	required

Returns:

Type	Description
bool	The mutated boolean gene.

Source code in deap_er/private/operators/mut_iso_line.py

def iso_line_bit(parent: bool, donor: bool, iso: float, sigma: float) -> bool:
    """Move toward ``donor`` along the line, then apply isotropic flips.

    ``t`` maps to a Bernoulli draw toward ``donor``. ``sigma`` is the
    probability of flipping the chosen bit afterward.

    Args:
        parent: Current gene value.
        donor: Elite donor value.
        iso: Line extension for ``t ~ Uniform(-iso, 1 + iso)``.
        sigma: Flip probability after the line draw.

    Returns:
        The mutated boolean gene.
    """
    t = _sample_t(iso)
    value = bool(donor) if rng.random() < max(0.0, min(1.0, t)) else bool(parent)
    if sigma > 0.0 and rng.random() < min(1.0, sigma):
        value = not value
    return value

iso_line_float(parent, donor, iso, sigma, *, low=None, up=None)

Interpolate toward donor and add isotropic Gaussian noise.

Parameters:

Name	Type	Description	Default
parent	float	Current gene value.	required
donor	float	Elite donor value.	required
iso	float	Line extension for t ~ Uniform(-iso, 1 + iso).	required
sigma	float	Standard deviation of the isotropic perturbation.	required
low	float | None	Optional lower bound applied after the draw.	None
up	float | None	Optional upper bound applied after the draw.	None

Returns:

Type	Description
float	The mutated gene value.

Source code in deap_er/private/operators/mut_iso_line.py

def iso_line_float(
    parent: float,
    donor: float,
    iso: float,
    sigma: float,
    *,
    low: float | None = None,
    up: float | None = None,
) -> float:
    """Interpolate toward ``donor`` and add isotropic Gaussian noise.

    Args:
        parent: Current gene value.
        donor: Elite donor value.
        iso: Line extension for ``t ~ Uniform(-iso, 1 + iso)``.
        sigma: Standard deviation of the isotropic perturbation.
        low: Optional lower bound applied after the draw.
        up: Optional upper bound applied after the draw.

    Returns:
        The mutated gene value.
    """
    gene = parent + _sample_t(iso) * (donor - parent) + rng.gauss(0.0, sigma)
    if low is not None and up is not None and up > low:
        gene = min(max(gene, low), up)
    return float(gene)

iso_line_int(parent, donor, iso, sigma, *, low, up)

Interpolate toward donor, add noise, round, and clamp.

Parameters:

Name	Type	Description	Default
parent	int	Current gene value.	required
donor	int	Elite donor value.	required
iso	float	Line extension for t ~ Uniform(-iso, 1 + iso).	required
sigma	float	Standard deviation of the isotropic perturbation.	required
low	int	Inclusive lower bound.	required
up	int	Inclusive upper bound.	required

Returns:

Type	Description
int	The mutated integer gene.

Source code in deap_er/private/operators/mut_iso_line.py

def iso_line_int(
    parent: int,
    donor: int,
    iso: float,
    sigma: float,
    *,
    low: int,
    up: int,
) -> int:
    """Interpolate toward ``donor``, add noise, round, and clamp.

    Args:
        parent: Current gene value.
        donor: Elite donor value.
        iso: Line extension for ``t ~ Uniform(-iso, 1 + iso)``.
        sigma: Standard deviation of the isotropic perturbation.
        low: Inclusive lower bound.
        up: Inclusive upper bound.

    Returns:
        The mutated integer gene.
    """
    raw = parent + _sample_t(iso) * (donor - parent) + rng.gauss(0.0, sigma)
    if up < low:
        return int(parent)
    return int(min(max(int(round(raw)), low), up))

mut_iso_line(individual, donor, iso, sigma, *, low=None, up=None)

Apply iso+line mutation toward an archive elite donor.

Each gene becomes parent + t * (donor - parent) + N(0, sigma) with t ~ Uniform(-iso, 1 + iso). Booleans use iso_line_bit; integers (that are not bool) round and clamp when bounds are given; other numeric genes clamp when bounds are given. The individual is modified in place.

Parameters:

Name	Type	Description	Default
individual	Individual	Parent to overwrite. Clone first when needed.	required
donor	Individual	Elite donor from archive.random_elites.	required
iso	float	Line extension parameter.	required
sigma	float	Isotropic noise standard deviation (flip rate for bool).	required
low	NumOrSeq | None	Lower search bound. Optional.	None
up	NumOrSeq | None	Upper search bound. Optional.	None

Returns:

Type	Description
Mutant	A one-element tuple containing the mutated individual.

Raises:

Type	Description
ValueError	If donor is shorter than individual, or if only one of low / up is set, or if a bound sequence is shorter than the individual.

Source code in deap_er/private/operators/mut_iso_line.py

def mut_iso_line(
    individual: Individual,
    donor: Individual,
    iso: float,
    sigma: float,
    *,
    low: NumOrSeq | None = None,
    up: NumOrSeq | None = None,
) -> Mutant:
    """Apply iso+line mutation toward an archive elite donor.

    Each gene becomes ``parent + t * (donor - parent) + N(0, sigma)``
    with ``t ~ Uniform(-iso, 1 + iso)``. Booleans use ``iso_line_bit``;
    integers (that are not bool) round and clamp when bounds are given;
    other numeric genes clamp when bounds are given. The individual is
    modified in place.

    Args:
        individual: Parent to overwrite. Clone first when needed.
        donor: Elite donor from ``archive.random_elites``.
        iso: Line extension parameter.
        sigma: Isotropic noise standard deviation (flip rate for bool).
        low: Lower search bound. Optional.
        up: Upper search bound. Optional.

    Returns:
        A one-element tuple containing the mutated individual.

    Raises:
        ValueError: If ``donor`` is shorter than ``individual``, or if
            only one of ``low`` / ``up`` is set, or if a bound sequence
            is shorter than the individual.
    """
    size = len(individual)
    if len(donor) < size:
        raise ValueError(f"{_DONOR_SHORT}: {len(donor)} < {size}")
    if (low is None) ^ (up is None):
        raise ValueError(_BOUNDS_PAIR)

    lows = ups = None
    if low is not None and up is not None:
        lows = broadcast_param("low", low, size)
        ups = broadcast_param("up", up, size)

    for index in range(size):
        xl = lows[index] if lows is not None else None
        xu = ups[index] if ups is not None else None
        individual[index] = _mutate_iso_line_gene(
            individual[index],
            donor[index],
            iso,
            sigma,
            low=xl,
            up=xu,
        )

    return (individual,)

mut_es_log_normal(individual, learn_rate, mut_prob)

Mutate an evolution strategy according to its strategy attribute.

The individual is modified in place. Genes are updated only when the individual has a strategy attribute.

Parameters:

Name	Type	Description	Default
individual	Individual	Individual to mutate.	required
learn_rate	float	Learning rate of the evolution strategy. For an evolution strategy of (10, 100) the recommended value is 1.	required
mut_prob	float	Probability of mutating each attribute.	required

Returns:

Type	Description
Mutant	A one-element tuple containing the mutated individual.

Source code in deap_er/private/operators/mut_various.py

def mut_es_log_normal(individual: Individual, learn_rate: float, mut_prob: float) -> Mutant:
    """Mutate an evolution strategy according to its ``strategy`` attribute.

    The individual is modified in place. Genes are updated only when
    the individual has a ``strategy`` attribute.

    Args:
        individual: Individual to mutate.
        learn_rate: Learning rate of the evolution strategy. For an
            evolution strategy of (10, 100) the recommended value is 1.
        mut_prob: Probability of mutating each attribute.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    size = len(individual)
    t = learn_rate / math.sqrt(2.0 * math.sqrt(size))
    t0 = learn_rate / math.sqrt(2.0 * size)
    n = rng.gauss(0, 1)
    t0_n = t0 * n

    for indx in range(size):
        if rng.random() < mut_prob and hasattr(individual, "strategy"):
            individual.strategy[indx] *= math.exp(t0_n + t * rng.gauss(0, 1))
            individual[indx] += individual.strategy[indx] * rng.gauss(0, 1)

    return (individual,)

mut_flip_bit(individual, mut_prob)

Flip the values of random attributes of the individual.

The individual is modified in place.

Parameters:

Name	Type	Description	Default
individual	Individual	Individual to mutate.	required
mut_prob	float	Probability of mutating each attribute.	required

Returns:

Type	Description
Mutant	A one-element tuple containing the mutated individual.

Source code in deap_er/private/operators/mut_various.py

def mut_flip_bit(individual: Individual, mut_prob: float) -> Mutant:
    """Flip the values of random attributes of the individual.

    The individual is modified in place.

    Args:
        individual: Individual to mutate.
        mut_prob: Probability of mutating each attribute.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    draws = rng.take_floats(len(individual))
    for i, u in enumerate(draws):
        if u < mut_prob:
            individual[i] = type(individual[i])(not individual[i])

    return (individual,)

mut_gaussian(individual, mu, sigma, mut_prob)

Apply a Gaussian mutation of mean mu and standard deviation sigma.

The individual is modified in place. mu and sigma may be scalars or per-gene sequences.

Parameters:

Name	Type	Description	Default
individual	Individual	Individual to mutate.	required
mu	NumOrSeq	Mean of the Gaussian mutation.	required
sigma	NumOrSeq	Standard deviation of the Gaussian mutation.	required
mut_prob	float	Probability of mutating each attribute.	required

Returns:

Type	Description
Mutant	A one-element tuple containing the mutated individual.

Raises:

Type	Description
ValueError	If mu or sigma is a sequence shorter than the individual.

Source code in deap_er/private/operators/mut_various.py

def mut_gaussian(individual: Individual, mu: NumOrSeq, sigma: NumOrSeq, mut_prob: float) -> Mutant:
    """Apply a Gaussian mutation of mean *mu* and standard deviation *sigma*.

    The individual is modified in place. ``mu`` and ``sigma`` may be
    scalars or per-gene sequences.

    Args:
        individual: Individual to mutate.
        mu: Mean of the Gaussian mutation.
        sigma: Standard deviation of the Gaussian mutation.
        mut_prob: Probability of mutating each attribute.

    Returns:
        A one-element tuple containing the mutated individual.

    Raises:
        ValueError: If ``mu`` or ``sigma`` is a sequence shorter than
            the individual.
    """
    size = len(individual)
    mu = broadcast_param("mu", mu, size)
    sigma = broadcast_param("sigma", sigma, size)

    idx = list(range(size))
    for i, m, s in zip(idx, mu, sigma, strict=False):
        if rng.random() < mut_prob:
            individual[i] += rng.gauss(m, s)

    return (individual,)

mut_polynomial_bounded(individual, eta, low, up, mut_prob)

Apply a bounded polynomial mutation with crowding degree eta.

The individual is modified in place. low and up may be scalars or per-gene sequences.

Parameters:

Name	Type	Description	Default
individual	Individual	Individual to mutate.	required
eta	float	Crowding degree of the mutation. Higher values produce children more similar to their parents; smaller values produce children more divergent from their parents.	required
low	NumOrSeq	Lower bound of the search space.	required
up	NumOrSeq	Upper bound of the search space.	required
mut_prob	float	Probability of mutating each attribute.	required

Returns:

Type	Description
Mutant	A one-element tuple containing the mutated individual.

Raises:

Type	Description
ValueError	If eta is not greater than 0, or if low or up is a sequence shorter than the individual.

Source code in deap_er/private/operators/mut_various.py

def mut_polynomial_bounded(
    individual: Individual, eta: float, low: NumOrSeq, up: NumOrSeq, mut_prob: float
) -> Mutant:
    """Apply a bounded polynomial mutation with crowding degree *eta*.

    The individual is modified in place. ``low`` and ``up`` may be
    scalars or per-gene sequences.

    Args:
        individual: Individual to mutate.
        eta: Crowding degree of the mutation. Higher values produce
            children more similar to their parents; smaller values
            produce children more divergent from their parents.
        low: Lower bound of the search space.
        up: Upper bound of the search space.
        mut_prob: Probability of mutating each attribute.

    Returns:
        A one-element tuple containing the mutated individual.

    Raises:
        ValueError: If ``eta`` is not greater than 0, or if ``low``
            or ``up`` is a sequence shorter than the individual.
    """
    require_positive_eta(eta)
    size = len(individual)
    low = broadcast_param("low", low, size)
    up = broadcast_param("up", up, size)

    idx = list(range(size))
    for i, xl, xu in zip(idx, low, up, strict=False):
        if rng.random() <= mut_prob:
            if xu <= xl:
                continue
            x = min(max(individual[i], xl), xu)
            delta_1 = (x - xl) / (xu - xl)
            delta_2 = (xu - x) / (xu - xl)
            rand = rng.random()
            mut_pow = 1.0 / (eta + 1.0)

            if rand < 0.5:
                xy = 1.0 - delta_1
                val = 2.0 * rand + (1.0 - 2.0 * rand) * xy ** (eta + 1)
                delta_q = val**mut_pow - 1.0
            else:
                xy = 1.0 - delta_2
                val = 2.0 * (1.0 - rand) + 2.0 * (rand - 0.5) * xy ** (eta + 1)
                delta_q = 1.0 - val**mut_pow

            x = x + delta_q * (xu - xl)
            x = min(max(x, xl), xu)
            individual[i] = x

    return (individual,)

mut_shuffle_indexes(individual, mut_prob)

Shuffle attributes of the individual.

The individual is modified in place.

Parameters:

Name	Type	Description	Default
individual	Individual	Individual to mutate.	required
mut_prob	float	Probability of mutating each attribute.	required

Returns:

Type	Description
Mutant	A one-element tuple containing the mutated individual.

Source code in deap_er/private/operators/mut_various.py

def mut_shuffle_indexes(individual: Individual, mut_prob: float) -> Mutant:
    """Shuffle attributes of the individual.

    The individual is modified in place.

    Args:
        individual: Individual to mutate.
        mut_prob: Probability of mutating each attribute.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    size = len(individual)
    if size < 2:
        return (individual,)
    for i in range(size):
        if rng.random() < mut_prob:
            swap_indx = rng.randint(0, size - 2)
            if swap_indx >= i:
                swap_indx += 1
            individual[i], individual[swap_indx] = individual[swap_indx], individual[i]

    return (individual,)

mut_uniform_int(individual, low, up, mut_prob)

Replace attributes with integers drawn uniformly from [low, up].

The individual is modified in place. Bounds are inclusive.

Parameters:

Name	Type	Description	Default
individual	Individual	Individual to mutate.	required
low	int	Lower bound of the search space.	required
up	int	Upper bound of the search space.	required
mut_prob	float	Probability of mutating each attribute.	required

Returns:

Type	Description
Mutant	A one-element tuple containing the mutated individual.

Raises:

Type	Description
ValueError	If low or up is a sequence shorter than the individual.

Source code in deap_er/private/operators/mut_various.py

def mut_uniform_int(individual: Individual, low: int, up: int, mut_prob: float) -> Mutant:
    """Replace attributes with integers drawn uniformly from [*low*, *up*].

    The individual is modified in place. Bounds are inclusive.

    Args:
        individual: Individual to mutate.
        low: Lower bound of the search space.
        up: Upper bound of the search space.
        mut_prob: Probability of mutating each attribute.

    Returns:
        A one-element tuple containing the mutated individual.

    Raises:
        ValueError: If ``low`` or ``up`` is a sequence shorter than
            the individual.
    """
    size = len(individual)
    lows = broadcast_param("low", low, size)
    ups = broadcast_param("up", up, size)

    idx = list(range(size))
    for i, xl, xu in zip(idx, lows, ups, strict=False):
        if rng.random() < mut_prob:
            individual[i] = rng.randint(int(xl), int(xu))

    return (individual,)

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

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

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)

guard_policy_fitness_exam(fitness_exam, *, held_out, n_cases, train_exams=None, mutated_exam=None)

Refuse policy fitness that targets a train or mutated exam.

Policy individuals must be scored only on held_out. Train-exam quality belongs in :func:~deap_er.tools.policy_observe, not in fitness assignment.

Parameters:

Name	Type	Description	Default
fitness_exam	CaseExam	Exam the caller would score for policy fitness.	required
held_out	CaseExam	Caller-marked held-out exam.	required
n_cases	int	Catalog length from the current elite pack.	required
train_exams	list[CaseExam] | None	Optional train exams that must not become the fitness target.	None
mutated_exam	CaseExam | None	Optional exam the policy action just varied.	None

Raises:

Type	Description
ValueError	If fitness_exam is not the held-out exam or matches a train or freshly mutated exam.

Source code in deap_er/private/operators/policy_fitness.py

def guard_policy_fitness_exam(
    fitness_exam: CaseExam,
    *,
    held_out: CaseExam,
    n_cases: int,
    train_exams: list[CaseExam] | None = None,
    mutated_exam: CaseExam | None = None,
) -> None:
    """Refuse policy fitness that targets a train or mutated exam.

    Policy individuals must be scored only on ``held_out``. Train-exam
    quality belongs in :func:`~deap_er.tools.policy_observe`, not in
    fitness assignment.

    Args:
        fitness_exam: Exam the caller would score for policy fitness.
        held_out: Caller-marked held-out exam.
        n_cases: Catalog length from the current elite pack.
        train_exams: Optional train exams that must not become the
            fitness target.
        mutated_exam: Optional exam the policy action just varied.

    Raises:
        ValueError: If ``fitness_exam`` is not the held-out exam or
            matches a train or freshly mutated exam.
    """
    held_cases = held_out.as_cases(n_cases)
    fitness_cases = fitness_exam.as_cases(n_cases)
    if train_exams is not None:
        for train in train_exams:
            if train is fitness_exam:
                raise ValueError("train-exam quality is an observation, not the policy objective")
            if train.as_cases(n_cases) == fitness_cases:
                raise ValueError("train-exam quality is an observation, not the policy objective")
    if fitness_cases != held_cases:
        raise ValueError("policy fitness must use the caller-marked held_out exam only")
    if mutated_exam is None:
        return
    if mutated_exam is fitness_exam or mutated_exam is held_out:
        raise ValueError("policy fitness must not reward the exam the policy just mutated")
    if mutated_exam.as_cases(n_cases) == fitness_cases:
        raise ValueError("policy fitness must not reward the exam the policy just mutated")

policy_held_out_fitness(elites, exams, *, held_out=None, fitness_exam=None, train_exams=None, mutated_exam=None, matrix=None, trust_matrix=False, solved=None, mode='unsolved')

Score policy individuals only on the held-out exam.

Train-exam difficulty is not part of the objective. Pair with :func:~deap_er.tools.policy_exam_scores and :func:~deap_er.records.policy_generalization_gap for observations and logbook chapters.

Parameters:

Name	Type	Description	Default
elites	list[Individual]	Evaluated tape individuals that supply the case pack.	required
exams	ExamLike	Train exams or a pool with a caller-marked held_out.	required
held_out	CaseExam | None	Optional held-out exam that overrides a pool marker.	None
fitness_exam	CaseExam | None	Optional exam the caller would assign fitness on. When set, it must match held_out.	None
train_exams	list[CaseExam] | None	Optional train exams checked by :func:guard_policy_fitness_exam.	None
mutated_exam	CaseExam | None	Optional exam varied by the last policy action.	None
matrix	ndarray | None	Optional (n_elites, n_cases) pack.	None
trust_matrix	bool	When True, matrix is accepted on shape alone.	False
solved	CaseSolved | None	Optional solve predicate. See :func:score_case_exams.	None
mode	DifficultyMode	unsolved or hamming difficulty.	'unsolved'

Returns:

Type	Description
float	Held-out exam difficulty as the policy fitness scalar.

Raises:

Type	Description
ValueError	If no held-out exam is marked or a guard refuses the requested fitness exam.

Source code in deap_er/private/operators/policy_fitness.py

def policy_held_out_fitness(
    elites: list[Individual],
    exams: ExamLike,
    *,
    held_out: CaseExam | None = None,
    fitness_exam: CaseExam | None = None,
    train_exams: list[CaseExam] | None = None,
    mutated_exam: CaseExam | None = None,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
    solved: CaseSolved | None = None,
    mode: DifficultyMode = "unsolved",
) -> float:
    """Score policy individuals only on the held-out exam.

    Train-exam difficulty is not part of the objective. Pair with
    :func:`~deap_er.tools.policy_exam_scores` and
    :func:`~deap_er.records.policy_generalization_gap` for
    observations and logbook chapters.

    Args:
        elites: Evaluated tape individuals that supply the case pack.
        exams: Train exams or a pool with a caller-marked ``held_out``.
        held_out: Optional held-out exam that overrides a pool marker.
        fitness_exam: Optional exam the caller would assign fitness on.
            When set, it must match ``held_out``.
        train_exams: Optional train exams checked by
            :func:`guard_policy_fitness_exam`.
        mutated_exam: Optional exam varied by the last policy action.
        matrix: Optional ``(n_elites, n_cases)`` pack.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape alone.
        solved: Optional solve predicate. See :func:`score_case_exams`.
        mode: ``unsolved`` or ``hamming`` difficulty.

    Returns:
        Held-out exam difficulty as the policy fitness scalar.

    Raises:
        ValueError: If no held-out exam is marked or a guard refuses
            the requested fitness exam.
    """
    resolved = resolve_policy_held_out(exams, held_out=held_out)
    n_cases, _ = elite_solve_matrix(elites, matrix, trust_matrix, solved)
    pool_train: list[CaseExam] | None = train_exams
    if pool_train is None and isinstance(exams, CaseExamPool):
        pool_train = exams.exams
    target = fitness_exam or resolved
    guard_policy_fitness_exam(
        target,
        held_out=resolved,
        n_cases=n_cases,
        train_exams=pool_train,
        mutated_exam=mutated_exam,
    )
    return float(
        score_case_exams(
            [resolved],
            elites,
            matrix=matrix,
            trust_matrix=trust_matrix,
            solved=solved,
            mode=mode,
        )[0]
    )

resolve_policy_held_out(exams, *, held_out=None)

Return the caller-marked held-out exam for policy fitness.

Train exams in a :class:~deap_er.records.CaseExamPool are never returned. Chronological meaning and marking stay on the caller.

Parameters:

Name	Type	Description	Default
exams	ExamLike	Train exams, a pool, or a single exam sequence.	required
held_out	CaseExam | None	Optional held-out exam that overrides a pool marker.	None

Returns:

Type	Description
CaseExam	The resolved held-out exam.

Raises:

Type	Description
ValueError	If no held-out exam is marked.

Source code in deap_er/private/operators/policy_fitness.py

def resolve_policy_held_out(
    exams: ExamLike,
    *,
    held_out: CaseExam | None = None,
) -> CaseExam:
    """Return the caller-marked held-out exam for policy fitness.

    Train exams in a :class:`~deap_er.records.CaseExamPool` are never
    returned. Chronological meaning and marking stay on the caller.

    Args:
        exams: Train exams, a pool, or a single exam sequence.
        held_out: Optional held-out exam that overrides a pool marker.

    Returns:
        The resolved held-out exam.

    Raises:
        ValueError: If no held-out exam is marked.
    """
    pool_held_out = held_out
    if isinstance(exams, CaseExamPool) and held_out is None:
        pool_held_out = exams.held_out
    if pool_held_out is None:
        raise ValueError("policy fitness requires a caller-marked held_out exam")
    return pool_held_out

sample_informed_cases(individuals, case_count, *, solved=None, matrix=None, trust_matrix=False)

Build a down-sample that prefers distinct fitness cases.

Two cases are synonymous when the same individuals solve them. Distance is the Hamming distance of those solve vectors. A random first case is kept, then farthest-first traversal adds the case farthest from the nearest already-chosen case. Ties, including a tail of zero distances, are broken at random.

A case is solved when fitness.values[case] is within 1e-12 of zero. Maximize-only scores and larger residuals need solved.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Population whose fitness vectors supply solve bits. Pass a fully scored parent sample if evaluation is sparse.	required
case_count	int	Number of case indices to return. Values above the number of cases are capped. case_count <= 0 returns an empty list.	required
solved	CaseSolved | None	Predicate (individual, case) -> bool. Optional. When not the default zero test, matrix is ignored.	None
matrix	ndarray | None	Optional (n_individuals, n_cases) case matrix. Used only with the default solved predicate.	None
trust_matrix	bool	When True, matrix is accepted on shape alone. Defaults to False.	False

Returns:

Type	Description
list[int]	Distinct fitness-case indices, in the order they were picked.

Raises:

Type	Description
ValueError	If individuals is empty or case_count is not an integer.
ValueError	If an individual has a missing or mismatched fitness length.

Source code in deap_er/private/operators/sample_informed_cases.py

def sample_informed_cases(
    individuals: list[Individual],
    case_count: int,
    *,
    solved: CaseSolved | None = None,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
) -> list[int]:
    """Build a down-sample that prefers distinct fitness cases.

    Two cases are synonymous when the same individuals solve them.
    Distance is the Hamming distance of those solve vectors. A random
    first case is kept, then farthest-first traversal adds the case
    farthest from the nearest already-chosen case. Ties, including a
    tail of zero distances, are broken at random.

    A case is solved when ``fitness.values[case]`` is within ``1e-12``
    of zero. Maximize-only scores and larger residuals need
    ``solved``.

    Args:
        individuals: Population whose fitness vectors supply solve bits.
            Pass a fully scored parent sample if evaluation is sparse.
        case_count: Number of case indices to return. Values above the
            number of cases are capped. ``case_count <= 0`` returns
            an empty list.
        solved: Predicate ``(individual, case) -> bool``. Optional.
            When not the default zero test, ``matrix`` is ignored.
        matrix: Optional ``(n_individuals, n_cases)`` case matrix.
            Used only with the default ``solved`` predicate.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape
            alone. Defaults to ``False``.

    Returns:
        Distinct fitness-case indices, in the order they were picked.

    Raises:
        ValueError: If ``individuals`` is empty or ``case_count`` is
            not an integer.
        ValueError: If an individual has a missing or mismatched
            fitness length.
    """
    if isinstance(case_count, bool) or not isinstance(case_count, Integral):
        raise ValueError("case_count must be an int")
    count = int(case_count)
    if not individuals:
        raise ValueError("individuals must be non-empty")
    if count <= 0:
        return []
    n_cases = len(individuals[0].fitness.values)
    if n_cases == 0:
        raise ValueError("every individual must have a valid fitness of the same length")
    size = min(count, n_cases)
    predicate = solved if solved is not None else _default_solved
    if matrix is not None and predicate is _default_solved:
        validate_case_matrix(matrix, individuals, trust=trust_matrix)
        solve = _solve_from_matrix(matrix)
    else:
        solve = _solve_matrix(individuals, n_cases, predicate)
    return _farthest_first_cases(solve, size)

sel_age_moea_2(individuals, sel_count, *, best_point=None, worst_point=None, extreme_points=None, nr_tol=0.001, nr_max_iter=100, _memory=None)

Select the next generation with AGE-MOEA-II.

Follows the environmental-selection loop of Panichella (GECCO 2022, Algorithm 2): non-dominated fronts are processed in order; when a front does not fit entirely, survivors are chosen by geometry-aware scores. The first front uses Newton-Raphson curvature and geodesic diversity; later fronts reuse the first front normalization and rank by inverse Minkowski distance to the ideal point.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required
best_point	ndarray | None	Ideal point of the previous generation. If omitted, it is taken from the current individuals.	None
worst_point	ndarray | None	Nadir point of the previous generation. If omitted, it is taken from the current individuals.	None
extreme_points	ndarray | None	Extreme points of the previous generation. If omitted, they are taken from the current individuals.	None
nr_tol	float	Newton-Raphson stopping tolerance for curvature.	0.001
nr_max_iter	int	Maximum Newton-Raphson iterations.	100
_memory	SelAGE2WithMemory | None	SelAGE2WithMemory instance that stores the updated normalization anchors. Not intended for manual use.	None

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_age_moea_2.py

def sel_age_moea_2(
    individuals: list[Individual],
    sel_count: int,
    *,
    best_point: ndarray | None = None,
    worst_point: ndarray | None = None,
    extreme_points: ndarray | None = None,
    nr_tol: float = 1e-3,
    nr_max_iter: int = 100,
    _memory: SelAGE2WithMemory | None = None,
) -> list[Individual]:
    """Select the next generation with AGE-MOEA-II.

    Follows the environmental-selection loop of Panichella (GECCO
    2022, Algorithm 2): non-dominated fronts are processed in order;
    when a front does not fit entirely, survivors are chosen by
    geometry-aware scores. The first front uses Newton-Raphson
    curvature and geodesic diversity; later fronts reuse the first
    front normalization and rank by inverse Minkowski distance to the
    ideal point.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.
        best_point: Ideal point of the previous generation. If
            omitted, it is taken from the current individuals.
        worst_point: Nadir point of the previous generation. If
            omitted, it is taken from the current individuals.
        extreme_points: Extreme points of the previous generation.
            If omitted, they are taken from the current individuals.
        nr_tol: Newton-Raphson stopping tolerance for curvature.
        nr_max_iter: Maximum Newton-Raphson iterations.
        _memory: ``SelAGE2WithMemory`` instance that stores the
            updated normalization anchors. Not intended for manual use.

    Returns:
        The selected individuals.
    """
    if not individuals or sel_count <= 0:
        return []
    if sel_count >= len(individuals):
        if isinstance(_memory, SelAGE2WithMemory):
            fitness = -numpy.array([ind.fitness.wvalues for ind in individuals], dtype=float)
            pareto_fronts = sort_non_dominated(individuals, len(individuals))
            index_map = {id(ind): idx for idx, ind in enumerate(individuals)}
            front_fit = pareto_front_fitness(fitness, pareto_fronts, index_map)
            best = merge_best(front_fit, best_point)
            worst = merge_worst(front_fit, worst_point)
            first_indices = [index_map[id(ind)] for ind in pareto_fronts[0]]
            first_front = fitness[first_indices]
            front_worst = numpy.max(first_front, axis=0)
            curvature, _ = estimate_geometry(
                first_front, best, worst, extreme_points, nr_tol, nr_max_iter, front_worst
            )
            _update_memory(_memory, best, worst, first_front, extreme_points, curvature)
        return list(individuals)

    pareto_fronts = sort_non_dominated(individuals, sel_count)
    fitness = -numpy.array([ind.fitness.wvalues for ind in individuals], dtype=float)
    index_map = {id(ind): idx for idx, ind in enumerate(individuals)}

    front_fit = pareto_front_fitness(fitness, pareto_fronts, index_map)
    best = merge_best(front_fit, best_point)
    worst = merge_worst(front_fit, worst_point)

    first_indices = [index_map[id(ind)] for ind in pareto_fronts[0]]
    first_front = fitness[first_indices]
    first_front_worst = numpy.max(first_front, axis=0)
    curvature, intercepts = estimate_geometry(
        first_front, best, worst, extreme_points, nr_tol, nr_max_iter, first_front_worst
    )

    chosen: list[Individual] = []
    for front_index, front in enumerate(pareto_fronts):
        if len(chosen) >= sel_count:
            break
        if len(chosen) + len(front) <= sel_count:
            chosen.extend(front)
            continue

        remaining = sel_count - len(chosen)
        front_indices = numpy.array([index_map[id(ind)] for ind in front], dtype=int)
        front_fitness = fitness[front_indices]
        local_worst = numpy.max(front_fitness, axis=0)
        scores = _front_survival_scores(
            front_fitness,
            front_index,
            best,
            worst,
            intercepts,
            curvature,
            extreme_points,
            local_worst,
        )
        order = numpy.argsort(scores)[::-1]
        chosen.extend(front[idx] for idx in order[:remaining])

    if isinstance(_memory, SelAGE2WithMemory):
        _update_memory(_memory, best, worst, first_front, extreme_points, curvature)

    return chosen[:sel_count]

sel_batch_epsilon_lexicase(individuals, sel_count, batch_size, epsilon=None, *, cases=None, matrix=None, trust_matrix=False, fit_weights=None, reduction=None)

Select individuals by epsilon-lexicase on batched case reductions.

Cases are shuffled and grouped into batches of at most batch_size. Each batch is reduced to one pseudo-case (mean squared error by default), then the usual epsilon-lexicase filter runs on the shorter matrix. A fresh shuffle and partition are drawn for every selected individual.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required
batch_size	int	Maximum cases per batch.	required
epsilon	float | None	Slack around the best batch value. If omitted, it is computed from the median absolute deviation of each batch column separately.	None
cases	Sequence[int] | None	Fitness-case indices to filter on. All cases are used when omitted.	None
matrix	ndarray | None	Optional (n_individuals, n_cases) case matrix. When omitted, values are read from fitness.values.	None
trust_matrix	bool	When True, matrix is accepted on shape alone. Defaults to False.	False
fit_weights	Sequence[float] | None	Optional per-column maximize/minimize signs. Required when matrix has more columns than fitness.values.	None
reduction	CaseReduction | None	Maps a (n_individuals, batch_width) block to one score per individual. Defaults to mean squared error.	None

Returns:

Type	Description
list[Individual]	The selected individuals.

Raises:

Type	Description
IndexError	If the population is empty or a case index is outside the fitness length.
ValueError	If batch_size is not a positive integer or matrix shape or values do not match fitness.

Source code in deap_er/private/operators/sel_batch_epsilon_lexicase.py

def sel_batch_epsilon_lexicase(
    individuals: list[Individual],
    sel_count: int,
    batch_size: int,
    epsilon: float | None = None,
    *,
    cases: Sequence[int] | None = None,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
    fit_weights: Sequence[float] | None = None,
    reduction: CaseReduction | None = None,
) -> list[Individual]:
    """Select individuals by epsilon-lexicase on batched case reductions.

    Cases are shuffled and grouped into batches of at most
    ``batch_size``. Each batch is reduced to one pseudo-case (mean
    squared error by default), then the usual epsilon-lexicase filter
    runs on the shorter matrix. A fresh shuffle and partition are
    drawn for every selected individual.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.
        batch_size: Maximum cases per batch.
        epsilon: Slack around the best batch value. If omitted, it is
            computed from the median absolute deviation of each batch
            column separately.
        cases: Fitness-case indices to filter on. All cases are used
            when omitted.
        matrix: Optional ``(n_individuals, n_cases)`` case matrix.
            When omitted, values are read from ``fitness.values``.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape
            alone. Defaults to ``False``.
        fit_weights: Optional per-column maximize/minimize signs.
            Required when ``matrix`` has more columns than
            ``fitness.values``.
        reduction: Maps a ``(n_individuals, batch_width)`` block to
            one score per individual. Defaults to mean squared error.

    Returns:
        The selected individuals.

    Raises:
        IndexError: If the population is empty or a case index is
            outside the fitness length.
        ValueError: If ``batch_size`` is not a positive integer or
            ``matrix`` shape or values do not match fitness.
    """
    if sel_count <= 0:
        return []
    require_population(individuals)
    partition_case_batches([], batch_size)
    packed = _resolve_matrix(individuals, matrix, trust_matrix=trust_matrix)
    n_cases = int(packed.shape[1])
    subset = case_subset(individuals, cases, n_cases=n_cases)
    weights = resolve_case_weights(individuals, packed, fit_weights)
    reduce = reduction if reduction is not None else reduce_case_mse
    mode = "epsilon_auto" if epsilon is None else "epsilon_fixed"
    selected: list[Individual] = []
    for _ in range(sel_count):
        batched, batch_weights = batch_case_matrix(packed, subset, weights, batch_size, reduce)
        selected.extend(
            lexicase_select_vectorized(
                individuals,
                1,
                batched,
                list(range(batched.shape[1])),
                batch_weights,
                mode=mode,
                epsilon=epsilon,
            )
        )
    return selected

assign_crowding_dist(individuals, *, use_weights=False)

Assign a crowding distance to each individual's fitness.

The distance is stored on the crowding_dist attribute of each individual's fitness. The individuals are modified in place.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals with Fitness attributes.	required
use_weights	bool	If True, crowd on wvalues instead of values. Defaults to False (Deb's NSGA-II).	False

Source code in deap_er/private/operators/sel_helpers.py

def assign_crowding_dist(individuals: list[Individual], *, use_weights: bool = False) -> None:
    """Assign a crowding distance to each individual's fitness.

    The distance is stored on the ``crowding_dist`` attribute of each
    individual's fitness. The individuals are modified in place.

    Args:
        individuals: Individuals with Fitness attributes.
        use_weights: If True, crowd on ``wvalues`` instead of
            ``values``. Defaults to False (Deb's NSGA-II).
    """
    if len(individuals) == 0:
        return

    distances = [0.0] * len(individuals)
    key = (lambda ind: ind.fitness.wvalues) if use_weights else (lambda ind: ind.fitness.values)
    crowd = [(key(ind), i) for i, ind in enumerate(individuals)]
    n_obj = len(individuals[0].fitness.values)

    for i in range(n_obj):
        crowd.sort(key=lambda element, obj_i=i: element[0][obj_i])
        distances[crowd[0][1]] = float("inf")
        distances[crowd[-1][1]] = float("inf")
        if crowd[-1][0][i] == crowd[0][0][i]:
            continue
        norm = n_obj * float(crowd[-1][0][i] - crowd[0][0][i])
        for prev, cur, next_ in zip(crowd[:-2], crowd[1:-1], crowd[2:], strict=False):
            distances[cur[1]] += (next_[0][i] - prev[0][i]) / norm

    for i, dist in enumerate(distances):
        individuals[i].fitness.crowding_dist = dist

uniform_reference_points(objectives, ref_ppo=4, scaling=None)

Generate reference points uniformly on the unit simplex.

Points lie on the hyperplane that intersects each axis at 1. scaling shrinks that layer toward the simplex center so several layers can be combined.

Parameters:

Name	Type	Description	Default
objectives	int	Number of objectives.	required
ref_ppo	int	Number of reference points per objective.	4
scaling	float | None	Optional scaling factor for combining layers.	None

Returns:

Type	Description
ndarray	Uniform reference points.

Source code in deap_er/private/operators/sel_helpers.py

def uniform_reference_points(
    objectives: int, ref_ppo: int = 4, scaling: float | None = None
) -> numpy.ndarray:
    """Generate reference points uniformly on the unit simplex.

    Points lie on the hyperplane that intersects each axis at 1.
    ``scaling`` shrinks that layer toward the simplex center so
    several layers can be combined.

    Args:
        objectives: Number of objectives.
        ref_ppo: Number of reference points per objective.
        scaling: Optional scaling factor for combining layers.

    Returns:
        Uniform reference points.
    """

    def _recursive(
        ref: numpy.ndarray, ovs: int, left: int, total: int, depth: int
    ) -> list[numpy.ndarray]:
        """Fill remaining objectives of one Das-Dennis reference point.

        Args:
            ref: Partial weight vector being filled.
            ovs: Number of objectives.
            left: Remaining integer budget to distribute.
            total: Total budget that sets the simplex spacing.
            depth: Objective index currently being assigned.

        Returns:
            Completed reference points generated from this prefix.
        """
        points = []
        if depth == ovs - 1:
            ref[depth] = left / total
            points.append(ref)
        else:
            for i in range(left + 1):
                ref[depth] = i / total
                rc = ref.copy()
                li = left - i
                d1 = depth + 1
                result = _recursive(rc, ovs, li, total, d1)
                points.extend(result)
        return points

    zeros = numpy.zeros(objectives)
    ref_points = _recursive(zeros, objectives, ref_ppo, ref_ppo, 0)
    ref_points = numpy.array(ref_points)

    if scaling is not None:
        ref_points *= scaling
        ref_points += (1 - scaling) / objectives

    return ref_points

sel_epsilon_lexicase(individuals, sel_count, epsilon=None, *, mode=None, cases=None, matrix=None, trust_matrix=False, fit_weights=None)

Select individuals by epsilon-lexicase filtering of fitness cases.

Each selected individual is the last remaining candidate after fitness cases are considered one at a time in random order. Candidates within epsilon of the best case value are kept. Pass cases to restrict the filter to a per-generation subset. Pass matrix when a packed case matrix is already available.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required
epsilon	float | None	Slack around the best case value. If omitted, it is computed from the median absolute deviation of the case values, separately for every case.	None
mode	LexicaseMode | None	Epsilon variant when epsilon is omitted: epsilon_auto or epsilon_static (population MAD and elite), epsilon_semi (population MAD, pool elite), or epsilon_dynamic (pool MAD and elite). Defaults to epsilon_auto.	None
cases	Sequence[int] | None	Fitness-case indices to filter on. All cases are used when omitted. Rebuild the subset each generation; do not freeze it on the toolbox.	None
matrix	ndarray | None	Optional (n_individuals, n_cases) case matrix. When omitted, values are read from fitness.values.	None
trust_matrix	bool	When True, matrix is accepted on shape alone. Defaults to False.	False
fit_weights	Sequence[float] | None	Optional per-column maximize/minimize signs. Required when matrix has more columns than fitness.values.	None

Returns:

Type	Description
list[Individual]	The selected individuals.

Raises:

Type	Description
IndexError	If the population is empty or a case index is outside the fitness length.
ValueError	If matrix shape or values do not match fitness.

Source code in deap_er/private/operators/sel_lexicase.py

def sel_epsilon_lexicase(
    individuals: list[Individual],
    sel_count: int,
    epsilon: float | None = None,
    *,
    mode: LexicaseMode | None = None,
    cases: Sequence[int] | None = None,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
    fit_weights: Sequence[float] | None = None,
) -> list[Individual]:
    """Select individuals by epsilon-lexicase filtering of fitness cases.

    Each selected individual is the last remaining candidate after
    fitness cases are considered one at a time in random order.
    Candidates within ``epsilon`` of the best case value are kept.
    Pass ``cases`` to restrict the filter to a per-generation subset.
    Pass ``matrix`` when a packed case matrix is already available.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.
        epsilon: Slack around the best case value. If omitted, it is
            computed from the median absolute deviation of the case
            values, separately for every case.
        mode: Epsilon variant when ``epsilon`` is omitted:
            ``epsilon_auto`` or ``epsilon_static`` (population MAD and
            elite), ``epsilon_semi`` (population MAD, pool elite), or
            ``epsilon_dynamic`` (pool MAD and elite). Defaults to
            ``epsilon_auto``.
        cases: Fitness-case indices to filter on. All cases are used
            when omitted. Rebuild the subset each generation; do not
            freeze it on the toolbox.
        matrix: Optional ``(n_individuals, n_cases)`` case matrix.
            When omitted, values are read from ``fitness.values``.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape
            alone. Defaults to ``False``.
        fit_weights: Optional per-column maximize/minimize signs.
            Required when ``matrix`` has more columns than
            ``fitness.values``.

    Returns:
        The selected individuals.

    Raises:
        IndexError: If the population is empty or a case index is
            outside the fitness length.
        ValueError: If ``matrix`` shape or values do not match fitness.
    """
    if sel_count <= 0:
        return []
    require_population(individuals)
    packed = _resolve_matrix(individuals, matrix, trust_matrix=trust_matrix)
    n_cases = int(packed.shape[1])
    subset = case_subset(individuals, cases, n_cases=n_cases)
    weights = resolve_case_weights(individuals, packed, fit_weights)
    if epsilon is not None:
        resolved = "epsilon_fixed"
    elif mode is None:
        resolved = "epsilon_auto"
    else:
        if mode not in _EPSILON_MODES:
            raise ValueError(f"mode must be one of {sorted(_EPSILON_MODES)}")
        resolved = mode
    return lexicase_select_vectorized(
        individuals,
        sel_count,
        packed,
        subset,
        weights,
        mode=resolved,
        epsilon=epsilon,
    )

sel_lexicase(individuals, sel_count, *, cases=None, matrix=None, trust_matrix=False, fit_weights=None)

Select individuals by lexicase filtering of fitness cases.

Each selected individual is the last remaining candidate after fitness cases are considered one at a time in random order. Pass cases to restrict the filter to a per-generation subset. Pass matrix when a packed case matrix is already available.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required
cases	Sequence[int] | None	Fitness-case indices to filter on. All cases are used when omitted. Rebuild the subset each generation; do not freeze it on the toolbox.	None
matrix	ndarray | None	Optional (n_individuals, n_cases) case matrix. When omitted, values are read from fitness.values.	None
trust_matrix	bool	When True, matrix is accepted on shape alone. Defaults to False.	False
fit_weights	Sequence[float] | None	Optional per-column maximize/minimize signs. Required when matrix has more columns than fitness.values.	None

Returns:

Type	Description
list[Individual]	The selected individuals.

Raises:

Type	Description
IndexError	If the population is empty or a case index is outside the fitness length.
ValueError	If matrix shape or values do not match fitness.

Source code in deap_er/private/operators/sel_lexicase.py

def sel_lexicase(
    individuals: list[Individual],
    sel_count: int,
    *,
    cases: Sequence[int] | None = None,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
    fit_weights: Sequence[float] | None = None,
) -> list[Individual]:
    """Select individuals by lexicase filtering of fitness cases.

    Each selected individual is the last remaining candidate after
    fitness cases are considered one at a time in random order.
    Pass ``cases`` to restrict the filter to a per-generation subset.
    Pass ``matrix`` when a packed case matrix is already available.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.
        cases: Fitness-case indices to filter on. All cases are used
            when omitted. Rebuild the subset each generation; do not
            freeze it on the toolbox.
        matrix: Optional ``(n_individuals, n_cases)`` case matrix.
            When omitted, values are read from ``fitness.values``.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape
            alone. Defaults to ``False``.
        fit_weights: Optional per-column maximize/minimize signs.
            Required when ``matrix`` has more columns than
            ``fitness.values``.

    Returns:
        The selected individuals.

    Raises:
        IndexError: If the population is empty or a case index is
            outside the fitness length.
        ValueError: If ``matrix`` shape or values do not match fitness.
    """
    if sel_count <= 0:
        return []
    require_population(individuals)
    packed = _resolve_matrix(individuals, matrix, trust_matrix=trust_matrix)
    n_cases = int(packed.shape[1])
    subset = case_subset(individuals, cases, n_cases=n_cases)
    weights = resolve_case_weights(individuals, packed, fit_weights)
    return lexicase_select_vectorized(
        individuals,
        sel_count,
        packed,
        subset,
        weights,
        mode="strict",
    )

fitness_case_matrix(individuals)

Pack fitness.values into a dense (n_individuals, n_cases) matrix.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Evaluated population. Zero-case fitness is packed as (n_individuals, 0).	required

Returns:

Type	Description
ndarray	Case values with one row per individual.

Raises:

Type	Description
ValueError	If individuals is empty or fitness lengths differ.

Source code in deap_er/private/operators/sel_lexicase_matrix.py

def fitness_case_matrix(individuals: list[Individual]) -> numpy.ndarray:
    """Pack ``fitness.values`` into a dense ``(n_individuals, n_cases)`` matrix.

    Args:
        individuals: Evaluated population. Zero-case fitness is packed
            as ``(n_individuals, 0)``.

    Returns:
        Case values with one row per individual.

    Raises:
        ValueError: If ``individuals`` is empty or fitness lengths differ.
    """
    if not individuals:
        raise ValueError("individuals must be non-empty")
    n_cases = len(individuals[0].fitness.values)
    matrix = numpy.empty((len(individuals), n_cases), dtype=numpy.float64)
    for row, individual in enumerate(individuals):
        values = individual.fitness.values
        if len(values) != n_cases:
            raise ValueError("every individual must have a valid fitness of the same length")
        if n_cases:
            matrix[row] = values
    return matrix

sel_moead(individuals, sel_count, weights, *, scalarization='tchebycheff', theta=5.0, ideal_point=None, _memory=None)

Select the next generation with MOEA/D decomposition.

Each weight vector defines a scalar subproblem. weights is typically uniform_reference_points. Subproblem winners prefer lower Pareto ranks; remaining slots are filled from complete lower fronts, then crowding distance on the last partial front of the leftover pool.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required
weights	ndarray	Decomposition weight vectors with shape (k, m).	required
scalarization	ScalarizationName | ScalarizationFn	"tchebycheff", "pbi", or a callable (fitness, weights, ideal_point) -> ndarray.	'tchebycheff'
theta	float	PBI penalty parameter.	5.0
ideal_point	ndarray | None	Ideal point from a previous generation. If omitted, it is taken from the current individuals.	None
_memory	SelMOEADWithMemory | None	SelMOEADWithMemory instance that stores the updated ideal point. Not intended for manual use.	None

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_moead.py

def sel_moead(
    individuals: list[Individual],
    sel_count: int,
    weights: ndarray,
    *,
    scalarization: ScalarizationName | ScalarizationFn = "tchebycheff",
    theta: float = 5.0,
    ideal_point: ndarray | None = None,
    _memory: SelMOEADWithMemory | None = None,
) -> list[Individual]:
    """Select the next generation with MOEA/D decomposition.

    Each weight vector defines a scalar subproblem. ``weights`` is
    typically ``uniform_reference_points``. Subproblem winners
    prefer lower Pareto ranks; remaining slots are filled from
    complete lower fronts, then crowding distance on the last
    partial front of the leftover pool.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.
        weights: Decomposition weight vectors with shape ``(k, m)``.
        scalarization: ``"tchebycheff"``, ``"pbi"``, or a callable
            ``(fitness, weights, ideal_point) -> ndarray``.
        theta: PBI penalty parameter.
        ideal_point: Ideal point from a previous generation. If
            omitted, it is taken from the current individuals.
        _memory: ``SelMOEADWithMemory`` instance that stores the
            updated ideal point. Not intended for manual use.

    Returns:
        The selected individuals.
    """
    if not individuals or sel_count <= 0:
        return []

    fitness = _minimize_fitness(individuals)

    prior = None
    if ideal_point is not None:
        prior = numpy.asarray(ideal_point, dtype=float).reshape(-1)
    elif isinstance(_memory, SelMOEADWithMemory):
        prior = numpy.asarray(_memory.ideal_point, dtype=float).reshape(-1)

    if sel_count >= len(individuals):
        if isinstance(_memory, SelMOEADWithMemory):
            z_star = _update_ideal_point(fitness, prior)
            _memory.ideal_point = numpy.asarray(z_star).reshape((1, -1))
        return list(individuals)

    scalar_fn = _resolve_scalarization(scalarization, theta)
    ranks = _pareto_ranks(individuals)

    z_star = _update_ideal_point(fitness, prior)
    chosen = _select_by_subproblems(
        individuals, fitness, weights, z_star, scalar_fn, sel_count, ranks
    )
    chosen = _fill_with_crowding(individuals, chosen, sel_count)

    if isinstance(_memory, SelMOEADWithMemory):
        _memory.ideal_point = numpy.asarray(z_star).reshape((1, -1))

    return chosen

moead_neighborhood(weights, n_neighbors)

Return sorted neighbor indices for each weight vector.

Each row lists the n_neighbors closest weight vectors by Euclidean distance, including the index itself.

Parameters:

Name	Type	Description	Default
weights	ndarray	Weight matrix with shape (k, m).	required
n_neighbors	int	Number of neighbors per weight vector.	required

Returns:

Type	Description
ndarray	Integer matrix with shape (k, n_neighbors).

Source code in deap_er/private/operators/sel_moead_helpers.py

def moead_neighborhood(weights: ndarray, n_neighbors: int) -> ndarray:
    """Return sorted neighbor indices for each weight vector.

    Each row lists the ``n_neighbors`` closest weight vectors by
    Euclidean distance, including the index itself.

    Args:
        weights: Weight matrix with shape ``(k, m)``.
        n_neighbors: Number of neighbors per weight vector.

    Returns:
        Integer matrix with shape ``(k, n_neighbors)``.
    """
    n_neighbors = min(n_neighbors, len(weights))
    dist = numpy.linalg.norm(weights[:, numpy.newaxis, :] - weights[numpy.newaxis, :, :], axis=2)
    return numpy.argsort(dist, axis=1, kind="quicksort")[:, :n_neighbors]

scalarization_pbi(fitness, weights, ideal_point, theta=5.0, eps=1e-16)

Return PBI scalarized values for each individual and weight.

Lower values are better. fitness and ideal_point are in minimize space.

Parameters:

Name	Type	Description	Default
fitness	ndarray	Objective matrix with shape (n_ind, m).	required
weights	ndarray	Weight matrix with shape (k, m).	required
ideal_point	ndarray	Ideal point with shape (m,).	required
theta	float	Penalty parameter balancing convergence and diversity.	5.0
eps	float	Small constant added to zero weights.	1e-16

Returns:

Type	Description
ndarray	Scalarized matrix with shape (n_ind, k).

Source code in deap_er/private/operators/sel_moead_helpers.py

def scalarization_pbi(
    fitness: ndarray,
    weights: ndarray,
    ideal_point: ndarray,
    theta: float = 5.0,
    eps: float = 1e-16,
) -> ndarray:
    """Return PBI scalarized values for each individual and weight.

    Lower values are better. ``fitness`` and ``ideal_point`` are in
    minimize space.

    Args:
        fitness: Objective matrix with shape ``(n_ind, m)``.
        weights: Weight matrix with shape ``(k, m)``.
        ideal_point: Ideal point with shape ``(m,)``.
        theta: Penalty parameter balancing convergence and diversity.
        eps: Small constant added to zero weights.

    Returns:
        Scalarized matrix with shape ``(n_ind, k)``.
    """
    diff = fitness - ideal_point
    w = numpy.where(numpy.abs(weights) < eps, eps, weights)
    norm_w = numpy.linalg.norm(w, axis=1)
    d1 = numpy.sum(diff[:, numpy.newaxis, :] * w[numpy.newaxis, :, :], axis=2) / norm_w
    unit = w / norm_w[:, numpy.newaxis]
    proj = ideal_point + d1[:, :, numpy.newaxis] * unit[numpy.newaxis, :, :]
    d2 = numpy.linalg.norm(fitness[:, numpy.newaxis, :] - proj, axis=2)
    return d1 + theta * d2

scalarization_tchebycheff(fitness, weights, ideal_point, eps=1e-16)

Return Tchebycheff scalarized values for each individual and weight.

Lower values are better. fitness and ideal_point are in minimize space.

Parameters:

Name	Type	Description	Default
fitness	ndarray	Objective matrix with shape (n_ind, m).	required
weights	ndarray	Weight matrix with shape (k, m).	required
ideal_point	ndarray	Ideal point with shape (m,).	required
eps	float	Small constant added to zero weights.	1e-16

Returns:

Type	Description
ndarray	Scalarized matrix with shape (n_ind, k).

Source code in deap_er/private/operators/sel_moead_helpers.py

def scalarization_tchebycheff(
    fitness: ndarray,
    weights: ndarray,
    ideal_point: ndarray,
    eps: float = 1e-16,
) -> ndarray:
    """Return Tchebycheff scalarized values for each individual and weight.

    Lower values are better. ``fitness`` and ``ideal_point`` are in
    minimize space.

    Args:
        fitness: Objective matrix with shape ``(n_ind, m)``.
        weights: Weight matrix with shape ``(k, m)``.
        ideal_point: Ideal point with shape ``(m,)``.
        eps: Small constant added to zero weights.

    Returns:
        Scalarized matrix with shape ``(n_ind, k)``.
    """
    diff = numpy.abs(fitness - ideal_point)
    w = numpy.where(numpy.abs(weights) < eps, eps, weights)
    return numpy.max(diff[:, numpy.newaxis, :] * w[numpy.newaxis, :, :], axis=2)

sel_novelty(individuals, sel_count, archive, descriptor_fn, *, k=15, metric='euclidean', valid=None)

Select individuals with the highest average distance to archive elites.

Novelty is the mean distance to the k nearest stored behavior descriptors. ind.fitness is not rewritten; only the ranking key changes. An empty archive falls back to uniform random selection.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to return. Non-positive values return an empty list.	required
archive	MapElitesArchive	MAP-Elites archive whose behavior coordinates define the reference set. Uses descriptors when present, else descriptor_fn on each stored elite.	required
descriptor_fn	Callable[[Individual], Sequence[float]]	Maps a pool member to its behavior coordinates.	required
k	int	Number of nearest archive neighbors averaged into the score.	15
metric	SemanticMetric	euclidean or cosine (same contract as semantic_distance).	'euclidean'
valid	ndarray | None	Optional per-dimension warmup mask passed to semantic_distance.	None

Returns:

Type	Description
list[Individual]	The most novel individuals. Ties keep the lowest pool index.

Raises:

Type	Description
ValueError	If k is less than 1.

Source code in deap_er/private/operators/sel_novelty.py

def sel_novelty(
    individuals: list[Individual],
    sel_count: int,
    archive: MapElitesArchive,
    descriptor_fn: Callable[[Individual], Sequence[float]],
    *,
    k: int = 15,
    metric: SemanticMetric = "euclidean",
    valid: numpy.ndarray | None = None,
) -> list[Individual]:
    """Select individuals with the highest average distance to archive elites.

    Novelty is the mean distance to the ``k`` nearest stored behavior
    descriptors. ``ind.fitness`` is not rewritten; only the ranking key
    changes. An empty archive falls back to uniform random selection.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to return. Non-positive values
            return an empty list.
        archive: MAP-Elites archive whose behavior coordinates define
            the reference set. Uses ``descriptors`` when present, else
            ``descriptor_fn`` on each stored elite.
        descriptor_fn: Maps a pool member to its behavior coordinates.
        k: Number of nearest archive neighbors averaged into the score.
        metric: ``euclidean`` or ``cosine`` (same contract as
            ``semantic_distance``).
        valid: Optional per-dimension warmup mask passed to
            ``semantic_distance``.

    Returns:
        The most novel individuals. Ties keep the lowest pool index.

    Raises:
        ValueError: If ``k`` is less than 1.
    """
    if sel_count <= 0:
        return []
    if not individuals:
        return []
    if k < 1:
        raise ValueError("k must be at least 1")

    archive_matrix = _archive_descriptor_matrix(archive, descriptor_fn)
    if archive_matrix.size == 0:
        return sel_random(individuals, sel_count)

    scores: list[tuple[float, int]] = []
    for index, individual in enumerate(individuals):
        query = numpy.asarray(descriptor_fn(individual), dtype=numpy.float64)
        novelty = _novelty_score(
            query,
            archive_matrix,
            k=k,
            metric=metric,
            valid=valid,
        )
        scores.append((novelty, index))

    order = sorted(scores, key=lambda item: (-item[0], item[1]))
    return [individuals[index] for _, index in order[:sel_count]]

sel_nsga_2(individuals, sel_count, *, feasible=None, violation=None)

Select the next generation with NSGA-II.

The pool is usually larger than sel_count. If the two sizes are equal, the population is sorted by Pareto front.

Optional feasible / violation switch ranking to Deb's constrained-domination rule: feasible individuals beat infeasible ones, two feasibles use ordinary Pareto and crowding, and two infeasibles prefer the smaller constraint violation. Omitted kwargs keep unconstrained NSGA-II. Fitness values are not rewritten.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required
feasible	Callable[[Individual], bool] | None	Predicate that reports whether an individual is feasible. Optional if violation is given.	None
violation	Callable[[Individual], float] | None	Function returning a scalar constraint violation. Optional if feasible is given. When this is the only constraint argument, <= 0 is feasible.	None

Returns:

Type	Description
list[Individual]	The selected individuals.

Raises:

Type	Description
TypeError	If a given constraint argument is not callable, or violation does not return a real scalar.
ValueError	If violation returns a non-finite number.

Source code in deap_er/private/operators/sel_nsga_2.py

def sel_nsga_2(
    individuals: list[Individual],
    sel_count: int,
    *,
    feasible: Callable[[Individual], bool] | None = None,
    violation: Callable[[Individual], float] | None = None,
) -> list[Individual]:
    """Select the next generation with NSGA-II.

    The pool is usually larger than ``sel_count``. If the two sizes
    are equal, the population is sorted by Pareto front.

    Optional ``feasible`` / ``violation`` switch ranking to Deb's
    constrained-domination rule: feasible individuals beat infeasible
    ones, two feasibles use ordinary Pareto and crowding, and two
    infeasibles prefer the smaller constraint violation. Omitted
    kwargs keep unconstrained NSGA-II. Fitness values are not
    rewritten.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.
        feasible: Predicate that reports whether an individual is
            feasible. Optional if ``violation`` is given.
        violation: Function returning a scalar constraint violation.
            Optional if ``feasible`` is given. When this is the only
            constraint argument, ``<= 0`` is feasible.

    Returns:
        The selected individuals.

    Raises:
        TypeError: If a given constraint argument is not callable, or
            ``violation`` does not return a real scalar.
        ValueError: If ``violation`` returns a non-finite number.
    """
    if not individuals or sel_count <= 0:
        return []
    if feasible is None and violation is None:
        pareto_fronts = sort_non_dominated(individuals, sel_count)
    else:
        pareto_fronts = sort_constraint_dominated(
            individuals, sel_count, feasible=feasible, violation=violation
        )

    for front in pareto_fronts:
        assign_crowding_dist(front)

    chosen = list(chain(*pareto_fronts[:-1]))
    sel_count = sel_count - len(chosen)
    if sel_count > 0:
        attr = attrgetter("fitness.crowding_dist")
        sorted_front = sorted(pareto_fronts[-1], key=attr, reverse=True)
        chosen.extend(sorted_front[:sel_count])

    return chosen

sel_nsga_3(individuals, sel_count, ref_points, best_point=None, worst_point=None, extreme_points=None, _memory=None)

Select the next generation with NSGA-III.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required
ref_points	ndarray	Reference points used for niche selection.	required
best_point	ndarray | None	Ideal point of the previous generation. If omitted, it is taken from the current individuals.	None
worst_point	ndarray | None	Nadir point of the previous generation. If omitted, it is taken from the current individuals.	None
extreme_points	ndarray | None	Extreme points of the previous generation. If omitted, they are taken from the current individuals.	None
_memory	SelNSGA3WithMemory | None	SelNSGA3WithMemory instance that stores the updated ideal, nadir, and extreme points. Not intended for manual use.	None

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_nsga_3.py

def sel_nsga_3(
    individuals: list[Individual],
    sel_count: int,
    ref_points: ndarray,
    best_point: ndarray | None = None,
    worst_point: ndarray | None = None,
    extreme_points: ndarray | None = None,
    _memory: SelNSGA3WithMemory | None = None,
) -> list[Individual]:
    """Select the next generation with NSGA-III.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.
        ref_points: Reference points used for niche selection.
        best_point: Ideal point of the previous generation. If
            omitted, it is taken from the current individuals.
        worst_point: Nadir point of the previous generation. If
            omitted, it is taken from the current individuals.
        extreme_points: Extreme points of the previous generation.
            If omitted, they are taken from the current individuals.
        _memory: ``SelNSGA3WithMemory`` instance that stores the
            updated ideal, nadir, and extreme points. Not intended
            for manual use.

    Returns:
        The selected individuals.
    """
    if not individuals or sel_count <= 0:
        return []
    pareto_fronts = sort_non_dominated(individuals, sel_count)

    fitness = numpy.array([ind.fitness.wvalues for f in pareto_fronts for ind in f])
    fitness *= -1

    if best_point is not None and worst_point is not None:
        best_point = numpy.min(numpy.concatenate((fitness, best_point), axis=0), axis=0)
        worst_point = numpy.max(numpy.concatenate((fitness, worst_point), axis=0), axis=0)
    else:
        best_point = numpy.min(fitness, axis=0)
        worst_point = numpy.max(fitness, axis=0)

    extreme_points = find_extreme_points(fitness, best_point, extreme_points)
    front_worst = numpy.max(fitness[: sum(len(f) for f in pareto_fronts), :], axis=0)
    intercepts = find_intercepts(extreme_points, best_point, worst_point, front_worst)
    niches, dist = associate_to_niche(fitness, ref_points, best_point, intercepts)

    niche_counts = numpy.zeros(len(ref_points), dtype=numpy.int64)
    index, counts = numpy.unique(niches[: -len(pareto_fronts[-1])], return_counts=True)
    niche_counts[index] = counts

    chosen = list(chain(*pareto_fronts[:-1]))
    selected = len(chosen)
    selected = select_from_niche(
        pareto_fronts[-1], sel_count - selected, niches[selected:], dist[selected:], niche_counts
    )
    chosen.extend(selected)

    if isinstance(_memory, SelNSGA3WithMemory):
        _memory.best_point = numpy.asarray(best_point).reshape((1, -1))
        _memory.worst_point = numpy.asarray(worst_point).reshape((1, -1))
        _memory.extreme_points = extreme_points

    return chosen

sel_sms_emoa(individuals, sel_count, ref_point=None)

Select the next generation with SMS-EMOA.

Non-dominated sorting ranks the pool. Complete fronts are kept. When the next front would exceed sel_count, individuals with the smallest hypervolume contribution on that critical front are removed one at a time until the quota is met. This is the Reduce operator from Beume, Naujoks, and Emmerich (2007).

Use on parents + offspring for generational search, or on parents + [child] for steady-state (mu + 1) selection.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Evaluated individuals to select from.	required
sel_count	int	Number of individuals to keep.	required
ref_point	list[float] | ndarray | None	Reference point in minimization space (the same convention as hypervolume and least_contrib: internally -wvalues). Optional. When omitted, the worst objective value in the input pool plus one is used and kept fixed for the whole truncation pass.	None

Returns:

Type	Description
list[Individual]	The selected individuals. Complete Pareto fronts appear first;
list[Individual]	survivors from the truncated critical front keep their relative
list[Individual]	list order after greedy removal (unlike sel_nsga_2, they
list[Individual]	are not crowding-sorted within the front).

Source code in deap_er/private/operators/sel_sms_emoa.py

def sel_sms_emoa(
    individuals: list[Individual],
    sel_count: int,
    ref_point: list[float] | numpy.ndarray | None = None,
) -> list[Individual]:
    """Select the next generation with SMS-EMOA.

    Non-dominated sorting ranks the pool. Complete fronts are kept.
    When the next front would exceed ``sel_count``, individuals with
    the smallest hypervolume contribution on that critical front are
    removed one at a time until the quota is met. This is the Reduce
    operator from Beume, Naujoks, and Emmerich (2007).

    Use on ``parents + offspring`` for generational search, or on
    ``parents + [child]`` for steady-state ``(mu + 1)`` selection.

    Args:
        individuals: Evaluated individuals to select from.
        sel_count: Number of individuals to keep.
        ref_point: Reference point in minimization space (the same
            convention as ``hypervolume`` and ``least_contrib``:
            internally ``-wvalues``). Optional. When omitted, the
            worst objective value in the input pool plus one is used
            and kept fixed for the whole truncation pass.

    Returns:
        The selected individuals. Complete Pareto fronts appear first;
        survivors from the truncated critical front keep their relative
        list order after greedy removal (unlike ``sel_nsga_2``, they
        are not crowding-sorted within the front).
    """
    if not individuals or sel_count <= 0:
        return []
    if sel_count >= len(individuals):
        return list(chain(*sort_non_dominated(individuals, len(individuals))))

    pareto_fronts = sort_non_dominated(individuals, sel_count)

    chosen = list(chain(*pareto_fronts[:-1]))
    need = sel_count - len(chosen)
    critical = pareto_fronts[-1]

    if need < len(critical):
        ref = (
            numpy.asarray(ref_point)
            if ref_point is not None
            else numpy.max(minimized_points(individuals), axis=0) + 1
        )
        while len(critical) > need:
            idx = least_contrib(critical, ref)
            critical.pop(idx)

    chosen.extend(critical)
    return chosen

sel_spea_2(individuals, sel_count)

Select the next generation with SPEA-II.

The pool is usually larger than sel_count. If the two sizes are equal, the population is sorted by Pareto front.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select. sel_count <= 0 returns an empty list.	required

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_spea_2.py

def sel_spea_2(individuals: list[Individual], sel_count: int) -> list[Individual]:
    """Select the next generation with SPEA-II.

    The pool is usually larger than ``sel_count``. If the two sizes
    are equal, the population is sorted by Pareto front.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select. ``sel_count <= 0``
            returns an empty list.

    Returns:
        The selected individuals.
    """
    if not individuals or sel_count <= 0:
        return []
    fits = raw_fitness(individuals)

    chosen = [i for i in range(len(individuals)) if fits[i] < 1]
    if len(chosen) < sel_count:
        chosen = fill_from_density(individuals, chosen, fits, sel_count)
    elif len(chosen) > sel_count:
        chosen = truncate_archive(individuals, chosen, sel_count)

    return [individuals[i] for i in chosen]

sel_team(individuals, sel_count, *, cases=None, matrix=None, trust_matrix=False)

Select a team by greedy maximum coverage of solved fitness cases.

A case is solved when its value is within 1e-12 of zero. Each added member is an unused pool individual that covers the most still-uncovered cases. Ties are broken at random. Member fitness is not rewritten; score the team on the caller.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Team size. Values above the pool length return the whole pool. sel_count <= 0 returns an empty list.	required
cases	Sequence[int] | None	Fitness-case indices to cover. All distinct cases are used when omitted. Duplicate indices are covered once.	None
matrix	ndarray | None	Optional (n_individuals, n_cases) case matrix. When omitted, values are read from fitness.values.	None
trust_matrix	bool	When True, matrix is accepted on shape alone. Defaults to False.	False

Returns:

Type	Description
list[Individual]	Distinct pool members in greedy-add order. sel_count == 1
list[Individual]	is the individual that solves the most selected cases.

Raises:

Type	Description
IndexError	If the population is empty or a case index is outside the fitness length.
ValueError	If matrix shape or values do not match fitness.

Source code in deap_er/private/operators/sel_team.py

def sel_team(
    individuals: list[Individual],
    sel_count: int,
    *,
    cases: Sequence[int] | None = None,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
) -> list[Individual]:
    """Select a team by greedy maximum coverage of solved fitness cases.

    A case is solved when its value is within ``1e-12`` of zero. Each
    added member is an unused pool individual that covers the most
    still-uncovered cases. Ties are broken at random. Member
    ``fitness`` is not rewritten; score the team on the caller.

    Args:
        individuals: Individuals to select from.
        sel_count: Team size. Values above the pool length return the
            whole pool. ``sel_count <= 0`` returns an empty list.
        cases: Fitness-case indices to cover. All distinct cases are
            used when omitted. Duplicate indices are covered once.
        matrix: Optional ``(n_individuals, n_cases)`` case matrix.
            When omitted, values are read from ``fitness.values``.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape
            alone. Defaults to ``False``.

    Returns:
        Distinct pool members in greedy-add order. ``sel_count == 1``
        is the individual that solves the most selected cases.

    Raises:
        IndexError: If the population is empty or a case index is
            outside the fitness length.
        ValueError: If ``matrix`` shape or values do not match fitness.
    """
    if sel_count <= 0:
        return []
    require_population(individuals)
    packed = _resolve_matrix(individuals, matrix, trust_matrix=trust_matrix)
    solve = _solve_columns(packed, case_subset(individuals, cases))
    taken = numpy.zeros(len(individuals), dtype=bool)
    uncovered = numpy.ones(solve.shape[1], dtype=bool)
    team: list[Individual] = []
    for _ in range(min(sel_count, len(individuals))):
        idx = _next_member(taken, solve, uncovered)
        team.append(individuals[idx])
        taken[idx] = True
        if uncovered.size:
            uncovered &= ~solve[idx]
    return team

sel_team_archive(archive, sel_count, *, cases=None, matrix=None, trust_matrix=False)

Select a team from occupied MAP-Elites archive cells.

The pool is list(archive) — live elites from filled cells, not random_elites copies. Delegates to :func:sel_team. Member fitness is not rewritten; score the team on the caller.

Parameters:

Name	Type	Description	Default
archive	MapElitesArchive	MAP-Elites archive with add and iteration support.	required
sel_count	int	Team size. Same semantics as :func:sel_team.	required
cases	Sequence[int] | None	Fitness-case indices to cover. Passed through to :func:sel_team.	None
matrix	ndarray | None	Optional (n_elites, n_cases) case matrix. When omitted, values are read from elite fitness.values. Row order must match list(archive) when trust_matrix=True.	None
trust_matrix	bool	When True, matrix is accepted on shape alone. Defaults to False.	False

Returns:

Type	Description
list[Individual]	Distinct archive elites in greedy-add order.

Raises:

Type	Description
IndexError	If the archive is empty or a case index is outside the fitness length.
ValueError	If matrix shape or values do not match fitness.

Source code in deap_er/private/operators/sel_team.py

def sel_team_archive(
    archive: MapElitesArchive,
    sel_count: int,
    *,
    cases: Sequence[int] | None = None,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
) -> list[Individual]:
    """Select a team from occupied MAP-Elites archive cells.

    The pool is ``list(archive)`` — live elites from filled cells, not
    ``random_elites`` copies. Delegates to :func:`sel_team`. Member
    ``fitness`` is not rewritten; score the team on the caller.

    Args:
        archive: MAP-Elites archive with ``add`` and iteration support.
        sel_count: Team size. Same semantics as :func:`sel_team`.
        cases: Fitness-case indices to cover. Passed through to
            :func:`sel_team`.
        matrix: Optional ``(n_elites, n_cases)`` case matrix. When
            omitted, values are read from elite ``fitness.values``.
            Row order must match ``list(archive)`` when
            ``trust_matrix=True``.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape
            alone. Defaults to ``False``.

    Returns:
        Distinct archive elites in greedy-add order.

    Raises:
        IndexError: If the archive is empty or a case index is outside
            the fitness length.
        ValueError: If ``matrix`` shape or values do not match fitness.
    """
    return sel_team(
        list(cast(Iterable[Any], archive)),
        sel_count,
        cases=cases,
        matrix=matrix,
        trust_matrix=trust_matrix,
    )

sel_double_tournament(individuals, rounds, fitness_size, parsimony_size, fitness_first, fit_attr='fitness')

Select with a fitness tournament and a size tournament.

The size contest can be used in genetic programming as a bloat control technique.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
rounds	int	Number of tournament rounds.	required
fitness_size	int	Number of individuals in each fitness tournament.	required
parsimony_size	float	Number of individuals in each size tournament. Must be in [1, 2].	required
fitness_first	bool	If True, run the fitness tournament first.	required
fit_attr	str	Attribute used as the fitness selection criterion.	'fitness'

Returns:

Type	Description
list[Individual]	The selected individuals. An empty pool or rounds <= 0
list[Individual]	returns an empty list.

Raises:

Type	Description
ValueError	If parsimony_size is outside [1, 2].

Source code in deap_er/private/operators/sel_tournament.py

def sel_double_tournament(
    individuals: list[Individual],
    rounds: int,
    fitness_size: int,
    parsimony_size: float,
    fitness_first: bool,
    fit_attr: str = "fitness",
) -> list[Individual]:
    """Select with a fitness tournament and a size tournament.

    The size contest can be used in genetic programming as a bloat
    control technique.

    Args:
        individuals: Individuals to select from.
        rounds: Number of tournament rounds.
        fitness_size: Number of individuals in each fitness tournament.
        parsimony_size: Number of individuals in each size tournament.
            Must be in ``[1, 2]``.
        fitness_first: If True, run the fitness tournament first.
        fit_attr: Attribute used as the fitness selection criterion.

    Returns:
        The selected individuals. An empty pool or ``rounds <= 0``
        returns an empty list.

    Raises:
        ValueError: If ``parsimony_size`` is outside ``[1, 2]``.
    """
    if not (1 <= parsimony_size <= 2):
        raise ValueError("Parsimony tournament size has to be in the range of [1, 2].")
    if rounds <= 0 or not individuals:
        return []

    def _size_tourney(
        pool: list[Individual], sel_count: int, select: Callable[..., Any]
    ) -> list[Individual]:
        """Run the parsimony (size) half of the double tournament.

        Args:
            pool: Individuals to select from.
            sel_count: Number of size contests to run.
            select: Selection callable used to pick the two contestants.

        Returns:
            Winners of the size contests.
        """
        chosen = []
        for _i in range(sel_count):
            prob = parsimony_size / 2.0
            ind1, ind2 = select(pool, sel_count=2)
            if len(ind1) > len(ind2):
                ind1, ind2 = ind2, ind1
            elif len(ind1) == len(ind2):
                prob = 0.5
            chosen.append(ind1 if rng.random() < prob else ind2)
        return chosen

    def _fit_tourney(
        pool: list[Individual], sel_count: int, select: Callable[..., Any]
    ) -> list[Individual]:
        """Run the fitness half of the double tournament.

        Args:
            pool: Individuals to select from.
            sel_count: Number of fitness contests to run.
            select: Selection callable used to pick the contestants.

        Returns:
            Winners of the fitness contests.
        """
        chosen = []
        for _i in range(sel_count):
            aspirants = select(pool, sel_count=fitness_size)
            chosen.append(max(aspirants, key=attrgetter(fit_attr)))
        return chosen

    if fitness_first:
        t_fit = partial(_fit_tourney, select=sel_random)
        return _size_tourney(individuals, rounds, t_fit)
    t_size = partial(_size_tourney, select=sel_random)
    return _fit_tourney(individuals, rounds, t_size)

sel_tournament(individuals, rounds, contestants, fit_attr='fitness')

Select the best of contestants random individuals, rounds times.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
rounds	int	Number of tournament rounds.	required
contestants	int	Number of individuals in each round.	required
fit_attr	str	Attribute used as the selection criterion.	'fitness'

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_tournament.py

def sel_tournament(
    individuals: list[Individual], rounds: int, contestants: int, fit_attr: str = "fitness"
) -> list[Individual]:
    """Select the best of ``contestants`` random individuals, ``rounds`` times.

    Args:
        individuals: Individuals to select from.
        rounds: Number of tournament rounds.
        contestants: Number of individuals in each round.
        fit_attr: Attribute used as the selection criterion.

    Returns:
        The selected individuals.
    """
    if rounds <= 0:
        return []
    n = len(individuals)
    if n == 0:
        raise IndexError("Cannot choose from an empty sequence")
    key = attrgetter(fit_attr)
    if contestants < 1:
        raise ValueError("contestants must be at least 1")
    idxs = rng.integers(0, n, size=rounds * contestants)
    total = rounds * contestants
    if contestants == 1:
        return [individuals[idx] for idx in idxs]
    if contestants == 2:
        return _sel_pair(individuals, idxs, key, total)
    if contestants == 3:
        return _sel_triple(individuals, idxs, key, total)
    chosen: list[Individual] = []
    for i in range(0, total, contestants):
        winner = individuals[idxs[i]]
        best = key(winner)
        for offset in range(1, contestants):
            candidate = individuals[idxs[i + offset]]
            score = key(candidate)
            if score > best:
                winner, best = candidate, score
        chosen.append(winner)
    return chosen

sel_tournament_cases(individuals, rounds, contestants, *, cases=None, case_count=None, matrix=None, trust_matrix=False, reduction=None)

Tournament selection on a down-sampled case aggregate.

Each individual is scored by reducing a case subset to one scalar (column mean by default), then standard tournament selection runs on those scores. The aggregate ranking uses the maximize/minimize sign of the first case index in the resolved subset; mixed per-case weights are not applied column-wise. When cases=[] or case_count <= 0, every individual ties on score zero and tournament rounds draw uniformly from the pool. Informed down-sampling stays on :func:~deap_er.tools.sample_informed_cases; pass its result as cases=.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
rounds	int	Number of tournament rounds.	required
contestants	int	Number of individuals in each round.	required
cases	Sequence[int] | None	Fitness-case indices to score. All cases are used when omitted and case_count is not set. An empty sequence resolves to uniform tournament draws.	None
case_count	int | None	When cases is omitted, draw this many distinct case indices at random. Ignored when cases is given. case_count <= 0 resolves to an empty subset (uniform tournament draws).	None
matrix	ndarray | None	Optional (n_individuals, n_cases) case matrix. When omitted, values are read from fitness.values.	None
trust_matrix	bool	When True, matrix is accepted on shape alone. Defaults to False.	False
reduction	CaseReduction | None	Maps a (n_individuals, n_cases) block to one score per individual. Defaults to the column mean.	None

Returns:

Type	Description
list[Individual]	The selected individuals.

Raises:

Type	Description
IndexError	If the population is empty.
ValueError	If contestants is not positive or matrix shape or values do not match fitness.

Source code in deap_er/private/operators/sel_tournament_cases.py

def sel_tournament_cases(
    individuals: list[Individual],
    rounds: int,
    contestants: int,
    *,
    cases: Sequence[int] | None = None,
    case_count: int | None = None,
    matrix: numpy.ndarray | None = None,
    trust_matrix: bool = False,
    reduction: CaseReduction | None = None,
) -> list[Individual]:
    """Tournament selection on a down-sampled case aggregate.

    Each individual is scored by reducing a case subset to one scalar
    (column mean by default), then standard tournament selection runs
    on those scores. The aggregate ranking uses the maximize/minimize
    sign of the **first** case index in the resolved subset; mixed
    per-case weights are not applied column-wise. When ``cases=[]`` or
    ``case_count <= 0``, every individual ties on score zero and
    tournament rounds draw uniformly from the pool. Informed
    down-sampling stays on
    :func:`~deap_er.tools.sample_informed_cases`; pass its result as
    ``cases=``.

    Args:
        individuals: Individuals to select from.
        rounds: Number of tournament rounds.
        contestants: Number of individuals in each round.
        cases: Fitness-case indices to score. All cases are used when
            omitted and ``case_count`` is not set. An empty sequence
            resolves to uniform tournament draws.
        case_count: When ``cases`` is omitted, draw this many distinct
            case indices at random. Ignored when ``cases`` is given.
            ``case_count <= 0`` resolves to an empty subset (uniform
            tournament draws).
        matrix: Optional ``(n_individuals, n_cases)`` case matrix.
            When omitted, values are read from ``fitness.values``.
        trust_matrix: When ``True``, ``matrix`` is accepted on shape
            alone. Defaults to ``False``.
        reduction: Maps a ``(n_individuals, n_cases)`` block to one
            score per individual. Defaults to the column mean.

    Returns:
        The selected individuals.

    Raises:
        IndexError: If the population is empty.
        ValueError: If ``contestants`` is not positive or ``matrix``
            shape or values do not match fitness.
    """
    if rounds <= 0:
        return []
    n = len(individuals)
    if n == 0:
        raise IndexError("Cannot choose from an empty sequence")
    if contestants < 1:
        raise ValueError("contestants must be at least 1")
    packed = _resolve_matrix(individuals, matrix, trust_matrix=trust_matrix)
    case_indices = _resolve_case_indices(individuals, cases, case_count)
    reduce = reduction if reduction is not None else reduce_case_mean
    scores = _tournament_scores(packed, case_indices, individuals[0].fitness.weights, reduce)
    idxs = rng.integers(0, n, size=rounds * contestants)
    if contestants == 1:
        return [individuals[idx] for idx in idxs]
    chosen: list[Individual] = []
    for i in range(0, rounds * contestants, contestants):
        winner = individuals[idxs[i]]
        best = scores[idxs[i]]
        for offset in range(1, contestants):
            candidate_idx = idxs[i + offset]
            score = scores[candidate_idx]
            if score > best:
                winner = individuals[candidate_idx]
                best = score
        chosen.append(winner)
    return chosen

sel_tournament_dcd(individuals, sel_count)

Select by pairwise dominance, breaking ties with crowding distance.

When sel_count is a multiple of four the original paired shuffle is used. Other counts run pairwise contests until enough winners are collected. Each individual must already have a crowding_dist attribute, which assign_crowding_dist can set.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required

Returns:

Type	Description
list[Individual]	The selected individuals.

Raises:

Type	Description
ValueError	If sel_count is larger than the pool.

Source code in deap_er/private/operators/sel_tournament_dcd.py

def sel_tournament_dcd(individuals: list[Individual], sel_count: int) -> list[Individual]:
    """Select by pairwise dominance, breaking ties with crowding distance.

    When ``sel_count`` is a multiple of four the original paired
    shuffle is used. Other counts run pairwise contests until enough
    winners are collected. Each individual must already have a
    ``crowding_dist`` attribute, which ``assign_crowding_dist`` can
    set.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.

    Returns:
        The selected individuals.

    Raises:
        ValueError: If ``sel_count`` is larger than the pool.
    """
    if sel_count <= 0:
        return []
    if sel_count > len(individuals):
        raise ValueError(
            "sel_tournament_dcd: count must be less than or equal to individuals length."
        )

    if sel_count % 4 == 0:
        individuals_1 = rng.sample(individuals, len(individuals))
        individuals_2 = rng.sample(individuals, len(individuals))

        chosen = []
        for i in range(0, sel_count, 4):
            chosen.append(_dcd_tourney(individuals_1[i], individuals_1[i + 1]))
            chosen.append(_dcd_tourney(individuals_1[i + 2], individuals_1[i + 3]))
            chosen.append(_dcd_tourney(individuals_2[i], individuals_2[i + 1]))
            chosen.append(_dcd_tourney(individuals_2[i + 2], individuals_2[i + 3]))
        return chosen

    if sel_count == 1 and len(individuals) == 1:
        return [individuals[0]]

    chosen = []
    pool = list(individuals)
    while len(chosen) < sel_count:
        rng.shuffle(pool)
        for i in range(0, len(pool) - 1, 2):
            chosen.append(_dcd_tourney(pool[i], pool[i + 1]))
            if len(chosen) >= sel_count:
                break
    return chosen

sel_best(individuals, sel_count, fit_attr='fitness')

Select the sel_count best individuals.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select. sel_count <= 0 returns an empty list.	required
fit_attr	str	Attribute used as the selection criterion.	'fitness'

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_various.py

def sel_best(
    individuals: list[Individual], sel_count: int, fit_attr: str = "fitness"
) -> list[Individual]:
    """Select the ``sel_count`` best individuals.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select. ``sel_count <= 0``
            returns an empty list.
        fit_attr: Attribute used as the selection criterion.

    Returns:
        The selected individuals.
    """
    if sel_count <= 0:
        return []
    key = attrgetter(fit_attr)
    return sorted(individuals, key=key, reverse=True)[:sel_count]

sel_random(individuals, sel_count)

Select sel_count individuals uniformly at random.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required

Returns:

Type	Description
list[Individual]	The selected individuals. An empty pool or sel_count <= 0
list[Individual]	returns an empty list.

Source code in deap_er/private/operators/sel_various.py

def sel_random(individuals: list[Individual], sel_count: int) -> list[Individual]:
    """Select ``sel_count`` individuals uniformly at random.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.

    Returns:
        The selected individuals. An empty pool or ``sel_count <= 0``
        returns an empty list.
    """
    if sel_count <= 0 or not individuals:
        return []
    return [rng.choice(individuals) for _ in range(sel_count)]

sel_roulette(individuals, sel_count, fit_attr='fitness')

Select sel_count individuals by roulette-wheel sampling.

Each draw uses only the first weighted objective of fit_attr. The returned list holds references to the input individuals.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required
fit_attr	str	Attribute used as the selection criterion.	'fitness'

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_various.py

def sel_roulette(
    individuals: list[Individual], sel_count: int, fit_attr: str = "fitness"
) -> list[Individual]:
    """Select ``sel_count`` individuals by roulette-wheel sampling.

    Each draw uses only the first weighted objective of ``fit_attr``.
    The returned list holds references to the input individuals.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.
        fit_attr: Attribute used as the selection criterion.

    Returns:
        The selected individuals.
    """
    if sel_count <= 0 or not individuals:
        return []
    wheel = _wheel_prefix(individuals, fit_attr)
    if wheel is None:
        return [rng.choice(individuals) for _ in range(sel_count)]
    sorted_, prefix, total = wheel
    chosen = []
    for _ in range(sel_count):
        idx = bisect.bisect_right(prefix, rng.random() * total)
        if idx >= len(sorted_):
            idx = len(sorted_) - 1
        chosen.append(sorted_[idx])
    return chosen

sel_stochastic_universal_sampling(individuals, sel_count, fit_attr='fitness')

Select sel_count individuals by stochastic universal sampling.

A single random offset samples the wheel at evenly spaced intervals. Only the first weighted objective of fit_attr is used. The returned list holds references to the input individuals.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select.	required
fit_attr	str	Attribute used as the selection criterion.	'fitness'

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_various.py

def sel_stochastic_universal_sampling(
    individuals: list[Individual], sel_count: int, fit_attr: str = "fitness"
) -> list[Individual]:
    """Select ``sel_count`` individuals by stochastic universal sampling.

    A single random offset samples the wheel at evenly spaced
    intervals. Only the first weighted objective of ``fit_attr`` is
    used. The returned list holds references to the input individuals.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select.
        fit_attr: Attribute used as the selection criterion.

    Returns:
        The selected individuals.
    """
    if sel_count <= 0 or not individuals:
        return []
    wheel = _wheel_prefix(individuals, fit_attr)
    if wheel is None:
        return [rng.choice(individuals) for _ in range(sel_count)]
    sorted_, prefix, total = wheel
    distance = total / float(sel_count)
    start = rng.uniform(0, distance)
    chosen = []
    for i in range(sel_count):
        idx = bisect.bisect_left(prefix, start + i * distance)
        if idx >= len(sorted_):
            idx = len(sorted_) - 1
        chosen.append(sorted_[idx])
    return chosen

sel_worst(individuals, sel_count, fit_attr='fitness')

Select the sel_count worst individuals.

Parameters:

Name	Type	Description	Default
individuals	list[Individual]	Individuals to select from.	required
sel_count	int	Number of individuals to select. sel_count <= 0 returns an empty list.	required
fit_attr	str	Attribute used as the selection criterion.	'fitness'

Returns:

Type	Description
list[Individual]	The selected individuals.

Source code in deap_er/private/operators/sel_various.py

def sel_worst(
    individuals: list[Individual], sel_count: int, fit_attr: str = "fitness"
) -> list[Individual]:
    """Select the ``sel_count`` worst individuals.

    Args:
        individuals: Individuals to select from.
        sel_count: Number of individuals to select. ``sel_count <= 0``
            returns an empty list.
        fit_attr: Attribute used as the selection criterion.

    Returns:
        The selected individuals.
    """
    if sel_count <= 0:
        return []
    key = attrgetter(fit_attr)
    return sorted(individuals, key=key)[:sel_count]

estimate_tune_ephemerals_evals(strategy, n_gen)

Estimate how many evaluations a memetic tune would spend.

Parameters:

Name	Type	Description	Default
strategy	Any	Strategy or StrategySeparable whose offsprings or lamb sets the batch size.	required
n_gen	int	Inner CMA generations.	required

Returns:

Type	Description
int	n_gen times the per-generation offspring count.

Source code in deap_er/private/programming/memetic_defaults.py

def estimate_tune_ephemerals_evals(strategy: Any, n_gen: int) -> int:
    """Estimate how many evaluations a memetic tune would spend.

    Args:
        strategy: ``Strategy`` or ``StrategySeparable`` whose
            ``offsprings`` or ``lamb`` sets the batch size.
        n_gen: Inner CMA generations.

    Returns:
        ``n_gen`` times the per-generation offspring count.
    """
    offsprings = getattr(strategy, "offsprings", None)
    if offsprings is None:
        offsprings = getattr(strategy, "lamb", 1)
    return int(n_gen) * int(offsprings)

structural_meta_case_columns(individuals, *, prim_set=None, predicted=None, columns=None)

Return cheap structural meta-case columns for a population.

Shape (len(individuals), len(columns)). Tree metrics need no predicted. non_finite_fraction needs one row per individual in predicted ((n_ind, n_rows) or a sequence of 1-D series). When predicted is missing, that column is filled with numpy.nan.

Parameters:

Name	Type	Description	Default
individuals	Sequence[Any]	Population whose genomes are PrimitiveTree or list-like node sequences.	required
prim_set	PrimitiveSetTyped | None	Primitive set for unique_opcodes and promote_hits. Required when either column is requested.	None
predicted	ndarray | Sequence[Sequence[float]] | None	Optional per-individual output series.	None
columns	Sequence[str] | None	Subset of :data:STRUCTURAL_META_CASES. All columns are used when omitted.	None

Returns:

Type	Description
ndarray	Structural scalars with one row per individual.

Raises:

Type	Description
ValueError	If individuals is empty, a column name is unknown, prim_set is missing for opcode or promote columns, or predicted has the wrong shape.

Source code in deap_er/private/various/structural_meta_case.py

def structural_meta_case_columns(
    individuals: Sequence[Any],
    *,
    prim_set: PrimitiveSetTyped | None = None,
    predicted: numpy.ndarray | Sequence[Sequence[float]] | None = None,
    columns: Sequence[str] | None = None,
) -> numpy.ndarray:
    """Return cheap structural meta-case columns for a population.

    Shape ``(len(individuals), len(columns))``. Tree metrics need no
    ``predicted``. ``non_finite_fraction`` needs one row per
    individual in ``predicted`` (``(n_ind, n_rows)`` or a sequence of
    1-D series). When ``predicted`` is missing, that column is filled
    with ``numpy.nan``.

    Args:
        individuals: Population whose genomes are ``PrimitiveTree`` or
            list-like node sequences.
        prim_set: Primitive set for ``unique_opcodes`` and
            ``promote_hits``. Required when either column is requested.
        predicted: Optional per-individual output series.
        columns: Subset of :data:`STRUCTURAL_META_CASES`. All columns
            are used when omitted.

    Returns:
        Structural scalars with one row per individual.

    Raises:
        ValueError: If ``individuals`` is empty, a column name is
            unknown, ``prim_set`` is missing for opcode or promote
            columns, or ``predicted`` has the wrong shape.
    """
    if not individuals:
        raise ValueError("individuals must be non-empty")
    names = _resolve_columns(columns)
    if _NEED_PRIM_SET.intersection(names) and prim_set is None:
        raise ValueError("prim_set is required for unique_opcodes and promote_hits")

    predicted_rows = (
        _predicted_rows(individuals, predicted) if "non_finite_fraction" in names else None
    )
    promoted = frozenset(promoted_names(prim_set)) if prim_set is not None else frozenset()
    tape_cache: dict[str, Any] = {}
    matrix = numpy.empty((len(individuals), len(names)), dtype=numpy.float64)

    for row, individual in enumerate(individuals):
        tree = _as_tree(individual)
        for col, name in enumerate(names):
            matrix[row, col] = _column_scalar(
                name,
                tree=tree,
                individual=individual,
                typed_set=prim_set,
                promoted=promoted,
                tape_cache=tape_cache,
                predicted_rows=predicted_rows,
                row=row,
            )
    return matrix

structural_meta_case_weights(columns=None)

Return default lexicase signs for structural meta-case columns.

Bloat metrics default to minimize (-1). non_finite_fraction defaults to maximize (+1) so lexicase prefers programs that are not finite on every bar.

Parameters:

Name	Type	Description	Default
columns	Sequence[str] | None	Subset of :data:STRUCTURAL_META_CASES. All columns are used when omitted.	None

Returns:

Type	Description
tuple[float, ...]	One maximize/minimize sign per column.

Raises:

Type	Description
ValueError	If a column name is unknown.

Source code in deap_er/private/various/structural_meta_case.py

def structural_meta_case_weights(
    columns: Sequence[str] | None = None,
) -> tuple[float, ...]:
    """Return default lexicase signs for structural meta-case columns.

    Bloat metrics default to minimize (-1). ``non_finite_fraction``
    defaults to maximize (+1) so lexicase prefers programs that are
    not finite on every bar.

    Args:
        columns: Subset of :data:`STRUCTURAL_META_CASES`. All
            columns are used when omitted.

    Returns:
        One maximize/minimize sign per column.

    Raises:
        ValueError: If a column name is unknown.
    """
    names = _resolve_columns(columns)
    return tuple(1.0 if name == "non_finite_fraction" else -1.0 for name in names)