# /// script
# requires-python = ">=3.10"
# dependencies = ["numpy"]
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Sliding maximum: naive, monotonic deque, and panes.

The maximum of the last w values at every step over n values, three ways:
max() over the window, a monotonic deque, and panes (maxima of w/10-value
chunks, 10 combined per output). Then the deque on decreasing and increasing
input, its worst and best shapes.
"""
import argparse, json, random, time
from collections import deque
import numpy as np

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--values", type=int, default=10_000_000, help="Values in the stream.")
p.add_argument("--window", type=int, default=10_000, help="Window w.")
p.add_argument("--naive-values", type=int, default=20_000, help="Values the naive version processes (it is w times slower).")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
rng = np.random.default_rng(args.seed)
w = args.window
xs = rng.random(args.values).tolist()

def naive(v):
    return [max(v[i - w + 1: i + 1]) for i in range(w - 1, len(v))]

def deq(v):
    d, out = deque(), []
    for i, x in enumerate(v):
        while d and v[d[-1]] <= x: d.pop()
        d.append(i)
        if d[0] <= i - w: d.popleft()
        if i >= w - 1: out.append(v[d[0]])
    return out

def panes(v):
    pw = w // 10
    pm = [max(v[i:i + pw]) for i in range(0, len(v) - pw + 1, pw)]           # each pane aggregated once
    return [max(pm[j - 9: j + 1]) for j in range(9, len(pm))]                 # 10 pane maxima per output (outputs at pane edges)

def timed(fn, v):
    t = time.perf_counter(); out = fn(v); return time.perf_counter() - t, out

rows = []
tn, outn = timed(naive, xs[: args.naive_values + w])
per_n = tn / len(outn)
td, outd = timed(deq, xs); tp, _ = timed(panes, xs)
assert outn == outd[: len(outn)]
for name, per in (("naive max over the window", per_n), ("monotonic deque", td / len(outd)), ("panes of w/10", tp / (len(outd)))):
    rows.append({"method": name, "ns_per_value": per * 1e9}); print(f"{name:<28} {per * 1e9:10.1f} ns per value")
shapes = []
for shape, v in (("decreasing", list(range(args.values // 10, 0, -1))), ("increasing", list(range(args.values // 10)))):
    t, _ = timed(deq, v); shapes.append({"input": shape, "ns_per_value": t / len(v) * 1e9})
    print(f"deque on {shape} input: {t / len(v) * 1e9:.1f} ns per value")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"methods": rows, "shapes": shapes, "summary": [{"naive_over_deque": rows[0]["ns_per_value"] / rows[1]["ns_per_value"],
                   "panes_over_deque": rows[2]["ns_per_value"] / rows[1]["ns_per_value"]}]}}, f)