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

n_phases: int = 1
phase(t, length)[source]
Parameters:
Return type:

int

to_dict()[source]
Return type:

dict[str, Any]

static from_dict(d)[source]
Parameters:

d (dict[str, Any])

Return type:

PhaseSchedule

class Homogeneous[source]

Bases: PhaseSchedule

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

n_phases: int = 1
phase(t, length)[source]
Parameters:
Return type:

int

to_dict()[source]
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]
Parameters:
Return type:

int

to_dict()[source]
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]
Parameters:
Return type:

int

to_dict()[source]
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]
Parameters:
Return type:

int

to_dict()[source]
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-density or log-mass at a single observation.

Parameters:

x (list[Any])

Return type:

float

seq_log_density(x)[source]

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

Parameters:

x (Any)

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for drawing observations from this distribution.

Parameters:

seed (int | None)

Return type:

ScheduledHMMSampler

estimator(pseudo_count=None)[source]

Return an estimator for fitting this distribution from data.

Parameters:

pseudo_count (float | None)

Return type:

ScheduledHMMEstimator

dist_to_encoder()[source]

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

Return type:

ScheduledHMMDataEncoder

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

Bases: DistributionSampler

Parameters:

  • dist (ScheduledHiddenMarkovModelDistribution)

  • seed (int | None)

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

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:
Return type:

Any

class ScheduledHMMDataEncoder[source]

Bases: DataSequenceEncoder

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (list[list[Any]])

Return type:

list[list[Any]]

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

Bases: SequenceEncodableStatisticAccumulator

Parameters:

  • n_states (int)

  • schedule (PhaseSchedule)

  • emission_factory (Any)

  • len_factory (Any)

update(x, weight, estimate)[source]
Parameters:

  • x (list[Any])

  • weight (float)

  • estimate (ScheduledHiddenMarkovModelDistribution)

Return type:

None

seq_update(x, weights, estimate)[source]
Parameters:

  • x (Any)

  • weights (ndarray)

  • estimate (ScheduledHiddenMarkovModelDistribution)

Return type:

None

initialize(x, weight, rng)[source]
Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]
Parameters:
Return type:

None

combine(other)[source]
Parameters:

other (Any)

Return type:

ScheduledHMMAccumulator

value()[source]
Return type:

tuple

from_value(value)[source]
Parameters:

value (tuple)

Return type:

ScheduledHMMAccumulator

acc_to_encoder()[source]
Return type:

ScheduledHMMDataEncoder

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

Bases: StatisticAccumulatorFactory

Parameters:

  • n_states (int)

  • schedule (PhaseSchedule)

  • emission_estimator (Any)

  • len_estimator (Any)

make()[source]
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 type:

ScheduledHMMAccumulatorFactory

estimate(nobs, suff_stat)[source]
Parameters:
Return type:

ScheduledHiddenMarkovModelDistribution