mixle.stats.latent.hierarchical_mixture module¶

Create, estimate, and sample from a hierarchical mixture distribution with K components consisting of sequence mixture distribution with L topics shared across all K components.

Defines the HierarchicalMixtureDistribution, HierarchicalMixtureSampler, HierarchicalMixtureEstimatorAccumulatorFactory, HierarchicalMixtureEstimatorAccumulator, HierarchicalMixtureEstimator, and the HierarchicalMixtureDataEncoder classes for use with mixle.

Data type: Sequence[T], where ‘T’ is the type of the topic distributions.

Note that this is a mixture with K ‘outer-mixture’ components consisting of L topic distributions {f_l(theta_l)}_{l=1}^{L}, with ‘inner-mixture’ weights {tau_{k,l}}_{l=1}^{L} for each of the K components.

Sampling proceeds as follows. Each sample is a sequence of length ‘N’ (where can be modeled with a length distribution P_len()) from an outer-mixture component k with probability w_k. Sampling from mixture component ‘k’ consists of sampling from a mixture with topics {f_l(theta_l)}_{l=1}^{L} and ‘inner-mixture’ weights {tau_{k,l}}_{l=1}^{L}.

Example: Let x = (x_1, x_2, x_3, …., x_N) be an observation from a hierarchical mixture distribution of length ‘N’. Let Z and U be a random variables s.t. p_mat(Z=k) = w_k and p_mat(U=l | Z = k) = tau_{k,l}. Then

alpha_i = x_i | Z = k ~ sum_{l=1}^{L} f_l(theta_l)*tau_{k,l}, for i = 1,2,…,N.

Further,

alpha_i | U=l ~ f_l(theta_l), for i = 1,2,3,…,N.

class HierarchicalMixtureDistribution(topics, mixture_weights, topic_weights, len_dist=NullDistribution(), name=None, keys=(None, None))[source]

Bases: SequenceEncodableProbabilityDistribution

HierarchicalMixtureDistribution object defining an outer mixture over sequence mixtures with shared topics.

Data type: Sequence[T], where T is the data type of the topic distributions.

Parameters:

topics (Sequence[SequenceEncodableProbabilityDistribution])
mixture_weights (list[float] | ndarray)
topic_weights (list[list[float]] | ndarray)
len_dist (SequenceEncodableProbabilityDistribution | None)
name (str | None)
keys (tuple[str | None, str | None] | None)

compute_capabilities()[source]

compute_declaration()[source]

density(x)[source]

Evaluate the density of an observation from hierarchical mixture distribution.

Parameters:: x (Sequence[T]) – A sequence of type data type T’s.
Returns:: Density evaluated at x.
Return type:: float

log_density(x)[source]

Evaluate the log density of an observation from hierarchical mixture distribution.

Note: Observation is a sequence.

Parameters:: x (Sequence[T]) – A sequence of type data type T’s.
Returns:: Log-density evaluated at x.
Return type:: float

posterior(x)[source]

Compute the posterior over the mixture components for the outer-mixture at observed value x.

Parameters:: x (Sequence[T]) – An observed sequence of data type T.
Returns:: Numpy array of length ‘num_mixtures’.
Return type:: ndarray

component_log_density(x)[source]

Evaluate the component-wise log-density for an observation from a hierarchical mixture model.

Parameters:: x (Sequence[T]) – An observation from a hierarchical mixture model.
Returns:: Numpy array length of ‘num_mixtures’.
Return type:: ndarray

to_mixture()[source]

Returns a MixtureDistribution object created from object instance.

Return type:: MixtureDistribution

seq_component_log_density(x)[source]

Vectorized evaluation of the outer-mixture component-wise log-density for an encoded sequence x.

This returns a numpy array with shape (rv[0], ‘num_mixtures’).

Note: This density is a Mixture of Sequence of Mixture, so the data must be bin-counted as last step in code.

Encoded sequence ‘x’ is a Tuple of length 5 containing:: x[0] (int): Number of independent observations. x[1] (ndarray[int]): Observation sequence index for each value in flattened x. x[2] (ndarray[int]): Length of each observation in x. x[3] (E): Encoded sequence of flattened observed values (has type E). x[4] (Optional[E2]): Encoded sequence of lengths (has type E2).

