mixle.evolve package

mixle.evolve – the self-improvement core: measure -> propose -> verify -> promote.

A pure, serving-agnostic orchestration algebra over capabilities that already exist in the library. It adds no new modeling capability: the proper scores and calibration diagnostics (mixle.inference.scoring / mixle.inference.calibration), the comparison tests (mixle.inference.model_comparison), the fitters (mixle.inference.estimation.optimize(), the streaming estimators, mixle.utils.automatic.get_estimator()), and the decision layer (mixle.inference.decision.bayes_action()) are all pre-existing. evolve wires them into a single anti-regression loop.

Core surface:

• measure – Objective + nll / log_score / crps / interval / calibration / decision_regret builders.

• propose – ImprovementOperator + Refit, OnlineUpdate, AutoSelect, Recalibrate, a scoped operator registry.

• verify – challenger_beats_champion() + Verdict (the anti-regression gate).

• drive – improve() + ImprovementResult, auto_select(), and the EvolutionLedger telemetry.

Search and population surface:

• search – search() over a typed Space (Real / Integer / Categorical), method='evolutionary' / 'bandit' / 'bo', returning a SearchResult.

• meta-search – Population + OperatorBandit (the policy that learns which operators help).

• structure search – Recompose / Mutate (genetic-programming structural moves over the model’s compositional tree) + structural_distance() (a tree-edit genotype distance) driving the population’s diversity; registered but off by default (structural + expensive).

L1 surface (closed-loop self-evolution, mixle.evolve.closed_loop):

• ClosedLoopSelfEvolution – harvest -> A5 acquire -> distill/refine/evolve challenger production -> challenger_beats_champion() -> deploy, as one budgeted background loop.

• OperatorCreditBandit – a per-context meta-bandit (over mixle.task.bandit.UCB1) crediting WHICH challenger-production operator wins for which kind of failure.

• GenealogyLedger – parent/operator/measured-gap receipts on every adopted champion, with a real lineage(model) walk-back.

improve(model, data, *, objective, operators=None, holdout=0.25, alpha=0.05, min_effect=0.0, budget=None, seed=0, ledger=None, parent_hash=None, require_calibration=True)

Propose challengers, gate them, and return the best verified improvement (or the champion).

Parameters:

• model (Any) – the champion (a fitted mixle distribution).

• data (Sequence[Any]) – the raw held-out dataset (split once into train/verify here).

• objective (Objective) – the Objective to improve on.

• operators (Sequence[ImprovementOperator] | None) – the proposal moves; defaults to the applicable subset of [Refit, OnlineUpdate, AutoSelect, Recalibrate].

• holdout (float | tuple) – train/verify split fraction, or an explicit (train, verify) tuple.

• alpha (float) – significance level for the verify gate.

• min_effect (float) – practical effect-size floor passed to the gate.

• budget (float | None) – optional cost ceiling – operators whose cost_hint exceeds the remaining budget are skipped after ascending-cost ordering.

• seed (int) – RNG seed for the split and the sampled objectives.

• ledger (EvolutionLedger | None) – optional EvolutionLedger to record every attempt into.

• parent_hash (str | None) – optional lineage hash for the champion (carried onto candidates and ledger rows).

• require_calibration (bool) – forward to the gate’s calibration no-regression check.

Returns:

An ImprovementResult. result.verified is True guarantees result.model beat the champion significantly and non-regressively on the verify split.

Return type:

ImprovementResult

class ImprovementResult(model, verified, operator, delta, verdict, evidence=<factory>, parent_hash=None)

Bases: object

The outcome of an improve() run.

Parameters:

• model (Any)

• verified (bool)

• operator (str | None)

• delta (float)

• verdict (Verdict | None)

• evidence (dict)

• parent_hash (str | None)

auto_select(data, *, space=None, criterion='bic', verify=True, holdout=0.25, seed=0, max_its=20)

Infer and fit a model from raw data, optionally gated by a held-out proper score.

Parameters:

• data (Sequence[Any]) – the raw dataset.

• space (Any | None) – reserved for typed search-space selection; must currently be None for auto_select.

• criterion (str | Objective) – 'bic' (delegate to the automatic in-sample pick) or a proper-score Objective (add the held-out verify gate on top of BIC).

• verify (bool) – when criterion is an Objective, whether to run the held-out gate (the BIC pick fitted on the train split is the champion; the BIC pick refitted on all data is the challenger, promoted only if it wins out of sample).

