mixle.stats.multivariate.diagonal_gaussian module

Create, estimate, and sample from a diagonal Gaussian distribution (independent-multivariate Gaussian).

Defines the DiagonalGaussianDistribution, DiagonalGaussianSampler, DiagonalGaussianAccumulatorFactory, DiagonalGaussianAccumulator, DiagonalGaussianEstimator, and the DiagonalGaussianDataEncoder classes for use with mixle.

The log-density of an ‘n’ dimensional diagonal-gaussian observation x = (x_1,x_2,…,x_n) with mean mu=(m_1,m_2,..,m_n), and diagonal covariance matrix given by covar = diag(s2_1, s2_2,…,s2_n).

log(p_mat(x)) = -0.5*sum_{i=1}^{n} (x_i-m_i)^2 / s2_i - 0.5*log(s2_i) - (n/2)*log(2*pi).

Data type: x (List[float], np.ndarray).

Reference: Mardia, Kent & Bibby, Multivariate Analysis (Academic Press, 1979).

class DiagonalGaussianFisherView(dist)

Bases: FixedFisherView

Parameters:

dist (Any)

class DiagonalGaussianDistribution(mu, covar=MISSING, name=None, keys=None, covariance=MISSING, prior=None)

Bases: SequenceEncodableProbabilityDistribution

Multivariate Gaussian distribution with independent components (diagonal covariance matrix).

Parameters:

• mu (Sequence[float] | ndarray)

• covar (Sequence[float] | ndarray)

• name (str | None)

• keys (str | None)

• covariance (Sequence[float] | ndarray)

• prior (SequenceEncodableProbabilityDistribution | None)

classmethod compute_capabilities()

classmethod compute_declaration()

static exp_family_sufficient_statistics(x, engine)

Return vector sufficient statistics for generated diagonal-Gaussian scoring.

Parameters:

• x (Any)

• engine (Any)

Return type:

tuple[Any, …]

static exp_family_natural_parameters(params, engine)

Return vector natural parameters for generated diagonal-Gaussian scoring.

Parameters:

• params (dict[str, Any])

• engine (Any)

Return type:

tuple[Any, …]

static exp_family_log_partition(params, engine)

Return the diagonal-Gaussian log partition for generated scoring.

Parameters:

• params (dict[str, Any])

• engine (Any)

Return type:

Any

static backend_legacy_sufficient_statistics(x, params, engine)

Return row-wise legacy accumulator statistics for generated resident reductions.

Parameters:

• x (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

tuple[Any, …]

set_prior(prior)

Attach a parameter prior and precompute conjugate-prior expectations.

With a MultivariateNormalGamma(mu0, lam, a, b) prior over (mu, tau=1/covar) this caches the expected natural parameters [ea, eb, e1, e2] with e1 = E[mu*tau] and e2 = -0.5*E[tau] per component (ea, eb scalars summed over components), so that expected_log_density(x) = x.e1 + (x*x).e2 - ea + eb. 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 | mu, tau)] under the prior.

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

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 at observation x.

See log_density() for details.

Parameters:

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

Returns:

Density at x.

log_density(x)

Evaluate the log-density at observation x.

The log-density is given by

log(p(x)) = -0.5*sum_{i=1}^{n} (x_i-m_i)^2 / s2_i - 0.5*log(s2_i) - (n/2)*log(2*pi).

Parameters:

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

Returns:

Log-density at x.

condition(observed)

Return the conditional over the unobserved dimensions given observed.

A diagonal Gaussian has independent coordinates, so conditioning on some of them leaves the rest unchanged: the result is just DiagonalGaussian(mu[unobserved], covar[unobserved]) (the observed values do not shift the unobserved mean or variance). Provided so diagonal-covariance Gaussian mixtures support MixtureDistribution.conditional() – there the responsibilities still update from how well each component explains the observed coordinates, even though the within-component coordinates are independent. Raises if no dimension is left unobserved.

Parameters:

observed (dict[int, float])

Return type:

DiagonalGaussianDistribution

marginal(keep)

Return the marginal over the dimensions keep: DiagonalGaussian(mu[keep], covar[keep]).

Marginalizing a diagonal Gaussian simply drops the other independent coordinates (order kept).

Parameters:

keep (Sequence[int])

Return type:

DiagonalGaussianDistribution

density_cumulative(x)

Exact probability-ordered cumulative G(x) = P(p(Y) >= p(x)) – the highest-density-region mass through x (multivariate analogue of a CDF). For a diagonal Gaussian the squared Mahalanobis distance sum_i (x_i-mu_i)^2/var_i is chi-square(dim), so G = chi2.cdf(maha2, dim). Used by mixle.enumeration.density_rank.density_rank() to return an EXACT cumulative.

Parameters:

x (Sequence[float] | ndarray)

Return type:

float

density_quantile(q)

Inverse of density_cumulative(): a representative point at cumulative-density index q.

q is the highest-density-region mass, whose boundary is the squared-Mahalanobis level chi2.ppf(q, dim); a representative point on that contour offsets the first coordinate by sqrt(level * var_0) (Mahalanobis distance exactly the level). Sweeping q enumerates the support in descending density.

Parameters:

q (float)

Return type:

ndarray

seq_log_density(x)

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

Parameters:

x (np.ndarray) – Encoded data matrix with shape (sz, dim) from DiagonalGaussianDataEncoder.seq_encode().

Returns:

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

Return type:

ndarray

static backend_log_density_from_params(x, mu, covar, engine)

Engine-neutral diagonal Gaussian log-density from explicit parameters.

Parameters:

• x (Any)

• mu (Any)

