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.

exception TransformDomainError

Bases: ValueError

Raised when an observation lies outside a transform’s declared domain.

exception TransformCompatibilityError

Bases: TypeError

Raised when a transform contract is incompatible with its child distribution.

class TransformContract(input_domain, output_domain, measure='adaptive', bijective=True)

Bases: object

Machine-checkable declaration for a deterministic bijective transform.

input_domain and output_domain use the same coarse vocabulary as compute-declaration support tags. measure is "adaptive" when the transform supports both counting measure and continuous change-of-variables, or "discrete" for a counting-measure-only mapping.

Parameters:

• input_domain (str)

• output_domain (str)

• measure (str)

• bijective (bool)

class TransformFitReceipt(accepted_weight, rejected_weight, rejected_fraction)

Bases: NamedTuple

Accepted/rejected evidence accounting attached to a fitted transform wrapper.

Parameters:

• accepted_weight (float)

• rejected_weight (float)

• rejected_fraction (float)

accepted_weight: float

Alias for field number 0

rejected_weight: float

Alias for field number 1

rejected_fraction: float

Alias for field number 2

class TransformStatistics(schema_version, child, accepted_weight, rejected_weight)

Bases: NamedTuple

Versioned child statistics plus transform-domain evidence accounting.

Parameters:

• schema_version (int)

• child (Any)

• accepted_weight (float)

• rejected_weight (float)

schema_version: int

Alias for field number 0

child: Any

Alias for field number 1

accepted_weight: float

Alias for field number 2

rejected_weight: float

Alias for field number 3

class IdentityTransform

Bases: object

Identity transform y = x.

forward(x)

Return x unchanged.

Parameters:

x (Any)

Return type:

Any

inverse(y)

Return y unchanged.

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)

Return the log absolute inverse-Jacobian determinant.

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()

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

Return type:

float

class AffineTransform(loc=0.0, scale=1.0)

Bases: object

Affine transform y = loc + scale * x.

Parameters:

• loc (float)

• scale (float)

forward(x)

Apply the affine map to a child-space value.

Parameters:

x (Any)

Return type:

Any

inverse(y)

Map a transformed-space value back to child space.

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)

Return the constant affine inverse-Jacobian correction.

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()

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

Return type:

float

class ExpTransform

Bases: object

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

forward(x)

Map a real value to the positive scale.

Parameters:

x (Any)

Return type:

Any

inverse(y)

Map a positive transformed value back to the real line.

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)

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

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()

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

Return type:

float

class LogTransform

Bases: object

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

forward(x)

Map a positive child value to the real line.

Parameters:

x (Any)

Return type:

Any

inverse(y)

Map a real transformed value back to the positive scale.

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)

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

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()

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

Return type:

float

class LogitTransform

Bases: object

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

forward(x)

Map a real value into the open unit interval.

Parameters:

x (Any)

Return type:

Any

inverse(y)

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

Parameters:

y (Any)

Return type:

Any

log_abs_det_inverse_jacobian(y)

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

Parameters:

y (Any)

Return type:

float

invalid_inverse_value()

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, fit_receipt=None)

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)

• fit_receipt (TransformFitReceipt | None)

mean()

Exact mean under an affine map: loc + scale * E[X].

An affine map is the one transform whose moments are exact in closed form, and the PPL’s .mean()/.var() fall back to Monte Carlo when a distribution defines neither – so 3 * Normal(0, 1) + 1 reported a sampled 1.02 and 8.95 beside prose calling the map exact (P08-F14). Every other transform has no closed form here and says so, which is what sends the caller back to the sampled estimate.

variance()

Exact variance under an affine map: scale**2 * Var[X].

compute_capabilities()

Return capabilities delegated from the child distribution where safe.

compute_declaration()

Return a declaration describing this distribution as a transformed child.

density(x)

Return the probability density or mass at a single observation.

Parameters:

x (Any)

Return type:

float

log_density(x)

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

Parameters:

x (Any)

Return type:

float

seq_log_density(x)

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

Parameters:

x (tuple[Any, ndarray, ndarray])

Return type:

ndarray

backend_seq_log_density(x, engine)

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

Parameters:

• x (tuple[Any, ndarray, ndarray])

• engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)

Return stacked child parameters for homogeneous fixed-transform mixtures.

Parameters:

• dists (Sequence[TransformDistribution])

• engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)

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

Parameters:

• x (tuple[Any, ndarray, ndarray])