• holdout (float) – held-out fraction for the proper-score gate.

• seed (int) – RNG seed for the split and sampled objectives.

• max_its (int) – EM iterations for the fits.

Returns:

An ImprovementResult. For criterion='bic' it carries the fitted automatic model with verified=False (no out-of-sample test was requested). For an Objective criterion with verify=True it carries the gate verdict and verified reflects whether the full-data model beats the train-only model out of sample.

Return type:

ImprovementResult

class Objective(*args, **kwargs)

Bases: Protocol

A lower-is-better-or-higher-with-a-flag scalar fitness with an optional paired vector.

pointwise(model, data)

The (n,) per-observation score, or None if the objective is scalar-only.

Parameters:

• model (Any)

• data (Any)

Return type:

ndarray | None

scalar(model, data)

The single comparable fitness (mean of pointwise by default).

Parameters:

• model (Any)

• data (Any)

Return type:

float

nll_objective()

Negative log-likelihood: per-obs -log p(y_i) (strictly proper, lower is better).

Return type:

Objective

log_score_objective()

Logarithmic score (log loss) of the predictive density at the realised outcomes.

Return type:

Objective

crps_objective(*, ensemble=256, seed=0)

Continuous Ranked Probability Score from a sampled predictive ensemble (lower is better).

Parameters:

• ensemble (int) – number of predictive draws per observation.

• seed (int) – RNG seed for the ensemble (reproducible).

Return type:

Objective

interval_objective(level=0.9, *, ensemble=256, seed=0)

Winkler interval score for the central level predictive interval (lower is better).

Parameters:

• level (float) – central coverage of the interval (e.g. 0.9 for a 90% interval).

• ensemble (int) – number of predictive draws used to read off the interval endpoints.

• seed (int) – RNG seed for the ensemble.

Return type:

Objective

calibration_objective(*, ensemble=256, seed=0, bins=10)

PIT calibration error of the predictive distribution (scalar-only, lower is better).

Uses the rank-based Probability Integral Transform of a sampled ensemble: under a calibrated continuous forecast the PIT values are Uniform(0, 1), and pit_calibration_error measures the histogram’s mean absolute deviation from uniform. There is no per-observation paired vector for a histogram statistic, so pointwise returns None and the verify gate scores this on the bootstrapped scalar.

(For classification models the natural calibration scalar is mixle.inference.calibration.expected_calibration_error(); this builder targets the continuous-predictive case, which is the common one for the streaming/auto-select loop.)

Parameters:

• ensemble (int)

• seed (int)

• bins (int)

Return type:

Objective

decision_regret_objective(loss, actions, *, n=2000, seed=0)

Realized decision regret under a fitted predictive posterior (scalar-only, lower is better).

For the chosen Bayes action a* = bayes_action(posterior(model), loss, actions) this reports the expected loss E_draw[ loss(a*, draw) ] – the realized cost of acting optimally under the model’s own belief. A better-calibrated model yields a lower expected loss for the same loss and action set, so it is a first-class promotion metric. Scalar-only (the regret is an expectation over posterior draws, not a per-observation quantity), so pointwise returns None.

Parameters:

• loss (Callable[[Any, Any], float]) – loss(action, draw) -> float (or a numpy-vectorized loss(action, draws) -> array).

• actions (Sequence[Any]) – the finite candidate-action set.

• n (int) – posterior draws for the Monte-Carlo expectation.

• seed (int) – RNG seed.

Return type:

Objective

challenger_beats_champion(champion, challenger, data, *, objective, alpha=0.05, min_effect=0.0, require_calibration=True, nonnested=False, multiplicity=None, calib_tol=1.0e-3, seed=0, elpd_pointwise=None)

Decide whether challenger significantly and non-regressively beats champion on data.

Parameters:

• champion (Any) – two fitted models scored on the same held-out data.

• challenger (Any) – two fitted models scored on the same held-out data.

• data (Any) – the held-out responses (one batch -> both models scored in the same order).

• objective (Objective) – the Objective to compare on.

• alpha (float) – significance level for the paired test (and the CI is at 1 - alpha).

• min_effect (float) – practical effect-size floor on |mean score difference|.

• require_calibration (bool) – if True, run the calibration no-regression check.

• nonnested (bool) – if True (a family swap), additionally require Vuong + Clarke to favor the challenger.

• multiplicity (str | None) – None or a mixle.inference.multiple_testing.adjust_pvalues() method ('bonferroni' / 'bh' / …) when this is one of many simultaneous challengers – adjusts the p-value before the gate.

