mixle.stats.latent.integer_hidden_association module¶

Integer hidden association models over grouped-count word sets.

An observation is a pair (previous_counts, emitted_counts) where both sides are sparse integer grouped counts. The model is a low-rank hidden association: words in the emitted set are generated through hidden states conditioned on the words in the previous set.

For previous words S1 and emitted words S2, cond_weights models p(state | word_in_S1) and state_prob_mat models p(word_in_S2 | state). The alpha parameter mixes this learned association with a uniform background over emitted values.

The grouped-count representation keeps repeated integer words compact while preserving the same density as an expanded iid sequence with counts.

class IntegerHiddenAssociationDistribution(state_prob_mat, cond_weights, alpha=0.0, prev_dist=NullDistribution(), len_dist=NullDistribution(), name=None, keys=(None, None), use_numba=False)[source]

Bases: SequenceEncodableProbabilityDistribution

Integer hidden association model: words of a second set are emitted through hidden states conditioned on words of a first set.

Parameters:

state_prob_mat (list[list[float]] | ndarray)
cond_weights (list[list[float]] | ndarray)
alpha (float)
prev_dist (SequenceEncodableProbabilityDistribution | None)
len_dist (SequenceEncodableProbabilityDistribution | None)
name (str | None)
keys (tuple[str | None, str | None])
use_numba (bool)

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

compute_declaration()[source]: Return the generated-compute declaration for integer hidden association.

log_density(x)[source]

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

For each emitted word in x[1], marginalizes over the given words in x[0] (weighted by count) and the hidden states, mixing with a uniform density with probability alpha. Adds the log-density of x[0] under prev_dist and of the total emission count under len_dist.

Parameters:: x (Tuple[List[Tuple[int, float]], List[Tuple[int, float]]]) – Grouped-count observation ([(S1 word, count)], [(S2 word, count)]).
Returns:: Log-density at observation x.
Return type:: float

seq_log_density(x)[source]

Vectorized evaluation of log-density at sequence encoded input x.

Parameters:: x (E) – Sequence encoded observations from IntegerHiddenAssociationDataEncoder.seq_encode(). Uses the numba kernel when the encoding was produced with use_numba=True.
Returns:: Numpy array of log-density values, one per encoded observation.
Return type:: ndarray

backend_seq_log_density(x, engine)[source]

Evaluate encoded log-densities using a backend-neutral compute engine.

Parameters:

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

Return type:

Any

conditional_word_log_probs(s1)[source]

Log of the per-emission word distribution q(.|S1) for a given S1 bag, or None if empty.

q(w|S1) = (1-alpha) * sum_u (c_u/n1) * sum_s cond_weights[u,s] * state_prob_mat[s,w] + alpha/W – the smoothed mixture the model uses to score each emitted word. Returns None for an empty S1 (n1 = 0), whose conditional is degenerate (the model’s own density is undefined there).

Parameters:: s1 (list[tuple[int, float]])
Return type:: ndarray | None

enumerator()[source]

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

The model factors as prev_dist(S1) * [prod_w q(w|S1)^{c_w}] * P_len(n2): the emitted bag S2 is a trial-count multinomial whose word distribution q(.|S1) depends on the given bag S1. Enumeration is a conditional product – the outer stream enumerates S1 from prev_dist and, for each S1, the inner stream enumerates S2 by the multinomial bag search under len_dist, merged by descending total score with prev_dist(S1) as the outer frontier bound. Requires an enumerable, non-null prev_dist so the S1 support is defined.

Return type:: DistributionEnumerator

sampler(seed=None)[source]

Return a sampler for iid integer hidden-association observations.

Requires non-null prev_dist and len_dist.

Parameters:: seed (int | None) – Optional random seed.
Returns:: A configured integer hidden-association sampler.
Return type:: IntegerHiddenAssociationSampler

estimator(pseudo_count=None)[source]

Return an estimator with matching dimensions and child estimators.

Parameters:: pseudo_count (float | None) – Unused; accepted for protocol consistency.
Returns:: A configured integer hidden-association estimator.
Return type:: IntegerHiddenAssociationEstimator

dist_to_encoder()[source]

Return an encoder for integer hidden-association observations.

Return type:: IntegerHiddenAssociationDataEncoder

class IntegerHiddenAssociationEnumerator(dist)[source]

Bases: DistributionEnumerator

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

Parameters:: dist (IntegerHiddenAssociationDistribution)

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

Bases: DistributionSampler

Sampler for grouped-count word-set pairs from an integer hidden-association distribution.

Parameters:

dist (IntegerHiddenAssociationDistribution)
seed (int | None)