• params (dict[str, Any])

• engine (Any)

Return type:

Any

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

Return distribution-owned state for autograd fitting.

Parameters:

• engine (Any)

• torch (Any)

• leaves (Any)

• recurse (Any)

• tensor_param (Any)

Return type:

Any

sampler(seed=None)

Return a sampler for drawing observations from this distribution.

Parameters:

seed (int | None)

Return type:

TransformSampler

estimator(pseudo_count=None)

Return an estimator for fitting this distribution from data.

Parameters:

pseudo_count (float | None)

Return type:

TransformEstimator

dist_to_encoder()

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

Return type:

TransformDataEncoder

enumerator()

Return an enumerator over the distribution support when available.

Return type:

TransformEnumerator

class TransformEnumerator(dist)

Bases: DistributionEnumerator

Enumerate transformed child support for discrete child distributions.

Parameters:

dist (TransformDistribution)

class TransformSampler(dist, seed=None)

Bases: DistributionSampler

Sampler that transforms draws from the child distribution.

Parameters:

• dist (TransformDistribution)

• seed (int | None)

sample(size=None, *, batched=True)

Draw child samples and map them through the transform.

Parameters:

• size (int | None)

• batched (bool)

class TransformAccumulator(accumulator, transform, density_correction=None, name=None, keys=None)

Bases: SequenceEncodableStatisticAccumulator

Accumulator that delegates inverse-transformed observations to the child.

Parameters:

• accumulator (SequenceEncodableStatisticAccumulator)

• transform (Any)

• density_correction (bool | None)

• name (str | None)

• keys (str | None)

update(x, weight, estimate)

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)

Accumulate a batch using validity-masked child weights.

Parameters:

• x (tuple[Any, ndarray, ndarray])

• weights (ndarray)

• estimate (TransformDistribution | None)

Return type:

None

seq_update_engine(x, weights, estimate, engine)

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:

• x (tuple[Any, ndarray, ndarray])

• weights (Any)

• estimate (TransformDistribution | None)

• engine (Any)

Return type:

None

initialize(x, weight, rng)

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

Parameters:

• x (Any)

• weight (float)

• rng (RandomState | None)

Return type:

None

seq_initialize(x, weights, rng)

Initialize from a validity-masked encoded batch.

Parameters:

• x (tuple[Any, ndarray, ndarray])

• weights (ndarray)

• rng (RandomState | None)

Return type:

None

combine(suff_stat)

Merge child sufficient statistics and evidence accounting.

Parameters:

suff_stat (TransformStatistics)

Return type:

TransformAccumulator

value()

Return versioned child statistics and transform-domain evidence accounting.

Return type:

TransformStatistics

from_value(x)

Restore child statistics and evidence accounting.

Parameters:

x (TransformStatistics)

Return type:

TransformAccumulator

scale(c)

Scale delegated sufficient statistics and evidence weights by c.

Parameters:

c (float)

Return type:

TransformAccumulator

key_merge(stats_dict)

Pool the wrapper’s full child statistics under its own key, or delegate when unkeyed.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)

Replace from the wrapper pool, or delegate when this wrapper is unkeyed.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()

Return the encoder associated with this accumulator.

Return type:

TransformDataEncoder

class TransformAccumulatorFactory(factory, transform, density_correction=None, name=None, keys=None)

Bases: StatisticAccumulatorFactory

Factory for TransformAccumulator.

Parameters:

• factory (StatisticAccumulatorFactory)

• transform (Any)

• density_correction (bool | None)

• name (str | None)

• keys (str | None)

make()

Create a fresh transform accumulator.

Return type:

TransformAccumulator

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

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

Return the accumulator factory for inverse-transformed observations.

Return type:

TransformAccumulatorFactory

estimate(nobs, suff_stat)

Estimate from accepted evidence and attach the rejected-weight receipt.

Parameters:

• nobs (float | None)

• suff_stat (Any)

Return type:

TransformDistribution

class TransformDataEncoder(encoder, transform, density_correction=True)

Bases: DataSequenceEncoder

Encode transformed observations as inverse child data plus Jacobian terms.

Parameters:

• encoder (DataSequenceEncoder)

• transform (Any)

• density_correction (bool | None)

seq_encode(x)

Encode observations as inverse child values, Jacobians, and validity flags.

Parameters:

x (Sequence[Any])

Return type:

tuple[Any, ndarray, ndarray]