• calib_tol (float) – tolerance on the calibration-error increase the challenger may carry.

• seed (int) – RNG seed for the (sampled) calibration scalars.

• elpd_pointwise (tuple[ndarray, ndarray] | None) – optional (pointwise_champion, pointwise_challenger) LOO/WAIC arrays; when given, the compare_elpd() 2-SE band is required to also favor the challenger.

Returns:

A Verdict. verdict.promote is the single promotion predicate.

Return type:

Verdict

class Verdict(favored, delta, p_value, ci, calibrated, evidence=<factory>)

Bases: object

The outcome of a single champion/challenger comparison.

Parameters:

• favored (str)

• delta (float)

• p_value (float)

• ci (tuple[float, float])

• calibrated (bool)

• evidence (dict)

property promote: bool

True iff the challenger is favored and passed calibration – the promotion predicate.

as_dict()

Serialize the verdict into JSON-compatible primitive fields.

Return type:

dict[str, Any]

class ImprovementOperator(*args, **kwargs)

Bases: Protocol

A uniform proposal move: an applicability pre-flight plus a fitted-challenger propose.

applicable(model, data, *, ctx)

Structural gate: whether this operator can run on the model and data.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

bool

propose(model, data, *, ctx)

Return a fitted challenger (or raise if the proposal cannot be built).

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

Candidate

class Candidate(model, operator, parent_hash=None, meta=<factory>)

Bases: object

A proposed, already-fitted challenger plus the provenance of how it was made.

Parameters:

• model (Any)

• operator (str)

• parent_hash (str | None)

• meta (dict)

class Refit(name='refit', cost_hint=1.0, max_its=20)

Bases: object

Re-fit the champion’s family on fresh data, warm-started from the champion’s parameters.

Parameters:

• name (str)

• cost_hint (float)

• max_its (int)

applicable(model, data, *, ctx)

Return whether model can be re-fit on a non-empty batch.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

bool

propose(model, data, *, ctx)

Fit the model family on data using model as the warm start.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

Candidate

class OnlineUpdate(mode='streaming', cost_hint=0.2)

Bases: object

Fold a fresh batch into the champion via a streaming estimator.

mode:

• 'streaming' – decay-mode StreamingEstimator (running-accumulator forgetting).

• 'incremental' – Neal-Hinton IncrementalEstimator (replace one chunk).

• 'posterior_carry' – exact recursive-Bayes BayesianStreamingEstimator (needs a

  conjugate family; applicable checks ConjugateUpdatable).

• 'forgetting' – power-prior BayesianStreamingEstimator.

Parameters:

• mode (str)

• cost_hint (float)

property name: str

Registry name including the selected online-update mode.

applicable(model, data, *, ctx)

Return whether the selected update mode is legal for model and data.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

bool

propose(model, data, *, ctx)

Apply the selected streaming update and return the updated challenger.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

Candidate

class AutoSelect(name='auto_select', cost_hint=3.0, max_its=20)

Bases: object

Infer an estimator from the raw data (get_estimator) and fit it – a possible family swap.

Parameters:

• name (str)

• cost_hint (float)

• max_its (int)

applicable(model, data, *, ctx)

Return whether there is data available for automatic family selection.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

bool

propose(model, data, *, ctx)

Infer an estimator from data, fit it, and return the fitted challenger.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

Candidate

class Recalibrate(name='recalibrate', cost_hint=0.5, ensemble=256, seed=0, grid=(0.6, 0.75, 0.9, 1.0, 1.1, 1.25, 1.5, 2.0))

Bases: object

Learn a predictive spread temperature T that flattens the PIT, no parameter refit.

applicable requires a sampler (used both to estimate the predictive center and to evaluate PIT calibration). The temperature is chosen on the train split by minimising the PIT calibration error over a small grid; T == 1 (the identity) is always in the grid, so the recalibrated model can never be worse-calibrated than the base on the fitting data.

Parameters:

• name (str)

• cost_hint (float)

• ensemble (int)

• seed (int)

• grid (tuple[float, ...])

applicable(model, data, *, ctx)

Return whether model can be sampled and the batch is non-empty.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

bool

propose(model, data, *, ctx)

Search the temperature grid and wrap model in the best calibration transform.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

Candidate

class Recompose(name='recompose', cost_hint=4.0, max_its=30)

Bases: object

Propose a richer two-component mixture structure for the champion family.

