Genetic Programming

deap_er.gp

USER_DISPATCH_SIGNATURE = '(op: int64, sp: int64, stack: float64[:, ::1], columns: float64[:, ::1], constants: float64[::1], scratch: float64[::1]) -> int64' module-attribute

Signature every consumer dispatch kernel must have.

The interpreter keeps one stack of column-length rows. sp is the number of occupied rows, so the top operand is stack[sp - 1] and a kernel of arity n reads stack[sp - n] through stack[sp - 1], writes its result into stack[sp - n], and returns sp - n + 1::

stack[sp - 1]  <- top operand      (n_rows values)
stack[sp - 2]  <- second operand
...
stack[sp - n]  <- first operand, and where the result goes
stack[sp]      <- free scratch row

constants is the tape constant pool, scratch is one spare row of n_rows values that a kernel may clobber freely, and stack[sp] is always allocated and free to use as well. Builtin rolling primitives fold their window length into the instruction, but a consumer primitive receives every argument on the stack, so a window arrives as a broadcast row and can be read from stack[sp - 1][0].

BUILTIN_OPCODES = {'vadd': Opcode.ADD, 'vsub': Opcode.SUB, 'vmul': Opcode.MUL, 'vdiv': Opcode.DIV, 'vneg': Opcode.NEG, 'vabs': Opcode.ABS, 'vlog': Opcode.LOG, 'vsqrt': Opcode.SQRT, 'vsin': Opcode.SIN, 'vcos': Opcode.COS, 'vgt': Opcode.GT, 'vlt': Opcode.LT, 'vge': Opcode.GE, 'vle': Opcode.LE, 'veq': Opcode.EQ, 'vand': Opcode.AND, 'vor': Opcode.OR, 'vnot': Opcode.NOT, 'vwhere': Opcode.WHERE, 'delay': Opcode.DELAY, 'diff': Opcode.DIFF, 'rolling_sum': Opcode.ROLL_SUM, 'rolling_mean': Opcode.ROLL_MEAN, 'rolling_std': Opcode.ROLL_STD, 'rolling_min': Opcode.ROLL_MIN, 'rolling_max': Opcode.ROLL_MAX, 'ema': Opcode.EMA, 'rolling_corr': Opcode.ROLL_CORR, 'rolling_cov': Opcode.ROLL_COV, 'rolling_beta': Opcode.ROLL_BETA, 'ts_rank': Opcode.TS_RANK, 'ts_argmax': Opcode.TS_ARGMAX, 'ts_argmin': Opcode.TS_ARGMIN} module-attribute

Opcode of every primitive registered by the builtin kits.

USER_BASE = 1000 module-attribute

First opcode value reserved for consumer-supplied kernels.

Array

Type tag for a one-dimensional float64 column.

Used as a strongly typed genetic programming type. Values that flow through nodes tagged with it are NumPy arrays, never instances of this class.

Mask

Type tag for a one-dimensional boolean condition.

Comparison and logic primitives return this tag. It keeps conditions out of arithmetic positions and gives where a well-defined first argument.

Window

Type tag for a positive integer window length.

Rolling and delay primitives take one argument of this tag. No primitive returns it, so a window is always a leaf and can be lowered to an immediate operand.

Opcode

Bases: IntEnum

Instruction set of the tree stack machine.

Every value below USER_BASE is owned by this library. Consumers bind their own kernels to values at or above USER_BASE.

Tape(opcodes, operands, constants, columns, depth, fill) dataclass

Flat, picklable form of a compiled expression tree.

Instructions are in postfix order. operands[i] is the column index of a COL_LOAD, the constant pool index of a CONST, the window length of a rolling instruction, and -1 otherwise.

Attributes:

Name	Type	Description
opcodes	ndarray	Instruction stream as int32.
operands	ndarray	Immediate operand of each instruction as int32.
constants	ndarray	Constant pool as float64.
columns	int	Number of columns the tape expects.
depth	int	Peak stack depth reached while running the tape.
fill	float	Fill used by the protected instructions.

Ephemeral()

Bases: Terminal, ABC

Terminal whose value is sampled when the instance is created.

Abstract base class. Subclasses must define a static method named func.

Sample func and initialize as a non-symbolic terminal.

Source code in deap_er/private/programming/primitives/primitive_nodes.py

def __init__(self) -> None:
    """Sample ``func`` and initialize as a non-symbolic terminal."""
    Terminal.__init__(self, self.func(), symbolic=False, ret_type=self.ret)

func() abstractmethod staticmethod

Produce a new ephemeral value.

Subclasses must override this static method.

Raises:

Type	Description
NotImplementedError	If the subclass does not define func.

Source code in deap_er/private/programming/primitives/primitive_nodes.py

@staticmethod
@abc.abstractmethod
def func() -> Any:
    """Produce a new ephemeral value.

    Subclasses must override this static method.

    Raises:
        NotImplementedError: If the subclass does not define ``func``.
    """
    raise NotImplementedError

Primitive(name, args, ret_type, weight=1.0)

Function node in a GP expression.

Formats as a Python call when given formatted child expressions.

Parameters:

Name	Type	Description	Default
name	str	Name of the primitive.	required
args	list[type]	Argument types of the primitive.	required
ret_type	type	Return type of the primitive.	required
weight	float	Relative sampling weight. Must be greater than 0.	1.0

Store the primitive name, argument types, and return type.

Source code in deap_er/private/programming/primitives/primitive_nodes.py

def __init__(self, name: str, args: list[type], ret_type: type, weight: float = 1.0) -> None:
    """Store the primitive name, argument types, and return type."""
    self.name = name
    self.arity = len(args)
    self.args = args
    self.ret = ret_type
    self.weight = weight
    placeholders = ", ".join(map("{{{0}}}".format, list(range(self.arity))))
    self.seq = f"{self.name}({placeholders})"

format(*args)

Format this primitive as a Python call.

Parameters:

Name	Type	Description	Default
*args	str	Formatted child expressions, one per argument.	()

Returns:

Type	Description
str	The primitive applied to args as source text.

Source code in deap_er/private/programming/primitives/primitive_nodes.py

def format(self, *args: str) -> str:
    """Format this primitive as a Python call.

    Args:
        *args: Formatted child expressions, one per argument.

    Returns:
        The primitive applied to ``args`` as source text.
    """
    return self.seq.format(*args)

__eq__(other)

Return whether other is a primitive with the same slots.

Source code in deap_er/private/programming/primitives/primitive_nodes.py

@override
def __eq__(self, other: object) -> bool:
    """Return whether ``other`` is a primitive with the same slots."""
    if type(self) is type(other):
        return all(getattr(self, slot) == getattr(other, slot) for slot in self.__slots__)
    else:
        return NotImplemented

Terminal(terminal, symbolic, ret_type, call_zero=False)

Leaf node in a GP expression.

A terminal is a value or a zero-arity function.

Parameters:

Name	Type	Description	Default
terminal	Any	Value or zero-arity function stored in the leaf.	required
symbolic	bool	If True, format the value with str; otherwise with repr.	required
ret_type	type	Return type of the terminal.	required

Store terminal as a named leaf of ret_type.

Parameters:

Name	Type	Description	Default
terminal	Any	Value or name stored in the leaf.	required
symbolic	bool	If True, format the value with str.	required
ret_type	type	Return type of the terminal.	required
call_zero	bool	If True, format as a zero-arity call name().	False

Source code in deap_er/private/programming/primitives/primitive_nodes.py

def __init__(
    self, terminal: Any, symbolic: bool, ret_type: type, call_zero: bool = False
) -> None:
    """Store ``terminal`` as a named leaf of ``ret_type``.

    Args:
        terminal: Value or name stored in the leaf.
        symbolic: If True, format the value with ``str``.
        ret_type: Return type of the terminal.
        call_zero: If True, format as a zero-arity call ``name()``.
    """
    self.ret = ret_type
    self.value = terminal
    self.name = str(terminal)
    self.conv_fct = str if symbolic else repr
    self.call_zero = call_zero

arity property

Number of arguments this terminal takes.

Always 0.

format()

Return the string form of the terminal value.

Source code in deap_er/private/programming/primitives/primitive_nodes.py

def format(self) -> str:
    """Return the string form of the terminal value."""
    text = self.conv_fct(self.value)
    if self.call_zero:
        return f"{text}()"
    return text

__eq__(other)

Return whether other is a terminal with the same slots.

Source code in deap_er/private/programming/primitives/primitive_nodes.py

@override
def __eq__(self, other: object) -> bool:
    """Return whether ``other`` is a terminal with the same slots."""
    if type(self) is type(other):
        return all(getattr(self, slot) == getattr(other, slot) for slot in self.__slots__)
    else:
        return NotImplemented

PrimitiveSet(name, arity, prefix='ARG')

Bases: PrimitiveSetTyped

Untyped primitive set.

Subclass of PrimitiveSetTyped that treats every type as object.

Parameters:

Name	Type	Description	Default
name	str	Name of the primitive set.	required
arity	int	Number of input arguments.	required
prefix	str	Prefix used to name input arguments.	'ARG'

Create an untyped set with arity inputs.

Source code in deap_er/private/programming/primitives/primitive_set.py

def __init__(self, name: str, arity: int, prefix: str = "ARG") -> None:
    """Create an untyped set with ``arity`` inputs."""
    args: list[type] = [object] * arity
    super().__init__(name, args, object, prefix)

add_primitive(primitive, arity, name=None, *_, weight=1.0, **__)

Add an untyped primitive of the given arity.

Parameters:

Name	Type	Description	Default
primitive	Callable[..., Any]	Callable to register.	required
arity	int	Number of arguments. Must be at least 1.	required
name	str | None	Optional name. Defaults to primitive.__name__.	None
weight	float	Relative sampling weight. Must be greater than 0.	1.0

Raises:

Type	Description
ValueError	If arity is less than 1, if name is already registered, or if weight is not greater than 0.

Source code in deap_er/private/programming/primitives/primitive_set.py

@override
def add_primitive(  # type: ignore[override]
    self,
    primitive: Callable[..., Any],
    arity: int,
    name: str | None = None,
    *_: Any,
    weight: float = 1.0,
    **__: Any,
) -> None:
    """Add an untyped primitive of the given arity.

    Args:
        primitive: Callable to register.
        arity: Number of arguments. Must be at least 1.
        name: Optional name. Defaults to ``primitive.__name__``.
        weight: Relative sampling weight. Must be greater than 0.

    Raises:
        ValueError: If ``arity`` is less than 1, if ``name`` is
            already registered, or if ``weight`` is not greater
            than 0.
    """
    if arity < 1:
        raise ValueError("arity should be >= 1")
    args: list[type] = [object] * arity
    super().add_primitive(primitive, args, object, name, weight=weight)

add_terminal(terminal, name=None, *_, call_zero=False, **__)

Add an untyped terminal to the set.

Parameters:

Name	Type	Description	Default
terminal	Any	Value or callable to register as a terminal.	required
name	str | None	Optional name. Defaults to terminal.__name__ when terminal is callable.	None
call_zero	bool	If True, format a callable terminal as name() so the default eval compile path calls it.	False

Source code in deap_er/private/programming/primitives/primitive_set.py

@override
def add_terminal(  # type: ignore[override]
    self,
    terminal: Any,
    name: str | None = None,
    *_: Any,
    call_zero: bool = False,
    **__: Any,
) -> None:
    """Add an untyped terminal to the set.

    Args:
        terminal: Value or callable to register as a terminal.
        name: Optional name. Defaults to ``terminal.__name__``
            when ``terminal`` is callable.
        call_zero: If True, format a callable terminal as
            ``name()`` so the default eval compile path calls it.
    """
    super().add_terminal(terminal, object, name, call_zero=call_zero)

add_ephemeral_constant(name, ephemeral, *_, **__)

Add an untyped ephemeral constant to the set.

Parameters:

Name	Type	Description	Default
name	str	Name of this ephemeral type.	required
ephemeral	Callable[..., Any]	Zero-arity callable that produces a value.	required

Source code in deap_er/private/programming/primitives/primitive_set.py

@override
def add_ephemeral_constant(  # type: ignore[override]
    self, name: str, ephemeral: Callable[..., Any], *_: Any, **__: Any
) -> None:
    """Add an untyped ephemeral constant to the set.

    Args:
        name: Name of this ephemeral type.
        ephemeral: Zero-arity callable that produces a value.
    """
    super().add_ephemeral_constant(name, ephemeral, object)

PrimitiveSetTyped(name, in_types, ret_type, prefix='ARG')

Primitive set for strongly typed genetic programming.

Parameters:

Name	Type	Description	Default
name	str	Name of the primitive set.	required
in_types	list[type]	Input types, one per argument.	required
ret_type	type	Return type of expressions built from this set.	required
prefix	str	Prefix used to name input arguments.	'ARG'

Create an empty typed set and register one terminal per input.

Source code in deap_er/private/programming/primitives/primitive_set_typed.py

def __init__(
    self, name: str, in_types: list[type], ret_type: type, prefix: str = "ARG"
) -> None:
    """Create an empty typed set and register one terminal per input."""
    self.name = name
    self.ins = in_types
    self.ret = ret_type

    self.terminals = defaultdict(list)
    self.primitives = defaultdict(list)
    self.context = {"__builtins__": None}
    self.arguments = []
    self.mapping = {}
    self.terms_count = 0
    self.prims_count = 0

    for i, type_ in enumerate(in_types):
        arg_str = f"{prefix}{i}"
        self.arguments.append(arg_str)
        term = Terminal(arg_str, True, type_)
        self._add_prim(term)
        self.terms_count += 1

terminal_ratio property

Ratio of terminals to all primitives in the set.

add_primitive(primitive, in_types, ret_type, name=None, weight=1.0)

Add a primitive to the set.

Parameters:

Name	Type	Description	Default
primitive	Callable[..., Any]	Callable to register.	required
in_types	list[type]	Argument types of the primitive.	required
ret_type	type	Type returned by the primitive.	required
name	str | None	Optional name. Defaults to primitive.__name__.	None
weight	float	Relative sampling weight. Must be greater than 0.	1.0

Raises:

Type	Description
ValueError	If name is already registered, or if weight is not greater than 0.

Source code in deap_er/private/programming/primitives/primitive_set_typed.py

def add_primitive(
    self,
    primitive: Callable[..., Any],
    in_types: list[type],
    ret_type: type,
    name: str | None = None,
    weight: float = 1.0,
) -> None:
    """Add a primitive to the set.

    Args:
        primitive: Callable to register.
        in_types: Argument types of the primitive.
        ret_type: Type returned by the primitive.
        name: Optional name. Defaults to ``primitive.__name__``.
        weight: Relative sampling weight. Must be greater than 0.

    Raises:
        ValueError: If ``name`` is already registered, or if
            ``weight`` is not greater than 0.
    """
    if weight <= 0:
        raise ValueError("Primitive weight must be greater than 0.")
    if name is None:
        raw_name = getattr(primitive, "__name__", None)
        if not isinstance(raw_name, str):
            raise TypeError("Primitive must have a name or a '__name__' attribute.")
        name = raw_name

    if name in self.arguments:
        raise ValueError(
            f"Primitive name '{name}' is also an argument of the primitive set. "
            f"A compiled lambda parameter would shadow the primitive."
        )
    if name in self.context:
        raise ValueError(
            f"Primitives are required to have a unique name. "
            f"Consider using the argument 'name' to "
            f"rename your second '{name}' primitive."
        )
    prim = Primitive(name, in_types, ret_type, weight=weight)

    self._add_prim(prim)
    self.context[prim.name] = primitive
    self.prims_count += 1

add_terminal(terminal, ret_type, name=None, *, call_zero=False)

Add a terminal to the set.

Parameters:

Name	Type	Description	Default
terminal	Any	Value or callable to register as a terminal.	required
ret_type	type	Type returned by the terminal.	required
name	str | None	Optional name. Defaults to terminal.__name__ when terminal is callable.	None
call_zero	bool	If True, format a callable terminal as name() so eval calls it. False keeps action terminals as names.	False

Raises:

Type	Description
ValueError	If name is already registered, or matches an argument of the primitive set.

Source code in deap_er/private/programming/primitives/primitive_set_typed.py

def add_terminal(
    self, terminal: Any, ret_type: type, name: str | None = None, *, call_zero: bool = False
) -> None:
    """Add a terminal to the set.

    Args:
        terminal: Value or callable to register as a terminal.
        ret_type: Type returned by the terminal.
        name: Optional name. Defaults to ``terminal.__name__``
            when ``terminal`` is callable.
        call_zero: If True, format a callable terminal as ``name()``
            so eval calls it. False keeps action terminals as names.

    Raises:
        ValueError: If ``name`` is already registered, or matches
            an argument of the primitive set.
    """
    symbolic = False
    if name is None and callable(terminal):
        raw_name = getattr(terminal, "__name__", None)
        name = raw_name if isinstance(raw_name, str) else None

    if name is not None and name in self.arguments:
        raise ValueError(
            f"Terminal name '{name}' is also an argument of the primitive set. "
            f"A compiled lambda parameter would shadow the terminal."
        )
    if name is not None and name in self.context:
        raise ValueError(
            f"Terminals are required to have a unique name. "
            f"Consider using the argument 'name' to "
            f"rename your second '{name}' terminal."
        )

    invoke_zero = False
    if name is not None:
        invoke_zero = call_zero and callable(terminal)
        self.context[name] = terminal
        terminal = name
        symbolic = True
    elif terminal in (True, False):
        self.context[str(terminal)] = terminal

    prim = Terminal(terminal, symbolic, ret_type, call_zero=invoke_zero)
    self._add_prim(prim)
    self.terms_count += 1

add_ephemeral_constant(name, ephemeral, ret_type)

Add an ephemeral constant to the set.

An ephemeral is a zero-arity function that returns a random value. Each tree samples its own immutable value.

Parameters:

Name	Type	Description	Default
name	str	Name of this ephemeral type.	required
ephemeral	Callable[..., Any]	Zero-arity callable that produces a value.	required
ret_type	type	Type returned by the ephemeral.	required

Raises:

Type	Description
TypeError	If name is already used by a different ephemeral or by another class in this module.

Source code in deap_er/private/programming/primitives/primitive_set_typed.py

def add_ephemeral_constant(
    self, name: str, ephemeral: Callable[..., Any], ret_type: type
) -> None:
    """Add an ephemeral constant to the set.

    An ephemeral is a zero-arity function that returns a random
    value. Each tree samples its own immutable value.

    Args:
        name: Name of this ephemeral type.
        ephemeral: Zero-arity callable that produces a value.
        ret_type: Type returned by the ephemeral.

    Raises:
        TypeError: If ``name`` is already used by a different
            ephemeral or by another class in this module.
    """
    module_gp = globals()
    if name not in module_gp:
        attrs = {"func": staticmethod(ephemeral), "ret": ret_type}
        class_ = type(name, (Ephemeral,), attrs)
        module_gp[name] = class_
    else:
        class_ = module_gp[name]
        if issubclass(class_, Ephemeral):
            if class_.func is not ephemeral:
                raise TypeError(
                    "Ephemera with different functions should be "
                    "named differently even between psets."
                )
            elif class_.ret is not ret_type:
                raise TypeError(
                    "Ephemera with the same name and function should "
                    "have the same type even between psets."
                )
        else:
            raise TypeError(
                "Ephemera should be named differently than classes defined in the gp module."
            )

    self._add_prim(class_)
    self.terms_count += 1

add_adf(prim_set)

Add an Automatically Defined Function (ADF) to the set.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Primitive set that defines the ADF name, inputs, and return type.	required

Source code in deap_er/private/programming/primitives/primitive_set_typed.py

def add_adf(self, prim_set: PrimitiveSetTyped) -> None:
    """Add an Automatically Defined Function (ADF) to the set.

    Args:
        prim_set: Primitive set that defines the ADF name, inputs,
            and return type.
    """
    prim = Primitive(prim_set.name, prim_set.ins, prim_set.ret)
    self._add_prim(prim)
    self.prims_count += 1

rename_arguments(**kwargs)

Rename input arguments using the given mapping.

Parameters:

Name	Type	Description	Default
**kwargs	str	Map of current argument names to new names. Names that are not current arguments are ignored.	{}

Raises:

Type	Description
ValueError	If a new name is already an argument, or is already used by a primitive or terminal.

Source code in deap_er/private/programming/primitives/primitive_set_typed.py

def rename_arguments(self, **kwargs: str) -> None:
    """Rename input arguments using the given mapping.

    Args:
        **kwargs: Map of current argument names to new names. Names
            that are not current arguments are ignored.

    Raises:
        ValueError: If a new name is already an argument, or is
            already used by a primitive or terminal.
    """
    apply_argument_renames(self, kwargs)

PrimitiveTree(content)

Bases: list[Any]

Prefix-ordered tree of primitives and terminals.

A list subclass used by genetic programming operators. Every node must expose an arity attribute.

Parameters:

Name	Type	Description	Default
content	Iterable[Any]	Primitives and terminals that form the tree.	required

Initialize the tree from content.

Source code in deap_er/private/programming/primitives/primitive_tree.py

def __init__(self, content: Iterable[Any]) -> None:
    """Initialize the tree from ``content``."""
    super().__init__(content)

height property

Height of the tree, which is the depth of the deepest node.

root property

Root node of the tree (the first element).

__deepcopy__(memo)

Return a deep copy of this tree.

Parameters:

Name	Type	Description	Default
memo	dict[int, Any]	Memo mapping used by copy.deepcopy.	required

Returns:

Type	Description
PrimitiveTree	A new tree with copied contents and attributes.

Source code in deap_er/private/programming/primitives/primitive_tree.py

def __deepcopy__(self, memo: dict[int, Any]) -> PrimitiveTree:
    """Return a deep copy of this tree.

    Args:
        memo: Memo mapping used by ``copy.deepcopy``.

    Returns:
        A new tree with copied contents and attributes.
    """
    new = self.__class__(self)
    new.__dict__.update(copy.deepcopy(self.__dict__, memo))
    return new

__setitem__(key, val)

Replace a node or subtree, preserving tree arity.

Parameters:

Name	Type	Description	Default
key	Any	Index or slice of the node or subtree to replace.	required
val	Any	Replacement node or sequence of nodes.	required

Raises:

Type	Description
IndexError	If a slice starts past the end of the tree.
ValueError	If the replacement would change the tree arity.

Source code in deap_er/private/programming/primitives/primitive_tree.py

@override
def __setitem__(self, key: Any, val: Any) -> None:  # type: ignore[override]
    """Replace a node or subtree, preserving tree arity.

    Args:
        key: Index or slice of the node or subtree to replace.
        val: Replacement node or sequence of nodes.

    Raises:
        IndexError: If a slice starts past the end of the tree.
        ValueError: If the replacement would change the tree arity.
    """
    if isinstance(key, slice):
        start = 0 if key.start is None else key.start
        if start >= len(self):
            raise IndexError(
                "Trying to set a slice larger than the size "
                "of the PrimitiveTree is not allowed."
            )
        total = val[0].arity
        for node in val[1:]:
            total += node.arity - 1
        if total != 0:
            raise ValueError(
                "Insertion of a subtree with an arity smaller "
                "than the PrimitiveTree is not allowed."
            )
    elif val.arity != self[key].arity:
        raise ValueError(
            "PrimitiveTree node replacement with a node of a different arity is not allowed."
        )
    list.__setitem__(self, key, val)

__str__()

Return the tree as a Python expression string.

Source code in deap_er/private/programming/primitives/primitive_tree.py

@override
def __str__(self) -> str:
    """Return the tree as a Python expression string."""
    string = ""
    stack: list[Any] = []
    for node in self:
        stack.append((node, []))
        while len(stack[-1][1]) == stack[-1][0].arity:
            prim, args = stack.pop()
            string = prim.format(*args)
            if len(stack) == 0:
                break
            stack[-1][1].append(string)
    return str(string)

from_string(string, prim_set) classmethod

Build a tree from a Python expression string.

prim_set must contain every primitive that appears in string.

Parameters:

Name	Type	Description	Default
string	str	Python expression to deserialize.	required
prim_set	PrimitiveSetTyped	Primitive set used to resolve names.	required

Returns:

Type	Description
PrimitiveTree	A tree populated with the deserialized primitives.

Raises:

Type	Description
TypeError	If a token is not a registered primitive and is not a Python literal, if a primitive or terminal type does not match the expected type, if a token arrives after the tree is complete, or if the stream still owes argument types. A Window slot accepts an int literal.

Source code in deap_er/private/programming/primitives/primitive_tree.py

