mixle.stats.latent.hierarchical_mixture module¶

Hierarchical mixtures over shared topic distributions.

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

This model has K outer-mixture components and L shared topic distributions {f_l(theta_l)}_{l=1}^{L}. Each outer component k has its own inner-mixture weights {tau_{k,l}}_{l=1}^{L}.

Sampling proceeds by drawing an outer component k with probability w_k, drawing a sequence length N from P_len when a length distribution is supplied, and then sampling each item from the topic mixture for component k.

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

Outer mixture over sequence mixtures with shared topic distributions.

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]: Declare generated-compute support inherited from topics and length model.

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

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]

Return the equivalent flat mixture of sequence distributions.

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]: Return exact-or-approximate density semantics joined from child models.

sampler(seed=None)[source]

Return a sampler for this hierarchical mixture.

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

estimator(pseudo_count=None)[source]

Create an estimator initialized from this hierarchical mixture.

Parameters:: pseudo_count (Optional[float]) – Prior mass used to smooth mixture and topic weights during estimation.
Returns:: Estimator configured with matching topic and length estimators.
Return type:: HierarchicalMixtureEstimator

dist_to_encoder()[source]

Return a data encoder for iid hierarchical-mixture 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

Sampler for sequences from a hierarchical mixture distribution.

Parameters:

dist (HierarchicalMixtureDistribution)
seed (int | None)

sample(size=None)[source]

Return samples from the underlying mixture sampler.

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

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

Bases: SequenceEncodableStatisticAccumulator

Accumulator for hierarchical-mixture sufficient statistics.

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) – Random state 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) – Random state 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 this accumulator into keyed sufficient statistics.

Merges comp_counts when weight_key is set, and merges topic accumulators when comp_key is set.

Also delegates keyed merging to the topic and length accumulators.

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

key_replace(stats_dict)[source]

Replace this accumulator’s statistics from matching keyed values.

Replaces comp_counts when weight_key is set, and replaces topic accumulators when comp_key is set.

Also delegates keyed replacement to the topic and length accumulators.

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

acc_to_encoder()[source]

Return a data encoder assembled from topic and length accumulators.

Return type:: HierarchicalMixtureDataEncoder

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

Bases: StatisticAccumulatorFactory

Factory for hierarchical-mixture estimator accumulators.

Parameters:

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

make()[source]

Return a new hierarchical-mixture estimator accumulator.

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

Estimate hierarchical-mixture distributions 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]

Return an accumulator factory configured from this estimator.

Return type:: HierarchicalMixtureEstimatorAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate a hierarchical mixture from aggregated sufficient statistics.

suff_stat is a three-item tuple containing:: suff_stat[0] (ndarray[float]): Aggregated component counts with shape (num_mixtures, num_topics). suff_stat[1] (Tuple[SS1,…]): One sufficient-statistic value per topic. suff_stat[2] (Optional[SS2]): Sufficient statistics for the length estimator.

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:

Estimated distribution.

Return type:

HierarchicalMixtureDistribution

class HierarchicalMixtureDataEncoder(topic_encoder, len_encoder)[source]

Bases: DataSequenceEncoder

Encode iid hierarchical-mixture observations for vectorized scoring.

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