mixle.stats.sets.integer_bernoulli_set module

Create, estimate, and sample from a integer set Bernoulli distribution.

Defines the IntegerBernoulliSetDistribution, IntegerBernoulliSetSampler, IntegerBernoulliSetAccumulatorFactory, IntegerBernoulliSetAccumulator, IntegerBernoulliSetEstimator, and the IntegerBernoulliSetDataEncoder classes for use with mixle.

Let S = {0,1,2,3…,N-1} be a set if integers. Let x_mat be a random subset of S. The Bernoulli set distribution models random subset of S as

p_k = p_mat(k is in x_mat) , k = 0,2,…,N-1.

The density for an observed subset of S, x=(x_1,x_2,..,x_m), for m < N) is given by

p_mat(x) = sum_{k=0}^{K-1}( p_k*(k in x) + (1-p_k)*(k not in x)).

class IntegerBernoulliSetDistribution(log_pvec, log_nvec=None, name=None, keys=None)

Bases: SequenceEncodableProbabilityDistribution

Distribution over finite sets of integer-valued Bernoulli outcomes.

Parameters:

• log_pvec (Sequence[float] | ndarray)

• log_nvec (Sequence[float] | ndarray | None)

• name (str | None)

• keys (str | None)

classmethod compute_capabilities()

classmethod compute_declaration()

density(x)

Return the probability density or mass at a single observation.

Parameters:

x (Sequence[int] | ndarray)

Return type:

float

log_density(x)

Return the log-density or log-mass at a single observation.

Parameters:

x (Sequence[int] | ndarray)

Return type:

float

seq_log_density(x)

Return vectorized log-density values for sequence-encoded observations.

Parameters:

x (tuple[int, ndarray, ndarray])

Return type:

ndarray

backend_seq_log_density(x, engine)

Engine-neutral log-density for encoded integer Bernoulli-set observations.

Parameters:

• x (tuple[int, ndarray, ndarray])

• engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)

Return stacked integer Bernoulli-set parameters for shared support size.

Parameters:

• dists (Sequence[IntegerBernoulliSetDistribution])

• engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)

Return an (n, k) matrix of integer Bernoulli-set log densities.

Parameters:

• x (tuple[int, ndarray, ndarray])

• params (dict[str, Any])

• engine (Any)

Return type:

Any

classmethod backend_stacked_sufficient_statistics(x, weights, params, engine)

Return component-stacked legacy (inclusion_counts, total_weight) statistics.

Parameters:

• x (tuple[int, ndarray, ndarray])

