mixle.stats.sequences.integer_markov_chain module

Integer Markov-chain distributions with optional lagged transitions.

The data type: Sequence[int].

For a sequence x of length n >= lag > 0, the log density is

log(P(x)) = log(P_init(x[0:lag]) + sum_{j=0}^{n-lag-1} log(p_mat(x[j + lag] | x[j],…,x[j+lag-1])) +

log(P_len(n)),

where P_len is evaluated at the actual length and P_init generates exactly lag bounded-integer states. Shorter sequences are outside the declared domain.

exception NonGenerativeIntegerMarkovChainError

Bases: TypeError

Raised when a conditional integer-chain factor is asked to generate sequences.

class IntegerMarkovChainStatistics(schema_version, num_values, lag, transition_counts, initial_nobs, initial, length_nobs, length)

Bases: NamedTuple

Immutable sufficient statistics for one fixed integer-chain support.

Parameters:

• schema_version (int)

• num_values (int)

• lag (int)

• transition_counts (tuple[tuple[tuple[int, ...], int, float], ...])

• initial_nobs (float)

• initial (Any | None)

• length_nobs (float)

• length (Any | None)

schema_version: int

Alias for field number 0

num_values: int

Alias for field number 1

lag: int

Alias for field number 2

transition_counts: tuple[tuple[tuple[int, ...], int, float], ...]

Alias for field number 3

initial_nobs: float

Alias for field number 4

initial: Any | None

Alias for field number 5

length_nobs: float

Alias for field number 6

length: Any | None

Alias for field number 7

class IntegerMarkovChainDistribution(num_values, cond_dist, lag=1, init_dist=NullDistribution(), len_dist=NullDistribution(), keys=None, name=None)

Bases: SequenceEncodableProbabilityDistribution

Markov-chain distribution over integer-valued states.

Parameters:

• num_values (int)

• cond_dist (list[list[float]] | ndarray)

• lag (int)

• init_dist (SequenceEncodableProbabilityDistribution | None)

• len_dist (SequenceEncodableProbabilityDistribution | None)

• keys (str | None)

• name (str | None)

compute_capabilities()

Declare generated-compute support inherited from initial and length distributions.

compute_declaration()

Return the generated-compute declaration for the integer Markov chain.

density_semantics()

Classify missing initial/length laws as conditional likelihood factors.

density(x)

Density of integer Markov chain evaluated at x.

See log_density() for details.

Parameters:

x (Sequence[int]) – An integer markov chain observation.

Returns:

Density evaluated at x.

Return type:

float

log_density(x)

Log-density of integer Markov chain evaluated at x.

The initial child scores exactly the first lag states, each later state is scored by its lag-indexed transition row, and the length child scores the actual sequence length. Shorter observations are rejected.

Parameters:

x (Sequence[int]) – An integer markov chain observation.

Returns:

Log-density evaluated at x.

Return type:

float

seq_log_density(x)

Vectorized evaluation of log-density at every observation in encoded sequence.

See log_density() for details on likelihood evaluation.

Sequence encoded arg ‘x’ is a Tuple of length 7 containing:

seq_len (ndarray[int]): Actual observed sequence lengths. init_idx (ndarray[int]): Observed sequence index of chains with lengths >= lag. seq_idx (ndarray[int]): Observed sequence index of chains with transitions. u_seq_idx (ndarray[object]): Numpy array of tuples containing the unique transitions. u_seq_values (ndarray[object]): Numpy array of tuples containing the transitions. init_enc (Optional[E]): Sequence encoding of initial values (has type E). len_enc (Optional[E2]): Sequence encoding of length values (has type E2).

Parameters:

x (tuple[ndarray, ndarray, ndarray, ndarray, ndarray, ndarray, E1 | None, E2 | None]) – See above for details.

Returns:

Log-density evaluated at each observation in encoded sequence.

Return type:

ndarray

backend_seq_log_density(x, engine)

Engine-neutral vectorized log-density for grouped integer Markov-chain encodings.

Parameters:

• x (tuple[ndarray, ndarray, ndarray, ndarray, ndarray, E1 | None, E2 | None])

• engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)