@classmethod
def from_string(cls, string: str, prim_set: PrimitiveSetTyped) -> PrimitiveTree:
    """Build a tree from a Python expression string.

    ``prim_set`` must contain every primitive that appears in
    ``string``.

    Args:
        string: Python expression to deserialize.
        prim_set: Primitive set used to resolve names.

    Returns:
        A tree populated with the deserialized primitives.

    Raises:
        TypeError: If a token is not a registered primitive
            and is not a Python literal, if a primitive or
            terminal type does not match the expected type, if
            a token arrives after the tree is complete, or if
            the stream still owes argument types. A ``Window``
            slot accepts an ``int`` literal.
    """
    tokens = re.split("[ \t\n\r\f\v(),]", string)
    expr = []
    ret_types = deque()
    for token in tokens:
        if token == "":
            continue
        if expr and not ret_types:
            raise TypeError(f"Unexpected extra token after a complete expression: {token}.")
        ret_type = ret_types.popleft() if ret_types else None
        if token in prim_set.mapping:
            primitive = primitive_from_token(token, prim_set, ret_type)
            expr.append(primitive)
            if isinstance(primitive, Primitive):
                ret_types.extendleft(reversed(primitive.args))
            continue
        expr.append(terminal_from_token(token, ret_type))
    if ret_types:
        raise TypeError("Expression is incomplete; missing arguments.")
    return cls(expr)

search_subtree(begin)

Return the slice of the subtree rooted at begin.

Parameters:

Name	Type	Description	Default
begin	int	Index of the subtree root.	required

Returns:

Type	Description
slice	Slice covering that subtree.

Source code in deap_er/private/programming/primitives/primitive_tree.py

def search_subtree(self, begin: int) -> slice:
    """Return the slice of the subtree rooted at ``begin``.

    Args:
        begin: Index of the subtree root.

    Returns:
        Slice covering that subtree.
    """
    end = begin + 1
    total = self[begin].arity
    while total > 0:
        total += self[end].arity - 1
        end += 1
    return slice(begin, end)

SlimTree(head, deltas=None)

SLIM geometric-semantic genotype: a head tree plus delta blocks.

This type exists only to back the SLIM inflate, deflate, and donor operators. Standard GP should continue to use PrimitiveTree.

Initialize a SLIM individual.

Parameters:

Name	Type	Description	Default
head	PrimitiveTree | list[Any]	Initial tree T placed at the list head.	required
deltas	list[PrimitiveTree] | None	Semantic delta blocks appended by inflate mutation.	None

Source code in deap_er/private/programming/slim/slim_tree.py

def __init__(
    self,
    head: PrimitiveTree | list[Any],
    deltas: list[PrimitiveTree] | None = None,
) -> None:
    """Initialize a SLIM individual.

    Args:
        head: Initial tree ``T`` placed at the list head.
        deltas: Semantic delta blocks appended by inflate mutation.
    """
    self.head = head if isinstance(head, PrimitiveTree) else PrimitiveTree(head)
    self.deltas = [
        delta if isinstance(delta, PrimitiveTree) else PrimitiveTree(delta)
        for delta in (deltas or [])
    ]

from_tree(tree) classmethod

Wrap a standard tree as a SLIM head with no deltas.

Parameters:

Name	Type	Description	Default
tree	SlimTree | PrimitiveTree | list[Any]	Tree to use as the SLIM head.	required

Returns:

Type	Description
SlimTree	A new SlimTree when tree is not already one.

Source code in deap_er/private/programming/slim/slim_tree.py

@classmethod
def from_tree(cls, tree: SlimTree | PrimitiveTree | list[Any]) -> SlimTree:
    """Wrap a standard tree as a SLIM head with no deltas.

    Args:
        tree: Tree to use as the SLIM head.

    Returns:
        A new ``SlimTree`` when ``tree`` is not already one.
    """
    if isinstance(tree, SlimTree):
        return tree
    return cls(PrimitiveTree(tree))

__len__()

Return the total node count across head and delta blocks.

Source code in deap_er/private/programming/slim/slim_tree.py

def __len__(self) -> int:
    """Return the total node count across head and delta blocks."""
    return len(self.head) + sum(len(delta) for delta in self.deltas)

__str__()

Return a Python expression that sums head and delta blocks.

Source code in deap_er/private/programming/slim/slim_tree.py

@override
def __str__(self) -> str:
    """Return a Python expression that sums head and delta blocks."""
    parts = [f"({self.head})"] + [f"({delta})" for delta in self.deltas]
    return " + ".join(parts)

__deepcopy__(memo)

Return a deep copy of this SLIM individual.

Parameters:

Name	Type	Description	Default
memo	dict[int, Any]	Memo mapping used by copy.deepcopy.	required

Returns:

Type	Description
SlimTree	A new SlimTree with copied head, deltas, and fitness.

Source code in deap_er/private/programming/slim/slim_tree.py

def __deepcopy__(self, memo: dict[int, Any]) -> SlimTree:
    """Return a deep copy of this SLIM individual.

    Args:
        memo: Memo mapping used by ``copy.deepcopy``.

    Returns:
        A new ``SlimTree`` with copied head, deltas, and fitness.
    """
    clone = SlimTree(
        deepcopy(self.head, memo),
        [deepcopy(delta, memo) for delta in self.deltas],
    )
    if hasattr(self, "fitness"):
        clone.fitness = deepcopy(self.fitness, memo)
    return clone

TapeFlags(all_nan, constant, hides_warmup) dataclass

Static certificates for a lowered tape.

skip_score property

Return whether scoring should skip interpret_tapes.

write_affine_scale(individual, intercept, slope, prim_set)

Write Keijzer a + b * f(x) back onto a GP expression.

A PrimitiveTree is wrapped as add(a, mul(b, tree)) with a and b as ephemeral leaves. A SlimTree wraps the head the same way and wraps each existing delta as mul(b, delta), so compile stays a + b * (head + sum(deltas)). Fitness and compile-cache entries for the previous expression are invalidated. This is the Lamarckian path next to :func:~deap_er.gp.tune_ephemerals; Darwinian callers use :func:~deap_er.tools.affine_scale or :func:~deap_er.tools.affine_case_errors for scoring only.

Parameters:

Name	Type	Description	Default
individual	Any	PrimitiveTree or SlimTree to wrap in place.	required
intercept	float	Fitted a.	required
slope	float	Fitted b.	required
prim_set	Any	Primitive set that must provide add / vadd and mul / vmul.	required

Returns:

Type	Description
Any	The same individual after write-back.

Raises:

Type	Description
TypeError	If add / mul (or the vector aliases) are missing from prim_set.

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

def write_affine_scale(
    individual: Any,
    intercept: float,
    slope: float,
    prim_set: Any,
) -> Any:
    """Write Keijzer ``a + b * f(x)`` back onto a GP expression.

    A ``PrimitiveTree`` is wrapped as ``add(a, mul(b, tree))`` with
    ``a`` and ``b`` as ephemeral leaves. A ``SlimTree`` wraps the head
    the same way and wraps each existing delta as ``mul(b, delta)``,
    so compile stays ``a + b * (head + sum(deltas))``. Fitness and
    compile-cache entries for the previous expression are invalidated.
    This is the Lamarckian path next to
    :func:`~deap_er.gp.tune_ephemerals`; Darwinian callers use
    :func:`~deap_er.tools.affine_scale` or
    :func:`~deap_er.tools.affine_case_errors` for scoring only.

    Args:
        individual: ``PrimitiveTree`` or ``SlimTree`` to wrap in place.
        intercept: Fitted ``a``.
        slope: Fitted ``b``.
        prim_set: Primitive set that must provide ``add`` / ``vadd``
            and ``mul`` / ``vmul``.

    Returns:
        The same ``individual`` after write-back.

    Raises:
        TypeError: If ``add`` / ``mul`` (or the vector aliases) are
            missing from ``prim_set``.
    """
    add_prim = _require_primitive(prim_set, _ADD_NAMES, "addition")
    mul_prim = _require_primitive(prim_set, _MUL_NAMES, "multiplication")
    ret_type = _leaf_ret(add_prim)
    template = _ephemeral_template(prim_set, ret_type)
    old_keys = _expression_keys(individual)
    if isinstance(individual, SlimTree):
        _wrap_slim(individual, intercept, slope, add_prim, mul_prim, ret_type, template)
    else:
        _wrap_tree(individual, intercept, slope, add_prim, mul_prim, ret_type, template)
    for key in dict.fromkeys(old_keys):
        invalidate_compiled(key)
    _clear_fitness(individual)
    return individual

make_column_pset(names, name='MAIN')

Build a typed primitive set with one Array input per column.

The resulting set expects and returns Array. Argument order is the column order: pset.arguments[i] is names[i], and a callable from compile_tree takes the columns as positional arguments in that same order.

Register operators onto the set with add_numpy_primitives, add_window_primitives, add_pair_window_primitives, and add_ts_primitives.

Parameters:

Name	Type	Description	Default
names	Sequence[str]	Column names, in the order the columns are passed to compiled trees. Each name must be a plain Python identifier that is not a keyword and does not look like the default ARG<n> argument prefix.	required
name	str	Name of the primitive set.	'MAIN'

Returns:

Type	Description
PrimitiveSetTyped	A typed primitive set with one renamed input terminal per
PrimitiveSetTyped	column.

Raises:

Type	Description
ValueError	If names is empty or holds an unusable name.

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

def make_column_pset(names: Sequence[str], name: str = "MAIN") -> PrimitiveSetTyped:
    """Build a typed primitive set with one ``Array`` input per column.

    The resulting set expects and returns ``Array``. Argument order is
    the column order: ``pset.arguments[i]`` is ``names[i]``, and a
    callable from ``compile_tree`` takes the columns as positional
    arguments in that same order.

    Register operators onto the set with ``add_numpy_primitives``,
    ``add_window_primitives``, ``add_pair_window_primitives``, and
    ``add_ts_primitives``.

    Args:
        names: Column names, in the order the columns are passed to
            compiled trees. Each name must be a plain Python
            identifier that is not a keyword and does not look like
            the default ``ARG<n>`` argument prefix.
        name: Name of the primitive set.

    Returns:
        A typed primitive set with one renamed input terminal per
        column.

    Raises:
        ValueError: If ``names`` is empty or holds an unusable name.
    """
    validate_names(names)
    in_types: list[type] = [Array] * len(names)
    prim_set = PrimitiveSetTyped(name, in_types, Array)
    prim_set.rename_arguments(**{f"ARG{i}": column for i, column in enumerate(names)})
    return prim_set

columnar_pset(names, *, name='MAIN', window=(2, 64), window_name='window', pair_windows=False, ts=False)

Build a columnar primitive set with the usual kits registered.

This is make_column_pset plus the vectorized and causal-window kits. Pair-window and time-series primitives are opt-in. A window ephemeral is added when window is a bound pair.

Parameters:

Name	Type	Description	Default
names	Sequence[str]	Column names, in the order compiled trees receive them.	required
name	str	Name of the primitive set.	'MAIN'
window	tuple[int, int] | None	Inclusive (low, high) window-ephemeral bounds. None skips the ephemeral.	(2, 64)
window_name	str	Ephemeral type name. Must be unique for a given bound pair across the process.	'window'
pair_windows	bool	If True, also register rolling_corr, rolling_cov, and rolling_beta.	False
ts	bool	If True, also register ts_rank, ts_argmax, and ts_argmin.	False

Returns:

Type	Description
PrimitiveSetTyped	A typed primitive set ready for register_gp.

Raises:

Type	Description
ValueError	If names is unusable or the window bounds are invalid.

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

def columnar_pset(
    names: Sequence[str],
    *,
    name: str = "MAIN",
    window: tuple[int, int] | None = (2, 64),
    window_name: str = "window",
    pair_windows: bool = False,
    ts: bool = False,
) -> PrimitiveSetTyped:
    """Build a columnar primitive set with the usual kits registered.

    This is ``make_column_pset`` plus the vectorized and causal-window
    kits. Pair-window and time-series primitives are opt-in. A window
    ephemeral is added when ``window`` is a bound pair.

    Args:
        names: Column names, in the order compiled trees receive them.
        name: Name of the primitive set.
        window: Inclusive ``(low, high)`` window-ephemeral bounds.
            ``None`` skips the ephemeral.
        window_name: Ephemeral type name. Must be unique for a given
            bound pair across the process.
        pair_windows: If True, also register ``rolling_corr``,
            ``rolling_cov``, and ``rolling_beta``.
        ts: If True, also register ``ts_rank``, ``ts_argmax``, and
            ``ts_argmin``.

    Returns:
        A typed primitive set ready for ``register_gp``.

    Raises:
        ValueError: If ``names`` is unusable or the window bounds
            are invalid.
    """
    pset = make_column_pset(names, name=name)
    add_numpy_primitives(pset)
    add_window_primitives(pset)
    if pair_windows:
        add_pair_window_primitives(pset)
    if ts:
        add_ts_primitives(pset)
    if window is not None:
        add_window_ephemeral(pset, window_name, window[0], window[1])
    return pset

evaluate_columnar(individuals, pset, matrix, target, *, cases=None, backend='opcode', parallel=False, empty=_DEFAULT_EMPTY, min_valid=None, reduce=True, static_filter=True)

Score trees against one packed column matrix.

Unique programs are lowered once by str(tree). The batch is run through interpret_tapes. Warmup nan samples are dropped from the MSE, matching the columnar fitness contract. Register the result as toolbox.evaluate_batch. When static_filter is true, tape_flags may skip interpret_tapes for identically nan, constant, or warmup-hiding programs and write empty instead. That path is what evaluate_invalid uses when evaluate_batch is this helper; nevals still counts the assignment.

Parameters:

Name	Type	Description	Default
individuals	Sequence[Any]	Trees to score. An empty sequence returns [].	required
pset	PrimitiveSetTyped	Primitive set used to lower each unique tree.	required
matrix	ndarray	Packed (n_rows, n_columns) column table.	required
target	ndarray	One-dimensional target series, length n_rows.	required
cases	Sequence[tuple[int, int]] | ndarray | None	Optional half-open (start, stop) ranges or a boolean mask forwarded to case_errors.	None
backend	str	Tape backend, 'opcode' or 'numba'.	'opcode'
parallel	bool	If True, run the Numba path with one workspace per thread.	False
empty	float	Fitness used when a prediction has too few finite samples, when cases is empty, or when every case is empty.	_DEFAULT_EMPTY
min_valid	int | None	Minimum finite overlap required when cases is omitted. Defaults to half the target length.	None
reduce	bool	When cases is set, return the mean of the case errors as a one-objective tuple, including a non-finite empty-case value. False returns the per-case tuple for lexicase.	True
static_filter	bool	When true, skip interpret_tapes for tapes that fail the static certificates from tape_flags.	True

Returns:

Type	Description
list[tuple[float, ...]]	One fitness tuple per individual, in input order.

Raises:

Type	Description
ValueError	If target is not a one-dimensional series whose length matches matrix rows.

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

def evaluate_columnar(
    individuals: Sequence[Any],
    pset: PrimitiveSetTyped,
    matrix: numpy.ndarray,
    target: numpy.ndarray,
    *,
    cases: Sequence[tuple[int, int]] | numpy.ndarray | None = None,
    backend: str = "opcode",
    parallel: bool = False,
    empty: float = _DEFAULT_EMPTY,
    min_valid: int | None = None,
    reduce: bool = True,
    static_filter: bool = True,
) -> list[tuple[float, ...]]:
    """Score trees against one packed column matrix.

    Unique programs are lowered once by ``str(tree)``. The batch is
    run through ``interpret_tapes``. Warmup ``nan`` samples are
    dropped from the MSE, matching the columnar fitness contract.
    Register the result as ``toolbox.evaluate_batch``. When
    ``static_filter`` is true, ``tape_flags`` may skip
    ``interpret_tapes`` for identically ``nan``, constant, or
    warmup-hiding programs and write ``empty`` instead. That path is
    what ``evaluate_invalid`` uses when ``evaluate_batch`` is this
    helper; ``nevals`` still counts the assignment.

    Args:
        individuals: Trees to score. An empty sequence returns ``[]``.
        pset: Primitive set used to lower each unique tree.
        matrix: Packed ``(n_rows, n_columns)`` column table.
        target: One-dimensional target series, length ``n_rows``.
        cases: Optional half-open ``(start, stop)`` ranges or a
            boolean mask forwarded to ``case_errors``.
        backend: Tape backend, ``'opcode'`` or ``'numba'``.
        parallel: If True, run the Numba path with one workspace
            per thread.
        empty: Fitness used when a prediction has too few finite
            samples, when ``cases`` is empty, or when every case
            is empty.
        min_valid: Minimum finite overlap required when ``cases``
            is omitted. Defaults to half the target length.
        reduce: When ``cases`` is set, return the mean of the case
            errors as a one-objective tuple, including a non-finite
            empty-case value. ``False`` returns the per-case tuple
            for lexicase.
        static_filter: When true, skip ``interpret_tapes`` for
            tapes that fail the static certificates from
            ``tape_flags``.

    Returns:
        One fitness tuple per individual, in input order.

    Raises:
        ValueError: If ``target`` is not a one-dimensional series
            whose length matches ``matrix`` rows.
    """
    expected = numpy.asarray(target, dtype=numpy.float64)
    if expected.ndim != 1:
        raise ValueError("target must be a one-dimensional series")
    packed = numpy.asarray(matrix, dtype=numpy.float64)
    if packed.ndim != 2 or packed.shape[0] != expected.shape[0]:
        raise ValueError("matrix rows must match the target length")
    if not individuals:
        return []
    tapes, index = _lower_unique(individuals, pset)
    floor = expected.size // 2 if min_valid is None else min_valid
    n_rows = int(packed.shape[0])
    if static_filter:
        bounds = bounds_from_matrix(packed)
        skip = [tape_skip_score(tape, bounds, n_rows=n_rows) for tape in tapes]
        score_tapes = [tape for tape, bad in zip(tapes, skip, strict=True) if not bad]
        predicted = (
            interpret_tapes(score_tapes, packed, backend=backend, parallel=parallel)
            if score_tapes
            else numpy.empty((0, n_rows), dtype=numpy.float64)
        )
        remap = _score_slot_map(skip)
        nan_row = numpy.full(n_rows, numpy.nan, dtype=numpy.float64)
        return [
            _score_prediction(
                nan_row if skip[slot] else predicted[remap[slot]],
                expected,
                cases,
                empty,
                floor,
                reduce,
            )
            for slot in index
        ]
    predicted = interpret_tapes(tapes, packed, backend=backend, parallel=parallel)
    return [
        _score_prediction(predicted[slot], expected, cases, empty, floor, reduce) for slot in index
    ]

build_tree_graph(expr)

Build a graph representation of a tree expression.

Parameters:

Name	Type	Description	Default
expr	GPExprTypes	Tree expression to convert.	required

Returns:

Type	Description
GPGraph	Nodes, edges, and a mapping of node indices to labels.

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

def build_tree_graph(expr: GPExprTypes) -> GPGraph:
    """Build a graph representation of a tree expression.

    Args:
        expr: Tree expression to convert.

    Returns:
        Nodes, edges, and a mapping of node indices to labels.
    """
    nodes = list(range(len(expr)))
    edges = []
    stack = []
    labels = {}

    for i, node in enumerate(expr):
        if stack:
            edges.append((stack[-1][0], i))
            stack[-1][1] -= 1
        if isinstance(node, Primitive):
            labels[i] = node.name
        elif hasattr(node, "value"):
            labels[i] = node.value
        else:
            labels[i] = str(node)
        stack.append([i, getattr(node, "arity", 0)])
        while stack and stack[-1][1] == 0:
            stack.pop()

    return nodes, edges, labels

clear_compile_cache()

Drop every compiled expression from the process-wide LRU cache.

Call this after mutating a primitive set so a later compile_tree cannot return a lambda compiled against the previous context. Also clears every live EvalCache so a language mutation cannot keep stale fitness.

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

def clear_compile_cache() -> None:
    """Drop every compiled expression from the process-wide LRU cache.

    Call this after mutating a primitive set so a later
    ``compile_tree`` cannot return a lambda compiled against the
    previous context. Also clears every live ``EvalCache`` so a
    language mutation cannot keep stale fitness.
    """
    _compile_cache.clear()
    clear_eval_caches()

compile_adf_tree(expr, prim_sets)

Compile a main tree together with its ADF trees.

The first element of expr is the main tree. The rest are automatically defined functions that the main tree may call.

Parameters:

Name	Type	Description	Default
expr	GPExprTypes	Sequence of expressions to compile, one per primitive set. Each item may be a string, a PrimitiveTree, or any object whose string form is valid Python.	required
prim_sets	GPTypedSets	Primitive sets aligned with expr. The first set is the main program and should refer to the ADFs; the following sets define those ADFs.	required

Returns:

Type	Description
Any	A callable if the main primitive set has one or more
Any	arguments, otherwise the result of the evaluation.

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

def compile_adf_tree(expr: GPExprTypes, prim_sets: GPTypedSets) -> Any:
    """Compile a main tree together with its ADF trees.

    The first element of ``expr`` is the main tree. The rest are
    automatically defined functions that the main tree may call.

    Args:
        expr: Sequence of expressions to compile, one per primitive
            set. Each item may be a string, a ``PrimitiveTree``, or
            any object whose string form is valid Python.
        prim_sets: Primitive sets aligned with ``expr``. The first
            set is the main program and should refer to the ADFs;
            the following sets define those ADFs.

    Returns:
        A callable if the main primitive set has one or more
        arguments, otherwise the result of the evaluation.
    """
    adf_dict = {}
    func = None
    for prim_set, sub_expr in reversed(list(zip(prim_sets, expr, strict=False))):
        prim_set.context.update(adf_dict)
        func = compile_tree(sub_expr, prim_set)
        adf_dict.update({prim_set.name: func})
    return func

compile_tree(expr, prim_set, *, backend='python', dispatch=None)

Evaluate expr against prim_set.

The default 'python' backend evaluates the expression as source text. The 'opcode' backend lowers the tree to a flat tape and runs it on a NumPy stack machine. The 'numba' backend runs the same tape through a compiled interpreter and needs the numba extra. Both tape backends require every primitive to carry an opcode, and reject the tree while lowering when one does not.

Compiled results are cached, keyed by a structural tree key or the source text, the argument names, the contents of the primitive set, the backend, the dispatcher, and the promoted-library generation. clear_compile_cache drops the table; promote_subtree does that on every mutation.

Parameters:

Name	Type	Description	Default
expr	GPExprTypes	Expression to compile. A string, a PrimitiveTree, or any object whose string form is valid Python.	required
prim_set	PrimitiveSetTyped	Primitive set that supplies the evaluation context.	required
backend	str	One of 'python', 'opcode', or 'numba'.	'python'
dispatch	Any	Compiled kernel that implements the consumer opcodes of the 'numba' backend. Ignored by the other backends.	None

Returns:

Type	Description
Any	A callable if prim_set has one or more arguments,
Any	otherwise the result of the evaluation.

Raises:

Type	Description
MemoryError	If evaluation exceeds the recursion limit.
NameError	If expr holds a primitive missing from prim_set.context.
ValueError	If the backend is unknown, or if a tape backend cannot lower the expression.

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

def compile_tree(
    expr: GPExprTypes,
    prim_set: PrimitiveSetTyped,
    *,
    backend: str = "python",
    dispatch: Any = None,
) -> Any:
    """Evaluate ``expr`` against ``prim_set``.

    The default ``'python'`` backend evaluates the expression as source
    text. The ``'opcode'`` backend lowers the tree to a flat tape and
    runs it on a NumPy stack machine. The ``'numba'`` backend runs the
    same tape through a compiled interpreter and needs the ``numba``
    extra. Both tape backends require every primitive to carry an
    opcode, and reject the tree while lowering when one does not.

    Compiled results are cached, keyed by a structural tree key or
    the source text, the argument names, the contents of the primitive
    set, the backend, the dispatcher, and the promoted-library
    generation. ``clear_compile_cache`` drops the table;
    ``promote_subtree`` does that on every mutation.

    Args:
        expr: Expression to compile. A string, a ``PrimitiveTree``,
            or any object whose string form is valid Python.
        prim_set: Primitive set that supplies the evaluation context.
        backend: One of ``'python'``, ``'opcode'``, or ``'numba'``.
        dispatch: Compiled kernel that implements the consumer opcodes
            of the ``'numba'`` backend. Ignored by the other backends.

    Returns:
        A callable if ``prim_set`` has one or more arguments,
        otherwise the result of the evaluation.

    Raises:
        MemoryError: If evaluation exceeds the recursion limit.
        NameError: If ``expr`` holds a primitive missing from
            ``prim_set.context``.
        ValueError: If the backend is unknown, or if a tape backend
            cannot lower the expression.
    """
    library = getattr(prim_set, "promoted_library", None)
    generation = 0 if library is None else library.generation
    cache_key = compile_cache_key(
        backend, dispatch, expr, prim_set.arguments, prim_set.context, generation
    )
    cached = _compile_cache.get(cache_key)
    if cached is not None:
        return cached

    _reject_unknown_primitive(expr, prim_set)
    if backend == "python":
        code = str(expr)
        if len(prim_set.arguments) > 0:
            args = ",".join(prim_set.arguments)
            code = f"lambda {args}: {code}"
        compiled = _compile_python(code, prim_set)
    elif backend in ("opcode", "numba"):
        compiled = _compile_tape(expr, prim_set, backend, dispatch)
    else:
        raise ValueError(
            f"Unknown compile backend '{backend}'. Use 'python', 'opcode', or 'numba'."
        )

    _compile_cache.set(cache_key, compiled)
    return compiled

