mixle.stats.processes.ctmc module

Continuous-time Markov chain (CTMC) over fully observed trajectories.

A CTMC on K states is governed by a generator matrix Q: off-diagonal q_ij >= 0 is the rate of jumping i -> j, and q_ii = -sum_{j!=i} q_ij. A fully-observed trajectory is the initial state plus the sequence of (dwell_time, next_state) jumps; its log-likelihood is

log L = sum_{i!=j} n_ij * log q_ij - sum_i q_i * T_i, q_i = -q_ii = sum_{j!=i} q_ij,

where n_ij is the number of observed i->j transitions and T_i the total time spent in i. This is a collection of independent Poisson-rate likelihoods, so the MLE is closed form and unique: q_ij = n_ij / T_i. The estimator therefore certifies GLOBAL_UNIQUE (see mixle.inference.certify(), which classifies this family). Data type: (s0, [(dt, s1), (dt, s2), ...]) – the initial state and the observed jumps.

The family follows the standard Mixle distribution contract (Distribution / Sampler / Accumulator / Factory / Estimator / DataEncoder) so it composes with optimize / seq_log_density / the PPL surface like every other family.

class ContinuousTimeMarkovChainDistribution(rates, initial_state=0, horizon=10.0, name=None, keys=None)[source]

Bases: SequenceEncodableProbabilityDistribution

CTMC on K states with generator Q (off-diagonal rates); MLE is closed form (GLOBAL_UNIQUE).

Parameters:

  • rates (np.ndarray)

  • initial_state (int)

  • horizon (float)

  • name (str | None)

  • keys (str | None)

property generator: ndarray

The generator matrix Q (off-diagonal rates, diagonal = -exit rate).

density(x)[source]

Return the probability density or mass at a single observation.

Concrete default: exponentiate log_density (the abstract method subclasses must provide). Leaves with a cheaper closed form may override this.

Parameters:

x (Any)

Return type:

float

log_density(x)[source]

Return the log-density or log-mass at a single observation.

Parameters:

x (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:

ContinuousTimeMarkovChainSampler

estimator(pseudo_count=None)[source]

Return an estimator for fitting this distribution from data.

Parameters:

pseudo_count (float | None)

Return type:

ContinuousTimeMarkovChainEstimator

dist_to_encoder()[source]

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

Return type:

ContinuousTimeMarkovChainDataEncoder

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

Bases: DistributionSampler

Exact Gillespie simulation of CTMC trajectories on [0, horizon].

Parameters:

  • dist (ContinuousTimeMarkovChainDistribution)

  • seed (int | None)

sample(size=None)[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:

size (int | None)

Return type:

Any

class ContinuousTimeMarkovChainAccumulator(num_states, name=None, keys=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulate the CTMC sufficient statistics: transition counts n_ij and dwell times T_i.

Parameters:

  • num_states (int)

  • name (str | None)

  • keys (str | None)

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

None

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

None

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

None

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

None

combine(suff_stat)[source]
Parameters:

suff_stat (tuple[ndarray, ndarray])

Return type:

ContinuousTimeMarkovChainAccumulator

value()[source]
Return type:

tuple[ndarray, ndarray]

from_value(x)[source]
Parameters:

x (tuple[ndarray, ndarray])

Return type:

ContinuousTimeMarkovChainAccumulator

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:

ContinuousTimeMarkovChainAccumulator

key_merge(stats_dict)[source]

Pool this accumulator’s statistics into stats_dict under its merge key.

The structural default implements the common single-key pattern: store the accumulator under self.keys the first time the key is seen, else combine into the one already there. Accumulators with several named keys (e.g. an HMM’s init/trans/state keys) or a non-accumulator stats payload override this. A keys of None (the default) is a no-op.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)[source]

Replace this accumulator’s statistics from the pooled stats_dict entry (see key_merge).

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()[source]
Return type:

ContinuousTimeMarkovChainDataEncoder

class ContinuousTimeMarkovChainAccumulatorFactory(num_states, name=None, keys=None)[source]

Bases: StatisticAccumulatorFactory

Parameters:

  • num_states (int)

  • name (str | None)

  • keys (str | None)

make()[source]
Return type:

ContinuousTimeMarkovChainAccumulator

class ContinuousTimeMarkovChainEstimator(num_states, pseudo_count=None, name=None, keys=None)[source]

Bases: ParameterEstimator

Closed-form rate MLE: q_ij = n_ij / T_i (independent Poisson rates, unique global optimum).

Parameters:

  • num_states (int)

  • pseudo_count (float | None)

  • name (str | None)

  • keys (str | None)

accumulator_factory()[source]
Return type:

ContinuousTimeMarkovChainAccumulatorFactory

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

ContinuousTimeMarkovChainDistribution

class ContinuousTimeMarkovChainDataEncoder(num_states)[source]

Bases: DataSequenceEncoder

Encode (s0, jumps) trajectories into per-trajectory (counts, dwell) sufficient statistics.

Parameters:

num_states (int)

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Any])

Return type:

list[tuple[ndarray, ndarray]]