mixle.stats.latent.scheduled_hidden_markov_model module

A length- and position-conditional (“scheduled”) hidden Markov model.

A standard HMM is time-homogeneous: the same initial distribution, transition matrix, and emissions apply at every position, and (with a len_dist) the length is drawn independently of the path – it only sets the count. This family makes the dynamics depend on where you are in the sequence and how long it is, through a serializable PhaseSchedule phi(t, L) that maps position t in a length-L sequence to a phase. Each phase has its own initial / transition / emission parameters; EM pools sufficient statistics by phase.

One mechanism covers every reasonable “length-conditional” model:

  • Homogeneous – phi(t, L) = 0 – the ordinary HMM (one phase).

  • ByLength – phi(t, L) = bucket(L) – a length-conditional HMM: short and long sequences use different dynamics (constant within a sequence).

  • ByRelativePosition – phi(t, L) = floor(B * t / L) – relative position: the chain knows how far through the sequence it is (e.g. winds down toward the end), regardless of absolute length.

  • ByPosition – phi(t, L) = min(t, cap-1) – absolute position (non-homogeneous in time).

The length itself is still drawn from len_dist (it remains a random variable); the schedule adds the conditioning of the content on length/position that the homogeneous model lacks. Emissions are per-phase too, so length/position can shape emissions, not just transitions.

This is a deliberately lean, numpy-only implementation (no numba / enumeration / terminal-state integration – those live on HiddenMarkovModelDistribution). It reuses the emission families’ own estimators for the M-step.

class PhaseSchedule[source]

Bases: object

Maps a position t in a length-L sequence to a phase index in [0, n_phases).

phase(t, length)[source]

Return the phase index for position t in a sequence of length.

Parameters:
Return type:

int

to_dict()[source]

Serialize the schedule to a JSON-compatible dictionary.

Return type:

dict[str, Any]

static from_dict(d)[source]

Deserialize a schedule produced by to_dict().

Parameters:

d (dict[str, Any])

Return type:

PhaseSchedule

class Homogeneous[source]

Bases: PhaseSchedule

One phase for everything – the ordinary time-homogeneous HMM.

phase(t, length)[source]

Return the single homogeneous phase.

Parameters:
Return type:

int

to_dict()[source]

Serialize the homogeneous schedule.

Return type:

dict[str, Any]

class ByPosition(cap)[source]

Bases: PhaseSchedule

Absolute position: phi(t, L) = min(t, cap - 1) (positions past cap-1 share the last phase).

Parameters:

cap (int)

phase(t, length)[source]

Return the absolute-position phase capped at the final phase.

Parameters:
Return type:

int

to_dict()[source]

Serialize the absolute-position schedule.

Return type:

dict[str, Any]

class ByRelativePosition(bins)[source]

Bases: PhaseSchedule

Relative position: phi(t, L) = min(bins - 1, floor(bins * t / L)) – progress through the sequence.

Parameters:

bins (int)

phase(t, length)[source]

Return the relative-position phase for t / length.

Parameters:
Return type:

int

to_dict()[source]

Serialize the relative-position schedule.

Return type:

dict[str, Any]

class ByLength(boundaries)[source]

Bases: PhaseSchedule

Length-conditional: phase is the bucket of L against sorted boundaries (constant within a seq).

With boundaries = [5, 10] there are three phases: L <= 5, 5 < L <= 10, L > 10.

Parameters:

boundaries (Sequence[int])

phase(t, length)[source]

Return the length-bucket phase for the sequence length.

Parameters:
Return type:

int

to_dict()[source]

Serialize the length-bucket schedule.

Return type:

dict[str, Any]

