mixle.stats.univariate.discrete.integer_categorical module

Create, estimate, and sample from a Categorical distribution defined on a range of integers starting a user defined minimum value.

Defines the IntegerCategoricalDistribution, IntegerCategoricalSampler, IntegerCategoricalAccumulatorFactory, IntegerCategoricalAccumulator, IntegerCategoricalEstimator, and the IntegerCategoricalDataEncoder classes for use with mixle.

Data type (int): The integer categorical distribution is defined through summary statistics min_val (int) and vector of probabilities p_vec (np.ndarray[float]) that sum to 1.0. The range of values is given by [min_val, min_val + len(p_vec) - ). The density is then,

P(x_mat=i) = p_vec[i]

for x in {min_val,min_val+1, …, min_val + length(p_vec) - 1}, else 0.0.

class IntegerCategoricalFisherView(dist)

Bases: CategoricalFisherView

Parameters:

dist (Any)

class IntegerCategoricalDistribution(min_val, p_vec=MISSING, name=None, prob_vec=MISSING, prior=None)

Bases: SequenceEncodableProbabilityDistribution

Categorical distribution over a bounded integer range.

Parameters:

• min_val (int)

• p_vec (list[float] | ndarray)

• name (str | None)

• prob_vec (list[float] | ndarray)

• prior (SequenceEncodableProbabilityDistribution | None)

classmethod compute_capabilities()

classmethod compute_declaration()

static exp_family_sufficient_statistics(x, engine)

Return raw values; category-aware one-hot statistics come from ..._from_params.

Parameters:

• x (Any)

• engine (Any)

Return type:

tuple[Any, …]

static exp_family_sufficient_statistics_from_params(x, params, engine)

Return the one-hot category indicator T(x) of shape (n, K) (zeros off support).

Parameters:

• x (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

tuple[Any, …]

static exp_family_natural_parameters(params, engine)

Return the natural parameter eta = log(p_vec) (one entry per category).

Parameters:

• params (dict[str, Any])

• engine (Any)

Return type:

tuple[Any, …]

static exp_family_log_partition(params, engine)

Return the log partition A = 0 (normalization is carried by eta = log p).

Parameters:

• params (dict[str, Any])

• engine (Any)

Return type:

Any

static exp_family_base_measure_from_params(x, params, engine)

Return log h(x) = 0 on the support [min_val, min_val+K) and -inf outside it.

Parameters:

• x (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

Any

get_parameters()

Return the probability vector p_vec (lets it be scored by a Dirichlet conjugate prior).

Return type:

ndarray

get_prior()

Return the conjugate parameter prior over the probability vector (or None).

Return type:

SequenceEncodableProbabilityDistribution | None

set_prior(prior)

Attach a parameter prior and precompute conjugate-prior expectations.

With a Dirichlet(alpha) (or SymmetricDirichlet(alpha)) prior over the probability vector this caches the variational expected log-probabilities E[log p_k] = digamma(alpha_k) - digamma(sum_k alpha_k) so that expected_log_density(x) = E[log p_{x - min_val}] - log(1 + default_value). Any other prior (including None) leaves the distribution a plain point model.

Parameters:

prior (SequenceEncodableProbabilityDistribution | None)

Return type:

None

expected_log_density(x)

Variational expectation E_q[log p(x)] under the (symmetric) Dirichlet prior.

Falls back to the plug-in log_density(x) when no conjugate prior is attached.

Parameters:

x (int)

Return type:

float

seq_expected_log_density(x)

Vectorized expected_log_density over sequence-encoded observations.

Parameters:

x (ndarray)

Return type:

ndarray

density(x)

Evaluate the density of the integer categorical at observation x.

p_mat(x_mat=x) = p_vec[x] if x in support [min_val, max_val], else 0.0.

Parameters:

x (int) – Integer value.

Returns:

Density at x.

Return type:

float

log_density(x)

Evaluate the log-density of the integer categorical at observation x.

log_p(x_mat=x) = log_p_vec[x] if x in support [min_val, max_val], else -np.inf.

Parameters:

x (int) – Integer value.

Returns:

Log-density at x.

Return type:

float

seq_log_density(x)

Vectorized evaluation of IntegerCategorical log_density() for sequence encoded iid observations x.

Parameters:

x (np.ndarray[int]) – Sequence encoded iid observation of integer categorical distribution.

Returns:

Numpy array of floats containing log_density() evaluated at each observation in x.

Return type:

ndarray

static backend_log_density_from_params(x, min_val, log_p_vec, engine)

Engine-neutral integer-categorical log-density from explicit parameters.

Parameters:

• x (Any)

• min_val (int)

• log_p_vec (Any)

• engine (Any)

Return type:

Any

backend_seq_log_density(x, engine)

Engine-neutral vectorized log-density for encoded data.

Parameters:

• x (Any)

• engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)

Return stacked integer-categorical parameters for a homogeneous mixture kernel.

Parameters:

• dists (Sequence[IntegerCategoricalDistribution])

• engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)

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

Parameters:

