mixle.stats.latent.joint_mixture module¶

Create, estimate, and sample from a Joint mixture distribution.

Defines the JointMixtureDistribution, JointMixtureSampler, JointMixtureAccumulatorFactory, JointMixtureAccumulator, JointMixtureEstimator, and the JointMixtureDataEncoder classes for use with mixle.

Data type: Tuple[T0, T1].

Consider a random variable X = (X_1, X_2). A joint mixture with N components for X_1, and M components for X_2 is given by

P(X) = sum_{i=1}^{N} w_i * f_i(X_1) * sum_{j=1}^{M} tau_{ij}*g_j(X_2),

where w_i is the probability of sampling X_1 from distribution f_i() (data type T0), tau_{ij} is the probability of sampling X_2 from g_j() (data type T1) given X_1 was sampled from f_i().

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

Bases: SequenceEncodableProbabilityDistribution

JointMixtureDistribution object defining a joint mixture over paired observations.

Data type: Tuple[T0, T1], where T0 and T1 are the data types of the components for X1 and X2.

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]

compute_declaration()[source]

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]

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.

sampler(seed=None)[source]

Create a JointMixtureSampler object for sampling from this distribution.

Parameters:: seed (Optional[int]) – Seed for the random number generator used in sampling.
Returns:: JointMixtureSampler object.
Return type:: JointMixtureSampler

estimator(pseudo_count=None)[source]

Create a JointMixtureEstimator object from the components of this distribution.

Parameters:: pseudo_count (Optional[float]) – If passed, used to re-weight the state counts in estimation.
Returns:: JointMixtureEstimator object.
Return type:: JointMixtureEstimator

dist_to_encoder()[source]

Return a JointMixtureDataEncoder object for encoding sequences of iid observations.

Return type:: DataSequenceEncoder

enumerator()[source]

Returns a JointMixtureEnumerator iterating (X1, X2) 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

JointMixtureSampler object for sampling (X1, X2) pairs from a JointMixtureDistribution.

Parameters:

dist (JointMixtureDistribution)
seed (int | None)

sample(size=None)[source]

Draw one or ‘size’ 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 (Optional[int]) – Number of samples to draw. If None, a single (X1, X2) tuple is returned.
Returns:: A Tuple (X1, X2) if size is None, else a list of ‘size’ such tuples.
Return type:: tuple[Any, Any] | Sequence[tuple[Any, Any]]

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

Bases: SequenceEncodableStatisticAccumulator

JointMixtureEstimatorAccumulator object for aggregating sufficient statistics of observed data.

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) – RandomState object used to seed member RandomState objects.

Returns:

None.

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 (RandomState) – RandomState object used to seed member RandomState objects.

Returns:

None.

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]

Combine the sufficient statistics of suff_stat with this accumulator.

Arg suff_stat is a Tuple of length 5 containing:: suff_stat[0] (np.ndarray): Component counts for the X1 mixture. suff_stat[1] (np.ndarray): Component counts for the X2 mixture. suff_stat[2] (np.ndarray): Joint counts of (X1 state, X2 state) pairs. suff_stat[3] (Tuple[SS0, …]): Sufficient statistics for the X1 components. suff_stat[4] (Tuple[SS1, …]): Sufficient statistics for the X2 components.

Parameters:: suff_stat (tuple[ndarray, ndarray, ndarray, tuple[E0, ...], tuple[E1, ...]]) – See above for details.
Returns:: JointMixtureEstimatorAccumulator object.
Return type:: JointMixtureEstimatorAccumulator

value()[source]

Returns sufficient statistics as a Tuple (see combine() for entry details).

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

from_value(x)[source]

Set the sufficient statistics of this accumulator to x.

Parameters:: x (tuple[ndarray, ndarray, ndarray, tuple[E0, ...], tuple[E1, ...]]) – Sufficient statistic Tuple (see combine() for entry details).
Returns:: JointMixtureEstimatorAccumulator object.
Return type:: JointMixtureEstimatorAccumulator

key_merge(stats_dict)[source]

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

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]) – Dictionary mapping keys to corresponding sufficient statistics.
Returns:: None.
Return type:: None

key_replace(stats_dict)[source]

Replace sufficient statistics of object instance with those of 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()[source]

Return a JointMixtureDataEncoder object for encoding sequences of iid observations.

Return type:: DataSequenceEncoder

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

Bases: StatisticAccumulatorFactory

JointMixtureEstimatorAccumulatorFactory object for creating JointMixtureEstimatorAccumulator objects.

Parameters:

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

make()[source]

Returns a JointMixtureEstimatorAccumulator object from attribute variables.

Return type:: JointMixtureEstimatorAccumulator

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

Bases: ParameterEstimator

JointMixtureEstimator object for estimating a JointMixtureDistribution from sufficient statistics.

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]

Returns a JointMixtureEstimatorAccumulatorFactory object from attribute variables.

Return type:: JointMixtureEstimatorAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate a Joint mixture distribution from aggregated sufficient statistics.

suff_stat is a Tuple containing:: suff_stat[0] (np.ndarray): Component counts for outer mixture. suff_stat[1] (np.ndarray): Component counts for the inner mixture. suff_stat[2] (np.ndarray): Component counts for the comps of inner mix given an outer mix component. suff_stat[3] (Tuple[E0,…]): Suff-stats for outer comps suff_stat[4] (Tuple[E1,…]): Suff-stats for the inner comps.

Parameters:

nobs (Optional[float]) – Weighted number of observations used in aggregation of suff_stats.
suff_stat (tuple[ndarray, ndarray, ndarray, tuple[E0, ...], tuple[E1, ...]]) – See above for details.

Returns:

JointMixtureDistribution object.

Return type:

JointMixtureDistribution

class JointMixtureDataEncoder(encoder1, encoder2)[source]

Bases: DataSequenceEncoder

JointMixtureDataEncoder object for encoding sequences of iid joint mixture observations.

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

log_density(x)[source]

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

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

Structured sufficient stats for one observation.

For latent-variable distributions, this is the posterior-expected complete-data sufficient statistic under estimate (or this view’s distribution when estimate is omitted).

Parameters:

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

Return type:

Any