mixle.stats.sequences.markov_chain module

Create, estimate, and sample from a Markov chain.

Defines the MarkovChainDistribution, MarkovChainDistributionSampler, MarkovChainDistributionAccumulatorFactory, MarkovChainDistributionAccumulator, MarkovChainDistributionEstimator, and the MarkovChainDistributionDataEncoder classes for use with mixle.

The assumed data type for the stats-space is T.

The density of Markov chain is given by for sequence of length n, x=[x[0],x[1],…,x[n-1]]

p_mat(x) = p_mat(x[0])*p_mat(x[1]|x[0])*…*p_mat(x[n-1]|x[n-2])*P_len(n)

where p_mat(x[i+1]|x[i]) is the transition probability, p_mat(x[0]) is the init-probability, and P_len(n) is given by the length distribution density.

Note if len(x) = 0, only log(P_len(0)) is returned.

markov_chain_dirichlet_default_prior(states)[source]

Returns the default (states, init_prior, row_priors) prior of unit-parameter Dirichlets.

Parameters:

states (Sequence[T]) – Ordered list of the S state values the priors range over.

Returns:

Tuple (list_of_states, DirichletDistribution, list_of_DirichletDistribution).

stationary_distribution(transitions)[source]

Stationary distribution pi of a row-stochastic matrix (pi A = pi, sum pi = 1).

Solved as the constrained least-squares system [(I - A^T); 1^T] pi = [0; 1] so it is robust for reducible/near-singular chains (it returns one valid stationary distribution). The result is clipped to non-negative and renormalized.

Parameters:

transitions (np.ndarray) – a square row-stochastic transition matrix.

Returns:

1-d numpy array of stationary probabilities (length = number of states).

Return type:

ndarray

class MarkovChainDistribution(init_prob_map, transition_map, len_dist=NullDistribution(), default_value=0.0, name=None, prior=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Markov-chain distribution over finite-state sequences.

Parameters:

  • init_prob_map (dict[T, float])

  • transition_map (dict[T, dict[T, float]])

  • len_dist (SequenceEncodableProbabilityDistribution | None)

  • default_value (float)

  • name (str | None)

get_prior()[source]

Returns the conjugate prior in (states, init_prior, row_priors) form (or None).

set_prior(prior)[source]

Set the conjugate Dirichlet prior and precompute its digamma expectations.

With Dirichlet init_prior and Dirichlet row_priors (each over the fixed ordered states) this caches the digamma expectations E[ln p_k] = psi(alpha_k) - psi(sum alpha) used by expected_log_density and sets has_conj_prior accordingly. prior=None leaves the distribution a plain point model.

Parameters:

prior – (states, init_prior, row_priors) tuple or None.

Return type:

None

expected_log_density(x)[source]

Variational E_q[log p(x)] under the Dirichlet priors over a state sequence.

Replaces the initial/transition log-probabilities with their digamma expectations E[ln p_k] = psi(alpha_k) - psi(sum alpha); the length term is added as in log_density(). Falls back to the plug-in log_density(x) when no conjugate prior is set.

Parameters:

x (List[T]) – An observed Markov chain state sequence.

Returns:

Expected log-density of the Markov chain at x.

Return type:

float

seq_expected_log_density(x)[source]

Vectorized expected_log_density() at sequence-encoded input x.

Falls back to seq_log_density(x) when no conjugate prior is set.

Parameters:

x (tuple[int, ndarray, ndarray, ndarray, ndarray, ndarray, ndarray, Any]) – Encoded sequences from seq_encode().

Returns:

Numpy array of expected log-densities, one per sequence.

Return type:

ndarray

compute_capabilities()[source]
compute_declaration()[source]
density(x)[source]

Return density of MarkovChainDistribution at observed sequence x.

Returns exponential of log_density(x). See log_density() for details.

Parameters:

x (List[T]) – An observed Markov chain sequence of data type T.

Returns:

Density of Markov chain at x.

Return type:

float

log_density(x)[source]

Return log-density of MarkovChainDistribution at observed sequence x.

Density of Markov chain is given by for sequence of length n, x=[x[0],x[1],…,x[n-1]]

p_mat(x) = p_mat(x[0])*p_mat(x[1]|x[0])*…*p_mat(x[n-1]|x[n-2])*P_len(n)

where p_mat(x[i+1]|x[i]) is the transition probability, p_mat(x[0]) is the init-probability, and P_len(n) is given by the length distribution density.

Note if len(x) = 0, only log(P_len(0)) is returned.

Parameters:

x (List[T]) – An observed Markov chain sequence of data type T.

Returns:

Log-density of Markov chain at x.

Return type:

float

seq_log_density(x)[source]

Vectorized evaluation of log_density of Markov Chain for an encoded sequence of observations x.

Computationally efficient implementation of log_density() for sequence encoded data x.

The arg value x is a Tuple of length 8 with entries:

x[0] (int): Number of total observations (number of Markov sequences). x[1] (ndarray[int]): Sequence index for initial state observations. x[2] (ndarray[int]): Sequence index for non-initial state observations in a sequence greater than len 1. x[3] (ndarray[int]): Numpy array of observations index in inv_key_map for initial states. x[4] (ndarray[int]): State-to-state index value of inv_key_map for initial state value. x[5] (ndarray[int]): State-to-state index value of inv_key_map for transition. x[6] (ndarray[T]): Maps integer index value to value in state-space (T). x[7] (Optional[T1]): Encoded sequence of lengths from len_encoder. None if no length distribution to be

estimated.

Parameters:

x (tuple[int, ndarray, ndarray, ndarray, ndarray, ndarray, ndarray, Any]) – See above for details.

Returns:

Numpy of length x[0], containing the log-density of Markov chain at each observation in x.

Return type:

ndarray

backend_seq_log_density(x, engine)[source]

Engine-neutral vectorized log-density for encoded Markov-chain sequences.

Parameters:
Return type:

Any

classmethod backend_stacked_params(dists, engine)[source]

Return stacked fixed-support Markov-chain parameters.

Parameters:

  • dists (Sequence[MarkovChainDistribution])

  • engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)[source]

