mixle.stats.combinator.conditional module

Conditional distributions over paired observations.

Data type: (Tuple[T0, T1]): The ConditionalDistribution if given by density,

P(X0,X1) = P_cond(X1|X0)*P_given(X0).

The ConditionalDistribution allows for user defined conditional distributions P_cond(X1|X0), and given distributions P_given(X0).

class ConditionalDistribution(dmap, default_dist=NullDistribution(), given_dist=NullDistribution(), name=None, keys=None, prior=None)

Bases: SequenceEncodableProbabilityDistribution

ConditionalDistribution models pairs (x0, x1) with density P_cond(x1 | x0) * P_given(x0), where the conditional distributions are looked up from a dictionary keyed by x0.

Parameters:

• dmap (dict[Any, SequenceEncodableProbabilityDistribution] | list[SequenceEncodableProbabilityDistribution])

• default_dist (SequenceEncodableProbabilityDistribution | None)

• given_dist (SequenceEncodableProbabilityDistribution | None)

• name (str | None)

• keys (str | None)

• prior (tuple[dict[Any, Any], Any, Any] | None)

get_prior()

Return the joint prior as (per_branch_priors, default_prior, given_prior).

per_branch_priors is a dict keyed like dmap of each conditional branch’s prior.

Return type:

tuple[dict[Any, Any], Any, Any]

set_prior(prior)

Distribute per-branch priors to the conditional branches, default, and given distributions.

prior=None is a no-op (children keep their existing priors, leaving the MLE path byte-identical). Otherwise prior is (per_branch_priors, default_prior, given_prior): each dmap branch prior is pushed to the matching child via set_prior, and the default/given priors are pushed to default_dist/given_dist.

Parameters:

prior (tuple[dict[Any, Any], Any, Any] | None)

Return type:

None

expected_log_density(x)

Prior-expected log-density: selected branch expected_log_density + given term.

Mirrors log_density: unmatched conditioning values with no default score -inf.

Parameters:

x (tuple[T0, T1])

Return type:

float

seq_expected_log_density(x)

Vectorized prior-expected log-density mirroring seq_log_density.

Parameters:

x (E0)

Return type:

ndarray

compute_capabilities()

Declare generated-compute support joined across branch, default, and given models.

compute_declaration()

Return the generated-compute declaration for the conditional distribution.

density(x)

Evaluates density of ConditionalDistribution at Tuple x.

Calls log_density() and returns the exponentiated result. See log_density() for details.

Parameters:

x (Tuple[T0, T1]) – T0 data type much match keys of dmap, T1 much match value of dmap distribution for key value.

Returns:

Density of ConditionalDistribution at Tuple x

Return type:

float

log_density(x)

Evaluate log-density of ConditionalDistribution at Tuple x.

Log-density:

log(P(x)) = log(P_cond(x[1] | x[0])) + log(P_given(x[0])), where log(P_cond(x[1] | x[0])) is defined from dmap, and log(P_given(x[0])) is defined from given_dist.

Note: Log-density is evaluated to -np.inf, if x[0] not in dmap and default_dist is NullDistribution().

Parameters:

x (Tuple[T0, T1]) – T0 data type much match keys of dmap, T1 much match value of dmap distribution for key value.

Returns:

Log-density of ConditionalDistribution at Tuple x.

Return type:

float

seq_log_density(x)

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

x Tuple of length 5:

x[0] (int): length of x (i.e. total observations). x[1] (Tuple[T0]): Unique conditional values in data. x[2] (Tuple[E0,…]): Tuple of encoded data sequences for each given key. x[3] (Tuple[ak.pdarray,…]): Tuple containing idxs for observation corresponding to x[1] values. x[4] (Optional[Encoded[T0]]): If the given_encoder is not the NullDataEncoder, the

observed conditional values of data type T0 are sequence encoded by given_encoder. Else return None.

Parameters:

x (E0) – See above for details.

Returns:

