mixle.stats.combinator.sequence module¶

Sequence distributions with iid elements and an optional length model.

Data type (T): Assume the sequence distribution has a base distribution ‘dist’ compatible with data type T and length distribution compatible with positive integers len_dist with respective densities P_dist() and P_len(). The density of the sequence distribution is given by

p_mat(x) = P_dist(x[0])*…*P_dist(x[n-1])*P_len(n),

for an observation x of data type Sequence[T] having length n.

class SequenceDistribution(dist, len_dist=NullDistribution(), len_normalized=False, name=None, prior=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Independent sequence distribution built from a component observation distribution.

Parameters:

dist (SequenceEncodableProbabilityDistribution)
len_dist (SequenceEncodableProbabilityDistribution | None)
len_normalized (bool | None)
name (str | None)
prior (tuple[Any, Any] | None)

compute_capabilities()[source]: Return compute-backend metadata inherited from the element and length distributions.

get_prior()[source]

Return the joint prior as (entry_prior, length_prior) from the wrapped children.

Return type:: tuple[Any, Any]

set_prior(prior)[source]

Distribute (entry_prior, length_prior) to the base and length distributions.

prior=None is a no-op (children keep their existing priors, leaving the MLE path byte-identical); otherwise the two-element prior is pushed to the base and length children via their own set_prior.

Parameters:: prior (tuple[Any, Any] | None)
Return type:: None

expected_log_density(x)[source]

Prior-expected log-density of the sequence x (sum over entries + length term).

Parameters:: x (Sequence[T])
Return type:: float

seq_expected_log_density(x)[source]

Vectorized prior-expected log-density over sequence-encoded input x.

Parameters:: x (tuple[ndarray, ndarray, ndarray, E1, E2 | None])
Return type:: ndarray

compute_declaration()[source]: Return the symbolic declaration for sequence elements and optional length statistics.

density(x)[source]

Evaluate the density of SequenceDistribution at observed sequence x.

Assume x is a Sequence of data type T with length n > 0. Assume P_dist() is the density for the base distribution with data type T of SequenceDistribution, and P_len() is the length distribution with data type int. Then,

P(x) = P_dist(x[0])*…*P_dist(x[n-1])*P_len(n), if len_normalize is False,

or,

P(x) = (P_dist(x[0])*…*P_dist(x[n-1])*P_len(n))^(1/n) if len_normalize is True.

Parameters:: x (Sequence[T]) – Sequence of iid observations from base distribution of SequenceDistribution.
Returns:: Density evaluated at observation x.
Return type:: float

density_semantics()[source]: Return the joined density semantics of the element distribution.

log_density(x)[source]

Evaluate the log-density of SequenceDistribution at observed sequence x.

See density() for details.

Parameters:: x (Sequence[T]) – Sequence of iid observations from base distribution of SequenceDistribution.
Returns:: Log-density evaluated at observation x.
Return type:: float

seq_ld_lambda()[source]: Return vectorized log-density callables for encoded data.

seq_log_density(x)[source]

Vectorized evaluation of SequenceDistribution.log-density evaluated on sequence encoded x.

Parameters:: x (E) – Sequence encoded data observation.
Returns:: Numpy array of log-density evaluated at each encoded observation value x.
Return type:: ndarray

backend_seq_log_density(x, engine)[source]

Engine-neutral vectorized log-density for encoded sequences.

Parameters:

x (tuple[ndarray, ndarray, ndarray, E1, E2 | None])
engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)[source]

Return stacked child routes for homogeneous sequence mixtures.

Parameters:

dists (Sequence[SequenceDistribution])
engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)[source]

Return an (n, k) matrix of sequence log densities.

Parameters:

x (tuple[ndarray, ndarray, ndarray, E1, E2 | None])
params (dict[str, Any])
engine (Any)

Return type:

Any

classmethod backend_stacked_sufficient_statistics_with_estimator(x, weights, params, engine, estimator)[source]

Return per-component legacy sequence sufficient statistics.

Parameters:

x (tuple[ndarray, ndarray, ndarray, E1, E2 | None])
weights (Any)
params (dict[str, Any])
engine (Any)
estimator (Any)

Return type:

tuple[Any, …]