static_limit(limiter, max_value)

Return a decorator that rejects oversized GP offspring.

May wrap crossover or mutation. An offspring whose measurement exceeds max_value is replaced by a randomly chosen parent.

Parameters:

Name	Type	Description	Default
limiter	Callable[..., Any]	Callable that measures an individual.	required
max_value	int | float	Maximum allowed measurement.	required

Returns:

Type	Description
Callable[..., Any]	A decorator for a GP operator registered on a Toolbox.

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

def static_limit(limiter: Callable[..., Any], max_value: int | float) -> Callable[..., Any]:
    """Return a decorator that rejects oversized GP offspring.

    May wrap crossover or mutation. An offspring whose measurement
    exceeds ``max_value`` is replaced by a randomly chosen parent.

    Args:
        limiter: Callable that measures an individual.
        max_value: Maximum allowed measurement.

    Returns:
        A decorator for a GP operator registered on a Toolbox.
    """

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> list[Any]:
            keep_inds = [clone_individual(ind) for ind in args]
            new_inds = list(func(*args, **kwargs))
            for i, ind in enumerate(new_inds):
                if keep_inds and limiter(ind) > max_value:
                    new_inds[i] = clone_individual(rng.choice(keep_inds))
            return new_inds

        return wrapper

    return decorator

cx_homologous(ind1, ind2)

Exchange subtrees at the same root-to-node path when types match.

A crossover point is chosen uniformly in ind1 (never the root). The same child-index path is resolved in ind2. When both nodes share a return type, their subtrees are swapped. Otherwise the operator falls back to random type-matched one-point crossover, the same contract as :func:cx_one_point.

Parameters:

Name	Type	Description	Default
ind1	GPIndividual	First individual to mate. Its crossover point anchors the homologous path.	required
ind2	GPIndividual	Second individual to mate.	required

Returns:

Type	Description
GPMates	The two individuals after subtree exchange.

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

def cx_homologous(ind1: GPIndividual, ind2: GPIndividual) -> GPMates:
    """Exchange subtrees at the same root-to-node path when types match.

    A crossover point is chosen uniformly in ``ind1`` (never the root).
    The same child-index path is resolved in ``ind2``. When both nodes
    share a return type, their subtrees are swapped. Otherwise the
    operator falls back to random type-matched one-point crossover, the
    same contract as :func:`cx_one_point`.

    Args:
        ind1: First individual to mate. Its crossover point anchors the
            homologous path.
        ind2: Second individual to mate.

    Returns:
        The two individuals after subtree exchange.
    """
    if len(ind1) < 2 or len(ind2) < 2:
        return ind1, ind2

    children1 = child_indices(list(ind1))
    index1 = int(rng.integers(1, len(ind1)))
    path = _path_from_root(children1, index1)
    index2 = _index_at_path(child_indices(list(ind2)), path)

    if index2 is not None and ind1[index1].ret == ind2[index2].ret:
        _swap_at(ind1, ind2, index1, index2)
    else:
        types1, types2, common_types = _common_type_candidates(ind1, ind2)
        _swap_subtrees(ind1, ind2, types1, types2, common_types)

    return ind1, ind2

cx_one_point(ind1, ind2)

Exchange a random subtree between two individuals.

A crossover point is chosen in each tree and the subtrees rooted there are swapped. Individuals shorter than two nodes are returned unchanged.

Parameters:

Name	Type	Description	Default
ind1	GPIndividual	First individual to mate.	required
ind2	GPIndividual	Second individual to mate.	required

Returns:

Type	Description
GPMates	The two individuals after subtree exchange.

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

def cx_one_point(ind1: GPIndividual, ind2: GPIndividual) -> GPMates:
    """Exchange a random subtree between two individuals.

    A crossover point is chosen in each tree and the subtrees rooted
    there are swapped. Individuals shorter than two nodes are returned
    unchanged.

    Args:
        ind1: First individual to mate.
        ind2: Second individual to mate.

    Returns:
        The two individuals after subtree exchange.
    """
    if len(ind1) < 2 or len(ind2) < 2:
        return ind1, ind2

    types1, types2, common_types = _common_type_candidates(ind1, ind2)
    _swap_subtrees(ind1, ind2, types1, types2, common_types)
    return ind1, ind2

cx_one_point_leaf_biased(ind1, ind2, term_prob)

Exchange a random subtree, biased toward terminals.

Same as one-point crossover, except each parent independently selects a terminal as the crossover point with probability term_prob.

Parameters:

Name	Type	Description	Default
ind1	GPIndividual	First individual to mate.	required
ind2	GPIndividual	Second individual to mate.	required
term_prob	float	Probability of choosing a terminal as the crossover point.	required

Returns:

Type	Description
GPMates	The two individuals after subtree exchange.

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

def cx_one_point_leaf_biased(ind1: GPIndividual, ind2: GPIndividual, term_prob: float) -> GPMates:
    """Exchange a random subtree, biased toward terminals.

    Same as one-point crossover, except each parent independently
    selects a terminal as the crossover point with probability
    ``term_prob``.

    Args:
        ind1: First individual to mate.
        ind2: Second individual to mate.
        term_prob: Probability of choosing a terminal as the
            crossover point.

    Returns:
        The two individuals after subtree exchange.
    """
    if len(ind1) < 2 or len(ind2) < 2:
        return ind1, ind2

    terminal_op = partial(eq, 0)
    primitive_op = partial(lt, 0)
    arity_op1 = terminal_op if rng.random() < term_prob else primitive_op
    arity_op2 = terminal_op if rng.random() < term_prob else primitive_op

    types1 = _collect_indices(ind1, arity_op1)
    types2 = _collect_indices(ind2, arity_op2)
    common_types = set(types1.keys()).intersection(types2.keys())
    _swap_subtrees(ind1, ind2, types1, types2, common_types)

    return ind1, ind2

cx_one_point_semantic(ind1, ind2, prim_set, matrix, *, valid=None, metric='euclidean')

Exchange subtrees at a semantically close type-matched pair.

One shared return type is chosen at random. ind1 donates a random crossover point of that type. ind2 donates the type-matched node whose interpret_tape row is nearest to the anchor subtree under metric. When every distance is infinite, ind2's partner is chosen uniformly among its type-matched candidates. Type-matched one-point (:func:~deap_er.gp.cx_one_point) remains the default toolbox mate; this operator is for columnar runs that already score interpret_tapes. Geometric semantic variation is :func:~deap_er.gp.cx_semantic.

Parameters:

Name	Type	Description	Default
ind1	GPIndividual	First individual to mate.	required
ind2	GPIndividual	Second individual to mate.	required
prim_set	PrimitiveSetTyped	Primitive set that can lower every subtree primitive.	required
matrix	ndarray	Packed (n_rows, n_columns) matrix for :func:~deap_er.gp.interpret_tapes.	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
metric	SemanticMetric	euclidean or cosine distance on finite rows.	'euclidean'

Returns:

Type	Description
GPMates	The two individuals after subtree exchange.

Raises:

Type	Description
ValueError	If a subtree cannot be lowered to a tape.

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

def cx_one_point_semantic(
    ind1: GPIndividual,
    ind2: GPIndividual,
    prim_set: PrimitiveSetTyped,
    matrix: numpy.ndarray,
    *,
    valid: numpy.ndarray | None = None,
    metric: SemanticMetric = "euclidean",
) -> GPMates:
    """Exchange subtrees at a semantically close type-matched pair.

    One shared return type is chosen at random. ``ind1`` donates a
    random crossover point of that type. ``ind2`` donates the
    type-matched node whose ``interpret_tape`` row is nearest to the
    anchor subtree under ``metric``. When every distance is infinite,
    ``ind2``'s partner is chosen uniformly among its type-matched
    candidates. Type-matched one-point (:func:`~deap_er.gp.cx_one_point`)
    remains the default toolbox mate; this operator is for columnar runs
    that already score ``interpret_tapes``. Geometric semantic variation
    is :func:`~deap_er.gp.cx_semantic`.

    Args:
        ind1: First individual to mate.
        ind2: Second individual to mate.
        prim_set: Primitive set that can lower every subtree primitive.
        matrix: Packed ``(n_rows, n_columns)`` matrix for
            :func:`~deap_er.gp.interpret_tapes`.
        valid: Optional per-row warmup mask of length ``n_rows``.
        metric: ``euclidean`` or ``cosine`` distance on finite rows.

    Returns:
        The two individuals after subtree exchange.

    Raises:
        ValueError: If a subtree cannot be lowered to a tape.
    """
    if len(ind1) < 2 or len(ind2) < 2:
        return ind1, ind2

    types1, types2, common_types = _common_type_candidates(ind1, ind2)
    if len(common_types) == 0:
        return ind1, ind2

    type_ = rng.choice(list(common_types))
    cands1 = types1[type_]
    cands2 = types2[type_]
    index1 = int(rng.choice(cands1))

    rows1 = _subtree_tape_rows([index1], ind1, prim_set, matrix)
    rows2 = _subtree_tape_rows(cands2, ind2, prim_set, matrix)
    nearest = semantic_nearest(rows1[0], rows2, k=1, metric=metric, valid=valid)
    index2 = int(rng.choice(cands2)) if nearest.size == 0 else cands2[int(nearest[0])]

    _swap_at(ind1, ind2, index1, index2)
    return ind1, ind2

gen_full(prim_set, min_depth, max_depth, ret_type=None)

Generate a full tree whose leaves share one depth.

The common leaf depth is drawn between min_depth and max_depth.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Primitive set from which nodes are selected.	required
min_depth	int	Minimum depth of the random tree.	required
max_depth	int	Maximum depth of the random tree.	required
ret_type	Any | None	Return type of the generated tree. Defaults to prim_set.ret.	None

Returns:

Type	Description
list[Any]	A full tree as a list of primitives and terminals.

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

def gen_full(
    prim_set: PrimitiveSetTyped, min_depth: int, max_depth: int, ret_type: Any | None = None
) -> list[Any]:
    """Generate a full tree whose leaves share one depth.

    The common leaf depth is drawn between ``min_depth`` and
    ``max_depth``.

    Args:
        prim_set: Primitive set from which nodes are selected.
        min_depth: Minimum depth of the random tree.
        max_depth: Maximum depth of the random tree.
        ret_type: Return type of the generated tree. Defaults to
            ``prim_set.ret``.

    Returns:
        A full tree as a list of primitives and terminals.
    """

    def condition(height: int, depth: int) -> bool:
        return height == depth

    return generate(prim_set, min_depth, max_depth, condition, ret_type)

gen_grow(prim_set, min_depth, max_depth, ret_type=None)

Generate a grown tree whose leaves may have different depths.

Each leaf depth lies between min_depth and max_depth.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Primitive set from which nodes are selected.	required
min_depth	int	Minimum depth of the random tree.	required
max_depth	int	Maximum depth of the random tree.	required
ret_type	Any | None	Return type of the generated tree. Defaults to prim_set.ret.	None

Returns:

Type	Description
list[Any]	A grown tree as a list of primitives and terminals.

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

def gen_grow(
    prim_set: PrimitiveSetTyped, min_depth: int, max_depth: int, ret_type: Any | None = None
) -> list[Any]:
    """Generate a grown tree whose leaves may have different depths.

    Each leaf depth lies between ``min_depth`` and ``max_depth``.

    Args:
        prim_set: Primitive set from which nodes are selected.
        min_depth: Minimum depth of the random tree.
        max_depth: Maximum depth of the random tree.
        ret_type: Return type of the generated tree. Defaults to
            ``prim_set.ret``.

    Returns:
        A grown tree as a list of primitives and terminals.
    """

    def condition(height: int, depth: int) -> bool:
        cond = rng.random() < prim_set.terminal_ratio
        return depth == height or (depth >= min_depth and cond)

    return generate(prim_set, min_depth, max_depth, condition, ret_type)

gen_half_and_half(prim_set, min_depth, max_depth, ret_type=None)

Generate a tree with either gen_grow or gen_full.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Primitive set from which nodes are selected.	required
min_depth	int	Minimum depth of the random tree.	required
max_depth	int	Maximum depth of the random tree.	required
ret_type	Any | None	Return type of the generated tree. Defaults to prim_set.ret.	None

Returns:

Type	Description
list[Any]	Either a full tree or a grown tree, as a list of primitives
list[Any]	and terminals.

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

def gen_half_and_half(
    prim_set: PrimitiveSetTyped, min_depth: int, max_depth: int, ret_type: Any | None = None
) -> list[Any]:
    """Generate a tree with either ``gen_grow`` or ``gen_full``.

    Args:
        prim_set: Primitive set from which nodes are selected.
        min_depth: Minimum depth of the random tree.
        max_depth: Maximum depth of the random tree.
        ret_type: Return type of the generated tree. Defaults to
            ``prim_set.ret``.

    Returns:
        Either a full tree or a grown tree, as a list of primitives
        and terminals.
    """
    choices = (gen_grow, gen_full)
    func = rng.choice(choices)
    return func(prim_set, min_depth, max_depth, ret_type)

generate(prim_set, min_depth, max_depth, condition, ret_type=None)

Grow a tree as a depth-first list of primitives and terminals.

Each branch grows until condition is true. A branch also stops early when its type has terminals but no primitives, which is how a strongly typed set expresses a leaf-only type such as a rolling window length. The list can be passed to PrimitiveTree to build a tree object.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Primitive set from which nodes are selected.	required
min_depth	int	Minimum depth of the random tree.	required
max_depth	int	Maximum depth of the random tree.	required
condition	Callable[..., bool]	Callable (height, depth) that decides when to stop growing a branch.	required
ret_type	Any | None	Return type of the generated tree. Defaults to prim_set.ret.	None

Returns:

Type	Description
list[Any]	A tree as a flat list of primitives and terminals, in
list[Any]	depth-first order.

Raises:

Type	Description
IndexError	If prim_set has no terminal or primitive of the required type.

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

def generate(
    prim_set: PrimitiveSetTyped,
    min_depth: int,
    max_depth: int,
    condition: Callable[..., bool],
    ret_type: Any | None = None,
) -> list[Any]:
    """Grow a tree as a depth-first list of primitives and terminals.

    Each branch grows until ``condition`` is true. A branch also stops
    early when its type has terminals but no primitives, which is how
    a strongly typed set expresses a leaf-only type such as a rolling
    window length. The list can be passed to ``PrimitiveTree`` to
    build a tree object.

    Args:
        prim_set: Primitive set from which nodes are selected.
        min_depth: Minimum depth of the random tree.
        max_depth: Maximum depth of the random tree.
        condition: Callable ``(height, depth)`` that decides when to
            stop growing a branch.
        ret_type: Return type of the generated tree. Defaults to
            ``prim_set.ret``.

    Returns:
        A tree as a flat list of primitives and terminals, in
        depth-first order.

    Raises:
        IndexError: If ``prim_set`` has no terminal or primitive of
            the required type.
    """
    if ret_type is None:
        ret_type = prim_set.ret
    expr = []
    height = rng.randint(min_depth, max_depth)
    stack = [(0, ret_type)]
    while len(stack) != 0:
        depth, ret_type = stack.pop()
        terminal_only = not prim_set.primitives[ret_type] and bool(prim_set.terminals[ret_type])
        if condition(height, depth) or terminal_only:
            expr.append(_choose_terminal(prim_set, ret_type))
        else:
            prim = _choose_primitive(prim_set, ret_type)
            expr.append(prim)
            for arg in reversed(prim.args):
                stack.append((depth + 1, arg))
    return expr

harm(toolbox, population, generations, cx_prob, mut_prob, hof=None, stats=None, verbose=False, **kwargs)

Evolve a GP population with HARM bloat control.

Default parameter values are recommended for most use cases. Requires mate, mutate, select, evaluate, clone, and map on toolbox.

Parameters:

Name	Type	Description	Default
toolbox	Toolbox	Toolbox with the evolution operators.	required
population	list[GPIndividual]	Individuals to evolve. Replaced in place.	required
generations	int	Number of generations to run.	required
cx_prob	float	Probability of mating two individuals.	required
mut_prob	float	Probability of mutating an individual.	required
hof	EvoRecords | None	Optional HallOfFame or ParetoFront to update.	None
stats	EvoStats | None	Optional Statistics or MultiStatistics to compile.	None
verbose	bool	If True, print the logbook stream each generation.	False
**kwargs	Any	HARM size-control knobs. Keyword-only. Accepted keys: alpha: Half-life of the exponential, scaled linearly with the cutoff. Higher values accept larger individuals. Default 0.05. beta: Minimum half-life, so growth remains possible while individuals are still small. Default 10.0. gamma: Fraction of individuals allowed past the cutoff. Default 0.25. rho: Fitness range used to place the cutoff. Higher values search more aggressively for slightly better solutions and may overfit. Default 0.9. nb_model: Individuals generated to model the natural size distribution. -1 uses max(2000, len(population)). Default -1. min_cutoff: Absolute minimum cutoff, to avoid shrinking the population too early. Default 20.	{}

Returns:

Type	Description
EvoAlgoResult	The final population and the logbook.

Raises:

Type	Description
TypeError	If kwargs contains an unknown name.

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

def harm(
    toolbox: Toolbox,
    population: list[GPIndividual],
    generations: int,
    cx_prob: float,
    mut_prob: float,
    hof: EvoRecords | None = None,
    stats: EvoStats | None = None,
    verbose: bool = False,
    **kwargs: Any,
) -> EvoAlgoResult:
    """Evolve a GP population with HARM bloat control.

    Default parameter values are recommended for most use cases.
    Requires ``mate``, ``mutate``, ``select``, ``evaluate``, ``clone``,
    and ``map`` on ``toolbox``.

    Args:
        toolbox: Toolbox with the evolution operators.
        population: Individuals to evolve. Replaced in place.
        generations: Number of generations to run.
        cx_prob: Probability of mating two individuals.
        mut_prob: Probability of mutating an individual.
        hof: Optional HallOfFame or ParetoFront to update.
        stats: Optional Statistics or MultiStatistics to compile.
        verbose: If True, print the logbook stream each generation.
        **kwargs: HARM size-control knobs. Keyword-only. Accepted
            keys:

            * ``alpha``: Half-life of the exponential, scaled
              linearly with the cutoff. Higher values accept larger
              individuals. Default ``0.05``.
            * ``beta``: Minimum half-life, so growth remains
              possible while individuals are still small. Default
              ``10.0``.
            * ``gamma``: Fraction of individuals allowed past the
              cutoff. Default ``0.25``.
            * ``rho``: Fitness range used to place the cutoff.
              Higher values search more aggressively for slightly
              better solutions and may overfit. Default ``0.9``.
            * ``nb_model``: Individuals generated to model the
              natural size distribution. ``-1`` uses
              ``max(2000, len(population))``. Default ``-1``.
            * ``min_cutoff``: Absolute minimum cutoff, to avoid
              shrinking the population too early. Default ``20``.

    Returns:
        The final population and the logbook.

    Raises:
        TypeError: If ``kwargs`` contains an unknown name.
    """
    unknown = kwargs.keys() - _HARM_DEFAULTS.keys()
    if unknown:
        names = ", ".join(repr(name) for name in sorted(unknown))
        raise TypeError(f"harm() got unexpected keyword argument(s): {names}")

    alpha = float(kwargs.get("alpha", _HARM_DEFAULTS["alpha"]))
    beta = float(kwargs.get("beta", _HARM_DEFAULTS["beta"]))
    gamma = float(kwargs.get("gamma", _HARM_DEFAULTS["gamma"]))
    rho = float(kwargs.get("rho", _HARM_DEFAULTS["rho"]))
    nb_model = int(kwargs.get("nb_model", _HARM_DEFAULTS["nb_model"]))
    min_cutoff = int(kwargs.get("min_cutoff", _HARM_DEFAULTS["min_cutoff"]))

    logbook = Logbook()
    logbook.header = ["gen", "nevals"] + (stats.fields if stats else [])

    nevals = evaluate_invalid(toolbox, population)

    if hof is not None:
        hof.update(population)

    record = stats.compile(population) if stats else {}
    logbook.record(gen=0, nevals=nevals, **record)

    if verbose:
        print(logbook.stream)

    if nb_model == -1:
        nb_model = max(2000, len(population))

    for gen in range(1, generations + 1):
        pop_len = len(population)
        natural_pop, natural_pop_sizes = produce(toolbox, population, nb_model, cx_prob, mut_prob)
        natural_hist = natural_histogram(natural_pop_sizes, pop_len, nb_model)
        _cutoff_size = cutoff_size(population, pop_len, rho, min_cutoff)

        def _target_prob(size: int, cutoff: int = _cutoff_size, length: int = pop_len) -> float:
            return target_prob(size, alpha, beta, gamma, length, cutoff)

        target_hist = target_histogram(natural_hist, _cutoff_size, _target_prob)
        accept_func = acceptance(natural_hist, target_hist, _target_prob)

        offspring, _ = produce(
            toolbox, population, pop_len, cx_prob, mut_prob, natural_pop, accept_func
        )

        nevals = evaluate_invalid(toolbox, offspring)

        if hof is not None:
            hof.update(offspring)

        population[:] = offspring
        record = stats.compile(population) if stats else {}
        logbook.record(gen=gen, nevals=nevals, **record)

        if verbose:
            print(logbook.stream)

    return population, logbook

tree_to_infix(expr)

Return a tree as an infix expression string.

Known arithmetic and logic names become infix operators. Other primitives stay in prefix name(args) form.

Parameters:

Name	Type	Description	Default
expr	Any	Prefix-ordered tree of primitives and terminals.	required

Returns:

Type	Description
str	The expression in infix notation.

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

def tree_to_infix(expr: Any) -> str:
    """Return a tree as an infix expression string.

    Known arithmetic and logic names become infix operators. Other
    primitives stay in prefix ``name(args)`` form.

    Args:
        expr: Prefix-ordered tree of primitives and terminals.

    Returns:
        The expression in infix notation.
    """
    string = ""
    stack: list[tuple[Any, list[str]]] = []
    for node in expr:
        stack.append((node, []))
        while stack and len(stack[-1][1]) == stack[-1][0].arity:
            prim, args = stack.pop()
            string = _format_node(prim, args)
            if not stack:
                break
            stack[-1][1].append(string)
    return str(string)

assign_ephemerals(individual, values)

Write repaired values onto numeric leaves, replacing each node.

Window coordinates are rounded to int and clamped to the ephemeral's inclusive legal range, or to >= 1 for a literal Window terminal. Nodes are replaced so a shallow clone does not alias values back onto the source tree.

Parameters:

Name	Type	Description	Default
individual	Any	PrimitiveTree or SlimTree to write.	required
values	Sequence[float]	Vector aligned with numeric_leaves.	required

Raises:

Type	Description
ValueError	If values is the wrong length.

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

def assign_ephemerals(individual: Any, values: Sequence[float]) -> None:
    """Write repaired values onto numeric leaves, replacing each node.

    Window coordinates are rounded to ``int`` and clamped to the
    ephemeral's inclusive legal range, or to ``>= 1`` for a literal
    ``Window`` terminal. Nodes are replaced so a shallow clone does
    not alias values back onto the source tree.

    Args:
        individual: ``PrimitiveTree`` or ``SlimTree`` to write.
        values: Vector aligned with ``numeric_leaves``.

    Raises:
        ValueError: If ``values`` is the wrong length.
    """
    leaves = numeric_leaves(individual)
    if len(values) != len(leaves):
        raise ValueError(f"Expected {len(leaves)} numeric-leaf values, got {len(values)}.")
    for (tree, index), value in zip(leaves, values, strict=True):
        tree[index] = _replaced_leaf(tree[index], value)

extract_ephemerals(individual)

Copy numeric-leaf values into a vector in walk order.

Parameters:

Name	Type	Description	Default
individual	Any	PrimitiveTree or SlimTree to read.	required

Returns:

Type	Description
ndarray	A length-n float array of current leaf values.

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

def extract_ephemerals(individual: Any) -> numpy.ndarray:
    """Copy numeric-leaf values into a vector in walk order.

    Args:
        individual: ``PrimitiveTree`` or ``SlimTree`` to read.

    Returns:
        A length-``n`` float array of current leaf values.
    """
    values = [float(tree[index].value) for tree, index in numeric_leaves(individual)]
    return numpy.asarray(values, dtype=float)

numeric_leaves(individual)

Return numeric-leaf locations in documented walk order.

A numeric leaf is an Ephemeral or a Terminal whose ret is Window. Order is prefix list order. A SlimTree walks head, then each delta, each in prefix order.

Parameters:

Name	Type	Description	Default
individual	Any	PrimitiveTree or SlimTree to walk.	required

Returns:

Type	Description
list[LeafLoc]	(tree, index) pairs in contract order.

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

