Sliding medians: expire heap entries by event identity

Last updated: 4 Oct 20269 min read

tutorial

A sliding-median index keeps a max heap for the lower half and a min heap for the upper half of active observations. The lower heap has either the same number of live items as the upper heap or one extra, so its top is the lower median. An expiring observation may sit below the top; deleting it from a binary heap by search would cost linear time. Instead, mark its unique event ID as delayed, decrement the live-size counter for its recorded heap side, and discard the stale record when it reaches a top. Rebalancing moves one visible top across when live counts diverge and updates that event's side map. The stream wrapper removes the event that just left the fixed-width window. Equal values remain distinct because heap entries carry unique IDs.

Operational case

Dispatch waits [47,19,61,29,83,37,19] yield lower medians [47,29,61,37,37] for width three. Width four yields [29,29,37,29], choosing the smaller of the two central sorted values. A stale value may remain in a heap array after its window exit, but it must not count toward balancing or become the reported median. Event IDs are never reused in this example; recycling an ID while an old stale record remains would make delayed-deletion bookkeeping ambiguous. A zero-width or wider-than-stream window is rejected.

Working Python program

import heapq


class ExpiringMedian:
    def __init__(self):
        self.lower = []
        self.upper = []
        self.side = {}
        self.delayed = set()
        self.seen = set()
        self.lower_live = 0
        self.upper_live = 0

    def _prune(self, heap):
        while heap and heap[0][1] in self.delayed:
            _, event_id = heapq.heappop(heap)
            self.delayed.remove(event_id)

    def _balance(self):
        self._prune(self.lower)
        self._prune(self.upper)
        if self.lower_live > self.upper_live + 1:
            negative_value, event_id = heapq.heappop(self.lower)
            heapq.heappush(self.upper, (-negative_value, event_id))
            self.side[event_id] = "upper"
            self.lower_live -= 1
            self.upper_live += 1
            self._prune(self.lower)
        elif self.lower_live < self.upper_live:
            value, event_id = heapq.heappop(self.upper)
            heapq.heappush(self.lower, (-value, event_id))
            self.side[event_id] = "lower"
            self.upper_live -= 1
            self.lower_live += 1
            self._prune(self.upper)

    def add(self, event_id, value):
        if event_id in self.seen:
            raise ValueError("event ID must be new")
        self.seen.add(event_id)
        self._prune(self.lower)
        if not self.lower or value <= -self.lower[0][0]:
            heapq.heappush(self.lower, (-value, event_id))
            self.side[event_id] = "lower"
            self.lower_live += 1
        else:
            heapq.heappush(self.upper, (value, event_id))
            self.side[event_id] = "upper"
            self.upper_live += 1
        self._balance()

    def remove(self, event_id):
        side = self.side.pop(event_id)
        self.delayed.add(event_id)
        if side == "lower":
            self.lower_live -= 1
            self._prune(self.lower)
        else:
            self.upper_live -= 1
            self._prune(self.upper)
        self._balance()

    def lower_median(self):
        if not self.side:
            return None
        self._prune(self.lower)
        return -self.lower[0][0]


def sliding_lower_medians(readings, width):
    if not 1 <= width <= len(readings):
        raise ValueError("window width outside stream")
    index = ExpiringMedian()
    answers = []
    for event_id, value in enumerate(readings):
        index.add(event_id, value)
        if event_id >= width:
            index.remove(event_id - width)
        if event_id >= width - 1:
            answers.append(index.lower_median())
    return answers


dispatch_waits = [47, 19, 61, 29, 83, 37, 19]
print(sliding_lower_medians(dispatch_waits, 3))
print(sliding_lower_medians(dispatch_waits, 4))

Output

[47, 29, 61, 37, 37]
[29, 29, 37, 29]

Time, space, and tradeoff

Across N arrivals, each arrival or expiry triggers at most one cross-heap move, and every heap record is eventually popped once, giving O(N log N) total heap work for the wrapper and O(log N) amortized work per arrival or expiry. A single call can pop many delayed records at once, so its worst-case latency can be O(N log N). Without periodic heap compaction, stale records and the set of seen IDs can occupy O(N) space even though only W events are live. A running-median heap without deletion cannot answer sliding windows correctly; a sorted multiset could instead maintain O(W) active memory with logarithmic edits.

Common Mistakes

Do not count a delayed heap record as a live member.
Do not assume a removed item is always at a heap top.
Do not reuse an event ID while old records may remain.
Do not promise O(W) memory without a stale-record compaction policy.

Connected lessons

Compare this operation boundary with Successor disjoint sets: skip permanently retired slots, Persistent range-distinct counts: keep only the latest position active, Editable substring fingerprints: join hashes in a segment tree, Ball trees: prune exact nearest-depot search with radius bounds, then complete the audit project and decision quiz.

data structures

range-query-structures