Implementation:Online ml River Stats Minimum

Knowledge Sources	Online_ml_River
Domains	Online_Learning, Statistics
Last Updated	2026-02-08 16:00 GMT

Overview

Minimum tracks the running minimum value observed in a data stream.

Description

This statistic maintains the smallest value seen so far in a streaming dataset. The implementation includes two variants: Min for the overall minimum across all observations, and RollingMin for tracking the minimum value within a sliding window. The basic Min class updates in constant time O(1), while RollingMin uses a sorted window structure for efficient windowed operations.

Usage

Use Minimum when you need to track the smallest value in streaming data. Common applications include monitoring lower bounds in sensor data, tracking minimum prices in financial data, finding extreme values for normalization and scaling, quality control, and detecting anomalies. The rolling variant is useful when only recent minimum values are relevant.

Code Reference

Source Location

• Repository: Online_ml_River
• File: river/stats/minimum.py

Signature

class Min(stats.base.Univariate):
    def __init__(self):
        self.min = math.inf

class RollingMin(stats.base.RollingUnivariate):
    def __init__(self, window_size: int):
        self.window = utils.SortedWindow(size=window_size)

Import

from river import stats

I/O Contract

Inputs

Name	Type	Required	Description
x	numbers.Number	Yes	Value to update the statistic with
window_size	int	Yes (for Rolling)	Size of the rolling window

Outputs

Name	Type	Description
get()	float	Current minimum value (or None for rolling if window not filled)

Usage Examples

from river import stats

# Running minimum
X = [1, -4, 3, -2, 5, -6]
minimum = stats.Min()

for x in X:
    minimum.update(x)
    print(f"Value: {x}, Min: {minimum.get()}")

# Output:
# Value: 1, Min: 1
# Value: -4, Min: -4
# Value: 3, Min: -4
# Value: -2, Min: -4
# Value: 5, Min: -4
# Value: -6, Min: -6

# Rolling minimum
X = [1, -4, 3, -2, 2, 1]
rolling_min = stats.RollingMin(2)

for x in X:
    rolling_min.update(x)
    print(f"Value: {x}, Rolling Min: {rolling_min.get()}")

# Output:
# Value: 1, Rolling Min: 1
# Value: -4, Rolling Min: -4
# Value: 3, Rolling Min: -4
# Value: -2, Rolling Min: -2
# Value: 2, Rolling Min: -2
# Value: 1, Rolling Min: 1

# Using min for data normalization
min_val = stats.Min()
max_val = stats.Max()

data = [10, 25, 15, 30, 5, 20]
for x in data:
    min_val.update(x)
    max_val.update(x)

# Normalize data to [0, 1]
min_v = min_val.get()
max_v = max_val.get()
normalized = [(x - min_v) / (max_v - min_v) for x in data]
print(f"Normalized: {[f'{n:.2f}' for n in normalized]}")

Related Pages

• Environment:Online_ml_River_Python_Runtime_Environment