• covar (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 diagonal-Gaussian parameters for a homogeneous mixture kernel.

Parameters:

• dists (Sequence[DiagonalGaussianDistribution])

• engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)

Return an (n, k) matrix of diagonal-Gaussian log densities.

Parameters:

• x (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

Any

to_fisher(**kwargs)

Return this distribution’s own Fisher view.

sampler(seed=None)

Create a DiagonalGaussianSampler for sampling from this distribution.

Parameters:

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

Returns:

DiagonalGaussianSampler object.

Return type:

DiagonalGaussianSampler

estimator(pseudo_count=None)

Create a DiagonalGaussianEstimator for estimating this distribution.

Parameters:

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

Returns:

DiagonalGaussianEstimator object.

Return type:

DiagonalGaussianEstimator

dist_to_encoder()

Returns a DiagonalGaussianDataEncoder object for encoding sequences of iid observations.

Return type:

DiagonalGaussianDataEncoder

class DiagonalGaussianSampler(dist, seed=None)

Bases: DistributionSampler

DiagonalGaussianSampler object for sampling from a DiagonalGaussianDistribution.

Parameters:

• dist (DiagonalGaussianDistribution)

• seed (int | None)

sample(size=None)

Draw iid samples from the diagonal Gaussian distribution.

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 a list of ‘size’ such arrays.

Return type:

Sequence[ndarray] | ndarray

class DiagonalGaussianAccumulator(dim=None, keys=None)

Bases: SequenceEncodableStatisticAccumulator

DiagonalGaussianAccumulator 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.

Parameters:

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

• weight (float) – Weight for the observation.

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

Returns:

None.

Return type:

None

initialize(x, weight, rng)

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

Parameters:

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

• weight (float) – Weight for the observation.

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

Returns:

None.

Return type:

None

seq_update(x, weights, estimate)

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

Parameters:

• x (np.ndarray) – Encoded data matrix with shape (sz, dim).

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

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

Returns:

None.

Return type:

None

seq_initialize(x, weights, rng)

Vectorized initialization of the accumulator. Calls seq_update().

Parameters:

• x (np.ndarray) – Encoded data matrix with shape (sz, dim).

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

• rng (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[np.ndarray, np.ndarray, float]) – Tuple of (weighted sum of observations, weighted sum of squared observations, sum of weights).

Returns:

DiagonalGaussianAccumulator object.

Return type:

DiagonalGaussianAccumulator

value()

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

Return type:

tuple[ndarray, ndarray, float]

from_value(x)

Set the sufficient statistics of the accumulator to x.

Parameters:

x (Tuple[np.ndarray, np.ndarray, float]) – Tuple of (weighted sum of observations, weighted sum of squared observations, sum of weights).

Returns:

DiagonalGaussianAccumulator object.

Return type:

DiagonalGaussianAccumulator

key_merge(stats_dict)

Combine sufficient statistics with other accumulators sharing a matching key.

Parameters:

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

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 aggregated statistics.

Returns:

None.

Return type:

None

acc_to_encoder()

Returns a DiagonalGaussianDataEncoder object for encoding sequences of iid observations.

Return type:

DiagonalGaussianDataEncoder

class DiagonalGaussianAccumulatorFactory(dim=None, keys=None)

Bases: StatisticAccumulatorFactory

DiagonalGaussianAccumulatorFactory object for creating DiagonalGaussianAccumulator objects.

Parameters:

• dim (int | None)

• keys (str | None)

make()

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

Return type:

DiagonalGaussianAccumulator

class DiagonalGaussianEstimator(dim=None, pseudo_count=(None, None), suff_stat=(None, None), name=None, keys=None, prior=None, min_covar=None, ridge=None)

Bases: ParameterEstimator

DiagonalGaussianEstimator object for estimating a diagonal Gaussian distribution from aggregated sufficient statistics.

Parameters:

• dim (int | None)

• pseudo_count (tuple[float | None, float | None])

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

• name (str | None)

• keys (str | None)

• prior (SequenceEncodableProbabilityDistribution | None)

• min_covar (float | None)

• ridge (float | None)

accumulator_factory()

Returns a DiagonalGaussianAccumulatorFactory built from the estimator’s attributes.

Return type:

DiagonalGaussianAccumulatorFactory

model_log_density(model)

Log-density of the model parameters under the MultivariateNormalGamma prior (ELBO global term).

The prior is over (mu, tau=1/covar), so the model’s covariance is inverted before scoring.

Parameters:

model (DiagonalGaussianDistribution)

Return type:

float

estimate(nobs, suff_stat)

Estimate a diagonal Gaussian distribution from aggregated sufficient statistics.

Suff_stat is a Tuple of size 3 containing:

suff_stat[0] (np.ndarray): Component-wise sum of weighted observation values. suff_stat[1] (np.ndarray): Component-wise sum of weighted squared observation values. suff_stat[2] (float): Sum of weights for each observation.

Parameters:

• nobs (Optional[float]) – Weighted number of observations used in aggregation of suff stats.

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

Returns:

DiagonalGaussianDistribution object.

Return type:

DiagonalGaussianDistribution

class DiagonalGaussianDataEncoder(dim=None)

Bases: DataSequenceEncoder

DiagonalGaussianDataEncoder object for encoding sequences of iid diagonal-Gaussian observations.

Parameters:

dim (int | None)

seq_encode(x)

Encode a sequence of iid length-dim observations for vectorized ‘seq_’ calls.

Parameters:

x (Sequence[Union[List[float], np.ndarray]]) – Sequence of length-dim observation vectors.

Returns:

Encoded data matrix with shape (len(x), dim).

Return type:

ndarray