mixle.stats.combinator.conditional module

Create, estimate, and sample from a Conditional distribution.

Defines the ConditionalDistribution, ConditionalDistributionSampler, ConditionalDistributionAccumulatorFactory, ConditionalDistributionAccumulator, ConditionalDistributionEstimator, and the ConditionalDistributionDataEncoder classes for use with mixle.

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

compute_declaration()

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)

Creates ConditionalDistributionSampler object for sampling from ConditionalDistribution instance.

Parameters:

seed (Optional[int]) – Set seed for sampling from ConditionalDistributionSampler object.

Returns:

ConditionalDistributionSampler object.

Return type:

ConditionalDistributionSampler

estimator(pseudo_count=None)

Creates ConditionalDistributionEstimator object from sufficient statistics of ConditionalDistribution object.

Used to estimate a ConditionalDistribution from data observations.

Parameters:

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

Returns:

ConditionalDistributionEstimator object.

Return type:

ConditionalDistributionEstimator

dist_to_encoder()

Creates ConditionalDistributionDataEncoder object for encoding sequences of ConditionalDistribution data.

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

Generates a simple sample from the ConditionalDistribution.

Returns Tuple of T0 and T1, where T1 is the data type of the conditional distribution, and T0 is the type of the given distribution.

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 conditional distribution of ConditionalDistribution object with given value x.

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 linear sufficient statistics in-place by c.

The structural default is correct for ordinary weighted sums, nested tuples/lists/dicts, and numeric arrays. Families whose value() payload includes non-linear metadata such as support bounds must override this method and leave that metadata unscaled.

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

Creates ConditionalDistributionDataEncoder object for encoding sequences of ConditionalDistribution data.

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 ConditionalAccumulator object from StatisticAccumulatorFactory member instances.

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

Creates ConditionalDistributionAccumulatorFactory from estimator member values.

Returns:

ConditionalDistributionAccumulatorFactory object.

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.

Returns a ConditionalDistribution object 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