sample_given(x)[source]

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

Parameters:: x (List[Tuple[int, float]]) – Given word set as (word, count) pairs.
Returns:: List of (emitted word, count) pairs.
Return type:: list[tuple[int, float]]

sample(size=None)[source]

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

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

class IntegerHiddenAssociationAccumulator(num_vals1, num_vals2, num_states, prev_acc=NullAccumulator(), size_acc=NullAccumulator(), use_numba=False, keys=(None, None))[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for hidden-state and emission counts from observed word-set pairs.

Parameters:

num_vals1 (int)
num_vals2 (int)
num_states (int)
prev_acc (SequenceEncodableStatisticAccumulator | None)
size_acc (SequenceEncodableStatisticAccumulator | None)
use_numba (bool)
keys (tuple[str | None, str | None] | None)

update(x, weight, estimate)[source]

Update sufficient statistics with posterior word/state assignments for the observation.

Parameters:

x (Tuple[List[Tuple[int, float]], List[Tuple[int, float]]]) – Grouped-count observation ([(S1 word, count)], [(S2 word, count)]).
weight (float) – Weight for the observation.
estimate (IntegerHiddenAssociationDistribution) – Previous estimate used to compute posteriors.

Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics with random (Dirichlet) state assignments.

Parameters:

x (Tuple[List[Tuple[int, float]], List[Tuple[int, float]]]) – Grouped-count observation ([(S1 word, count)], [(S2 word, 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]

Vectorized initialization of sufficient statistics from sequence encoded observations.

Parameters:

x (E) – Sequence encoded observations from IntegerHiddenAssociationDataEncoder.seq_encode().
weights (np.ndarray) – Weights, one per encoded observation.
rng (np.random.RandomState) – Random number generator for the random assignments.

Return type:

None

seq_update(x, weights, estimate)[source]

Vectorized update of sufficient statistics from sequence encoded observations.

Parameters:

x (E) – Sequence encoded observations from IntegerHiddenAssociationDataEncoder.seq_encode(). Uses the numba kernel when the encoding was produced with use_numba=True.
weights (np.ndarray) – Weights, one per encoded observation.
estimate (IntegerHiddenAssociationDistribution) – Previous estimate used to compute posteriors.

Return type:

None

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

Engine-resident E-step for the pure (non-numba) blocked encoding.

Mirrors the numpy branch of seq_update: for each observation the (given word x state x emitted word) responsibility tensor is built and normalized with the alpha smoothing on the active engine (numpy or torch), and the initial/weight/state counts are scattered into engine-resident accumulators via index_add. Only the per-observation orchestration runs in Python; all tensor arithmetic and accumulation are on the engine.

Parameters:

x (tuple[tuple[list[tuple[ndarray, ...]], E2 | None, E1 | None], None] | tuple[None, tuple[tuple[ndarray, ...], E1 | None, E2 | None]])
weights (ndarray)
estimate (IntegerHiddenAssociationDistribution)
engine (Any)

Return type:

None

combine(suff_stat)[source]

Merge sufficient statistics of suff_stat into this accumulator.

Parameters:: suff_stat (Tuple[np.ndarray, np.ndarray, np.ndarray, Optional[SS1], Optional[SS2]]) – Init counts, weight counts, state counts, prev suff stats, and size suff stats.
Returns:: This IntegerHiddenAssociationAccumulator.
Return type:: IntegerHiddenAssociationAccumulator

value()[source]

Returns the sufficient statistics: (init counts, weight counts, state counts, prev, size).

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

from_value(x)[source]

Set the sufficient statistics of this accumulator from x.

Parameters:: x (Tuple[np.ndarray, np.ndarray, np.ndarray, Optional[SS1], Optional[SS2]]) – Init counts, weight counts, state counts, prev suff stats, and size suff stats.
Returns:: This IntegerHiddenAssociationAccumulator.
Return type:: IntegerHiddenAssociationAccumulator

scale(c)[source]

Scale linear association counts and delegate child accumulators.

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

key_merge(stats_dict)[source]

Merge this accumulator’s weight and state counts into stats_dict under their keys, if keyed.

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

key_replace(stats_dict)[source]

Replace this accumulator’s weight and state counts with the keyed statistics in stats_dict, if keyed.

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

acc_to_encoder()[source]

Return an encoder compatible with integer hidden-association observations.

Return type:: DataSequenceEncoder

class IntegerHiddenAssociationAccumulatorFactory(num_vals1, num_vals2, num_states, prev_factory=NullAccumulatorFactory(), len_factory=NullAccumulatorFactory(), use_numba=False, keys=(None, None))[source]

Bases: StatisticAccumulatorFactory

Factory for integer hidden-association accumulators.

Parameters:

num_vals1 (int)
num_vals2 (int)
num_states (int)
prev_factory (StatisticAccumulatorFactory | None)
len_factory (StatisticAccumulatorFactory | None)
use_numba (bool)
keys (tuple[str | None, str | None])

make()[source]

Return a fresh integer hidden-association accumulator.

Return type:: IntegerHiddenAssociationAccumulator

class IntegerHiddenAssociationEstimator(num_vals, num_states, alpha=0.0, prev_estimator=NullEstimator(), len_estimator=NullEstimator(), suff_stat=None, pseudo_count=None, use_numba=None, name=None, keys=(None, None))[source]

Bases: ParameterEstimator

Estimator for integer hidden-association distributions.

Parameters:

num_vals (list[int] | tuple[int, int] | int)
num_states (int)
alpha (float)
prev_estimator (ParameterEstimator | None)
len_estimator (ParameterEstimator | None)
suff_stat (Any | None)
pseudo_count (float | None)
use_numba (bool | None)
name (str | None)
keys (tuple[str | None, str | None] | None)

accumulator_factory()[source]

Return an accumulator factory matching this estimator.

Return type:: IntegerHiddenAssociationAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate an IntegerHiddenAssociationDistribution from aggregated sufficient statistics.

Parameters:

nobs (Optional[float]) – Number of observations, passed to the prev and length estimators.
suff_stat (Tuple[np.ndarray, np.ndarray, np.ndarray, Optional[SS1], Optional[SS2]]) – Init counts, weight counts, state counts, prev suff stats, and size suff stats.

Returns:

IntegerHiddenAssociationDistribution object.

Return type:

IntegerHiddenAssociationDistribution

class IntegerHiddenAssociationDataEncoder(prev_encoder, len_encoder, use_numba)[source]

Bases: DataSequenceEncoder

Encode grouped-count integer hidden-association observations.

Parameters:

prev_encoder (DataSequenceEncoder)
len_encoder (DataSequenceEncoder)
use_numba (bool)

seq_encode(x)[source]

Sequence encoding for integer hidden association observations.

If numba is not used see _seq_encode(). Else the following is returned a Tuple of the following form is returned None, ((s0, s1, x0, x1, c0, c1, w0), xv, nn) with,

s0 (np.ndarray): Numpy array of lengths for length of x[i][0] s1 (np.ndarray): Numpy array of lengths for length of x[i][1]. x0 (np.ndarray): Flattened numpy array of values from x[i][0]. x1 (np.ndarray): Flattened numpy array of values from x[i][1]. c0 (np.ndarray): Flattened numpy array of counts from x[i][0]. c1 (np.ndarray): Flattened numpy array of counts from x[i][1]. w0 (np.ndarray): Numpy array of sum of counts for each x[i][0]. xv (E1): Sequence encoded flattened values of x[i][0]. nn (E2): Sequence encoded values of lengths (counts).

Parameters:: x (Sequence[tuple[list[tuple[int, float]], list[tuple[int, float]]]]) – Sequence of iid integer hidden association observations.
Returns:: See above.
Return type:: tuple[tuple[list[tuple[ndarray, …]], Any | None, Any | None], None] | tuple[None, tuple[ndarray, …], Any | None, Any | None]

numba_seq_log_density(num_states, max_len1, t0, t1, x0, x1, c0, c1, w0, cond_weights, state_prob_mat, init_prob_vec, a, b, out)[source]: Numba kernel computing per-observation log-densities into out from flattened encodings.

numba_seq_update(num_states, max_len1, t0, t1, x0, x1, c0, c1, w0, cond_weights, state_prob_mat, weight_count, state_count, init_count, weights, a, b)[source]: Numba kernel accumulating posterior weight/state counts in place from flattened encodings.

vec_bincount1(x, w, out)[source]

Numba bincount on the rows of matrix w for groups x.

Parameters:

x (np.ndarray[np.float64]) – Group ids of rows
w (np.ndarray[np.float64]) – N by S numpy array with rows corresponding to x
out (np.ndarray[np.float64]) – Unique values in support of x by S.

Returns:

Numpy 2-d array.

vec_bincount2(x, w, out)[source]

Numba bincount on the rows of matrix w for groups x.

N = len(x) S = number of states. U = unique values in x can take on.

Parameters:

x (np.ndarray[np.float64]) – Group ids of columns of w.
w (np.ndarray[np.float64]) – S by N numpy array with cols corresponding to x
out (np.ndarray[np.float64]) – S by U matrix.

Returns:

Numpy 2-d array.