def numeric_leaves(individual: Any) -> list[LeafLoc]:
    """Return numeric-leaf locations in documented walk order.

    A numeric leaf is an ``Ephemeral`` or a ``Terminal`` whose
    ``ret is Window``. Order is prefix list order. A ``SlimTree``
    walks ``head``, then each delta, each in prefix order.

    Args:
        individual: ``PrimitiveTree`` or ``SlimTree`` to walk.

    Returns:
        ``(tree, index)`` pairs in contract order.
    """
    leaves: list[LeafLoc] = []
    if isinstance(individual, SlimTree):
        trees = [individual.head, *individual.deltas]
    else:
        trees = [individual]
    for tree in trees:
        for index, node in enumerate(tree):
            if isinstance(node, Ephemeral) or (isinstance(node, Terminal) and node.ret is Window):
                leaves.append((tree, index))
    return leaves

tune_ephemerals(individual, strategy, evaluate=None, n_gen=5, *, evaluate_batch=None, clone=None)

Polish numeric leaves with a short boxed CMA run.

Extracts ephemeral floats and Window ints, runs n_gen generate / update steps on strategy, writes the repaired centroid back, then invalidates fitness, the compile cache, and matching EvalCache keys for the previous expression. Evaluation is the caller's evaluate on clones, or evaluate_batch on a pack of clones. Provide one of those callables; when both are set, the batch path is used.

Parameters:

Name	Type	Description	Default
individual	Any	PrimitiveTree or SlimTree to tune in place.	required
strategy	Any	Strategy or StrategySeparable whose dim matches the leaf count.	required
evaluate	Callable[[Any], Any] | None	callable(ind) -> fitness tuple. Optional when evaluate_batch is given.	None
n_gen	int	Inner CMA generations. Default 5.	5
evaluate_batch	Callable[[list[Any]], Any] | None	Optional callable(inds) -> fitness tuples.	None
clone	Callable[[Any], Any] | None	Individual copier. Defaults to clone_individual.	None

Returns:

Type	Description
Any	The same individual after write-back.

Raises:

Type	Description
ValueError	If n_gen < 1, strategy.dim does not match the number of numeric leaves, or neither evaluate nor evaluate_batch is given.

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

def tune_ephemerals(
    individual: Any,
    strategy: Any,
    evaluate: Callable[[Any], Any] | None = None,
    n_gen: int = 5,
    *,
    evaluate_batch: Callable[[list[Any]], Any] | None = None,
    clone: Callable[[Any], Any] | None = None,
) -> Any:
    """Polish numeric leaves with a short boxed CMA run.

    Extracts ephemeral floats and ``Window`` ints, runs ``n_gen``
    ``generate`` / ``update`` steps on ``strategy``, writes the
    repaired centroid back, then invalidates fitness, the compile
    cache, and matching ``EvalCache`` keys for the previous
    expression. Evaluation is the caller's
    ``evaluate`` on clones, or ``evaluate_batch`` on a pack of clones.
    Provide one of those callables; when both are set, the batch path
    is used.

    Args:
        individual: ``PrimitiveTree`` or ``SlimTree`` to tune in place.
        strategy: ``Strategy`` or ``StrategySeparable`` whose
            ``dim`` matches the leaf count.
        evaluate: ``callable(ind) ->`` fitness tuple. Optional when
            ``evaluate_batch`` is given.
        n_gen: Inner CMA generations. Default ``5``.
        evaluate_batch: Optional ``callable(inds) ->`` fitness tuples.
        clone: Individual copier. Defaults to ``clone_individual``.

    Returns:
        The same ``individual`` after write-back.

    Raises:
        ValueError: If ``n_gen < 1``, ``strategy.dim`` does not match
            the number of numeric leaves, or neither ``evaluate`` nor
            ``evaluate_batch`` is given.
    """
    if n_gen < 1:
        raise ValueError(f"n_gen must be at least 1, got {n_gen}.")
    leaves = numeric_leaves(individual)
    if not leaves:
        return individual
    if evaluate is None and evaluate_batch is None:
        raise ValueError("Provide evaluate or evaluate_batch.")
    if getattr(strategy, "dim", None) != len(leaves):
        raise ValueError(
            f"strategy.dim is {getattr(strategy, 'dim', None)}, "
            f"but the individual has {len(leaves)} numeric leaves."
        )
    strategy.centroid = numpy.asarray(extract_ephemerals(individual), dtype=float)
    _box_strategy(strategy, [tree[index] for tree, index in leaves])
    copier = clone_individual if clone is None else clone
    for _ in range(n_gen):
        trials = strategy.generate(_trial_init(individual))
        _score_trials(individual, trials, evaluate, evaluate_batch, copier)
        strategy.update(trials)
    old_keys = [expression_key(individual)]
    old_keys.extend(expression_key(tree) for tree, _ in leaves)
    assign_ephemerals(individual, _clipped_centroid(strategy).tolist())
    for key in dict.fromkeys(old_keys):
        invalidate_compiled(key)
    fitness = getattr(individual, "fitness", None)
    if fitness is not None and fitness.is_valid():
        del individual.fitness.values
    return individual

tune_ephemerals_budget(individual, strategy, evaluate=None, *, n_gen=None, evaluate_batch=None, clone=None, n_evals=None, nevals_used=0, exams=None, held_out=None, held_out_evaluate=None, held_out_evaluate_batch=None)

Polish numeric leaves under a memetic and evaluation leash.

Caps inner n_gen to :data:~deap_er.gp.MEMETIC_MAX_N_GEN and, when n_evals is set, to the remaining evaluation budget. Defaults to :data:~deap_er.gp.MEMETIC_DEFAULT_N_GEN inner generations.

When a caller-marked held-out exam is available from held_out or exams, trials are judged with held_out_evaluate or held_out_evaluate_batch instead of the train evaluate path.

Parameters:

Name	Type	Description	Default
individual	Any	PrimitiveTree or SlimTree to tune in place.	required
strategy	Any	Strategy or StrategySeparable whose dim matches the leaf count.	required
evaluate	Callable[[Any], Any] | None	callable(ind) -> fitness tuple for train scoring.	None
n_gen	int | None	Requested inner generations. Defaults to :data:~deap_er.gp.MEMETIC_DEFAULT_N_GEN.	None
evaluate_batch	Callable[[list[Any]], Any] | None	Optional callable(inds) -> fitness tuples for train scoring.	None
clone	Callable[[Any], Any] | None	Individual copier passed through to :func:~deap_er.gp.tune_ephemerals.	None
n_evals	int | None	Optional evaluation budget for the outer run.	None
nevals_used	int	Evaluations already charged to the run.	0
exams	Any	Optional exam pool whose held_out marker is read when held_out is omitted.	None
held_out	CaseExam | None	Optional held-out exam that overrides a pool marker.	None
held_out_evaluate	Callable[[Any], Any] | None	Held-out judge when a held-out exam exists.	None
held_out_evaluate_batch	Callable[[list[Any]], Any] | None	Batch held-out judge when a held-out exam exists.	None

Returns:

Type	Description
Any	(individual, evals_spent). evals_spent is 0 when the
int	budget is already spent or no numeric leaves exist.

Raises:

Type	Description
ValueError	If a held-out exam is marked but no held-out judge is given, or if neither train nor held-out evaluation is available.

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

def tune_ephemerals_budget(
    individual: Any,
    strategy: Any,
    evaluate: Callable[[Any], Any] | None = None,
    *,
    n_gen: int | None = None,
    evaluate_batch: Callable[[list[Any]], Any] | None = None,
    clone: Callable[[Any], Any] | None = None,
    n_evals: int | None = None,
    nevals_used: int = 0,
    exams: Any = None,
    held_out: CaseExam | None = None,
    held_out_evaluate: Callable[[Any], Any] | None = None,
    held_out_evaluate_batch: Callable[[list[Any]], Any] | None = None,
) -> tuple[Any, int]:
    """Polish numeric leaves under a memetic and evaluation leash.

    Caps inner ``n_gen`` to :data:`~deap_er.gp.MEMETIC_MAX_N_GEN` and,
    when ``n_evals`` is set, to the remaining evaluation budget. Defaults
    to :data:`~deap_er.gp.MEMETIC_DEFAULT_N_GEN` inner generations.

    When a caller-marked held-out exam is available from ``held_out`` or
    ``exams``, trials are judged with ``held_out_evaluate`` or
    ``held_out_evaluate_batch`` instead of the train ``evaluate`` path.

    Args:
        individual: ``PrimitiveTree`` or ``SlimTree`` to tune in place.
        strategy: ``Strategy`` or ``StrategySeparable`` whose ``dim``
            matches the leaf count.
        evaluate: ``callable(ind) ->`` fitness tuple for train scoring.
        n_gen: Requested inner generations. Defaults to
            :data:`~deap_er.gp.MEMETIC_DEFAULT_N_GEN`.
        evaluate_batch: Optional ``callable(inds) ->`` fitness tuples
            for train scoring.
        clone: Individual copier passed through to
            :func:`~deap_er.gp.tune_ephemerals`.
        n_evals: Optional evaluation budget for the outer run.
        nevals_used: Evaluations already charged to the run.
        exams: Optional exam pool whose ``held_out`` marker is read
            when ``held_out`` is omitted.
        held_out: Optional held-out exam that overrides a pool marker.
        held_out_evaluate: Held-out judge when a held-out exam exists.
        held_out_evaluate_batch: Batch held-out judge when a held-out
            exam exists.

    Returns:
        ``(individual, evals_spent)``. ``evals_spent`` is ``0`` when the
        budget is already spent or no numeric leaves exist.

    Raises:
        ValueError: If a held-out exam is marked but no held-out judge
            is given, or if neither train nor held-out evaluation is
            available.
    """
    resolved_held_out = resolve_tune_held_out(exams, held_out=held_out)
    judge_evaluate = evaluate
    judge_batch = evaluate_batch
    if resolved_held_out is not None:
        if held_out_evaluate is None and held_out_evaluate_batch is None:
            raise ValueError("held_out exam requires held_out_evaluate or held_out_evaluate_batch")
        judge_evaluate = held_out_evaluate
        judge_batch = held_out_evaluate_batch
    if judge_evaluate is None and judge_batch is None:
        raise ValueError("Provide evaluate or evaluate_batch.")
    requested = MEMETIC_DEFAULT_N_GEN if n_gen is None else int(n_gen)
    capped = cap_tune_n_gen(
        strategy,
        requested,
        n_evals=n_evals,
        nevals_used=nevals_used,
    )
    if capped < 1 or not numeric_leaves(individual):
        return individual, 0
    tune_ephemerals(
        individual,
        strategy,
        evaluate=judge_evaluate,
        n_gen=capped,
        evaluate_batch=judge_batch,
        clone=clone,
    )
    return individual, estimate_tune_ephemerals_evals(strategy, capped)

cap_tune_n_gen(strategy, n_gen, *, n_evals=None, nevals_used=0, max_n_gen=MEMETIC_MAX_N_GEN)

Cap inner tune generations to a memetic leash and optional budget.

Parameters:

Name	Type	Description	Default
strategy	Any	Strategy passed to :func:~deap_er.gp.tune_ephemerals.	required
n_gen	int	Requested inner generations.	required
n_evals	int | None	Optional evaluation budget. When set, the return value never exceeds what nevals_used can still afford.	None
nevals_used	int	Evaluations already charged to the run.	0
max_n_gen	int	Hard ceiling on inner generations. Defaults to :data:MEMETIC_MAX_N_GEN.	MEMETIC_MAX_N_GEN

Returns:

Type	Description
int	A non-negative generation count. 0 means tune should no-op.

Raises:

Type	Description
ValueError	If nevals_used is negative or max_n_gen is less than 1.

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

def cap_tune_n_gen(
    strategy: Any,
    n_gen: int,
    *,
    n_evals: int | None = None,
    nevals_used: int = 0,
    max_n_gen: int = MEMETIC_MAX_N_GEN,
) -> int:
    """Cap inner tune generations to a memetic leash and optional budget.

    Args:
        strategy: ``Strategy`` passed to
            :func:`~deap_er.gp.tune_ephemerals`.
        n_gen: Requested inner generations.
        n_evals: Optional evaluation budget. When set, the return
            value never exceeds what ``nevals_used`` can still afford.
        nevals_used: Evaluations already charged to the run.
        max_n_gen: Hard ceiling on inner generations. Defaults to
            :data:`MEMETIC_MAX_N_GEN`.

    Returns:
        A non-negative generation count. ``0`` means tune should no-op.

    Raises:
        ValueError: If ``nevals_used`` is negative or ``max_n_gen`` is
            less than ``1``.
    """
    if nevals_used < 0:
        raise ValueError("nevals_used must be at least 0")
    if max_n_gen < 1:
        raise ValueError("max_n_gen must be at least 1")
    capped = min(int(n_gen), max_n_gen)
    if n_evals is None:
        return capped
    remaining = int(n_evals) - nevals_used
    if remaining <= 0:
        return 0
    per_gen = estimate_tune_ephemerals_evals(strategy, 1)
    if per_gen <= 0:
        return 0
    return min(capped, remaining // per_gen)

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)

mut_ephemeral(individual, mode='all')

Resample one or all ephemeral constants in the tree.

Parameters:

Name	Type	Description	Default
individual	GPIndividual	GP tree to mutate.	required
mode	str	'one' to replace a single random ephemeral, or 'all' to replace every ephemeral.	'all'

Returns:

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

Raises:

Type	Description
ValueError	If mode is not 'one' or 'all'.

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

def mut_ephemeral(individual: GPIndividual, mode: str = "all") -> GPMutant:
    """Resample one or all ephemeral constants in the tree.

    Args:
        individual: GP tree to mutate.
        mode: ``'one'`` to replace a single random ephemeral, or
            ``'all'`` to replace every ephemeral.

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

    Raises:
        ValueError: If ``mode`` is not ``'one'`` or ``'all'``.
    """
    if mode not in ["one", "all"]:
        raise ValueError("Mode must be one of 'one' or 'all'.")

    ephemera_idx = []
    for index, node in enumerate(individual):
        if isinstance(node, Ephemeral):
            ephemera_idx.append(index)

    if len(ephemera_idx) > 0:
        if mode == "one":
            ephemera_idx = (rng.choice(ephemera_idx),)

        for i in ephemera_idx:
            individual[i] = type(individual[i])()

    return (individual,)

mut_insert(individual, prim_set)

Insert a new primitive branch at a random position.

Parameters:

Name	Type	Description	Default
individual	GPIndividual	GP tree to mutate.	required
prim_set	PrimitiveSetTyped	Primitive set used to choose the inserted branch.	required

Returns:

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

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

def mut_insert(individual: GPIndividual, prim_set: PrimitiveSetTyped) -> GPMutant:
    """Insert a new primitive branch at a random position.

    Args:
        individual: GP tree to mutate.
        prim_set: Primitive set used to choose the inserted branch.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    index = rng.randrange(len(individual))
    node = individual[index]
    slice_ = individual.search_subtree(index)

    primitives = []
    for p in prim_set.primitives[node.ret]:
        if node.ret in p.args:
            primitives.append(p)

    if len(primitives) == 0:
        return (individual,)

    new_node = choose_weighted(primitives)
    note_promoted_use(prim_set, new_node.name)
    new_subtree = [None] * len(new_node.args)

    choices = []
    for i, a in enumerate(new_node.args):
        if a == node.ret:
            choices.append(i)
    position = rng.choice(choices)

    if not _fill_sibling_terminals(new_subtree, new_node, prim_set, position):
        return (individual,)

    new_subtree[position : position + 1] = individual[slice_]
    new_subtree.insert(0, new_node)
    individual[slice_] = new_subtree

    return (individual,)

mut_node_replacement(individual, prim_set)

Replace a random node with a compatible node from prim_set.

Parameters:

Name	Type	Description	Default
individual	GPIndividual	GP tree to mutate.	required
prim_set	PrimitiveSetTyped	Primitive set to sample the replacement from.	required

Returns:

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

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

def mut_node_replacement(individual: GPIndividual, prim_set: PrimitiveSetTyped) -> GPMutant:
    """Replace a random node with a compatible node from ``prim_set``.

    Args:
        individual: GP tree to mutate.
        prim_set: Primitive set to sample the replacement from.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    if len(individual) < 2:
        return (individual,)

    index = rng.randrange(1, len(individual))
    node = individual[index]

    if node.arity == 0:
        term = rng.choice(prim_set.terminals[node.ret])
        if isclass(term):
            term = term()
        individual[index] = term
    else:
        node_ret = prim_set.primitives[node.ret]
        prims = [p for p in node_ret if p.args == node.args]
        individual[index] = choose_weighted(prims)
        note_promoted_use(prim_set, individual[index].name)

    return (individual,)

mut_shrink(individual)

Replace a random branch with one of its arguments.

Parameters:

Name	Type	Description	Default
individual	GPIndividual	GP tree to mutate.	required

Returns:

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

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

def mut_shrink(individual: GPIndividual) -> GPMutant:
    """Replace a random branch with one of its arguments.

    Args:
        individual: GP tree to mutate.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    if len(individual) < 3 or individual.height <= 1:
        return (individual,)

    i_prims = []
    for i, node in enumerate(individual[1:], 1):
        if isinstance(node, Primitive) and node.ret in node.args:
            i_prims.append((i, node))

    if len(i_prims) != 0:
        index, prim = rng.choice(i_prims)
        choices = []
        for i, type_ in enumerate(prim.args):
            if type_ == prim.ret:
                choices.append(i)
        arg_idx = rng.choice(choices)
        r_index = index + 1
        subtree = []
        for _ in range(arg_idx + 1):
            r_slice = individual.search_subtree(r_index)
            subtree = individual[r_slice]
            r_index += len(subtree)

        i_slice = individual.search_subtree(index)
        individual[i_slice] = subtree

    return (individual,)

mut_uniform(individual, expr, prim_set)

Replace a random subtree with an expression from expr.

Parameters:

Name	Type	Description	Default
individual	GPIndividual	GP tree to mutate.	required
expr	Callable[..., Any]	Callable that returns a random subtree.	required
prim_set	PrimitiveSetTyped	Primitive set passed to expr.	required

Returns:

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

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

def mut_uniform(
    individual: GPIndividual, expr: Callable[..., Any], prim_set: PrimitiveSetTyped
) -> GPMutant:
    """Replace a random subtree with an expression from ``expr``.

    Args:
        individual: GP tree to mutate.
        expr: Callable that returns a random subtree.
        prim_set: Primitive set passed to ``expr``.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    index = rng.randrange(len(individual))
    i_slice = individual.search_subtree(index)
    ret_type = individual[index].ret
    individual[i_slice] = expr(prim_set=prim_set, ret_type=ret_type)
    return (individual,)

bind_tape(tape, dispatch=None)

Bind a tape to the compiled interpreter.

The interpreter is compiled once per process and once more for each distinct dispatcher, never per tree. It is single threaded on purpose: parallelize across individuals through toolbox.map rather than inside a tree.

Every compiled tape shares one process-wide workspace, so the returned callable must not be run from several threads at once. The result it returns is a fresh array.

Parameters:

Name	Type	Description	Default
tape	Tape	Tape produced by lower_tree.	required
dispatch	Any	Compiled kernel implementing the opcodes at or above USER_BASE, following USER_DISPATCH_SIGNATURE.	None

Returns:

Type	Description
Callable[..., ndarray]	A callable that takes one array per column, or a single
Callable[..., ndarray]	(n_rows, n_columns) matrix, and returns the result.

Raises:

Type	Description
ImportError	If the numba extra is not installed.
ValueError	If the tape expects no columns, or if it holds consumer opcodes but no dispatcher was given.

Source code in deap_er/private/programming/numba/numba_ops.py

def bind_tape(tape: Tape, dispatch: Any = None) -> Callable[..., numpy.ndarray]:
    """Bind a tape to the compiled interpreter.

    The interpreter is compiled once per process and once more for each
    distinct dispatcher, never per tree. It is single threaded on
    purpose: parallelize across individuals through ``toolbox.map``
    rather than inside a tree.

    Every compiled tape shares one process-wide workspace, so the
    returned callable must not be run from several threads at once. The
    result it returns is a fresh array.

    Args:
        tape: Tape produced by ``lower_tree``.
        dispatch: Compiled kernel implementing the opcodes at or above
            ``USER_BASE``, following ``USER_DISPATCH_SIGNATURE``.

    Returns:
        A callable that takes one array per column, or a single
        ``(n_rows, n_columns)`` matrix, and returns the result.

    Raises:
        ImportError: If the ``numba`` extra is not installed.
        ValueError: If the tape expects no columns, or if it holds
            consumer opcodes but no dispatcher was given.
    """
    if tape.columns == 0:
        raise ValueError(
            "The numba backend evaluates a tape over columns and cannot size a "
            "result without them. Use backend='python' or backend='opcode' for a "
            "primitive set that takes no arguments."
        )
    run, idle = build()
    if dispatch is None:
        unknown = tape.opcodes[tape.opcodes >= USER_BASE]
        if unknown.size:
            raise ValueError(
                f"The tape holds consumer opcode {int(unknown[0])} but no dispatch "
                "kernel was given. Pass dispatch= to compile_tree."
            )
        dispatch = idle

    def call(*columns: Any) -> numpy.ndarray:
        matrix = as_matrix(columns, tape.columns)
        stack, scratch = reserve(tape.depth, matrix.shape[0])
        run(
            tape.opcodes,
            tape.operands,
            tape.constants,
            matrix,
            tape.fill,
            stack,
            scratch,
            dispatch,
        )
        return stack[0].copy()

    return call

numba_available()

Report whether the optional Numba dependency is importable.

Returns:

Type	Description
bool	True when the numba extra is installed.

Source code in deap_er/private/programming/numba/numba_ops.py

def numba_available() -> bool:
    """Report whether the optional Numba dependency is importable.

    Returns:
        True when the ``numba`` extra is installed.
    """
    try:
        import numba  # noqa: F401  # optional extra, probed without requiring it
    except ImportError:
        return False
    return True

warmup_numba(*, parallel=False, dispatch=None)

Compile the Numba tape interpreter for this process.

Runs a trivial column-load tape so the interpreter and the serial batch kernel are specialized before the first real evaluation. When NUMBA_CACHE_DIR is unset, a writable directory under the user's cache (~/.cache/deap-er/numba, or $XDG_CACHE_HOME/deap-er/numba) is used so spawned workers with a different working directory can reload the interpreter from disk instead of recompiling it.

Parameters:

Name	Type	Description	Default
parallel	bool	If True, also specialize the prange batch kernel.	False
dispatch	Any	Consumer kernel to specialize. None uses the idle dispatcher.	None

Raises:

Type	Description
ImportError	If the numba extra is not installed.

Source code in deap_er/private/programming/numba/numba_ops.py

def warmup_numba(*, parallel: bool = False, dispatch: Any = None) -> None:
    """Compile the Numba tape interpreter for this process.

    Runs a trivial column-load tape so the interpreter and the serial
    batch kernel are specialized before the first real evaluation.
    When ``NUMBA_CACHE_DIR`` is unset, a writable directory under the
    user's cache (``~/.cache/deap-er/numba``, or
    ``$XDG_CACHE_HOME/deap-er/numba``) is used so spawned workers with
    a different working directory can reload the interpreter from disk
    instead of recompiling it.

    Args:
        parallel: If True, also specialize the ``prange`` batch kernel.
        dispatch: Consumer kernel to specialize. ``None`` uses the idle
            dispatcher.

    Raises:
        ImportError: If the ``numba`` extra is not installed.
    """
    ensure_numba_cache_dir()
    pset = make_column_pset(["first"])
    add_numpy_primitives(pset)
    tape = lower_tree(PrimitiveTree([pset.mapping["first"]]), pset)
    matrix = numpy.zeros((2, 1), dtype=numpy.float64)
    interpret_tapes([tape], matrix, backend="numba", dispatch=dispatch)
    if parallel:
        interpret_tapes([tape], matrix, backend="numba", dispatch=dispatch, parallel=True)

vabs(value)

Take the elementwise absolute value of a series.

Parameters:

Name	Type	Description	Default
value	Any	Operand.	required

Returns:

Type	Description
Any	The elementwise absolute value.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vabs(value: Any) -> Any:
    """Take the elementwise absolute value of a series.

    Args:
        value: Operand.

    Returns:
        The elementwise absolute value.
    """
    return numpy.absolute(value)

vadd(left, right)

Add two series elementwise.

Parameters:

Name	Type	Description	Default
left	Any	Left operand.	required
right	Any	Right operand.	required

Returns:

Type	Description
Any	The elementwise sum.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vadd(left: Any, right: Any) -> Any:
    """Add two series elementwise.

    Args:
        left: Left operand.
        right: Right operand.

    Returns:
        The elementwise sum.
    """
    return numpy.add(left, right)

vcos(value)

Take the elementwise cosine of a series.

Parameters:

Name	Type	Description	Default
value	Any	Operand in radians.	required

Returns:

