mixle.stats.latent.joint_mixture module¶

Joint mixtures over paired observations.

This module models observations of the form (x1, x2) with separate component families for each side and a learned conditional association between their latent component states.

For components f_i on X1 and g_j on X2, the paired density is:

p(x1, x2) = sum_i w1_i f_i(x1) sum_j tau12_ij g_j(x2)

The reverse conditional table tau21 is stored as well so the fitted object can expose both directions of the paired latent association.

class JointMixtureDistribution(components1, components2, w1, w2, taus12, taus21, keys=(None, None, None), name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Joint mixture distribution over paired observations.

Observations are (x1, x2) tuples. The first tuple element is scored by components1 and the second by components2.

Parameters:

components1 (Sequence[SequenceEncodableProbabilityDistribution])
components2 (Sequence[SequenceEncodableProbabilityDistribution])
w1 (Sequence[float] | np.ndarray)
w2 (Sequence[float] | np.ndarray)
taus12 (list[list[float]] | np.ndarray)
taus21 (list[list[float]] | np.ndarray)
keys (tuple[str | None, str | None, str | None] | None)
name (str | None)

compute_capabilities()[source]: Intersect generated-compute backend support across all child components.

compute_declaration()[source]: Return the generated-compute declaration for the paired latent mixture.

density(x)[source]

Evaluate the density of a joint mixture observation x.

See log_density() for details.

Parameters:: x (Tuple[T0, T1]) – A single (X1, X2) observation.
Returns:: Density evaluated at x.
Return type:: float

log_density(x)[source]

Evaluate the log-density of a joint mixture observation x.

The log-density at x = (x1, x2) is

log(sum_{i=1}^{N} w_i * f_i(x1) * sum_{j=1}^{M} tau12_{ij} * g_j(x2)),

evaluated with a log-sum-exp for numerical stability.

Parameters:: x (Tuple[T0, T1]) – A single (X1, X2) observation.
Returns:: Log-density evaluated at x.
Return type:: float

seq_log_density(x)[source]

Vectorized evaluation of the log-density for an encoded sequence of observations x.

Encoded sequence ‘x’ is a Tuple of length 3 containing:: x[0] (int): Number of observations. x[1] (E0): Encoded sequence of X1 values. x[2] (E1): Encoded sequence of X2 values.

Parameters:: x (tuple[int, E0, E1]) – Encoded sequence of iid joint mixture observations.
Returns:: Log-density evaluated at each observation in the encoded sequence x.
Return type:: ndarray

backend_seq_log_density(x, engine)[source]

Engine-neutral log-density for encoded joint-mixture observations.

Parameters:

x (tuple[int, E0, E1])
engine (Any)

Return type:

Any

to_fisher(**kwargs)[source]: Structural Fisher view for the joint mixture.

density_semantics()[source]: Return exact-or-approximate density semantics joined from child components.

sampler(seed=None)[source]

Return a sampler for iid draws from this distribution.

Parameters:: seed (int | None) – Optional random seed.
Returns:: A configured JointMixtureSampler.
Return type:: JointMixtureSampler

estimator(pseudo_count=None)[source]

Return an estimator initialized from this distribution’s components.

Parameters:: pseudo_count (float | None) – Optional smoothing count for latent-state counts.
Returns:: A JointMixtureEstimator.
Return type:: JointMixtureEstimator

dist_to_encoder()[source]

Return an encoder for paired joint-mixture observations.

Return type:: DataSequenceEncoder

enumerator()[source]

Return an enumerator over pairs in descending probability order.

Return type:: JointMixtureEnumerator

class JointMixtureEnumerator(dist)[source]

Bases: DistributionEnumerator

Enumerates the support of a JointMixtureDistribution in descending probability order.

Parameters:: dist (JointMixtureDistribution)

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

Bases: DistributionSampler

Sampler for paired observations from a joint mixture distribution.

Parameters:

dist (JointMixtureDistribution)
seed (int | None)

sample(size=None)[source]

Draw iid (X1, X2) samples from the joint mixture.

The X1 component state is drawn from w1, X1 is sampled from that component, the X2 component state is drawn from taus12 given the X1 state, and X2 is sampled from the corresponding X2 component.

Parameters:: size (int | None) – Number of iid samples to draw. None returns a scalar pair.
Returns:: A scalar pair when size is None; otherwise a list of pairs.
Return type:: tuple[Any, Any] | Sequence[tuple[Any, Any]]

class JointMixtureEstimatorAccumulator(accumulators1, accumulators2, keys=(None, None, None), name=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for joint-mixture EM sufficient statistics.

Parameters:

accumulators1 (Sequence[SequenceEncodableStatisticAccumulator])
accumulators2 (Sequence[SequenceEncodableStatisticAccumulator])
keys (tuple[str | None, str | None, str | None] | None)
name (str | None)

update(x, weight, estimate)[source]

Update sufficient statistics with a single weighted observation.

Encodes the single observation and delegates to seq_update() so that the scalar and vectorized estimation paths agree.

Parameters:

x (Tuple[T0, T1]) – A single (X1, X2) observation.
weight (float) – Weight for the observation.
estimate (JointMixtureDistribution) – Previous estimate from EM algorithm.

Returns:

None.

Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics with a single weighted observation.

A component state is drawn uniformly at random for each of X1 and X2, and the corresponding component accumulators are initialized.

Parameters:

x (Tuple[T0, T1]) – A single (X1, X2) observation.
weight (float) – Weight for the observation.
rng (RandomState) – Random state used to seed child accumulator initializers.

Return type:

None

seq_initialize(x, weights, rng)[source]

Vectorized initialization of sufficient statistics from an encoded sequence x.

Note: Calls _rng_initialize() to ensure equivalence between seq_initialize() and initialize().

Parameters:

x (Tuple[int, E0, E1]) – Encoded sequence of iid joint mixture observations.
weights (np.ndarray) – Weights for the observations.
rng – Random state used to seed child accumulator initializers.

Return type:

None

seq_update(x, weights, estimate)[source]

Vectorized update of sufficient statistics from an encoded sequence x.

The joint posterior over component pairs (i, j) is computed under the previous estimate, and the marginal posteriors are passed as weights into the component accumulators.

Parameters:

x (Tuple[int, E0, E1]) – Encoded sequence of iid joint mixture observations.
weights (np.ndarray) – Weights for the observations.
estimate (JointMixtureDistribution) – Previous estimate from EM algorithm.

Returns:

None.

Return type:

None

seq_update_engine(x, weights, estimate, engine)[source]: Engine-resident E-step: component scoring and the joint-posterior arithmetic run on the active engine (numpy or torch); the marginal/joint counts and the per-component responsibility weights match the host seq_update.

combine(suff_stat)[source]

Merge aggregated joint-mixture sufficient statistics into this accumulator.

The tuple is interpreted as (x1_counts, x2_counts, joint_counts, x1_child_stats, x2_child_stats).

Parameters:: suff_stat (tuple[ndarray, ndarray, ndarray, tuple[E0, ...], tuple[E1, ...]]) – Aggregated sufficient statistics.
Returns:: This accumulator.
Return type:: JointMixtureEstimatorAccumulator

value()[source]

Return accumulated sufficient statistics.

Return type:: tuple[ndarray, ndarray, ndarray, tuple[Any, …], tuple[Any, …]]

from_value(x)[source]

Replace this accumulator’s sufficient statistics.

Parameters:: x (tuple[ndarray, ndarray, ndarray, tuple[E0, ...], tuple[E1, ...]]) – Aggregated sufficient statistics in value format.
Returns:: This accumulator.
Return type:: JointMixtureEstimatorAccumulator

key_merge(stats_dict)[source]

Merge this accumulator into stats_dict under configured keys.

Merges the count statistics if the weight key is set, and the X1/X2 component sufficient statistics if the corresponding accumulator keys are set.

Parameters:: stats_dict (dict[str, Any]) – Mapping from merge keys to sufficient statistics.
Return type:: None

key_replace(stats_dict)[source]

Replace this accumulator’s sufficient statistics from matching keys.

Parameters:: stats_dict (dict[str, Any]) – Mapping from merge keys to sufficient statistics.
Return type:: None

acc_to_encoder()[source]

Return an encoder compatible with paired joint-mixture observations.

Return type:: DataSequenceEncoder

class JointMixtureEstimatorAccumulatorFactory(factories1, factories2, keys=(None, None, None), name=None)[source]

Bases: StatisticAccumulatorFactory

Factory for joint-mixture EM accumulators.

Parameters:

factories1 (Sequence[StatisticAccumulatorFactory])
factories2 (Sequence[StatisticAccumulatorFactory])
keys (tuple[str | None, str | None, str | None] | None)
name (str | None)

make()[source]

Return a fresh joint-mixture accumulator.

Return type:: JointMixtureEstimatorAccumulator

class JointMixtureEstimator(estimators1, estimators2, suff_stat=None, pseudo_count=None, keys=(None, None, None), name=None)[source]

Bases: ParameterEstimator

Estimator for paired latent mixture distributions.

Parameters:

estimators1 (Sequence[ParameterEstimator])
estimators2 (Sequence[ParameterEstimator])
suff_stat (tuple[np.ndarray, np.ndarray, np.ndarray, tuple[E0, ...], tuple[E1, ...]] | None)
pseudo_count (tuple[float, float, float] | None)
keys (tuple[str | None, str | None, str | None] | None)
name (str | None)

accumulator_factory()[source]

Return an accumulator factory matching this estimator.

Return type:: JointMixtureEstimatorAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate a joint mixture distribution from aggregated sufficient statistics.

The tuple is interpreted as (x1_counts, x2_counts, joint_counts, x1_child_stats, x2_child_stats).

Parameters:

nobs – Weighted number of observations, accepted for the estimator interface.
suff_stat (tuple[ndarray, ndarray, ndarray, tuple[E0, ...], tuple[E1, ...]]) – Aggregated joint-mixture sufficient statistics.

Returns:

A fitted joint mixture distribution.

Return type:

JointMixtureDistribution

class JointMixtureDataEncoder(encoder1, encoder2)[source]

Bases: DataSequenceEncoder

Encode paired observations for vectorized joint-mixture scoring and EM.

Parameters:

encoder1 (DataSequenceEncoder)
encoder2 (DataSequenceEncoder)

seq_encode(x)[source]

Encode a sequence of iid joint mixture observations for vectorized functions.

Return value ‘rv’ is a Tuple containing:: rv[0] (int): Number of observations. rv[1] (E0): Encoded sequence of X1 values. rv[2] (E1): Encoded sequence of X2 values.

Parameters:: x (Sequence[Tuple[T0, T1]]) – Sequence of (X1, X2) observations.
Returns:: See above for details.
Return type:: tuple[int, Any, Any]

JointMixtureAccumulator: alias of JointMixtureEstimatorAccumulator

JointMixtureAccumulatorFactory: alias of JointMixtureEstimatorAccumulatorFactory

class JointMixtureFisherView(dist)[source]

Bases: MixtureFisherView

Complete-data Fisher view for joint mixtures without concrete proxies.

Parameters:: dist (Any)

property num_pairs: int: Number of positive-weight component pairs represented by this Fisher view.

log_density(x)[source]

Evaluate the Fisher-view mixture log-density for one paired observation.

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

structured_statistics(x, estimate=None, weight=1.0)[source]

Return posterior pair weights and child sufficient statistics for one observation.

Parameters:

x (Any)
estimate (Any | None)
weight (float)

Return type:

Any