mixle.stats.latent.integer_hidden_association module

Create, estimate, and sample from an integer hidden association model.

Defines the IntegerHiddenAssociationDistribution, IntegerHiddenAssociationSampler, IntegerHiddenAssociationAccumulatorFactory, IntegerHiddenAssociationAccumulator, IntegerHiddenAssociationEstimator, and the IntegerHiddenAssociationDataEncoder classes for use with mixle.

The k-rank variant of SparseMarkovAssociation.

Data type: Tuple[List[Tuple[int, float]], List[Tuple[int, float]]].

The SparseMarkovAssociation model is a generative model for two sets of words S_1 ={w_{1,1},…,w_{1,n}} and S_2 ={w_{2,1},…,w_{2,m}} over W possible words. The model assumes a hidden set of states H_2 = {h_{2,1},…,h_{2,m}} where h_{2,j} takes on values in {1,2,…,k} and a hidden set of assignments A_2 = {a_{2,1},…,a_{2,m}} where a_{2,j} takes on values in {1,2,…,m}. The observed likelihood function is computed from P(S_1, S_2) = P(S_2 | S_1) P(S_1), where

1. log(P(S_2|S_1)) = sum_{i=1}^{m} log(P(w_{2,i}|w_{1,1},…,w_{1,n}) = sum_{i=1}^{m} log( (1/m)*sum_{j=1}^{n} (1-alpha)*sum_{k=1}^{K}P(w_{2,i} | h_{2,k})*P(h_{2,k}|w_{1,j}) + alpha/W).

2. log(P(S_1)) = sum_{j=1}^{n} log((1-alpha)*P(w_{1,j}) + alpha/W ).

This model is great when the conditional probability matrix is both large and dense. It can also be nested inside other graphical models like a mixture model.

Note: This is the k-rank equivalent of SparseMarkovAssociationModel.

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

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()

Return backend capability metadata for this concrete integer association model.

compute_declaration()

log_density(x)

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)

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)

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)

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()

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)

Create an IntegerHiddenAssociationSampler object from this distribution.

Requires non-null prev_dist and len_dist.

Parameters:

seed (Optional[int]) – Used to set seed in random sampler.

Returns:

IntegerHiddenAssociationSampler object.

Return type:

IntegerHiddenAssociationSampler

estimator(pseudo_count=None)

Create an IntegerHiddenAssociationEstimator with matching dimensions and component estimators.

Parameters:

pseudo_count (Optional[float]) – Unused (kept for protocol consistency).

Returns:

IntegerHiddenAssociationEstimator object.

Return type:

IntegerHiddenAssociationEstimator

dist_to_encoder()

Returns an IntegerHiddenAssociationDataEncoder object for encoding sequences of data.

Return type:

IntegerHiddenAssociationDataEncoder

class IntegerHiddenAssociationEnumerator(dist)

Bases: DistributionEnumerator

Parameters:

dist (IntegerHiddenAssociationDistribution)

class IntegerHiddenAssociationSampler(dist, seed=None)

Bases: DistributionSampler

IntegerHiddenAssociationSampler object for drawing grouped-count word set pairs from an IntegerHiddenAssociationDistribution instance.

Parameters:

• dist (IntegerHiddenAssociationDistribution)

• seed (int | None)

sample_given(x)

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)

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))

Bases: SequenceEncodableStatisticAccumulator

IntegerHiddenAssociationAccumulator object for accumulating 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)

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)

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)

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)

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)

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)

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()

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)

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)

Scale linear association counts and delegate child accumulators.

Parameters:

c (float)

Return type:

IntegerHiddenAssociationAccumulator

key_merge(stats_dict)

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)

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()

Returns an IntegerHiddenAssociationDataEncoder object for encoding sequences of data.

Return type:

DataSequenceEncoder

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

Bases: StatisticAccumulatorFactory

IntegerHiddenAssociationAccumulatorFactory object for creating IntegerHiddenAssociationAccumulator objects.

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()

Returns a new IntegerHiddenAssociationAccumulator object.

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))

Bases: ParameterEstimator

IntegerHiddenAssociationEstimator object for estimating an IntegerHiddenAssociationDistribution from aggregated sufficient statistics.

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()

Returns an IntegerHiddenAssociationAccumulatorFactory for creating accumulator objects.

Return type:

IntegerHiddenAssociationAccumulatorFactory

estimate(nobs, suff_stat)

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)

Bases: DataSequenceEncoder

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

Parameters:

• prev_encoder (DataSequenceEncoder)

• len_encoder (DataSequenceEncoder)

• use_numba (bool)

seq_encode(x)

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)

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)

Numba kernel accumulating posterior weight/state counts in place from flattened encodings.

vec_bincount1(x, w, out)

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)

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.