mixle.stats.sets.bernoulli_set module¶

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

Defines the BernoulliSetDistribution, BernoulliSetSampler, BernoulliSetAccumulatorFactory, BernoulliSetAccumulator, BernoulliSetEstimator, BernoulliSetDataEncoder, and the BernoulliSetEnumerator classes for use with mixle.

Data type: Sequence[Any]: An observation is a set (any iterable of distinct hashable values) drawn from a finite support S = {s_1,s_2,….,s_N}. Let x be a random subset of S. Each element s_k is included in x independently with probability

p_k = P(s_k is in x) , k = 1,2,…,N,

so the density of an observed set x is

p(x) = prod_{s_k in x} p_k * prod_{s_k not in x} (1-p_k).

A comment on estimation: Note that probability of an element s_k belonging to the set is 0 if we do not encounter any elements an observation sequence. For this reason, we need not state the support of the state-space in estimation.

class BernoulliSetDistribution(pmap=MISSING, min_prob=1.0e-128, name=None, keys=None, prob_map=MISSING, prior=None, posteriors=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Bernoulli set distribution: each support element is included in an observed set independently.

Parameters:

pmap (dict[Any, float])
min_prob (float)
name (str | None)
keys (str | None)
prob_map (dict[Any, float])
prior (SequenceEncodableProbabilityDistribution | None)
posteriors (dict[Any, tuple[float, float]] | None)

classmethod compute_capabilities()[source]

classmethod compute_declaration()[source]

set_prior(prior, posteriors=None)[source]

Attach a (per-element) Beta prior and precompute conjugate-prior expectations.

With a shared Beta(a, b) prior on each element’s inclusion probability p_k this caches the digamma expectations so that expected_log_density evaluates the variational Bayes term E_q[log p(x | p)] via E[log p_k] = digamma(a_k) - digamma(a_k + b_k) and E[log(1 - p_k)] = digamma(b_k) - digamma(a_k + b_k). When posteriors (element -> (a_k, b_k)) is supplied, those per-element posterior Beta parameters are used; otherwise the shared prior parameters are broadcast over the support in pmap. Any other prior (including None) leaves the distribution a plain point model.

Parameters:

prior (SequenceEncodableProbabilityDistribution | None) – A shared BetaDistribution prior, or None.
posteriors (dict[Any, tuple[float, float]] | None) – Optional per-element posterior Beta parameters carried forward from a conjugate update.

Return type:

None

get_prior()[source]

Return the shared Beta prior (or None).

Return type:: SequenceEncodableProbabilityDistribution | None

get_posteriors()[source]

Return the per-element posterior Beta parameters (or None).

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

expected_log_density(x)[source]

Variational expectation E_q[log p(x | p)] under the per-element Beta prior.

Sums E[log p_k] over elements present in x plus E[log(1 - p_k)] over the remaining support. Falls back to the plug-in log_density(x) when no conjugate prior is attached.

Parameters:: x (Sequence[Any])
Return type:: float

seq_expected_log_density(x)[source]

Vectorized expected_log_density over sequence-encoded observations.

Parameters:: x (tuple[int, ndarray, ndarray, ndarray])
Return type:: ndarray

density(x)[source]

Density of the Bernoulli set distribution at observed set x.

See log_density() for details.

Parameters:: x (Sequence[Any]) – Observed set of distinct elements from the support of pmap.
Returns:: Density at observation x.
Return type:: float

log_density(x)[source]

Log-density of the Bernoulli set distribution at observed set x.

Sums log(p_k / (1-p_k)) over the elements present in x, plus the constant sum_k log(1-p_k). Returns -inf if x is missing a required element (an element with p_k = 1 when min_prob is 0).

Parameters:: x (Sequence[Any]) – Observed set of distinct elements from the support of pmap.
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 (Tuple[int, np.ndarray, np.ndarray, np.ndarray]) – Sequence encoded set observations from BernoulliSetDataEncoder.seq_encode().
Returns:: Numpy array of log-density values, one per encoded observation.
Return type:: ndarray

backend_seq_log_density(x, engine)[source]

Engine-neutral vectorized log-density for encoded object-valued sets.

Parameters:

x (tuple[int, ndarray, ndarray, ndarray])
engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)[source]