Parameters:

x (tuple[int, ndarray, ndarray, E1, E2 | None]) – Encoded sequence of iid hierarchical mixture model observations.

Returns:

Numpy array of dimensions ‘rv[0]’ by ‘num_mixtures’, containing the log-density for each component of the: outer mixture.

Return type:

ndarray

seq_log_density(x)[source]

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

Encoded sequence ‘x’ is a Tuple of length 5 containing:: x[0] (int): Number of independent observations. x[1] (ndarray[int]): Observation sequence index for each value in flattened x. x[2] (ndarray[int]): Length of each observation in x. x[3] (E): Encoded sequence of flattened observed values (has type E). x[4] (Optional[E2]): Encoded sequence of lengths (has type E2).

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

backend_seq_component_log_density(x, engine)[source]

Engine-neutral outer-component log densities for hierarchical-mixture encoded sequences.

Parameters:

x (tuple[int, ndarray, ndarray, E1, E2 | None])
engine (Any)

Return type:

Any

backend_seq_log_density(x, engine)[source]

Engine-neutral hierarchical-mixture log-density for encoded sequences.

Parameters:

x (tuple[int, ndarray, ndarray, E1, E2 | None])
engine (Any)

Return type:

Any

seq_posterior(x)[source]

Vectorized evaluation of the posterior over each outer-mixture component for an encoded sequence x.

Encoded sequence ‘x’ is a Tuple of length 5 containing:: x[0] (int): Number of independent observations. x[1] (ndarray[int]): Observation sequence index for each value in flattened x. x[2] (ndarray[int]): Length of each observation in x. x[3] (E): Encoded sequence of flattened observed values (has type E). x[4] (Optional[E2]): Encoded sequence of lengths (has type E2).

Parameters:: x (tuple[int, ndarray, ndarray, E1, E2 | None]) – See above for details.
Returns:: Numpy array of dimension (x[0], ‘num_mixtures’).
Return type:: ndarray

to_fisher(**kwargs)[source]: Reuse the equivalent flat mixture’s Fisher view.

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]

Return HierarchicalMixtureSampler object created from attribute variables.

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

estimator(pseudo_count=None)[source]

Create an HierarchicalMixtureEstimator object from attributes variables.

Parameters:: pseudo_count (Optional[float]) – Re-weight sufficient statistics in estimation step of EM.
Returns:: HierarchicalMixtureEstimator object.
Return type:: HierarchicalMixtureEstimator

dist_to_encoder()[source]

Return an HierarchicalMixtureDataEncoder object for encoding sequences of iid observations.

Return type:: HierarchicalMixtureDataEncoder

enumerator()[source]

Returns a HierarchicalMixtureEnumerator iterating sequences in descending probability order.

Return type:: HierarchicalMixtureEnumerator

class HierarchicalMixtureEnumerator(dist)[source]

Bases: DistributionEnumerator

Enumerates the support of a HierarchicalMixtureDistribution in descending probability order.

Parameters:: dist (HierarchicalMixtureDistribution)

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

Bases: DistributionSampler

HierarchicalMixtureSampler object for sampling sequences from a HierarchicalMixtureDistribution.

Parameters:

dist (HierarchicalMixtureDistribution)
seed (int | None)

sample(size=None)[source]

Returns samples from MixtureSampler.

Parameters:: size (int | None)
Return type:: Sequence[Any] | Any

class HierarchicalMixtureEstimatorAccumulator(accumulators, num_mixtures, len_accumulator=NullAccumulator(), keys=(None, None))[source]

Bases: SequenceEncodableStatisticAccumulator

HierarchicalMixtureEstimatorAccumulator object for aggregating sufficient statistics of observed data.

Parameters:

accumulators (Sequence[SequenceEncodableStatisticAccumulator])
num_mixtures (int)
len_accumulator (SequenceEncodableStatisticAccumulator | None)
keys (tuple[str | None, str | None] | None)

update(x, weight, estimate)[source]

Update sufficient statistics with an observation x.

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

Parameters:

x (Sequence[T]) – An observation from hierarchical mixture mode with data type T.
weight (float) – Observation weight.
estimate (HierarchicalMixtureDistribution) – Previous estimate from EM algorithm.

