mixle.stats.latent.lookback_hidden_markov_model module

Evaluate, estimate, and sample from a lookback hidden Markov model (typed rewrite).

Defines the LookbackHiddenMarkovModelDistribution, LookbackHiddenMarkovModelSampler, LookbackHiddenMarkovModelEstimatorAccumulator, LookbackHiddenMarkovModelEstimatorAccumulatorFactory, LookbackHiddenMarkovModelEstimator, and the LookbackHiddenMarkovModelDataEncoder classes for use with mixle.

A lookback hidden Markov model is a hidden Markov model whose emission distributions condition on the previous lag observations: with hidden states Z(t) following a Markov chain with initial state probabilities w and transition matrix A,

P(X(1),…,X(n)) = sum_z P(X(1:lag) | Z(1)=z_1) * w[z_1]

• prod_{t=lag+1}^{n} P(X(t) | X(t-lag:t-1), Z(t)=z_t) * A[z_{t-1}, z_t],

where the per-state topics distributions model windows x[t-lag:t+1] of length lag+1 (e.g. IntegerMarkovChainDistribution), and the per-state init_dist distributions model the first lag observations. An optional length distribution models the number of hidden positions: len(x) - lag + 1 (initial segment plus emission windows) when lag > 0, and len(x) when lag == 0.

With lag == 0 the model reduces to an ordinary hidden Markov model: there is no initial segment, init_dist is never evaluated, the first state is drawn from w and emits the window x[0:1], and each subsequent state emits x[t:t+1].

Data type: Sequence[T] - each observation is a sequence (e.g. a list) whose length-(lag+1) sliding windows have the data type accepted by the topic distributions, and whose first lag entries have the data type accepted by the initial distributions.

Note: This is the typed rewrite of the sibling module mixle.stats.lookback_hmm, which is the original implementation kept stable for the example scripts and external users. The math is identical, but the two modules differ slightly in their handling of optional arguments: this module substitutes Null* objects (NullDistribution, NullEstimator, NullDataEncoder, …) for an absent len_dist/init_dist, while the sibling uses None (and omits the length term from densities). The LookbackHiddenMarkovModelDataEncoder constructor signatures also differ (here: encoder first with an encoder attribute; sibling: lag first with a topic_encoder attribute).

class LookbackHiddenMarkovModelDistribution(topics, w=MISSING, transitions=MISSING, lag=0, init_dist=None, len_dist=NullDistribution(), name=None, weights=MISSING, terminal_states=None)

Bases: SequenceEncodableProbabilityDistribution

Hidden Markov model whose state emissions condition on the previous lag observations.

Parameters:

• topics (Sequence[SequenceEncodableProbabilityDistribution])

• w (ndarray)

• lag (int)

• init_dist (Sequence[SequenceEncodableProbabilityDistribution] | None)

• len_dist (SequenceEncodableProbabilityDistribution | None)

• name (str | None)

• weights (ndarray)

• terminal_states (set[int] | Sequence[int] | None)

density(x)

Evaluate the density of the distribution at sequence x.

Parameters:

x (Sequence[T]) – Observed sequence.

Returns:

Density at x.

Return type:

float

log_density(x)

Evaluate the log-density of the distribution at sequence x.

Marginalizes the hidden state path with a scaled forward pass. The initial segment x[:lag] is scored by init_dist, each window x[t-lag:t+1] by the topic distributions, and the number of hidden positions by len_dist. When lag == 0 there is no initial segment: the first state emits the window x[0:1] directly (ordinary HMM).

Parameters:

x (Sequence[T]) – Observed sequence with len(x) >= lag.

Returns:

Log-density at x.

Return type:

float

viterbi_sequence(x)

Compute the most likely hidden state sequence for observed sequence x.

Parameters:

x (Sequence[T]) – Observed sequence with len(x) >= lag.

Returns:

Integer array of len(x) - lag + 1 (len(x) when lag == 0) most likely hidden

state indices.

Return type:

np.ndarray

seq_log_density(x)

Vectorized evaluation of the log-density at encoded sequences x.

Parameters:

x – Encoded sequence data produced by seq_encode() / dist_to_encoder().

Returns:

Log-density value for each encoded sequence.

Return type:

np.ndarray

compute_capabilities()

Return backend capability metadata for this concrete lookback-HMM instance.

backend_seq_log_density(x, engine)

Engine-neutral lookback-HMM scoring via the shared HMM forward pass.

seq_posterior(x)

Compute posterior hidden state probabilities for encoded sequences x.

Parameters:

x – Encoded sequence data produced by seq_encode() / dist_to_encoder().

Returns:

For each sequence, an array of per-position posterior state

probabilities with shape (num_windows, num_states).

Return type:

List[np.ndarray]

density_semantics()

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.

sampler(seed=None)

Create a LookbackHiddenMarkovModelSampler for this distribution.

Parameters:

seed (Optional[int]) – Seed for random number generator.

Returns:

