mixle.stats.latent.semi_supervised_mixture module¶

Semi-supervised mixture distributions with per-observation component priors.

Observations have the form (y, prior), where y is an observation for the component distributions and prior is either None or a sequence of (component_index, probability) pairs. When prior information is present, the mixture weights are reweighted for that observation before scoring or responsibility updates. When prior is None, the model behaves like an ordinary mixture.

seq_initialize() falls back to scalar initialize() calls on raw observations, so initialization 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]: Declare generated-compute support inherited from mixture components.

compute_declaration()[source]: Return the generated-compute declaration for the semi-supervised mixture.

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]: Return exact-or-approximate density semantics joined from component models.

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 (low-overhead, 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]

Create an accumulator factory from the child estimators.

Returns:: Factory configured for this estimator.
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:

Estimated distribution.

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