Returns:

None.

Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics with an observation x.

Parameters:

x (Sequence[T]) – An observation from hierarchical mixture mode with data type T.
weight (float) – Observation weight.
rng (RandomState) – RandomState object for initializing sufficient statistics.

Returns:

None.

Return type:

None

seq_initialize(x, weights, rng)[source]

Vectorized initialization of sufficient statistics from an encoded sequence of observations in x.

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

Encoded sequence ‘x’ is a Tuple of length 5 containing:: x[0] (int): Number of independent observations. x[1] (ndarray[int]): Observation sequence index for each value in flattened x. x[2] (ndarray[int]): Length of each observation in x. x[3] (E): Encoded sequence of flattened observed values (has type E). x[4] (Optional[E2]): Encoded sequence of lengths (has type E2).

Parameters:

x (tuple[int, ndarray, ndarray, E1, E2 | None]) – Encoded sequence of observations of hierarchical mixture model.
weights (ndarray) – Weights for observations.
rng (RandomState) – RandomState object for initializing sufficient statistics.

Returns:

None.

Return type:

None

seq_update(x, weights, estimate)[source]

Vectorized update of sufficient statistics from an encoded sequence x.

Encoded sequence ‘x’ is a Tuple of length 5 containing:: x[0] (int): Number of independent observations. x[1] (ndarray[int]): Observation sequence index for each value in flattened x. x[2] (ndarray[int]): Length of each observation in x. x[3] (E): Encoded sequence of flattened observed values (has type E). x[4] (Optional[E2]): Encoded sequence of lengths (has type E2).

Parameters:

x (tuple[int, ndarray, ndarray, E1, E2 | None]) – Encoded sequence of observations of hierarchical mixture model.
weights (ndarray) – Weights for observations.
estimate (HierarchicalMixtureDistribution) – Previous estimate from EM algorithm.

Returns:

None.

Return type:

None

seq_update_engine(x, weights, estimate, engine)[source]: Engine-resident E-step: topic scoring and the outer/topic posterior arithmetic run on the active engine (numpy or torch); component counts and per-item topic responsibilities are produced on the engine and fed to the child accumulators. Matches host seq_update.

combine(suff_stat)[source]

Combine the sufficient statistics of ‘suff_stat; with attribute variables.

Arg suff_stat is a Tuple of length 3 containing,: suff_stat[0] (ndarray[float]): Aggregated component counts with shape (num_mixtures, num_topics). suff_stat[1] (Tuple[SS1,…]): Tuple of ‘num_topics’ sufficient statistics for the topics. suff_stat[2] (Optional[SS2]): Optional sufficient statistic for length accumulator.

Parameters:: suff_stat (Tuple[np.ndarray, Tuple[SS1, ...], Optional[SS2]]) – See above for details.
Returns:: HierarchicalMixtureEstimatorAccumulator object.
Return type:: HierarchicalMixtureEstimatorAccumulator

value()[source]

Returns sufficient statistics of type Tuple[np.ndarray, Tuple[SS1,…], Optional[SS2]].

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

from_value(x)[source]

Set the attribute variables for sufficient statistics to arg ‘x’.

Arg ‘x’ is a Tuple of length 3 containing,: x[0] (ndarray[float]): Aggregated component counts with shape (num_mixtures, num_topics). x[1] (Tuple[SS1,…]): Tuple of ‘num_topics’ sufficient statistics for the topics. x[2] (Optional[SS2]): Optional sufficient statistic for length accumulator.

Parameters:: x (Tuple[np.ndarray, Tuple[SS1, ...], Optional[SS2]]) – See above for details.
Returns:: HierarchicalMixtureEstimatorAccumulator object.
Return type:: HierarchicalMixtureEstimatorAccumulator

scale(c)[source]

Scale linear counts and delegate child/length sufficient statistics.

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

key_merge(stats_dict)[source]

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

Merges comp_counts if weight_key is set and found in stats_dict. Merges topic accumulators if ‘comp_key’ is set and found in stats_dict.

Calls key_merge() of the length accumulator.

Parameters:: stats_dict (Dict[str, Any]) – Dictionary mapping keys to corresponding sufficient statistics.
Returns:: None.
Return type:: None

