mixle.stats.latent.lookback_hidden_markov_model module

Lookback hidden Markov models with emissions conditioned on recent observations.

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()

Return exact-or-approximate density semantics joined from child models.

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 terminal-aware lookback HMM sequences as lists.

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 all accumulated lookback-HMM sufficient statistics in place.

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