class ScheduledHiddenMarkovModelDistribution(inits, transitions, emissions, schedule, len_dist=None, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Phase-indexed (length-/position-conditional) HMM. See the module docstring for the modeling story.

Parameters:

  • inits (np.ndarray)

  • transitions (np.ndarray)

  • emissions (list[list[Any]])

  • schedule (PhaseSchedule)

  • len_dist (Any)

  • name (str | None)

log_density(x)[source]

Return the log likelihood of one scheduled HMM sequence.

Parameters:

x (list[Any])

Return type:

float

seq_log_density(x)[source]

Score a batch of scheduled HMM sequences.

Parameters:

x (Any)

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for scheduled HMM sequences.

Parameters:

seed (int | None)

Return type:

ScheduledHMMSampler

estimator(pseudo_count=None)[source]

Raise because emission and length estimators must be supplied explicitly.

Parameters:

pseudo_count (float | None)

Return type:

ScheduledHMMEstimator

dist_to_encoder()[source]

Return the pass-through scheduled HMM encoder.

Return type:

ScheduledHMMDataEncoder

class ScheduledHMMSampler(dist, seed=None)[source]

Bases: DistributionSampler

Sampler for scheduled HMM sequences.

Parameters:

  • dist (ScheduledHiddenMarkovModelDistribution)

  • seed (int | None)

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

Draw one sequence or a list of sequences.

Parameters:
Return type:

Any

class ScheduledHMMDataEncoder[source]

Bases: DataSequenceEncoder

Pass-through encoder for scheduled HMM sequence observations.

seq_encode(x)[source]

Encode scheduled HMM records as sequence lists.

Parameters:

x (list[list[Any]])

Return type:

list[list[Any]]

class ScheduledHMMAccumulator(n_states, schedule, emission_factory, len_factory=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for phase-pooled scheduled HMM EM sufficient statistics.

Parameters:

  • n_states (int)

  • schedule (PhaseSchedule)

  • emission_factory (Any)

  • len_factory (Any)

update(x, weight, estimate)[source]

Accumulate sufficient statistics from one weighted sequence.

Parameters:

  • x (list[Any])

  • weight (float)

  • estimate (ScheduledHiddenMarkovModelDistribution)

Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate weighted sufficient statistics from a batch.

Parameters:

  • x (Any)

  • weights (ndarray)

  • estimate (ScheduledHiddenMarkovModelDistribution)

Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics with random soft state responsibilities.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from a weighted batch.

Parameters:
Return type:

None

combine(other)[source]

Merge serialized scheduled HMM sufficient statistics.

Parameters:

other (Any)

Return type:

ScheduledHMMAccumulator

value()[source]

Return serialized scheduled HMM sufficient statistics.

Return type:

tuple

from_value(value)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

value (tuple)

Return type:

ScheduledHMMAccumulator

acc_to_encoder()[source]

Return the encoder associated with this accumulator.

Return type:

ScheduledHMMDataEncoder

class ScheduledHMMAccumulatorFactory(n_states, schedule, emission_estimator, len_estimator=None)[source]

Bases: StatisticAccumulatorFactory

Factory for scheduled HMM accumulators.

Parameters:

  • n_states (int)

  • schedule (PhaseSchedule)

  • emission_estimator (Any)

  • len_estimator (Any)

make()[source]

Create a fresh scheduled HMM accumulator.

Return type:

ScheduledHMMAccumulator

class ScheduledHMMEstimator(n_states, schedule, emission_estimator, len_estimator=None, pseudo_count=1e-8, name=None)[source]

Bases: ParameterEstimator

EM estimator for a ScheduledHiddenMarkovModelDistribution with a fixed schedule.

emission_estimator is the estimator for ONE emission distribution (reused for every phase x state); len_estimator (optional) estimates the length distribution. The schedule is fixed (it defines the parameter sharing); only the per-phase parameters are learned.

Parameters:

  • n_states (int)

  • schedule (PhaseSchedule)

  • emission_estimator (Any)

  • len_estimator (Any)

  • pseudo_count (float)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used by this estimator.

Return type:

ScheduledHMMAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate phase-indexed initial, transition, emission, and length models.

Parameters:
Return type:

ScheduledHiddenMarkovModelDistribution