mixle.stats.bayes.dirichlet module

Create, estimate, and sample from a Dirichlet distribution with concentration parameters alpha.

Defines the DirichletDistribution, DirichletSampler, DirichletAccumulatorFactory, DirichletAccumulator, DirichletEstimator, and the DirichletDataEncoder classes for use with mixle.

Data type: Union[List[float], np.ndarray[float]]

The log-density of a Dirichlet with dim = K, is given by

log(p_mat(x)) = -log(Const) + sum_{k=0}^{K-1} (alpha_k -1)*log(x_k), for sum_k x_k = 1.0,

else -inf. In above,

log(Const) = sum_{k=0}^{K-1} log(Gamma(alpha_k)) - log(Gamma(sum_{k=0}^{K-1} alpha_k)).

dirichlet_param_solve(alpha, mean_log_p, delta, max_iter=_MAX_DIRICHLET_ITERATIONS)

Iteratively solve for alpha of a Dirichlet distribution.

Parameters:

• alpha (np.ndarray) – Numpy array of Dirichlet parameters (all entries should be non-negative).

• mean_log_p (np.ndarray) – Sufficient statistic (1/N) sum_{i=1}^{N} log(x_{i,k}), where N is the number of observations.

• delta (float) – Tolerance for convergence of Newton-Method.

• max_iter (int)

Returns:

Tuple[np.ndarray, int] containing the estimates of alpha and numer of iterations in solver.

Return type:

tuple[ndarray, int]

mpe(x0, f, eps, max_iter=1000)

Minimal polynomial extrapolation for accelerating the fixed-point iteration x_{n+1} = f(x_n).

Parameters:

• x0 (np.ndarray) – Starting point for the fixed-point iteration.

• f (Callable[[np.ndarray], np.ndarray]) – Fixed-point map being iterated.

• eps (float) – Tolerance on the absolute change between extrapolated iterates.

• max_iter (int)

Returns:

Tuple[np.ndarray, int] containing the extrapolated fixed point and the iteration count.

Return type:

tuple[ndarray, int]

alpha_seq_lambda(mean_log_p)

Returns the fixed-point map for the Dirichlet alpha given sufficient statistic mean_log_p.

Parameters:

mean_log_p (np.ndarray) – Mean of the log of the observed proportions.

Returns:

Callable mapping the current alpha to the next alpha iterate.

Return type:

Callable[[ndarray], ndarray]

find_alpha(current_alpha, mlp, thresh)

Solve for the Dirichlet alpha with MPE-accelerated fixed-point iteration.

Parameters:

• current_alpha (np.ndarray) – Initial estimate of the Dirichlet parameters.

• mlp (np.ndarray) – Mean of the log of the observed proportions (sufficient statistic).

• thresh (float) – Convergence tolerance.

Returns:

Tuple[np.ndarray, int] containing the estimate of alpha and the iteration count.

Return type:

tuple[ndarray, int]

class DirichletFisherView(dist)

Bases: FixedFisherView

Parameters:

dist (Any)

class DirichletDistribution(alpha, name=None, keys=None)

Bases: SequenceEncodableProbabilityDistribution

Dirichlet distribution over probability vectors, with concentration parameters alpha.

Parameters:

• alpha (list[float] | ndarray)

• name (str | None)

• keys (str | None)

classmethod compute_capabilities()

classmethod compute_declaration()

get_parameters()

Return the concentration vector alpha.

Lets a DirichletDistribution serve as a conjugate prior (on a Categorical/Mixture weight simplex) under the unified Bayesian estimation protocol.

Return type:

ndarray

cross_entropy(dist)

Cross entropy -E_self[log dist(x)] for a Dirichlet argument.

Accepts another DirichletDistribution (full concentration vector) or a SymmetricDirichletDistribution (scalar concentration broadcast to this distribution’s dimension). Both arise as the conjugate prior/posterior over the same simplex during variational Bayes (e.g. the ELBO global term in DPM).

Parameters:

dist (SequenceEncodableProbabilityDistribution)

Return type:

float

entropy()

Returns the differential entropy in nats.

Return type:

float

density(x)

Evaluate the density of a dirichlet observation.

See log_density() for details.

Parameters:

x (Union[List[float], np.ndarray]) – A single dirichlet observation.

Returns:

Density evaluated at x.

Return type:

float

log_density(x)

Evaluate the log-density of a dirichlet observation.

The log-density of a Dirichlet with dim = K, is given by

log(p_mat(x)) = -log(Const) + sum_{k=0}^{K-1} (alpha_k -1)*log(x_k), for sum_k x_k = 1.0,