Type	Description
Any	The elementwise cosine.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vcos(value: Any) -> Any:
    """Take the elementwise cosine of a series.

    Args:
        value: Operand in radians.

    Returns:
        The elementwise cosine.
    """
    return numpy.cos(value)

vdiv(left, right, fill=DEFAULT_FILL)

Divide two series elementwise, protecting against zero divisors.

Positions where finite operands produced a non-finite quotient are replaced by fill. A non-finite value that came from an operand is preserved.

Parameters:

Name	Type	Description	Default
left	Any	Numerator.	required
right	Any	Denominator.	required
fill	float	Value substituted for a fabricated non-finite result.	DEFAULT_FILL

Returns:

Type	Description
Any	The protected elementwise quotient.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vdiv(left: Any, right: Any, fill: float = DEFAULT_FILL) -> Any:
    """Divide two series elementwise, protecting against zero divisors.

    Positions where finite operands produced a non-finite quotient are
    replaced by ``fill``. A non-finite value that came from an operand
    is preserved.

    Args:
        left: Numerator.
        right: Denominator.
        fill: Value substituted for a fabricated non-finite result.

    Returns:
        The protected elementwise quotient.
    """
    with numpy.errstate(divide="ignore", invalid="ignore"):
        result = numpy.divide(left, right)
    return protect(result, fill, left, right)

vlog(value, fill=DEFAULT_FILL)

Take the natural logarithm of a series, protecting the domain.

Parameters:

Name	Type	Description	Default
value	Any	Operand.	required
fill	float	Value substituted for a fabricated non-finite result.	DEFAULT_FILL

Returns:

Type	Description
Any	The protected elementwise logarithm.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vlog(value: Any, fill: float = DEFAULT_FILL) -> Any:
    """Take the natural logarithm of a series, protecting the domain.

    Args:
        value: Operand.
        fill: Value substituted for a fabricated non-finite result.

    Returns:
        The protected elementwise logarithm.
    """
    with numpy.errstate(divide="ignore", invalid="ignore"):
        result = numpy.log(value)
    return protect(result, fill, value)

vmul(left, right)

Multiply two series elementwise.

Parameters:

Name	Type	Description	Default
left	Any	Left operand.	required
right	Any	Right operand.	required

Returns:

Type	Description
Any	The elementwise product.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vmul(left: Any, right: Any) -> Any:
    """Multiply two series elementwise.

    Args:
        left: Left operand.
        right: Right operand.

    Returns:
        The elementwise product.
    """
    return numpy.multiply(left, right)

vneg(value)

Negate a series elementwise.

Parameters:

Name	Type	Description	Default
value	Any	Operand to negate.	required

Returns:

Type	Description
Any	The elementwise negation.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vneg(value: Any) -> Any:
    """Negate a series elementwise.

    Args:
        value: Operand to negate.

    Returns:
        The elementwise negation.
    """
    return numpy.negative(value)

vsin(value)

Take the elementwise sine of a series.

Parameters:

Name	Type	Description	Default
value	Any	Operand in radians.	required

Returns:

Type	Description
Any	The elementwise sine.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vsin(value: Any) -> Any:
    """Take the elementwise sine of a series.

    Args:
        value: Operand in radians.

    Returns:
        The elementwise sine.
    """
    return numpy.sin(value)

vsqrt(value, fill=DEFAULT_FILL)

Take the square root of a series, protecting the domain.

Parameters:

Name	Type	Description	Default
value	Any	Operand.	required
fill	float	Value substituted for a fabricated non-finite result.	DEFAULT_FILL

Returns:

Type	Description
Any	The protected elementwise square root.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vsqrt(value: Any, fill: float = DEFAULT_FILL) -> Any:
    """Take the square root of a series, protecting the domain.

    Args:
        value: Operand.
        fill: Value substituted for a fabricated non-finite result.

    Returns:
        The protected elementwise square root.
    """
    with numpy.errstate(invalid="ignore"):
        result = numpy.sqrt(value)
    return protect(result, fill, value)

vsub(left, right)

Subtract two series elementwise.

Parameters:

Name	Type	Description	Default
left	Any	Left operand.	required
right	Any	Right operand.	required

Returns:

Type	Description
Any	The elementwise difference.

Source code in deap_er/private/programming/numpy/numpy_arith.py

def vsub(left: Any, right: Any) -> Any:
    """Subtract two series elementwise.

    Args:
        left: Left operand.
        right: Right operand.

    Returns:
        The elementwise difference.
    """
    return numpy.subtract(left, right)

vand(left, right)

Combine two masks elementwise with logical and.

Parameters:

Name	Type	Description	Default
left	Any	Left mask.	required
right	Any	Right mask.	required

Returns:

Type	Description
Any	A boolean mask.

Source code in deap_er/private/programming/numpy/numpy_logic.py

def vand(left: Any, right: Any) -> Any:
    """Combine two masks elementwise with logical and.

    Args:
        left: Left mask.
        right: Right mask.

    Returns:
        A boolean mask.
    """
    return numpy.logical_and(left, right)

veq(left, right)

Compare two series elementwise with ==.

Parameters:

Name	Type	Description	Default
left	Any	Left operand.	required
right	Any	Right operand.	required

Returns:

Type	Description
Any	A boolean mask.

Source code in deap_er/private/programming/numpy/numpy_logic.py

def veq(left: Any, right: Any) -> Any:
    """Compare two series elementwise with ``==``.

    Args:
        left: Left operand.
        right: Right operand.

    Returns:
        A boolean mask.
    """
    return numpy.equal(left, right)

vge(left, right)

Compare two series elementwise with >=.

Parameters:

Name	Type	Description	Default
left	Any	Left operand.	required
right	Any	Right operand.	required

Returns:

Type	Description
Any	A boolean mask.

Source code in deap_er/private/programming/numpy/numpy_logic.py

def vge(left: Any, right: Any) -> Any:
    """Compare two series elementwise with ``>=``.

    Args:
        left: Left operand.
        right: Right operand.

    Returns:
        A boolean mask.
    """
    return numpy.greater_equal(left, right)

vgt(left, right)

Compare two series elementwise with >.

Parameters:

Name	Type	Description	Default
left	Any	Left operand.	required
right	Any	Right operand.	required

Returns:

Type	Description
Any	A boolean mask.

Source code in deap_er/private/programming/numpy/numpy_logic.py

def vgt(left: Any, right: Any) -> Any:
    """Compare two series elementwise with ``>``.

    Args:
        left: Left operand.
        right: Right operand.

    Returns:
        A boolean mask.
    """
    return numpy.greater(left, right)

vle(left, right)

Compare two series elementwise with <=.

Parameters:

Name	Type	Description	Default
left	Any	Left operand.	required
right	Any	Right operand.	required

Returns:

Type	Description
Any	A boolean mask.

Source code in deap_er/private/programming/numpy/numpy_logic.py

def vle(left: Any, right: Any) -> Any:
    """Compare two series elementwise with ``<=``.

    Args:
        left: Left operand.
        right: Right operand.

    Returns:
        A boolean mask.
    """
    return numpy.less_equal(left, right)

vlt(left, right)

Compare two series elementwise with <.

Parameters:

Name	Type	Description	Default
left	Any	Left operand.	required
right	Any	Right operand.	required

Returns:

Type	Description
Any	A boolean mask.

Source code in deap_er/private/programming/numpy/numpy_logic.py

def vlt(left: Any, right: Any) -> Any:
    """Compare two series elementwise with ``<``.

    Args:
        left: Left operand.
        right: Right operand.

    Returns:
        A boolean mask.
    """
    return numpy.less(left, right)

vnot(value)

Invert a mask elementwise.

Parameters:

Name	Type	Description	Default
value	Any	Mask to invert.	required

Returns:

Type	Description
Any	A boolean mask.

Source code in deap_er/private/programming/numpy/numpy_logic.py

def vnot(value: Any) -> Any:
    """Invert a mask elementwise.

    Args:
        value: Mask to invert.

    Returns:
        A boolean mask.
    """
    return numpy.logical_not(value)

vor(left, right)

Combine two masks elementwise with logical or.

Parameters:

Name	Type	Description	Default
left	Any	Left mask.	required
right	Any	Right mask.	required

Returns:

Type	Description
Any	A boolean mask.

Source code in deap_er/private/programming/numpy/numpy_logic.py

def vor(left: Any, right: Any) -> Any:
    """Combine two masks elementwise with logical or.

    Args:
        left: Left mask.
        right: Right mask.

    Returns:
        A boolean mask.
    """
    return numpy.logical_or(left, right)

vwhere(condition, on_true, on_false)

Select elementwise between two series.

This is how a program turns a condition into a value without a Python if.

Parameters:

Name	Type	Description	Default
condition	Any	Mask that selects the branch.	required
on_true	Any	Values taken where the mask is true.	required
on_false	Any	Values taken where the mask is false.	required

Returns:

Type	Description
Any	The selected series.

Source code in deap_er/private/programming/numpy/numpy_logic.py

def vwhere(condition: Any, on_true: Any, on_false: Any) -> Any:
    """Select elementwise between two series.

    This is how a program turns a condition into a value without a
    Python ``if``.

    Args:
        condition: Mask that selects the branch.
        on_true: Values taken where the mask is true.
        on_false: Values taken where the mask is false.

    Returns:
        The selected series.
    """
    return numpy.where(condition, on_true, on_false)

add_numpy_primitives(prim_set, *, fill=DEFAULT_FILL)

Register the vectorized primitive kit on a typed primitive set.

Adds arithmetic, protected division and domain-limited unary operations, comparisons that return Mask, mask logic, and vwhere. The constants True and False are registered as Mask terminals so tree generation can terminate a mask branch.

The chosen fill is bound into the protected primitives, where infer_fill can read it back so the other backends reproduce the same protection.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Typed primitive set built by make_column_pset or an equivalent set over Array and Mask.	required
fill	float	Value substituted for a non-finite result of a protected operation that was computed from finite operands.	DEFAULT_FILL

Raises:

Type	Description
ValueError	If a primitive name collides with an argument of prim_set, or if a name is already registered.

Source code in deap_er/private/programming/numpy/numpy_ops.py

def add_numpy_primitives(prim_set: PrimitiveSetTyped, *, fill: float = DEFAULT_FILL) -> None:
    """Register the vectorized primitive kit on a typed primitive set.

    Adds arithmetic, protected division and domain-limited unary
    operations, comparisons that return ``Mask``, mask logic, and
    ``vwhere``. The constants ``True`` and ``False`` are registered as
    ``Mask`` terminals so tree generation can terminate a mask branch.

    The chosen ``fill`` is bound into the protected primitives, where
    ``infer_fill`` can read it back so the other backends reproduce the
    same protection.

    Args:
        prim_set: Typed primitive set built by ``make_column_pset`` or
            an equivalent set over ``Array`` and ``Mask``.
        fill: Value substituted for a non-finite result of a protected
            operation that was computed from finite operands.

    Raises:
        ValueError: If a primitive name collides with an argument of
            ``prim_set``, or if a name is already registered.
    """
    binary: list[type] = [Array, Array]
    unary: list[type] = [Array]

    plain: dict[str, tuple[Callable[..., Any], list[type]]] = {
        "vadd": (vadd, binary),
        "vsub": (vsub, binary),
        "vmul": (vmul, binary),
        "vneg": (vneg, unary),
        "vabs": (vabs, unary),
        "vsin": (vsin, unary),
        "vcos": (vcos, unary),
    }
    protected: dict[str, tuple[Callable[..., Any], list[type]]] = {
        "vdiv": (vdiv, binary),
        "vlog": (vlog, unary),
        "vsqrt": (vsqrt, unary),
    }
    compares: dict[str, Callable[..., Any]] = {
        "vgt": vgt,
        "vlt": vlt,
        "vge": vge,
        "vle": vle,
        "veq": veq,
    }
    logic: dict[str, tuple[Callable[..., Any], list[type]]] = {
        "vand": (vand, [Mask, Mask]),
        "vor": (vor, [Mask, Mask]),
        "vnot": (vnot, [Mask]),
    }

    names = [*plain, *protected, *compares, *logic, "vwhere"]
    reject_shadowed(prim_set, names)

    for name, (func, in_types) in plain.items():
        prim_set.add_primitive(func, in_types, Array, name)
    for name, (func, in_types) in protected.items():
        prim_set.add_primitive(partial(func, fill=fill), in_types, Array, name)
    for name, func in compares.items():
        prim_set.add_primitive(func, binary, Mask, name)
    for name, (func, in_types) in logic.items():
        prim_set.add_primitive(func, in_types, Mask, name)

    prim_set.add_primitive(vwhere, [Mask, Array, Array], Array, "vwhere")
    prim_set.add_terminal(True, Mask)
    prim_set.add_terminal(False, Mask)

bind_numba_opcode(name, opcode)

Bind a primitive name to a consumer-supplied opcode.

The binding is used only while lowering a tree. It is never read inside the interpreter loop. Persist the bindings alongside a run: replaying a checkpointed tree against different bindings decodes the tree into different instructions.

Parameters:

Name	Type	Description	Default
name	str	Name of a primitive registered on the primitive set.	required
opcode	int	Opcode value, at or above USER_BASE.	required

Raises:

Type	Description
ValueError	If the opcode is below USER_BASE, if the name belongs to a builtin primitive, or if the name is already bound to a different opcode.

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

def bind_numba_opcode(name: str, opcode: int) -> None:
    """Bind a primitive name to a consumer-supplied opcode.

    The binding is used only while lowering a tree. It is never read
    inside the interpreter loop. Persist the bindings alongside a run:
    replaying a checkpointed tree against different bindings decodes
    the tree into different instructions.

    Args:
        name: Name of a primitive registered on the primitive set.
        opcode: Opcode value, at or above ``USER_BASE``.

    Raises:
        ValueError: If the opcode is below ``USER_BASE``, if the name
            belongs to a builtin primitive, or if the name is already
            bound to a different opcode.
    """
    if opcode < USER_BASE:
        raise ValueError(f"Consumer opcodes must be at least {USER_BASE}, got {opcode}.")
    if name in BUILTIN_OPCODES:
        raise ValueError(f"The primitive '{name}' already has a builtin opcode.")
    known = _user_opcodes.get(name)
    if known is not None and known != opcode:
        raise ValueError(f"The primitive '{name}' is already bound to opcode {known}.")
    _user_opcodes[name] = opcode

interpret_tape(tape, columns)

Run a tape over column inputs.

Evaluates through the same functions as the default backend, so the two agree by construction.

Parameters:

Name	Type	Description	Default
tape	Tape	Tape produced by lower_tree.	required
columns	Sequence[Any] | ndarray	One array per column, in the order the primitive set declares them, or one packed (n_rows, n_columns) matrix.	required

Returns:

Type	Description
Any	The result of the expression.

Raises:

Type	Description
ValueError	If the column count does not match the tape, if the tape holds an instruction the interpreter does not know, or if the tape underflows or leaves no result.

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

def interpret_tape(tape: Tape, columns: Sequence[Any] | numpy.ndarray) -> Any:
    """Run a tape over column inputs.

    Evaluates through the same functions as the default backend, so
    the two agree by construction.

    Args:
        tape: Tape produced by ``lower_tree``.
        columns: One array per column, in the order the primitive set
            declares them, or one packed ``(n_rows, n_columns)``
            matrix.

    Returns:
        The result of the expression.

    Raises:
        ValueError: If the column count does not match the tape, if
            the tape holds an instruction the interpreter does not
            know, or if the tape underflows or leaves no result.
    """
    column_count = _column_count(columns)
    if column_count != tape.columns:
        raise ValueError(f"The tape expects {tape.columns} columns, got {column_count}.")

    stack: list[Any] = []
    for step in range(tape.opcodes.size):
        opcode = int(tape.opcodes[step])
        operand = int(tape.operands[step])
        _apply_opcode(stack, columns, tape, opcode, operand)
    if not stack:
        raise ValueError("The tape is malformed and leaves no result.")
    return stack[-1]

lower_tree(expr, prim_set, *, fill=None)

Lower an expression tree to a flat instruction tape.

Only primitives with a builtin opcode or a binding made through bind_numba_opcode can be lowered. Everything else fails here rather than at evaluation time.

Parameters:

Name	Type	Description	Default
expr	GPExprTypes	Expression to lower. A PrimitiveTree, a sequence of nodes, or a string that PrimitiveTree.from_string can parse against prim_set.	required
prim_set	PrimitiveSetTyped	Primitive set the expression was built from.	required
fill	float | None	Fill for the protected instructions. Read back from prim_set when omitted.	None

Returns:

Type	Description
Tape	The lowered tape.

Raises:

Type	Description
ValueError	If a primitive has no opcode, if a window argument is not an integer leaf, if a terminal is neither a column nor a number, or if the tree is empty, holds unreachable nodes, or does not balance the evaluation stack.

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

def lower_tree(
    expr: GPExprTypes, prim_set: PrimitiveSetTyped, *, fill: float | None = None
) -> Tape:
    """Lower an expression tree to a flat instruction tape.

    Only primitives with a builtin opcode or a binding made through
    ``bind_numba_opcode`` can be lowered. Everything else fails here
    rather than at evaluation time.

    Args:
        expr: Expression to lower. A ``PrimitiveTree``, a sequence of
            nodes, or a string that ``PrimitiveTree.from_string`` can
            parse against ``prim_set``.
        prim_set: Primitive set the expression was built from.
        fill: Fill for the protected instructions. Read back from
            ``prim_set`` when omitted.

    Returns:
        The lowered tape.

    Raises:
        ValueError: If a primitive has no opcode, if a window argument
            is not an integer leaf, if a terminal is neither a column
            nor a number, or if the tree is empty, holds unreachable
            nodes, or does not balance the evaluation stack.
    """
    if isinstance(expr, str):
        expr = PrimitiveTree.from_string(expr, prim_set)
    nodes = list(expr)
    if not nodes:
        raise ValueError("An empty expression cannot be lowered.")
    nodes = expand_promoted(nodes, prim_set)

    children = child_indices(nodes)
    windows, folded = immediate_windows(nodes, children)
    order = postfix_order(children, folded)
    if len(order) + len(folded) != len(nodes):
        raise ValueError("The expression holds nodes that are not reachable from the root.")

    opcodes: list[int] = []
    operands: list[int] = []
    constants: list[float] = []
    pool: dict[float, int] = {}
    pointer = 0
    depth = 0

    for index in order:
        node = nodes[index]
        if not isinstance(node, Primitive):
            opcode, operand = leaf_instruction(node, prim_set, pool, constants)
            pointer += 1
        else:
            opcode = opcode_of(node.name)
            operand = windows.get(index, -1)
            # Builtins are all in _ARITY, and they are the only nodes
            # that fold an operand into the instruction. A consumer
            # kernel takes every argument from the stack.
            pointer -= OPCODES_ARITY.get(opcode, node.arity) - 1
        if pointer < 1:
            raise ValueError("The expression is malformed and underflows the evaluation stack.")
        depth = max(depth, pointer)
        opcodes.append(opcode)
        operands.append(operand)

    if pointer != 1:
        raise ValueError("The expression is malformed and leaves more than one result.")

    return Tape(
        opcodes=numpy.array(opcodes, dtype=numpy.int32),
        operands=numpy.array(operands, dtype=numpy.int32),
        constants=numpy.array(constants, dtype=numpy.float64),
        columns=len(prim_set.arguments),
        depth=depth,
        fill=numpy_ops.infer_fill(prim_set) if fill is None else float(fill),
    )

numba_opcodes()

Return the consumer opcode bindings made so far.

Returns:

Type	Description
dict[str, int]	A copy of the name to opcode map, suitable for storing with a
dict[str, int]	checkpoint.

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

def numba_opcodes() -> dict[str, int]:
    """Return the consumer opcode bindings made so far.

    Returns:
        A copy of the name to opcode map, suitable for storing with a
        checkpoint.
    """
    return dict(_user_opcodes)

promote_subtree(prim_set, expr, index=0, *, max_library=32, prefix='promo', weight=1.0)

Lift a complete typed subtree into prim_set as a primitive.

The caller fires this helper. It does not run inside ea_*. Formals are the set's argument terminals that appear in the subtree, in prim_set.arguments order. Constants and ephemerals stay baked into the body. Nested promoted names are expanded before the body is stored, so evicting an inner name does not leave USER_BASE in later tapes. On a columnar set the name is bound at or above USER_BASE and lower_tree expands the body so tapes stay on builtin opcodes. Promotion clears the compile cache and every live EvalCache.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Set that receives the new primitive.	required
expr	PrimitiveTree | Sequence[Any]	Prefix tree holding the subtree.	required
index	int	Root index of the subtree to lift.	0
max_library	int	Maximum number of promoted names to keep.	32
prefix	str	Prefix of the generated name (promo0, …).	'promo'
weight	float	Sampling weight of the new primitive.	1.0

Returns:

Type	Description
str	The generated primitive name.

Raises:

Type	Description
IndexError	If index is outside expr.
TypeError	If a node type does not match its parent slot.
ValueError	If max_library is less than 1, if weight is not greater than 0, if the slice is not a complete typed tree, if it is a lone argument terminal, if it contains an ADF, if the body cannot be lowered, or if an untyped set would receive a zero-arity primitive.

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

def promote_subtree(
    prim_set: PrimitiveSetTyped,
    expr: PrimitiveTree | Sequence[Any],
    index: int = 0,
    *,
    max_library: int = 32,
    prefix: str = "promo",
    weight: float = 1.0,
) -> str:
    """Lift a complete typed subtree into ``prim_set`` as a primitive.

    The caller fires this helper. It does not run inside ``ea_*``.
    Formals are the set's argument terminals that appear in the
    subtree, in ``prim_set.arguments`` order. Constants and
    ephemerals stay baked into the body. Nested promoted names are
    expanded before the body is stored, so evicting an inner name
    does not leave ``USER_BASE`` in later tapes. On a columnar set
    the name is bound at or above ``USER_BASE`` and ``lower_tree``
    expands the body so tapes stay on builtin opcodes. Promotion
    clears the compile cache and every live ``EvalCache``.

    Args:
        prim_set: Set that receives the new primitive.
        expr: Prefix tree holding the subtree.
        index: Root index of the subtree to lift.
        max_library: Maximum number of promoted names to keep.
        prefix: Prefix of the generated name (``promo0``, …).
        weight: Sampling weight of the new primitive.

    Returns:
        The generated primitive name.

    Raises:
        IndexError: If ``index`` is outside ``expr``.
        TypeError: If a node type does not match its parent slot.
        ValueError: If ``max_library`` is less than 1, if ``weight``
            is not greater than 0, if the slice is not a complete
            typed tree, if it is a lone argument terminal, if it
            contains an ADF, if the body cannot be lowered, or if
            an untyped set would receive a zero-arity primitive.
    """
    if max_library < 1:
        raise ValueError("max_library must be at least 1.")
    if weight <= 0:
        raise ValueError("Primitive weight must be greater than 0.")
    tree = expr if isinstance(expr, PrimitiveTree) else PrimitiveTree(expr)
    if index < 0 or index >= len(tree):
        raise IndexError(f"Subtree index {index} is outside the expression.")
    try:
        bound = tree.search_subtree(index)
    except IndexError as err:
        raise ValueError("The extracted nodes are not a complete typed tree.") from err
    if index == 0 and bound.stop != len(tree):
        raise ValueError("The extracted nodes are not a complete typed tree.")
    nodes = list(tree[bound])
    validate_subtree(nodes, prim_set)
    used = used_arguments(nodes, prim_set)
    if isinstance(prim_set, PrimitiveSet) and not used:
        raise ValueError("Untyped primitive sets require a primitive of arity at least 1.")
    in_types = [prim_set.mapping[name].ret for name in used]
    body, formals = rewrite_body(nodes, used)
    body = PrimitiveTree(expand_promoted(list(body), prim_set))
    func = compile_body(body, in_types, nodes[0].ret, prim_set)
    columnar = body_is_columnar(body, in_types, nodes[0].ret, prim_set)
    library = library_of(prim_set)
    detached: list[tuple[PromotedRecord, Any, Any]] = []
    name = ""
    try:
        while len(library.records) >= max_library:
            detached.append(detach_least_used(prim_set, library))
        name = next_promo_name(prim_set, library, prefix)
        register_promoted(prim_set, func, in_types, nodes[0].ret, name, weight)
        opcode = next_columnar_opcode(name) if columnar else None
        library.records[name] = PromotedRecord(
            name=name,
            uses=0,
            serial=library.serial,
            body=body,
            formals=formals,
            opcode=opcode,
        )
        library.generation += 1
        if opcode is not None:
            bind_numba_opcode(name, opcode)
        clear_compile_cache()
    except Exception:
        rollback_promote(prim_set, library, name, detached)
        raise
    return name

promoted_names(prim_set)

Return currently registered promoted names, oldest first.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Primitive set that may hold a promoted library.	required

Returns:

Type	Description
list[str]	Promoted names still on the set. Empty when none were added.

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

