mixle.stats.latent.semi_supervised_mixture module¶

Create, estimate, and sample from a semi-supervised mixture distribution.

Defines the SemiSupervisedMixtureDistribution, SemiSupervisedMixtureSampler, SemiSupervisedMixtureAccumulatorFactory, SemiSupervisedMixtureEstimatorAccumulator, SemiSupervisedMixtureEstimator, and the SemiSupervisedMixtureDataEncoder classes for use with mixle.

Data type (Tuple[T, Optional[Sequence[Tuple[int, float]]]): T is the data type of the mixture components. The optional Sequence of tuples contain labels for the observations coming from the component (0,1,2,…num_components-1) and an associated probability for the label.

The likelihood for an observation x = (y, prior) is simply a mixture distribution with the weights of the mixture re-weighted to account for the prior knowledge that x was observed from components in prior with probs in prior as well.

If no prior is provided, the likelihood is simply a mixture.

Note: seq_initialize() falls back to scalar initialize() calls on the raw observations, so it is not vectorized.

class SemiSupervisedMixtureDistribution(components, w=MISSING, name=None, weights=MISSING)[source]

Bases: SequenceEncodableProbabilityDistribution

SemiSupervisedMixtureDistribution models observations (value, prior) where the optional prior labels re-weight the mixture weights over the listed components.

Parameters:

components (Sequence[SequenceEncodableProbabilityDistribution])
w (list[float] | ndarray)
name (str | None)
weights (list[float] | ndarray)

compute_capabilities()[source]

compute_declaration()[source]

density(x)[source]

Density of the semi-supervised mixture at observation x.

See log_density() for details.

Parameters:: x (Tuple[T0, Optional[Sequence[Tuple[int, T1]]]]) – Observation (value, prior).
Returns:: Density at x.
Return type:: float

log_density(x)[source]

Log-density of the semi-supervised mixture at observation x = (value, prior).

If prior is None this is the standard mixture log-density. Otherwise the mixture weights are restricted to the components listed in the prior, re-weighted by the prior probabilities, and re-normalized before mixing the component log-densities.

Parameters:: x (Tuple[T0, Optional[Sequence[Tuple[int, T1]]]]) – Observation (value, prior), where prior is an optional sequence of (component index, probability) pairs.
Returns:: Log-density at x.
Return type:: float

posterior(x)[source]

Posterior probability of each component for observation x = (value, prior).

Components not listed in the prior (when a prior is present) receive posterior 0.

Parameters:: x (Tuple[T0, Optional[Sequence[Tuple[int, T1]]]]) – Observation (value, prior).
Returns:: Numpy array of length num_components containing the component posteriors.
Return type:: ndarray

seq_log_density(x)[source]

Vectorized evaluation of the log-density on sequence encoded data x.

Parameters:: x (E) – Sequence encoded data produced by SemiSupervisedMixtureDataEncoder.seq_encode().
Returns:: Numpy array of log-densities, one entry per encoded observation.
Return type:: ndarray

backend_seq_log_density(x, engine)[source]

Engine-neutral semi-supervised mixture log-density for encoded observations.

Parameters:

x (tuple[int, E0, tuple[tuple[ndarray, ndarray, ndarray, ndarray], ndarray, ndarray], Sequence[tuple[T0, Sequence[tuple[int, T1]] | None]]])
engine (Any)

Return type:

Any

seq_posterior(x)[source]

Vectorized component posteriors on sequence encoded data x.

Parameters:: x (E) – Sequence encoded data produced by SemiSupervisedMixtureDataEncoder.seq_encode().
Returns:: Numpy array of shape (number of observations, num_components) of posteriors.
Return type:: ndarray

density_semantics()[source]

What log_density returns relative to the true log-density (default: exact).

Override to declare that this distribution’s log_density is a variational lower bound (ELBO), an upper bound, or an approximation rather than the exact log p(x). This is surfaced as the ExactDensity capability and noted in mixle.describe(), so code that needs an exact likelihood can require(x, ExactDensity) instead of silently trusting a bound.

sampler(seed=None)[source]

Creates a SemiSupervisedMixtureSampler for sampling component values.

Parameters:: seed (Optional[int]) – Seed for the random number generator used in sampling.
Returns:: SemiSupervisedMixtureSampler object.
Return type:: SemiSupervisedMixtureSampler

estimator(pseudo_count=None)[source]

Creates a SemiSupervisedMixtureEstimator with one child estimator per component.

Parameters:: pseudo_count (Optional[float]) – Used to inflate the sufficient statistics of the mixture weights.
Returns:: SemiSupervisedMixtureEstimator object.
Return type:: SemiSupervisedMixtureEstimator

dist_to_encoder()[source]

Creates a SemiSupervisedMixtureDataEncoder for encoding sequences of (value, prior) observations.

Returns:: SemiSupervisedMixtureDataEncoder object.
Return type:: SemiSupervisedMixtureDataEncoder

enumerator()[source]

Enumeration is not well-defined for semi-supervised mixtures.

Observations pair a component value with exogenous prior labels: the model defines no distribution over the prior part, so the support over (value, prior) pairs cannot be enumerated with consistent probabilities.

Raises:: EnumerationError always. –
Return type:: DistributionEnumerator

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

Bases: DistributionSampler

SemiSupervisedMixtureSampler draws component values from a SemiSupervisedMixtureDistribution.

Parameters:

dist (SemiSupervisedMixtureDistribution)
seed (int | None)

sample(size=None)[source]

Draw ‘size’ component values from the mixture (no prior labels are generated).

Parameters:: size (Optional[int]) – Number of independent samples. If None a single value is returned.
Returns:: A single component value if size is None, else a list of ‘size’ component values.
Return type:: Sequence[Any] | Any

class SemiSupervisedMixtureEstimatorAccumulator(accumulators, keys=(None, None), name=None)[source]

Bases: SequenceEncodableStatisticAccumulator

SemiSupervisedMixtureEstimatorAccumulator accumulates posterior-weighted sufficient statistics for the mixture weights and each component.

Parameters:

accumulators (Sequence[SequenceEncodableStatisticAccumulator])
keys (tuple[str | None, str | None] | None)
name (str | None)

update(x, weight, estimate)[source]

Update the sufficient statistics with one weighted observation x = (value, prior).

The component posteriors are computed from the current estimate (the prior labels restrict and re-weight them), and each component accumulator receives the value with weight posterior * weight.

Parameters:

x (Tuple[T0, Optional[Sequence[Tuple[int, T1]]]]) – Observation (value, prior).
weight (float) – Weight for the observation.
estimate (SemiSupervisedMixtureDistribution) – Current mixture estimate used to compute the component posteriors. Required.

Returns:

None.

Return type:

None

initialize(x, weight, rng)[source]

Initialize the accumulator with one weighted observation x = (value, prior).

If a prior is present the value is assigned to the listed components with the prior probabilities as weights; otherwise a random component receives almost all the weight.

Parameters:

x (Tuple[T0, Optional[Sequence[Tuple[int, T1]]]]) – Observation (value, prior).
weight (float) – Weight for the observation.
rng (RandomState) – RandomState used to seed the member RandomStates.

Returns:

None.

Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize the accumulator from sequence encoded data x.

Note: falls back to scalar initialize() on the raw observations carried in the encoding, so it is not vectorized.

Parameters:

x (E) – Sequence encoded data produced by SemiSupervisedMixtureDataEncoder.seq_encode().
weights (np.ndarray) – Weights for each encoded observation.
rng (RandomState) – RandomState used to seed the member RandomStates.

Returns:

None.

Return type:

None

seq_update(x, weights, estimate)[source]

Vectorized update of the sufficient statistics from sequence encoded data x.

Computes the prior-adjusted component posteriors for all observations and passes the posterior-weighted encoded data to each component accumulator’s seq_update.

Parameters:

x (E) – Sequence encoded data produced by SemiSupervisedMixtureDataEncoder.seq_encode().
weights (np.ndarray) – Weights for each encoded observation.
estimate (SemiSupervisedMixtureDistribution) – Current mixture estimate used to compute the component posteriors. Required.

Returns:

None.

Return type:

None

seq_update_engine(x, weights, estimate, engine)[source]: Engine-resident E-step: component scoring and the responsibility softmax run on the active engine; the (cheap, index-based) semi-supervised prior adjustment is built host-side. Matches the host seq_update.

combine(suff_stat)[source]

Aggregate sufficient statistics suff_stat with this accumulator’s statistics.

Parameters:: suff_stat (Tuple[np.ndarray, Tuple[SS0, ...]]) – Component counts and component sufficient statistics, as returned by value().
Returns:: SemiSupervisedMixtureEstimatorAccumulator with combined sufficient statistics.
Return type:: SemiSupervisedMixtureEstimatorAccumulator

value()[source]

Returns the sufficient statistics: (component counts, component values).

Return type:: tuple[ndarray, tuple[Any, …]]

from_value(x)[source]

Set the accumulator’s sufficient statistics to x.

Parameters:: x (Tuple[np.ndarray, Tuple[SS0, ...]]) – Component counts and component sufficient statistics, as returned by value().
Returns:: SemiSupervisedMixtureEstimatorAccumulator object.
Return type:: SemiSupervisedMixtureEstimatorAccumulator

key_merge(stats_dict)[source]

Merge the weight and component sufficient statistics for matching keys.

Parameters:: stats_dict (Dict[str, Any]) – Maps keys to shared sufficient statistics.
Returns:: None.
Return type:: None

key_replace(stats_dict)[source]

Replace the weight and component sufficient statistics with keyed values.

Parameters:: stats_dict (Dict[str, Any]) – Maps keys to shared sufficient statistics.
Returns:: None.
Return type:: None

acc_to_encoder()[source]

Creates a SemiSupervisedMixtureDataEncoder for encoding sequences of (value, prior) observations.

Returns:: SemiSupervisedMixtureDataEncoder object.
Return type:: SemiSupervisedMixtureDataEncoder

class SemiSupervisedMixtureEstimatorAccumulatorFactory(factories, dim, keys=(None, None), name=None)[source]

Bases: StatisticAccumulatorFactory

SemiSupervisedMixtureEstimatorAccumulatorFactory creates SemiSupervisedMixtureEstimatorAccumulator objects from the component factories.

Parameters:

factories (Sequence[StatisticAccumulatorFactory])
dim (int)
keys (tuple[str | None, str | None] | None)
name (str | None)

make()[source]

Creates a SemiSupervisedMixtureEstimatorAccumulator with one accumulator per component.

Returns:: SemiSupervisedMixtureEstimatorAccumulator object.
Return type:: SemiSupervisedMixtureEstimatorAccumulator

class SemiSupervisedMixtureEstimator(estimators, suff_stat=None, pseudo_count=None, keys=(None, None), name=None)[source]

Bases: ParameterEstimator

SemiSupervisedMixtureEstimator estimates a SemiSupervisedMixtureDistribution from aggregated sufficient statistics.

Parameters:

estimators (Sequence[ParameterEstimator])
suff_stat (ndarray | None)
pseudo_count (float | None)
keys (tuple[str | None, str | None] | None)
name (str | None)

accumulator_factory()[source]

Creates a SemiSupervisedMixtureEstimatorAccumulatorFactory from the child estimators.

Returns:: SemiSupervisedMixtureEstimatorAccumulatorFactory object.
Return type:: SemiSupervisedMixtureEstimatorAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate a SemiSupervisedMixtureDistribution from aggregated sufficient statistics.

The mixture weights are the normalized component counts, optionally regularized by pseudo_count and the stored suff_stat weights.

Parameters:

nobs (Optional[float]) – Not used. Kept for consistency.
suff_stat (Tuple[np.ndarray, Tuple[SS0, ...]]) – Component counts and component sufficient statistics, as returned by SemiSupervisedMixtureEstimatorAccumulator.value().

Returns:

SemiSupervisedMixtureDistribution object.

Return type:

SemiSupervisedMixtureDistribution

class SemiSupervisedMixtureDataEncoder(encoder, num_components=None)[source]

Bases: DataSequenceEncoder

SemiSupervisedMixtureDataEncoder encodes sequences of (value, prior) observations using a shared component encoder for the values and flat arrays for the prior labels.

Parameters:

encoder (DataSequenceEncoder)
num_components (int | None)

seq_encode(x)[source]

Encode a sequence of iid (value, prior) observations for vectorized “seq_” calls.

The encoding is a tuple of length 4:

rv[0] (int): Number of observations. rv[1]: The values encoded by the shared component encoder. rv[2]: Prior arrays ((row index, component index, prob, log prob), per-row prior

sums, per-row has-prior flags).

rv[3]: The raw observations (used by seq_initialize).

Parameters:: x (Sequence[Tuple[T0, Optional[Sequence[Tuple[int, T1]]]]]) – Observations.
Returns:: See description above.
Return type:: tuple[int, Any, tuple[tuple[ndarray, ndarray, ndarray, ndarray], ndarray, ndarray], Sequence[tuple[T0, Sequence[tuple[int, T1]] | None]]]

SemiSupervisedMixtureAccumulator: alias of SemiSupervisedMixtureEstimatorAccumulator

SemiSupervisedMixtureAccumulatorFactory: alias of SemiSupervisedMixtureEstimatorAccumulatorFactory