Sliding window conversation scores with percentiles

A support agent platform scores every finished conversation (say 0 to 100 from an automated grader). Operations:

• record(timestamp, score): timestamps arrive in non-decreasing order.
• percentile(p, now): the p-th percentile of the scores recorded in the window (now - W, now].

Ask:

• Which percentile definition? Use nearest rank: sort the n scores and return the one at rank ceil(p / 100 × n) (rank 1 for p = 0). It always returns a real score, which dashboards prefer. Interpolated percentiles are a follow-up.
• What range do scores have? Integers 0 to 100 here. That matters a lot: a bounded integer range allows a counting structure.
• Query rate versus record rate. Dashboards query often (p50, p90, p99 every few seconds); records arrive at conversation rate.

Brute force: keep the window in a deque, and on each query copy it, sort it, pick the rank. O(n log n) per query.

Bounded integer scores: counts plus a Fenwick tree.

• A deque of (timestamp, score) holds the window in arrival order. Evicting expired entries from the left is O(1) each, and every entry is evicted at most once.
• A Fenwick tree over the 101 possible scores holds how many window entries have each score. record adds 1 at score; eviction adds −1.
• A percentile query is "find the smallest score s whose count of entries <= s reaches the rank". A Fenwick tree answers that with a binary descent in O(log R), where R = 101.

With R this small, a plain count array and a linear scan (101 steps) is also fine and simpler. Say that, then show the Fenwick version as the answer that still works for R = 10,000 (scores with two decimals).

Arbitrary real scores need an order-statistic structure: a sorted list with binary search (Python's bisect.insort is O(n) per insert but a fast memmove; fine to about 10^5 entries), or a balanced tree or skip list with subtree sizes for O(log n). For millions of entries per window, approximate with a t-digest or a histogram sketch and say what error you accept.

import bisect
import math
from collections import deque
from typing import Deque, List, Optional, Tuple


class ScoreWindow:
    """Percentiles over integer scores in [0, max_score] within a time window."""

    def __init__(self, window: float, max_score: int = 100) -> None:
        self.window = window
        self.size = max_score + 1
        self.tree = [0] * (self.size + 1)     # Fenwick over score values, 1-based
        self.events: Deque[Tuple[float, int]] = deque()
        self.top = 1 << (self.size.bit_length() - 1)

    def record(self, timestamp: float, score: int) -> None:
        if not 0 <= score < self.size:
            raise ValueError("score out of range")
        if self.events and timestamp < self.events[-1][0]:
            raise ValueError("timestamps must not go backwards")
        self.events.append((timestamp, score))
        self._add(score, 1)

    def count(self, now: float) -> int:
        self._expire(now)
        return len(self.events)

    def percentile(self, p: float, now: float) -> Optional[int]:
        self._expire(now)
        n = len(self.events)
        if n == 0:
            return None
        rank = max(1, math.ceil(p / 100 * n))
        return self._kth(rank)

    # ---- internals ----------------------------------------------------------
    def _expire(self, now: float) -> None:
        while self.events and self.events[0][0] <= now - self.window:
            _, score = self.events.popleft()
            self._add(score, -1)

    def _add(self, score: int, delta: int) -> None:
        i = score + 1
        while i <= self.size:
            self.tree[i] += delta
            i += i & -i

    def _kth(self, k: int) -> int:
        """Smallest score whose cumulative count reaches k (binary descent)."""
        pos, step = 0, self.top
        while step:
            nxt = pos + step
            if nxt <= self.size and self.tree[nxt] < k:
                pos = nxt
                k -= self.tree[nxt]
            step >>= 1
        return pos  # 1-based position pos + 1 is score pos


class RealScoreWindow:
    """Any real-valued score: sorted list, O(log n) search, O(n) memmove insert."""

    def __init__(self, window: float) -> None:
        self.window = window
        self.events: Deque[Tuple[float, float]] = deque()
        self.sorted: List[float] = []

    def record(self, timestamp: float, score: float) -> None:
        self.events.append((timestamp, score))
        bisect.insort(self.sorted, score)

    def percentile(self, p: float, now: float) -> Optional[float]:
        while self.events and self.events[0][0] <= now - self.window:
            _, score = self.events.popleft()
            del self.sorted[bisect.bisect_left(self.sorted, score)]
        n = len(self.sorted)
        if n == 0:
            return None
        return self.sorted[max(1, math.ceil(p / 100 * n)) - 1]

A platform handling 2 million conversations a day sees about 23 a second, so a one-hour window holds about 83,000 scores.

• Brute force: sorting 83,000 values per query is about 83,000 × 17 ≈ 1.4M comparisons, roughly 15 ms in Python. Three percentiles every 5 seconds is tolerable for one dashboard, not for thousands of tenants.
• Counts plus Fenwick: each record is 7 tree updates (log2 of 101), each query 7 steps. Memory is the deque (83,000 entries) plus 102 counters.
• If the per-tenant memory of the deque matters, bucket time instead: keep one count array per minute (60 buckets × 101 counters), and a query sums the live buckets. Exact at minute granularity, constant memory per tenant.

1. Out-of-order timestamps. Late events (a conversation scored after a delay) break the deque's order. Use time buckets keyed by minute: a late event goes into its own bucket if that bucket is still in the window, and is dropped otherwise.
2. Interpolated percentiles. Linear interpolation between ranks floor and ceil of p/100 × (n − 1) needs the two neighboring order statistics: two _kth calls.
3. Many tenants and segments. Percentiles per customer, per intent and per channel multiply the structures. Time-bucketed histograms merge by adding counts, so store per-segment buckets and merge at query time.
4. Alerting. "Page someone when p10 drops below 40 for 10 minutes": evaluate on a schedule from the same structure and require the condition to hold across consecutive checks to avoid flapping.
5. Approximate at scale. For unbounded real scores and huge windows, t-digest or DDSketch give mergeable percentiles with bounded relative error and kilobytes of memory.

Mid-level (L4)

• Correct windowing with a deque and a correct nearest-rank percentile, even if each query sorts.
• Handles an empty window, p = 0 and p = 100, and the window boundary.
• Tests against a brute-force reference.

Senior (L5)

• Uses the bounded score range: counts plus a Fenwick tree (or a count array), with O(log R) or O(R) queries, and explains when the simpler array is enough.
• Handles real-valued scores with an order-statistic structure and states its costs.
• Works through window size and query cost with real numbers.

Staff (L6+)

• Moves to time-bucketed, mergeable histograms for late events, many segments and constant memory.
• Knows the approximate options (t-digest, DDSketch) and their error guarantees.
• Connects it to the product: quality dashboards per customer, alert thresholds without flapping.