def promoted_names(prim_set: PrimitiveSetTyped) -> list[str]:
    """Return currently registered promoted names, oldest first.

    Args:
        prim_set: Primitive set that may hold a promoted library.

    Returns:
        Promoted names still on the set. Empty when none were added.
    """
    library = getattr(prim_set, "promoted_library", None)
    if library is None:
        return []
    return list(library.records)

register_gp(toolbox, pset, *, individual=None, min_depth=1, max_depth=2, mut_min_depth=0, mut_max_depth=2, height_limit=17, backend=None, select=True, contestants=3)

Register the standard tree-GP operators on toolbox.

Wires the aliases every prefix-tree run needs and that callers commonly omit: clone_individual instead of deepcopy, compile_tree, half-and-half initialization, one-point crossover, uniform mutation, and a height static_limit. Does not register evaluate or evaluate_batch — fitness stays on the caller.

Parameters:

Name	Type	Description	Default
toolbox	Toolbox	Toolbox to mutate.	required
pset	PrimitiveSetTyped	Primitive set used for init, compile, and mutation.	required
individual	type | None	Optional individual type. When given, registers expr, individual, and population.	None
min_depth	int	Minimum depth for half-and-half initialization.	1
max_depth	int	Maximum depth for half-and-half initialization.	2
mut_min_depth	int	Minimum depth of the subtree grown by mutation.	0
mut_max_depth	int	Maximum depth of the subtree grown by mutation.	2
height_limit	int | None	Height cap applied to mate and mutate. None skips the decorator.	17
backend	str | None	Optional compile_tree backend. The compile default is used when omitted.	None
select	bool | Callable[..., Any]	If True, register tournament selection. If False, leave select unset. A callable is registered as select instead.	True
contestants	int	Tournament size when select is True.	3

Returns:

Type	Description
Toolbox	The same toolbox, for chaining.

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

def register_gp(
    toolbox: Toolbox,
    pset: PrimitiveSetTyped,
    *,
    individual: type | None = None,
    min_depth: int = 1,
    max_depth: int = 2,
    mut_min_depth: int = 0,
    mut_max_depth: int = 2,
    height_limit: int | None = 17,
    backend: str | None = None,
    select: bool | Callable[..., Any] = True,
    contestants: int = 3,
) -> Toolbox:
    """Register the standard tree-GP operators on ``toolbox``.

    Wires the aliases every prefix-tree run needs and that callers
    commonly omit: ``clone_individual`` instead of ``deepcopy``,
    ``compile_tree``, half-and-half initialization, one-point
    crossover, uniform mutation, and a height ``static_limit``.
    Does not register ``evaluate`` or ``evaluate_batch`` — fitness
    stays on the caller.

    Args:
        toolbox: Toolbox to mutate.
        pset: Primitive set used for init, compile, and mutation.
        individual: Optional individual type. When given, registers
            ``expr``, ``individual``, and ``population``.
        min_depth: Minimum depth for half-and-half initialization.
        max_depth: Maximum depth for half-and-half initialization.
        mut_min_depth: Minimum depth of the subtree grown by mutation.
        mut_max_depth: Maximum depth of the subtree grown by mutation.
        height_limit: Height cap applied to ``mate`` and ``mutate``.
            ``None`` skips the decorator.
        backend: Optional ``compile_tree`` backend. The compile
            default is used when omitted.
        select: If True, register tournament selection. If False,
            leave ``select`` unset. A callable is registered as
            ``select`` instead.
        contestants: Tournament size when ``select`` is True.

    Returns:
        The same ``toolbox``, for chaining.
    """
    if individual is not None:
        toolbox.register(
            "expr",
            gen_half_and_half,
            prim_set=pset,
            min_depth=min_depth,
            max_depth=max_depth,
        )
        toolbox.register("individual", init_iterate, individual, toolbox.expr)
        toolbox.register("population", init_repeat, list, toolbox.individual)
    toolbox.register("clone", clone_individual)
    if backend is None:
        toolbox.register("compile", compile_tree, prim_set=pset)
    else:
        toolbox.register("compile", compile_tree, prim_set=pset, backend=backend)
    toolbox.register("mate", cx_one_point)
    toolbox.register("expr_mut", gen_full, min_depth=mut_min_depth, max_depth=mut_max_depth)
    toolbox.register("mutate", mut_uniform, expr=toolbox.expr_mut, prim_set=pset)
    if height_limit is not None:
        limit = static_limit(_HEIGHT, height_limit)
        toolbox.decorate("mate", limit)
        toolbox.decorate("mutate", limit)
    if select is True:
        toolbox.register("select", sel_tournament, contestants=contestants)
    elif select is not False:
        toolbox.register("select", select)
    return toolbox

cx_semantic(ind1, ind2, prim_set, min_depth=2, max_depth=6, gen_func=gen_grow)

Mate two individuals by a semantic crossover.

Parameters:

Name	Type	Description	Default
ind1	list[Any]	First individual to mate.	required
ind2	list[Any]	Second individual to mate.	required
prim_set	PrimitiveSetTyped	Primitive set used to build the random tree.	required
min_depth	int	Minimum depth of the random tree.	2
max_depth	int	Maximum depth of the random tree.	6
gen_func	Callable[..., Any]	Tree generator. Defaults to gen_grow.	gen_grow

Returns:

Type	Description
tuple[list[Any], list[Any]]	The two individuals after crossover.

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

def cx_semantic(
    ind1: list[Any],
    ind2: list[Any],
    prim_set: PrimitiveSetTyped,
    min_depth: int = 2,
    max_depth: int = 6,
    gen_func: Callable[..., Any] = gen_grow,
) -> tuple[list[Any], list[Any]]:
    """Mate two individuals by a semantic crossover.

    Args:
        ind1: First individual to mate.
        ind2: Second individual to mate.
        prim_set: Primitive set used to build the random tree.
        min_depth: Minimum depth of the random tree.
        max_depth: Maximum depth of the random tree.
        gen_func: Tree generator. Defaults to ``gen_grow``.

    Returns:
        The two individuals after crossover.
    """
    _check(prim_set, "crossover")

    tr = gen_func(prim_set, min_depth, max_depth)
    tr.insert(0, prim_set.mapping["lf"])

    def create_ind(ind: list[Any], ind_ext: list[Any]) -> list[Any]:
        new_ind = ind
        new_ind.insert(0, prim_set.mapping["mul"])
        new_ind.insert(0, prim_set.mapping["add"])
        new_ind.extend(tr)
        new_ind.append(prim_set.mapping["mul"])
        new_ind.append(prim_set.mapping["sub"])
        new_ind.append(Terminal(1.0, False, object))
        new_ind.extend(tr)
        new_ind.extend(ind_ext)
        return new_ind

    parent1 = list(ind1)
    parent2 = list(ind2)
    new_ind1 = create_ind(ind1, parent2)
    new_ind2 = create_ind(ind2, parent1)
    return new_ind1, new_ind2

mut_semantic(individual, prim_set, min_depth=2, max_depth=6, gen_func=None, mut_step=None)

Mutate an individual by a semantic mutation.

Parameters:

Name	Type	Description	Default
individual	list[Any]	Individual to mutate.	required
prim_set	PrimitiveSetTyped	Primitive set used to build the random trees.	required
min_depth	int	Minimum depth of each random tree.	2
max_depth	int	Maximum depth of each random tree.	6
gen_func	Callable[..., Any] | None	Tree generator. Defaults to gen_grow.	None
mut_step	float | None	Mutation step. Drawn uniformly from [0, 2] when omitted.	None

Returns:

Type	Description
tuple[list[Any]]	A one-element tuple containing the mutated individual.

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

def mut_semantic(
    individual: list[Any],
    prim_set: PrimitiveSetTyped,
    min_depth: int = 2,
    max_depth: int = 6,
    gen_func: Callable[..., Any] | None = None,
    mut_step: float | None = None,
) -> tuple[list[Any]]:
    """Mutate an individual by a semantic mutation.

    Args:
        individual: Individual to mutate.
        prim_set: Primitive set used to build the random trees.
        min_depth: Minimum depth of each random tree.
        max_depth: Maximum depth of each random tree.
        gen_func: Tree generator. Defaults to ``gen_grow``.
        mut_step: Mutation step. Drawn uniformly from ``[0, 2]``
            when omitted.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    _check(prim_set, "mutation")

    if gen_func is None:
        gen_func = gen_grow

    if mut_step is None:
        mut_step = rng.uniform(0, 2)

    new_ind = individual
    new_ind.insert(0, prim_set.mapping["add"])
    new_ind.extend(build_sig2_delta(prim_set, min_depth, max_depth, gen_func, mut_step))

    return (new_ind,)

cx_slim_donor(ind1, ind2, prim_set, *, best_donor=True)

Mate two SLIM individuals by donor crossover (XODn / XOBDn).

One delta block moves from the donor parent to the receiver. Offspring sizes stay close to the parents because one block is removed and one is added. When best_donor is True and both parents have valid fitness, the fitter parent donates (XOBDn); otherwise the donor is chosen uniformly (XODn). When the donor has no delta blocks, the parents are returned unchanged.

Parameters:

Name	Type	Description	Default
ind1	SlimTree	First parent SlimTree.	required
ind2	SlimTree	Second parent SlimTree.	required
prim_set	PrimitiveSetTyped	Primitive set checked for required semantic operators.	required
best_donor	bool	Prefer the fitter parent as donor when fitness is valid on both parents.	True

Returns:

Type	Description
tuple[SlimTree, SlimTree]	The two parents after crossover.

Source code in deap_er/private/programming/slim/slim_ops.py

def cx_slim_donor(
    ind1: SlimTree,
    ind2: SlimTree,
    prim_set: PrimitiveSetTyped,
    *,
    best_donor: bool = True,
) -> tuple[SlimTree, SlimTree]:
    """Mate two SLIM individuals by donor crossover (XODn / XOBDn).

    One delta block moves from the donor parent to the receiver.
    Offspring sizes stay close to the parents because one block is
    removed and one is added. When ``best_donor`` is True and both
    parents have valid fitness, the fitter parent donates (XOBDn);
    otherwise the donor is chosen uniformly (XODn). When the donor
    has no delta blocks, the parents are returned unchanged.

    Args:
        ind1: First parent ``SlimTree``.
        ind2: Second parent ``SlimTree``.
        prim_set: Primitive set checked for required semantic operators.
        best_donor: Prefer the fitter parent as donor when fitness is
            valid on both parents.

    Returns:
        The two parents after crossover.
    """
    _check(prim_set, "crossover")
    slim1 = _require_slim(ind1, "donor crossover")
    slim2 = _require_slim(ind2, "donor crossover")
    donor_idx = _pick_donor_index(slim1, slim2, best_donor)
    donor = slim1 if donor_idx == 0 else slim2
    receiver = slim2 if donor_idx == 0 else slim1
    if donor.deltas:
        block = donor.deltas.pop(rng.randrange(len(donor.deltas)))
        receiver.deltas.append(block)
    return slim1, slim2

mut_slim(individual, prim_set, *, inflate_prob=0.3, min_depth=2, max_depth=6, gen_func=None, mut_step=None)

Apply inflate or deflate mutation with fixed probability.

Parameters:

Name	Type	Description	Default
individual	SlimTree	SLIM genotype to mutate in place.	required
prim_set	PrimitiveSetTyped	Primitive set used by inflate mutation.	required
inflate_prob	float	Probability of inflate mutation. Deflate is used otherwise.	0.3
min_depth	int	Minimum depth of random trees for inflate.	2
max_depth	int	Maximum depth of random trees for inflate.	6
gen_func	Callable[..., Any] | None	Tree generator for inflate. Defaults to gen_grow.	None
mut_step	float | None	Mutation step for inflate. Drawn uniformly from [0, 2] when omitted.	None

Returns:

Type	Description
tuple[SlimTree]	A one-element tuple containing the mutated individual.

Source code in deap_er/private/programming/slim/slim_ops.py

def mut_slim(
    individual: SlimTree,
    prim_set: PrimitiveSetTyped,
    *,
    inflate_prob: float = 0.3,
    min_depth: int = 2,
    max_depth: int = 6,
    gen_func: Callable[..., Any] | None = None,
    mut_step: float | None = None,
) -> tuple[SlimTree]:
    """Apply inflate or deflate mutation with fixed probability.

    Args:
        individual: SLIM genotype to mutate in place.
        prim_set: Primitive set used by inflate mutation.
        inflate_prob: Probability of inflate mutation. Deflate is used
            otherwise.
        min_depth: Minimum depth of random trees for inflate.
        max_depth: Maximum depth of random trees for inflate.
        gen_func: Tree generator for inflate. Defaults to ``gen_grow``.
        mut_step: Mutation step for inflate. Drawn uniformly from
            ``[0, 2]`` when omitted.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    if rng.random() < inflate_prob:
        return mut_slim_inflate(
            individual,
            prim_set,
            min_depth=min_depth,
            max_depth=max_depth,
            gen_func=gen_func,
            mut_step=mut_step,
        )
    return mut_slim_deflate(individual)

mut_slim_deflate(individual)

Remove a random delta block (SLIM deflate mutation).

The head tree is never removed. When there are no delta blocks, the individual is returned unchanged.

Parameters:

Name	Type	Description	Default
individual	SlimTree	SLIM genotype to mutate in place.	required

Returns:

Type	Description
tuple[SlimTree]	A one-element tuple containing the mutated individual.

Source code in deap_er/private/programming/slim/slim_ops.py

