Written
on 2025-10-25
Problem statement
I’ve been trying to build out and optimize
tobes-ui’s wavelength calibration.
It’s surprisingly involved. So being the lazy ’uman I am, I turned to
mechanized help.
What is surprising about this story is that adding one sentence to the prompt
turned the output from a steaming pile of bovine waste to… well, at least
as good code as mine; if not better.
Intrigued? Let’s dig in.
Prompt
I used the following prompt for the LLMs:
I want to compute streaming max from incoming values, over a defined time
window. the rate of incoming changes, and the window should be ad-hoc (runtime)
reconfigurable. Can you sketch a python class, say
SlidingMax, that would do that in O(1)? One public method add to add a new
value (and return max), plus window_size property to control the window size
at runtime…
Results
For that prompt I’ve got the following results… and I’m betting you don’t
find them great. I know I was underwhelmed.
ChatGPT-5
from collections import deque
from time import time
class SlidingMax:
def __init__(self, window_size: int):
self.window_size = window_size
self.data = deque() # Will hold pairs of (index, value)
self.window = deque() # Will hold indices of values in decreasing order
self.time = deque() # Hold timestamp to track window expiration
self.counter = 0 # To use as a unique index for each add
@property
def window_size(self):
return self._window_size
@window_size.setter
def window_size(self, size: int):
self._window_size = size
# When window size is changed, clear outdated data in the deque.
self.trim_window()
def trim_window(self):
# Remove outdated values based on window size
while self.time and self.time[0] < self.counter - self._window_size:
self.time.popleft()
self.data.popleft()
self.window.popleft()
def add(self, value: float) -> float:
# Update the time counter and store the new value
self.counter += 1
self.time.append(self.counter)
self.data.append(value)
# Maintain the decreasing order in the deque
while self.window and self.data[self.window[-1]] <= value:
self.window.pop()
# Add the current index (the position of the new value in data)
self.window.append(self.counter)
# Trim the window if it exceeds the size
self.trim_window()
# Return the current maximum value (value at the front of the deque)
return self.data[self.window[0]]
# Usage example:
sm = SlidingMax(5)
# Adding values
print(sm.add(10)) # Max = 10
print(sm.add(5)) # Max = 10
print(sm.add(20)) # Max = 20
print(sm.add(3)) # Max = 20
print(sm.add(15)) # Max = 20
# Reconfigure window size at runtime
sm.window_size = 3 # Reduce window size
print(sm.add(8)) # Max = 15 (New window size)
Claude Sonnet 4.5
from collections import deque
from time import time
class SlidingMax:
def __init__(self, window_size=60.0):
self._window_size = window_size
self._values = deque() # (timestamp, value)
self._max_deque = deque() # (timestamp, value) in decreasing order
@property
def window_size(self):
return self._window_size
@window_size.setter
def window_size(self, new_size):
self._window_size = new_size
self._cleanup(time())
def add(self, value, timestamp=None):
if timestamp is None:
timestamp = time()
self._cleanup(timestamp)
# Remove elements from back that are smaller than current value
while self._max_deque and self._max_deque[-1][1] <= value:
self._max_deque.pop()
self._values.append((timestamp, value))
self._max_deque.append((timestamp, value))
return self._max_deque[0][1] if self._max_deque else None
def _cleanup(self, current_time):
cutoff = current_time - self._window_size
# Remove expired values
while self._values and self._values[0][0] <= cutoff:
self._values.popleft()
# Remove expired max candidates
while self._max_deque and self._max_deque[0][0] <= cutoff:
self._max_deque.popleft()
’Uman-80
This is actually the code I wrote:
class SlidingMax:
"""Sliding max over a window_size seconds; useful e.g. to avoid jumpy Y axis."""
def __init__(self, window_size=5.0):
# (timestamp, value); timestamps increasing, values decreasing
self._max_deque = deque()
self.window_size = window_size # setter to enforce positivity
@property
def window_size(self) -> float:
"""Get current window size"""
return self._window_size
@window_size.setter
def window_size(self, new_size: float) -> float:
"""Resize the window"""
if new_size <= 0:
raise ValueError(f'window_size expected positive, got {new_size}')
self._window_size = new_size
self._remove_expired_entries(time.time())
def add(self, value):
"""Add new value, return max."""
timestamp = time.time()
self._remove_expired_entries(timestamp)
# Remove elements from back that are smaller than current value
# This is fine, as a value will be added, and that has > ts
# (assuming monotonic time)
while self._max_deque and self._max_deque[-1][1] <= value:
self._max_deque.pop()
self._max_deque.append((timestamp, value))
return self._max_deque[0][1] if self._max_deque else None
def _remove_expired_entries(self, current_time: float):
"""Removes expired entries from the dbl-ended queue"""
cutoff = current_time - self._window_size
# Note:
# Monotonic time is assumed; might lead to subtle "bugs" in rare cases;
# so don't use this to run a nuclear power plant, ok? ;)
while self._max_deque and self._max_deque[0][0] <= cutoff:
self._max_deque.popleft()
Discussion
As you see from the code above, it’s roughly the same, but LLM output is
littered with dead code, and illogical bullshit.
Even weirder is that ChatGPT knows the correct answer.
When prompted:
I want to compute streaming max from incoming values, over a defined time
window. the rate of incoming changes, and the window should be ad-hoc (runtime)
reconfigurable. How would I do that in O(1) time?
which is pretty much most of my original prompt, just devoid of the shape of the
class, it either spits out:
from collections import deque
class StreamingMax:
def __init__(self, window_size):
self.window_size = window_size
self.deque = deque() # Stores (value, timestamp or index)
def update(self, value, timestamp):
# Remove elements outside the window size
while self.deque and self.deque[0][1] <= timestamp - self.window_size:
self.deque.popleft()
# Remove all smaller elements from the back of the deque
while self.deque and self.deque[-1][0] <= value:
self.deque.pop()
# Add the new value along with the timestamp or index
self.deque.append((value, timestamp))
def get_max(self):
# The front of the deque is the maximum value
return self.deque[0][0]
def reconfigure_window(self, new_window_size):
self.window_size = new_window_size
# Clean up the deque based on the new window size
while self.deque and self.deque[0][1] <= timestamp - self.window_size:
self.deque.popleft()
# Usage
stream = StreamingMax(window_size=5)
# Simulate data stream with timestamps (indexing for simplicity)
for timestamp in range(10):
value = some_data_source[timestamp]
stream.update(value, timestamp)
print(f"Max at time {timestamp}: {stream.get_max()}")
# Dynamically change the window size
stream.reconfigure_window(10)
or solid reasoning and pseudo-code
as to why a single deque is the right answer. But I tl;dr’d that on the first
few tries… only to waste many a precious minute doing it all in my head.
But ask it to do stuff your way, and the results are shit. And Claude
slaps in a bunch of code that’s useless. What is self._values for, again?
Anyway, I fought with both of them for a while, before finally sitting down
and thinking through the deque solution.
Why does single deque work?
If you pop expired entries from the left (which you absolutely should; they’re
useless), there’s going to be max at the leftmost position… provided you
also: pop from the right all the values smaller than the one you’re currently
adding.
Why does it work? The one you’re adding has a higher timestamp than any of the
previous ones, so you’re getting rid of useless bulk (chaff), and only
keeping the stuff you need (wheat).
Plus, this makes the complexity much better than some of the other “solutions”
I got from the LLMs; for example the one where max() was always computed by
iterating over an array holding all values for the whole window, etc.
So, this deque solution is actually pretty neat, if you think about it.
Worst case complexity:
- Space-wise: Worst-case is, I think, when you put in a max value, and then you
always insert smaller values. That’s when the deque is going to be longest.
So, O(n) at most;
- Time-wise: Seems roughly O(1) to me. Sure, not constant latency
(sometimes you gotta pop a bunch more elements etc), but ammortized it’s
definitely in the O(1) ballpark.
The holy-shit moment
I had the brilliant idea to add a single “Write production-ready code.”
line to the original prompt. In my mind, shouldn’t change much, eh?
Holy crap, is there a difference or what…
ChatGPT-5
from collections import deque
import time
from typing import Optional, Union
class SlidingMax:
"""
Streaming sliding-window maximum over a time-based window.
- O(1) amortized per add() operation.
- Time-based (not count-based) window.
- Window size can be adjusted at runtime via the `window_size` property.
Example:
sm = SlidingMax(window_size=5.0)
for v in [3, 1, 5, 2]:
print(sm.add(v)) # current max in last 5 seconds
"""
def __init__(self, window_size: float):
if window_size <= 0:
raise ValueError("window_size must be positive (seconds)")
self._window_size = float(window_size)
self._dq = deque() # stores (timestamp, value)
@property
def window_size(self) -> float:
"""Current time window size in seconds."""
return self._window_size
@window_size.setter
def window_size(self, new_size: Union[int, float]):
if new_size <= 0:
raise ValueError("window_size must be positive (seconds)")
self._window_size = float(new_size)
self._evict_old(time.time()) # prune immediately using current time
def add(self, value: float, timestamp: Optional[float] = None) -> float:
"""
Add a new value with an optional timestamp (defaults to time.time()).
Returns the current maximum within the active time window.
"""
now = timestamp if timestamp is not None else time.time()
# Remove elements outside the current window.
self._evict_old(now)
# Maintain deque in decreasing order of values.
while self._dq and self._dq[-1][1] <= value:
self._dq.pop()
self._dq.append((now, value))
return self._dq[0][1] # max value (front of deque)
def _evict_old(self, now: float):
"""Remove elements older than the current window."""
cutoff = now - self._window_size
while self._dq and self._dq[0][0] < cutoff:
self._dq.popleft()
# Example usage
if __name__ == "__main__":
sm = SlidingMax(window_size=3.0)
sm.add(5)
time.sleep(1)
sm.add(2)
time.sleep(1)
sm.add(8)
print(sm.add(6)) # Should print 8
sm.window_size = 1.0 # Shrink window dynamically
print(sm.add(7)) # Recomputed max within new window
Claude Sonnet 4.5
from collections import deque
from typing import Optional
import time
class SlidingMax:
"""
Maintains the maximum value over a sliding time window with O(1) amortized complexity.
Uses a monotonic deque to track potential maximum values efficiently.
Thread-safe for single-writer scenarios.
"""
def __init__(self, window_size: float):
"""
Initialize the sliding maximum tracker.
Args:
window_size: Time window in seconds (must be positive)
Raises:
ValueError: If window_size is not positive
"""
if window_size <= 0:
raise ValueError("window_size must be positive")
self._window_size = float(window_size)
self._deque: deque[tuple[float, float]] = deque() # (timestamp, value)
@property
def window_size(self) -> float:
"""Get the current window size in seconds."""
return self._window_size
@window_size.setter
def window_size(self, size: float) -> None:
"""
Set a new window size and clean up stale entries.
Args:
size: New window size in seconds (must be positive)
Raises:
ValueError: If size is not positive
"""
if size <= 0:
raise ValueError("window_size must be positive")
self._window_size = float(size)
self._cleanup(time.time())
def add(self, value: float, timestamp: Optional[float] = None) -> Optional[float]:
"""
Add a new value and return the current maximum.
Args:
value: The value to add
timestamp: Optional timestamp (defaults to current time)
Returns:
The maximum value in the current window, or None if window is empty
Raises:
ValueError: If timestamp is provided and is older than the most recent entry
"""
if timestamp is None:
timestamp = time.time()
else:
timestamp = float(timestamp)
if self._deque and timestamp < self._deque[-1][0]:
raise ValueError("Timestamps must be monotonically increasing")
self._cleanup(timestamp)
while self._deque and self._deque[-1][1] <= value:
self._deque.pop()
self._deque.append((timestamp, value))
return self._deque[0][1] if self._deque else None
def get_max(self) -> Optional[float]:
"""
Get the current maximum without adding a value.
Returns:
The maximum value in the current window, or None if window is empty
"""
self._cleanup(time.time())
return self._deque[0][1] if self._deque else None
def _cleanup(self, current_time: float) -> None:
"""Remove entries outside the current window."""
cutoff = current_time - self._window_size
while self._deque and self._deque[0][0] <= cutoff:
self._deque.popleft()
def clear(self) -> None:
"""Remove all entries from the window."""
self._deque.clear()
def __len__(self) -> int:
"""Return the number of entries currently in the window."""
self._cleanup(time.time())
return len(self._deque)
def __repr__(self) -> str:
max_val = self.get_max()
return f"SlidingMax(window_size={self._window_size}, max={max_val}, entries={len(self)})"
Closing words
Honestly, I’m confused. One line, that I would consider absolutely
inconsequential, changes the output from a steaming pile of slop, to
roughly the code I would actually write. Not to mention their’s has better
docstrings, type annotations, and other niceties.
Mind boggles.
But, should “Write production-ready code.” be my standard prelude now?
.png)
2 hours ago
1