Return an (n, k) matrix of Markov-chain sequence log densities.

Parameters:
Return type:

Any

classmethod backend_stacked_sufficient_statistics_with_estimator(x, weights, params, engine, estimator)[source]

Return per-component legacy (initial_counts, transition_counts, length_stat) statistics.

Parameters:
Return type:

tuple[Any, …]

gradient_fit_state(engine, torch, leaves, recurse, tensor_param)[source]

Return distribution-owned state for fixed-support autograd fitting.

Parameters:
Return type:

Any

sampler(seed=None)[source]

Create MarkovChainSampler from MarkovChainDistribution instance.

Raises exception if length distribution (len_dist) was not specified in initialization.

Parameters:

seed (Optional[int]) – Used to set the seed of random number generator for sampling.

Returns:

MarkovChainSampler object.

Return type:

MarkovChainSampler

estimator(pseudo_count=None)[source]

Create MarkovChainEstimator from instance of MarkovChainDistribution object.

Parameters:

pseudo_count (Optional[float]) – Used to re-weight the sufficient statistics of MarkovChainDistribution.

Returns:

MarkovChainEstimator object.

Return type:

MarkovChainEstimator

dist_to_encoder()[source]

Create MarkovChainDataEncoder object for encoding sequences of MarkovChainDistribution observations.

Note: len_encoder is passed as NullDataEncoder() if len_dist is not to be estimated.

Returns:

MarkovChainDataEncoder object.

Return type:

MarkovChainDataEncoder

enumerator()[source]

Returns MarkovChainEnumerator iterating state sequences in descending probability order.

Return type:

MarkovChainEnumerator

quantized_count_index(quantizer, max_fine_bucket)[source]

Structural count index: a forward DP carrying a count histogram per (length, end-state).

log p(x) = log p_init(x0) + sum_i log p_trans(x_i|x_{i-1}) + log p(len). The forward recursion is lifted into the count semiring: alpha[t][s] is the histogram (over the fine bucket of accumulated log probability) of length-t prefixes ending in state s, with alpha[1][s] = delta(bucket(log p_init(s))) and alpha[t+1][s'] = sum_s alpha[t][s].shift(bucket(log p_trans(s'|s))). Per length L the sequence histogram pools the end states and shifts by the length term; the total pools lengths. Sequences are unranked by choosing the end state, then walking the trellis backward choosing predecessors by count.

Parameters:

max_fine_bucket (int)

class MarkovChainEnumerator(dist)[source]

Bases: DistributionEnumerator

Parameters:

dist (MarkovChainDistribution)

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

Bases: DistributionSampler

Parameters:

  • dist (MarkovChainDistribution)

  • seed (int | None)

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

Draw iid samples from Markov chain distribution.

If size is None, sample N from len_sampler() and return a List[T] of length N, where T is the data type of the Markov chain. If size > 0, return a list of length size, containing List[T] data types.