Each component is warm-fit on a different half of the data so EM starts away from the identical-component solution. The normal verification gate decides whether the additional structure improves held-out evidence enough to promote. The operator is registered but omitted from the default set because it is intentionally more expensive than the conservative updates.

Parameters:

• name (str)

• cost_hint (float)

• max_its (int)

applicable(model, data, *, ctx)

Return whether the model can be re-estimated on enough rows for a split.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

bool

propose(model, data, *, ctx)

Fit a two-component mixture challenger initialized from data splits.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

Candidate

class Mutate(name='mutate', cost_hint=4.0, max_its=30)

Bases: object

Apply a random structural mutation to the champion and refit.

Available moves are grow for adding a bootstrap-fit component, shrink for dropping the lowest-weight component from an existing mixture, and perturb for a bootstrap re-fit. Repeated use inside a Population gives the search driver a structure-induction move, while the verification gate remains responsible for promotion. The operator is registered but omitted from the default set because it is comparatively expensive.

Parameters:

• name (str)

• cost_hint (float)

• max_its (int)

applicable(model, data, *, ctx)

Return whether the model can be structurally mutated and re-fit.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

bool

propose(model, data, *, ctx)

Sample one structural move, refit it, and return the resulting challenger.

Parameters:

• model (Any)

• data (Any)

• ctx (dict)

Return type:

Candidate

structural_distance(a, b)

A [0, 1] genotype distance between two models’ structures (tree-edit distance, size-normalized).

Parameters:

• a (Any)

• b (Any)

Return type:

float

model_signature(model)

The compositional structure of model as a (type_label, [child signatures]) tree.

Parameters:

model (Any)

Return type:

tuple[str, list]

tree_edit_distance(a, b)

Unordered tree-edit distance: relabel cost (0/1) + greedy min-cost matching of children, with unmatched subtrees fully inserted/deleted. Exact for identical trees; a standard greedy approximation otherwise.

Parameters:

• a (tuple[str, list])

• b (tuple[str, list])

Return type:

int

register_operator(operator)

Register operator in the scoped evolve operator registry (returns it for decorator use).

Parameters:

operator (ImprovementOperator)

Return type:

ImprovementOperator

unregister_operator(name)

Remove a previously-registered operator by name (no-op if absent).

Parameters:

name (str)

Return type:

None

registered_operators()

A copy of the current scoped operator registry.

Return type:

dict[str, ImprovementOperator]

default_operators()

The Phase-1 default operator set: refit, online update, auto-select, recalibrate. (Recompose + Mutate are structural + expensive, so they are available via the registry but not enabled by default.)

Return type:

list[ImprovementOperator]

class EvolutionLedger(rows=<factory>)

Bases: object

An ordered, JSON-serializable log of improvement attempts.

Parameters:

rows (list[dict[str, Any]])

record(*, operator, delta, verdict, cost, parent_hash, meta=None)

Append one attempt row and return it.

Parameters:

• operator (str)

• delta (float)

• verdict (dict | None)

• cost (float)

• parent_hash (str | None)

• meta (dict | None)

Return type:

dict[str, Any]

to_json(**dumps_kwargs)

Serialize the full ledger to a JSON string (rows are already plain dicts).

Parameters:

dumps_kwargs (Any)

Return type:

str

search(space, data, *, objective, build_fn, method='bo', n_iter=25, holdout=0.25, seed=0, **method_kwargs)

Search space for the config whose build_fn model scores best on a held-out split.

Parameters:

• space (Space) – the typed Space to search.

• data (Sequence[Any]) – the raw dataset (split once into train/val here for the inner objective).

• objective (Objective) – the held-out Objective (lower-is-better aware).

• build_fn (Callable[[dict[str, Any]], Any]) – caller-supplied config -> fitted model (the search is family-agnostic).

• method (str) – 'bo' (Bayesian optimization over the numeric box), 'evolutionary' (a (mu + lambda) loop over sample / neighbors), or 'bandit' (delegate the operator policy to an OperatorBandit).

• n_iter (int) – search budget (BO acquisition steps / evolutionary generations / bandit generations).

• holdout (float) – held-out fraction for the inner objective.

• seed (int) – RNG seed.

• method_kwargs (Any) – backend-specific knobs (e.g. mu / lam for the evolutionary loop, operators / size for the bandit population).

Returns:

A SearchResult with best_config / best_model / best_score (native orientation) / history.

Return type:

SearchResult

