mixle.stats.univariate.discrete.categorical module¶

Create, estimate, and sample from a Categorical distribution.

Defines the CategoricalDistribution, CategoricalSampler, CategoricalAccumulatorFactory, CategoricalAccumulator, CategoricalEstimator, and the CategoricalDataEncoder classes for use with mixle.

Data type: Any. The data type is taken as the categorical object and a probability is estimated.

If Data type is int, consider using mixle.stats.univariate.discrete.integer_categorical (IntegerCategoricalDistribution) instead.

Reference: Johnson, Kemp & Kotz, Univariate Discrete Distributions (3rd ed., Wiley, 2005).

class CategoricalFisherView(dist, keys, probs)[source]

Bases: FixedFisherView

Parameters:

dist (Any)
keys (Sequence[Any])
probs (Sequence[float])

class CategoricalDistribution(pmap=MISSING, default_value=0.0, name=None, prob_map=MISSING, prior=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Categorical distribution over hashable labels.

Parameters:

pmap (dict[Any, float])
default_value (float)
name (str | None)
prob_map (dict[Any, float])
prior (SequenceEncodableProbabilityDistribution | None)

classmethod compute_capabilities()[source]

classmethod compute_declaration()[source]

static exp_family_sufficient_statistics(x, engine)[source]

Return a shape-only fallback; category-aware statistics come from ..._from_params.

Parameters:

x (Any)
engine (Any)

Return type:

tuple[Any, …]

static exp_family_sufficient_statistics_from_params(x, params, engine)[source]

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

Categories are ordered canonically by sorted(pmap, key=repr) so the columns line up with exp_family_natural_parameters().

Parameters:

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

Return type:

tuple[Any, …]

static exp_family_natural_parameters(params, engine)[source]

Return the natural parameter eta = log(pmap) over categories in canonical key order.

Parameters:

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

Return type:

tuple[Any, …]

static exp_family_log_partition(params, engine)[source]

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)[source]

Return log h(x) = 0 on the support (a key of pmap) and -inf for off-support labels.

Parameters:

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

Return type:

Any

get_prior()[source]

Return the conjugate parameter prior over the category-probability simplex (or None).

Return type:: SequenceEncodableProbabilityDistribution | None

set_prior(prior)[source]

Attach a parameter prior and precompute conjugate-prior expectations.

With a DictDirichlet(alpha) prior over the category probabilities this caches the variational expected log-probabilities E[log p_k] = digamma(alpha_k) - digamma(sum_k alpha_k) for each key of pmap so that expected_log_density(x) = E[log p_x] - log(1 + default_value). A scalar alpha is treated as a symmetric Dirichlet of dimension len(pmap). Any other prior (including None) leaves the distribution a plain point model.

Parameters:: prior (SequenceEncodableProbabilityDistribution | None)
Return type:: None

expected_log_density(x)[source]

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

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

Parameters:: x (Any)
Return type:: float

seq_expected_log_density(x)[source]

Vectorized expected_log_density over sequence-encoded observations.

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

density(x)[source]

Density evaluation of CategoricalDistribution.

p_mat(x) = p_i, if x in pmap.keys(), else p_mat(x) = default_value.

Parameters:: x (Any) – Evaluate CategoricalDistribution density value at x.
Returns:: float density value at x
Return type:: float

log_density(x)[source]

Log-Density evaluation of CategoricalDistribution.

log(p_mat(x)) = log(p_i), if x in pmap.keys(), else log(p_mat(x)) = log(default_value).

Parameters:: x (Any) – Evaluate CategoricalDistribution density value at x.
Returns:: Log-density of Categorical distribution evaluated at x.
Return type:: float

seq_log_density(x)[source]

Vectorized evaluation of log-density for sequence encoded data.

Input value x must be obtained from a call to CategoricalDataEncoder.seq_encode(data). Returns numpy array of log-density evaluated at all observations contained in encoded data x.

Parameters:: x (tuple[ndarray, ndarray]) – (Tuple[np.ndarray,np.ndarray]): Tuple of numpy indices for unique categories, and numpy array unique objects that index xs maps to.
Returns:: Numpy array of log-density evaluated at all observations contained in encoded data x.
Return type:: ndarray

backend_seq_log_density(x, engine)[source]

Engine-neutral log-density for encoded object categories.

The object-to-index lookup remains Python-side at the encoding boundary; the selected log-probability vector is an engine tensor, so simplex-map parameters can still participate in autograd.