• weights (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

tuple[Any, Any]

sampler(seed=None)

Return a sampler for drawing observations from this distribution.

Parameters:

seed (int | None)

Return type:

IntegerBernoulliSetSampler

estimator(pseudo_count=None)

Return an estimator for fitting this distribution from data.

Parameters:

pseudo_count (float | None)

Return type:

IntegerBernoulliSetEstimator

dist_to_encoder()

Return the data encoder used by this distribution for vectorized methods.

Return type:

IntegerBernoulliSetDataEncoder

enumerator()

Returns IntegerBernoulliSetEnumerator iterating subsets in descending probability order.

Return type:

IntegerBernoulliSetEnumerator

class IntegerBernoulliSetEnumerator(dist)

Bases: DistributionEnumerator

Parameters:

dist (IntegerBernoulliSetDistribution)

class IntegerBernoulliSetSampler(dist, seed=None)

Bases: DistributionSampler

Parameters:

• dist (IntegerBernoulliSetDistribution)

• seed (int | None)

sample(size=None)

Draw observations.

Combinator samplers (mixture/sequence/…) accept batched. With batched=True (the default) each child stream is drawn in one vectorized call instead of a per-draw Python loop – far faster. Because every child sampler owns an independent RandomState, batching consumes each stream in the same order as the loop, so the draws are identical to the legacy path. batched=False forces that legacy per-draw loop as a guaranteed- stable reference. Leaf samplers are already vectorized and ignore the flag.

Parameters:

size (int | None)

Return type:

list[Sequence[int]] | Sequence[int]

class IntegerBernoulliSetAccumulator(num_vals, keys=None)

Bases: SequenceEncodableStatisticAccumulator

Parameters:

• num_vals (int)

• keys (str | None)

update(x, weight, estimate)

Parameters:

• x (Sequence[int] | ndarray)

• weight (float)

• estimate (IntegerBernoulliSetDistribution | None)

Return type:

None

initialize(x, weight, rng)

Parameters:

• x (Sequence[int] | ndarray)

• weight (float)

• rng (RandomState | None)

Return type:

None

seq_update(x, weights, estimate)

Parameters:

• x (tuple[int, ndarray, ndarray])

• weights (ndarray)

• estimate (IntegerBernoulliSetDistribution | None)

Return type:

None

seq_update_engine(x, weights, estimate, engine)

Engine-resident accumulation of per-integer inclusion counts (numpy or torch).

The weighted integer histogram is reduced on the active engine; the fixed-size count vector is host bookkeeping. Matches seq_update.

Parameters:

• x (tuple[int, ndarray, ndarray])

• weights (Any)

• estimate (IntegerBernoulliSetDistribution | None)

• engine (Any)

Return type:

None

seq_initialize(x, weights, rng)

Parameters:

• x (tuple[int, ndarray, ndarray])

• weights (ndarray)

• rng (RandomState | None)

Return type:

None

combine(suff_stat)

Parameters:

suff_stat (tuple[ndarray, float])

Return type:

IntegerBernoulliSetAccumulator

value()

Return type:

tuple[ndarray, float]

from_value(x)

Parameters:

x (tuple[ndarray, float])

Return type:

IntegerBernoulliSetAccumulator

scale(c)

Scale linear sufficient statistics in-place by c.

The structural default is correct for ordinary weighted sums, nested tuples/lists/dicts, and numeric arrays. Families whose value() payload includes non-linear metadata such as support bounds must override this method and leave that metadata unscaled.

Parameters:

c (float)

Return type:

IntegerBernoulliSetAccumulator

key_merge(stats_dict)

Pool this accumulator’s statistics into stats_dict under its merge key.

The structural default implements the common single-key pattern: store the accumulator under self.keys the first time the key is seen, else combine into the one already there. Accumulators with several named keys (e.g. an HMM’s init/trans/state keys) or a non-accumulator stats payload override this. A keys of None (the default) is a no-op.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)

Replace this accumulator’s statistics from the pooled stats_dict entry (see key_merge).

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()

Return type:

IntegerBernoulliSetDataEncoder

class IntegerBernoulliSetAccumulatorFactory(num_vals, keys=None)

Bases: StatisticAccumulatorFactory

Parameters:

• num_vals (int)

• keys (str | None)

make()

Return type:

IntegerBernoulliSetAccumulator

class IntegerBernoulliSetEstimator(num_vals=MISSING, min_prob=1.0e-128, pseudo_count=None, suff_stat=None, name=None, keys=None, num_values=MISSING)

Bases: ParameterEstimator

Parameters:

• num_vals (int)

• min_prob (float)

• pseudo_count (float | None)

• suff_stat (ndarray | None)

• name (str | None)

• keys (str | None)

• num_values (int)

accumulator_factory()

Return type:

IntegerBernoulliSetAccumulatorFactory

estimate(nobs, suff_stat=None)

Parameters:

• nobs (float | None)

• suff_stat (ndarray | None)

Return type:

IntegerBernoulliSetDistribution

class IntegerBernoulliSetDataEncoder

Bases: DataSequenceEncoder

IntegerBernoulliSetDataEncoder object for encoding sequences of iid integer Bernoulli set observations.

seq_encode(x)

Encode sequences of iid observations for vectorized calculations.

Returns ‘rv’:

rv[0] (int): Total number of observations. rv[1] (np.ndarray): Index for flattened values of observations. rv[2] (np.ndarray): Flattened numpy array of integer values.

Parameters:

x (Sequence[Sequence[int]]) – Sequence of integer set observations.

Returns:

See above for details.

Return type:

tuple[int, ndarray, ndarray]