mixle.stats.latent.segmental_hidden_markov_model module

Hidden Markov model over arbitrary emission segments.

Each hidden state emits one segment object. The segment can be any data type accepted by that state’s emission distribution: a scalar, tuple, set, sequence, or another composable mixle.stats object. To model variable-length emissions, use a SequenceDistribution (or any other distribution over sequences) as an emission distribution.

Unlike HiddenMarkovModelDistribution, each state keeps its own encoder, so emission distributions may use different distribution classes as long as they can all score the same raw segment observations.

class SegmentalHiddenMarkovModelDistribution(emissions, w=MISSING, transitions=MISSING, len_dist=NullDistribution(), name=None, weights=MISSING, terminal_states=None)

Bases: SequenceEncodableProbabilityDistribution

HMM whose states emit arbitrary segment-valued distributions.

Observations are lists of segment objects. For example, with SequenceDistribution(GaussianDistribution(...), len_dist=...) as an emission, each state emits a variable-length list of real values.

Parameters:

• emissions (Sequence[SequenceEncodableProbabilityDistribution])

• w (Sequence[float] | ndarray)

• transitions (Sequence[Sequence[float]] | ndarray)

• len_dist (SequenceEncodableProbabilityDistribution | None)

• name (str | None)

• weights (Sequence[float] | ndarray)

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

compute_capabilities()

compute_declaration()

property topics: list[SequenceEncodableProbabilityDistribution]

Compatibility alias with HiddenMarkovModelDistribution terminology.

density(x)

Return the probability density or mass at a single observation.

Parameters:

x (Sequence[Any])

Return type:

float

log_density(x)

Return the log-density or log-mass at a single observation.

Parameters:

x (Sequence[Any])

Return type:

float

seq_log_density(x)

Return vectorized log-density values for sequence-encoded observations.

Parameters:

x (tuple[ndarray, ndarray, tuple[Any, ...], Any | None])

Return type:

ndarray

backend_seq_log_density(x, engine)

Engine-neutral segmental-HMM forward scoring for encoded segment sequences.

Parameters:

• x (tuple[ndarray, ndarray, tuple[Any, ...], Any | None])

• engine (Any)

Return type:

Any

sampler(seed=None)

Return a sampler for drawing observations from this distribution.

Parameters:

seed (int | None)

Return type:

SegmentalHiddenMarkovSampler

estimator(pseudo_count=None)

Return an estimator for fitting this distribution from data.

Parameters:

pseudo_count (float | None)

Return type:

SegmentalHiddenMarkovEstimator

dist_to_encoder()

Return the data encoder used by this distribution for vectorized methods.

Return type:

SegmentalHiddenMarkovDataEncoder

enumerator()

Enumerate segment sequences in descending marginal probability order.

The segmental HMM has the standard HMM forward semantics – each position emits one segment from its state’s distribution, scored independently – so it reuses HiddenMarkovModelEnumerator directly via its per-state emission (topics), log_w, log_transitions, and len_dist. Each segment is drawn from the union of the per-state emission supports, so every emission distribution must itself support enumeration (and a length distribution must be modeled).

Return type:

DistributionEnumerator

class SegmentalHiddenMarkovSampler(dist, seed=None)

Bases: DistributionSampler

Draw iid segmental-HMM observations.

Parameters:

• dist (SegmentalHiddenMarkovModelDistribution)

• seed (int | None)

sample(size=None)

Draw observations.

Combinator samplers (mixture/sequence/…) accept batched. With batched=True (the default) each child stream is drawn in one vectorized call instead of a per-draw Python loop – far faster. Because every child sampler owns an independent RandomState, batching consumes each stream in the same order as the loop, so the draws are identical to the legacy path. batched=False forces that legacy per-draw loop as a guaranteed- stable reference. Leaf samplers are already vectorized and ignore the flag.

Parameters:

size (int | None)

Return type:

list[Any] | list[list[Any]]

class SegmentalHiddenMarkovAccumulator(accumulators, len_accumulator=NullAccumulator(), keys=(None, None, None), name=None)

Bases: SequenceEncodableStatisticAccumulator

Baum-Welch accumulator for segmental HMMs.

Parameters:

• accumulators (Sequence[SequenceEncodableStatisticAccumulator])

• len_accumulator (SequenceEncodableStatisticAccumulator | None)

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

• name (str | None)

update(x, weight, estimate)

Parameters:

• x (Sequence[Any])

