mixle.stats.sets.integer_bernoulli_edit module¶

Create, estimate, and sample from an integer Bernoulli edit set distribution.

Defines the IntegerBernoulliEditDistribution, IntegerBernoulliEditSampler, IntegerBernoulliEditAccumulatorFactory, IntegerBernoulliEditAccumulator, IntegerBernoulliEditEstimator, and the IntegerBernoulliEditDataEncoder classes for use with mixle.

Data type: Tuple[Sequence[int], Sequence[int]]: An observation x = (x1, x2) is a pair of integer sets (prev set, next set), each a subset of S = {0,1,2,…N-1}.

Assume S = {0,1,2,…N-1} is a set of integers. The Bernoulli edit set distribution considers transitions between two random subsets. That is, let X1 and X2 be a random subsets of unique integers from S, s.t. X1 and X2 have at most N elements.

Consider observed subsets of S x1 and x2. The density is given by

p_mat(x2 | x1) = sum_{k=0}^{N-1} p_mat(k in x2 | k in x1) + p_mat(k in x2 | k not in x1) + p_mat(k not in x2 | k in x1) + p_mat(k not in x2 | k not in x1).

p_mat(x1,x2) = P_init(x1)*p_mat(x2|x1).

Note: In (1) only one of the summation terms in non-zero for a given value of k. In (2), P_init() is a distribution defining probabilities for an integer 0<=k<N being in a set (Generally a BernoulliSetDistribution is a good choice).

class IntegerBernoulliEditDistribution(log_edit_pmat, init_dist=NullDistribution(), name=None, keys=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Bernoulli edit set distribution: each integer independently transitions in/out between two sets.

Parameters:

log_edit_pmat (Sequence[tuple[float, float]] | ndarray)
init_dist (SequenceEncodableProbabilityDistribution | None)
name (str | None)
keys (str | None)

classmethod compute_capabilities()[source]

density(x)[source]

Density of the Bernoulli edit set distribution at observation x.

See log_density() for details.

Parameters:: x (Tuple[Sequence[int], Sequence[int]]) – Observed (prev set, next set) pair of integer sets.
Returns:: Density at observation x.
Return type:: float

log_density(x)[source]

Log-density of the joint observation (x[0], x[1]).

Computes log p(x[1] | x[0]) by summing per-integer edit log-probabilities for kept, added, and removed elements, plus log p(x[0]) under init_dist.

Parameters:: x (Tuple[Sequence[int], Sequence[int]]) – Observed (prev set, next set) pair of integer sets.
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 (prev set, next set) observations from IntegerBernoulliEditDataEncoder.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 integer edit-set observations.

Parameters:

x (tuple[int, ndarray, ndarray, ndarray, tuple[ndarray, ndarray, ndarray], E1 | None])
engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)[source]

Return stacked integer edit-set parameters for shared support and init policy.

Parameters:

dists (Sequence[IntegerBernoulliEditDistribution])
engine (Any)

Return type:

dict[str, Any]

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

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

Parameters:

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

Return type:

Any

classmethod backend_stacked_sufficient_statistics_with_estimator(x, weights, params, engine, estimator)[source]

Return per-component legacy (edit_counts, total_weight, init_stat) statistics.

Parameters:

x (tuple[int, ndarray, ndarray, ndarray, tuple[ndarray, ndarray, ndarray], E1 | None])
weights (Any)
params (dict[str, Any])
engine (Any)
estimator (Any)

Return type:

tuple[Any, …]

sampler(seed=None)[source]

Create an IntegerBernoulliEditSampler object from this distribution.

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

estimator(pseudo_count=None)[source]

Create an IntegerBernoulliEditEstimator with matching num_vals.

Parameters:: pseudo_count (Optional[float]) – Used to re-weight sufficient statistics in estimation.
Returns:: IntegerBernoulliEditEstimator object.
Return type:: IntegerBernoulliEditEstimator

dist_to_encoder()[source]

Returns an IntegerBernoulliEditDataEncoder object for encoding sequences of data.

Return type:: IntegerBernoulliEditDataEncoder

enumerator()[source]

Returns IntegerBernoulliEditEnumerator iterating set-pairs in descending probability order.

Return type:: IntegerBernoulliEditEnumerator

class IntegerBernoulliEditEnumerator(dist)[source]

Bases: DistributionEnumerator

Enumerates finite previous/next integer-set pairs in descending probability order.

Parameters:: dist (IntegerBernoulliEditDistribution)

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

Bases: DistributionSampler

IntegerBernoulliEditSampler object for drawing (prev set, next set) pairs from an IntegerBernoulliEditDistribution instance.

Parameters:

dist (IntegerBernoulliEditDistribution)
seed (int | None)

sample(size=None)[source]

Draw iid (prev set, next set) observations from the distribution.

Parameters:: size (Optional[int]) – Number of pairs to draw. If None, a single pair is returned.
Returns:: A (prev set, next set) tuple of integer lists if size is None, else a list of such tuples.
Return type:: list[tuple[list[int], list[int]]] | tuple[list[int], list[int]]

sample_given(x)[source]

Draw a next set conditioned on the last set in x.

Parameters:: x (Sequence[Sequence[int]]) – History of integer sets; only the last set x[-1] is conditioned on.
Returns:: List of integers sampled for the next set.
Return type:: list[int]

class IntegerBernoulliEditAccumulator(num_vals, init_acc=NullAccumulator(), keys=None)[source]

Bases: SequenceEncodableStatisticAccumulator

IntegerBernoulliEditAccumulator object for accumulating removed/added/kept counts from observed set pairs.

Parameters:

num_vals (int)
init_acc (SequenceEncodableStatisticAccumulator | None)
keys (str | None)

