mixle.inference.streaming module

Online / streaming estimation over batched sufficient statistics.

streaming_accumulate folds one batch’s sufficient statistics; StreamingEstimator and IncrementalEstimator drive incremental fitting across a stream of batches (with optional forgetting/step schedules). Distinct from the Bayesian BayesianStreamingEstimator in estimation.py.

streaming_accumulate(enc_data, estimator, model)

Return one batch’s globally tied sufficient-stat accumulator.

Encoded-data handles can implement pysp_stream_accumulate to do the local/distributed fold themselves. Plain encoded chunks use the legacy in-process seq_update loop.

Parameters:

• enc_data (Any)

• estimator (ParameterEstimator)

• model (SequenceEncodableProbabilityDistribution)

Return type:

tuple[float, Any]

class StreamingEstimator(estimator, schedule=None, model=None, init_estimator=None, init_p=0.1, rng=None, encoder=None, num_chunks=1, dataset_size=None)

Bases: _StreamingBase

Decay-mode online estimator built from accumulator scaling and M-steps.

Parameters:

• estimator (ParameterEstimator)

• model (SequenceEncodableProbabilityDistribution | None)

• init_estimator (ParameterEstimator | None)

• init_p (float)

• rng (RandomState | None)

• num_chunks (int)

• dataset_size (float | None)

update(data=None, *, enc_data=None)

Consume one batch and return the updated model.

Every scale and count is validated finite before it touches the running accumulator, and the running state is committed only once the M-step has succeeded – so a failed update leaves the estimator exactly where it was and the caller can retry.

Parameters:

• data (Sequence[T] | None)

• enc_data (list[tuple[int, E0]] | None)

Return type:

SequenceEncodableProbabilityDistribution

class IncrementalEstimator(estimator, model=None, init_estimator=None, init_p=0.1, rng=None, encoder=None, num_chunks=1)

Bases: _StreamingBase

Neal-Hinton style incremental EM over replaceable data chunks.

Each chunk contributes a sufficient-statistic payload computed under the current model. Revisiting a chunk replaces that payload and rebuilds the pooled accumulator from the stored chunk summaries before running the ordinary M-step. Rebuilding is essential because mergeable statistics form a monoid, not necessarily a group: sums can be subtracted, but support minima/maxima, weighted medians, sketches, and similar summaries cannot.

Parameters:

• estimator (ParameterEstimator)

• model (SequenceEncodableProbabilityDistribution | None)

• init_estimator (ParameterEstimator | None)

• init_p (float)

• rng (RandomState | None)

• num_chunks (int)

update(data=None, *, enc_data=None, chunk_id=None)

Replace one chunk contribution and return the updated model.

chunk_id is keyword-only so this matches StreamingEstimator.update()’s (data, *, enc_data) shape across the streaming surface; it is required (a None chunk_id raises) because the Neal-Hinton update keys each batch’s contribution by it.

The chunk store, running statistics and model are committed only once the M-step has succeeded, so a failed update leaves the estimator exactly where it was – in particular the previous payload for a revisited chunk_id survives – and the caller can retry.

Parameters:

• data (Sequence[T] | None)

• enc_data (list[tuple[int, E0]] | None)

• chunk_id (Any)

Return type:

SequenceEncodableProbabilityDistribution

chunk_value(chunk_id)

Return a copy of one stored chunk contribution.

Parameters:

chunk_id (Any)

reset()

Drop all chunk contributions and fitted model state.

Return type:

None