Return stacked integer Markov-chain parameters for shared support/lag.

Parameters:

• dists (Sequence[IntegerMarkovChainDistribution])

• engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)

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

Parameters:

• x (tuple[ndarray, ndarray, ndarray, ndarray, ndarray, E1 | None, E2 | None])

• params (dict[str, Any])

• engine (Any)

Return type:

Any

classmethod backend_stacked_sufficient_statistics_with_estimator(x, weights, params, engine, estimator)

Return per-component versioned fixed-support sufficient statistics.

Parameters:

• x (tuple[ndarray, ndarray, ndarray, ndarray, ndarray, E1 | None, E2 | None])

• weights (Any)

• params (dict[str, Any])

• engine (Any)

• estimator (Any)

Return type:

tuple[Any, …]

sampler(seed=None)

Return a sampler for this integer Markov chain.

Parameters:

seed (int | None)

Return type:

IntegerMarkovChainSampler

transition_sampler(seed=None)

Return a sampler exposing only sample_given for conditional transition use.

Parameters:

seed (int | None)

Return type:

IntegerMarkovChainSampler

estimator(pseudo_count=None)

Return an estimator initialized from this integer Markov chain.

Parameters:

pseudo_count (float | None)

dist_to_encoder()

Return a data encoder for iid integer Markov-chain observations.

Return type:

IntegerMarkovChainDataEncoder

enumerator()

Returns IntegerMarkovChainEnumerator iterating integer sequences in descending probability order.

Return type:

IntegerMarkovChainEnumerator

class IntegerMarkovChainEnumerator(dist)

Bases: DistributionEnumerator

Enumerates integer Markov-chain sequences in descending probability order.

Parameters:

dist (IntegerMarkovChainDistribution)

class IntegerMarkovChainSampler(dist, seed)

Bases: DistributionSampler

Draw integer-valued sequences from an IntegerMarkovChainDistribution.

Parameters:

• dist (IntegerMarkovChainDistribution)

• seed (int | None)

single_sample()

Returns a single sample from the integer Markov chain distribution.

Return type:

Sequence[int]

sample(size=None, *, batched=True)

Draw iid samples from an integer Markov chain distribution.

With batched=True (default) and size not None, the lengths and initial states are drawn per chain (byte-identical to the loop) and the lag-conditional transitions are vectorized across all live chains at each time index. The transition draws change RNG consumption order, so the output is statistically equivalent but NOT byte-identical to batched=False. Set batched=False to reproduce the exact legacy per-sequence output for a given seed.

Parameters:

• size (Optional[int]) – If None, size is taken to be 0.

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

Returns:

Sequence[int] if size is None, else List[Sequence[int]] with length equal to size.

Return type:

list[Sequence[int]] | Sequence[int]

sample_given(x)

Sample from the Markov chain conditioned on a given value ‘x’.

Parameters:

x (Sequence[int]) – Sample from Markov chain conditioned on observing ‘x’.

Returns:

Single sample transition from integer Markov chain.

Return type:

int

class IntegerMarkovChainAccumulator(num_values, lag, init_accumulator=NullAccumulator(), len_accumulator=NullAccumulator(), keys=None, name=None)

Bases: SequenceEncodableStatisticAccumulator

Accumulate transition, initial-state, and length statistics for Markov chains.

Parameters:

• num_values (int)

• lag (int)

• init_accumulator (SequenceEncodableStatisticAccumulator | None)

• len_accumulator (SequenceEncodableStatisticAccumulator | None)

• keys (str | None)

• name (str | None)

update(x, weight, estimate)

Update sufficient statistics with a single weighted observation.

Parameters:

• x (Sequence[int]) – An observation from an integer Markov chain.

• weight (float) – Observation weight.

• estimate (Optional[IntegerMarkovChainDistribution]) – Optional previous estimate.

Returns:

None.

Return type:

None

initialize(x, weight, rng)

Initialize sufficient statistics from a single observation.

Note: Calls _rng_initialize() to ensure consistency with seq_initialize() function.

Parameters:

• x (Sequence[int]) – An observation from an integer Markov chain.

