Sliding-window maximums in Python: a monotonic deque avoids repeated scans

Problem. A program that needs the maximum of every overlapping window can keep scanning values it inspected one step ago. Baseline complexity. The direct Python implementation below takes O((n – k + 1)k) time for n values and window width k, and creates one k-item slice per result. Solution. A monotonic deque retains only indices that can still become a window maximum. Measured result. On 80,000 seeded integers with k = 2,000, the executed run produced equal outputs; its best repeated-scan time was 6.158 s and the deque version took 0.0642 s, a 95.86× ratio on this host.

Reproduce the result

Complete code

from __future__ import annotations
from collections import deque
import gc
import json
import platform
import random
import sys
import timeit
import tracemalloc
N = 80_000
WINDOW = 2_000
SEED = 20260902
REPEATS = 5
def baseline_maxima(values: list[int], width: int) -> list[int]:
    if not 1 <= width <= len(values):
        raise ValueError("width must be between 1 and len(values)")
    return [max(values[start:start + width]) for start in range(len(values) - width + 1)]
def monotonic_maxima(values: list[int], width: int) -> list[int]:
    if not 1 <= width <= len(values):
        raise ValueError("width must be between 1 and len(values)")
    candidates: deque[int] = deque()
    maxima: list[int] = []
    for index, value in enumerate(values):
        while candidates and values[candidates[-1]] <= value:
            candidates.pop()
        candidates.append(index)
        first_index = index - width + 1
        if candidates[0] < first_index:
            candidates.popleft()
        if first_index >= 0:
            maxima.append(values[candidates[0]])
    return maxima
def counted_monotonic_maxima(values: list[int], width: int) -> tuple[list[int], int]:
    if not 1 <= width <= len(values):
        raise ValueError("width must be between 1 and len(values)")
    candidates: deque[int] = deque()
    maxima: list[int] = []
    value_comparisons = 0
    for index, value in enumerate(values):
        while candidates:
            value_comparisons += 1
            if values[candidates[-1]] > value:
                break
            candidates.pop()
        candidates.append(index)
        first_index = index - width + 1
        if candidates[0] < first_index:
            candidates.popleft()
        if first_index >= 0:
            maxima.append(values[candidates[0]])
    return maxima, value_comparisons
def peak_traced_bytes(function, values: list[int], width: int) -> int:
    gc.collect()
    tracemalloc.start()
    function(values, width)
    _, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return peak
def main() -> None:
    random_source = random.Random(SEED)
    values = [random_source.randrange(-1_000_000, 1_000_001) for _ in range(N)]
    baseline = baseline_maxima(values, WINDOW)
    optimised = monotonic_maxima(values, WINDOW)
    counted, candidate_comparisons = counted_monotonic_maxima(values, WINDOW)
    assert baseline == optimised == counted
    assert monotonic_maxima([5], 1) == [5]
    assert monotonic_maxima([2, 2, 2], 2) == [2, 2]
    for invalid_width in (0, N + 1):
        try:
            monotonic_maxima(values, invalid_width)
        except ValueError:
            pass
        else:
            raise AssertionError("invalid width was accepted")
    gc.collect()
    baseline_samples = timeit.repeat(
        lambda: baseline_maxima(values, WINDOW), repeat=REPEATS, number=1
    )
    gc.collect()
    monotonic_samples = timeit.repeat(
        lambda: monotonic_maxima(values, WINDOW), repeat=REPEATS, number=1
    )
    baseline_peak = peak_traced_bytes(baseline_maxima, values, WINDOW)
    monotonic_peak = peak_traced_bytes(monotonic_maxima, values, WINDOW)
    baseline_best = min(baseline_samples)
    monotonic_best = min(monotonic_samples)
    result = {
        "python": sys.version.split()[0],
        "platform": platform.platform(),
        "n": N,
        "window": WINDOW,
        "seed": SEED,
        "windows_produced": len(baseline),
        "all_maxima_equal": baseline == optimised == counted,
        "edge_checks": "single_value, duplicate_values, invalid_widths",
        "baseline_value_inspections_theoretical": (N - WINDOW + 1) * WINDOW,
        "monotonic_candidate_value_comparisons_observed": candidate_comparisons,
        "baseline_seconds_repeat_5": baseline_samples,
        "monotonic_seconds_repeat_5": monotonic_samples,
        "baseline_best_seconds": baseline_best,
        "monotonic_best_seconds": monotonic_best,
        "best_time_ratio_baseline_over_monotonic": baseline_best / monotonic_best,
        "baseline_peak_traced_bytes": baseline_peak,
        "monotonic_peak_traced_bytes": monotonic_peak,
    }
    print(json.dumps(result, indent=2, sort_keys=True))
if __name__ == "__main__":
    main()

Output

{
  "all_maxima_equal": true,
  "baseline_best_seconds": 6.15825610794127,
  "baseline_peak_traced_bytes": 649000,
  "baseline_seconds_repeat_5": [
    6.173405050998554,
    6.223466970957816,
    6.179453025804833,
    6.1667492219712585,
    6.15825610794127
  ],
  "baseline_value_inspections_theoretical": 156002000,
  "best_time_ratio_baseline_over_monotonic": 95.8564241095773,
  "edge_checks": "single_value, duplicate_values, invalid_widths",
  "monotonic_best_seconds": 0.06424458418041468,
  "monotonic_candidate_value_comparisons_observed": 159908,
  "monotonic_peak_traced_bytes": 634888,
  "monotonic_seconds_repeat_5": [
    0.06424458418041468,
    0.06521174195222557,
    0.06448776600882411,
    0.06483835610561073,
    0.06464950600638986
  ],
  "n": 80000,
  "platform": "Linux-6.12.47+rpt-rpi-v8-aarch64-with-glibc2.41",
  "python": "3.13.5",
  "seed": 20260902,
  "window": 2000,
  "windows_produced": 78001
}