class SearchResult(best_config, best_model, best_score, history=<factory>)

Bases: object

The outcome of a search() (or Population.run()) run.

Parameters:

• best_config (dict[str, Any])

• best_model (Any)

• best_score (float)

• history (list[dict[str, Any]])

class Space(dims)

Bases: object

A typed, named search space over Real / Integer / Categorical dims.

Construct from a dict mapping each parameter name to its dimension:

space = Space({"mu": Real(-5, 5), "k": Integer(1, 4), "family": Categorical(["a", "b"])})

Dimension order is the insertion order of the dict; encode() / decode() and to_bounds() all use that same fixed order so the numeric vector and the box align.

Parameters:

dims (dict[str, Dimension])

property ndim: int

Number of dimensions in the fixed search-space order.

to_bounds()

The continuous (low, high) box for the BO backend (categoricals as integer indices).

Return type:

list[tuple[float, float]]

sample(rng)

Draw a random config dict, each dimension sampled natively.

Parameters:

rng (RandomState)

Return type:

dict[str, Any]

encode(config)

Encode a config dict into the numeric vector (in the fixed dimension order).

Parameters:

config (dict[str, Any])

Return type:

ndarray

decode(vector)

Decode a numeric vector (BO proposal) back into a config dict (rounding / clipping).

Parameters:

vector (Sequence[float])

Return type:

dict[str, Any]

neighbors(point)

All configs one local move away in exactly one dimension (the evolutionary mutation set).

Parameters:

point (dict[str, Any])

Return type:

list[dict[str, Any]]

class Real(lo, hi)

Bases: object

A continuous dimension over [lo, hi].

Parameters:

• lo (float)

• hi (float)

bounds()

Return the numeric bounds used by continuous optimizers.

Return type:

tuple[float, float]

sample(rng)

Draw a uniformly distributed value from the interval.

Parameters:

rng (RandomState)

Return type:

float

encode(value)

Clip and encode value as a floating-point coordinate.

Parameters:

value (Any)

Return type:

float

decode(x)

Clip a numeric optimizer coordinate back into the interval.

Parameters:

x (float)

Return type:

float

neighbors(value)

A coarse local move: +/- 10% of the range, clipped to bounds.

Parameters:

value (Any)

Return type:

list[float]

class Integer(lo, hi)

Bases: object

An integer dimension over the inclusive range [lo, hi].

Parameters:

• lo (int)

• hi (int)

bounds()

Return widened numeric bounds so rounding covers each integer level.

Return type:

tuple[float, float]

sample(rng)

Draw an integer uniformly from the inclusive range.

Parameters:

rng (RandomState)

Return type:

int

encode(value)

Round, clip, and encode value as a numeric coordinate.

Parameters:

value (Any)

Return type:

float

decode(x)

Round and clip a numeric optimizer coordinate to an integer value.

Parameters:

x (float)

Return type:

int

neighbors(value)

Return adjacent integer values within the range.

Parameters:

value (Any)

Return type:

list[int]

class Categorical(choices)

Bases: object

An unordered categorical dimension over a finite list of choices.

Parameters:

choices (tuple[Any, ...])

bounds()

Return widened index bounds so rounding covers each categorical choice.

Return type:

tuple[float, float]

sample(rng)

Draw one choice uniformly at random.

Parameters:

rng (RandomState)

Return type:

Any

encode(value)

Encode a choice as its floating-point index.

Parameters:

value (Any)

Return type:

float

decode(x)

Round and clip an optimizer coordinate back to a choice.

Parameters:

x (float)

Return type:

Any

neighbors(value)

Return all choices except value.

Parameters:

value (Any)

Return type:

list[Any]

class Population(seeds, *, objective, operators=None, bandit=None, size=12, diversity_quota=2, seed=0)

Bases: object

A diversity-preserving population of model structures, evolved by the OperatorBandit.

seeds are fitted models (the starting structures). Each step() selects operators via the bandit, applies them to parents chosen by fitness, gates the challengers with the Phase-1 champion/challenger rule (Benjamini-Hochberg multiplicity, since a generation produces many challengers at once), rewards the bandit with the verified deltas, and keeps the verified-best plus a coarse capability-diversity quota.

Parameters:

• seeds (Sequence[Any]) – the initial fitted models (at least one).

• objective (Objective) – the Objective to optimize (lower-is-better aware).

• operators (Sequence[ImprovementOperator] | None) – the proposal-move pool; defaults to the Phase-1 safe set.