With batched=True (default) the state paths for the whole batch are drawn by looping over time and advancing all live chains at once through the transition matrix, instead of N x T scalar draws. This consumes the RNG in a different order than the legacy per-draw loop, so the draws are statistically equivalent but NOT byte-identical to batched=False. Set batched=False to reproduce the exact legacy output for a given seed.

Parameters:

  • size (Optional[int]) – Number of samples to draw. Draws 1 sample if None.

  • batched (bool) – Vectorize state-path draws across chains (default); set False for the legacy per-draw loop.

Returns:

List[T] or List[List[T]], depending on size arg.

Return type:

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

sample_paths(lengths)[source]

Vectorized batch of state paths, one per requested length.

Loops over time and advances all live chains at once through the transition matrix. The RNG consumption order differs from per-sequence sample_seq calls, so paths are statistically equivalent but not byte-identical. Used by HiddenMarkovSampler for batched state-path draws.

Parameters:

lengths (Sequence[int]) – Length of each chain to sample.

Returns:

List of state-sequences (List[T]), one per entry in lengths.

Return type:

list[list[Any]]

sample_seq(size=None, v0=None, *, batched=False)[source]

Sample a Markov chain sequence of length ‘size’ conditioned on initial state ‘v0’.

If size is None, draw a sequence of length 1, returning as type T.

If size is not None, draw a sequence of length size, returning as type List[T].

If v0 is None, v0 is sampled from member variable ‘init_prob’.

This is the legacy per-step path (one rng.choice per transition); the batched flag is accepted for API symmetry but does not change behavior here. For vectorized batches of whole chains use sample_paths() or sample(..., batched=True).

Parameters:

  • size (Optional[int]) – Length of Markov chain sequence to sample.

  • v0 (Optional[T]) – Initial state of Markov chain sequence to sample from.

  • batched (bool) – Accepted for API symmetry; sample_seq is always the per-step path.

Returns:

T or List[T] depending on arg size.

Return type:

T | list[T]