update(x, weight, estimate)[source]

Add weight to the removed/added/kept counts for the observed (prev set, next set) pair.

Parameters:

x (Tuple[Sequence[int], Sequence[int]]) – Observed (prev set, next set) pair of integer sets.
weight (float) – Weight for the observation.
estimate (Optional[IntegerBernoulliEditDistribution]) – Previous estimate passed to the init accumulator.

Return type:

None

initialize(x, weight, rng)[source]

Initialize the accumulator with a weighted observation.

Parameters:

x (Tuple[Sequence[int], Sequence[int]]) – Observed (prev set, next set) pair of integer sets.
weight (float) – Weight for the observation.
rng (RandomState) – Random number generator passed to the init accumulator.

Return type:

None

seq_update(x, weights, estimate)[source]

Vectorized update of sufficient statistics from sequence encoded observations.

Parameters:

x (E) – Sequence encoded (prev set, next set) observations from IntegerBernoulliEditDataEncoder.seq_encode().
weights (np.ndarray) – Weights, one per encoded observation.
estimate (Optional[IntegerBernoulliEditDistribution]) – Previous estimate passed to the init accumulator.

Return type:

None

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

Engine-resident accumulation of removed/added/kept edit counts (numpy or torch).

The three weighted edit-type histograms are reduced on the active engine; the fixed-size count matrix is host bookkeeping. The init child is routed through the engine. Matches seq_update.

Parameters:

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

Return type:

None

seq_initialize(x, weights, rng)[source]

Vectorized initialization of sufficient statistics from sequence encoded observations.

Parameters:

x (E) – Sequence encoded (prev set, next set) observations from IntegerBernoulliEditDataEncoder.seq_encode().
weights (np.ndarray) – Weights, one per encoded observation.
rng (np.random.RandomState) – Random number generator passed to the init accumulator.

Return type:

None

combine(suff_stat)[source]

Merge sufficient statistics of suff_stat into this accumulator.

Parameters:: suff_stat (Tuple[np.ndarray, float, Optional[SS1]]) – Edit counts, total weight, and init suff stats.
Returns:: This IntegerBernoulliEditAccumulator.
Return type:: IntegerBernoulliEditAccumulator

value()[source]

Returns the sufficient statistics: (edit counts, total weight, init suff stats).

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

from_value(x)[source]

Set the sufficient statistics of this accumulator from x.

Parameters:: x (Tuple[np.ndarray, float, Optional[SS1]]) – Edit counts, total weight, and init suff stats.
Returns:: This IntegerBernoulliEditAccumulator.
Return type:: IntegerBernoulliEditAccumulator

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 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 sufficient statistics.
Return type:: None

acc_to_encoder()[source]

Returns an IntegerBernoulliEditDataEncoder object for encoding sequences of data.

Return type:: IntegerBernoulliEditDataEncoder

class IntegerBernoulliEditAccumulatorFactory(num_vals, init_factory=None, keys=None)[source]

Bases: StatisticAccumulatorFactory

IntegerBernoulliEditAccumulatorFactory object for creating IntegerBernoulliEditAccumulator objects.

Parameters:

num_vals (int)
init_factory (StatisticAccumulatorFactory | None)
keys (str | None)

make()[source]

Returns a new IntegerBernoulliEditAccumulator object.

Return type:: IntegerBernoulliEditAccumulator

class IntegerBernoulliEditEstimator(num_vals=MISSING, init_estimator=NullEstimator(), min_prob=1.0e-128, pseudo_count=None, suff_stat=None, name=None, keys=None, num_values=MISSING)[source]

Bases: ParameterEstimator

IntegerBernoulliEditEstimator object for estimating an IntegerBernoulliEditDistribution from aggregated sufficient statistics.

Parameters:

num_vals (int)
init_estimator (ParameterEstimator | None)
min_prob (float)
pseudo_count (float | None)
suff_stat (ndarray | None)
name (str | None)
keys (str | None)
num_values (int)

accumulator_factory()[source]

Returns an IntegerBernoulliEditAccumulatorFactory for creating IntegerBernoulliEditAccumulator objects.

Return type:: IntegerBernoulliEditAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate an IntegerBernoulliEditDistribution from aggregated sufficient statistics.

Parameters:

nobs (Optional[float]) – Unused (kept for protocol consistency).
suff_stat (Tuple[np.ndarray, float, Optional[SS1]]) – Edit counts, total weight, and init suff stats.

Returns:

IntegerBernoulliEditDistribution object.

Return type:

IntegerBernoulliEditDistribution

class IntegerBernoulliEditDataEncoder(init_encoder)[source]

Bases: DataSequenceEncoder

IntegerBernoulliEditDataEncoder object for encoding sequences of iid (prev set, next set) observations.

Parameters:: init_encoder (DataSequenceEncoder)

seq_encode(x)[source]

Encode a sequence of iid (prev set, next set) observations for vectorized calculations.

Return value ‘rv’ is a Tuple of length 6 containing:: rv[0] (int): Number of observed pairs. rv[1] (np.ndarray): Observation index for each flattened edit entry. rv[2] (np.ndarray): Flattened integer values of edited elements. rv[3] (np.ndarray): Edit type per entry: 0 (removed), 1 (added), 2 (kept). rv[4] (Tuple[np.ndarray, np.ndarray, np.ndarray]): Indices of entries with each edit type. rv[5] (Optional[Any]): Sequence encoding of the previous sets from init_encoder.

Parameters:: x (Sequence[Tuple[Sequence[int], Sequence[int]]]) – Sequence of iid (prev set, next set) observations.
Returns:: See ‘rv’ above.
Return type:: tuple[int, ndarray, ndarray, ndarray, tuple[ndarray, ndarray, ndarray], Any | None]