• bandit (OperatorBandit | None) – an OperatorBandit over operators (built with the default policy if omitted).

• size (int) – the carrying capacity of the population.

• diversity_quota (int) – how many of size slots are reserved for capability-diverse members (the rest go to the fittest); the quota keeps the search from collapsing onto one structure too early.

• seed (int) – RNG seed for parent sampling and the bandit.

step(data)

Run one generation: select -> propose -> gate -> reward -> survivor selection.

Parameters:

data (Any)

Return type:

GenerationReport

run(data, generations=5)

Evolve for generations steps; return a SearchResult.

The returned best_model is the run incumbent, guaranteed no worse than the best seed on the objective (anti-regression). history is one row per generation (proposals / verified / score).

Parameters:

• data (Any)

• generations (int)

Return type:

Any

property champion: Any

The current run incumbent (the best model seen, anti-regression guaranteed).

class OperatorBandit(operators, *, policy='thompson', decay=0.97, prior_cost_aware=True, seed=0)

Bases: object

A non-stationary bandit over a fixed pool of ImprovementOperator.

select(k) returns the k highest-value operators under the chosen policy; reward folds a verified delta + cost back into the chosen arm; report is the “which operators help” artifact.

Policies:

• 'thompson' – Beta-Bernoulli Thompson sampling on the win indicator (reward > 0), scaled by the mean positive reward, so an operator that wins rarely but big and one that wins often but small are compared on expected verified delta.

• 'ucb' – UCB1 on the mean reward with a sqrt(2 ln N / n) exploration bonus.

Non-stationarity: each reward first multiplies every arm’s statistics by decay (a forgetting factor in (0, 1]), so old evidence fades and the policy can track a shifting best operator.

Parameters:

• operators (Sequence[ImprovementOperator])

• policy (str)

• decay (float)

• prior_cost_aware (bool)

• seed (int)

value(name)

The current policy value of operator name (a Thompson draw or the UCB index).

Parameters:

name (str)

Return type:

float

select(k=1)

Return the k operators with the highest policy value (a fresh Thompson draw each call).

Parameters:

k (int)

Return type:

list[ImprovementOperator]

reward(op_name, delta, cost)

Fold a verified delta (0 if the challenger was rejected) and cost into the arm.

Decays every arm first (non-stationarity), then updates the chosen arm. A delta is clipped at 0 below – a rejected challenger is a zero reward, never negative, matching the anti-regression guarantee (we never punish an operator for a rejected proposal beyond not rewarding it).

Parameters:

• op_name (str)

• delta (float)

• cost (float)

Return type:

None

report()

Per-operator win-rate, mean verified delta, mean cost, and (decayed) pull count.

Return type:

dict[str, Any]

class ClosedLoopSelfEvolution(champion, *, objective, operators=None, context_fn=None, acquire_k=32, acquire_strategy='entropy', bandit_c=1.0, seed=0)

Bases: object

The whole L1 loop, one budgeted step at a time: harvest -> acquire -> propose -> gate -> deploy, with per-context operator credit and genealogy receipts on every adoption.

Parameters:

• champion (Any) – the initial fitted model (e.g. a CategoricalDistribution).

• objective (Objective) – the Objective the loop optimizes (e.g. accuracy_objective()).

• operators (dict[str, ImprovementOperator] | None) – challenger-production operators by name; defaults to default_challenger_operators().

• context_fn (Callable[[Sequence[Any]], str] | None) – batch -> context key for the per-context bandit; defaults to a single global context ("default") when the caller has no domain/failure-type signal to condition on.

• acquire_k (int) – how many harvested failures A5’s acquire prioritizes per step.

• acquire_strategy (str) – the acquire() ranking strategy (default "entropy", the one that works for a marginal/non-conditional champion via _ConstantProbaAdapter).

• seed (int) – RNG seed threaded through the bandit and the gate.

• bandit_c (float)

step(batch, *, verify=None, context=None)

One cycle of the loop on one arriving batch of held-out-shaped observations.

1. harvest failures under the CURRENT champion,

2. rank them with A5’s acquire,

3. pick a challenger-production operator via the per-context bandit,

4. propose + gate the challenger,

5. reward the bandit with the verified (anti-regression) delta,

6. deploy + record a genealogy receipt iff the gate promotes.

Parameters:

• batch (Sequence[Any])

• verify (Sequence[Any] | None)

• context (str | None)

Return type:

LoopStepResult

