mixle.stats.latent.dirac_length module

Create, estimate, and sample from a mixture over a dirac delta at v and a length distribution.

Defines the DiracLengthMixtureDistribution, DiracLengthMixtureSampler, DiracLengthMixtureAccumulatorFactory, DiracLengthMixtureAccumulator, DiracLengthMixtureEstimator, and the DiracLengthMixtureDataEncoder classes for use with mixle.

The DiracLengthMixtureDistribution is defined by the density of the form,

P(Y) = p*P_1(Y) + (1-p)*Delta_{v}(Y),

where P_1() is a length distribution with support on non-negative integers, or a subset of them, and Delta_{v}(x) = 1 if x = v, else 0.

class DiracLengthMixtureDistribution(len_dist, p, v=0, name=None)

Bases: SequenceEncodableProbabilityDistribution

DiracLengthMixtureDistribution object defined by a length distribution, choice of dirac value, and p.

Parameters:

• p (float) – Probability of being drawn from length distribution. Must be between 0 and 1.

• len_dist (SequenceEncodableProbabilityDistribution) – Distribution with support on non-negative integers.

• name (Optional[str]) – Set name for object instance.

• v (int)

p

Probability of being drawn from length distribution. Must be between 0 and 1.

Type:

float

len_dist

Distribution with support on non-negative integers.

Type:

SequenceEncodableProbabilityDistribution

name

Name for object instance.

Type:

Optional[str]

compute_capabilities()

compute_declaration()

density(x)

Evaluate density of length Dirac mixture distribution at observation x.

See log_density() for details.

Parameters:

x (int) – Integer value.

Returns:

Density at x.

Return type:

float

log_density(x)

Evaluate the log-density of length Dirac mixture distribution at observation x.

log(P(x)) = log( p*P_1(x) + (1-p)*Delta_{v}(x) ),

Parameters:

x (int) – Integer value.

Returns:

log-density at x.

Return type:

float

component_log_density(x)

Log-density of each mixture component (length distribution, dirac at v) at x.

Parameters:

x (int) – Integer value.

Returns:

Numpy array of the two component log-densities.

Return type:

ndarray

posterior(x)

Posterior probability of each mixture component given observation x.

Parameters:

x (int) – Integer value.

Returns:

Numpy array of the two component posterior probabilities (sums to one).

Return type:

ndarray

seq_component_log_density(x)

Vectorized component log-densities at sequence encoded input x.

Parameters:

x (E) – Sequence encoded data from DiracLengthMixtureDataEncoder.

Returns:

Numpy array of shape (len(x), 2) of component log-densities.

Return type:

ndarray

seq_log_density(x)

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

Parameters:

x (E) – Sequence encoded data from DiracLengthMixtureDataEncoder.

Returns:

Numpy array of log-density (float) of len(x).

Return type:

ndarray

backend_seq_component_log_density(x, engine)

Engine-neutral component log densities for encoded length/dirac mixtures.

Parameters:

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

• engine (Any)

Return type:

Any

backend_seq_log_density(x, engine)

Engine-neutral mixture log-density for encoded length/dirac observations.

Parameters:

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

• engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)

Return stacked parameters for shared-dirac length mixtures.

Parameters:

• dists (Sequence[DiracLengthMixtureDistribution])

• engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)

Return an (n, k) matrix of length/dirac mixture log densities.

Parameters:

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

• params (dict[str, Any])

• engine (Any)

Return type:

Any

classmethod backend_stacked_sufficient_statistics_with_estimator(x, weights, params, engine, estimator)

Return per-component legacy (component_counts, length_stat) statistics.

Parameters:

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

• weights (Any)

• params (dict[str, Any])

• engine (Any)

• estimator (Any)

Return type:

tuple[Any, …]

seq_posterior(x)

Vectorized component posterior probabilities at sequence encoded input x.

Parameters:

x (E) – Sequence encoded data from DiracLengthMixtureDataEncoder.

Returns:

Numpy array of shape (len(x), 2) of component posteriors.

Return type:

ndarray

sampler(seed=None)

Create a DiracLengthMixtureSampler from parameters of this distribution.

Parameters:

seed (Optional[int]) – Used to set seed in random sampler.

Returns:

DiracLengthMixtureSampler object.

Return type:

DiracLengthMixtureSampler

estimator(pseudo_count=None)

