Sliding window average with updates

The stream gets values one at a time, each with an index (0, 1, 2, …) in arrival order. Operations:

• add(value): append a value; returns its index.
• update(index, value): change a value that already arrived (a late correction, say a conversation's score revised by a reviewer).
• average(): the mean of the last k values (fewer while the stream is shorter than k).

Ask: can an update target a value that has left the window? (Yes; it must not change the current average, but if a later query covers it, it should count. Part 2 has such queries.) Empty stream? (average() returns None.) Integer or float values? (Use floats for the answer; keep the running sum exact when inputs are integers.)

Brute force: sum the last k values on every average(): O(k) per query.

Running window sum, O(1) everything. Keep all values in an array and a running sum of the window [n - k, n):

• add(v): append, add v to the sum, and if the window now holds k + 1 values, subtract the one that fell out (values[n - k - 1]).
• update(i, v): if i is inside the window, adjust the sum by v - values[i]. Then store v.
• average(): sum / min(n, k).

The trap is forgetting that an update outside the window must not touch the sum, and that the value used when the old element leaves must be its updated value. Storing the current value in the array and reading it at eviction time handles both.

If values are floats, a running sum drifts after millions of adds and subtracts. Either recompute the window sum from scratch every k steps (amortized O(1)) or keep integer inputs exact.

Part 2: averages over any range. "Average of indexes l..r" with updates anywhere needs prefix sums that tolerate point updates: a Fenwick (binary indexed) tree. update and range_sum are both O(log n), and the last-k average is just range_sum(n - k, n - 1).

from typing import List, Optional


class SlidingWindowAverage:
    def __init__(self, k: int) -> None:
        if k <= 0:
            raise ValueError("k must be positive")
        self.k = k
        self.values: List[float] = []
        self.window_sum = 0.0

    def add(self, value: float) -> int:
        self.values.append(value)
        self.window_sum += value
        n = len(self.values)
        if n > self.k:
            self.window_sum -= self.values[n - self.k - 1]
        return n - 1

    def update(self, index: int, value: float) -> None:
        if not 0 <= index < len(self.values):
            raise IndexError(index)
        if index >= len(self.values) - self.k:      # inside the window
            self.window_sum += value - self.values[index]
        self.values[index] = value

    def average(self) -> Optional[float]:
        if not self.values:
            return None
        return self.window_sum / min(len(self.values), self.k)


class Fenwick:
    def __init__(self) -> None:
        self.tree = [0.0]                 # 1-based

    def append(self, value: float) -> None:
        i = len(self.tree)
        # node i covers (i - lowbit(i), i]: sum the children it already has
        total, child, low = value, i - 1, i & -i
        while child > i - low:
            total += self.tree[child]
            child -= child & -child
        self.tree.append(total)

    def add(self, i: int, delta: float) -> None:   # 0-based i
        i += 1
        while i < len(self.tree):
            self.tree[i] += delta
            i += i & -i

    def prefix(self, i: int) -> float:             # sum of [0, i)
        total = 0.0
        while i > 0:
            total += self.tree[i]
            i -= i & -i
        return total


class RangeAverage:
    """Part 2: point updates and the average over any index range."""

    def __init__(self) -> None:
        self.values: List[float] = []
        self.fw = Fenwick()

    def add(self, value: float) -> int:
        self.values.append(value)
        self.fw.append(value)
        return len(self.values) - 1

    def update(self, index: int, value: float) -> None:
        self.fw.add(index, value - self.values[index])
        self.values[index] = value

    def range_average(self, lo: int, hi: int) -> Optional[float]:
        lo, hi = max(lo, 0), min(hi, len(self.values) - 1)
        if lo > hi:
            return None
        return (self.fw.prefix(hi + 1) - self.fw.prefix(lo)) / (hi - lo + 1)

    def last_k_average(self, k: int) -> Optional[float]:
        n = len(self.values)
        return self.range_average(n - k, n - 1)

Fenwick.append builds the new node in O(log n) by summing the existing nodes it covers, so the tree grows with the stream instead of being sized up front.

Space is O(n) for both. If only the window matters and updates outside it are ignored, part 1 can keep just the last k values in a ring buffer: O(k) space.

1. Time-based windows. "Average over the last 5 minutes" instead of the last k values: keep (timestamp, value) in a deque and evict from the left while the oldest is older than now - W. Updates need to find the entry by id, so keep a map from id to its deque slot, or store values in an array and the deque holds indexes.
2. Weighted or exponential averages. An exponentially weighted moving average needs O(1) state and no window at all, but a late update to an old value cannot be applied exactly without the history. Say so.
3. Other window statistics. Max or min over the window needs a monotonic deque (O(1) amortized), but point updates break it; use a segment tree instead. Percentiles need an order-statistic structure (see the percentile variant of this question).
4. Many streams. One window per conversation or per agent: a map from stream id to a window object, with idle streams expired. Memory is the number of active streams times k.
5. Distributed. Partition streams by id across workers so each stream's updates land on one process, which keeps the O(1) logic local.

Mid-level (L4)

• O(1) add and average with a running sum, correct while the stream is shorter than k.
• Handles updates inside the window and ignores updates outside it.
• Writes tests including an update followed by eviction of the updated value.

Senior (L5)

• Explains why eviction must read the updated value and why the array design gets that right.
• Mentions floating-point drift and a fix.
• Moves to a Fenwick or segment tree when queries can cover any range, with O(log n) costs.

Staff (L6+)

• Adapts to time-based windows, many concurrent streams and partitioning across workers.
• Picks structures per statistic (monotonic deque, segment tree, order-statistic tree) and notes which ones survive point updates.
• Ties the design to the use case: late score corrections, dashboards and alert thresholds.