else -inf. In above

log(Const) = sum_{k=0}^{K-1} log(Gamma(alpha_k)) - log(Gamma(sum_{k=0}^{K-1} alpha_k)).

Parameters:

x (Union[List[float], np.ndarray]) – A single dirichlet observation.

Returns:

Log-density evaluated at x.

Return type:

float

seq_log_density(x)

Vectorized evaluation of the log-density at a sequence-encoded input x.

Parameters:

x (Tuple[np.ndarray, np.ndarray, np.ndarray]) – Encoded data from DirichletDataEncoder.seq_encode(), a tuple of (log of observations, observations, squared observations).

Returns:

Numpy array containing the log-density of each encoded observation.

Return type:

ndarray

static backend_log_density_from_params(log_x, alpha, log_const, engine)

Engine-neutral Dirichlet log-density from encoded log observations.

Parameters:

• log_x (Any)

• alpha (Any)

• log_const (Any)

• engine (Any)

Return type:

Any

backend_seq_log_density(x, engine)

Engine-neutral vectorized log-density for encoded Dirichlet observations.

Parameters:

• x (tuple[Any, Any, Any])

• engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)

Return stacked parameters for equal-dimensional Dirichlet mixtures.

Parameters:

• dists (Sequence[DirichletDistribution])

• engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)

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

Parameters:

• x (tuple[Any, Any, Any])

• params (dict[str, Any])

• engine (Any)

Return type:

Any

classmethod backend_stacked_sufficient_statistics(x, weights, params, engine)

Return component-stacked legacy Dirichlet sufficient statistics.

Parameters:

• x (tuple[Any, Any, Any])

