mixle.stats.sequences.integer_markov_chain module

Create, estimate, and sample from a Markov chain with support on the integers (chain can include a lag).

Defines the IntegerMarkovChainDistribution, IntegerMarkovChainSampler, IntegerMarkovChainAccumulatorFactory, IntegerMarkovChainAccumulator, IntegerMarkovChainEstimator, and the IntegerMarkovChainDataEncoder classes for use with mixle.

The data type: Sequence[int].

Consider a sequence of length n > 0 s.t. x = (x[0],x[1],…,x[n-1]). With lag > 0, we have the integer Markov chain has a log-density given by:

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

log(P_len(n)),

where P_len(n) is the density for the length distribution evaluated for length ‘n’, and P_init() is the density for the initial distribution. If the sequence length is less than the lag, i.e. len(x) < lag, then

log(P(x)) = log(P_len(n)).

Note: P_len() should be compatible with non-negative integers. P_init() must be compatible with sequences of ints.

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

compute_declaration()

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.

Consider a sequence of length n > 0 s.t. x = (x[0],x[1],…,x[n-1]). With lag > 0, we have log-density given by:

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

log(P_len(n)),

where P_len(n) is the density for the length distribution evaluated for length ‘n’, and P_init() is the density for the initial distribution. If the sequence length is less than the lag, i.e. len(x) < lag, then

log(P(x)) = log(P_len(n)).

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]): Lengths of chains - lag. If less than lag length is 0. 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 legacy (transition_counts, initial_stat, length_stat) 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)

Returns an IntegerMarkovChainSampler object.

Parameters:

seed (int | None)

Return type:

IntegerMarkovChainSampler

estimator(pseudo_count=None)

Returns an IntegerMarkovChainEstimator object.

Parameters:

pseudo_count (float | None)

dist_to_encoder()

Returns an IntegerMarkovChainDataEncoder object for encoding sequences of 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

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(lag, init_accumulator=NullAccumulator(), len_accumulator=NullAccumulator(), keys=None, name=None)

Bases: SequenceEncodableStatisticAccumulator

Parameters:

• lag (int)

• init_accumulator (SequenceEncodableStatisticAccumulator | None)

• len_accumulator (SequenceEncodableStatisticAccumulator | None)

• keys (str | None)

• name (str | None)

update(x, weight, estimate)

Update the sufficient statistics of object instance with a single 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]): Lengths of chains - lag. If less than lag length is 0. 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]): Lengths of chains - lag. If less than lag length is 0. 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 sufficient statistics with object instance.

Arg suff_stat is a Tuple of length 3 containing:

suff_stat[0] (Dict[Tuple[Tuple[int, …], int], float]): Dictionary mapping state transition counts. suff_stat[1] (Optional[SS1]): Optional sufficient statistics for init accumulator of type SS1. suff_stat[2] (Optional[SS2]): Optional sufficient statistics for length accumulator of type SS2.

Parameters:

suff_stat (tuple[dict[tuple[tuple[int, ...], int], float], SS1 | None, SS2 | None]) – See above for details.

Returns:

IntegerMarkovAccumulator object.

Return type:

IntegerMarkovChainAccumulator

value()

Returns sufficient statistics of integer Markov chain.

Returned suff_stat is a Tuple of length 3 containing:

suff_stat[0] (Dict[Tuple[Tuple[int, …], int], float]): Dictionary mapping state transition counts. suff_stat[1] (Optional[SS1]): Optional sufficient statistics for init accumulator of type SS1. suff_stat[2] (Optional[SS2]): Optional sufficient statistics for length accumulator of type SS2.

Returns:

Tuple[Dict[Tuple[Tuple[int, …], int], float], Optional[SS1], Optional[SS2]].

Return type:

tuple[dict[tuple[tuple[int, …], int], float], Any | None, Any | None]

from_value(x)

Set sufficient statistics of object instance to aggregated sufficient statistics in arg ‘x’.

Arg value ‘x’ is a Tuple of length 3 containing:

x[0] (Dict[Tuple[Tuple[int, …], int], float]): Dictionary mapping state transition counts. x[1] (Optional[SS1]): Optional sufficient statistics for init accumulator of type SS1. x[2] (Optional[SS2]): Optional sufficient statistics for length accumulator of type SS2.

Parameters:

x (tuple[dict[tuple[tuple[int, ...], int], float], SS1 | None, SS2 | None]) – See above for details.

Returns:

IntegerMarkovChainAccumulator object.

Return type:

IntegerMarkovChainAccumulator

key_merge(stats_dict)

Merge sufficient statistics of object instance with matching keys in stats_dict.

Parameters:

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

Returns:

None.

Return type:

None

key_replace(stats_dict)

Replace the sufficient statistics of object instance with those containing matching keys in ‘stats_dict’.

Parameters:

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

Returns:

None.

Return type:

None

acc_to_encoder()

Returns IntegerMarkovChainDataEncoder object.

Return type:

IntegerMarkovChainDataEncoder

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

Bases: StatisticAccumulatorFactory

Parameters:

• lag (int)

• init_factory (StatisticAccumulatorFactory | None)

• len_factory (StatisticAccumulatorFactory | None)

• keys (str | None)

• name (str | None)

make()

Returns an IntegerMarkovChainAccumulator object from instance.

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

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)

accumulator_factory()

Returns an IntegerMarkovChainAccumulatorFactory object from attributes values.

Return type:

IntegerMarkovChainAccumulatorFactory

estimate(nobs, suff_stat)

Estimate IntegerMarkovChainDistribution object from aggregated sufficient statistics in arg ‘suff_stat’.

Arg ‘suff_stat’ is a Tuple of length 3 containing:

suff_stat[0] (Dict[Tuple[Tuple[int, …], int], float]): Dictionary mapping state transition counts. suff_stat[1] (Optional[SS1]): Optional sufficient statistics for init accumulator of type SS1. suff_stat[2] (Optional[SS2]): Optional sufficient statistics for length accumulator of type SS2.

Parameters:

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

• suff_stat (tuple[dict[tuple[tuple[int, ...], int], float], SS1 | None, SS2 | None]) – See above for details.

Returns:

IntegerMarkovChainDistribution object.

Return type:

IntegerMarkovChainDistribution

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

Bases: DataSequenceEncoder

Parameters:

• lag (int)

• init_encoder (DataSequenceEncoder)

• len_encoder (DataSequenceEncoder)

seq_encode(x)

Encode sequence of iid observations from integer Markov chain.

Returns a Tuple of length 7 containing:

seq_len (ndarray[int]): Lengths of chains - lag. If less than lag length is 0. 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]