gradient_fit_state(engine, torch, leaves, recurse, tensor_param)[source]

Return distribution-owned state for autograd fitting.

Parameters:

engine (Any)
torch (Any)
leaves (list[Any])
recurse (Any)
tensor_param (Any)

Return type:

Any

to_fisher(**kwargs)[source]: Structural Fisher view for the sequence.

to_exponential_family(engine=None)[source]

Return the iid exponential-family view, or None.

An iid sequence of an exponential family is itself an exponential family with the shared element eta and T(x) = sum_t T_0(x_t). This holds only when the length is not separately modeled and not length-normalized (a length term or normalization breaks the single-exp-family form); otherwise returns None, as does a non-exp-family element.

Parameters:: engine (Any)

sampler(seed=None)[source]

Create a sampler for this sequence distribution.

Note: If member len_dist (SequenceEncodableDistribution) is NullDistribution() and or not compatible with data type int, an error is thrown.

Parameters:: seed (Optional[int]) – Used to set seed of random number generator used to sample.
Returns:: SequenceSampler object.
Return type:: SequenceSampler

estimator(pseudo_count=None)[source]

Return an estimator for iid sequence observations and their lengths.

Parameters:: pseudo_count (float | None)
Return type:: SequenceEstimator

decomposition()[source]: Sequences are iid: this names the data (sequence) axis, sufficient stats SUM-reduce. The planner sizes this axis from N (the data), not a fixed model count; the base dist may itself decompose.

dist_to_encoder()[source]

Return an encoder for iid sequence observations.

Base distribution DataSequenceEncoder and length distribution DataSequenceEncoder objects are passed.

Returns:: SequenceDataEncoder object.
Return type:: SequenceDataEncoder

enumerator()[source]

Returns SequenceEnumerator iterating sequences in descending probability order.

Return type:: SequenceEnumerator

structural_fine_bucket(value, quantizer)[source]

Sum of per-element buckets plus the length term – mirrors the count index’s per-length L-fold element convolution shifted by the length-term bucket.

Return type:: int

quantized_count_index(quantizer, max_fine_bucket)[source]

Structural count index: per-length L-fold self-convolution of the element histogram.

log p(x) = sum_i log p(x_i) + log p(len(x)). For each length L the count histogram of the L-element sum is the L-fold self-convolution of the element count histogram, shifted by the length term’s bucket; the total is the pooled sum over lengths. Lengths come from the length distribution’s enumerator in descending probability, so once the length term alone exceeds the depth bound every later length does too and we stop. Sequences are unranked by resolving the contributing length, then the per-position element buckets via the convolution unranker.

Parameters:: max_fine_bucket (int)

class SequenceEnumerator(dist)[source]

Bases: DistributionEnumerator

Best-first enumerator for finite-length iid sequences with a modeled length law.

Parameters:: dist (SequenceDistribution)

class SequenceSampler(dist, len_dist, seed=None)[source]

Bases: DistributionSampler

Sampler for iid sequences whose lengths are drawn from a length distribution.

Parameters:

dist (SequenceEncodableProbabilityDistribution)
len_dist (SequenceEncodableProbabilityDistribution)
seed (int | None)

sample(size=None, *, batched=True)[source]

Generate iid sequence samples.

If size is None, the length n is sampled from len_sampler and then n iid elements are drawn from the base distribution sampler.

If size is positive, the process is repeated and a list of sequences is returned.

With batched=True (default) all lengths are drawn at once and all elements across every sequence are drawn in a single vectorized call, then split by length. Because len_sampler and dist_sampler own independent RandomState streams, this yields the same draws as the legacy per-element loop (batched=False) but far faster.

Parameters:

size (Optional[int])
batched (bool) – Vectorize element draws (default); set False for the legacy per-draw loop.

Returns:

List[T] or List[List[T]] with length(size).

Return type:

list[Any]