Return stacked Bernoulli-set parameters for shared object support.

Parameters:

dists (Sequence[BernoulliSetDistribution])
engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)[source]

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

Parameters:

x (tuple[int, ndarray, ndarray, ndarray])
params (dict[str, Any])
engine (Any)

Return type:

Any

classmethod backend_stacked_sufficient_statistics(x, weights, params, engine)[source]

Return per-component legacy (count_map, total_weight) statistics.

Parameters:

x (tuple[int, ndarray, ndarray, ndarray])
weights (Any)
params (dict[str, Any])
engine (Any)

Return type:

tuple[tuple[dict[Any, float], float], …]

sampler(seed=None)[source]

Create a BernoulliSetSampler object from parameters of BernoulliSetDistribution instance.

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

estimator(pseudo_count=None)[source]

Create a BernoulliSetEstimator, passing pmap as suff_stat if pseudo_count is given.

Parameters:: pseudo_count (Optional[float]) – Used to re-weight the distribution’s pmap in estimation.
Returns:: BernoulliSetEstimator object.
Return type:: BernoulliSetEstimator

dist_to_encoder()[source]

Returns a BernoulliSetDataEncoder object for encoding sequences of data.

Return type:: BernoulliSetDataEncoder

enumerator()[source]

Returns BernoulliSetEnumerator iterating subsets of the support in descending probability order.

Return type:: BernoulliSetEnumerator

class BernoulliSetEnumerator(dist)[source]

Bases: DistributionEnumerator

Enumerates subsets of the pmap support in descending probability order.

Parameters:: dist (BernoulliSetDistribution)

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

Bases: DistributionSampler

BernoulliSetSampler object for drawing random sets from a BernoulliSetDistribution instance.

Parameters:

dist (BernoulliSetDistribution)
seed (int | None)

sample(size=None)[source]

Draw iid set observations from the BernoulliSetDistribution instance.

Parameters:: size (Optional[int]) – Number of sets to draw. If None, a single set is returned.
Returns:: A list of included elements if size is None, else a list of such lists of length size.
Return type:: Sequence[Any] | list[Sequence[Any]]

class BernoulliSetAccumulator(keys=None)[source]

Bases: SequenceEncodableStatisticAccumulator

BernoulliSetAccumulator object for aggregating per-element inclusion counts from observed sets.

Parameters:: keys (str | None)

update(x, weight, estimate)[source]

Add weight to the inclusion count of each element of the observed set x.

Parameters:

x (Sequence[Any]) – Observed set of distinct elements.
weight (float) – Weight for the observation.
estimate (Optional[BernoulliSetDistribution]) – Unused (kept for protocol consistency).

Return type:

None

initialize(x, weight, rng)[source]

Initialize the accumulator with a weighted observation. Calls update().

Parameters:

x (Sequence[Any]) – Observed set of distinct elements.
weight (float) – Weight for the observation.
rng (Optional[RandomState]) – Unused (kept for protocol consistency).

Return type:

None

seq_update(x, weights, estimate)[source]

Vectorized update of sufficient statistics from sequence encoded observations.

Parameters:

x (Tuple[int, np.ndarray, np.ndarray, np.ndarray]) – Sequence encoded set observations from BernoulliSetDataEncoder.seq_encode().
weights (np.ndarray) – Weights, one per encoded observation.
estimate (Optional[BernoulliSetDistribution]) – Unused (kept for protocol consistency).

Return type:

None

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

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

The weighted element histogram is reduced on the active engine; the object-keyed count dict is host bookkeeping. Matches seq_update.

Parameters:

x (tuple[int, ndarray, ndarray, ndarray])
weights (Any)
estimate (BernoulliSetDistribution | None)
engine (Any)

Return type:

None

seq_initialize(x, weights, rng)[source]

Vectorized initialization of sufficient statistics. Calls seq_update().