key_replace(stats_dict)[source]

Replace the sufficient statistics of object instance with those containing matching keys in ‘stats_dict’.

Replaces comp_counts if weight_key is set and found in stats_dict. Replaces topic accumulators if ‘comp_key’ is set and found in stats_dict.

Calls key_replace() of the length accumulator.

Parameters:: stats_dict (Dict[str, Any]) – Dictionary mapping keys to corresponding sufficient statistics.
Returns:: None.
Return type:: None

acc_to_encoder()[source]

Return an HierarchicalMixtureDataEncoder object for encoding sequences of iid observations.

Return type:: HierarchicalMixtureDataEncoder

class HierarchicalMixtureEstimatorAccumulatorFactory(factories, num_mixtures, len_factory=NullAccumulatorFactory(), keys=(None, None))[source]

Bases: StatisticAccumulatorFactory

HierarchicalMixtureEstimatorAccumulatorFactory object for creating HierarchicalMixtureEstimatorAccumulator objects.

Parameters:

factories (Sequence[StatisticAccumulatorFactory])
num_mixtures (int)
len_factory (StatisticAccumulatorFactory | None)
keys (tuple[str | None, str | None] | None)

make()[source]

Returns an HierarchicalMixtureEstimatorAccumulator object from attributes variables.

Return type:: HierarchicalMixtureEstimatorAccumulator

class HierarchicalMixtureEstimator(estimators, num_mixtures, len_estimator=NullEstimator(), len_dist=None, suff_stat=None, pseudo_count=None, name=None, keys=(None, None))[source]

Bases: ParameterEstimator

HierarchicalMixtureEstimator object for estimating a HierarchicalMixtureDistribution from sufficient statistics.

Parameters:

estimators (Sequence[ParameterEstimator])
num_mixtures (int)
len_estimator (ParameterEstimator | None)
len_dist (SequenceEncodableProbabilityDistribution | None)
suff_stat (ndarray | None)
pseudo_count (float | None)
name (str | None)
keys (tuple[str | None, str | None] | None)

accumulator_factory()[source]

Create an HierarchicalMixtureEstimatorAccumulator from object instance.

Return type:: HierarchicalMixtureEstimatorAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate HierarchicalMixtureDistribution from aggregated sufficient statistics.

Arg suff_stat is a Tuple of length 3 containing,: suff_stat[0] (ndarray[float]): Aggregated component counts with shape (num_mixtures, num_topics). suff_stat[1] (Tuple[SS1,…]): Tuple of ‘num_topics’ sufficient statistics for the topics. suff_stat[2] (Optional[SS2]): Optional sufficient statistic for length accumulator.

Parameters:

nobs (Optional[float]) – Number of observations used in accumulation of ‘suff_stat’.
suff_stat (tuple[ndarray, SS1, SS2 | None]) – See above for details.

Returns:

HierarchicalMixtureDistribution object.

Return type:

HierarchicalMixtureDistribution

class HierarchicalMixtureDataEncoder(topic_encoder, len_encoder)[source]

Bases: DataSequenceEncoder

HierarchicalMixtureDataEncoder object for encoding sequences of iid hierarchical mixture observations.

Parameters:

topic_encoder (DataSequenceEncoder)
len_encoder (DataSequenceEncoder)

seq_encode(x)[source]

Encode a sequence of iid observations from a hierarchical mixture model.

Returns ‘rv’ as a Tuple of length 5 containing:: rv[0] (int): Number of independent observations. rv[1] (ndarray[int]): Observation sequence index for each value in flattened x. rv[2] (ndarray[int]): Length of each observation in x. rv[3] (E): Encoded sequence of flattened observed values (has type E). rv[4] (Optional[E2]): Encoded sequence of lengths (has type E2).

Parameters:: x (Sequence[Sequence[T]]) – Sequence of hierarchical mixture model observations.
Returns:: See above.
Return type:: tuple[int, ndarray, ndarray, Any, Any | None]

HierarchicalMixtureAccumulator: alias of HierarchicalMixtureEstimatorAccumulator

HierarchicalMixtureAccumulatorFactory: alias of HierarchicalMixtureEstimatorAccumulatorFactory