Numpy array of log-density evaluated at each encoded data point.

Return type:

ndarray

backend_seq_log_density(x, engine)

Engine-neutral vectorized log-density for grouped conditional encodings.

Parameters:

• x (E0)

• engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)

Return stacked child routes for homogeneous conditional mixtures.

Parameters:

• dists (Sequence[ConditionalDistribution])

• engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)

Return an (n, k) matrix of conditional log densities.

Parameters:

• x (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 conditional sufficient statistics.

Parameters:

• x (E0)

• weights (Any)

• params (dict[str, Any])

• engine (Any)

• estimator (Any)

Return type:

tuple[dict[Any, Any], Any, Any]

gradient_fit_state(engine, torch, leaves, recurse, tensor_param)

Return distribution-owned state for autograd fitting.

Parameters:

• engine (Any)

• torch (Any)

• leaves (list[Any])

• recurse (Any)

• tensor_param (Any)

Return type:

Any

sampler(seed=None)

Create a sampler for this conditional distribution.

Parameters:

seed (Optional[int]) – Seed for the sampler’s random number generator.

Returns:

ConditionalDistributionSampler configured from this distribution.

Return type:

ConditionalDistributionSampler

estimator(pseudo_count=None)

Create an estimator initialized from this conditional distribution.

Used to estimate a ConditionalDistribution from data observations.

Parameters:

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

Returns:

ConditionalDistributionEstimator with estimators for the branch, default, and given distributions.

Return type:

ConditionalDistributionEstimator

dist_to_encoder()

Create a data encoder for conditional-distribution observations.

Return type:

ConditionalDistributionDataEncoder

enumerator()

Creates a ConditionalDistributionEnumerator iterating (given, value) pairs in descending joint probability order.

Requires an enumerable given distribution and enumerable conditional distributions; raises EnumerationError otherwise.

Returns:

ConditionalDistributionEnumerator object.

Return type:

ConditionalDistributionEnumerator

class ConditionalDistributionEnumerator(dist)

Bases: DistributionEnumerator

Enumerates (given, value) pairs of a ConditionalDistribution in descending joint probability order.

Parameters:

dist (ConditionalDistribution)

class ConditionalDistributionSampler(dist, seed=None)

Bases: ConditionalSampler, DistributionSampler

ConditionalDistributionSampler draws (given, value) pairs from a ConditionalDistribution, or values conditioned on a fixed given value via sample_given().

Parameters:

• dist (ConditionalDistribution)

• seed (int | None)

single_sample()

Draw one (given, conditional) pair from the conditional distribution.

The first element is sampled from the given distribution; the second is sampled from the branch selected by that value, or from the default branch when the value is not present in dmap.

Returns:

Tuple[T0, T1] as defined from dmap and given_distribution types in dist (ConditionalDistribution instance).

Return type:

tuple[Any, Any]

sample(size=None)

Sample ‘size’ independent samples from ConditionalDistribution.

Sequence of ‘size’ calls to single_sample(). If size is None, size is taken to be 1.

Data type returned is a Tuple[T0, T1], where T0 and T1 are the respective data types of the given_dist and dmap defined in the CompositeDistribution instance ‘dist’.

Parameters:

size (Optional[int]) – Number of independent samples to draw from ConditionalDistribution.

Returns:

A list of ‘size’ tuples of Tuple[T0, T1], or a single Tuple[T0, T1].

Return type:

tuple[Any, Any] | list[tuple[Any, Any]]

sample_given(x)

Sample from the conditional distribution for a supplied given value.

Return data type T1 as defined for dictionary of ConditionalDistribution instance.

Parameters:

x (T0) – Value of given/conditional value for ConditionalDistribution.

Returns:

Single sample from ConditionalDistribution object ‘dist.dmap’ given x.

Return type:

Any

class ConditionalDistributionAccumulator(accumulator_map, default_accumulator=NullAccumulator(), given_accumulator=NullAccumulator(), keys=None)

Bases: SequenceEncodableStatisticAccumulator

ConditionalDistributionAccumulator accumulates sufficient statistics for each conditional distribution, the default distribution, and the given distribution.

Parameters:

• accumulator_map (dict[T0, SequenceEncodableStatisticAccumulator])

• default_accumulator (SequenceEncodableStatisticAccumulator | None)

• given_accumulator (SequenceEncodableStatisticAccumulator | None)

• keys (str | None)

update(x, weight, estimate)

Updates sufficient statistics of ConditionalDistributionAccumulator for one weighted observation x.

Single weighted observation used to update the sufficient statistics of ConditionalDistributionAccumulator.

Parameters:

• x (Tuple[T0, T1]) – Tuple observation of ConditionalDistribution.

• weight (float) – Weight for observation.

• estimate (Optional['ConditionalDistribution']) – Sufficient statistics from ConditionalDistribution are aggregated with weighted observation x.

Returns:

None.

Return type:

None

initialize(x, weight, rng)

Initialize ConditionalDistributionAccumulator with single weighted observation.

Note: _rng_initialize is called if _init_rng is False. This allows consistency between seq_initialize and initialize.

Parameters:

• x (Tuple[T0, T1]) – Tuple observation of ConditionalDistribution.

• weight (float) – Weight for observation.

• rng (RandomState) – RandomState used to set seed in initialize calls to member accumulators.

Returns:

None

Return type:

None

seq_initialize(x, weights, rng)

Vectorized initialize ConditionalDistributionAccumulator for a sequence encoded x.

Input x must be an encoded sequence produces from ConditionalDistributionDataEncoder.seq_encode() called on a list iid sequence of ConditionalDistribution observations.

Calls seq_initialize on accumulator_map, default_accumulator, and given_accumulator.

Note: _rng_initialize is called if _init_rng is False. This allows consistency between seq_initialize and initialize.

E Tuple of length 5:

E[0] (int): length of x (i.e. total observations). E[1] (Tuple[T0]): Unique conditional values in data. E[2] (Tuple[Encoded[T1]): Tuple of sequence encoded data of type T1 encoded by

encoder_map[key] or default_encoder if key not in default_encoder and default_encoder is not the NullDataEncoder.

E[3] (Tuple[np.ndarray,…]): Tuple of length equal to the number of unique conditional

values encountered in the data. Each entry contains a numpy array for the indices of x that correspond to a unique conditional value.

E[4] (Optional[Encoded[T0]]): If the given_encoder is not the NullDataEncoder, the

observed conditional values of data type T0 are sequence encoded by given_encoder. Else return None.

Parameters:

• x (E0) – See description above for details.

• weights (ndarray) – Numpy array of floats containing weights for each observation.

• rng (RandomState) – RandomState used to set seed in initialize calls to member accumulators.

Returns:

None.

Return type:

None

seq_update(x, weights, estimate)

Vectorized update of sufficient statistics of ConditionalDistributionAccumulator for a sequence encoded x.

Input x must be an encoded sequence produces from ConditionalDistributionDataEncoder.seq_encode() called on a list iid sequence of ConditionalDistribution observations.

Calls seq_update on accumulator_map, default_accumulator, and given_accumualtor.

E Tuple of length 5:

E[0] (int): length of x (i.e. total observations). E[1] (Tuple[T0]): Unique conditional values in data. E[2] (Tuple[Encoded[T1]): Tuple of sequence encoded data of type T1 encoded by

encoder_map[key] or default_encoder if key not in default_encoder and default_encoder is not the NullDataEncoder.

E[3] (Tuple[np.ndarray,…]): Tuple of length equal to the number of unique conditional

values encountered in the data. Each entry contains a numpy array for the indices of x that correspond to a unique conditional value.

E[4] (Optional[Encoded[T0]]): If the given_encoder is not the NullDataEncoder, the

observed conditional values of data type T0 are sequence encoded by given_encoder. Else return None.

Parameters:

• x (E0) – See description above for details.

• weights (ndarray) – Numpy array of floats containing weights for each observation.

• estimate (Optional['ConditionalDistribution']) – Sufficient statistics from ConditionalDistribution are used in merged with aggregated statistics from input x.

Returns:

None.

Return type:

None

seq_update_engine(x, weights, estimate, engine)

Engine-resident E-step: per-conditional-value subgroup weights are gathered on the active engine and the matching child accumulators (and the given accumulator) are routed through the engine. Matches seq_update.

Parameters:

• x (E0)

• weights (Any)

• estimate (ConditionalDistribution)

• engine (Any)

Return type:

None

combine(suff_stat)

Aggregate sufficient statistics (suff_stat) with sufficient statistics of ConditionalDistributionAccumulator

instance.

Parameters:

suff_stat (tuple[dict[T0, SS0], SS1 | None, SS2 | None]) – Tuple of length 3 containing the sufficient statistics of the conditional distributions, default distribution, and given distribution.

Returns:

ConditionalDistributionAccumulator with aggregated sufficient statistics.

Return type:

ConditionalDistributionAccumulator

value()

Get sufficient statistics of CompositeDistributionAccumulator.

Return type:

tuple[dict[Any, Any], Any | None, Any | None]

from_value(x)

Set ConditionalDistributionAccumulator member instances to x.

Input x must be sufficient statistic tuple compatible with ConditionalDistributionAccumulator.

Parameters:

x (tuple[dict[T0, SS0], SS1 | None, SS1 | None]) – Tuple of length 3 containing the sufficient statistics of ConditionalDistributionAccumulator.

Returns:

ConditionalDistributionAccumulator object.

Return type:

ConditionalDistributionAccumulator

scale(c)

Scale every branch, default, and given accumulator statistic in place.

Parameters:

c (float)

Return type:

ConditionalDistributionAccumulator

key_merge(stats_dict)

Aggregate the sufficient statistics of ConditionalDistributionAccumulator with member instance key in

stats_dict.

Parameters:

stats_dict (Dict[str, Any]) – Key of dict are the ‘keys’ for ConditionalDistributionAccumulator that represent the same distribution.

Returns:

None

Return type:

None

key_replace(stats_dict)

Invoke key_replace on each member SequenceEncodableStatisticAccumulator of

ConditionalDistributionAccumulator instance.

Parameters:

stats_dict (Dict[str, Any]) – Key of dict are the ‘keys’ for ConditionalDistributionAccumulator that represent the same distribution.

Returns:

None

Return type:

None

acc_to_encoder()

Create a data encoder from the branch accumulator encoders.

Return type:

ConditionalDistributionDataEncoder

class ConditionalDistributionAccumulatorFactory(factory_map, default_factory=NullAccumulatorFactory(), given_factory=NullAccumulatorFactory(), keys=None)

Bases: StatisticAccumulatorFactory

ConditionalDistributionAccumulatorFactory creates ConditionalDistributionAccumulator objects from the per-key, default, and given factories.

Parameters:

• factory_map (dict[T0, StatisticAccumulatorFactory])

• default_factory (StatisticAccumulatorFactory)

• given_factory (StatisticAccumulatorFactory)

• keys (str | None)

make()

Create a conditional accumulator from member factories.

SequenceEncodableStatisticAccumulator objects are created for accumulator_map, default_accumulator, and given_accumulator in ConditionalAccumulator object.

Returns:

ConditionalDistributionAccumulator

Return type:

ConditionalDistributionAccumulator

class ConditionalDistributionEstimator(estimator_map, default_estimator=NullEstimator(), given_estimator=NullEstimator(), name=None, keys=None, prior=None)

Bases: ParameterEstimator

ConditionalDistributionEstimator estimates a ConditionalDistribution from aggregated sufficient statistics.

Parameters:

• estimator_map (dict[T0, ParameterEstimator])

• default_estimator (ParameterEstimator | None)

• given_estimator (ParameterEstimator | None)

• name (str | None)

• keys (str | None)

• prior (tuple[dict[Any, Any], Any, Any] | None)

get_prior()

Return the joint prior as (per_branch_priors, default_prior, given_prior) from child estimators.

Return type:

tuple[dict[Any, Any], Any, Any]

set_prior(prior)

Distribute per-branch priors to the child estimators (branches, default, given).

prior=None is a no-op. Each branch prior is pushed to the matching estimator via set_prior; default/given priors go to the default/given estimators.

Parameters:

prior (tuple[dict[Any, Any], Any, Any] | None)

Return type:

None

model_log_density(model)

Sum each branch’s estimator model_log_density plus the default and given terms.

Parameters:

model (ConditionalDistribution)

Return type:

float

accumulator_factory()

Return an accumulator factory for the given, branch, and default estimators.

Return type:

ConditionalDistributionAccumulatorFactory

estimate(nobs, suff_stat)

Estimate a ConditionalDistribution from aggregated data.

Calls the estimate() member function of each ParameterEstimator instance for estimator_map, default_estimator, and given_estimator.

Input suff_stat if a Tuple of size three containing sufficient statistics compatible with each respective ParameterEstimator. Entry one of the Tuple must be a dict with keys of data type T0, matching the data type for the given distribution.

Return the conditional distribution estimated from the sufficient statistics in suff_stat.

Parameters:

• nobs (Optional[float]) – Not used. Kept for consistency.

• suff_stat (tuple[dict[T0, SS0], SS1 | None, SS2 | None]) – See description above.

Returns:

ConditionalDistribution object.

Return type:

ConditionalDistribution

class ConditionalDistributionDataEncoder(encoder_map, default_encoder=NullDataEncoder(), given_encoder=NullDataEncoder())

Bases: DataSequenceEncoder

ConditionalDistributionDataEncoder encodes sequences of (given, value) pairs, grouping the values by given value and delegating each group to the matching conditional encoder.

Parameters:

• encoder_map (dict[T0, DataSequenceEncoder])

• default_encoder (DataSequenceEncoder)

• given_encoder (DataSequenceEncoder)

seq_encode(x)

Encode sequence of iid observations from ConditionalDistribution for vectorized “seq_” function calls.

Data must be a List of Tuple of two types, T0 and T1. T0 is the data type compatible with the conditional values of the ConditionalDistribution. T1 must be consistent with the data type of the conditional distributions.

E Tuple of length 5:

E[0] (int): length of x (i.e. total observations). E[1] (Tuple[T0]): Unique conditional values in data. E[2] (Tuple[Encoded[T1]): Tuple of sequence encoded data of type T1 encoded by

encoder_map[key] or default_encoder if key not in default_encoder and default_encoder is not the NullDataEncoder.

E[3] (Tuple[np.ndarray,…]): Tuple of length equal to the number of unique conditional

values encountered in the data. Each entry contains a numpy array for the indices of x that correspond to a unique conditional value.

E[4] (Optional[Encoded[T0]]): If the given_encoder is not the NullDataEncoder, the

observed conditional values of data type T0 are sequence encoded by given_encoder. Else return None.

Parameters:

x (List[Tuple[T0, T1]]) – List of data observations.

Returns:

Returns rv (see description for details)

Return type:

tuple[int, tuple[Any, …], tuple[Any, …], tuple[ndarray, …], Any | None]

ConditionalAccumulator

alias of ConditionalDistributionAccumulator

ConditionalAccumulatorFactory

alias of ConditionalDistributionAccumulatorFactory

ConditionalDataEncoder

alias of ConditionalDistributionDataEncoder

ConditionalEnumerator

alias of ConditionalDistributionEnumerator

ConditionalEstimator

alias of ConditionalDistributionEstimator