python 35 lines · 8 steps

Running median with two heaps

A pair of complementary heaps keeps the median available in constant time as values stream in.

Explained by highlit
Advanced heaps streaming invariants data-structures generators
1import heapq
2 
3 
4class MovingMedian:
5 def __init__(self):
6 self._low = []
7 self._high = []
8 
9 def add(self, value):
10 if not self._low or value <= -self._low[0]:
11 heapq.heappush(self._low, -value)
12 else:
13 heapq.heappush(self._high, value)
14 self._rebalance()
15 
16 def _rebalance(self):
17 if len(self._low) > len(self._high) + 1:
18 heapq.heappush(self._high, -heapq.heappop(self._low))
19 elif len(self._high) > len(self._low):
20 heapq.heappush(self._low, -heapq.heappop(self._high))
21 
22 @property
23 def median(self):
24 if not self._low:
25 raise ValueError("no data points yet")
26 if len(self._low) > len(self._high):
27 return float(-self._low[0])
28 return (-self._low[0] + self._high[0]) / 2.0
29 
30 
31def stream_medians(values):
32 tracker = MovingMedian()
33 for value in values:
34 tracker.add(value)
35 yield tracker.median
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Two heaps split around the middle let you read the median from the roots in constant time.
  2. 2Python only ships a min-heap, so negating values turns `_low` into a max-heap.
  3. 3Enforcing a size invariant after every insert is what keeps the median at the heap roots.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
python 
import re
from functools import total_ordering
from typing import Optional
 

Parsing and comparing semantic versions

regex operator-overloading sorting
Intermediate 7 steps
python 
from typing import Any, Sequence, Mapping
 
 
def render_markdown_table(

Rendering an aligned Markdown table in Python

string formatting data transformation closures
Intermediate 8 steps

Share this explainer

Here's the card — post it anywhere.

Running median with two heaps — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code