9. Top-K Trending Intents in a Sliding Window

Problem

A voice platform wants a live "what are people asking about right now" dashboard: at each query moment, report the k most frequent intents among requests seen in the trailing window.

Events and query times both arrive in non-decreasing timestamp order, so this should run as a single forward sweep rather than re-scanning history on every query.

Source: src/9_top_k_trending_intents.py

def top_k_trending(
    events: list[tuple[float, str]],
    window_seconds: float,
    queries: list[float],
    k: int,
) -> list[list[str]]: ...

>>> events = [(0, "weather"), (1, "weather"), (2, "news"), (3, "weather")]
>>> top_k_trending(events, window_seconds=5, queries=[3], k=2)
[['weather', 'news']]

Step-by-Step Approach

1. Maintain a running frequency count (a hashmap from intent to count) and a deque of the events currently inside the window, oldest first.
2. Process queries in order (they're already sorted). For each query time q, first admit every not-yet-seen event with timestamp <= q: append it to the deque and bump its intent's count.
3. Then evict from the front of the deque any event whose timestamp has fallen outside [q - window_seconds, q], decrementing its intent's count (and dropping the key entirely once its count hits zero, so stale zero-count entries don't leak into the top-k step).
4. Compute the top-k intents by count descending, tie-broken by intent name ascending — heapq.nsmallest(k, items, key=...) with a negated-count key does both orderings in one pass.
5. Append that query's top-k list to the results and move to the next query. Because both the event pointer and the deque only ever move forward, each event is admitted once and evicted at most once across the whole run.

The key insight is that because both the events and the queries are pre-sorted, the window's contents can be maintained incrementally (admit new events, evict stale ones) in amortized O(1) per event rather than re-filtering the full event list on every query, which is what turns an O(q × n) naive approach into O(n + q log k).

Reference solution

from collections import defaultdict, deque
import heapq


def top_k_trending(
    events: list[tuple[float, str]],
    window_seconds: float,
    queries: list[float],
    k: int,
) -> list[list[str]]:
    counts: dict[str, int] = defaultdict(int)
    buf: deque[tuple[float, str]] = deque()  # events currently inside the window, oldest first
    ei = 0
    n = len(events)
    results: list[list[str]] = []

    for q in queries:
        # admit every event up to and including this query's timestamp
        while ei < n and events[ei][0] <= q:
            t, intent = events[ei]
            buf.append((t, intent))
            counts[intent] += 1
            ei += 1
        # evict events that fell out of the trailing window [q - window_seconds, q]
        while buf and buf[0][0] < q - window_seconds:
            t, intent = buf.popleft()
            counts[intent] -= 1
            if counts[intent] == 0:
                del counts[intent]
        # top-k by count desc, tie-broken by name asc -- nsmallest on a negated
        # count key gives us both orderings in one O(m log k) pass
        top = heapq.nsmallest(k, counts.items(), key=lambda kv: (-kv[1], kv[0]))
        results.append([name for name, _ in top])

    return results

Key Functions & Tricks

• collections.deque — O(1) append/popleft, ideal for a sliding window's ordered contents.
• collections.defaultdict(int) — frequency counting without existence checks.
• heapq.nsmallest(k, items, key=...) — gets the top-k by a composite sort key in O(m log k) without fully sorting all items.
• Deleting zero-count keys from counts — keeps the top-k step from ever considering intents that have fully aged out of the window.
• Two monotonic pointers (ei for admission, the deque's front for eviction) — the mechanism that makes the whole sweep amortized O(1) per event.

How to Recognize This Pattern