Environment

The program ran with CPython 3.13.5 on Linux 6.12.47+rpt-rpi-v8-aarch64-with-glibc2.41. It uses only the standard library. The timeit module times small code fragments with perf_counter() by default, and its timing calls temporarily disable garbage collection unless the setup enables it.^[3] The memory values come from tracemalloc, which traces Python allocation blocks rather than a process’s resident set size.^[4]

Methodology

The source creates 80,000 deterministic integers with seed 20260902 and asks for maxima in 78,001 overlapping windows of width 2,000. It checks full output equality before timing, then checks a single-value input, duplicate values and invalid widths. Each implementation runs once in each of five timeit.repeat samples; the article uses the minimum, while retaining the full vectors in Output. The baseline’s theoretical inspection count is 156,002,000. A separately counted deque pass made 159,908 candidate-value comparisons. Allocation tracing starts immediately before each function call and stops immediately after it returns, so the two peak values are narrow Python-allocation probes, not whole-machine memory measurements.

The overlapping-window problem

A moving maximum appears in monitoring, signal processing and time-series work. Lemire describes the related running maximum-minimum filter as computing extrema over moving windows, with applications in signal processing and time series analysis.^[1]

The straightforward Python version is pleasantly short: take values[start:start + width], call max, repeat. It is also doing a surprising amount of recycling. Adjacent windows of width 2,000 share 1,999 positions, yet the direct code evaluates a new slice and a new maximum for each starting point.

That is a valid choice for a handful of windows, for tiny widths, or when clarity matters more than throughput. The fault is not in max. The workload has changed around it.

Baseline complexity and real costs

There are n – k + 1 complete windows when 1 ≤ k ≤ n. Scanning each one takes O(k), so the baseline is O((n – k + 1)k), commonly written O(nk). The executed input makes that exact product 156,002,000 value positions. It is a count for this implementation’s conceptual scans, not a CPU-instruction count.

Python adds costs that the asymptotic notation does not show. values[start:start + width] allocates a fresh list of references, then max walks it. The result list is retained by both versions. The allocation probe reported 649,000 peak traced bytes for the baseline and 634,888 for the deque version. That small difference does not mean the two methods have the same memory behaviour in every process. The baseline slice is temporary, the output list dominates this particular probe, and tracemalloc deliberately sees only Python-traced allocations.^[4]

Keep only candidates that can still win

The deque holds indices, not values. Its values are strictly decreasing from front to back. The front is therefore the largest candidate for the current window.

When a new value arrives, smaller or equal values at the back leave the deque. They are behind the new value and cannot become a later maximum before the new value itself expires. Then the new index joins the back. If the front index falls before the window’s left edge, it leaves from the front. Python documents collections.deque as a list-like container with fast appends and pops at either end, which is the operation pattern used here.^[2]

Each index is appended once. An index can be removed from the back at most once, or later from the front at most once. The resulting candidate maintenance is O(n) amortised time and O(k) deque space. The separate counted run found 159,908 candidate-value comparisons for 80,000 inputs. That number is an observation for this seeded data, while the one-entry, one-removal argument is the general bound.

The <= condition is deliberate. With equal values, retaining the newer index means it remains valid for longer; the duplicate-value check in the program confirms the output [2, 2] for [2, 2, 2] with width 2.

What this measurement shows

The raw run is in Output. The baseline samples ranged from 6.158 s to 6.223 s. The deque samples ranged from 0.0642 s to 0.0652 s. Their minimum-sample ratio was 95.86. Python’s documentation advises inspecting the complete repeat vector and says the lowest value is generally the useful lower bound when other processes disturb timing.^[3]

This is evidence for one data shape, one window width, one Python build and one aarch64 host. It does not establish a fixed speed-up for every machine, every k or a production system with I/O and concurrent work. The direct implementation calls built-in max after producing a slice, whereas the deque algorithm performs its bookkeeping in Python bytecode. For a very narrow window, the fixed bookkeeping can matter more than the avoided scans. That boundary was not measured here.

Where it stops helping

The deque answers a maximum for every fixed-width step through an ordered sequence. It is not a general range-maximum index. If queries can ask arbitrary ranges in arbitrary order, a sparse table or segment tree may fit better. If the window width changes per query, this queue no longer describes the task.

It also does not make old results update themselves. Changing a value already inside the stream changes later windows and requires recalculation from an appropriate point. For a single maximum of one list, max(values) is clearer and has no queue to maintain. The deque earns its extra lines when a long sequence generates many overlapping fixed-width maxima, which is exactly where the measured scan count becomes hard to ignore.

Sources

[1] Lemire, Streaming Maximum-Minimum Filter Using No More than Three Comparisons per Element

[2] Python documentation: collections.deque

[3] Python documentation: timeit

[4] Python documentation: tracemalloc