Parameters:

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

Return type:

Any

gradient_fit_state(engine, torch, leaves, recurse, tensor_param)[source]

Return distribution-owned state for autograd fitting.

Parameters:

engine (Any)
torch (Any)
leaves (list[Any])
recurse (Any)
tensor_param (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)[source]

Return stacked categorical probabilities for shared finite supports.

Parameters:

dists (Sequence[CategoricalDistribution])
engine (Any)

Return type:

dict[str, Any]

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

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

Parameters:

x (tuple[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 maps from engine-resident posterior weights.

Parameters:

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

Return type:

tuple[dict[Any, float], …]

support_size()[source]

Number of categories in the support.

Return type:: int

to_fisher(**kwargs)[source]: Return the categorical’s one-hot Fisher view (generic fallback for default-augmented maps).

sampler(seed=None)[source]

Creates CategoricalSampler for sampling from CategoricalDistribution.

Parameters:: seed (Optional[int]) – Seed for setting random number generator used to sample.
Returns:: CategoricalSampler object.
Return type:: CategoricalSampler

estimator(pseudo_count=None)[source]

Creates a CategoricalEstimator for estimating parameters of CategoricalDistribution.

Parameters:: pseudo_count (Optional[float]) – If set, inflates counts for currently set sufficient statistic (pmap).
Returns:: CategoricalEstimator object.
Return type:: CategoricalEstimator

dist_to_encoder()[source]

Creates a CategoricalDataEncoder object for sequence encoding data.

Returns:: CategoricalDataEncoder object.
Return type:: CategoricalDataEncoder

enumerator()[source]

Creates a CategoricalEnumerator iterating the support in descending probability order.

Returns:: CategoricalEnumerator object.
Return type:: CategoricalEnumerator

quantized_index(max_bits, bin_width_bits=1.0)[source]

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

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 for finite categorical maps.

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 for two finite categorical maps.

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

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

Bases: DistributionSampler

Parameters:

dist (CategoricalDistribution)
seed (int | None)

sample(size=None)[source]

Draw size-number of samples from CategoricalSampler object.

If size is not provided, size is assumed = 1. If size > 1, a list is returned.

Parameters:: size (Optional[int]) – Number of samples to be draw. If size is None, size = 1.
Returns:: List of levels if size > 1, else a single sample from levels with prob probs.
Return type:: Any | list[Any]

class CategoricalEnumerator(dist)[source]

Bases: DistributionEnumerator

Parameters:: dist (CategoricalDistribution)

class CategoricalAccumulator(keys=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Parameters:: keys (str | None)

update(x, weight, estimate)[source]

Adds weight to the category_count for category x.

If x is new Category label, a new key in the dict count_map is created and the count is incremented by weight.

Parameters:

x (Any) – Category label.
weight (float) – Weight for the observation x.
estimate (Optional['CategoricalDistribution']) – Kept for consistency with update method in SequenceEncodableStatisticAccumulator.

Returns:

None, updates sufficient_stat of Accumulator, count_map.

Return type:

None

initialize(x, weight, rng)[source]

Initializes the CategoricalAccumulator sufficient statistics one observation at a time.

Note: this is just a call to update, since there is no randomness in initialization.

Parameters:

x (Any) – Category label.
weight (float) – Weight incrementing suff stat count_map counts for the observation x.
rng (Optional[RandomState]) – Kept for consistency with update method in SequenceEncodableStatisticAccumulator.

Returns:

None, initializes sufficient_stat of Accumulator, count_map.

Return type:

None

get_seq_lambda()[source]

seq_update(x, weights, estimate)[source]

Vectorized accumulation of Categorical sufficient statistics from encoded sequence of data.

Requires data as encoded sequence from CategoricalDataEncoder.seq_encode(data).

Parameters:

x (Tuple[np.ndarray,np.ndarray]) – Tuple of numpy indices for unique categories, and numpy array unique objects that index xs maps to.
weights (np.ndarray) – weights for each observation in encoded data set.
estimate (Optional['CategoricalDistribution']) – Kept for consistency with update method in SequenceEncodableStatisticAccumulator.

Returns:

None

Return type:

None

seq_initialize(x, weights, rng)[source]

Vectorized initialization of Categorical sufficient statistics from encoded sequence of data.

Requires data as encoded sequence from CategoricalDataEncoder.seq_encode(data). Note: this is just a call to seq_update, since there is no randomness in initialization.

Parameters:

x (Tuple[np.ndarray,np.ndarray]) – Tuple of numpy indices for unique categories, and numpy array unique objects that index xs maps to.
weights (np.ndarray) – weights for each observation in encoded data set.
rng (Optional[RandomState]) – Kept for consistency with update method in SequenceEncodableStatisticAccumulator.

Returns:

None

Return type:

None

combine(suff_stat)[source]

Combine the sufficient statistics of CategoricalAccumulator with suff_stat.

Parameters:: suff_stat (Dict[Any, float]) – Prior data observations aggregated into dictionary with category levels as keys and counts as values.
Returns:: None, updates the count_map of CategoricalAccumulator.
Return type:: CategoricalAccumulator

value()[source]

Returns sufficient statistic of CategoricalAccumulator.

Sufficient statistic value is a dictionary with category as keys and counts of categories as values.

Returns:: Dict[Any, float] of sufficient statistic.
Return type:: dict[Any, float]

from_value(x)[source]

Set CategoricalAccumulator sufficient statistics and member variables from suff_stat dict defined in value().

Takes sufficient statistic value from dictionary with category as keys and counts of categories as values. Sets count_map to the passed value x.

Parameters:: x (Dict[Any, float]) – Dictionary with category as keys and counts of categories as values
Returns:: CategoricalAccumulator with member variable sufficient statistics set to x.
Return type:: CategoricalAccumulator

key_merge(stats_dict)[source]

Combines the sufficient statistics of CategoricalAccumulators that have the same key value.

Parameters:: stats_dict (Dict[str, Any]) – Dictionary for mapping keys to CategoricalAccumulators.
Returns:: None
Return type:: None

key_replace(stats_dict)[source]

Set CategoricalAccumulator sufficient statistic member variables to the value of stats_dict: accumualator with same stats_dict key as member variable key.

Parameters:: stats_dict (Dict[str, Any]) – Maps member variable key to CategoricalAccumulator with same key.
Returns:: None
Return type:: None

acc_to_encoder()[source]

Creates a CategoricalDataEncoder object for sequence encoding data.

Returns:: CategoricalDataEncoder object.
Return type:: CategoricalDataEncoder

class CategoricalAccumulatorFactory(keys=None)[source]

Bases: StatisticAccumulatorFactory

Parameters:: keys (str | None)

make()[source]

Return a CategoricalAccumulator with keys passed.

Returns:: CategoricalAccumulator
Return type:: CategoricalAccumulator

class CategoricalEstimator(pseudo_count=None, suff_stat=None, default_value=False, name=None, keys=None, prior=None)[source]

Bases: ParameterEstimator

Parameters:

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

get_prior()[source]

Return the conjugate parameter prior over the category-probability simplex (or None).

Return type:: SequenceEncodableProbabilityDistribution | None

set_prior(prior)[source]

Set the conjugate parameter prior over the category-probability simplex.

Parameters:: prior (SequenceEncodableProbabilityDistribution | None)
Return type:: None

model_log_density(model)[source]

Log-density of the model probability map under the DictDirichlet prior (ELBO global term).

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

accumulator_factory()[source]

Create CategoricalAccumulatorFactory with keys passed is set.

Returns:: CategoricalAccumulatorFactory
Return type:: CategoricalAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate a CategoricalDistribution from suff_stat value.

If default_value is True, we estimate a default value from the suff_stat counts. Else, it is set to 0.0.

pseudo_count is used to averaged over the number of levels and added to the corresponding counts.

If suff_stat member value is None, estimate for CategoricalDistribution is formed from the suff_stat passed. Otherwise, the suff_stat member value is combined with the suff_stat values passed to estimate.

Parameters:

nobs (Optional[float]) – Not used. Kept for consistency with ParameterEstimator.estimate.
suff_stat (Dict[Any, float]) – Dict with categories as keys and counts as values from accumulated data.

Returns:

CategoricalDistribution estimated from passed in suff_stat value and sufficient statistic member variable: (if it is not None).

Return type:

CategoricalDistribution

class CategoricalDataEncoder[source]

Bases: DataSequenceEncoder

CategoricalDataEncoder for encoding Categorical data for use with vectorized “seq_” functions.

seq_encode(x)[source]

Sequence encode list of categories for use with vectorized “seq_” functions.

Parameters:: x (List[Any]) – List of category labels.
Returns:: Tuple of numpy indicies for unique categories in x, and numpy array unique objects that index xs maps to.
Return type:: tuple[ndarray, ndarray]