Written on 

LLM: Write production-ready code!

llm programming quickie

Problem statement

I’ve been trying to build out and optimize tobes-ui’s wavelength calibration. It’s surprisingly involved1. 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 waste2 to… well, at least as good code as mine; if not better.

Intrigued? Let’s dig in3.

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 wrote4:

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)

or5 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 ones6, 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:

The holy-shit moment

I had the brilliant idea7 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 inconsequential8, changes the output from a steaming pile of slop, to roughly the code I would actually write9. 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?10

  1. Devil, details, all that. ↩

  2. According to my biases… ehm, I mean… subjective standards? ↩

  3. tl;dr: “Write production-ready code.” in prompt made all the difference. ↩

  4. And I don’t deny I stole re-used bunch of stuff from the LLM output. ↩

  5. Depending on current unix ts in nanoseconds, barometric pressure, and position of planets. ↩

  6. Fine, fine, not strictly true, clocks might not be monotonic sometimes. And if this were a pacemaker, I would care. I don’t care for making Y axis of tobes-ui wavelength calibration less jumpy. ↩

  7. Tongue in cheek, really, when chatting with a friend of mine… ↩

  8. More of a joke, really. ↩

  9. On cursory glance; haven’t read it all in depth. Claude’s pass all my unit tests, ChatGPT’s fail in one case. ↩

  10. NGL, it already is. Every little bit helps. ↩