mixle.enumeration.algorithms module

Back-compat facade for the smart-enumeration machinery.

The implementation was split by concern (it had grown into one ~1100-line module mixing three distinct things):

• generic stream primitives -> mixle.enumeration.streams

• best-first / product search -> mixle.enumeration.best_first

• the quantized seek / unrank index -> mixle.enumeration.quantization.seek

This module re-exports them so from mixle.enumeration.algorithms import X keeps working unchanged. New code can import from the modules above directly. See mixle.stats.compute.pdist.DistributionEnumerator for the enumeration contract.

class BufferedStream(it)

Bases: object

Random access by rank into a lazy stream of (value, log_prob) pairs.

get(i) extends an internal buffer as needed and returns the i-th item, or None if the stream has fewer than i+1 items. The underlying stream is consumed at most once regardless of how many consumers share this object.

Parameters:

it (Iterator[tuple[Any, float]])

get(i)

Parameters:

i (int)

Return type:

tuple[Any, float] | None

freeze(x)

Return a canonical hashable key for x, for de-duplicating support values.

Lists/tuples freeze element-wise to tuples, dicts to frozensets of (key, value) pairs, sets to frozensets, numpy arrays to (shape, bytes), numpy scalars to their python equivalents, and NaN to a shared sentinel (so nan == nan for dedup purposes). Raises TypeError for values that cannot be canonicalized.

Parameters:

x (Any)

Return type:

Hashable

merge_enumerators(streams, offsets)

Lazy k-way merge of sorted (value, log_prob) streams with per-stream offsets.

Stream k’s log probs are shifted by offsets[k]. Correct only when the streams have pairwise disjoint supports (no de-duplication or re-scoring is performed).

Parameters:

• streams (Sequence[Iterator[tuple[Any, float]]])

• offsets (Sequence[float])

Return type:

Iterator[tuple[Any, float]]

supports_enumeration(dist)

Return True if dist.enumerator() can be constructed.

Return type:

bool

class ProductEnumerator(streams, combine=tuple, offset=0.0)

Bases: object

Best-first enumeration of the Cartesian product of sorted child streams.

Yields (combine(values), log_prob) with log_prob = offset + sum of child log probs, in non-increasing order. Standard k-best lattice search: a max-heap over index tuples, with successors advancing one coordinate; correctness follows from each child stream being sorted (coordinate-wise monotonicity).

Duplicate-free successor rule: each heap entry carries the lowest coordinate it is allowed to advance, and a node only spawns successors for coordinates at or above that index. Every tuple is then reached by exactly one path – the one that increments coordinates in non-decreasing index order – so no visited set (and no O(n) per-successor tuple hashing) is needed. Best-first order is preserved because each successor’s score is <= its generating node’s (a sorted coordinate only decreases when advanced), so canonical edges are score-monotone.

Parameters:

• streams (Sequence[BufferedStream])

• combine (Callable[[tuple[Any, ...]], Any])

• offset (float)

class LengthFrontierMerge(len_stream, make_stream)

Bases: object

Merge per-length sorted streams, instantiating lengths lazily from a sorted length stream.

len_stream yields (length, log_prob_of_length) in descending order. make_stream(length, lp_len) returns a sorted iterator of (value, log_prob) whose log probs already include lp_len and never exceed it (true whenever per-element contributions are log probs <= 0). The next un-instantiated length’s lp_len is then a valid upper bound on anything its stream could produce, so lengths are instantiated only when they can beat the best instantiated head. Supports of distinct lengths must be disjoint (no de-duplication).

Parameters:

• len_stream (BufferedStream)

• make_stream (Callable[[int, float], Iterator[tuple[Any, float]]])

frontier_merge(outer_stream, make_stream)

Merge per-outer-key inner streams, instantiating outer keys lazily from a sorted outer stream.

The general form of LengthFrontierMerge for arbitrary (not necessarily integer) outer keys: outer_stream yields (key, lp_key) in descending lp_key order, and make_stream(key, lp_key) returns a sorted (value, log_prob) iterator whose log-probs are <= lp_key (true whenever the inner contribution is <= 0 and lp_key is folded in as an offset). The next un-instantiated key’s lp_key is then an upper bound on anything its stream could produce, so keys are instantiated only when they can beat the best instantiated head. Supports of distinct keys must be disjoint (no de-duplication). Used for conditional products where the inner distribution depends on the outer value (e.g. hidden-association S2 given S1).

Parameters:

• outer_stream (BufferedStream)

• make_stream (Callable[[Any, float], Iterator[tuple[Any, float]]])

Return type:

Iterator[tuple[Any, float]]

best_first_union(streams, log_offsets, exact_log_density, tol=1.0e-10)