Parameters:

x (Tuple[int, np.ndarray, np.ndarray, np.ndarray]) – Sequence encoded set observations from BernoulliSetDataEncoder.seq_encode().
weights (np.ndarray) – Weights, one per encoded observation.
rng (np.random.RandomState) – Unused (kept for protocol consistency).

Return type:

None

combine(suff_stat)[source]

Merge sufficient statistics of suff_stat into this accumulator.

Parameters:: suff_stat (Tuple[Dict[Any, float], float]) – Inclusion counts by element and total weight.
Returns:: This BernoulliSetAccumulator.
Return type:: BernoulliSetAccumulator

value()[source]

Returns the sufficient statistics: (inclusion counts by element, total weight).

Return type:: tuple[dict[Any, float], float]

from_value(x)[source]

Set the sufficient statistics of this accumulator from x.

Parameters:: x (Tuple[Dict[Any, float], float]) – Inclusion counts by element and total weight.
Returns:: This BernoulliSetAccumulator.
Return type:: BernoulliSetAccumulator

key_merge(stats_dict)[source]

Merge this accumulator’s statistics into stats_dict under its key, if keyed.

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

key_replace(stats_dict)[source]

Replace this accumulator’s statistics with the keyed statistics in stats_dict, if keyed.

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

acc_to_encoder()[source]

Returns a BernoulliSetDataEncoder object for encoding sequences of data.

Return type:: BernoulliSetDataEncoder

class BernoulliSetAccumulatorFactory(keys=None)[source]

Bases: StatisticAccumulatorFactory

BernoulliSetAccumulatorFactory object for creating BernoulliSetAccumulator objects.

Parameters:: keys (str | None)

make()[source]

Returns a new BernoulliSetAccumulator object.

Return type:: BernoulliSetAccumulator

class BernoulliSetEstimator(min_prob=1.0e-128, pseudo_count=None, suff_stat=None, name=None, keys=None, prior=None)[source]

Bases: ParameterEstimator

BernoulliSetEstimator object for estimating a BernoulliSetDistribution from aggregated sufficient statistics.

Parameters:

min_prob (float)
pseudo_count (float | None)
suff_stat (dict[Any, float] | None)
name (str | None)
keys (str | None)
prior (SequenceEncodableProbabilityDistribution | None)

accumulator_factory()[source]

Returns a BernoulliSetAccumulatorFactory for creating BernoulliSetAccumulator objects.

Return type:: BernoulliSetAccumulatorFactory

model_log_density(model)[source]

Log-density of the model’s per-element probabilities under the shared Beta prior.

This is the global ELBO term: with a shared Beta(a, b) prior the contribution is the sum over support elements of log Beta(p_k; a, b). Returns 0.0 when no conjugate prior is attached.

Parameters:: model (BernoulliSetDistribution)
Return type:: float

estimate(nobs, suff_stat)[source]

Estimate a BernoulliSetDistribution from aggregated sufficient statistics.

Parameters:

nobs (Optional[float]) – Unused (kept for protocol consistency).
suff_stat (Tuple[Dict[Any, float], float]) – Inclusion counts by element and total weight.

Returns:

BernoulliSetDistribution object.

Return type:

BernoulliSetDistribution

class BernoulliSetDataEncoder[source]

Bases: DataSequenceEncoder

BernoulliSetDataEncoder for encoding sequences of iid observations.

seq_encode(x)[source]

Encode a sequence of iid observations for use with vectorized functions.

Return value ‘rv’ is a Tuple of length 4 containing:: rv[0] (int): Number of observed sets. rv[1] (np.ndarray): Numpy array of integer indices for flattened array of values. rv[2] (np.ndarray): Numpy array of unique values. (dtype is object). rv[3] (np.ndarray): Numpy array of val_map (rv[2]) integer indices for flattened array of values.

Parameters:: x (Sequence[Sequence[Any]]) – A sequence of iid Bernoulli set observations.
Returns:: See ‘rv’ above.
Return type:: tuple[int, ndarray, ndarray, ndarray]