mixle.stats.latent.hidden_association module¶

Create, estimate, and sample from a hidden association model.

Defines the HiddenAssociationDistribution, HiddenAssociationSampler, HiddenAssociationAccumulatorFactory, HiddenAssociationAccumulator, HiddenAssociationEstimator, and the HiddenAssociationDataEncoder classes for use with mixle.

Consider a set of value V = {v_1,v_2,…,v_K} with data type T. Let the given density be discrete probability density over the values in V,

P_g(X_i = v_k) = p_g(k), for k = 1,2,….,K

where sum_k p_g(k) = 1.0. Consider M samples from P_g() denoted x = (x_1,x_2,…,x_M). We then introduce the latent variable U, where

p_k(x) = p_mat(U = v_k | x) = (# of x_1,…,x_M that are = to v_k) / M, for k = 1,2,…,K.

We then draw N a positive integer N from distribution P_len(), then draw N samples from the density above to get z = (z_1, z_2, …., z_N). Last we sample from the conditional distribution defined for P_c(Y = v_k | z_i) to obtain y = (y_1,…,y_N).

The log_density is given by,

log(p_mat(x,y)) = sum_{i=1}^{N} log(sum_{k=1}^{K} p_k(x)*P_c(y_i|v_k)) + log(P_g(x)) + log(P_len(N)).

Note: That in this model we consider grouped-counts. So the given data type is

x: Tuple[List[Tuple[T, float]], List[Tuple[T, float]]] = [x[0], x[1]],

where x[0] = [(value, count)] for the unique values of x_mat = (X_1,X_2,…,X_M) in V, and x[1] = [(value, count)] for the unique values of Y = (Y_1,…,Y_N) in V as well.

class HiddenAssociationDistribution(cond_dist, given_dist=NullDistribution(), len_dist=NullDistribution(), name=None, keys=(None, None))[source]

Bases: SequenceEncodableProbabilityDistribution

Hidden association model: values of a second set are emitted conditionally on values drawn from a first set.

Parameters:

cond_dist (ConditionalDistribution)
given_dist (SequenceEncodableProbabilityDistribution | None)
len_dist (SequenceEncodableProbabilityDistribution | None)
name (str | None)
keys (tuple[str | None, str | None] | None)

compute_capabilities()[source]: Return backend capability metadata for this concrete hidden association model.

compute_declaration()[source]

log_density(x)[source]

Log-density of the hidden association model at observation x.

For each emitted value in x[1], marginalizes the conditional emission density over the given values in x[0] weighted by their counts, then adds the log-density of the given set under given_dist and of the total emission count under len_dist.

Parameters:: x (Tuple[List[Tuple[T, float]], List[Tuple[T, float]]]) – Grouped-count observation ([(given value, count)], [(emitted value, count)]).
Returns:: Log-density at observation x.
Return type:: float

seq_log_density(x)[source]

Evaluation of log-density at sequence encoded input x (loops over log_density).

Parameters:: x (List[Tuple[List[Tuple[T, float]], List[Tuple[T, float]]]]) – Sequence encoded observations from HiddenAssociationDataEncoder.seq_encode() (the observations themselves).
Returns:: Numpy array of log-density values, one per observation.
Return type:: ndarray

backend_seq_log_density(x, engine)[source]

Evaluate encoded log-densities through distribution-owned backend composition.

Parameters:

x (Sequence[tuple[list[tuple[T, float]], list[tuple[T, float]]]])
engine (Any)

Return type:

Any

emission_mixture(s1)[source]

The per-emission distribution q(.|S1) as a mixture, or None for an empty/degenerate S1.

q(emitted|S1) = sum_u (c_u/n1) P(emitted|u) is a finite mixture of the conditional emission distributions cond_dist.dmap[u] weighted by the given-bag’s normalized counts – enumerable whenever those component distributions are. Requires cond_dist to be a ConditionalDistribution (so the per-given components are available).

Parameters:: s1 (list[tuple[T, float]])
Return type:: MixtureDistribution | None

enumerator()[source]

Enumerate (S1, S2) observations in descending probability order.

The model factors as given_dist(S1) * [prod_e q(e|S1)^{c_e}] * P_len(n): the emitted bag S2 is drawn iid from the per-given mixture q(.|S1) (see emission_mixture()). Enumeration is a conditional product – the outer stream enumerates S1 from given_dist and, for each S1, the inner stream enumerates S2 as a multiset best-first search over q(.|S1)’s own enumeration under len_dist, merged by descending total score with given_dist(S1) as the frontier bound. Requires an enumerable non-null given_dist and a ConditionalDistribution cond_dist.

Return type:: DistributionEnumerator

sampler(seed=None)[source]

Create a HiddenAssociationSampler object from this distribution.

Requires non-null given_dist and len_dist.

Parameters:: seed (Optional[int]) – Used to set seed in random sampler.
Returns:: HiddenAssociationSampler object.
Return type:: HiddenAssociationSampler

estimator(pseudo_count=None)[source]

Create a HiddenAssociationEstimator from the component distributions’ estimators.

Parameters:: pseudo_count (Optional[float]) – Unused (kept for protocol consistency).
Returns:: HiddenAssociationEstimator object.
Return type:: HiddenAssociationEstimator

dist_to_encoder()[source]

Returns a HiddenAssociationDataEncoder object for encoding sequences of data.

Return type:: HiddenAssociationDataEncoder

class HiddenAssociationEnumerator(dist)[source]

Bases: DistributionEnumerator

Parameters:: dist (HiddenAssociationDistribution)

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

Bases: DistributionSampler

HiddenAssociationSampler object for drawing grouped-count set pairs from a HiddenAssociationDistribution.

Parameters:

dist (HiddenAssociationDistribution)
seed (int | None)

sample(size=None)[source]

Draw iid grouped-count observations from the hidden association model.

Parameters:: size (Optional[int]) – Number of observations to draw. If None, a single observation is returned.
Returns:: A ([(given value, count)], [(emitted value, count)]) tuple if size is None, else a list of such tuples of length size.
Return type:: Sequence[tuple[list[tuple[Any, float]], list[tuple[Any, float]]]] | tuple[list[tuple[Any, float]], list[tuple[Any, float]]]

sample_given(x)[source]

Draw an emitted grouped-count set conditioned on the given set x.

Parameters:: x (List[Tuple[T, float]]) – Given set as (value, count) pairs.
Returns:: List of (emitted value, count) pairs.

class HiddenAssociationAccumulator(cond_acc, given_acc=NullAccumulator(), size_acc=NullAccumulator(), name=None, keys=(None, None))[source]

Bases: SequenceEncodableStatisticAccumulator

HiddenAssociationAccumulator object for accumulating sufficient statistics from observed set pairs.

Parameters:

cond_acc (ConditionalDistributionAccumulator)
given_acc (SequenceEncodableStatisticAccumulator | None)
size_acc (SequenceEncodableStatisticAccumulator | None)
name (str | None)
keys (tuple[str | None, str | None] | None)

update(x, weight, estimate)[source]

Update sufficient statistics with the posterior assignment of emitted values to given values.

Parameters:

x (Tuple[List[Tuple[T, float]], List[Tuple[T, float]]]) – Grouped-count observation ([(given value, count)], [(emitted value, count)]).
weight (float) – Weight for the observation.
estimate (HiddenAssociationDistribution) – Previous estimate used to compute posteriors.

Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics with random (Dirichlet) assignments of emitted to given values.

Parameters:

x (Tuple[List[Tuple[T, float]], List[Tuple[T, float]]]) – Grouped-count observation ([(given value, count)], [(emitted value, count)]).
weight (float) – Weight for the observation.
rng (np.random.RandomState) – Random number generator for the random assignments.

Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from sequence encoded observations (loops over initialize()).

Parameters:

x (Sequence[Tuple[List[Tuple[T, float]], List[Tuple[T, float]]]]) – Sequence encoded observations.
weights (np.ndarray) – Weights, one per observation.
rng (np.random.RandomState) – Random number generator for the random assignments.

Return type:

None

seq_update(x, weights, estimate)[source]

Update sufficient statistics from sequence encoded observations (loops over update()).

Parameters:

x (Sequence[Tuple[List[Tuple[T, float]], List[Tuple[T, float]]]]) – Sequence encoded observations.
weights (np.ndarray) – Weights, one per observation.
estimate (HiddenAssociationDistribution) – Previous estimate used to compute posteriors.

Return type:

None

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

Engine-resident E-step for the hidden association model.

Builds the batched cross product of (given, emitted) pairs across the whole minibatch, scores them through the conditional distribution’s engine kernel, and forms the per-emitted posterior over given values with a segmented softmax (global-max shift + index_add) on the active engine. The engine-computed posterior weights are fed to the conditional accumulator; the given/size accumulators are updated per observation. Mirrors update.

Parameters:

x (Sequence[tuple[list[tuple[T, float]], list[tuple[T, float]]]])
weights (ndarray)
estimate (HiddenAssociationDistribution)
engine (Any)

Return type:

None

combine(suff_stat)[source]

Merge sufficient statistics of suff_stat into this accumulator.

Parameters:: suff_stat (Tuple[SS1, Optional[SS2], Optional[SS3]]) – Conditional, given, and size suff stats.
Returns:: This HiddenAssociationAccumulator.
Return type:: HiddenAssociationAccumulator

value()[source]

Returns the sufficient statistics: (conditional, given, size) accumulator values.

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

from_value(x)[source]

Set the sufficient statistics of this accumulator from x.

Parameters:: x (Tuple[SS1, Optional[SS2], Optional[SS3]]) – Conditional, given, and size suff stats.
Returns:: This HiddenAssociationAccumulator.
Return type:: HiddenAssociationAccumulator

scale(c)[source]

Scale sufficient statistics by delegating to child accumulators.

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

key_merge(stats_dict)[source]

Merge keyed statistics of the conditional, given, and size accumulators into stats_dict.

Parameters:: stats_dict (Dict[str, Any]) – Maps keys to merged sufficient statistics.
Return type:: None

key_replace(stats_dict)[source]

Replace keyed statistics of the conditional, given, and size accumulators with those in stats_dict.

Parameters:: stats_dict (Dict[str, Any]) – Maps keys to merged sufficient statistics.
Return type:: None

acc_to_encoder()[source]

Returns a HiddenAssociationDataEncoder object for encoding sequences of data.

Return type:: HiddenAssociationDataEncoder

class HiddenAssociationAccumulatorFactory(cond_factory, given_factory=NullAccumulatorFactory(), len_factory=NullAccumulatorFactory(), name=None, keys=(None, None))[source]

Bases: StatisticAccumulatorFactory

HiddenAssociationAccumulatorFactory object for creating HiddenAssociationAccumulator objects.

Parameters:

cond_factory (ConditionalDistributionAccumulatorFactory)
given_factory (StatisticAccumulatorFactory | None)
len_factory (StatisticAccumulatorFactory | None)
name (str | None)
keys (tuple[str | None, str | None] | None)

make()[source]

Returns a new HiddenAssociationAccumulator object.

Return type:: HiddenAssociationAccumulator

class HiddenAssociationEstimator(cond_estimator, given_estimator=NullEstimator(), len_estimator=NullEstimator(), pseudo_count=None, name=None, keys=(None, None))[source]

Bases: ParameterEstimator

HiddenAssociationEstimator object for estimating a HiddenAssociationDistribution from aggregated sufficient statistics.

Parameters:

cond_estimator (ConditionalDistributionEstimator)
given_estimator (ParameterEstimator | None)
len_estimator (ParameterEstimator | None)
pseudo_count (float | None)
name (str | None)
keys (tuple[str | None, str | None] | None)

accumulator_factory()[source]

Returns a HiddenAssociationAccumulatorFactory for creating HiddenAssociationAccumulator objects.

Return type:: HiddenAssociationAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate a HiddenAssociationDistribution from aggregated sufficient statistics.

Parameters:

nobs (Optional[float]) – Number of observations, passed to the given and length estimators.
suff_stat (Tuple[SS1, Optional[SS2], Optional[SS3]]) – Conditional, given, and size suff stats.

Returns:

HiddenAssociationDistribution object.

Return type:

HiddenAssociationDistribution

class HiddenAssociationDataEncoder[source]

Bases: DataSequenceEncoder

HiddenAssociationDataEncoder object for encoding sequences of iid grouped-count set pair observations.

seq_encode(x)[source]

Encode a sequence of iid grouped-count observations (identity encoding).

Parameters:: x (Sequence[Tuple[List[Tuple[T, float]], List[Tuple[T, float]]]]) – Sequence of iid ([(given value, count)], [(emitted value, count)]) observations.
Returns:: The observations unchanged (seq_log_density and seq_update loop over them).
Return type:: Sequence[tuple[list[tuple[T, float]], list[tuple[T, float]]]]