mixle.stats.univariate.discrete.integer_uniform_spike module¶

Evaluate, estimate, and sample from a uniform distribution over integers in range [min_val, max_val] with a spike: placed on the integer value k.

Defines the IntegerUniformSpikeDistribution, IntegerUniformSpikeSampler, IntegerUniformSpikeAccumulatorFactory, IntegerUniformSpikeAccumulator, IntegerUniformSpikeEstimator, and the IntegerUniformSpikeDataEncoder classes for use with mixle.

Data type: (int): The IntegerUniformSpikeDistribution with a range [min_val, max_val] = [a,b], and spike placed on integer value k with probability p, is given by

P(x_i = k) = p, P(x_i = x) = (1-p)/(b-a), x in [a,b] {k}, P(x_i = else) = 0.0.

class IntegerUniformSpikeDistribution(k, num_vals, p, min_val=0, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

IntegerUniformSpikeDistribution object: uniform over an integer range with a spike of mass p at k.

Parameters:

k (int)
num_vals (int)
p (float)
min_val (int | None)
name (str | None)

classmethod compute_capabilities()[source]

classmethod compute_declaration()[source]

density(x)[source]

Density of the integer uniform spike distribution at observation x.

See log_density() for details.

Parameters:: x (int) – Integer observation.
Returns:: Density at x.
Return type:: float

log_density(x)[source]

Log-density of the integer uniform spike distribution at observation x.

Returns log(p) if x equals the spike value k, log((1-p)/(num_vals-1)) for any other integer in [min_val, max_val], and -inf outside the range.

Parameters:: x (int) – Integer observation.
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 (np.ndarray) – Numpy array of integer observations.
Returns:: Numpy array of log-density (float) of len(x).
Return type:: ndarray

backend_seq_log_density(x, engine)[source]

Engine-neutral log-density for encoded integer spike observations.

Parameters:

x (ndarray)
engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)[source]

Return stacked integer-uniform-spike parameters for a shared support.

Parameters:

dists (list[IntegerUniformSpikeDistribution])
engine (Any)

Return type:

dict[str, Any]

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

Return an (n, k) matrix of integer-uniform-spike log densities.

Parameters:

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

Return type:

Any

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

Return component-stacked legacy (min_val, count_vec) statistics.

Parameters:

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

Return type:

tuple[Any, Any]

sampler(seed=None)[source]

Create an IntegerUniformSpikeSampler from parameters of this distribution.

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

estimator(pseudo_count=None)[source]

Create an IntegerUniformSpikeEstimator for the current integer range.

Parameters:: pseudo_count (Optional[float]) – Used to inflate sufficient statistics.
Returns:: IntegerUniformSpikeEstimator object.
Return type:: IntegerUniformSpikeEstimator

dist_to_encoder()[source]

Returns an IntegerUniformSpikeDataEncoder for encoding sequences of iid integer observations.

Return type:: IntegerUniformSpikeDataEncoder

enumerator()[source]

Returns an IntegerUniformSpikeEnumerator iterating the support in descending probability order.

Return type:: IntegerUniformSpikeEnumerator

quantized_index(max_bits, bin_width_bits=1.0)[source]

Build a bounded bit-quantized index directly from the finite integer support.

Parameters:

max_bits (float)
bin_width_bits (float)

Return type:

QuantizedEnumerationIndex

quantized_multi_cross_index(others, max_bits, bin_width_bits=1.0)[source]

Build an exact aligned cross-bin view over finite integer spike supports.

Parameters:: bin_width_bits (float)
Return type:: QuantizedCrossIndex

quantized_cross_index(other, max_bits, bin_width_bits=1.0)[source]

Build an exact aligned cross-bin view over two integer spike supports.

Parameters:: bin_width_bits (float)
Return type:: QuantizedCrossIndex

class IntegerUniformSpikeEnumerator(dist)[source]

Bases: DistributionEnumerator

Enumerates the support [min_val, max_val] in descending probability order.

The spike value k is yielded first when p >= (1-p)/(num_vals-1), otherwise last; the remaining values share the same probability and are yielded in ascending integer order. Zero-probability values are skipped.

Parameters:: dist (IntegerUniformSpikeDistribution)

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

Bases: DistributionSampler

IntegerUniformSpikeSampler object for sampling from an IntegerUniformSpikeDistribution.

Parameters:

dist (IntegerUniformSpikeDistribution)
seed (int | None)

dist

Distribution to sample from.

Type:: IntegerUniformSpikeDistribution

rng

Seeded RandomState for sampling.

Type:: RandomState

non_k

Integers of the support excluding the spike value k.

Type:: np.ndarray

sample(size=None)[source]

Draw iid samples from the integer uniform spike distribution.

Parameters:: size (Optional[int]) – Number of iid samples to draw.
Returns:: A single int if size is None, else a numpy array of ints with length size.
Return type:: int | ndarray

class IntegerUniformSpikeAccumulator(min_val, max_val, keys=None, name=None)[source]

Bases: SequenceEncodableStatisticAccumulator

IntegerUniformSpikeAccumulator object for accumulating weighted integer counts over a growing range.

Parameters:

min_val (int | None)
max_val (int | None)
keys (str | None)
name (str | None)

min_val

Smallest integer observed (or configured) so far.

Type:: Optional[int]

max_val

Largest integer observed (or configured) so far.

Type:: Optional[int]

count_vec

Weighted counts for each integer in [min_val, max_val].