class SequenceAccumulator(accumulator, len_accumulator=NullAccumulator(), len_normalized=False, keys=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator that delegates element and length sufficient statistics to child accumulators.

Parameters:

accumulator (SequenceEncodableStatisticAccumulator)
len_accumulator (SequenceEncodableStatisticAccumulator)
len_normalized (bool | None)
keys (str | None)

update(x, weight, estimate)[source]

Update sequence sufficient statistics with one weighted observation.

The element accumulator receives the sequence contents and the length accumulator receives the sequence length.

Parameters:

x (Sequence[T]) – A sequence of iid observations of data type T.
weight (float) – Weight for sequence observation.
estimate (Optional[SequenceDistribution]) – SequenceDistribution instance to aggregate sufficient statistics with.

Returns:

None.

Return type:

None

initialize(x, weight, rng)[source]

Initialize sequence sufficient statistics from one weighted observation.

Note: Calls _rng_initialize() method if _len_rng has not been set. This ensures consistency between initialize and seq_initialize calls.

Method invokes calls to accumulator.initialize() and len_accumulator.initialize() if len_accumulator is not NullAccumulator.

Parameters:

x (Sequence[T]) – Sequence of iid observations from base distribution.
weight (float) – Weight for sequence observation.
rng (RandomState) – Random state used during initialization.

Returns:

None.

Return type:

None

seq_initialize(x, weights, rng)[source]

Vectorized initialization of SequenceAccumulator sufficient statistics from sequence encoded x.

Parameters:

x (E) – Encoded data sequence.
weights (np.ndarray[float]) – Numpy array of floats for weighting observations.
rng (RandomState) – Random state used during initialization.

Returns:

None.

Return type:

None

seq_update(x, weights, estimate)[source]

Vectorized update of SequenceAccumulator sufficient statistics from sequence encoded x.

Parameters:

x (E) – Encoded data sequence.
weights (np.ndarray[float]) – Numpy array of floats for weighting observations.
estimate (Optional[SequenceDistribution]) – SequenceDistribution instance to aggregate sufficient statistics with.

Returns:

None.

Return type:

None

seq_update_engine(x, weights, estimate, engine)[source]

Engine-resident E-step: per-element weights are gathered/normalized on the active engine and the base/length accumulators are routed through the engine. Matches seq_update.

Parameters:

x (tuple[ndarray, ndarray, ndarray, E1, E2 | None])
weights (Any)
estimate (SequenceDistribution | None)
engine (Any)

Return type:

None

combine(suff_stat)[source]

Combine the sufficient statistics of SequenceAccumulator instance with suff_stat arg.

Parameters:: suff_stat (Tuple[SS1, Optional[SS2]]) – Tuple of sufficient statistics of base distribution and value for length distribution.
Returns:: SequenceAccumulator object.
Return type:: SequenceAccumulator

value()[source]

Return Tuple[SS1, Optional[SS2]], sufficient statistics of base accumulator and length accumulator.

Return type:: tuple[Any, Any | None]

from_value(x)[source]

Set the SequenceAccumulator base accumulator and length accumulator to values of x.

Parameters:: x (Tuple[SS1, Optional[SS2]]) – Tuple of sufficient statistics of base distribution and value for length distribution.
Returns:: SequenceAccumulator object.
Return type:: SequenceAccumulator

scale(c)[source]

Scale element and length sufficient statistics through their accumulators.

Parameters:: c (float)
Return type:: SequenceAccumulator

get_seq_lambda()[source]: Return low-level sequence kernels from the element and length accumulators.

key_merge(stats_dict)[source]

Merges member sufficient statistics with sufficient statistics that contain matching keys.

Parameters:: stats_dict (Dict[str, Any]) – Dictionary mapping keys to sufficient statistics.
Returns:: None.
Return type:: None

key_replace(stats_dict)[source]

Set member sufficient statistics to values of objects with matching keys.

Parameters:: stats_dict (Dict[str, Any]) – Dictionary mapping keys to sufficient statistics.
Returns:: None.
Return type:: None

acc_to_encoder()[source]

Return an encoder for iid sequence observations.

Base distribution DataSequenceEncoder and length distribution DataSequenceEncoder objects are passed.

Returns:: SequenceDataEncoder object.
Return type:: SequenceDataEncoder

class SequenceAccumulatorFactory(dist_factory, len_factory=NullAccumulatorFactory(), len_normalized=False, keys=None)[source]

Bases: StatisticAccumulatorFactory

Factory for sequence accumulators composed from element and length accumulator factories.

Parameters:

dist_factory (StatisticAccumulatorFactory)
len_factory (StatisticAccumulatorFactory)
len_normalized (bool | None)
keys (str | None)

make()[source]

Return a fresh sequence accumulator from the element and length factories.

Return type:: SequenceAccumulator

class SequenceEstimator(estimator, len_estimator=NullEstimator(), len_dist=NullDistribution(), len_normalized=False, name=None, keys=None, prior=None)[source]

Bases: ParameterEstimator

Estimator for iid sequence distributions from element and optional length sufficient statistics.

Parameters:

estimator (ParameterEstimator)
len_estimator (ParameterEstimator | None)
len_dist (SequenceEncodableProbabilityDistribution | None)
len_normalized (bool | None)
name (str | None)
keys (str | None)
prior (tuple[Any, Any] | None)

get_prior()[source]

Return the joint prior as (entry_prior, length_prior) from the child estimators.

Return type:: tuple[Any, Any]

set_prior(prior)[source]

Distribute (entry_prior, length_prior) to the entry and length estimators.

prior=None is a no-op. The prior is pushed to the entry/length estimators (not to a fixed len_dist), so each child performs its own conjugate update.

Parameters:: prior (tuple[Any, Any] | None)
Return type:: None

model_log_density(model)[source]

Sum the entry and length estimators’ model_log_density on the corresponding children.

Parameters:: model (SequenceDistribution)
Return type:: float

accumulator_factory()[source]

Return SequenceAccumulatorFactory from len_estimator and estimator member variables with keys passed.

Return type:: SequenceAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate the element distribution and, when configured, the length distribution.

Parameters:

nobs (float | None)
suff_stat (tuple[Any, Any | None])

Return type:

SequenceDistribution

class SequenceDataEncoder(encoders)[source]

Bases: DataSequenceEncoder

Encoder that flattens sequence elements and separately encodes sequence lengths.

Parameters:: encoders (tuple[DataSequenceEncoder, DataSequenceEncoder])

seq_encode(x)[source]

Encode iid sequence observations for vectorized seq_* methods.

Data ‘x’ must be a Sequence of Sequences containing data types T consistent with the distribution encoder (DataSequenceEncoder) object ‘encoder’. That is x: Sequence containing ‘N’ objects of xx: Sequence[T].

Consider example data x = [ [0,1,2], [],[3,4]]. Then x: Sequence[Sequence[int]].

Assume the data type returned by ‘encoder.seq_encode()’ is T1, and ‘len_encoder.seq_encode()’ is T2.

rv1 (ndarray[int]): Index for values of positive length sequence entries. I.e. x produces -> [0,0,0,2,2] rv2 (ndarray[float]): Inverse of sequence lengths. I.e. x -> [1/3,1/3,1/3,0,1/2,1/2] rv3 (ndarray[bool]): True if length of sequence is not 0. I.e. x -> [True,True, True, False, True,True] rv4 (T1): Sequence encoding resulting from encoder.seq_encode() on list of all observed values. rv5 (Optional[T2]): Sequence encoding resulting len_encoder.seq_encode() on all sequence length values.

Parameters:: x (Sequence[Sequence[T]]) – Sequence of Sequence[T], where T is compatible with base distribution of sequence distribution. Sequence of iid sequence observations.
Return type:: tuple[ndarray, ndarray, ndarray, tuple[Any, …], Any | None]

Returns:

class SequenceFisherView(dist)[source]

Bases: FixedFisherView

Structured Fisher view for iid sequence distributions.

Parameters:: dist (Any)

fisher_information(stats=None, diagonal=False, ridge=1.0e-8, **kwargs)[source]

Return Fisher information, falling back to the generic Fisher view when needed.

Parameters:

stats (ndarray | None)
diagonal (bool)
ridge (float)
kwargs (Any)

Return type:

ndarray

fisher_vectors(stats=None, metric='diagonal', center=None, fisher=None, ridge=1.0e-8, **kwargs)[source]

Return Fisher-normalized sequence statistics, with a generic fallback.

Parameters:

stats (ndarray | None)
metric (str)
center (ndarray | None)
fisher (ndarray | None)
ridge (float)
kwargs (Any)

Return type:

ndarray