run(stream, *, verify_batches=None, contexts=None, budget=None)

Run the loop over a SEQUENCE of batches (a stream) – the budgeted background loop’s outer iteration. budget (if given) is a ceiling on the total cost_hint of operators actually applied (proposal attempts that were skipped as inapplicable don’t spend budget); the loop stops early once it is exhausted, leaving the champion at whatever it last became.

Parameters:

• stream (Sequence[Sequence[Any]])

• verify_batches (Sequence[Sequence[Any]] | None)

• contexts (Sequence[str] | None)

• budget (float | None)

Return type:

list[LoopStepResult]

class LoopStepResult(context, operator, promoted, delta, champion_score, champion)

Bases: object

One ClosedLoopSelfEvolution.step() outcome.

Parameters:

• context (str)

• operator (str)

• promoted (bool)

• delta (float)

• champion_score (float)

• champion (Any)

class OperatorCreditBandit(operator_names, *, c=1.0, seed=0)

Bases: object

A per-context meta-bandit over challenger-production operators, wrapping one mixle.task.bandit.UCB1 per context (e.g. a failure type/domain) – so the loop learns, independently for each context, which operator actually produces winning challengers there.

Parameters:

• operator_names (Sequence[str])

• c (float)

• seed (int)

select(context)

The bandit-chosen operator name for context.

Parameters:

context (str)

Return type:

str

reward(context, operator, reward)

Fold an observed (non-negative, anti-regression) reward back into operator’s arm.

Parameters:

• context (str)

• operator (str)

• reward (float)

Return type:

None

report()

Per-context, per-operator mean reward and pull count.

Return type:

dict[str, dict[str, float]]

class GenealogyLedger(ledger=<factory>, _model_ids=<factory>, _counter=0)

Bases: object

Genealogy receipts for adopted (gate-passing) champions, built directly on EvolutionLedger (never storing model objects in the ledger rows themselves – only their operator, measured gap, and a stable id – exactly the ledger’s own JSON-serializability discipline).

Parameters:

• ledger (EvolutionLedger)

• _model_ids (dict[int, str])

• _counter (int)

record_adoption(*, parent, child, operator, gap, context, meta=None)

Record ONE adoption: child replaced parent via operator, measured gap (the verified challenger-beats-champion delta). parent=None marks the root of a lineage.

Parameters:

• parent (Any | None)

• child (Any)

• operator (str)

• gap (float)

• context (str)

• meta (dict | None)

Return type:

dict[str, Any]

lineage(model)

Reconstruct model’s full lineage: the ordered (root-first) chain of adoption receipts {operator, delta (the measured gap), parent_hash, meta: {child_hash, context, ...}} back to the first recorded ancestor. Returns [] if model was never recorded as an adoption.

Parameters:

model (Any)

Return type:

list[dict[str, Any]]

accuracy_objective()

Higher-is-better accuracy for a model exposing a single “most likely label” prediction (a fitted CategoricalDistribution’s argmax(pmap)): per-observation 1{y == predicted_label}.

Return type:

Objective

harvest_failures(champion, batch, objective)

The objective-generic analog of mixle.task.router.Router.harvested(): the observations in batch where champion did NOT score at its own best-attainable pointwise value under objective – for accuracy_objective() this is exactly “the cases it got wrong.” A scalar-only objective (no honest per-observation vector) harvests the whole batch – there is no finer signal to rank on.

Parameters:

• champion (Any)

• batch (Sequence[Any])

• objective (Objective)

Return type:

list[Any]

harvested_from_router(router)

Pull the REAL harvested failure pool from a live mixle.task.router.Router: every input that escalated all the way to the frontier, paired with its frontier label – exactly router.harvested(), just zipped into one pool for mixle.task.acquire.acquire().

Parameters:

router (Any)

Return type:

list[tuple[Any, Any]]

default_challenger_operators()

The three L1 challenger-production operators, each a reused ImprovementOperator (no bespoke fitting logic):

• "distill" -> AutoSelect – cold-start: fit a fresh model from the harvested pool, the “train a new model from teacher-labeled data” shape of distillation.

• "refine" -> Refit – warm-started refit of the champion’s own parameters on the harvested pool.

• "evolve" -> Mutate – genetic-programming structure edit (grow/shrink/perturb), mixle.evolve’s own structure-search operator.

Return type:

dict[str, ImprovementOperator]

principled_crossover(model_a, model_b, *, weight_a=0.5)

