Implementation:Online ml River Stats Maximum

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

Overview

Maximum tracks the running maximum value observed in a data stream.

Description

This statistic maintains the largest value seen so far in a streaming dataset. The implementation includes multiple variants: Max for the overall maximum, RollingMax for the maximum over a sliding window, AbsMax for the maximum absolute value, and RollingAbsMax for the maximum absolute value over a window. All variants update in constant time O(1) for the non-rolling versions.

Usage

Use Maximum when you need to track the largest value in streaming data. Common applications include monitoring peak values in sensor data, tracking maximum prices in financial data, finding extreme values for normalization, and detecting anomalies. The rolling variants are useful when only recent maximum values are relevant.

Code Reference

Source Location

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

Signature

class Max(stats.base.Univariate):
    def __init__(self):
        self.max = -math.inf

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

class AbsMax(stats.base.Univariate):
    def __init__(self):
        self.abs_max = 0.0

class RollingAbsMax(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 maximum value (or None for rolling if window not filled)

Usage Examples

from river import stats

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

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

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

# Rolling maximum
X = [1, -4, 3, -2, 2, 1]
rolling_max = stats.RollingMax(window_size=2)

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

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

# Absolute maximum
X = [1, -4, 3, -2, 5, -6]
abs_max = stats.AbsMax()

for x in X:
    abs_max.update(x)
    print(f"Value: {x}, Abs Max: {abs_max.get()}")

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

Related Pages

• Environment:Online_ml_River_Python_Runtime_Environment