Create a DiracLengthMixtureEstimator with matching dirac value v.

Parameters:

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

Returns:

DiracLengthMixtureEstimator object.

Return type:

DiracLengthMixtureEstimator

dist_to_encoder()

Returns a DiracLengthMixtureDataEncoder for encoding sequences of iid integer observations.

Return type:

DiracLengthMixtureDataEncoder

enumerator()

Returns a DiracLengthMixtureEnumerator iterating the union of the length-distribution support and the dirac point v in descending probability order.

Return type:

DiracLengthMixtureEnumerator

class DiracLengthMixtureEnumerator(dist)

Bases: DistributionEnumerator

Enumerates the union of the length-distribution support and the dirac point v.

The model is a two-component mixture: the length distribution with weight p and a dirac delta at v with weight 1-p. The dirac component contributes the trivial single-point stream [(v, 0.0)]. Supports may overlap (the length distribution can also emit v), so candidates are de-duplicated and re-scored exactly with the mixture log-density.

Parameters:

dist (DiracLengthMixtureDistribution)

class DiracLengthMixtureSampler(dist, seed=None)

Bases: DistributionSampler

DiracLengthMixtureSampler object for sampling from a DiracLengthMixtureDistribution.

Parameters:

• dist (DiracLengthMixtureDistribution)

• seed (int | None)

sample(size=None)

Draw iid samples from a DiracLengthMixture distribution.

Parameters:

size (Optional[int]) – Number of iid samples to draw.

Returns:

Int or List[int] depending on size = None or size (int).

Return type:

list[int] | int

class DiracLengthMixtureAccumulator(accumulator, v=0, keys=(None, None), name=None)

Bases: SequenceEncodableStatisticAccumulator

DiracLengthMixtureAccumulator object for accumulating component counts and length-distribution statistics.

Parameters:

• accumulator (SequenceEncodableStatisticAccumulator) – Accumulator for the length distribution.

• v (int) – Dirac location.

• keys (Tuple[Optional[str], Optional[str]]) – Keys for the mixture weights and component statistics.

• name (Optional[str]) – Set name for object instance.

accumulator

Accumulator for the length distribution.

Type:

SequenceEncodableStatisticAccumulator

comp_counts

Posterior-weighted counts for the two components.

Type:

np.ndarray

weight_key

Key for merging mixture weight counts.

Type:

Optional[str]

comp_key

Key for merging component sufficient statistics.

Type:

Optional[str]

v

Dirac location.

Type:

int

name

Name for object instance.

Type:

Optional[str]

seq_update(x, weights, estimate)

Vectorized accumulation of posterior-weighted statistics from encoded observations x.

Parameters:

• x (E) – Sequence encoded data from DiracLengthMixtureDataEncoder.

• weights (np.ndarray) – Weights on the observations.

• estimate (DiracLengthMixtureDistribution) – Previous estimate used for posteriors.

seq_update_engine(x, weights, estimate, engine)

Engine-resident E-step: the length-distribution scoring (the heavy term) runs through the active engine; the cheap two-component (length vs. Dirac) responsibility bookkeeping mirrors the host seq_update exactly.

update(x, weight, estimate)

Add one observation’s posterior-weighted contribution to the sufficient statistics.

Parameters:

• x (int) – Integer observation.

• weight (float) – Weight on the observation.

• estimate (DiracLengthMixtureDistribution) – Previous estimate used for posteriors.

Return type:

None

initialize(x, weight, rng)

Initialize the accumulator with observation x, randomly splitting weight at the dirac point.

Parameters:

• x (int) – Integer observation.

• weight (float) – Weight on the observation.

• rng (RandomState) – Random number generator for initialization.

seq_initialize(x, weights, rng)

Vectorized initialization from encoded observations x with random splits at the dirac point.

Parameters:

• x (E) – Sequence encoded data from DiracLengthMixtureDataEncoder.

• weights (np.ndarray) – Weights on the observations.

• rng (RandomState) – Random number generator for initialization.

Return type:

None

combine(suff_stat)

Combine sufficient statistics (component counts, length-dist stats) with this accumulator.

Parameters:

suff_stat (Tuple[np.ndarray, SS0]) – Component counts and length-distribution statistics.

Returns:

This DiracLengthMixtureAccumulator.

Return type:

DiracLengthMixtureAccumulator

value()

Returns sufficient statistics as a tuple (component counts, length-distribution statistics).

