mixle.stats.combinator.finite_stochastic_transform module

Finite stochastic-transform combinator: push a finite discrete source through a noisy channel.

FiniteStochasticTransformDistribution models Y | X = x ~ Categorical(kernel[x, :]) where the source X ~ dist ranges over {0, ..., m-1} and only the output Y in {0, ..., n-1} is observed. It is the stochastic analogue of TransformDistribution: the transform is a fixed m x n row-stochastic kernel (a confusion matrix / discrete channel) rather than a deterministic invertible map, so the source marginal is

P(Y = y) = sum_x P(X = x) * kernel[x, y].

Estimation recovers the source dist from observed Y without inverting the channel. The naive route – solve kernel^T p_X = P_hat(Y) for p_X (an m x n (pseudo-)inverse, O(m^3), ill-conditioned, and blind to the simplex / to a structured source) – is short-circuited two ways:

  1. The data’s only sufficient statistic is the length-n vector of output counts, so the work is collapsed from the N observations to the n distinct outputs up front (one bincount).

  2. The deconvolution is then a single E-step on those counts: the posterior responsibilities R[x, y] = P(X = x) kernel[x, y] / P(Y = y) give the expected source counts c = R @ n_y by a matrix-vector product (O(n*m), independent of N), which are fed straight to the source’s own estimator. Iterating fit is EM / Richardson-Lucy deconvolution – monotone, constraint-aware, and reusing whatever (possibly structured) estimator the source provides.

class FiniteStochasticTransformDistribution(dist, kernel, name=None, keys=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Finite discrete source pushed through a fixed finite stochastic kernel (noisy channel).

Parameters:

  • dist (SequenceEncodableProbabilityDistribution)

  • kernel (Any)

  • name (str | None)

  • keys (str | None)

density(x)[source]

Return P(Y = x).

Parameters:

x (int)

Return type:

float

log_density(x)[source]

Return log P(Y = x) for an integer output x in {0, ..., n-1}.

Parameters:

x (int)

Return type:

float

seq_log_density(x)[source]

Return per-observation log P(Y = y) for an encoded batch of outputs.

Parameters:

x (tuple[ndarray, ndarray])

Return type:

ndarray

support_size()[source]

Number of finite outputs {0, ..., n-1}.

Return type:

int

sampler(seed=None)[source]

Return a sampler drawing X ~ dist then Y ~ Categorical(kernel[X]).

Parameters:

seed (int | None)

Return type:

FiniteStochasticTransformSampler

estimator(pseudo_count=None)[source]

Return an estimator that recovers the source via the channel-inversion-free E-step.

Parameters:

pseudo_count (float | None)

Return type:

FiniteStochasticTransformEstimator

dist_to_encoder()[source]

Return the data encoder for integer outputs.

Return type:

FiniteStochasticTransformDataEncoder

enumerator()[source]

Enumerate the finite outputs y in descending P(Y = y) order.

Return type:

FiniteStochasticTransformEnumerator

class FiniteStochasticTransformEnumerator(dist)[source]

Bases: DistributionEnumerator

Enumerate the finite output support in descending marginal probability.

Parameters:

dist (FiniteStochasticTransformDistribution)

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

Bases: DistributionSampler

Sample X from the source, then Y from that source state’s kernel row.

Parameters:

  • dist (FiniteStochasticTransformDistribution)

  • seed (int | None)

sample(size=None)[source]

Draw one output (or a list of size outputs).

Parameters:

size (int | None)

class FiniteStochasticTransformAccumulator(source_accumulator, kernel, keys=None)[source]

Bases: SingleChildAccumulator

Accumulate expected source counts via one channel E-step over aggregated output counts.

Parameters:

  • source_accumulator (SequenceEncodableStatisticAccumulator)

  • kernel (ndarray)

  • keys (str | None)

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

  • x (Any)

  • weight (float)

  • estimate (FiniteStochasticTransformDistribution | None)

Return type:

None

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

None

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

None

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

None

acc_to_encoder()[source]
Return type:

FiniteStochasticTransformDataEncoder

class FiniteStochasticTransformAccumulatorFactory(source_factory, kernel, keys=None)[source]

Bases: StatisticAccumulatorFactory

Factory for FiniteStochasticTransformAccumulator.

Parameters:

  • source_factory (StatisticAccumulatorFactory)

  • kernel (ndarray)

  • keys (str | None)

make()[source]
Return type:

FiniteStochasticTransformAccumulator

class FiniteStochasticTransformEstimator(source_estimator, kernel, name=None, keys=None)[source]

Bases: ParameterEstimator

Estimator for a fixed-kernel finite stochastic transform (recovers the source).

Parameters:

  • source_estimator (ParameterEstimator)

  • kernel (Any)

  • name (str | None)

  • keys (str | None)

accumulator_factory()[source]
Return type:

FiniteStochasticTransformAccumulatorFactory

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

FiniteStochasticTransformDistribution

class FiniteStochasticTransformDataEncoder(num_output)[source]

Bases: DataSequenceEncoder

Encode integer outputs as an index array plus an in-range validity mask.

Parameters:

num_output (int)

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Any])

Return type:

tuple[ndarray, ndarray]