• weight (float) – Observation weight.

• rng (RandomState) – RandomState for initializing sufficient statistics.

Returns:

None.

Return type:

None

seq_update(x, weights, estimate)

Vectorized update of sufficient statistics from an encoded sequence of observations ‘x’.

Sequence encoded arg ‘x’ is a Tuple of length 7 containing:

seq_len (ndarray[int]): Actual observed sequence lengths. init_idx (ndarray[int]): Observed sequence index of chains with lengths >= lag. seq_idx (ndarray[int]): Observed sequence index of chains with transitions. u_seq_idx (ndarray[object]): Numpy array of tuples containing the unique transitions. u_seq_values (ndarray[object]): Numpy array of tuples containing the transitions. init_enc (Optional[E]): Sequence encoding of initial values (has type E). len_enc (Optional[E2]): Sequence encoding of length values (has type E2).

Parameters:

• x (tuple[ndarray, ndarray, ndarray, ndarray, ndarray, ndarray, E1 | None, E2 | None]) – See above for details.

• weights (np.ndarray) – Numpy array of observation weights.

• estimate (Optional[IntegerMarkovChainDistribution]) – Optional previous estimate.

Returns:

None.

Return type:

None

seq_update_engine(x, weights, estimate, engine)

Engine-resident E-step: per-unique-transition counts are reduced on the active engine before being scattered into the sparse transition dict; the init/len children are routed through the engine. Matches seq_update.

Parameters:

• x (tuple[ndarray, ndarray, ndarray, ndarray, ndarray, ndarray, E1 | None, E2 | None])

• weights (Any)

• estimate (IntegerMarkovChainDistribution | None)

• engine (Any)

Return type:

None

seq_initialize(x, weights, rng)

Vectorized initialization of sufficient statistics from an encoded sequence of observations in ‘x’.

Note: Calls _rng_initialize() to ensure consistency with seq_initialize() function.

Sequence encoded arg ‘x’ is a Tuple of length 7 containing:

seq_len (ndarray[int]): Actual observed sequence lengths. init_idx (ndarray[int]): Observed sequence index of chains with lengths >= lag. seq_idx (ndarray[int]): Observed sequence index of chains with transitions. u_seq_idx (ndarray[object]): Numpy array of tuples containing the unique transitions. u_seq_values (ndarray[object]): Numpy array of tuples containing the transitions. init_enc (Optional[E]): Sequence encoding of initial values (has type E). len_enc (Optional[E2]): Sequence encoding of length values (has type E2).

Parameters:

• x (tuple[ndarray, ndarray, ndarray, ndarray, ndarray, ndarray, E1 | None, E2 | None]) – See above for details.

• weights (np.ndarray) – Numpy array of observation weights.

• rng (RandomState) – RandomState for initializing sufficient statistics.

Returns:

None.

Return type:

None

combine(suff_stat)

Combine another versioned fixed-support statistic into this accumulator.

Parameters:

suff_stat (IntegerMarkovChainStatistics) – See above for details.

Returns:

This accumulator after combination.

Return type:

IntegerMarkovChainAccumulator

value()

Return copied child statistics and immutable canonical transition counts.

Return type:

IntegerMarkovChainStatistics

from_value(x)

Restore accumulator state from a versioned fixed-support statistic.

Parameters:

x (IntegerMarkovChainStatistics) – See above for details.

Returns:

This accumulator after restoration.

Return type:

IntegerMarkovChainAccumulator

scale(c)

Scale numeric evidence while preserving support and schema metadata.

Parameters:

c (float)

Return type:

IntegerMarkovChainAccumulator

key_merge(stats_dict)

Merge this accumulator into keyed sufficient statistics.

Parameters:

stats_dict (Dict[str, Any]) – Dictionary mapping keys to corresponding sufficient statistics.

Returns:

None.

Return type:

None

key_replace(stats_dict)

Replace this accumulator’s statistics from matching keyed values.

Parameters:

stats_dict (Dict[str, Any]) – Dictionary mapping keys to corresponding sufficient statistics.

Returns:

None.

Return type:

None

acc_to_encoder()