Return type:

tuple[ndarray, Any]

from_value(x)

Set sufficient statistics from a (component counts, length-distribution statistics) tuple.

Parameters:

x (Tuple[np.ndarray, SS0]) – Component counts and length-distribution statistics.

Returns:

This DiracLengthMixtureAccumulator.

Return type:

DiracLengthMixtureAccumulator

key_merge(stats_dict)

Merge keyed sufficient statistics into stats_dict under the weight and component keys.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)

Replace keyed sufficient statistics from stats_dict under the weight and component keys.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()

Returns a DiracLengthMixtureDataEncoder for encoding sequences of iid integer observations.

Return type:

DiracLengthMixtureDataEncoder

class DiracLengthMixtureAccumulatorFactory(factory, v=0, keys=(None, None), name=None)

Bases: StatisticAccumulatorFactory

DiracLengthMixtureAccumulatorFactory object for creating DiracLengthMixtureAccumulator objects.

Parameters:

• factory (StatisticAccumulatorFactory) – Accumulator factory for the length distribution.

• v (int) – Dirac location.

• keys (Tuple[Optional[str], Optional[str]]) – Keys for the mixture weights and component statistics.

• name (Optional[str]) – Set name for object instance.

factory

Accumulator factory for the length distribution.

Type:

StatisticAccumulatorFactory

v

Dirac location.

Type:

int

keys

Keys for the mixture weights and component statistics.

Type:

Tuple[Optional[str], Optional[str]]

name

Name for object instance.

Type:

Optional[str]

make()

Returns a new DiracLengthMixtureAccumulator wrapping a fresh length-distribution accumulator.

Return type:

DiracLengthMixtureAccumulator

class DiracLengthMixtureEstimator(estimator, v=0, fixed_p=None, suff_stat=None, pseudo_count=None, name=None, keys=(None, None))

Bases: ParameterEstimator

DiracLengthMixtureEstimator object for estimating DiracLengthMixtureDistribution objects.

Parameters:

• estimator (ParameterEstimator) – Estimator for the length distribution.

• v (int) – Dirac location.

• fixed_p (Optional[float]) – Hold the length-distribution weight p fixed at this value.

• suff_stat (Optional[float]) – Prior value of p used with pseudo_count for regularization.

• pseudo_count (Optional[float]) – Used to inflate the component count statistics.

• name (Optional[str]) – Set name for object instance.

• keys (Tuple[Optional[str], Optional[str]]) – Keys for the mixture weights and component statistics.

estimator

Estimator for the length distribution.

Type:

ParameterEstimator

v

Dirac location.

Type:

int

pseudo_count

Used to inflate the component count statistics.

Type:

Optional[float]

suff_stat

Prior value of p used with pseudo_count for regularization.

Type:

Optional[float]

keys

Keys for the mixture weights and component statistics.

Type:

Tuple[Optional[str], Optional[str]]

name

Name for object instance.

Type:

Optional[str]

fixed_p_vec

Fixed component weights [p, 1-p] when fixed_p is given.

Type:

Optional[np.ndarray]

accumulator_factory()

Returns a DiracLengthMixtureAccumulatorFactory consistent with this estimator.

Return type:

DiracLengthMixtureAccumulatorFactory

estimate(nobs, suff_stat)

Estimate a DiracLengthMixtureDistribution from accumulated sufficient statistics.

Parameters:

• nobs (Optional[float]) – Weighted number of observations.

• suff_stat (Tuple[np.ndarray, SS0]) – Component counts and length-distribution statistics.

Returns:

DiracLengthMixtureDistribution object.

Return type:

DiracLengthMixtureDistribution

class DiracLengthMixtureDataEncoder(encoder, v=0)

Bases: DataSequenceEncoder

DiracLengthMixtureDataEncoder object for encoding sequences of iid integer observations.

Parameters:

• encoder (DataSequenceEncoder) – Encoder for the length distribution.

• v (int) – Dirac location.

encoder

Encoder for the length distribution.

Type:

DataSequenceEncoder

v

Dirac location.

Type:

int

seq_encode(x)

Encode a sequence of iid integer observations for vectorized use.

Parameters:

x (Sequence[int]) – Sequence of iid integer observations.

Returns:

Tuple of (sequence length, indices equal to v, indices not equal to v, base-encoded data).

Return type:

tuple[int, ndarray, ndarray, Any]