Enumerate the union of sorted streams with overlapping supports.

Candidate values are pulled from the streams (stream k shifted by log_offsets[k]), de-duplicated via freeze, re-scored exactly with exact_log_density, and buffered until their exact score is at least the upper bound on any not-yet-seen value: logsumexp_k(log_offsets[k] + head_lp_k). This is the mixture algorithm: any unseen x satisfies p_k(x) <= head_k for every k, hence sum_k w_k p_k(x) <= exp(bound).

Parameters:

• streams (Sequence[BufferedStream])

• log_offsets (Sequence[float])

• exact_log_density (Callable[[Any], float])

• tol (float)

Return type:

Iterator[tuple[Any, float]]

best_first_union_max(streams, log_offsets, exact_log_density, tol=1.0e-10)

Like best_first_union, but for a max-scored union (bound = max over heads).

Used to enumerate a deduped symbol pool ordered by max-over-states emission probability for markov/HMM enumerators.

Parameters:

• streams (Sequence[BufferedStream])

• log_offsets (Sequence[float])

• exact_log_density (Callable[[Any], float])

• tol (float)

Return type:

Iterator[tuple[Any, float]]

bounded_best_first_union_index(streams, log_offsets, exact_log_density, max_bits, bin_width_bits=1.0, tol=1.0e-10, component_log_density=None)

Build a quantized index from a globally bounded best-first union.

This is the index-building counterpart of best_first_union. It is useful for mixtures: if each stream head bounds one component contribution, then the log-sum of live heads bounds every unseen value. Once that global frontier is below the requested probability threshold, all remaining qualifying values must already be in the exact-score buffer, so the index can stop without enumerating a looser per-component candidate prefix.

Parameters:

• streams (Sequence[BufferedStream])

• log_offsets (Sequence[float])

• exact_log_density (Callable[[Any], float])

• max_bits (float)

• bin_width_bits (float)

• tol (float)

• component_log_density (Callable[[int, Any], float] | None)

Return type:

QuantizedEnumerationIndex

sound_top_k(dist, k, start=0, budget_bits=40.0, oversample=8, bin_width_bits=1.0, max_budget_bits=512.0, total_mass=1.0, tol=1.0e-12)

Exact true-descending observations ranked [start, start+k) for ANY normalized model.

Mass-threshold certificate, correct regardless of the seek stream’s ordering (so it holds for deep/nested mixtures and HMMs where the tropical order is badly displaced): pull distinct (value, log_prob) from the count-budget index, accumulate exact probability mass, and keep the best start+k by probability. Since every unpulled item’s probability is at most the remaining mass total_mass - accumulated, once remaining drops below the (start+k)-th best probability the heap is exactly the true top start+k; return sorted_desc[start:start+k]. If the in-budget stream is exhausted first, the budget is doubled (up to max_budget_bits). total_mass (default 1.0) may be a known upper bound for sub-normalized models.

Cost is O(number pulled) – small for peaked distributions, large for flat ones, so for top-k from rank 0 dist.enumerator() is usually leaner; this adds the soundness certificate and the arbitrary start offset the sequential enumerator lacks.

Parameters:

• dist (Any)

• k (int)

• start (int)

• budget_bits (float)

• oversample (int)

• bin_width_bits (float)

• max_budget_bits (float)

• total_mass (float)

• tol (float)

Return type:

list[tuple[Any, float]]

class QuantizedEnumerationIndex(bins, bin_width_bits, max_bits, truncated)

Bases: object

Bounded, indexable view of an exact probability-ordered enumeration.

Items are grouped into log-probability bins measured in bits:

bits(x) = -log2(p(x)) bin(x) = floor(bits(x) / bin_width_bits)

Only items whose bit cost is at most max_bits are indexed. The returned (value, log_prob) pairs always carry the original exact log probability; the quantized bins are used only for counting and rank lookup.

The generic implementation can build from an exact enumerator stream, so it works for every currently enumerable discrete stats model. Simple distributions may also build directly from scored support items, and distribution-specific dynamic programs can produce the same bin layout without relying on an exact global stream.

Parameters:

• bins (Sequence[tuple[int, list[tuple[Any, float]]]])

• bin_width_bits (float)

• max_bits (float)

• truncated (bool)

static bin_for_log_prob(log_prob, bin_width_bits=1.0)

Return the quantized bit bin for a log probability.

Parameters:

• log_prob (float)

• bin_width_bits (float)

Return type:

int

classmethod from_enumerator(enum, max_bits, bin_width_bits=1.0)

Build an index from an exact non-increasing-probability enumerator.

Parameters:

• enum (Iterator[tuple[Any, float]]) – Iterator yielding (value, log_prob) pairs in exact descending order.

