mixle.stats.multivariate.diagonal_gaussian module¶

Diagonal Gaussian distributions, estimators, accumulators, and encoders.

The log-density of an n-dimensional diagonal Gaussian observation x = (x_1, x_2, ..., x_n) with mean mu and diagonal covariance covar = diag(s2_1, s2_2, ..., s2_n) is:

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

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

class DiagonalGaussianFisherView(dist)[source]

Bases: FixedFisherView

Fisher view over per-dimension first and second moments for a diagonal Gaussian.

Parameters:: dist (Any)

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

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()[source]: Declare backend support for diagonal Gaussian generated kernels.

classmethod compute_declaration()[source]: Return the generated-compute declaration for the diagonal Gaussian.

static exp_family_sufficient_statistics(x, engine)[source]

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

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

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

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

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

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

Vectorized expected_log_density over sequence-encoded observations.

Parameters:: x (ndarray)
Return type:: ndarray

density(x)[source]

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

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

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

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

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

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

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

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

Engine-neutral vectorized log-density for encoded data.

Parameters:

x (Any)
engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)[source]

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

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)[source]: Return this distribution’s own Fisher view.

sampler(seed=None)[source]

Return a sampler for iid draws from this distribution.

Parameters:: seed (int | None) – Optional seed for the sampler’s random state.
Returns:: A configured DiagonalGaussianSampler.
Return type:: DiagonalGaussianSampler

estimator(pseudo_count=None)[source]

Return an estimator initialized from this distribution’s shape.

Parameters:: pseudo_count (float | None) – Optional smoothing count applied to mean and variance estimates.
Returns:: A DiagonalGaussianEstimator.
Return type:: DiagonalGaussianEstimator

dist_to_encoder()[source]

Return an encoder for iid diagonal Gaussian observations.

Return type:: DiagonalGaussianDataEncoder

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

Bases: DistributionSampler

Sampler for iid diagonal Gaussian observations.

Parameters:

dist (DiagonalGaussianDistribution)
seed (int | None)

sample(size=None)[source]

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

Bases: SequenceEncodableStatisticAccumulator

Accumulator for diagonal Gaussian sufficient statistics.

Parameters:

dim (int | None)
keys (str | None)

update(x, weight, estimate)[source]

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

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

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

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

Merge sufficient statistics 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:: This accumulator.
Return type:: DiagonalGaussianAccumulator

value()[source]

Return (sum, sum_squares, count) sufficient statistics.

Return type:: tuple[ndarray, ndarray, float]

from_value(x)[source]

Replace this accumulator’s sufficient statistics.

Parameters:: x (Tuple[np.ndarray, np.ndarray, float]) – Tuple of (weighted sum of observations, weighted sum of squared observations, sum of weights).
Returns:: This accumulator.
Return type:: DiagonalGaussianAccumulator

key_merge(stats_dict)[source]

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

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

Return an encoder compatible with this accumulator’s dimension.

Return type:: DiagonalGaussianDataEncoder

class DiagonalGaussianAccumulatorFactory(dim=None, keys=None)[source]

Bases: StatisticAccumulatorFactory

Factory for diagonal Gaussian accumulators.

Parameters:

dim (int | None)
keys (str | None)

make()[source]

Return a fresh accumulator with the factory configuration.

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

Bases: ParameterEstimator

Estimator for diagonal Gaussian distributions.

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

Return an accumulator factory matching this estimator.

Return type:: DiagonalGaussianAccumulatorFactory

model_log_density(model)[source]

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

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

Bases: DataSequenceEncoder

Encoder for iid diagonal Gaussian observations.

Parameters:: dim (int | None)

seq_encode(x)[source]

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