mixle.stats.combinator.optional module¶

Create, estimate, and sample from an Optional distribution.

Defines the OptionalDistribution, OptionalSampler, OptionalAccumulatorFactory, OptionalAccumulator, OptionalEstimator, and the OptionalDataEncoder classes for use with mixle.

This distribution assigns a probability (p) to data being missing. With probability (1-p) the data is assumed to come from a base distribution set by the user.

The OptionalDistribution allows for potentially missing data. The value p (the probability of being missing) must be specified to sample from the distribution.

class OptionalDistribution(dist, p=None, missing_value=None, name=None, prior=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Mixture-style wrapper that models missing observations explicitly.

Parameters:

dist (SequenceEncodableProbabilityDistribution)
p (float | None)
missing_value (Any)
name (str | None)
prior (tuple[Any, Any] | None)

compute_capabilities()[source]

get_prior()[source]

Return the joint prior as (p_prior, dist_prior).

Return type:: tuple[Any, Any]

set_prior(prior)[source]

Distribute the joint prior (p_prior, dist_prior) to the missing probability and base dist.

prior=None is a no-op (point model, existing behavior byte-identical). Otherwise the first element is a conjugate Beta prior on p (caching the digamma expectations used by expected_log_density) and the second is pushed to the base distribution’s set_prior.

Parameters:: prior (tuple[Any, Any] | None)
Return type:: None

expected_log_density(x)[source]

Posterior-expected log-density E_q[log p(x)] at x.

With a conjugate Beta prior on p the expectation over p is available in closed form via digamma terms (missing => da - dab; observed => db - dab + dist.expected_log_density(x)); otherwise this falls back to the plug-in log_density.

Parameters:: x (T)
Return type:: float

seq_expected_log_density(x)[source]

Vectorized posterior-expected log-density; falls back to seq_log_density without a prior.

Parameters:: x (tuple[int, ndarray, ndarray, E])
Return type:: ndarray

compute_declaration()[source]

density(x)[source]

Evaluate the density of the Optional distribution at x.

See log_density() for details.

Parameters:: x (T) – Observation from base dist or missing value.
Returns:: Density at x.
Return type:: float

density_semantics()[source]

What log_density returns relative to the true log-density (default: exact).

Override to declare that this distribution’s log_density is a variational lower bound (ELBO), an upper bound, or an approximation rather than the exact log p(x). This is surfaced as the ExactDensity capability and noted in mixle.describe(), so code that needs an exact likelihood can require(x, ExactDensity) instead of silently trusting a bound.

log_density(x)[source]

Evalute the log density of the Optional distribution at x.

If x is a missing value: return log(p) if p is not None, else return 0.0 If x is not the missing_value: if p is not None, return the log_denisty(x) at base dist + log(1-p) else: return

log_density(x).

Parameters:: x (T) – Observation from base dist or missing value.
Returns:: Log-density at x.
Return type:: float

seq_log_density(x)[source]

Return vectorized log-density values for sequence-encoded observations.

Parameters:: x (tuple[int, ndarray, ndarray, E])
Return type:: ndarray

backend_seq_log_density(x, engine)[source]

Engine-neutral vectorized log-density for optional encoded data.

Parameters:

x (tuple[int, ndarray, ndarray, E])
engine (Any)

Return type:

Any

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

Return distribution-owned state for autograd fitting.

Parameters:

engine (Any)
torch (Any)
leaves (list[Any])
recurse (Any)
tensor_param (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)[source]

Return stacked optional-wrapper parameters for homogeneous mixture kernels.

Parameters:

dists (Sequence[OptionalDistribution])
engine (Any)

Return type:

dict[str, Any]

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

Return an (n, k) matrix of optional-wrapper log densities.

Parameters:

x (tuple[int, ndarray, ndarray, E])
params (dict[str, Any])
engine (Any)

Return type:

Any

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

Return per-component legacy optional-wrapper sufficient statistics.

Parameters:

x (tuple[int, ndarray, ndarray, E])
weights (Any)
params (dict[str, Any])
engine (Any)
estimator (Any)

Return type:

tuple[Any, …]

to_fisher(**kwargs)[source]: Fisher view for the optional/missing-gate.

sampler(seed=None)[source]

Return a sampler for drawing observations from this distribution.

Parameters:: seed (int | None)
Return type:: OptionalSampler

estimator(pseudo_count=None)[source]

Return an estimator for fitting this distribution from data.

Parameters:: pseudo_count (float | None)
Return type:: OptionalEstimator

dist_to_encoder()[source]

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

Return type:: OptionalDataEncoder

enumerator()[source]

Returns an OptionalEnumerator iterating the support (including the missing value) in descending probability order.

Return type:: OptionalEnumerator

class OptionalEnumerator(dist)[source]

Bases: DistributionEnumerator

Parameters:: dist (OptionalDistribution)

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

Bases: DistributionSampler

Parameters:

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

class OptionalEstimatorAccumulator(accumulator, missing_value=None, name=None, keys=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Parameters:

accumulator (SequenceEncodableStatisticAccumulator)
missing_value (Any)
name (str | None)
keys (str | None)

update(x, weight, estimate)[source]

Parameters:

x (T)
weight (float)
estimate (OptionalDistribution)

Return type:

None

initialize(x, weight, rng)[source]

Parameters:

x (T)
weight (float)
rng (RandomState)

Return type:

None

seq_update(x, weights, estimate)[source]

Parameters:

x (tuple[int, ndarray, ndarray, E])
weights (ndarray)
estimate (OptionalDistribution)

Return type:

None

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

Engine-resident E-step: missing/observed mass is summed on the active engine and the observed child accumulator is routed through the engine. Matches seq_update.

Parameters:

x (tuple[int, ndarray, ndarray, E])
weights (Any)
estimate (OptionalDistribution)
engine (Any)

Return type:

None

seq_initialize(x, weights, rng)[source]

Parameters:

x (tuple[int, ndarray, ndarray, E])
weights (ndarray)
rng (RandomState)

Return type:

None

combine(suff_stat)[source]

Parameters:: suff_stat (tuple[list[float], SS])
Return type:: OptionalEstimatorAccumulator

value()[source]

Return type:: tuple[list[float], Any]

from_value(x)[source]

Parameters:: x (tuple[list[float], SS])
Return type:: OptionalEstimatorAccumulator

scale(c)[source]

Scale missing/observed weights and delegate observed statistics.

Parameters:: c (float)
Return type:: OptionalEstimatorAccumulator

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

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

acc_to_encoder()[source]

Return type:: OptionalDataEncoder

class OptionalEstimatorAccumulatorFactory(estimator, missing_value=None, keys=None, name=None)[source]

Bases: StatisticAccumulatorFactory

Parameters:

estimator (ParameterEstimator)
missing_value (Any)
keys (str | None)
name (str | None)

make()[source]

Return type:: OptionalEstimatorAccumulator

class OptionalEstimator(estimator, missing_value=None, est_prob=False, pseudo_count=None, name=None, keys=None, prior=None)[source]

Bases: ParameterEstimator

Parameters:

estimator (ParameterEstimator)
missing_value (Any)
est_prob (bool)
pseudo_count (float | None)
name (str | None)
keys (str | None)
prior (tuple[Any, Any] | None)

accumulator_factory()[source]

Return type:: OptionalEstimatorAccumulatorFactory

get_prior()[source]

Return the joint prior as (p_prior, dist_prior) from this estimator and the base estimator.

Return type:: tuple[Any, Any]

set_prior(prior)[source]

Distribute (p_prior, dist_prior) to this estimator’s p prior and the base estimator.

prior=None is a no-op (empirical/pseudo-count path stays byte-identical). The first element is a conjugate Beta prior on p; the second is pushed to the base estimator via set_prior.

Parameters:: prior (tuple[Any, Any] | None)
Return type:: None

model_log_density(model)[source]

Sum the Beta-prior log-density at p and the base estimator’s term (ELBO global term).

Parameters:: model (OptionalDistribution)
Return type:: float

estimate(nobs, suff_stat)[source]

Parameters:

nobs (float | None)
suff_stat (tuple[list[float], SS] | None)

Return type:

OptionalDistribution

class OptionalDataEncoder(encoder, missing_value=None)[source]

Bases: DataSequenceEncoder

Parameters:

encoder (DataSequenceEncoder)
missing_value (Any)

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:: x (Sequence[T])
Return type:: tuple[int, ndarray, ndarray, Any]

OptionalAccumulator: alias of OptionalEstimatorAccumulator

OptionalAccumulatorFactory: alias of OptionalEstimatorAccumulatorFactory

class OptionalFisherView(dist)[source]

Bases: EmpiricalMetricFixedFisherView

Parameters:: dist (Any)

mean_statistics(stats=None, model=True, **kwargs)[source]

Parameters:

stats (ndarray | None)
model (bool)
kwargs (Any)

Return type:

ndarray

fisher_information(stats=None, diagonal=False, ridge=1.0e-8, **kwargs)[source]

Empirical Fisher approximation from per-observation statistic vectors.

Parameters:

stats (ndarray | None)
diagonal (bool)
ridge (float)
kwargs (Any)

Return type:

ndarray

fisher_vectors(stats=None, metric='diagonal', center=None, fisher=None, ridge=1.0e-8, **kwargs)[source]

Return centered/whitened sufficient-statistic vectors.

metric=’identity’ returns centered statistics, ‘diagonal’ divides by per-coordinate Fisher standard deviations, and ‘full’ applies an empirical full-matrix whitening transform.

Parameters:

stats (ndarray | None)
metric (str)
center (ndarray | None)
fisher (ndarray | None)
ridge (float)
kwargs (Any)

Return type:

ndarray