BatchStats

BatchStats computes statistics on data that arrives in batches, so you can stream or process large datasets without loading everything into memory. Its incremental algorithms expose a small NumPy-friendly API and support merging independently computed accumulators.

BatchStats requires Python 3.10 or newer.

Installation

pip install batchstats

Or with conda/mamba:

conda install -c conda-forge batchstats

Quick Start

import numpy as np
from batchstats import BatchMean, BatchVar

rng = np.random.default_rng(0)
data_stream = (rng.standard_normal((100, 10)) for _ in range(10))

batch_mean = BatchMean()
batch_var = BatchVar()

for batch in data_stream:
    batch_mean.update_batch(batch)
    batch_var.update_batch(batch)

mean = batch_mean()
variance = batch_var()

print(f"Mean shape: {mean.shape}")
print(f"Variance shape: {variance.shape}")

Available Statistics

• BatchSum / BatchNanSum
• BatchWeightedSum
• BatchMean / BatchNanMean
• BatchWeightedMean
• BatchMin / BatchNanMin
• BatchMax / BatchNanMax
• BatchPeakToPeak / BatchNanPeakToPeak
• BatchVar
• BatchStd
• BatchCov
• BatchCorr
• BatchTopK / BatchNanTopK (exact extreme ranks and linear tail quantiles)

BatchNanSum and BatchNanMean support merging with +. These classes and the top-k accumulators support versioned to_state() / from_state() and NPZ checkpoints with save() / load(), without pickle.

from batchstats import BatchNanTopK, required_k

extremes = BatchNanTopK(required_k(0.998, 8784), axis=0)
for block in data_stream:  # (time, *spatial_shape), same spatial positions
    extremes.update_batch(block)
p998 = extremes.quantile(0.998)  # raises if retained capacity is insufficient
extremes.save("extremes.npz")

Top-k outputs have a leading rank axis and mask unavailable ranks while preserving the selected dtype. BatchTopK rejects NaNs; BatchNanTopK ignores them per cell. See the streaming and checkpoint guide.

Docs: https://batchstats.readthedocs.io

Development

Install the development dependencies and run the local quality gates:

python -m pip install -e ".[dev]"
python -m ruff check .
python -m ruff format --check .
python -m pytest --cov=batchstats
python -m build
python -m twine check dist/*