class MarkovChainAccumulator(len_accumulator=NullAccumulator(), keys=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Parameters:

  • len_accumulator (SequenceEncodableStatisticAccumulator | None)

  • keys (str | None)

update(x, weight, estimate)[source]

Update sufficient statistics of MarkovChainAccumulator with weighted observation.

Aggregates suff stats by checking initial state of sequence, and counting all transitions. Passes length of sequence x to len_accumulator.

Parameters:

  • x (List[T])

  • weight (float) – Weight for observation.

  • estimate (Optional[MarkovChainDistribution]) – Previous estimate for MarkovChainDistribution or None.

Returns:

None.

Return type:

None

initialize(x, weight, rng)[source]
Initialize MarkovChainAccumulator with Markov chain observation x and random number generator rng passed

to len_accumulator.initialize().

Parameters:

  • x (List[T]) – Single Markov chain observation.

  • weight (float) – Weight for observation.

  • rng (RandomState) – RandomState object for passing to len_accumulator.initialize().

Returns:

None.

Return type:

None

seq_initialize(x, weights, rng)[source]

Vectorized initialization of MarkovChainAccumulator sufficient statistics from a sequence of encoded data x.

Note that this is the same as seq_update() for the transition and initial state updates. For len_accumulator, a call to seq_initialize() must be made.

The arg value x is a Tuple of length 8 with entries:

x[0] (int): Number of total observations (number of Markov sequences). x[1] (ndarray[int]): Sequence index for initial state observations. x[2] (ndarray[int]): Sequence index for non-initial state observations in a sequence greater than len 1. x[3] (ndarray[int]): Numpy array of observations index in inv_key_map for initial states. x[4] (ndarray[int]): State-to-state index value of inv_key_map for initial state value. x[5] (ndarray[int]): State-to-state index value of inv_key_map for transition. x[6] (ndarray[T]): Maps integer index value to value in state-space (T). x[7] (Optional[T1]): Encoded sequence of lengths from len_encoder. None if no length distribution to be

estimated.

Parameters:
Returns:

None.

Return type:

None

seq_update(x, weights, estimate)[source]

Vectorized update of Markov chain sufficient statistics for a sequence encoded x.

Computationally efficient update of MarkovChainAccumulator object using vectorized numpy operations.

Note that estimate must be passed, as the ‘estimate’ argument of len_accumulator.seq_update() may require estimate parameter to not be None.

The arg value x is a Tuple of length 8 with entries:

x[0] (int): Number of total observations (number of Markov sequences). x[1] (ndarray[int]): Sequence index for initial state observations. x[2] (ndarray[int]): Sequence index for non-initial state observations in a sequence greater than len 1. x[3] (ndarray[int]): Numpy array of observations index in inv_key_map for initial states. x[4] (ndarray[int]): State-to-state index value of inv_key_map for initial state value. x[5] (ndarray[int]): State-to-state index value of inv_key_map for transition. x[6] (ndarray[T]): Maps integer index value to value in state-space (T). x[7] (Optional[T1]): Encoded sequence of lengths from len_encoder. None if no length distribution to be

estimated.

Parameters:
Returns:

None.

Return type:

None

seq_update_engine(x, weights, estimate, engine)[source]

Engine-resident E-step: initial-state counts are reduced on the active engine and the transition weights are gathered on the engine before filling the sparse count maps; the length accumulator is routed through the engine. Matches seq_update.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge the sufficient statistics of arg suff_stat with MarkovChainAccumulator.

Arg suff_stat is a Tuple of length three containing,

suff_stat[0] (Dict[T, float]): Maps initial state values to their corresponding counts. suff_stat[1] (Dict[T, Dict[T, List[float]]]): Maps state to state transition counts. suff_stat[2] (T1): Sufficient statistic value of length accumulator. (Assumed type T1).

Parameters:

suff_stat (tuple[dict[T, float], dict[T, dict[T, float]], Any | None]) – See above for details.

Returns:

MarkovChainAccumulator obejct.

Return type:

MarkovChainAccumulator

value()[source]

Returns Tuple of length three containing sufficient statistic member variables of MarkovChainAccumulator.

Return type:

tuple[dict[T, float], dict[T, dict[T, float]], Any | None]

from_value(x)[source]

Assign MarkovChainAccumulator sufficient statistics to value of x.

Arg x is a Tuple of length three containing,

x[0] (Dict[T, float]): Maps initial state values to their corresponding counts. x[1] (Dict[T, Dict[T, List[float]]]): Maps state to state transition counts. x[2] (T1): Sufficient statistic value of length accumulator. (Assumed type T1).

Parameters:

x (tuple[dict[T, float], dict[T, dict[T, float]], Any | None]) – See above for details.

Returns:

MarkovChainAccumulator object.

Return type:

MarkovChainAccumulator

scale(c)[source]

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:

MarkovChainAccumulator

key_merge(stats_dict)[source]
Aggregate the sufficient statistics of MarkovChainAccumulator with member instance key in

stats_dict.

Parameters:

stats_dict (Dict[str, MarkovChainAccumulator]) – Key of dict are the ‘keys’ for MarkovChainAccumulator that represent the same distribution.

Returns:

None.

Return type:

None

key_replace(stats_dict)[source]
Set MarkovChainAccumulator sufficient statistic member variables to the value of stats_dict with

matching keys.

Checks if member variable key is already in the stats_dict. If it is, set the accumulator to have same member variable sufficient statistics as MarkovChainAccumulators with matching keys.

Parameters:

stats_dict (Dict[str, MarkovChainAccumulator]) – Maps member variable key to MarkovChainAccumulator with same key.

Returns:

None.

Return type:

None

acc_to_encoder()[source]

Create MarkovChainDataEncoder object for encoding sequences of MarkovChainDistribution observations.

Note: len_encoder is passed as NullDataEncoder() if len_dist is not to be estimated.

Returns:

MarkovChainDataEncoder object.

Return type:

MarkovChainDataEncoder

class MarkovChainAccumulatorFactory(len_factory=NullAccumulatorFactory(), keys=None)[source]

Bases: StatisticAccumulatorFactory

Parameters:

  • len_factory (StatisticAccumulatorFactory)

  • keys (str | None)

make()[source]

Returns MarkovChainAccumulator object for accumulating sufficient statistics of Markov chain.

Return type:

MarkovChainAccumulator

class MarkovChainEstimator(pseudo_count=None, levels=None, len_estimator=NullEstimator(), name=None, keys=None, prior=None)[source]

Bases: ParameterEstimator

Parameters:

  • pseudo_count (float | None)

  • levels (Iterable[T] | None)

  • len_estimator (ParameterEstimator | None)

  • name (str | None)

  • keys (str | None)

accumulator_factory()[source]

Returns MarkovChainAccumulatorFactory for creating MarkovChainAccumulator.

Return type:

MarkovChainAccumulatorFactory

get_prior()[source]

Returns the conjugate prior in (states, init_prior, row_priors) form (or None).

set_prior(prior)[source]

Set the conjugate Dirichlet prior and flag whether it admits the conjugate update.

Parameters:

prior – (states, init_prior, row_priors) tuple or None; has_conj_prior is set when all priors are Dirichlet.

Return type:

None

model_log_density(model)[source]

Log-density of the model’s probabilities under the Dirichlet priors.

Sums the Dirichlet log-densities of the initial-state probabilities and each transition row (floored at a tiny constant so MAP estimates that sit on the simplex boundary score finitely). Returns 0.0 without a conjugate prior.

Parameters:

model (MarkovChainDistribution) – Model to score.

Returns:

Prior log-density of the model parameters.

Return type:

float

estimate(nobs, suff_stat)[source]

Estimate MarkovChainDistribution from aggregated sufficient statistics from observed data.

Arg suff_stat is a Tuple of length three containing,

suff_stat[0] (Dict[T, float]): Maps initial state values to their aggregated counts. suff_stat[1] (Dict[T, Dict[T, List[float]]]): Maps state to state transition counts. suff_stat[2] (T1): Sufficient statistic value of length accumulator. (Assumed type T1).

If member variable pseudo_count is set estimate1() is called to aggregated weighted sufficient statistics. Else estimate0() is called to obtain estimates for MarkovChainDistribution directly from arg ‘suff_stat’.

Parameters:

  • nobs (Optional[float]) – Number of observations. Passed to estimate1() or estimate2().

  • suff_stat (tuple[dict[T, float], dict[T, dict[T, float]], Any | None]) – Seed above for details.

Returns:

MarkovChainDistribution object.

Return type:

MarkovChainDistribution

estimate0(nobs, suff_stat)[source]

Estimate MarkovChainDistribution from aggregated sufficient statistics from observed data.

Maximum likelihood estimates for initial state probabilities, transition probabilities, and the length distribution are obtained directly from aggregated data in ‘suff_stat’.

Arg suff_stat is a Tuple of length three containing,

suff_stat[0] (Dict[T, float]): Maps initial state values to their aggregated counts. suff_stat[1] (Dict[T, Dict[T, List[float]]]): Maps state to state transition counts. suff_stat[2] (T1): Sufficient statistic value of length accumulator. (Assumed type T1).

Parameters:

  • nobs (Optional[float]) – Number of observations. Passed to estimate1() or estimate2().

  • suff_stat (tuple[dict[T, float], dict[T, dict[T, float]], Any | None]) – Seed above for details.

Returns:

MarkovChainDistribution object.

Return type:

MarkovChainDistribution

estimate1(nobs, suff_stat)[source]

Estimate MarkovChainDistribution from aggregated sufficient statistics from observed data.

Maximum likelihood estimates for initial state probabilities, transition probabilities, and the length distribution are obtained by a weighted aggregation of sufficient statistics in ‘suff_stat’, and member variables of MarkovChainEstimator object.

Arg suff_stat is a Tuple of length three containing,

suff_stat[0] (Dict[T, float]): Maps initial state values to their aggregated counts. suff_stat[1] (Dict[T, Dict[T, List[float]]]): Maps state to state transition counts. suff_stat[2] (T1): Sufficient statistic value of length accumulator. (Assumed type T1).

Parameters:

  • nobs (Optional[float]) – Number of observations. Passed to estimate1() or estimate2().

  • suff_stat (tuple[dict[T, float], dict[T, dict[T, float]], Any | None]) – Seed above for details.

Returns:

MarkovChainDistribution object.

Return type:

MarkovChainDistribution

class MarkovChainDataEncoder(len_encoder=NullDataEncoder())[source]

Bases: DataSequenceEncoder

Parameters:

len_encoder (DataSequenceEncoder)

seq_encode(x)[source]

Sequence encoding a sequence of iid Markov chain observations with data type T.

The returned value is (rv) is a Tuple of length 8 with entries:

rv[0] (int): Number of total observations (number of Markov sequences). rv[1] (ndarray[int]): Sequence index for initial state observations. rv[2] (ndarray[int]): Sequence index for non-initial state observations in a sequence greater than len 1. rv[3] (ndarray[int]): Numpy array of observations index in inv_key_map for initial states. rv[4] (ndarray[int]): State-to-state index value of inv_key_map for initial state value. rv[5] (ndarray[int]): State-to-state index value of inv_key_map for transition. rv[6] (ndarray[T]): Maps integer index value to value in state-space (T). rv[7] (Optional[T1]): Encoded sequence of lengths from len_encoder. None if no length distributon to be

estimated.

Parameters:

x (List[List[T]]) – Sequence of iid observations of Markov chain sequences.

Returns:

Tuple of length 8. See above for details.

Return type:

tuple[int, ndarray, ndarray, ndarray, ndarray, ndarray, ndarray, Any]