• weight (float)

• estimate (SegmentalHiddenMarkovModelDistribution)

Return type:

None

initialize(x, weight, rng)

Parameters:

• x (Sequence[Any])

• weight (float)

• rng (RandomState)

Return type:

None

seq_initialize(x, weights, rng)

Parameters:

• x (tuple[ndarray, ndarray, tuple[Any, ...], Any | None])

• weights (ndarray)

• rng (RandomState)

Return type:

None

seq_update(x, weights, estimate)

Parameters:

• x (tuple[ndarray, ndarray, tuple[Any, ...], Any | None])

• weights (ndarray)

• estimate (SegmentalHiddenMarkovModelDistribution)

Return type:

None

seq_update_engine(x, weights, estimate, engine)

Engine-resident Baum-Welch E-step (numpy or torch).

Reuses hmm_engine_forward_backward over the sequence-contiguous segment encoding, producing the same initial/state/transition/emission statistics as the host seq_update.

combine(suff_stat)

Parameters:

suff_stat (tuple[int, ndarray, ndarray, ndarray, Sequence[Any], Any | None])

Return type:

SegmentalHiddenMarkovAccumulator

value()

Return type:

tuple[int, ndarray, ndarray, ndarray, tuple[Any, …], Any | None]

from_value(x)

Parameters:

x (tuple[int, ndarray, ndarray, ndarray, Sequence[Any], Any | None])

Return type:

SegmentalHiddenMarkovAccumulator

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:

SegmentalHiddenMarkovAccumulator

key_merge(stats_dict)

Pool this accumulator’s statistics into stats_dict under its merge key.

The structural default implements the common single-key pattern: store the accumulator under self.keys the first time the key is seen, else combine into the one already there. Accumulators with several named keys (e.g. an HMM’s init/trans/state keys) or a non-accumulator stats payload override this. A keys of None (the default) is a no-op.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)

Replace this accumulator’s statistics from the pooled stats_dict entry (see key_merge).

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()

Return type:

SegmentalHiddenMarkovDataEncoder

class SegmentalHiddenMarkovAccumulatorFactory(factories, len_factory=NullAccumulatorFactory(), keys=(None, None, None), name=None)

Bases: StatisticAccumulatorFactory

Factory for SegmentalHiddenMarkovAccumulator.

Parameters:

• factories (Sequence[StatisticAccumulatorFactory])

• len_factory (StatisticAccumulatorFactory | None)

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

• name (str | None)

make()

Return type:

SegmentalHiddenMarkovAccumulator

class SegmentalHiddenMarkovEstimator(estimators, len_estimator=NullEstimator(), pseudo_count=(None, None), name=None, keys=(None, None, None), terminal_states=None)

Bases: ParameterEstimator

Baum-Welch estimator for SegmentalHiddenMarkovModelDistribution.

Parameters:

• estimators (Sequence[ParameterEstimator])

• len_estimator (ParameterEstimator | None)

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

• name (str | None)

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

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

accumulator_factory()

Return type:

SegmentalHiddenMarkovAccumulatorFactory

estimate(nobs, suff_stat)

Parameters:

• nobs (float | None)

• suff_stat (tuple[int, ndarray, ndarray, ndarray, Sequence[Any], Any | None])

Return type:

SegmentalHiddenMarkovModelDistribution

class SegmentalHiddenMarkovDataEncoder(emission_encoders, len_encoder=NullDataEncoder())

Bases: DataSequenceEncoder

Encode a batch of segment sequences for a segmental HMM.

Parameters:

• emission_encoders (Sequence[DataSequenceEncoder])

• len_encoder (DataSequenceEncoder | None)

seq_encode(x)

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Sequence[Any]])

Return type:

tuple[ndarray, ndarray, tuple[Any, …], Any | None]

SegmentalHiddenMarkovDistribution

alias of SegmentalHiddenMarkovModelDistribution

SegmentalHiddenMarkovModelAccumulator

alias of SegmentalHiddenMarkovAccumulator

SegmentalHiddenMarkovModelAccumulatorFactory

alias of SegmentalHiddenMarkovAccumulatorFactory

SegmentalHiddenMarkovModelDataEncoder

alias of SegmentalHiddenMarkovDataEncoder

SegmentalHiddenMarkovModelEstimator

alias of SegmentalHiddenMarkovEstimator

SegmentalHiddenMarkovModelSampler

alias of SegmentalHiddenMarkovSampler