Sampler object (requires a non-null len_dist).

Return type:

LookbackHiddenMarkovModelSampler

enumerator()

Not supported: lookback emissions condition on the previous lag observations.

Unlike the standard / segmental HMM, each emission depends on the preceding lag symbols, so the effective state is (hidden_state, last lag observations). Best-first enumeration over that augmented (and, for large/continuous alphabets, unbounded) state space is not implemented; use sampler() or the exact log_density / viterbi_sequence instead.

Return type:

DistributionEnumerator

estimator(pseudo_count=None)

Create a LookbackHiddenMarkovModelEstimator from this distribution.

Parameters:

pseudo_count (Optional[float]) – Regularize the initial-state and transition estimates.

Returns:

Estimator built from the topic, initial-segment, and

length distributions, preserving the lag.

Return type:

LookbackHiddenMarkovModelEstimator

seq_encode(x)

Encode a sequence of observed sequences for vectorized ‘seq_’ calls.

Parameters:

x (Sequence[Sequence[T]]) – Sequence of iid observed sequences.

Returns:

Encoded data consistent with seq_log_density(), seq_posterior(), and seq_update().

dist_to_encoder()

Return a LookbackHiddenMarkovModelDataEncoder for encoding sequences of iid observations.

Returns:

Encoder built from the topic, initial, and length

distributions of this instance.

Return type:

LookbackHiddenMarkovModelDataEncoder

class LookbackTerminalDataEncoder

Bases: DataSequenceEncoder

Passthrough encoder for terminal-state lookback HMMs: keeps raw sequences (scored per sequence).

seq_encode(x)

Encode the iid observation sequence x for vectorized evaluation.

class LookbackHiddenMarkovModelSampler(dist, seed=None)

Bases: DistributionSampler

Sampler for LookbackHiddenMarkovModelDistribution. Requires non-null init_dist and len_dist.

Parameters:

• dist (LookbackHiddenMarkovModelDistribution)

• seed (int | None)

sample(size=None)

Draw iid sequences from the lookback hidden Markov distribution.

Parameters:

size (Optional[int]) – Number of sequences to draw. If None, a single sequence is returned.

Returns:

One sampled sequence if size is None, else a list of

size sampled sequences.

Return type:

Union[List[T], List[List[T]]]

class LookbackHiddenMarkovModelEstimatorAccumulator(seq_accumulators, init_accumulators=None, lag=0, len_accumulator=None, keys=(None, None, None), terminal_states=None)

Bases: SequenceEncodableStatisticAccumulator

Accumulator for sufficient statistics of a lookback hidden Markov model.

update(x, weight, estimate)

Update sufficient statistics with one observed sequence and weight.

Parameters:

• x (Sequence[T]) – Observed sequence.

• weight (float) – Weight for the observation.

• estimate (LookbackHiddenMarkovModelDistribution) – Current estimate used for the E-step.

initialize(x, weight, rng)

Initialize sufficient statistics with one observed sequence using random state weights.

Parameters:

• x (Sequence[T]) – Observed sequence.

• weight (float) – Weight for the observation.

• rng (np.random.RandomState) – Random number generator for the random state assignment.

seq_initialize(x, weights, rng)

Vectorized initialization of sufficient statistics with encoded sequences.

Parameters:

• x – Encoded sequence data produced by acc_to_encoder() (or a matching dist encoder).

• weights (np.ndarray) – Weight for each encoded sequence.

• rng (np.random.RandomState) – Random number generator for the random state assignment.

acc_to_encoder()

Return a LookbackHiddenMarkovModelDataEncoder consistent with this accumulator.

Returns:

Encoder built from the member accumulators.

Return type:

LookbackHiddenMarkovModelDataEncoder

seq_update(x, weights, estimate)

Vectorized Baum-Welch update of sufficient statistics with encoded sequences.

Parameters:

• x – Encoded sequence data produced by acc_to_encoder() (or a matching dist encoder).

• weights (np.ndarray) – Weight for each encoded sequence.

• estimate (LookbackHiddenMarkovModelDistribution) – Current estimate used for the E-step.

seq_update_engine(x, weights, estimate, engine)

Engine-resident Baum-Welch E-step via the shared HMM forward-backward (numpy or torch).

Emissions (init segment + windowed topics) are scored on the active engine, the forward-backward runs on the engine, and the resulting posteriors are routed to the init / topic / length accumulators. Mirrors seq_update.

combine(suff_stat)

Aggregate sufficient statistics from suff_stat (a value() tuple) into this accumulator.

Parameters:

suff_stat (Tuple) – Sufficient statistics in the format returned by value().

Returns:

This accumulator after aggregation.

Return type:

LookbackHiddenMarkovModelEstimatorAccumulator

value()

Return the sufficient statistics of this accumulator.

Returns:

(lag, num_states, init_counts, state_counts, trans_counts, seq_acc_values,

init_acc_values, len_acc_value).

Return type:

Tuple