Type:: Optional[np.ndarray]

count

Total weighted observation count.

Type:: float

key

Key for merging sufficient statistics across accumulators.

Type:: Optional[str]

name

Name for object instance.

Type:: Optional[str]

update(x, weight, estimate)[source]

Add weight to the count for integer x, growing the count vector if x is out of range.

Parameters:

x (int) – Integer observation.
weight (float) – Weight on the observation.
estimate (Optional[IntegerUniformSpikeDistribution]) – Unused previous estimate.

Return type:

None

initialize(x, weight, rng)[source]

Initialize the accumulator with observation x and weight (delegates to update).

Parameters:

x (int)
weight (float)
rng (RandomState)

Return type:

None

seq_initialize(x, weights, rng)[source]

Vectorized initialization from encoded observations x (delegates to seq_update).

Parameters:

x (tuple[int, ndarray, ndarray])
weights (ndarray)
rng (RandomState)

Return type:

None

seq_update(x, weights, estimate)[source]

Vectorized accumulation of weighted counts from encoded observations x.

Parameters:

x (np.ndarray) – Sequence encoded integer observations.
weights (np.ndarray) – Weights on the observations.
estimate (Optional[IntegerUniformSpikeDistribution]) – Unused previous estimate.

Return type:

None

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

Engine-resident accumulation: the weighted value histogram is reduced on the active engine (numpy or torch); the dynamic support range is host bookkeeping. Matches seq_update.

Parameters:

x (ndarray)
weights (Any)
estimate (IntegerUniformSpikeDistribution | None)
engine (Any)

Return type:

None

combine(suff_stat)[source]

Combine sufficient statistics (min_val, count_vec) with this accumulator, aligning ranges.

Parameters:: suff_stat (Tuple[int, np.ndarray]) – Minimum value and count vector of another accumulator.
Returns:: This IntegerUniformSpikeAccumulator.
Return type:: IntegerUniformSpikeAccumulator

value()[source]

Returns sufficient statistics as a tuple (min_val, count_vec).

Return type:: tuple[int, ndarray]

from_value(x)[source]

Set sufficient statistics from a (min_val, count_vec) tuple.

Parameters:: x (Tuple[int, np.ndarray]) – Minimum value and count vector.
Returns:: This IntegerUniformSpikeAccumulator.
Return type:: IntegerUniformSpikeAccumulator

scale(c)[source]

Scale linear counts while preserving the integer support offset.

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

key_merge(stats_dict)[source]

Merge this accumulator’s sufficient statistics into stats_dict under its key.

Parameters:: stats_dict (dict[str, Any])
Return type:: None

key_replace(stats_dict)[source]

Replace this accumulator’s sufficient statistics from stats_dict under its key.

Parameters:: stats_dict (dict[str, Any])
Return type:: None

acc_to_encoder()[source]

Returns an IntegerUniformSpikeDataEncoder for encoding sequences of iid integer observations.

Return type:: IntegerUniformSpikeDataEncoder

class IntegerUniformSpikeAccumulatorFactory(min_val=None, max_val=None, keys=None, name=None)[source]

Bases: StatisticAccumulatorFactory

IntegerUniformSpikeAccumulatorFactory object for creating IntegerUniformSpikeAccumulator objects.

Parameters:

min_val (Optional[int]) – Smallest integer value in the range, if known.
max_val (Optional[int]) – Largest integer value in the range, if known.
keys (Optional[str]) – Set key for merging sufficient statistics.
name (Optional[str]) – Set name for object instance.

min_val

Smallest integer value in the range, if known.

Type:: Optional[int]

max_val

Largest integer value in the range, if known.

Type:: Optional[int]

keys

Key for merging sufficient statistics.

Type:: Optional[str]

name

Name for object instance.

Type:: Optional[str]

make()[source]

Returns a new IntegerUniformSpikeAccumulator object.

Return type:: IntegerUniformSpikeAccumulator

class IntegerUniformSpikeEstimator(min_val=None, max_val=None, pseudo_count=None, suff_stat=None, name=None, keys=None)[source]

Bases: ParameterEstimator

IntegerUniformSpikeEstimator object for estimating IntegerUniformSpikeDistribution objects from counts.

Parameters:

min_val (int | None)
max_val (int | None)
pseudo_count (float | None)
suff_stat (tuple[int, float | None] | None)
name (str | None)
keys (str | None)

accumulator_factory()[source]

Returns an IntegerUniformSpikeAccumulatorFactory consistent with this estimator.

Return type:: IntegerUniformSpikeAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate an IntegerUniformSpikeDistribution by maximizing the spike location and weight.

The spike location k is chosen to maximize the likelihood of the accumulated counts (with optional pseudo_count regularization from the estimator configuration).

Parameters:

nobs (Optional[float]) – Weighted number of observations.
suff_stat (Tuple[int, np.ndarray]) – Minimum value and count vector.

Returns:

IntegerUniformSpikeDistribution object.

Return type:

IntegerUniformSpikeDistribution

class IntegerUniformSpikeDataEncoder[source]

Bases: DataSequenceEncoder

IntegerUniformSpikeDataEncoder object for encoding sequences of iid integer observations.

seq_encode(x)[source]

Encode a sequence of iid integer observations as a numpy integer array.

Parameters:: x (Union[List[int], np.ndarray]) – Sequence of iid integer observations.
Returns:: Numpy array of ints.
Return type:: ndarray