• weights (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

tuple[Any, Any, Any, Any]

to_fisher(**kwargs)

Return the Dirichlet’s log-coordinate Fisher view (generic fallback for degenerate alpha).

sampler(seed=None)

Create a DirichletSampler for sampling from this distribution.

Parameters:

seed (Optional[int]) – Seed to set for sampling with RandomState.

Returns:

DirichletSampler object.

Return type:

DirichletSampler

estimator(pseudo_count=None)

Create a DirichletEstimator for estimating this distribution.

If pseudo_count is passed, the current normalized alpha is used to regularize the estimate.

Parameters:

pseudo_count (Optional[float]) – Used to inflate sufficient statistics in estimation.

Returns:

DirichletEstimator object.

Return type:

DirichletEstimator

dist_to_encoder()

Create DirichletDataEncoder object for encoding sequences of iid Dirichlet observations.

Return type:

DirichletDataEncoder

class DirichletSampler(dist, seed=None)

Bases: DistributionSampler

DirichletSampler object for sampling from a DirichletDistribution.

Parameters:

• dist (DirichletDistribution)

• seed (int | None)

sample(size=None)

Draw iid samples from the Dirichlet distribution.

Entries with non-positive alpha are fixed at zero and the remaining entries are sampled from the Dirichlet restricted to the valid alpha values.

Parameters:

size (Optional[int]) – Number of iid samples to draw. If None, a single sample is drawn.

Returns:

Numpy array with shape (dim,) if size is None, else with shape (size, dim).

Return type:

ndarray

class DirichletAccumulator(dim=None, keys=None)

Bases: SequenceEncodableStatisticAccumulator

DirichletAccumulator object for aggregating sufficient statistics from iid observations.

Parameters:

• dim (int | None)

• keys (str | None)

update(x, weight, estimate)

Update sufficient statistics with a single weighted observation.

Zero-valued entries of x are excluded from the sum of logs.

Parameters:

• x (Union[np.ndarray, List[float]]) – Length-dim probability vector observation.

• weight (float) – Weight for the observation.

• estimate (Optional[DirichletDistribution]) – Kept for consistency with SequenceEncodableStatisticAccumulator (not used).

Returns:

None.

Return type:

None

initialize(x, weight, estimate)

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

Parameters:

• x (Union[np.ndarray, List[float]]) – Length-dim probability vector observation.

• weight (float) – Weight for the observation.

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

Returns:

None.

Return type:

None

get_seq_lambda()

Returns a list containing the seq_update member function.

seq_update(x, weights, estimate)

Vectorized update of sufficient statistics with an encoded sequence of observations.

Parameters:

• x (Tuple[np.ndarray, np.ndarray, np.ndarray]) – Encoded data from DirichletDataEncoder.seq_encode().

• weights (np.ndarray) – Numpy array of observation weights.

• estimate (Optional[DirichletDistribution]) – Kept for consistency (not used).

Returns:

None.

Return type:

None

seq_update_engine(x, weights, estimate, engine)

Engine-resident accumulation of Dirichlet moment statistics (numpy or torch).

The weighted vector moments (sum of logs, sum, sum of squares) are reduced via a weight-vector / observation-matrix product on the active engine. Matches seq_update.

Parameters:

• x (tuple[ndarray, ndarray, ndarray])

• weights (Any)

• estimate (DirichletDistribution | None)

• engine (Any)

Return type:

None

seq_initialize(x, weights, rng)

Vectorized initialization of the accumulator. Calls seq_update().

Parameters:

• x (Tuple[np.ndarray, np.ndarray, np.ndarray]) – Encoded data from DirichletDataEncoder.seq_encode().

• weights (np.ndarray) – Numpy array of observation weights.

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

Returns:

None.

Return type:

None

combine(suff_stat)

Merge the sufficient statistics of suff_stat into this accumulator.

Parameters:

suff_stat (Tuple[int, np.ndarray, np.ndarray, np.ndarray]) – Tuple of (counts, sum of logs, sum of observations, sum of squared observations).

Returns:

DirichletAccumulator object.

Return type:

DirichletAccumulator

value()

Returns the sufficient statistics (counts, sum of logs, sum, sum of squares) of the accumulator.

Return type:

tuple[int, ndarray, ndarray, ndarray]

from_value(x)

Set the sufficient statistics of the accumulator to x.

Parameters:

x (Tuple[int, np.ndarray, np.ndarray, np.ndarray]) – Tuple of (counts, sum of logs, sum of observations, sum of squared observations).

Returns:

None.

key_merge(stats_dict)

Combine sufficient statistics with other accumulators sharing a matching key.

Parameters:

stats_dict (Dict[str, Any]) – Dictionary mapping keys to accumulators.

Returns:

None.

Return type:

None

key_replace(stats_dict)

Replace sufficient statistics with values from stats_dict for a matching key.

Parameters:

stats_dict (Dict[str, Any]) – Dictionary mapping keys to accumulators.

Returns:

None.

Return type:

None

acc_to_encoder()

Create DirichletDataEncoder object for encoding sequences of iid Dirichlet observations.

Return type:

DirichletDataEncoder

class DirichletAccumulatorFactory(dim=None, keys=None)

Bases: StatisticAccumulatorFactory

DirichletAccumulatorFactory object for creating DirichletAccumulator objects.

Parameters:

• dim (int | None)

• keys (str | None)

make()

Returns a new DirichletAccumulator with the factory’s dim and keys.

Return type:

DirichletAccumulator

class DirichletEstimator(dim=None, pseudo_count=None, suff_stat=None, delta=1.0e-8, keys=None, use_mpe=False, name=None)

Bases: ParameterEstimator

DirichletEstimator object for estimating a Dirichlet distribution from aggregated sufficient statistics.

Parameters:

• dim (int | None)

• pseudo_count (float | None)

• suff_stat (ndarray | None)

• delta (float | None)

• keys (str | None)

• use_mpe (bool)

• name (str | None)

accumulator_factory()

Create DirichletAccumulator object from attributes variables.

Return type:

DirichletAccumulatorFactory

estimate(nobs, suff_stat)

Estimate a Dirichlet distribution from aggregated sufficient statistics.

Suff_stat is a Tuple of size 4 containing:

suff_stat[0] (float): Sum of observation weights. suff_stat[1] (np.ndarray): Weighted sum of the log of observation vectors. suff_stat[2] (np.ndarray): Weighted sum of observation vectors. suff_stat[3] (np.ndarray): Weighted sum of squared observation vectors.

The concentration parameters are solved from the mean-log-probability sufficient statistic with a fixed-point solver (dirichlet_param_solve, or find_alpha when use_mpe is set).

Parameters:

• nobs (Optional[float]) – Not used. Counts are taken from suff_stat[0].

• suff_stat (Tuple[int, np.ndarray, np.ndarray, np.ndarray]) – See above for details.

Returns:

DirichletDistribution object.

Return type:

DirichletDistribution

class DirichletDataEncoder

Bases: DataSequenceEncoder

DirichletDataEncoder object for encoding sequences of iid Dirichlet observations.

seq_encode(x)

Encode a sequence of iid probability-vector observations for vectorized ‘seq_’ calls.

Parameters:

x (Sequence[Sequence[float]]) – Sequence of length-dim probability vectors.

Returns:

Tuple of (log of observations clipped away from zero, observations, squared observations).