from_value(x)

Set the sufficient statistics of this accumulator from a value() tuple.

Parameters:

x (Tuple) – Sufficient statistics in the format returned by value().

Returns:

This accumulator after assignment.

Return type:

LookbackHiddenMarkovModelEstimatorAccumulator

scale(c)

Scale linear sufficient statistics in-place by c.

The structural default is correct for ordinary weighted sums, nested tuples/lists/dicts, and numeric arrays. Families whose value() payload includes non-linear metadata such as support bounds must override this method and leave that metadata unscaled.

Parameters:

c (float)

Return type:

LookbackHiddenMarkovModelEstimatorAccumulator

key_merge(stats_dict)

Merge keyed sufficient statistics of this accumulator into stats_dict.

Parameters:

stats_dict (Dict[str, Any]) – Dictionary mapping keys to merged sufficient statistics.

key_replace(stats_dict)

Replace keyed sufficient statistics of this accumulator with values from stats_dict.

Parameters:

stats_dict (Dict[str, Any]) – Dictionary mapping keys to merged sufficient statistics.

class LookbackHiddenMarkovModelEstimatorAccumulatorFactory(lag, seq_factories, init_factories=None, len_factory=NullAccumulatorFactory(), keys=(None, None, None), terminal_states=None)

Bases: StatisticAccumulatorFactory

Factory for creating LookbackHiddenMarkovModelEstimatorAccumulator objects.

Parameters:

• lag (int)

• seq_factories (Sequence[StatisticAccumulatorFactory])

• init_factories (Sequence[StatisticAccumulatorFactory] | None)

• len_factory (StatisticAccumulatorFactory | None)

• keys (tuple[str | None, str | None, str | None] | None)

make()

Create a new LookbackHiddenMarkovModelEstimatorAccumulator from the member factories.

Returns:

Accumulator with zeroed sufficient statistics.

Return type:

LookbackHiddenMarkovModelEstimatorAccumulator

class LookbackHiddenMarkovModelEstimator(estimators, lag=0, init_estimators=None, len_estimator=NullEstimator(), suff_stat=None, pseudo_count=(None, None), name=None, keys=(None, None, None), terminal_states=None)

Bases: ParameterEstimator

Estimator for a lookback hidden Markov model from aggregated sufficient statistics.

Parameters:

• estimators (Sequence[ParameterEstimator])

• lag (int)

• init_estimators (Sequence[ParameterEstimator] | None)

• len_estimator (ParameterEstimator | None)

• pseudo_count (tuple[float | None, float | None] | None)

• name (str | None)

• keys (tuple[str | None, str | None, str | None] | None)

accumulator_factory()

Create a LookbackHiddenMarkovModelEstimatorAccumulatorFactory from the member estimators.

Returns:

Factory for accumulators consistent

with this estimator.

Return type:

LookbackHiddenMarkovModelEstimatorAccumulatorFactory

estimate(nobs, suff_stat)

Estimate a LookbackHiddenMarkovModelDistribution from aggregated sufficient statistics.

Parameters:

• nobs (Optional[float]) – Weighted number of observations (passed to the length estimator).

• suff_stat (Tuple) – Sufficient statistics in the format returned by LookbackHiddenMarkovModelEstimatorAccumulator.value().

Returns:

M-step estimate of the distribution.

Return type:

LookbackHiddenMarkovModelDistribution

class LookbackHiddenMarkovModelDataEncoder(encoder, lag, len_encoder=NullDataEncoder(), init_encoder=NullDataEncoder())

Bases: DataSequenceEncoder

Encoder for sequences of iid lookback-HMM observations (each a Sequence[T]).

Parameters:

• encoder (DataSequenceEncoder)

• lag (int)

• len_encoder (DataSequenceEncoder | None)

• init_encoder (DataSequenceEncoder | None)

seq_encode(x)

Encode a sequence of iid observed sequences for vectorized processing.

Each sequence x[i] is split into its initial segment x[i][:lag] and its sliding windows x[i][j-lag:j+1] for j in [lag, len(x[i])); index arrays track which flattened position belongs to which sequence and which positions are initial segments vs emission windows. When lag == 0 there are no initial segments: every position is an emission window (ordinary HMM).

Parameters:

x (Sequence[Sequence[T]]) – Sequence of iid observed sequences.

Returns:

((ids, idi, ims, imi, sz, enc_windows, enc_inits), len_enc) where ids/idi map

windows/initial segments to sequence indices, ims/imi give their flattened positions, sz holds per-sequence position counts, enc_windows/enc_inits are the encoded windows and initial segments (enc_inits is None when lag == 0), and len_enc is the encoded position counts.

Return type:

Tuple

LookbackHiddenMarkovModelAccumulator

alias of LookbackHiddenMarkovModelEstimatorAccumulator

LookbackHiddenMarkovModelAccumulatorFactory

alias of LookbackHiddenMarkovModelEstimatorAccumulatorFactory