def mut_slim_deflate(
    individual: SlimTree,
) -> tuple[SlimTree]:
    """Remove a random delta block (SLIM deflate mutation).

    The head tree is never removed. When there are no delta blocks,
    the individual is returned unchanged.

    Args:
        individual: SLIM genotype to mutate in place.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    slim = _require_slim(individual, "deflate mutation")
    if not slim.deltas:
        return (slim,)
    slim.deltas.pop(rng.randrange(len(slim.deltas)))
    return (slim,)

mut_slim_inflate(individual, prim_set, min_depth=2, max_depth=6, gen_func=None, mut_step=None)

Append a geometric semantic delta block (SLIM inflate mutation).

Parameters:

Name	Type	Description	Default
individual	SlimTree	SLIM genotype to mutate in place.	required
prim_set	PrimitiveSetTyped	Primitive set used to build the random trees.	required
min_depth	int	Minimum depth of each random tree.	2
max_depth	int	Maximum depth of each random tree.	6
gen_func	Callable[..., Any] | None	Tree generator. Defaults to gen_grow.	None
mut_step	float | None	Mutation step. Drawn uniformly from [0, 2] when omitted.	None

Returns:

Type	Description
tuple[SlimTree]	A one-element tuple containing the mutated individual.

Source code in deap_er/private/programming/slim/slim_ops.py

def mut_slim_inflate(
    individual: SlimTree,
    prim_set: PrimitiveSetTyped,
    min_depth: int = 2,
    max_depth: int = 6,
    gen_func: Callable[..., Any] | None = None,
    mut_step: float | None = None,
) -> tuple[SlimTree]:
    """Append a geometric semantic delta block (SLIM inflate mutation).

    Args:
        individual: SLIM genotype to mutate in place.
        prim_set: Primitive set used to build the random trees.
        min_depth: Minimum depth of each random tree.
        max_depth: Maximum depth of each random tree.
        gen_func: Tree generator. Defaults to ``gen_grow``.
        mut_step: Mutation step. Drawn uniformly from ``[0, 2]``
            when omitted.

    Returns:
        A one-element tuple containing the mutated individual.
    """
    _check(prim_set, "mutation")
    slim = _require_slim(individual, "inflate mutation")
    if gen_func is None:
        gen_func = gen_grow
    if mut_step is None:
        mut_step = rng.uniform(0, 2)
    slim.deltas.append(build_sig2_delta(prim_set, min_depth, max_depth, gen_func, mut_step))
    return (slim,)

compile_slim_tree(slim, prim_set, *, backend='python', dispatch=None)

Compile a SLIM individual for evaluation.

Parameters:

Name	Type	Description	Default
slim	SlimTree	SLIM genotype to compile.	required
prim_set	PrimitiveSetTyped	Primitive set that supplies the evaluation context.	required
backend	str	One of 'python', 'opcode', or 'numba'.	'python'
dispatch	Any	Compiled kernel for the 'numba' backend.	None

Returns:

Type	Description
Any	A callable when prim_set has one or more arguments,
Any	otherwise the evaluated scalar for the expression.

Source code in deap_er/private/programming/slim/slim_tree.py

def compile_slim_tree(
    slim: SlimTree,
    prim_set: PrimitiveSetTyped,
    *,
    backend: str = "python",
    dispatch: Any = None,
) -> Any:
    """Compile a SLIM individual for evaluation.

    Args:
        slim: SLIM genotype to compile.
        prim_set: Primitive set that supplies the evaluation context.
        backend: One of ``'python'``, ``'opcode'``, or ``'numba'``.
        dispatch: Compiled kernel for the ``'numba'`` backend.

    Returns:
        A callable when ``prim_set`` has one or more arguments,
        otherwise the evaluated scalar for the expression.
    """
    head_fn = compile_tree(slim.head, prim_set, backend=backend, dispatch=dispatch)
    if not slim.deltas:
        return head_fn
    delta_fns = [
        compile_tree(delta, prim_set, backend=backend, dispatch=dispatch) for delta in slim.deltas
    ]
    if len(prim_set.arguments) == 0:
        return head_fn() + sum(delta_fn() for delta_fn in delta_fns)

    def combined(*args: Any, **kwargs: Any) -> Any:
        total = head_fn(*args, **kwargs)
        for delta_fn in delta_fns:
            total += delta_fn(*args, **kwargs)
        return total

    return combined

suffix_rescore(tapes, matrix, prefix, *, n_new=None, lookback=None, backend='opcode', dispatch=None, parallel=False)

Rescore lookback + n_new trailing rows onto a cached prefix.

After an append-only vstack, pass the grown packed matrix and the scores from before the append. The helper keeps those prefix columns (new rows are the present) and writes only the last n_new outputs of a suffix interpret_tapes call. The full series matches a one-shot interpret_tapes on matrix, including warmup nan s.

A lookback smaller than :func:tape_lookback is rejected so a short suffix cannot silently drop history. ema is IIR: the helper scores the full pack so the recurrence matches the oracle.

Parameters:

Name	Type	Description	Default
tapes	Iterable[Tape]	Tapes in output-row order. A one-shot iterable is consumed once.	required
matrix	Any	Grown packed (n_rows, n_columns) table.	required
prefix	ndarray	Cached scores of shape (n_tapes, n_prefix_rows).	required
n_new	int | None	Rows appended after prefix. Inferred from the matrix and prefix when omitted.	None
lookback	int | None	History rows to keep in front of the new block. Defaults to the max :func:tape_lookback of tapes.	None
backend	str	Forwarded to :func:interpret_tapes.	'opcode'
dispatch	Any	Forwarded to :func:interpret_tapes.	None
parallel	bool	Forwarded to :func:interpret_tapes.	False

Returns:

Type	Description
ndarray	A new (n_tapes, n_rows) series. It does not alias
ndarray	prefix or matrix.

Raises:

Type	Description
ValueError	If n_new or lookback is invalid, if prefix does not line up with the tapes and matrix, or if lookback is below a tape's bound.

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

def suffix_rescore(
    tapes: Iterable[Tape],
    matrix: Any,
    prefix: numpy.ndarray,
    *,
    n_new: int | None = None,
    lookback: int | None = None,
    backend: str = "opcode",
    dispatch: Any = None,
    parallel: bool = False,
) -> numpy.ndarray:
    """Rescore ``lookback + n_new`` trailing rows onto a cached prefix.

    After an append-only ``vstack``, pass the grown packed matrix and
    the scores from before the append. The helper keeps those prefix
    columns (new rows are the present) and writes only the last
    ``n_new`` outputs of a suffix ``interpret_tapes`` call. The full
    series matches a one-shot ``interpret_tapes`` on ``matrix``,
    including warmup ``nan`` s.

    A lookback smaller than :func:`tape_lookback` is rejected so a
    short suffix cannot silently drop history. ``ema`` is IIR: the
    helper scores the full pack so the recurrence matches the oracle.

    Args:
        tapes: Tapes in output-row order. A one-shot iterable is
            consumed once.
        matrix: Grown packed ``(n_rows, n_columns)`` table.
        prefix: Cached scores of shape ``(n_tapes, n_prefix_rows)``.
        n_new: Rows appended after ``prefix``. Inferred from the
            matrix and prefix when omitted.
        lookback: History rows to keep in front of the new block.
            Defaults to the max :func:`tape_lookback` of ``tapes``.
        backend: Forwarded to :func:`interpret_tapes`.
        dispatch: Forwarded to :func:`interpret_tapes`.
        parallel: Forwarded to :func:`interpret_tapes`.

    Returns:
        A new ``(n_tapes, n_rows)`` series. It does not alias
        ``prefix`` or ``matrix``.

    Raises:
        ValueError: If ``n_new`` or ``lookback`` is invalid, if
            ``prefix`` does not line up with the tapes and matrix, or
            if ``lookback`` is below a tape's bound.
    """
    tapes = tuple(tapes)
    packed = _as_grown_matrix(matrix)
    rows = packed.shape[0]
    added = _new_rows(n_new, rows, prefix)
    bound = _required_lookback(tapes)
    used = bound if lookback is None else lookback
    if used < 0:
        raise ValueError(f"lookback must be at least 0, got {used}.")
    if used < bound:
        raise ValueError(_SHORT_LOOKBACK.format(lookback=used, bound=bound))
    _check_prefix(prefix, len(tapes), rows - added)
    span = rows if _has_ema(tapes) else min(rows, used + added)
    scored = interpret_tapes(
        tapes, packed[-span:], backend=backend, dispatch=dispatch, parallel=parallel
    )
    out = numpy.empty((len(tapes), rows), dtype=numpy.float64)
    kept = rows - added
    if kept:
        out[:, :kept] = numpy.asarray(prefix, dtype=numpy.float64)[:, :kept]
    if added:
        out[:, kept:] = scored[:, -added:]
    return out

interpret_tapes(tapes, matrix, *, backend='opcode', dispatch=None, parallel=False)

Run many tapes over one packed column matrix.

The result has one row per tape and one column per sample. It is a new array: it does not alias matrix or the interpreter workspace. Cache unique programs by str(tree). A stringified Window leaf is an int; from_string restores that tag, but an ephemeral class identity is not part of the text.

The 'opcode' backend unpacks the matrix columns once and runs the NumPy stack machine with common-subexpression elimination across the batch. The 'numba' backend uses the same CSE plan on the serial path; parallel=True keeps the compiled per-tape loop. parallel=True is not available on the opcode backend.

Parameters:

Name	Type	Description	Default
tapes	Iterable[Tape]	Tapes produced by lower_tree, in the order of the output rows. A one-shot iterable is consumed once.	required
matrix	Any	Packed (n_rows, n_columns) column table. A sequence of columns is rejected.	required
backend	str	Either 'opcode' or 'numba'.	'opcode'
dispatch	Any	Compiled kernel that implements the consumer opcodes of the 'numba' backend. Ignored by 'opcode'. With parallel=True it must be safe on several stacks at once — no process-global buffer.	None
parallel	bool	If True, run the Numba path with one workspace per thread. Requires backend='numba'.	False

Returns:

Type	Description
ndarray	A C-contiguous float64 array of shape
ndarray	(len(tapes), n_rows).

Raises:

Type	Description
ValueError	If the backend is unknown, if parallel is set on the opcode backend, if matrix is not a packed table, or if a tape does not match the matrix.
ImportError	If backend='numba' and the numba extra is not installed.

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

def interpret_tapes(
    tapes: Iterable[Tape],
    matrix: Any,
    *,
    backend: str = "opcode",
    dispatch: Any = None,
    parallel: bool = False,
) -> numpy.ndarray:
    """Run many tapes over one packed column matrix.

    The result has one row per tape and one column per sample. It is a
    new array: it does not alias ``matrix`` or the interpreter
    workspace. Cache unique programs by ``str(tree)``. A stringified
    ``Window`` leaf is an ``int``; ``from_string`` restores that tag,
    but an ephemeral class identity is not part of the text.

    The ``'opcode'`` backend unpacks the matrix columns once and runs
    the NumPy stack machine with common-subexpression elimination
    across the batch. The ``'numba'`` backend uses the same CSE plan
    on the serial path; ``parallel=True`` keeps the compiled per-tape
    loop. ``parallel=True`` is not available on the opcode backend.

    Args:
        tapes: Tapes produced by ``lower_tree``, in the order of the
            output rows. A one-shot iterable is consumed once.
        matrix: Packed ``(n_rows, n_columns)`` column table. A sequence
            of columns is rejected.
        backend: Either ``'opcode'`` or ``'numba'``.
        dispatch: Compiled kernel that implements the consumer opcodes
            of the ``'numba'`` backend. Ignored by ``'opcode'``.
            With ``parallel=True`` it must be safe on several stacks
            at once — no process-global buffer.
        parallel: If True, run the Numba path with one workspace per
            thread. Requires ``backend='numba'``.

    Returns:
        A C-contiguous ``float64`` array of shape
        ``(len(tapes), n_rows)``.

    Raises:
        ValueError: If the backend is unknown, if ``parallel`` is set
            on the opcode backend, if ``matrix`` is not a packed
            table, or if a tape does not match the matrix.
        ImportError: If ``backend='numba'`` and the ``numba`` extra
            is not installed.
    """
    tapes = tuple(tapes)
    packed = _as_batch_matrix(matrix)
    _check_tapes(tapes, packed.shape[1])
    if backend == "opcode":
        if parallel:
            raise ValueError("parallel=True requires backend='numba'.")
        return _run_opcode(tapes, packed)
    if backend == "numba":
        try:
            # Optional extra: imported only when the Numba batch path is asked for.
            from .numba.numba_batch import run_tapes
        except ImportError as err:
            raise ImportError(_NUMBA_MISSING) from err
        return run_tapes(tapes, packed, dispatch=dispatch, parallel=parallel)
    raise ValueError(f"Unknown compile backend '{backend}'. Use 'opcode' or 'numba'.")

bounds_from_matrix(matrix)

Return empirical (low, high) bounds for each matrix column.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Packed (n_rows, n_columns) table.	required

Returns:

Type	Description
ndarray	A (n_columns, 2) float64 array of column bounds.

Raises:

Type	Description
ValueError	If matrix is not two-dimensional.

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

def bounds_from_matrix(matrix: numpy.ndarray | Sequence[Sequence[float]]) -> numpy.ndarray:
    """Return empirical ``(low, high)`` bounds for each matrix column.

    Args:
        matrix: Packed ``(n_rows, n_columns)`` table.

    Returns:
        A ``(n_columns, 2)`` ``float64`` array of column bounds.

    Raises:
        ValueError: If ``matrix`` is not two-dimensional.
    """
    packed = numpy.asarray(matrix, dtype=numpy.float64)
    if packed.ndim != 2:
        raise ValueError("matrix must be a two-dimensional array")
    bounds = numpy.empty((packed.shape[1], 2), dtype=numpy.float64)
    for column in range(packed.shape[1]):
        series = packed[:, column]
        bounds[column, 0] = numpy.nanmin(series)
        bounds[column, 1] = numpy.nanmax(series)
    return bounds

tape_flags(tape, column_bounds, *, n_rows)

Return static certificates for a tape before scoring.

Parameters:

Name	Type	Description	Default
tape	Tape	Tape produced by lower_tree.	required
column_bounds	ndarray | Sequence[tuple[float, float]]	Per-column (low, high) bounds.	required
n_rows	int	Row count of the matrix the tape would run on.	required

Returns:

Type	Description
TapeFlags	Flags for identically nan, constant, or warmup-hiding programs.

Raises:

Type	Description
ValueError	If n_rows is negative, the tape is malformed, or the tape holds a consumer opcode.

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

def tape_flags(
    tape: Tape,
    column_bounds: numpy.ndarray | Sequence[tuple[float, float]],
    *,
    n_rows: int,
) -> TapeFlags:
    """Return static certificates for a tape before scoring.

    Args:
        tape: Tape produced by ``lower_tree``.
        column_bounds: Per-column ``(low, high)`` bounds.
        n_rows: Row count of the matrix the tape would run on.

    Returns:
        Flags for identically ``nan``, constant, or warmup-hiding programs.

    Raises:
        ValueError: If ``n_rows`` is negative, the tape is malformed, or
            the tape holds a consumer opcode.
    """
    if n_rows < 0:
        raise ValueError("n_rows must be at least 0.")
    walked = walk_tape(tape, normalize_bounds(column_bounds, tape.columns))
    summary = walked.summary
    scorable = summary.can_finite and summary.first_finite < n_rows
    return TapeFlags(
        all_nan=not scorable,
        constant=scorable and summary.const and summary.lo == summary.hi,
        hides_warmup=walked.warmup_hidden,
    )

tape_interval(tape, column_bounds)

Return a conservative output interval for a tape.

Parameters:

Name	Type	Description	Default
tape	Tape	Tape produced by lower_tree.	required
column_bounds	ndarray | Sequence[tuple[float, float]]	Per-column (low, high) bounds.	required

Returns:

Type	Description
tuple[float, float]	The (low, high) envelope of finite outputs.

Raises:

Type	Description
ValueError	If the tape is malformed or holds a consumer opcode.

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

def tape_interval(
    tape: Tape,
    column_bounds: numpy.ndarray | Sequence[tuple[float, float]],
) -> tuple[float, float]:
    """Return a conservative output interval for a tape.

    Args:
        tape: Tape produced by ``lower_tree``.
        column_bounds: Per-column ``(low, high)`` bounds.

    Returns:
        The ``(low, high)`` envelope of finite outputs.

    Raises:
        ValueError: If the tape is malformed or holds a consumer opcode.
    """
    walked = walk_tape(tape, normalize_bounds(column_bounds, tape.columns))
    return walked.summary.lo, walked.summary.hi

tape_skip_score(tape, column_bounds, *, n_rows)

Return whether evaluate_columnar should skip scoring a tape.

Parameters:

Name	Type	Description	Default
tape	Tape	Tape produced by lower_tree.	required
column_bounds	ndarray | Sequence[tuple[float, float]]	Per-column (low, high) bounds.	required
n_rows	int	Row count of the matrix the tape would run on.	required

Returns:

Type	Description
bool	True when any static certificate fires.

Raises:

Type	Description
ValueError	If n_rows is negative, the tape is malformed, or the tape holds a consumer opcode.

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

def tape_skip_score(
    tape: Tape,
    column_bounds: numpy.ndarray | Sequence[tuple[float, float]],
    *,
    n_rows: int,
) -> bool:
    """Return whether ``evaluate_columnar`` should skip scoring a tape.

    Args:
        tape: Tape produced by ``lower_tree``.
        column_bounds: Per-column ``(low, high)`` bounds.
        n_rows: Row count of the matrix the tape would run on.

    Returns:
        ``True`` when any static certificate fires.

    Raises:
        ValueError: If ``n_rows`` is negative, the tape is malformed, or
            the tape holds a consumer opcode.
    """
    return tape_flags(tape, column_bounds, n_rows=n_rows).skip_score

tape_lookback(tape)

Return the causal lookback bound of a tape.

Walks the postfix tape and composes per-opcode lookbacks: windowed instructions add their bound to the child, and pointwise instructions take the max of their arguments. The result is the shortest prefix of earlier rows that a legal suffix rescore must keep as history.

Parameters:

Name	Type	Description	Default
tape	Tape	Tape produced by lower_tree.	required

Returns:

Type	Description
int	The program's lookback in rows. 0 for a pointwise tape.

Raises:

Type	Description
ValueError	If the tape underflows, leaves no result, or holds a consumer opcode.

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

def tape_lookback(tape: Tape) -> int:
    """Return the causal lookback bound of a tape.

    Walks the postfix tape and composes per-opcode lookbacks: windowed
    instructions add their bound to the child, and pointwise
    instructions take the max of their arguments. The result is the
    shortest prefix of earlier rows that a legal suffix rescore must
    keep as history.

    Args:
        tape: Tape produced by ``lower_tree``.

    Returns:
        The program's lookback in rows. ``0`` for a pointwise tape.

    Raises:
        ValueError: If the tape underflows, leaves no result, or holds
            a consumer opcode.
    """
    stack: list[int] = []
    for step in range(tape.opcodes.size):
        opcode = int(tape.opcodes[step])
        extra = opcode_lookback(opcode, int(tape.operands[step]))
        arity = OPCODES_ARITY.get(opcode)
        if opcode in {int(Opcode.COL_LOAD), int(Opcode.CONST)}:
            stack.append(0)
            continue
        if arity == 1:
            stack[-1] = _peek(stack) + extra
            continue
        if arity == 2:
            right = _pop(stack)
            stack[-1] = max(_peek(stack), right) + extra
            continue
        if arity == 3:
            on_false = _pop(stack)
            on_true = _pop(stack)
            stack[-1] = max(_peek(stack), on_true, on_false)
            continue
        raise ValueError(f"Opcode {opcode} has no lookback certificate.")
    if not stack:
        raise ValueError("The tape is malformed and leaves no result.")
    return stack[-1]

add_window_ephemeral(prim_set, name, low, high)

Register a random window length as an ephemeral constant.

Each tree samples its own immutable window length from the closed interval [low, high]. The sampler is memoized by name, because a primitive set rejects two ephemerals that share a name but not a function.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Typed primitive set to register on.	required
name	str	Name of this ephemeral type. Must be unique across every primitive set in the process.	required
low	int	Smallest window length that may be sampled. At least 1.	required
high	int	Largest window length that may be sampled.	required

Raises:

Type	Description
ValueError	If the bounds are invalid, or if name was already used with different bounds.

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

def add_window_ephemeral(prim_set: PrimitiveSetTyped, name: str, low: int, high: int) -> None:
    """Register a random window length as an ephemeral constant.

    Each tree samples its own immutable window length from the closed
    interval ``[low, high]``. The sampler is memoized by name, because
    a primitive set rejects two ephemerals that share a name but not a
    function.

    Args:
        prim_set: Typed primitive set to register on.
        name: Name of this ephemeral type. Must be unique across every
            primitive set in the process.
        low: Smallest window length that may be sampled. At least 1.
        high: Largest window length that may be sampled.

    Raises:
        ValueError: If the bounds are invalid, or if ``name`` was
            already used with different bounds.
    """
    if low < 1:
        raise ValueError(f"The lowest window length must be at least 1, got {low}.")
    if high < low:
        raise ValueError(f"Window bounds are inverted: [{low}, {high}].")

    known = _samplers.get(name)
    if known is None:

        def sampler() -> int:
            return rng.randint(low, high)

        _samplers[name] = (low, high, sampler)
    elif known[:2] != (low, high):
        raise ValueError(
            f"The window ephemeral '{name}' was already registered with "
            f"bounds [{known[0]}, {known[1]}]. Use a different name."
        )

    draw = _samplers[name][2]
    setattr(draw, "low", low)  # noqa: B010
    setattr(draw, "high", high)  # noqa: B010
    prim_set.add_ephemeral_constant(name, draw, Window)

add_window_primitives(prim_set)

Register the causal window primitive kit on a typed primitive set.

Every primitive takes an Array and a Window and returns an Array. All of them are causal: an output sample is a function of that sample and earlier ones only, and samples without enough history are nan rather than zero, so a consumer can mask the warmup instead of trading on fabricated values.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Typed primitive set built by make_column_pset or an equivalent set over Array and Window.	required

Raises:

Type	Description
ValueError	If a primitive name collides with an argument of prim_set, or if a name is already registered.

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

def add_window_primitives(prim_set: PrimitiveSetTyped) -> None:
    """Register the causal window primitive kit on a typed primitive set.

    Every primitive takes an ``Array`` and a ``Window`` and returns an
    ``Array``. All of them are causal: an output sample is a function
    of that sample and earlier ones only, and samples without enough
    history are ``nan`` rather than zero, so a consumer can mask the
    warmup instead of trading on fabricated values.

    Args:
        prim_set: Typed primitive set built by ``make_column_pset`` or
            an equivalent set over ``Array`` and ``Window``.

    Raises:
        ValueError: If a primitive name collides with an argument of
            ``prim_set``, or if a name is already registered.
    """
    windowed: dict[str, Callable[..., Any]] = {
        "delay": delay,
        "diff": diff,
        "rolling_sum": rolling_sum,
        "rolling_mean": rolling_mean,
        "rolling_std": rolling_std,
        "rolling_min": rolling_min,
        "rolling_max": rolling_max,
        "ema": ema,
    }
    reject_shadowed(prim_set, list(windowed))

    in_types: list[type] = [Array, Window]
    for name, func in windowed.items():
        prim_set.add_primitive(func, in_types, Array, name)

ema(value, window)

Take a causal exponential moving average of a series.

The recurrence is y[t] = a * x[t] + (1 - a) * y[t - 1] with a = 2 / (window + 1), seeded so that y equals x at the first finite sample. The first window - 1 samples after that seed are reported as nan to match the rolling primitives. A nan inside the series propagates to every later sample.

Parameters:

Name	Type	Description	Default
value	Any	Series to filter.	required
window	Any	Span of the average. At least 1.	required

Returns:

Type	Description
ndarray	The exponential moving average.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def ema(value: Any, window: Any) -> numpy.ndarray:
    """Take a causal exponential moving average of a series.

    The recurrence is ``y[t] = a * x[t] + (1 - a) * y[t - 1]`` with
    ``a = 2 / (window + 1)``, seeded so that ``y`` equals ``x`` at the
    first finite sample. The first ``window - 1`` samples after that
    seed are reported as ``nan`` to match the rolling primitives. A
    ``nan`` inside the series propagates to every later sample.

    Args:
        value: Series to filter.
        window: Span of the average. At least 1.

    Returns:
        The exponential moving average.

    Raises:
        ValueError: If the window length is less than 1.
    """
    # Deferred so that importing deap_er does not pull in scipy.signal,
    # which no other part of the package needs.
    from scipy.signal import lfilter, lfilter_zi

    series, length = as_series(value, window)
    result = numpy.full(series.shape, numpy.nan, dtype=numpy.float64)
    if length > series.size:
        return result

    finite = numpy.flatnonzero(numpy.isfinite(series))
    if finite.size == 0:
        return result

    start = int(finite[0])
    tail = series[start:]
    alpha = 2.0 / (length + 1.0)
    numer = numpy.array([alpha], dtype=numpy.float64)
    denom = numpy.array([1.0, alpha - 1.0], dtype=numpy.float64)
    state = lfilter_zi(numer, denom) * tail[0]
    filtered, _ = lfilter(numer, denom, tail, zi=state)

    result[start:] = filtered
    result[: start + length - 1] = numpy.nan
    return result

add_pair_window_primitives(prim_set)

Register the two-input causal window kit on a typed primitive set.

Every primitive takes two Array arguments and a Window and returns an Array. All of them are causal: an output sample is a function of that sample and earlier ones only, and samples without enough history are nan.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Typed primitive set built by make_column_pset or an equivalent set over Array and Window.	required

Raises:

Type	Description
ValueError	If a primitive name collides with an argument of prim_set, or if a name is already registered.

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

def add_pair_window_primitives(prim_set: PrimitiveSetTyped) -> None:
    """Register the two-input causal window kit on a typed primitive set.

    Every primitive takes two ``Array`` arguments and a ``Window`` and
    returns an ``Array``. All of them are causal: an output sample is a
    function of that sample and earlier ones only, and samples without
    enough history are ``nan``.

    Args:
        prim_set: Typed primitive set built by ``make_column_pset`` or
            an equivalent set over ``Array`` and ``Window``.

    Raises:
        ValueError: If a primitive name collides with an argument of
            ``prim_set``, or if a name is already registered.
    """
    windowed: dict[str, Callable[..., Any]] = {
        "rolling_corr": rolling_corr,
        "rolling_cov": rolling_cov,
        "rolling_beta": rolling_beta,
    }
    reject_shadowed(prim_set, list(windowed))

    in_types: list[type] = [Array, Array, Window]
    for name, func in windowed.items():
        prim_set.add_primitive(func, in_types, Array, name)

rolling_beta(left, right, window)

Take the OLS slope of left on right over a trailing window.

The slope is the population covariance divided by the population variance of right. A window where right is constant is nan. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
left	Any	Dependent series.	required
right	Any	Independent series.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The rolling beta of left versus right.

Raises:

Type	Description
ValueError	If the window length is less than 1, or if the operands are not one-dimensional series of the same length.

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

def rolling_beta(left: Any, right: Any, window: Any) -> numpy.ndarray:
    """Take the OLS slope of ``left`` on ``right`` over a trailing window.

    The slope is the population covariance divided by the population
    variance of ``right``. A window where ``right`` is constant is
    ``nan``. The first ``window - 1`` samples are ``nan``.

    Args:
        left: Dependent series.
        right: Independent series.
        window: Trailing window length. At least 1.

    Returns:
        The rolling beta of ``left`` versus ``right``.

    Raises:
        ValueError: If the window length is less than 1, or if the
            operands are not one-dimensional series of the same length.
    """
    result, length, cov, _, var_y = pair_moments(left, right, window)
    if cov is None or var_y is None:
        return result
    with numpy.errstate(invalid="ignore", divide="ignore"):
        values = cov / var_y
    values[var_y <= 0.0] = numpy.nan
    result[length - 1 :] = values
    return result

rolling_corr(left, right, window)

Take the population correlation of two series over a trailing window.

The window is [t - window + 1, t] inclusive. A window where either series has zero variance is nan. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
left	Any	First series.	required
right	Any	Second series.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The rolling correlation.

Raises:

Type	Description
ValueError	If the window length is less than 1, or if the operands are not one-dimensional series of the same length.

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

def rolling_corr(left: Any, right: Any, window: Any) -> numpy.ndarray:
    """Take the population correlation of two series over a trailing window.

    The window is ``[t - window + 1, t]`` inclusive. A window where
    either series has zero variance is ``nan``. The first
    ``window - 1`` samples are ``nan``.

    Args:
        left: First series.
        right: Second series.
        window: Trailing window length. At least 1.

    Returns:
        The rolling correlation.

    Raises:
        ValueError: If the window length is less than 1, or if the
            operands are not one-dimensional series of the same length.
    """
    result, length, cov, var_x, var_y = pair_moments(left, right, window)
    if cov is None or var_x is None or var_y is None:
        return result
    denom = numpy.sqrt(var_x * var_y)
    with numpy.errstate(invalid="ignore", divide="ignore"):
        values = cov / denom
    values[denom <= 0.0] = numpy.nan
    result[length - 1 :] = values
    return result

rolling_cov(left, right, window)

Take the population covariance of two series over a trailing window.

The window is [t - window + 1, t] inclusive and the divisor is the window length. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
left	Any	First series.	required
right	Any	Second series.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The rolling covariance.

Raises:

Type	Description
ValueError	If the window length is less than 1, or if the operands are not one-dimensional series of the same length.

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

def rolling_cov(left: Any, right: Any, window: Any) -> numpy.ndarray:
    """Take the population covariance of two series over a trailing window.

    The window is ``[t - window + 1, t]`` inclusive and the divisor is
    the window length. The first ``window - 1`` samples are ``nan``.

    Args:
        left: First series.
        right: Second series.
        window: Trailing window length. At least 1.

    Returns:
        The rolling covariance.

    Raises:
        ValueError: If the window length is less than 1, or if the
            operands are not one-dimensional series of the same length.
    """
    result, length, cov, _, _ = pair_moments(left, right, window)
    if cov is not None:
        result[length - 1 :] = cov
    return result

rolling_max(value, window)

Take the maximum of a trailing window of a series.

The window is [t - window + 1, t] inclusive. A window that holds a nan reduces to nan. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to reduce.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The rolling maximum.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def rolling_max(value: Any, window: Any) -> numpy.ndarray:
    """Take the maximum of a trailing window of a series.

    The window is ``[t - window + 1, t]`` inclusive. A window that
    holds a ``nan`` reduces to ``nan``. The first ``window - 1``
    samples are ``nan``.

    Args:
        value: Series to reduce.
        window: Trailing window length. At least 1.

    Returns:
        The rolling maximum.

    Raises:
        ValueError: If the window length is less than 1.
    """
    result, length, reduced = rolling(value, window, numpy.maximum)
    if reduced is not None:
        result[length - 1 :] = reduced
    return result

rolling_mean(value, window)

Average a trailing window of a series.

The window is [t - window + 1, t] inclusive. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to reduce.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The rolling mean.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def rolling_mean(value: Any, window: Any) -> numpy.ndarray:
    """Average a trailing window of a series.

    The window is ``[t - window + 1, t]`` inclusive. The first
    ``window - 1`` samples are ``nan``.

    Args:
        value: Series to reduce.
        window: Trailing window length. At least 1.

    Returns:
        The rolling mean.

    Raises:
        ValueError: If the window length is less than 1.
    """
    result, length, reduced = rolling(value, window, numpy.add)
    if reduced is not None:
        result[length - 1 :] = reduced / length
    return result

rolling_min(value, window)

Take the minimum of a trailing window of a series.

The window is [t - window + 1, t] inclusive. A window that holds a nan reduces to nan. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to reduce.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The rolling minimum.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def rolling_min(value: Any, window: Any) -> numpy.ndarray:
    """Take the minimum of a trailing window of a series.

    The window is ``[t - window + 1, t]`` inclusive. A window that
    holds a ``nan`` reduces to ``nan``. The first ``window - 1``
    samples are ``nan``.

    Args:
        value: Series to reduce.
        window: Trailing window length. At least 1.

    Returns:
        The rolling minimum.

    Raises:
        ValueError: If the window length is less than 1.
    """
    result, length, reduced = rolling(value, window, numpy.minimum)
    if reduced is not None:
        result[length - 1 :] = reduced
    return result

rolling_std(value, window)

Take the population standard deviation of a trailing window.

The window is [t - window + 1, t] inclusive and the divisor is the window length. The first window - 1 samples are nan.

Each window is centered on its own mean before squaring, so the result stays accurate however far the samples sit from zero. The windows are centered in row blocks, which keeps peak memory proportional to the block rather than to the whole series. A window that holds an infinity is nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to reduce.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The rolling standard deviation.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def rolling_std(value: Any, window: Any) -> numpy.ndarray:
    """Take the population standard deviation of a trailing window.

    The window is ``[t - window + 1, t]`` inclusive and the divisor is
    the window length. The first ``window - 1`` samples are ``nan``.

    Each window is centered on its own mean before squaring, so the
    result stays accurate however far the samples sit from zero. The
    windows are centered in row blocks, which keeps peak memory
    proportional to the block rather than to the whole series. A window
    that holds an infinity is ``nan``.

    Args:
        value: Series to reduce.
        window: Trailing window length. At least 1.

    Returns:
        The rolling standard deviation.

    Raises:
        ValueError: If the window length is less than 1.
    """
    series, length = as_series(value, window)
    result = numpy.full(series.shape, numpy.nan, dtype=numpy.float64)
    if length > series.size:
        return result
    view = sliding_window_view(series, length)
    output = result[length - 1 :]
    for start in range(0, view.shape[0], CHUNK_ROWS):
        block = view[start : start + CHUNK_ROWS]
        deviations = window_deviations(block, length)
        with numpy.errstate(invalid="ignore"):
            residue = numpy.add.reduce(deviations, axis=-1) / length
            squares = numpy.add.reduce(deviations * deviations, axis=-1) / length
            variance = squares - residue * residue
        numpy.maximum(variance, 0.0, out=variance)
        output[start : start + CHUNK_ROWS] = numpy.sqrt(variance)
    return result

rolling_sum(value, window)

Sum a trailing window of a series.

The window is [t - window + 1, t] inclusive. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to reduce.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The rolling sum.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def rolling_sum(value: Any, window: Any) -> numpy.ndarray:
    """Sum a trailing window of a series.

    The window is ``[t - window + 1, t]`` inclusive. The first
    ``window - 1`` samples are ``nan``.

    Args:
        value: Series to reduce.
        window: Trailing window length. At least 1.

    Returns:
        The rolling sum.

    Raises:
        ValueError: If the window length is less than 1.
    """
    result, length, reduced = rolling(value, window, numpy.add)
    if reduced is not None:
        result[length - 1 :] = reduced
    return result

delay(value, window)

Shift a series into the past by window samples.

y[t] is x[t - window]. The first window samples have no past to read and are nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to shift.	required
window	Any	Number of samples to shift by. At least 1.	required

Returns:

Type	Description
ndarray	The shifted series.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def delay(value: Any, window: Any) -> numpy.ndarray:
    """Shift a series into the past by ``window`` samples.

    ``y[t]`` is ``x[t - window]``. The first ``window`` samples have no
    past to read and are ``nan``.

    Args:
        value: Series to shift.
        window: Number of samples to shift by. At least 1.

    Returns:
        The shifted series.

    Raises:
        ValueError: If the window length is less than 1.
    """
    series, length = as_series(value, window)
    result = numpy.full(series.shape, numpy.nan, dtype=numpy.float64)
    if length < series.size:
        result[length:] = series[:-length]
    return result

diff(value, window)

Subtract a delayed copy of a series from itself.

y[t] is x[t] - x[t - window]. The first window samples are nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to difference.	required
window	Any	Number of samples to look back. At least 1.	required

Returns:

Type	Description
ndarray	The differenced series.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def diff(value: Any, window: Any) -> numpy.ndarray:
    """Subtract a delayed copy of a series from itself.

    ``y[t]`` is ``x[t] - x[t - window]``. The first ``window`` samples
    are ``nan``.

    Args:
        value: Series to difference.
        window: Number of samples to look back. At least 1.

    Returns:
        The differenced series.

    Raises:
        ValueError: If the window length is less than 1.
    """
    series, length = as_series(value, window)
    return series - delay(series, length)

add_ts_primitives(prim_set)

Register the causal time-series unary kit on a typed primitive set.

Every primitive takes an Array and a Window and returns an Array. All of them are causal: an output sample is a function of that sample and earlier ones only, and samples without enough history are nan.

Parameters:

Name	Type	Description	Default
prim_set	PrimitiveSetTyped	Typed primitive set built by make_column_pset or an equivalent set over Array and Window.	required

Raises:

Type	Description
ValueError	If a primitive name collides with an argument of prim_set, or if a name is already registered.

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

def add_ts_primitives(prim_set: PrimitiveSetTyped) -> None:
    """Register the causal time-series unary kit on a typed primitive set.

    Every primitive takes an ``Array`` and a ``Window`` and returns an
    ``Array``. All of them are causal: an output sample is a function
    of that sample and earlier ones only, and samples without enough
    history are ``nan``.

    Args:
        prim_set: Typed primitive set built by ``make_column_pset`` or
            an equivalent set over ``Array`` and ``Window``.

    Raises:
        ValueError: If a primitive name collides with an argument of
            ``prim_set``, or if a name is already registered.
    """
    windowed: dict[str, Callable[..., Any]] = {
        "ts_rank": ts_rank,
        "ts_argmax": ts_argmax,
        "ts_argmin": ts_argmin,
    }
    reject_shadowed(prim_set, list(windowed))

    in_types: list[type] = [Array, Window]
    for name, func in windowed.items():
        prim_set.add_primitive(func, in_types, Array, name)

ts_argmax(value, window)

Take how many samples ago the trailing-window maximum occurred.

The window is [t - window + 1, t] inclusive. 0 means the current sample is the maximum. A tie keeps the most recent maximum. A window that holds a nan is nan. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to search.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The age of the window maximum.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def ts_argmax(value: Any, window: Any) -> numpy.ndarray:
    """Take how many samples ago the trailing-window maximum occurred.

    The window is ``[t - window + 1, t]`` inclusive. ``0`` means the
    current sample is the maximum. A tie keeps the most recent
    maximum. A window that holds a ``nan`` is ``nan``. The first
    ``window - 1`` samples are ``nan``.

    Args:
        value: Series to search.
        window: Trailing window length. At least 1.

    Returns:
        The age of the window maximum.

    Raises:
        ValueError: If the window length is less than 1.
    """
    return _ts_arg(value, window, numpy.argmax)

ts_argmin(value, window)

Take how many samples ago the trailing-window minimum occurred.

The window is [t - window + 1, t] inclusive. 0 means the current sample is the minimum. A tie keeps the most recent minimum. A window that holds a nan is nan. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to search.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The age of the window minimum.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def ts_argmin(value: Any, window: Any) -> numpy.ndarray:
    """Take how many samples ago the trailing-window minimum occurred.

    The window is ``[t - window + 1, t]`` inclusive. ``0`` means the
    current sample is the minimum. A tie keeps the most recent
    minimum. A window that holds a ``nan`` is ``nan``. The first
    ``window - 1`` samples are ``nan``.

    Args:
        value: Series to search.
        window: Trailing window length. At least 1.

    Returns:
        The age of the window minimum.

    Raises:
        ValueError: If the window length is less than 1.
    """
    return _ts_arg(value, window, numpy.argmin)

ts_rank(value, window)

Take the percentile rank of the current sample in a trailing window.

The window is [t - window + 1, t] inclusive. The value is the 1-based average rank of x[t] among those samples, scaled by (rank - 1) / (window - 1). A unique window low is 0.0, a unique window high is 1.0, and an all-tie window is 0.5. A window of 1 is nan. A window that holds a nan is nan. The first window - 1 samples are nan.

Parameters:

Name	Type	Description	Default
value	Any	Series to rank.	required
window	Any	Trailing window length. At least 1.	required

Returns:

Type	Description
ndarray	The rolling percentile rank.

Raises:

Type	Description
ValueError	If the window length is less than 1.

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

def ts_rank(value: Any, window: Any) -> numpy.ndarray:
    """Take the percentile rank of the current sample in a trailing window.

    The window is ``[t - window + 1, t]`` inclusive. The value is the
    1-based average rank of ``x[t]`` among those samples, scaled by
    ``(rank - 1) / (window - 1)``. A unique window low is ``0.0``, a
    unique window high is ``1.0``, and an all-tie window is ``0.5``.
    A window of 1 is ``nan``. A window that holds a ``nan`` is
    ``nan``. The first ``window - 1`` samples are ``nan``.

    Args:
        value: Series to rank.
        window: Trailing window length. At least 1.

    Returns:
        The rolling percentile rank.

    Raises:
        ValueError: If the window length is less than 1.
    """
    result, length, view = _windows(value, window)
    if view is None or length == 1:
        return result
    current = view[:, -1:]
    less = numpy.add.reduce(view < current, axis=-1)
    equal = numpy.add.reduce(view == current, axis=-1)
    rank = less + (equal + 1.0) / 2.0
    values = (rank - 1.0) / (length - 1.0)
    values[numpy.isnan(view).any(axis=-1)] = numpy.nan
    result[length - 1 :] = values
    return result

affine_scale(predicted, target, *, valid=None)

Fit Keijzer intercept and slope for a + b * f(x).

Least-squares a and b are computed on the same scorable samples :func:~deap_er.tools.case_errors uses: an optional valid= mask intersected with the finite check on both series. Darwinian callers apply a + b * predicted only when writing fitness or case errors; the tree is unchanged.

A series with no scorable samples returns the identity (0.0, 1.0). A constant prediction returns an intercept-only shift (mean(target) - mean(predicted), 1.0).

Parameters:

Name	Type	Description	Default
predicted	ndarray	Predicted series f(x).	required
target	ndarray	Target series, same length as predicted.	required
valid	ndarray | None	Optional per-sample mask. Same contract as :func:~deap_er.tools.case_errors.	None

Returns:

Type	Description
tuple[float, float]	(a, b) so the scaled series is a + b * predicted.

Raises:

Type	Description
ValueError	If the inputs are not aligned one-dimensional arrays, or if valid has the wrong shape.

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

def affine_scale(
    predicted: numpy.ndarray,
    target: numpy.ndarray,
    *,
    valid: numpy.ndarray | None = None,
) -> tuple[float, float]:
    """Fit Keijzer intercept and slope for ``a + b * f(x)``.

    Least-squares ``a`` and ``b`` are computed on the same scorable
    samples :func:`~deap_er.tools.case_errors` uses: an optional
    ``valid=`` mask intersected with the finite check on both series.
    Darwinian callers apply ``a + b * predicted`` only when writing
    fitness or case errors; the tree is unchanged.

    A series with no scorable samples returns the identity
    ``(0.0, 1.0)``. A constant prediction returns an intercept-only
    shift ``(mean(target) - mean(predicted), 1.0)``.

    Args:
        predicted: Predicted series ``f(x)``.
        target: Target series, same length as ``predicted``.
        valid: Optional per-sample mask. Same contract as
            :func:`~deap_er.tools.case_errors`.

    Returns:
        ``(a, b)`` so the scaled series is ``a + b * predicted``.

    Raises:
        ValueError: If the inputs are not aligned one-dimensional
            arrays, or if ``valid`` has the wrong shape.
    """
    predicted, target = _as_series(predicted, target)
    sample_valid = case_valid_mask(predicted, target, valid)
    if not numpy.any(sample_valid):
        return 0.0, 1.0
    forecast = predicted[sample_valid]
    observed = target[sample_valid]
    forecast_mean = float(numpy.mean(forecast))
    observed_mean = float(numpy.mean(observed))
    centered = forecast - forecast_mean
    denom = float(numpy.dot(centered, centered))
    if not numpy.isfinite(denom) or denom <= 0.0:
        return observed_mean - forecast_mean, 1.0
    slope = float(numpy.dot(observed - observed_mean, centered) / denom)
    intercept = observed_mean - slope * forecast_mean
    return intercept, slope

semantic_descriptors(matrix, *, kind='moments', valid=None, target=None, ranges=None, basis=None, center=None, empty=float('inf'), individuals=None, trust_matrix=False)

Dispatch a semantic pack to moments, solve bits, or a projection.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
kind	DescriptorKind	moments, solve, or project.	'moments'
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
target	ndarray | None	Target series. Required for solve.	None
ranges	Sequence[tuple[int, int]] | ndarray | None	Case bounds. Required for solve.	None
basis	ndarray | None	Projection matrix. Required for project.	None
center	ndarray | None	Optional center passed to :func:semantic_project.	None
empty	float	Empty-case MSE for solve.	float('inf')
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Descriptor array whose width depends on kind.

Raises:

Type	Description
ValueError	If kind is unknown or a required argument is missing.

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

def semantic_descriptors(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    *,
    kind: DescriptorKind = "moments",
    valid: numpy.ndarray | None = None,
    target: numpy.ndarray | None = None,
    ranges: Sequence[tuple[int, int]] | numpy.ndarray | None = None,
    basis: numpy.ndarray | None = None,
    center: numpy.ndarray | None = None,
    empty: float = float("inf"),
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Dispatch a semantic pack to moments, solve bits, or a projection.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        kind: ``moments``, ``solve``, or ``project``.
        valid: Optional per-row warmup mask of length ``n_rows``.
        target: Target series. Required for ``solve``.
        ranges: Case bounds. Required for ``solve``.
        basis: Projection matrix. Required for ``project``.
        center: Optional center passed to :func:`semantic_project`.
        empty: Empty-case MSE for ``solve``.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Descriptor array whose width depends on ``kind``.

    Raises:
        ValueError: If ``kind`` is unknown or a required argument is
            missing.
    """
    if kind == "moments":
        return semantic_moments(
            matrix, valid=valid, individuals=individuals, trust_matrix=trust_matrix
        )
    if kind == "solve":
        if target is None or ranges is None:
            raise ValueError("kind='solve' requires target and ranges")
        return semantic_solve_bits(
            matrix,
            target,
            ranges,
            valid=valid,
            empty=empty,
            individuals=individuals,
            trust_matrix=trust_matrix,
        )
    if kind == "project":
        if basis is None:
            raise ValueError("kind='project' requires basis")
        return semantic_project(
            matrix,
            basis,
            valid=valid,
            target=target,
            center=center,
            individuals=individuals,
            trust_matrix=trust_matrix,
        )
    raise ValueError(f"unknown descriptor kind {kind!r}")