• max_bits (float) – Include only values with -log2(p) <= max_bits.

• bin_width_bits (float) – Width of each quantized probability bin in bits.

Returns:

QuantizedEnumerationIndex over the bounded prefix/domain slice.

Return type:

QuantizedEnumerationIndex

classmethod from_items(items, max_bits, bin_width_bits=1.0, sorted_items=False, truncated=None)

Build an index from known support items and exact log probabilities.

Parameters:

• items (Sequence[tuple[Any, float]]) – Finite collection of (value, log_prob) pairs. Zero-probability items with log_prob == -inf are ignored.

• max_bits (float) – Include only values with -log2(p) <= max_bits.

• bin_width_bits (float) – Width of each quantized probability bin in bits.

• sorted_items (bool) – If True, preserve the input order. Otherwise, stable-sort included items by descending log probability.

• truncated (bool | None) – Optional explicit truncation flag. If omitted, it is True when any finite-probability item was outside the bit bound.

Returns:

QuantizedEnumerationIndex over the bounded finite item set.

Return type:

QuantizedEnumerationIndex

bin_for_index(index)

Return (bin_id, offset_within_bin) for a bounded quantized rank.

Parameters:

index (int)

Return type:

tuple[int, int]

get(index)

Return the indexed (value, exact_log_prob) pair.

Parameters:

index (int)

Return type:

tuple[Any, float]

slice(start, k)

Return up to k indexed pairs starting at start.

Parameters:

• start (int)

• k (int)

Return type:

list[tuple[Any, float]]

iter_from(start=0)

Iterate indexed pairs from start to the end of the bounded index.

Parameters:

start (int)

Return type:

Iterator[tuple[Any, float]]

bin_items(bin_id)

Return all indexed items in a quantized probability bin.

Parameters:

bin_id (int)

Return type:

list[tuple[Any, float]]

summary()

Return a compact description of the bounded index.

Return type:

dict[str, Any]

class LazyQuantizedEnumerationIndex(counts, bin_width_bits, max_bits, truncated, getter)

Bases: QuantizedEnumerationIndex

Quantized index whose bin counts are precomputed but items are unranked lazily.

This supports compositional distributions where a dynamic program can count how many values fall in each quantized log-density bin without materializing every Cartesian-product value. The getter receives a bin id and offset within that bin and returns the exact (value, log_prob) pair for that quantized rank.

Parameters:

• counts (dict[int, int])

• bin_width_bits (float)

• max_bits (float)

• truncated (bool)

• getter (Callable[[int, int], tuple[Any, float]])

bin_for_index(index)

Return (bin_id, offset_within_bin) for a bounded quantized rank.

Parameters:

index (int)

Return type:

tuple[int, int]

get(index)

Return the indexed (value, exact_log_prob) pair.

Parameters:

index (int)

Return type:

tuple[Any, float]

iter_from(start=0)

Iterate indexed pairs from start to the end of the bounded index.

Parameters:

start (int)

Return type:

Iterator[tuple[Any, float]]

bin_items(bin_id)

Return all indexed items in a quantized probability bin.

Parameters:

bin_id (int)

Return type:

list[tuple[Any, float]]

class QuantizedCrossIndex(items, max_bits, bin_width_bits, truncated=False)

Bases: object

Aligned support rows for multiple distributions under quantized bit bounds.

Each row is (value, log_probs), where log_probs[j] is the exact log-density of value under component j, or -inf when that component assigns zero mass. Rows are included when any component’s bit cost is within its requested bound. The joint bin counts expose the cross-bin alignment that mixtures need and marginal bin counts cannot provide.

Parameters:

• items (Sequence[tuple[Any, Sequence[float]]])

• max_bits (Sequence[float])

• bin_width_bits (float)

• truncated (bool)

static bits_for_log_prob(log_prob)

Return -log2(p), using inf for zero probability.

Parameters:

log_prob (float)

Return type:

float

classmethod from_items(items, max_bits, bin_width_bits=1.0, truncated=False)

Build a cross index from exact aligned support rows.

Parameters:

• items (Sequence[tuple[Any, Sequence[float]]])

• max_bits (Sequence[float])

• bin_width_bits (float)

• truncated (bool)

Return type:

QuantizedCrossIndex

iter_items()

Iterate aligned support rows.

Return type:

Iterator[tuple[Any, tuple[float, …]]]

summary()

Return a compact description of the cross index.

Return type:

dict[str, Any]

quantized_index(enum, max_bits, bin_width_bits=1.0)

Convenience wrapper for QuantizedEnumerationIndex.from_enumerator.

Parameters:

• enum (Iterator[tuple[Any, float]])

• max_bits (float)

• bin_width_bits (float)

Return type:

QuantizedEnumerationIndex