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)

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)

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

Consume one batch and return the updated model.

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 subtracts that chunk’s previous payload, adds the new payload, and runs the ordinary estimator M-step on the pooled statistics. No distribution-specific estimation code lives here; the class only uses scale(-1), combine(), and estimate().

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.

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