#1: Implement Time-Windowed Rate Limiter (coding implementation)
Winner: Answer A (Deepseek V4 Flash - high)| Deepseek V4 Flash - high | Deepseek V4 Flash - max | |
|---|---|---|
| Score | 9.0 | 5.0 |
| Time | 17.3s | 16.1s |
| Tokens | 2615 | 2568 |
| Speed | 152 tok/s | 160 tok/s |
| Restarts | 0 | 0 |
Winner: Answer A (Deepseek V4 Flash - high)
Judge reasoning:
Answer A correctly implements per-key deques, validates positive limit and window_seconds, uses time.monotonic() when needed, prunes expired timestamps with the required <= now - window_seconds boundary, removes empty buckets, and includes a self-test covering two keys and boundary expiration. Minor issues are mostly cosmetic, such as confusing comments in the self-test. Answer B is efficient and mostly structured well, but it prunes with < cutoff instead of <= cutoff, violating the required boundary behavior. Its self-test also contains incorrect assertions and would fail when run, so it does not meet the self-test requirement reliably.
Prompt
Implement a single-file Python 3 module containing a class SlidingWindowRateLimiter. It must support __init__(limit: int, window_seconds: float), allow(key: str, now: float | None = None) -> bool, and remaining(key: str, now: float | None = None) -> int. A key is allowed at most limit events in any rolling window of length window_seconds. If now is None, use time.monotonic(). The implementation must be efficient for many keys and many calls: expired timestamps should not remain indefinitely. Edge cases: limit must be positive, window_seconds must be positive, timestamps may be non-integer floats, different keys are independent, and calls with explicit now values are nondecreasing per key but may differ across keys. Include a small self-test block under if __name__ == '__main__': demonstrating behavior for at least two keys and boundary expiration. Do not use third-party packages.Deepseek V4 Flash - high answer (Answer A)
import time
from collections import deque
from typing import Dict, Optional
class SlidingWindowRateLimiter:
"""Tracks per-key event rates within a sliding time window.
For each key, at most `limit` events are allowed in any rolling window
of length `window_seconds`. The implementation uses a dictionary of
deques to store event timestamps and prunes expired entries on each
call, ensuring memory does not grow unboundedly.
"""
def __init__(self, limit: int, window_seconds: float) -> None:
if limit <= 0:
raise ValueError(f"limit must be positive, got {limit}")
if window_seconds <= 0:
raise ValueError(f"window_seconds must be positive, got {window_seconds}")
self._limit = limit
self._window = window_seconds
self._buckets: Dict[str, deque] = {} # key -> deque of timestamps
def _prune(self, key: str, now: float) -> None:
"""Remove timestamps that are outside the current window."""
if key not in self._buckets:
return
dq = self._buckets[key]
cutoff = now - self._window
# Since `now` is non‑decreasing per key, we can safely pop from the left.
while dq and dq[0] <= cutoff:
dq.popleft()
if not dq:
del self._buckets[key]
def _get_now(self, now: Optional[float]) -> float:
return time.monotonic() if now is None else now
def allow(self, key: str, now: Optional[float] = None) -> bool:
"""Record an event for *key* at time *now* (or monotonic time if None).
Returns True if the event is allowed (i.e. the count in the current
window is below the limit), otherwise False.
"""
now = self._get_now(now)
self._prune(key, now)
dq = self._buckets.get(key)
if dq is None:
dq = deque()
self._buckets[key] = dq
if len(dq) < self._limit:
dq.append(now)
return True
return False
def remaining(self, key: str, now: Optional[float] = None) -> int:
"""Return how many events for *key* can still be allowed at *now*.
This method does **not** record an event.
"""
now = self._get_now(now)
self._prune(key, now)
dq = self._buckets.get(key)
count = len(dq) if dq else 0
return self._limit - count
if __name__ == "__main__":
# Self‑test: two keys, boundary expiration, and independence.
limiter = SlidingWindowRateLimiter(limit=2, window_seconds=1.0)
# --- Key "A" ---
# t=0.0: allow two events
assert limiter.allow("A", 0.0) == True
assert limiter.allow("A", 0.0) == True
# t=0.0: third event should be denied
assert limiter.allow("A", 0.0) == False
# remaining should be 0
assert limiter.remaining("A", 0.0) == 0
# t=0.5: still within window of first two events -> denied
assert limiter.allow("A", 0.5) == False
assert limiter.remaining("A", 0.5) == 0
# t=1.0: first event (t=0.0) is exactly at the boundary (now - window = 0.0)
# and should be pruned because we use > cutoff (strictly greater).
# After pruning, only the second event (t=0.0) remains? Wait, both
# events at t=0.0 are pruned because cutoff = 1.0 - 1.0 = 0.0,
# and 0.0 <= 0.0 is true. So both are removed. Then the deque is empty.
# So allow should succeed.
assert limiter.allow("A", 1.0) == True
assert limiter.remaining("A", 1.0) == 1 # one event just added
# t=1.0: second event allowed
assert limiter.allow("A", 1.0) == True
assert limiter.remaining("A", 1.0) == 0
# t=1.0: third denied
assert limiter.allow("A", 1.0) == False
# --- Key "B" (independent) ---
# t=0.0: allow three events (limit=2, so only two succeed)
assert limiter.allow("B", 0.0) == True
assert limiter.allow("B", 0.0) == True
assert limiter.allow("B", 0.0) == False
assert limiter.remaining("B", 0.0) == 0
# t=1.0: both events at t=0.0 are pruned, so two new events allowed
assert limiter.allow("B", 1.0) == True
assert limiter.allow("B", 1.0) == True
assert limiter.allow("B", 1.0) == False
# Verify key "A" is unaffected by key "B"
assert limiter.remaining("A", 1.0) == 0 # A had two events at t=1.0
# --- Edge: non‑integer timestamps ---
limiter2 = SlidingWindowRateLimiter(limit=1, window_seconds=0.5)
assert limiter2.allow("X", 0.1) == True
assert limiter2.allow("X", 0.2) == False # still within 0.5s window
assert limiter2.allow("X", 0.6001) == True # first event expired
print("All tests passed.")
Deepseek V4 Flash - max answer (Answer B)
import time
from collections import deque
class SlidingWindowRateLimiter:
"""Tracks rate limits per key using a sliding window of fixed duration."""
def __init__(self, limit: int, window_seconds: float):
"""
Initialize the rate limiter.
Args:
limit: Maximum number of events allowed within the window.
window_seconds: Length of the sliding window in seconds.
Raises:
ValueError: If limit or window_seconds is not positive.
"""
if limit <= 0:
raise ValueError(f"limit must be positive, got {limit}")
if window_seconds <= 0:
raise ValueError(f"window_seconds must be positive, got {window_seconds}")
self._limit = limit
self._window = window_seconds
# key -> deque of event timestamps (sorted in chronological order)
self._buckets: dict[str, deque[float]] = {}
def _now(self, now: float | None) -> float:
"""Return the current time or the provided timestamp."""
return time.monotonic() if now is None else now
def _cleanup(self, key: str, now: float) -> int:
"""
Remove expired timestamps for the given key and return the current count.
If the deque becomes empty, the key is removed from storage.
"""
bucket = self._buckets.get(key)
if bucket is None:
return 0
cutoff = now - self._window
# Remove timestamps that are strictly before the cutoff.
# Since timestamps are non‑decreasing, we can pop from the left.
while bucket and bucket[0] < cutoff:
bucket.popleft()
count = len(bucket)
if count == 0:
# No active events → remove the key entirely to avoid memory leaks.
del self._buckets[key]
return count
def allow(self, key: str, now: float | None = None) -> bool:
"""
Check if an event is allowed and record it if so.
Args:
key: Identifier for the rate limit bucket.
now: Explicit timestamp (must be non‑decreasing for the same key).
Returns:
True if the event is within the limit, False otherwise.
"""
t = self._now(now)
count = self._cleanup(key, t)
if count < self._limit:
# Record the event.
bucket = self._buckets.get(key)
if bucket is None:
bucket = deque()
self._buckets[key] = bucket
bucket.append(t)
return True
return False
def remaining(self, key: str, now: float | None = None) -> int:
"""
Return the number of remaining slots for the given key without recording an event.
Args:
key: Identifier for the rate limit bucket.
now: Explicit timestamp (must be non‑decreasing for the same key).
Returns:
Number of additional events allowed at this moment.
"""
t = self._now(now)
count = self._cleanup(key, t)
return self._limit - count
if __name__ == "__main__":
# Self‑test: demonstrate basic behaviour with two keys and boundary expiration.
limiter = SlidingWindowRateLimiter(limit=2, window_seconds=1.0)
# Use explicit timestamps to control time precisely.
base = 1000.0
# Key "A": allow two events, then a third should be denied.
assert limiter.allow("A", base) is True
assert limiter.remaining("A", base) == 1
assert limiter.allow("A", base + 0.2) is True
assert limiter.remaining("A", base + 0.2) == 0
assert limiter.allow("A", base + 0.3) is False # limit reached
assert limiter.remaining("A", base + 0.3) == 0
# Key "B": independent from A.
assert limiter.allow("B", base) is True
assert limiter.remaining("B", base) == 1
assert limiter.allow("B", base + 0.1) is True
assert limiter.allow("B", base + 0.1) is False # limit reached
# After the window passes, events expire.
# For key A, the first event at base is now older than 1 second.
assert limiter.allow("A", base + 1.1) is True # one slot freed
assert limiter.remaining("A", base + 1.1) == 1 # one event still active (base+0.2)
# The second event (base+0.2) expires after base+1.2.
assert limiter.allow("A", base + 1.3) is True # another slot freed
assert limiter.remaining("A", base + 1.3) == 1 # only the event at base+1.1 remains
# Key B: after window, both events expired.
assert limiter.remaining("B", base + 1.2) == 2 # fully available
# Verify that keys with no active events are removed from storage.
assert "A" in limiter._buckets # still has one event
# Advance time so that all events for A expire.
limiter.allow("A", base + 2.5) # this cleans up old events and adds a new one
assert "A" in limiter._buckets
# Let that last event expire.
limiter.remaining("A", base + 4.0) # cleanup removes it
assert "A" not in limiter._buckets
# Test that limit and window validation works.
try:
SlidingWindowRateLimiter(0, 1.0)
assert False, "Should have raised ValueError"
except ValueError:
pass
try:
SlidingWindowRateLimiter(1, -1.0)
assert False, "Should have raised ValueError"
except ValueError:
pass
print("All tests passed.")