mixle.stats.combinator.transform module¶

Invertible-transform wrappers for sequence-encodable distributions.

The module implements identity, affine, exponential, logit, and custom transforms plus the distribution, sampler, accumulator, and encoder plumbing needed to apply them inside Mixle combinators.

class IdentityTransform[source]

Bases: object

Identity transform y = x.

forward(x)[source]

Return x unchanged.

Parameters:

x (Any)

Return type:

Any

inverse(y)[source]

Return y unchanged.

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]

Return the log absolute inverse-Jacobian determinant.

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]

Return a safe child-space fill value for invalid inverses.

Return type:

float

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

Bases: object

Affine transform y = loc + scale * x.

Parameters:
forward(x)[source]

Apply the affine map to a child-space value.

Parameters:

x (Any)

Return type:

Any

inverse(y)[source]

Map a transformed-space value back to child space.

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]

Return the constant affine inverse-Jacobian correction.

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]

Return a safe child-space fill value for invalid inverses.

Return type:

float

class ExpTransform[source]

Bases: object

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

forward(x)[source]

Map a real value to the positive scale.

Parameters:

x (Any)

Return type:

Any

inverse(y)[source]

Map a positive transformed value back to the real line.

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]

Return the log inverse-Jacobian correction for log(y).

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]

Return a safe child-space fill value for invalid inverses.

Return type:

float

class LogTransform[source]

Bases: object

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

forward(x)[source]

Map a positive child value to the real line.

Parameters:

x (Any)

Return type:

Any

inverse(y)[source]

Map a real transformed value back to the positive scale.

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]

Return the log inverse-Jacobian correction for exp(y).

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]

Return a safe positive child-space fill value for invalid inverses.

Return type:

float

class LogitTransform[source]

Bases: object

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

forward(x)[source]

Map a real value into the open unit interval.

Parameters:

x (Any)

Return type:

Any

inverse(y)[source]

Map a unit-interval value back to the real line.

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)[source]

Return the log inverse-Jacobian correction for the logit map.

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()[source]

Return a safe child-space fill value for invalid inverses.

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]

Return capabilities delegated from the child distribution where safe.

compute_declaration()[source]

Return a declaration describing this distribution as a transformed child.

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 child samples and map them through the transform.

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]

Accumulate one inverse-transformed observation when it is valid.

Parameters:

  • x (Any)

  • weight (float)

  • estimate (TransformDistribution | None)

Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate a batch using validity-masked child weights.

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]

Initialize from one inverse-transformed observation when it is valid.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize from a validity-masked encoded batch.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge child sufficient statistics.

Parameters:

suff_stat (Any)

Return type:

TransformAccumulator

value()[source]

Return the child accumulator’s serialized sufficient statistics.

Return type:

Any

from_value(x)[source]

Restore the child accumulator from serialized sufficient statistics.

Parameters:

x (Any)

Return type:

TransformAccumulator

scale(c)[source]

Scale delegated sufficient statistics by c.

Parameters:

c (float)

Return type:

TransformAccumulator

key_merge(stats_dict)[source]

Delegate keyed statistic merging to the child accumulator.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)[source]

Delegate keyed statistic replacement to the child accumulator.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()[source]

Return the encoder associated with this accumulator.

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]

Create a fresh transform accumulator.

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 the accumulator factory for inverse-transformed observations.

Return type:

TransformAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate the child distribution and wrap it with the fixed transform.

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 observations as inverse child values, Jacobians, and validity flags.

Parameters:

x (Sequence[Any])

Return type:

tuple[Any, ndarray, ndarray]