• x (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

Any

classmethod backend_stacked_sufficient_statistics(x, weights, params, engine)

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

Parameters:

• x (Any)

• weights (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

tuple[Any, Any]

support_size()

Number of integer values in the range.

Return type:

int

to_fisher(**kwargs)

Return the integer-categorical one-hot Fisher view.

sampler(seed=None)

IntegerCategoricalSampler object for sampling from IntegerCategoricalDistribution instance.

Parameters:

seed (Optional[int]) – Set seed for drawing random samples.

Returns:

IntegerCategoricalSampler object.

Return type:

IntegerCategoricalSampler

estimator(pseudo_count=None)

IntegerCategoricalEstimator object from instance of IntegerCategoricalDistribution object.

If pseudo_count is not None, pass min_val and p_vec as sufficient statistics for aggregated estimaton.

Parameters:

pseudo_count (Optional[float]) – Used to re-weight sufficient statistics of IntegerCategoricalDistribution instance in estimation.

Returns:

IntegerCategoricalEstimator object.

Return type:

IntegerCategoricalEstimator

dist_to_encoder()

Return IntegerCategoricalDataEncoder object for encoding sequences of iid integer categorical observations.

Return type:

IntegerCategoricalDataEncoder

enumerator()

Return IntegerCategoricalEnumerator iterating the support in descending probability order.

Return type:

IntegerCategoricalEnumerator

quantized_index(max_bits, bin_width_bits=1.0)

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)

Build an exact aligned cross-bin view over integer categorical ranges.

Parameters:

bin_width_bits (float)

Return type:

QuantizedCrossIndex

quantized_cross_index(other, max_bits, bin_width_bits=1.0)

Build an exact aligned cross-bin view over two integer categorical ranges.

Parameters:

bin_width_bits (float)

Return type:

QuantizedCrossIndex

class IntegerCategoricalEnumerator(dist)

Bases: DistributionEnumerator

Parameters:

dist (IntegerCategoricalDistribution)

class IntegerCategoricalSampler(dist, seed=None)

Bases: DistributionSampler

Parameters:

• dist (IntegerCategoricalDistribution)

• seed (int | None)

sample(size=None)

Draw iid samples from IntegerCategoricalSampler object.

Note: If size is None, a single sample is returned as an integer. If size > 0, a List of integers with length equal to size is returned.

Parameters:

size (Optional[int]) – Number of iid samples to draw.

Returns:

Integer or List[int] of iid samples from IntegerCategoricalSampler instance.

Return type:

int | list[int]

class IntegerCategoricalAccumulator(min_val=None, max_val=None, keys=None)

Bases: SequenceEncodableStatisticAccumulator

Parameters:

• min_val (int | None)

• max_val (int | None)

• keys (str | None)

update(x, weight, estimate)

Update sufficient statistics for IntegerCategoricalAccumulator with one weighted observation.

If min_val and max_val are not set, count_vec is created. If x is larger than max_val of x is less than min_val a new value for max_val/min_val is set, and count_vec is increased to account for the new support range.

Parameters:

• x (int) – Observation from integer categorical distribution.

• weight (float) – Weight for observation.

• estimate (Optional[ntegerCategoricalDistribution]) – Kept for consistency with SequenceEncodableStatisticAccumulator.

Returns:

None.

Return type:

None

initialize(x, weight, rng)

Initialize IntegerCategoricalAccumulator object with weighted observation

Note: Just calls update().

Parameters:

• x (int) – Observation from integer categorical distribution.

• weight (float) – Weight for observation.

• rng (Optional[RandomState]) – Kept for consistency with SequenceEncodableStatisticAccumulator.

Returns:

None.

Return type:

None

seq_initialize(x, weights, rng)

Vectorized initialization of IntegerCategoricalAccumulator sufficient statistics with weighted observations.

Note: Just calls seq_update().

Parameters:

• x (np.ndarray[int]) – Sequence encoded iid observations of integer categorical distribution.

• weights (ndarray) – Numpy array of positive floats.

• rng (Optional[RandomState]) – Kept for consistency with SequenceEncodableStatisticAccumulator.

Returns:

None.

Return type:

None

seq_update(x, weights, estimate)

Vectorized update of IntegerCategoricalAccumulator sufficient statistics with sequence encoded iid

observations x.

Note: Determines the range (support) of integer categorical from the sequence encoded data.

Parameters:

• x (np.ndarray[int]) – Sequence encoded iid observations of integer categorical distribution.

• weights (ndarray) – Numpy array of positive floats.

• estimate (Optional[IntegerCategoricalDistribution]) – Previous estimate of IntegerCategoricalDistribution.

Returns:

None.

Return type:

None

seq_update_engine(x, weights, estimate, engine)

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

• engine (Any)

Return type:

None

combine(suff_stat)

Combine aggregated sufficient statistics with sufficient statistics of IntegerCategoricalAccumulator

instance.

Arg passed suff_stat is sufficient statistics a Tuple of length two containing:

suff_stat[0] (int): Minimum value of the integer categorical, suff_stat[1] (np.ndarray[float]): Numpy array containing probabilities for each integer value. This also

sets the support of integer categorical to have a maximum value of suff_stat[0] + len(suff_stat[0]) - 1.

Member variables min_val, max_val, and count_vec are set from suff_stat arg if count_vec is None. Else, suff_stat is combined with the values of min_val, max_val, and count_vec.

Parameters:

suff_stat (tuple[int | None, ndarray | None]) – See above for details.

Returns:

IntegerCategoricalAccumulator object.

Return type:

IntegerCategoricalAccumulator

value()

Returns member sufficient statistics Tuple[int, np.ndarray[float]] of IntegerCategoricalAccumulator

instance.

Entry 0 of returned value is the minimum value, and entry 1 is the probability weights for each integer value in the support.

Return type:

tuple[int, ndarray]

from_value(x)

Sets IntegerCategoricalAccumulator instance sufficient statistic member variables to x.

Arg passed x is sufficient statistics a Tuple of length two containing:

x[0] (int): Minimum value of the integer categorical, x[1] (np.ndarray[float]): Numpy array containing probabilities for each integer value. This also sets the

support of integer categorical to have a maximum value of x[0] + len(x[0]) - 1.

Parameters:

x (Tuple[int, np.ndarray[float]]) – See above for details.

Returns:

IntegerCategoricalAccumulator object.

Return type:

IntegerCategoricalAccumulator

scale(c)

Scale count vector while preserving integer support metadata.

Parameters:

c (float)

Return type:

IntegerCategoricalAccumulator

key_merge(stats_dict)

Aggregate member sufficient statistics with sufficient statistics of objects with matching keys.

Parameters:

stats_dict (Dict[str, Any]) – Dict mapping keys to corresponding sufficient stats.

Returns:

None.

Return type:

None

key_replace(stats_dict)

Set member sufficient statistics to suff stats with matching keys.

Parameters:

stats_dict (Dict[str, Any]) – Dict mapping keys to corresponding sufficient stats.

Returns:

None.

Return type:

None

acc_to_encoder()

Return IntegerCategoricalDataEncoder object for encoding sequences of iid integer categorical observations.

Return type:

IntegerCategoricalDataEncoder

class IntegerCategoricalAccumulatorFactory(min_val=None, max_val=None, keys=None)

Bases: StatisticAccumulatorFactory

Parameters:

• min_val (int | None)

• max_val (int | None)

• keys (str | None)

make()

Returns IntegerCategoricalAccumulator object with min_val, max_val, and keys passed.

Return type:

IntegerCategoricalAccumulator

class IntegerCategoricalEstimator(min_val=None, max_val=None, pseudo_count=None, suff_stat=None, name=None, keys=None, prior=None)

Bases: ParameterEstimator

Parameters:

• min_val (int | None)

• max_val (int | None)

• pseudo_count (float | None)

• suff_stat (tuple[int, ndarray] | None)

• name (str | None)

• keys (str | None)

• prior (SequenceEncodableProbabilityDistribution | None)

get_prior()

Return the conjugate parameter prior over the probability vector (or None).

Return type:

SequenceEncodableProbabilityDistribution | None

set_prior(prior)

Set the conjugate parameter prior over the probability vector.

Parameters:

prior (SequenceEncodableProbabilityDistribution | None)

Return type:

None

model_log_density(model)

Log-density of the model probability vector under the (symmetric) Dirichlet prior.

Parameters:

model (IntegerCategoricalDistribution)

Return type:

float

accumulator_factory()

Returns IntegerCategoricalAccumulatorFactory object from member sufficient statistics of

IntegerCategoricalEstimator.

Note: If min_val and max_val are BOTH not None, these values are passed to IntegerCategoricalAccumulatorFactory. Else, they are obtained from member variable suff_stat. One of these conditions must be satisfied.

Returns:

Return type:

IntegerCategoricalAccumulatorFactory

estimate(nobs, suff_stat)

Estimate an IntegerCategoricalDistribution object from aggregating sufficient statistics.

Arg ‘suff_stat’ is a Tuple of int and np.ndarray[float],

suff_stat[0] (int): Minimum value of the integer categorical distribution, suff_stat[1] (ndarray[float]): Probabilities for each integer observation in range [suff_stat[0], suff_stat[0] + len(suff_stat[1])-1).

Arg suff_stat is aggregated sufficient statistics obtained from observations of integer categorical data, that is used to estimate the integer categorical distribution. If pseudo_count is not None, the integer categorical is estimated by a combing arg suff_stat and a re-weighted member variable ‘suff_stat’.

Parameters:

• nobs (Optional[float]) – Not used. Kept for consistency with ParameterEstimator.

• suff_stat (tuple[int, ndarray] | None)

Returns:

IntegerCategoricalDistribution object.

Return type:

IntegerCategoricalDistribution

class IntegerCategoricalDataEncoder

Bases: DataSequenceEncoder

IntegerCategoricalDataEncoder object for encoding sequences of iid integer categorical observations.

seq_encode(x)

Sequence encode iid integer categorical observations for “seq_” functions.

Parameters:

x (Union[List[int], np.ndarray]) – Assumed int observations of integer categorical.

Returns:

Numpy array of integers.

Return type:

ndarray