Monotonic deque in Python
Deque kept in monotone order so the window extremum is read off in O(1) per step.
- File
monotonic_deque.py, 81 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/monotonic-deque/. The authors are listed in prov/S8NDWC.jsonl.
from collections import deque
from typing import Generic, Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: "Ordered", /) -> bool: ...
T = TypeVar("T", bound=Ordered)
class MonotonicDeque(Generic[T]):
"""A queue with a minimum query: push at the back, pop at the front, min.
The deque holds (position, item) pairs of the strict suffix minima of the
queue, so the items increase from the front to the back and the front is
the minimum. push costs O(1) amortized, pop and min cost O(1) worst case.
pop does not return the item, because not every item is stored.
"""
def __init__(self) -> None:
self.d: deque[tuple[int, T]] = deque() # @line L1
self.h = 0
self.t = 0
def push(self, x: T) -> None:
"""Append x at the back of the queue. Cost O(1) amortized."""
d = self.d
while d and not d[-1][1] < x: # @line L2: back item >= x
d.pop() # @line L3
d.append((self.t, x)) # @line L4
self.t += 1 # @line L5
def pop(self) -> None:
"""Remove the oldest item. Precondition: the queue is not empty. Cost O(1)."""
if self.h >= self.t:
raise IndexError("pop from an empty queue")
if self.d[0][0] == self.h: # @line L6
self.d.popleft()
self.h += 1 # @line L7
def min(self) -> T:
"""Return the least item of the queue (the newest one on ties). Cost O(1)."""
if self.h >= self.t:
raise IndexError("min of an empty queue")
return self.d[0][1] # @line L8
def __len__(self) -> int:
return self.t - self.h
def check_invariant(self) -> None:
"""Check the parts of the invariant that do not need the popped items."""
pos = [j for j, _ in self.d]
items = [x for _, x in self.d]
assert all(self.h <= j < self.t for j in pos)
assert all(a < b for a, b in zip(pos, pos[1:]))
assert all(a < b for a, b in zip(items, items[1:]))
assert (len(self) == 0) == (len(self.d) == 0)
if self.d:
assert pos[-1] == self.t - 1
def sliding_min(a: list[T], k: int) -> list[T]:
"""Return the minimum of each window a[r - k + 1 .. r], for k - 1 <= r < len(a).
Precondition: k >= 1. Cost Theta(len(a)) time and O(k) extra space.
"""
if k < 1:
raise ValueError("k must be at least 1")
q: MonotonicDeque[T] = MonotonicDeque()
out: list[T] = []
for r in range(len(a)):
q.push(a[r])
if len(q) > k:
q.pop()
if r >= k - 1:
out.append(q.min())
return out
test_monotonic_deque.py 86 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/monotonic-deque/. The authors are listed in prov/S8NDWC.jsonl.
import random
import pytest
from monotonic_deque import MonotonicDeque, sliding_min
def test_example() -> None:
assert sliding_min([4, 2, 12, 3, 8, 5, 1, 7], 3) == [2, 2, 3, 3, 1, 1]
def test_example_states() -> None:
q: MonotonicDeque[int] = MonotonicDeque()
states = []
for r, x in enumerate([4, 2, 12, 3, 8, 5, 1, 7]):
q.push(x)
if len(q) > 3:
q.pop()
q.check_invariant()
states.append(list(q.d))
assert states[3] == [(1, 2), (3, 3)]
assert states[4] == [(3, 3), (4, 8)]
assert states[6] == [(6, 1)]
def test_edges() -> None:
assert sliding_min([], 1) == []
assert sliding_min([5], 1) == [5]
assert sliding_min([3, 1, 2], 1) == [3, 1, 2]
assert sliding_min([3, 1, 2], 3) == [1]
assert sliding_min([3, 1], 5) == []
assert sliding_min([7, 7, 7, 7], 2) == [7, 7, 7]
assert sliding_min([1, 2, 3, 4], 2) == [1, 2, 3]
assert sliding_min([4, 3, 2, 1], 2) == [3, 2, 1]
assert sliding_min([-(2**63), 2**63 - 1, 0], 2) == [-(2**63), 0]
def test_pitfalls() -> None:
# tie rule: two equal items, the first leaves, a 2 is still in the window
assert sliding_min([2, 2, 5], 2) == [2, 2]
# window test off by one
assert sliding_min([1, 3], 2) == [1]
def test_empty_errors() -> None:
q: MonotonicDeque[int] = MonotonicDeque()
with pytest.raises(IndexError):
q.pop()
with pytest.raises(IndexError):
q.min()
with pytest.raises(ValueError):
sliding_min([1], 0)
def test_random_operations_against_list() -> None:
rng = random.Random(4711)
for _ in range(500):
q: MonotonicDeque[int] = MonotonicDeque()
model: list[int] = []
for _ in range(rng.randint(0, 60)):
op = rng.random()
if op < 0.55 or not model:
x = rng.randint(-5, 5)
q.push(x)
model.append(x)
elif op < 0.8:
q.pop()
model.pop(0)
else:
assert q.min() == min(model)
q.check_invariant()
assert len(q) == len(model)
if model:
assert q.min() == min(model)
def test_random_windows_against_brute_force() -> None:
rng = random.Random(90210)
for _ in range(500):
a = [rng.randint(-20, 20) for _ in range(rng.randint(0, 30))]
k = rng.randint(1, 8)
want = [min(a[i : i + k]) for i in range(len(a) - k + 1)]
assert sliding_min(a, k) == want
What the next level needs
- the tests did not run in this build