The Event Window

A medium Python interview practice problem on DataDriven. Write the Python and run it against test cases, with instant feedback.

Domain
Python
Difficulty
medium
Seniority
L4
Asked in
Citadel

Problem

A request-tracking service for an API needs to answer one question on demand: how many hits landed in the last five minutes. Implement a HitCounter whose hit(timestamp) records an event at the given second and whose get_hits(timestamp) returns how many recorded events fall inside the trailing 300-second window ending at and including timestamp, so a hit at exactly timestamp - 300 has already aged out. Timestamps arrive in non-decreasing order.

Then implement the graded entrypoint event_window(operations, arguments): given a list of operation names ("HitCounter", "hit", "get_hits") and a parallel list of argument lists, replay them against a single counter and return one result per operation, None for "HitCounter" and "hit" and the integer count for each "get_hits".

A five-minute window is all that matters.

Example

Input:

arguments = [[],[1],[2],[3],[4],[300],[300],[301]], operations = ["HitCounter","hit","hit","hit","get_hits","hit","get_hits","get_hits"]

Output:

[null,null,null,null,3,null,4,3]

Worked solution and explanation

What this really is

Strip the costume and this is a rolling-window count over a sorted event stream: hold the timestamps you have seen, and on each query report how many still fall inside the trailing 300 seconds. Anyone can store every hit and re-scan the whole list per query. What separates a strong candidate is noticing the stream is non-decreasing, so once a timestamp leaves the window it never returns, and you can drop it for good from the front. Get the boundary wrong and you count a hit that aged out exactly 300 seconds ago, or evict one that is still alive.

Break down the requirements

Step 1: Record timestamps on hit

Store each hit timestamp in an ordered structure (a deque) as it arrives.

Step 2: Query hits within the last 300 seconds

Count all stored timestamps h that satisfy timestamp - 300 < h <= timestamp, i.e. those within the trailing 300-second window ending inclusively at timestamp.

Step 3: Optimize for repeated queries

Lazily discard expired timestamps from the front of the queue on each query, so each timestamp is added and removed at most once.

Step 4: Expose a replay driver

Wrap the class in a driver event_window(operations, arguments) that replays the op sequence against one instance and collects the results, since the grader calls a single top-level function.

The solution

Queue-based hit counter with expiration pruning, driven by a replay function
from collections import deque


class HitCounter:
    def __init__(self):
        self.hits = deque()

    def hit(self, timestamp: int):
        self.hits.append(timestamp)

    def get_hits(self, timestamp: int) -> int:
        cutoff = timestamp - 299
        while self.hits and self.hits[0] < cutoff:
            self.hits.popleft()
        return len(self.hits)


def event_window(operations, arguments):
    counter = None
    results = []
    for op, args in zip(operations, arguments):
        if op == "HitCounter":
            counter = HitCounter()
            results.append(None)
        elif op == "hit":
            counter.hit(*args)
            results.append(None)
        elif op == "get_hits":
            results.append(counter.get_hits(*args))
    return results

Common follow-up questions