mixle.stats.combinator.sequence module¶

Create, estimate, and sample from a sequence of iid sequence of base distribution ‘dist’ with data type T. A length distribution for the lengths of the iid sequences can be specified as a discrete distribution compatible with non-negative integer values.

Defines the SequenceDistribution, SequenceSampler, SequenceAccumulatorFactory, SequenceAccumulator, SequenceEstimator, and the SequenceDataEncoder classes for use with mixle.

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]

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]

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]

What log_density returns relative to the true log-density (default: exact).

Override to declare that this distribution’s log_density is a variational lower bound (ELBO), an upper bound, or an approximation rather than the exact log p(x). This is surfaced as the ExactDensity capability and noted in mixle.describe(), so code that needs an exact likelihood can require(x, ExactDensity) instead of silently trusting a bound.

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 SequenceSampler object from instance of SequenceDistribution.

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]

Create SequenceEstimator from instance of SequenceDistribution with pseudo_count passed if not None.

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]

Create SequenceDataEncoder for encoding sequences of iid observations of SequenceDistribution.

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

Parameters:: dist (SequenceDistribution)

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

Bases: DistributionSampler

Parameters:

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

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

Generate iid samples from SequenceSampler object.

If size is None, the length ‘n’ of the iid sequence is sampled from len_sampler. Then ‘n’ iid samples are drawn from the base dist sampled ‘dist_sampler’.

If size > 0, above is repeated size times and a List of size List[T] is retured.

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

Parameters:

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

update(x, weight, estimate)[source]

Update sufficient statistics for SequenceAccumulator object.

Aggregate sufficient statistics of base accumulator and length accumulator with sufficient statistics of ‘estimate’ if passed.

Note: Calls update() of accumulator and len_accumulator.

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 SequenceAccumulator object with 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) – RandomState object used to set seed in random 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) – RandomState object used to set seed in random 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]

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]

Create SequenceDataEncoder for encoding sequences of iid observations of SequenceDistribution.

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

Parameters:

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

make()[source]

Return SequenceAccumulator with SequenceEncodableStatisticAccumulator objects created from dist_factory and len_factory.

Return type:: SequenceAccumulator

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

Bases: ParameterEstimator

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]

Parameters:

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

Return type:

SequenceDistribution

class SequenceDataEncoder(encoders)[source]

Bases: DataSequenceEncoder

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

seq_encode(x)[source]

Encode iid observations of Sequence distribution for use with vectorized “seq_” functions.

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]

Empirical Fisher approximation from per-observation statistic vectors.

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 centered/whitened sufficient-statistic vectors.

metric=’identity’ returns centered statistics, ‘diagonal’ divides by per-coordinate Fisher standard deviations, and ‘full’ applies an empirical full-matrix whitening transform.

Parameters:

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

Return type:

ndarray