Combine two champions the ONLY principled way this framework allows: mixture composition (mixle.ops.mixture()), never gene-splicing (cutting/pasting incompatible weight sub-blocks). Returns an unfitted mixture prototype; refit it against data (e.g. via its own .estimator()) before treating it as a challenger.

Parameters:

• model_a (Any)

• model_b (Any)

• weight_a (float)

Return type:

Any

class ConceptLibrary(base_families=('gaussian',))

Bases: object

A small, inspectable registry of admitted concept families – the “library” of CARD L6.

Starts with base_families (present from the outset, never revocable via revoke() – they are the starting library, not something this loop discovered). Everything admitted afterwards is a first-class, named, evidenced entry that can be queried by task signature and, if it turns out to be a mistake, genuinely revoked – removed from both the active family set and the design-prior ledger that backs query(), not merely hidden.

Parameters:

base_families (Sequence[str])

property families: tuple[str, ...]

the starting base plus everything still admitted.

Type:

Every family currently in the library

admit(family, evidence, *, task_signature, task_index, quality)

Admit family to the library, receipted with evidence and recorded in the design-prior ledger under task_signature so query() can recommend it later.

Parameters:

• family (str)

• evidence (dict[str, Any])

• task_signature (str)

• task_index (int)

• quality (float)

Return type:

None

revoke(family, *, task_index=-1, reason='')

Undo an admission: family leaves the active family set AND the design-prior ledger, so query() can never recommend it again – genuinely reversible, not soft-hidden.

Parameters:

• family (str)

• task_index (int)

• reason (str)

Return type:

None

query(signature)

The design-prior recommendation for signature: the best-recorded, still-admitted family seen for a matching task signature, or None if nothing qualifies.

Parameters:

signature (str)

Return type:

str | None

class AdmissionEvent(action, family, task_index, evidence=<factory>)

Bases: object

One receipted, timestamped-by-task-index entry in a ConceptLibrary’s audit log.

Parameters:

• action (str)

• family (str)

• task_index (int)

• evidence (dict[str, Any])

class TaskResult(task_index, task_signature, discrepancy, challenger_family, reused_concept, admitted_family, verdict, mdl_gain_bits)

Bases: object

One task’s pass through the loop – the receipt the acceptance test reads.

Parameters:

• task_index (int)

• task_signature (str)

• discrepancy (float)

• challenger_family (str | None)

• reused_concept (bool)

• admitted_family (str | None)

• verdict (Verdict | None)

• mdl_gain_bits (float)

run_concept_discovery_loop(tasks, *, library=None, champion_family='gaussian', objective=None, train_frac=0.6, recurrence_window=2, discrepancy_threshold=0.003, min_effect=0.01, nonnested=False)

Run discrepancy -> propose -> verify -> adopt -> remember over tasks, in order.

For every task: fit the champion_family on a train split, measure discrepancy against a held-out split (mixle.epistemic.discrepancy.discrepancy_report()). If the library already has an admitted concept recommended for this task’s signature (ConceptLibrary.query()), that concept is tried directly – no re-search. Otherwise, once high discrepancy has recurred for recurrence_window consecutive tasks sharing a signature with no admitted concept yet, the automatic profiler proposes a new family; it is gated via mixle.evolve.verify.challenger_beats_champion() and, only if it passes, admitted.

Returns the (possibly newly-created) ConceptLibrary and one TaskResult per task.

Parameters:

• tasks (Sequence[Sequence[float]])

• library (ConceptLibrary | None)

• champion_family (str)

• objective (Objective | None)

• train_frac (float)

• recurrence_window (int)

• discrepancy_threshold (float)

• min_effect (float)

• nonnested (bool)

Return type:

tuple[ConceptLibrary, list[TaskResult]]

task_signature(data)

A coarse, reusable descriptor of a task’s data-generating shape.

Deliberately coarse (kind + skew sign, not a fingerprint of the exact data): the point is that different tasks sharing the same hidden family land on the same signature, so ConceptLibrary.query() can recommend a concept discovered on an earlier task to a later one.

Parameters:

data (Sequence[float])

Return type:

str

Submodules

• mixle.evolve.closed_loop module
• mixle.evolve.concept_discovery module
• mixle.evolve.improve module
• mixle.evolve.ledger module
• mixle.evolve.objective module
• mixle.evolve.operators module
• mixle.evolve.population module
• mixle.evolve.search module
• mixle.evolve.space module
• mixle.evolve.structure module
• mixle.evolve.verify module