Return a data encoder built from the child accumulators.

Return type:

IntegerMarkovChainDataEncoder

class IntegerMarkovChainAccumulatorFactory(num_values, lag, init_factory=NullAccumulatorFactory(), len_factory=NullAccumulatorFactory(), keys=None, name=None)

Bases: StatisticAccumulatorFactory

Create integer Markov-chain accumulators with child accumulator factories.

Parameters:

• num_values (int)

• lag (int)

• init_factory (StatisticAccumulatorFactory | None)

• len_factory (StatisticAccumulatorFactory | None)

• keys (str | None)

• name (str | None)

make()

Return a new integer Markov-chain accumulator.

Return type:

IntegerMarkovChainAccumulator

class IntegerMarkovChainEstimator(num_values, lag=1, init_estimator=NullEstimator(), len_estimator=NullEstimator(), init_dist=None, len_dist=None, pseudo_count=None, name=None, keys=None)

Bases: ParameterEstimator

Estimate integer Markov-chain transition probabilities and child models.

Parameters:

• num_values (int)

• lag (int)

• init_estimator (ParameterEstimator | None)

• len_estimator (ParameterEstimator | None)

• init_dist (SequenceEncodableProbabilityDistribution | None)

• len_dist (SequenceEncodableProbabilityDistribution | None)

• pseudo_count (float | None)

• name (str | None)

• keys (str | None)

get_prior()

Returns the implicit per-row symmetric-Dirichlet prior pseudo_count induces, or None.

pseudo_count smoothing (cond_mat += pseudo_count before row-normalizing in estimate()) is the exact MAP point estimate under an independent SymmetricDirichletDistribution(pseudo_count + 1) prior on each row of the conditional matrix – see model_log_density(). Lets mixle.inference.estimation.optimize() auto-detect the 'map' objective instead of silently tracking plain (unpenalized) MLE for a pseudo-count-regularized fit.

model_log_density(model)

Log-density of the fitted conditional matrix under the implicit row prior (see get_prior()), plus init_dist/len_dist’s own model_log_density when the estimator that actually fit them (rather than a fixed, caller-supplied distribution) exposes one. Returns 0.0 (a plain MLE fit) when pseudo_count is None.

Parameters:

model (IntegerMarkovChainDistribution)

Return type:

float

accumulator_factory()

Return an accumulator factory configured from this estimator.

Return type:

IntegerMarkovChainAccumulatorFactory

estimate(nobs, suff_stat)

Estimate an integer Markov-chain distribution from versioned sufficient statistics.

Parameters:

• nobs (Optional[float]) – Number of observations used in aggregation of ‘suff_stat’.

• suff_stat (IntegerMarkovChainStatistics) – See above for details.

Returns:

IntegerMarkovChainDistribution object.

Return type:

IntegerMarkovChainDistribution

class IntegerMarkovChainDataEncoder(num_values, lag, init_encoder=NullDataEncoder(), len_encoder=NullDataEncoder())

Bases: DataSequenceEncoder

Encode integer-valued sequences for vectorized Markov-chain scoring.

Parameters:

• num_values (int)

• lag (int)

• init_encoder (DataSequenceEncoder)

• len_encoder (DataSequenceEncoder)

seq_encode(x)

Encode iid observations from an integer Markov chain.

The returned tuple contains:

seq_len (ndarray[int]): Actual observed sequence lengths. init_idx (ndarray[int]): Observed sequence index of chains with lengths >= lag. seq_idx (ndarray[int]): Observed sequence index of chains with transitions. u_seq_idx (ndarray[object]): Numpy array of tuples containing the unique transitions. u_seq_values (ndarray[object]): Numpy array of tuples containing the transitions. init_enc (Optional[E]): Sequence encoding of initial values (has type E). len_enc (Optional[E2]): Sequence encoding of length values (has type E2).

Parameters:

x (List[Sequence[int]]) – Sequence of iid observations from integer markov chain distribution.

Returns:

See above for details.

Return type:

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

row_count(x)

Return the number of rows after validating integer-chain index geometry.

Parameters:

x (Any)

Return type:

int