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 of one fully observed trajectory.

Parameters:

x (Any)

Return type:

float

log_density(x)[source]

Return the log-likelihood of one fully observed CTMC trajectory.

Parameters:

x (Any)

Return type:

float

seq_log_density(x)[source]

Return vectorized log-likelihoods for encoded trajectory statistics.

Parameters:

x (Any)

Return type:

ndarray

sampler(seed=None)[source]

Return a Gillespie sampler for trajectories from this CTMC.

Parameters:

seed (int | None)

Return type:

ContinuousTimeMarkovChainSampler

estimator(pseudo_count=None)[source]

Return the closed-form rate estimator for this state space.

Parameters:

pseudo_count (float | None)

Return type:

ContinuousTimeMarkovChainEstimator

dist_to_encoder()[source]

Return the trajectory-statistics encoder used by 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 one trajectory or a list of trajectories over the configured horizon.

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]

Update transition-count and dwell-time statistics from one trajectory.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics from one trajectory.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Update sufficient statistics from encoded trajectory statistics.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from encoded trajectory statistics.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge transition-count and dwell-time sufficient statistics.

Parameters:

suff_stat (tuple[ndarray, ndarray])

Return type:

ContinuousTimeMarkovChainAccumulator

value()[source]

Return copies of transition counts and dwell times.

Return type:

tuple[ndarray, ndarray]

from_value(x)[source]

Restore transition-count and dwell-time sufficient statistics.

Parameters:

x (tuple[ndarray, ndarray])

Return type:

ContinuousTimeMarkovChainAccumulator

scale(c)[source]

Scale accumulated sufficient statistics by a constant.

Parameters:

c (float)

Return type:

ContinuousTimeMarkovChainAccumulator

key_merge(stats_dict)[source]

Merge keyed transition and dwell statistics into this accumulator.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)[source]

Replace this accumulator’s state from keyed statistics when present.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()[source]

Return the encoder compatible with CTMC sufficient statistics.

Return type:

ContinuousTimeMarkovChainDataEncoder

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

Bases: StatisticAccumulatorFactory

Create accumulators for CTMC transition-count and dwell-time statistics.

Parameters:

  • num_states (int)

  • name (str | None)

  • keys (str | None)

make()[source]

Create an empty CTMC accumulator.

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 a factory for CTMC sufficient-statistic accumulators.

Return type:

ContinuousTimeMarkovChainAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate off-diagonal generator rates from transition counts and dwell times.

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 trajectories as per-trajectory transition counts and dwell times.

Parameters:

x (Sequence[Any])

Return type:

list[tuple[ndarray, ndarray]]