semantic_moments(matrix, *, valid=None, individuals=None, trust_matrix=False)

Return per-individual mean, population std, min, and max.

Moments use samples where valid (broadcast) and the row are finite. An empty row is all-NaN. A single sample has std = 0.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Descriptor array of shape (n_individuals, 4).

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

def semantic_moments(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    *,
    valid: numpy.ndarray | None = None,
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Return per-individual mean, population std, min, and max.

    Moments use samples where ``valid`` (broadcast) and the row are
    finite. An empty row is all-NaN. A single sample has ``std = 0``.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        valid: Optional per-row warmup mask of length ``n_rows``.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Descriptor array of shape ``(n_individuals, 4)``.
    """
    packed = packed_semantics(matrix, individuals, trust_matrix)
    return _row_moments(packed, semantic_valid_mask(packed, valid))

semantic_solve_bits(matrix, target, ranges, *, valid=None, empty=float('inf'), individuals=None, trust_matrix=False)

Return one solved-case bit per individual and case.

A case is solved when its MSE is isclose to 0 with abs_tol=1e-12, matching :func:~deap_er.tools.sample_informed_cases. Segment bounds and valid= follow :func:~deap_er.tools.case_errors.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Predicted pack of shape (n_individuals, n_rows).	required
target	ndarray	Target series of length n_rows.	required
ranges	Sequence[tuple[int, int]] | ndarray	Case bounds accepted by :func:~deap_er.tools.case_intervals.	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
empty	float	MSE used when a case has no scorable samples.	float('inf')
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Float 0/1 array of shape (n_individuals, n_cases).

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

def semantic_solve_bits(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    target: numpy.ndarray,
    ranges: Sequence[tuple[int, int]] | numpy.ndarray,
    *,
    valid: numpy.ndarray | None = None,
    empty: float = float("inf"),
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Return one solved-case bit per individual and case.

    A case is solved when its MSE is ``isclose`` to ``0`` with
    ``abs_tol=1e-12``, matching :func:`~deap_er.tools.sample_informed_cases`.
    Segment bounds and ``valid=`` follow :func:`~deap_er.tools.case_errors`.

    Args:
        matrix: Predicted pack of shape ``(n_individuals, n_rows)``.
        target: Target series of length ``n_rows``.
        ranges: Case bounds accepted by :func:`~deap_er.tools.case_intervals`.
        valid: Optional per-row warmup mask of length ``n_rows``.
        empty: MSE used when a case has no scorable samples.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Float ``0/1`` array of shape ``(n_individuals, n_cases)``.
    """
    packed = packed_semantics(matrix, individuals, trust_matrix)
    series = numpy.asarray(target, dtype=numpy.float64)
    if series.ndim != 1 or series.shape[0] != packed.shape[1]:
        raise ValueError("target must be a one-dimensional series matching n_rows")
    intervals = case_intervals(ranges, packed.shape[1])
    row_valid = semantic_valid_mask(packed, valid) & numpy.isfinite(series)
    bits = numpy.empty((packed.shape[0], len(intervals)), dtype=numpy.float64)
    for index, (start, stop) in enumerate(intervals):
        sample = row_valid[:, start:stop]
        diff = packed[:, start:stop] - series[start:stop]
        sq = numpy.where(sample, diff * diff, 0.0)
        counts = sample.sum(axis=1)
        mse = numpy.where(counts > 0, sq.sum(axis=1) / counts, empty)
        bits[:, index] = numpy.isclose(mse, 0.0, atol=1e-12)
    return bits

semantic_distance(query, matrix, *, metric='euclidean', valid=None)

Return finite-mask distances from query to each packed row.

Only coordinates that are finite on both sides and marked valid enter the distance. An empty overlap, or a zero cosine norm, is +inf. A one-dimensional matrix is treated as a single row and still returns a length-1 array (not a Python float) so the return type stays ndarray for the type checker.

Parameters:

Name	Type	Description	Default
query	ndarray | Sequence[float]	Semantic row of length n_rows.	required
matrix	ndarray | Sequence[Sequence[float]] | Sequence[float]	Pack of shape (n_individuals, n_rows), or one row of length n_rows.	required
metric	SemanticMetric	euclidean or cosine (1 - cosine similarity).	'euclidean'
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None

Returns:

Type	Description
ndarray	Distances of length n_individuals. One-dimensional
ndarray	matrix yields shape (1,).

Raises:

Type	Description
ValueError	If shapes do not match or metric is unknown.

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

def semantic_distance(
    query: numpy.ndarray | Sequence[float],
    matrix: numpy.ndarray | Sequence[Sequence[float]] | Sequence[float],
    *,
    metric: SemanticMetric = "euclidean",
    valid: numpy.ndarray | None = None,
) -> numpy.ndarray:
    """Return finite-mask distances from ``query`` to each packed row.

    Only coordinates that are finite on both sides and marked ``valid``
    enter the distance. An empty overlap, or a zero cosine norm, is
    ``+inf``. A one-dimensional ``matrix`` is treated as a single row
    and still returns a length-1 array (not a Python float) so the
    return type stays ``ndarray`` for the type checker.

    Args:
        query: Semantic row of length ``n_rows``.
        matrix: Pack of shape ``(n_individuals, n_rows)``, or one row of
            length ``n_rows``.
        metric: ``euclidean`` or ``cosine`` (``1 -`` cosine similarity).
        valid: Optional per-row warmup mask of length ``n_rows``.

    Returns:
        Distances of length ``n_individuals``. One-dimensional
        ``matrix`` yields shape ``(1,)``.

    Raises:
        ValueError: If shapes do not match or ``metric`` is unknown.
    """
    packed = numpy.asarray(matrix, dtype=numpy.float64)
    if packed.ndim == 1:
        packed = packed.reshape(1, -1)
    packed = as_semantic_matrix(packed)
    row = _query_row(query, packed.shape[1])
    return _masked_distance(packed, row, _pair_mask(packed, row, valid), metric)

semantic_nearest(query, matrix, *, k=1, metric='euclidean', valid=None, individuals=None, trust_matrix=False)

Return the lowest-index nearest neighbors of query.

Infinite distances are skipped. Ties keep the lowest pack index. individuals is trust-alignment only: the return value is always integer pack indices, never the individual objects.

Parameters:

Name	Type	Description	Default
query	ndarray | Sequence[float]	Semantic row of length n_rows.	required
matrix	ndarray | Sequence[Sequence[float]]	Pack of shape (n_individuals, n_rows).	required
k	int	Maximum number of neighbors to return.	1
metric	SemanticMetric	euclidean or cosine.	'euclidean'
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Neighbor indices in increasing distance order, length at most
ndarray	k. Empty when every distance is infinite.

Raises:

Type	Description
ValueError	If k is less than 1, or the pack does not match individuals.

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

def semantic_nearest(
    query: numpy.ndarray | Sequence[float],
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    *,
    k: int = 1,
    metric: SemanticMetric = "euclidean",
    valid: numpy.ndarray | None = None,
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Return the lowest-index nearest neighbors of ``query``.

    Infinite distances are skipped. Ties keep the lowest pack index.
    ``individuals`` is trust-alignment only: the return value is always
    integer pack indices, never the individual objects.

    Args:
        query: Semantic row of length ``n_rows``.
        matrix: Pack of shape ``(n_individuals, n_rows)``.
        k: Maximum number of neighbors to return.
        metric: ``euclidean`` or ``cosine``.
        valid: Optional per-row warmup mask of length ``n_rows``.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Neighbor indices in increasing distance order, length at most
        ``k``. Empty when every distance is infinite.

    Raises:
        ValueError: If ``k`` is less than 1, or the pack does not match
            ``individuals``.
    """
    if k < 1:
        raise ValueError("k must be at least 1")
    if individuals is None:
        packed = as_semantic_matrix(matrix)
    else:
        packed = validate_semantic_matrix(matrix, individuals, trust_matrix=trust_matrix)
    dist = semantic_distance(query, packed, metric=metric, valid=valid)
    finite = numpy.flatnonzero(numpy.isfinite(dist))
    if finite.size == 0:
        return numpy.empty(0, dtype=int)
    order = finite[numpy.argsort(dist[finite], kind="stable")]
    return order[:k]

semantic_pca_basis(matrix, n_dims, *, valid=None, target=None)

Return a thin-SVD basis and column center for :func:semantic_project.

Columns outside valid (and non-finite target samples) are dropped before centering. Unused rows of the returned basis and center are zero.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
n_dims	int	Number of components to keep.	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
target	ndarray | None	Optional target whose non-finite samples are dropped.	None

Returns:

Type	Description
ndarray	(basis, center) with shapes (n_rows, n_dims) and
ndarray	(n_rows,).

Raises:

Type	Description
ValueError	If n_dims is not positive, or no finite row remains on the kept columns.

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

def semantic_pca_basis(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    n_dims: int,
    *,
    valid: numpy.ndarray | None = None,
    target: numpy.ndarray | None = None,
) -> tuple[numpy.ndarray, numpy.ndarray]:
    """Return a thin-SVD basis and column center for :func:`semantic_project`.

    Columns outside ``valid`` (and non-finite ``target`` samples) are
    dropped before centering. Unused rows of the returned basis and
    center are zero.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        n_dims: Number of components to keep.
        valid: Optional per-row warmup mask of length ``n_rows``.
        target: Optional target whose non-finite samples are dropped.

    Returns:
        ``(basis, center)`` with shapes ``(n_rows, n_dims)`` and
        ``(n_rows,)``.

    Raises:
        ValueError: If ``n_dims`` is not positive, or no finite row
            remains on the kept columns.
    """
    if n_dims < 1:
        raise ValueError("n_dims must be positive")
    packed = as_semantic_matrix(matrix)
    keep = semantic_column_keep(packed.shape[1], valid, target)
    if not numpy.any(keep):
        raise ValueError("semantic_pca_basis needs at least one finite row on kept columns")
    kept = packed[:, keep]
    finite_rows = numpy.all(numpy.isfinite(kept), axis=1)
    kept = kept[finite_rows]
    if kept.shape[0] == 0:
        raise ValueError("semantic_pca_basis needs at least one finite row on kept columns")
    center_keep = kept.mean(axis=0)
    _, _, vt = numpy.linalg.svd(kept - center_keep, full_matrices=False)
    n_comp = min(n_dims, vt.shape[0])
    basis = numpy.zeros((packed.shape[1], n_dims), dtype=numpy.float64)
    center = numpy.zeros(packed.shape[1], dtype=numpy.float64)
    center[keep] = center_keep
    if n_comp:
        basis[numpy.ix_(numpy.flatnonzero(keep), numpy.arange(n_comp))] = vt[:n_comp].T
    return basis, center

semantic_project(matrix, basis, *, valid=None, target=None, center=None, individuals=None, trust_matrix=False)

Project a semantic pack through a caller-supplied basis.

Columns outside valid (and non-finite target samples) are zeroed so warmup does not enter the product. A row that is still non-finite on a kept column yields a non-finite descriptor.

Parameters:

Name	Type	Description	Default
matrix	ndarray | Sequence[Sequence[float]]	Semantic pack of shape (n_individuals, n_rows).	required
basis	ndarray	Projection matrix of shape (n_rows, n_dims).	required
valid	ndarray | None	Optional per-row warmup mask of length n_rows.	None
target	ndarray | None	Optional target whose non-finite samples are dropped.	None
center	ndarray | None	Optional length-n_rows vector subtracted before the product. Used with :func:semantic_pca_basis.	None
individuals	Sequence[Any] | None	Optional population used by trust_matrix.	None
trust_matrix	bool	When True, accept matrix on shape alone.	False

Returns:

Type	Description
ndarray	Descriptor array of shape (n_individuals, n_dims).

Raises:

Type	Description
ValueError	If basis or center does not match n_rows.

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

def semantic_project(
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    basis: numpy.ndarray,
    *,
    valid: numpy.ndarray | None = None,
    target: numpy.ndarray | None = None,
    center: numpy.ndarray | None = None,
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> numpy.ndarray:
    """Project a semantic pack through a caller-supplied basis.

    Columns outside ``valid`` (and non-finite ``target`` samples) are
    zeroed so warmup does not enter the product. A row that is still
    non-finite on a kept column yields a non-finite descriptor.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        basis: Projection matrix of shape ``(n_rows, n_dims)``.
        valid: Optional per-row warmup mask of length ``n_rows``.
        target: Optional target whose non-finite samples are dropped.
        center: Optional length-``n_rows`` vector subtracted before the
            product. Used with :func:`semantic_pca_basis`.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Returns:
        Descriptor array of shape ``(n_individuals, n_dims)``.

    Raises:
        ValueError: If ``basis`` or ``center`` does not match ``n_rows``.
    """
    packed = packed_semantics(matrix, individuals, trust_matrix)
    components = numpy.asarray(basis, dtype=numpy.float64)
    if components.ndim != 2 or components.shape[0] != packed.shape[1]:
        raise ValueError(
            f"basis must have shape ({packed.shape[1]}, n_dims), got {components.shape}"
        )
    keep = semantic_column_keep(packed.shape[1], valid, target)
    shifted = packed.copy()
    if center is not None:
        mean = numpy.asarray(center, dtype=numpy.float64)
        if mean.ndim != 1 or mean.shape[0] != packed.shape[1]:
            raise ValueError("center must be a one-dimensional vector matching n_rows")
        shifted = shifted - mean
    shifted[:, ~keep] = 0.0
    return shifted @ components

semantic_random_basis(n_rows, n_dims)

Return a Gaussian random-projection basis.

Columns are scaled by 1 / sqrt(n_dims). Draws use the process :data:~deap_er.tools.rng.

Parameters:

Name	Type	Description	Default
n_rows	int	Number of semantic coordinates (rows of the pack).	required
n_dims	int	Number of projected dimensions.	required

Returns:

Type	Description
ndarray	Basis of shape (n_rows, n_dims).

Raises:

Type	Description
ValueError	If n_rows or n_dims is not positive.

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

def semantic_random_basis(n_rows: int, n_dims: int) -> numpy.ndarray:
    """Return a Gaussian random-projection basis.

    Columns are scaled by ``1 / sqrt(n_dims)``. Draws use the process
    :data:`~deap_er.tools.rng`.

    Args:
        n_rows: Number of semantic coordinates (rows of the pack).
        n_dims: Number of projected dimensions.

    Returns:
        Basis of shape ``(n_rows, n_dims)``.

    Raises:
        ValueError: If ``n_rows`` or ``n_dims`` is not positive.
    """
    if n_rows < 1 or n_dims < 1:
        raise ValueError("n_rows and n_dims must be positive")
    draw = rng.standard_normal((n_rows, n_dims))
    return draw / numpy.sqrt(n_dims)