mixle.stats.combinator.transform module

Fixed invertible-transform wrapper for sequence-encodable distributions.

class IdentityTransform[source]

Bases: object

Identity transform y = x.

forward(x)[source]
Parameters:

x (Any)

Return type:

Any

inverse(y)[source]
Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]
Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]
Return type:

float

class AffineTransform(loc=0.0, scale=1.0)[source]

Bases: object

Affine transform y = loc + scale * x.

Parameters:
forward(x)[source]
Parameters:

x (Any)

Return type:

Any

inverse(y)[source]
Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]
Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]
Return type:

float

class ExpTransform[source]

Bases: object

Exponential transform y = exp(x), mapping real x to positive y.

forward(x)[source]
Parameters:

x (Any)

Return type:

Any

inverse(y)[source]
Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]
Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]
Return type:

float

class LogTransform[source]

Bases: object

Log transform y = log(x), mapping positive x to real y.

forward(x)[source]
Parameters:

x (Any)

Return type:

Any

inverse(y)[source]
Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]
Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]
Return type:

float

class LogitTransform[source]

Bases: object

Logistic transform y = 1 / (1 + exp(-x)), mapping real x to (0, 1).

forward(x)[source]
Parameters:

x (Any)

Return type:

Any

inverse(y)[source]
Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]
Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]
Return type:

float

class TransformDistribution(dist, transform=None, density_correction=None, name=None, keys=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Push a child distribution through a fixed invertible transform.

Observations live in transformed space. For fixed continuous transforms, log-density uses the inverse transform and adds the inverse-Jacobian term. The transform is not learned; estimation inverse-transforms observations and delegates sufficient statistics to the child estimator.

Parameters:

  • dist (SequenceEncodableProbabilityDistribution)

  • transform (Any | None)

  • density_correction (bool | None)

  • name (str | None)

  • keys (str | None)

compute_capabilities()[source]
compute_declaration()[source]
density(x)[source]

Return the probability density or mass at a single observation.

Parameters:

x (Any)

Return type:

float

log_density(x)[source]

Return the log-density or log-mass at a single observation.

Parameters:

x (Any)

Return type:

float

seq_log_density(x)[source]

Return vectorized log-density values for sequence-encoded observations.

Parameters:

x (tuple[Any, ndarray, ndarray])

Return type:

ndarray

backend_seq_log_density(x, engine)[source]

Engine-neutral vectorized log-density for inverse-encoded observations.

Parameters:
Return type:

Any

classmethod backend_stacked_params(dists, engine)[source]

Return stacked child parameters for homogeneous fixed-transform mixtures.

Parameters:
Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)[source]

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

Parameters:
Return type:

Any

classmethod backend_stacked_sufficient_statistics_with_estimator(x, weights, params, engine, estimator)[source]

Return child legacy statistics for valid inverse-transformed observations.

Parameters:
Return type:

Any

gradient_fit_state(engine, torch, leaves, recurse, tensor_param)[source]

Return distribution-owned state for autograd fitting.

Parameters:
Return type:

Any

sampler(seed=None)[source]

Return a sampler for drawing observations from this distribution.

Parameters:

seed (int | None)

Return type:

TransformSampler

estimator(pseudo_count=None)[source]

Return an estimator for fitting this distribution from data.

Parameters:

pseudo_count (float | None)

Return type:

TransformEstimator

dist_to_encoder()[source]

Return the data encoder used by this distribution for vectorized methods.

Return type:

TransformDataEncoder

enumerator()[source]

Return an enumerator over the distribution support when available.

Return type:

TransformEnumerator

class TransformEnumerator(dist)[source]

Bases: DistributionEnumerator

Enumerate transformed child support for discrete child distributions.

Parameters:

dist (TransformDistribution)

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

Bases: DistributionSampler

Sampler that transforms draws from the child distribution.

Parameters:

  • dist (TransformDistribution)

  • seed (int | None)

sample(size=None)[source]

Draw observations.

Combinator samplers (mixture/sequence/…) accept batched. With batched=True (the default) each child stream is drawn in one vectorized call instead of a per-draw Python loop – far faster. Because every child sampler owns an independent RandomState, batching consumes each stream in the same order as the loop, so the draws are identical to the legacy path. batched=False forces that legacy per-draw loop as a guaranteed- stable reference. Leaf samplers are already vectorized and ignore the flag.

Parameters:

size (int | None)

class TransformAccumulator(accumulator, transform, density_correction=None, name=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator that delegates inverse-transformed observations to the child.

Parameters:

  • accumulator (SequenceEncodableStatisticAccumulator)

  • transform (Any)

  • density_correction (bool | None)

  • name (str | None)

update(x, weight, estimate)[source]
Parameters:

  • x (Any)

  • weight (float)

  • estimate (TransformDistribution | None)

Return type:

None

seq_update(x, weights, estimate)[source]
Parameters:
Return type:

None

seq_update_engine(x, weights, estimate, engine)[source]

Engine-resident E-step: the validity-masked weights are formed on the active engine and the child accumulator is routed through the engine. Matches seq_update.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]
Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]
Parameters:
Return type:

None

combine(suff_stat)[source]
Parameters:

suff_stat (Any)

Return type:

TransformAccumulator

value()[source]
Return type:

Any

from_value(x)[source]
Parameters:

x (Any)

Return type:

TransformAccumulator

scale(c)[source]

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:

TransformAccumulator

key_merge(stats_dict)[source]

Pool this accumulator’s statistics into stats_dict under its merge key.

The structural default implements the common single-key pattern: store the accumulator under self.keys the first time the key is seen, else combine into the one already there. Accumulators with several named keys (e.g. an HMM’s init/trans/state keys) or a non-accumulator stats payload override this. A keys of None (the default) is a no-op.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)[source]

Replace this accumulator’s statistics from the pooled stats_dict entry (see key_merge).

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()[source]
Return type:

TransformDataEncoder

class TransformAccumulatorFactory(factory, transform, density_correction=None, name=None)[source]

Bases: StatisticAccumulatorFactory

Factory for TransformAccumulator.

Parameters:

  • factory (StatisticAccumulatorFactory)

  • transform (Any)

  • density_correction (bool | None)

  • name (str | None)

make()[source]
Return type:

TransformAccumulator

class TransformEstimator(estimator, transform=None, density_correction=None, name=None, keys=None)[source]

Bases: ParameterEstimator

Estimator for fixed-transform distributions.

Parameters:

  • estimator (ParameterEstimator)

  • transform (Any | None)

  • density_correction (bool | None)

  • name (str | None)

  • keys (str | None)

accumulator_factory()[source]
Return type:

TransformAccumulatorFactory

estimate(nobs, suff_stat)[source]
Parameters:
Return type:

TransformDistribution

class TransformDataEncoder(encoder, transform, density_correction=True)[source]

Bases: DataSequenceEncoder

Encode transformed observations as inverse child data plus Jacobian terms.

Parameters:

  • encoder (DataSequenceEncoder)

  • transform (Any)

  • density_correction (bool | None)

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Any])

Return type:

tuple[Any, ndarray, ndarray]