mixle.stats.latent.hidden_association module¶

Hidden association models over grouped-count set pairs.

An observation is a pair (given_counts, emitted_counts) where each side is a list of (value, count) entries. The first side defines an empirical mixture over latent values, and the second side is generated from a conditional emission model marginalized over that empirical mixture.

For grouped counts x and emissions y, the model scores:

log p(x, y) = log p_given(x) + log p_len(|y|)

sum_y count_y log sum_x empirical_x cond(y | x)

This representation keeps repeated values compact while preserving the same distributional contract as iid samples expanded by count.

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]: Return the generated-compute declaration for hidden association.

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]

Return a sampler for iid grouped-count pairs.

Requires non-null given_dist and len_dist.

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

estimator(pseudo_count=None)[source]

Return an estimator built from the child estimators.

Parameters:: pseudo_count (float | None) – Unused; accepted for protocol consistency.
Returns:: A configured HiddenAssociationEstimator.
Return type:: HiddenAssociationEstimator

dist_to_encoder()[source]

Return an encoder for grouped-count hidden-association observations.

Return type:: HiddenAssociationDataEncoder

class HiddenAssociationEnumerator(dist)[source]

Bases: DistributionEnumerator

Enumerate hidden-association support pairs in best-first order.

Parameters:: dist (HiddenAssociationDistribution)

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

Bases: DistributionSampler

Sampler for grouped-count set pairs from a hidden-association distribution.

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

Accumulator for hidden-association sufficient statistics.

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]

Return an encoder compatible with hidden-association observations.

Return type:: HiddenAssociationDataEncoder

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

Bases: StatisticAccumulatorFactory

Factory for hidden-association accumulators.

Parameters:

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

make()[source]

Return a fresh hidden-association accumulator.

Return type:: HiddenAssociationAccumulator

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

Bases: ParameterEstimator

Estimator for hidden association distributions.

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]

Return an accumulator factory matching this estimator.

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

Encode grouped-count hidden-association 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]]]]