Top-K and streaming selection

Understand the idea

For largest k values, keep a min-heap of at most k candidates. A new value no greater than the root cannot enter the retained top k. This uses memory proportional to k.

Work through an example

For [3,1,5,2,4], k=2, candidate heaps are [3], [1,3], [3,5], unchanged, [4,5].

Time and space complexity

O(n log(k+1)) stream work/O(k) space; final sorted output adds O(k log k).

Why it works

The heap contains the largest min(k,processed_count) values seen, counting duplicates.

Common mistakes

• A max-heap of all n values uses more storage.
• Define k=0 and k>n.

Chapter lab · Python implementation

A heap gives fast access to one extreme, without sorting all stored values. Python heapq is a min-heap API. Our lab keeps the largest k stream elements in a size-k min-heap; the root is the smallest retained candidate and therefore the kth-largest seen value.

This shared lab connects the chapter's topics. Follow the current lesson's worked example first, then trace the complete implementation. Additional algorithms described in topic exercises are variations you can implement independently.

import heapq

def top_k(values,k):
    if k<0: raise ValueError('k must be nonnegative')
    if k==0: return []
    heap=[]
    for x in values:
        if len(heap)<k: heapq.heappush(heap,x)
        elif x>heap[0]: heapq.heapreplace(heap,x)
    return sorted(heap,reverse=True)

class Median:
    def __init__(self): self.low,self.high=[],[]
    def add(self,x):
        heapq.heappush(self.low,-x)
        heapq.heappush(self.high,-heapq.heappop(self.low))
        if len(self.high)>len(self.low):
            heapq.heappush(self.low,-heapq.heappop(self.high))
    def value(self):
        if not self.low: raise ValueError('empty stream')
        if len(self.low)>len(self.high): return -self.low[0]
        return (-self.low[0]+self.high[0])/2

Run and test the lab

Copy the implementation and these assertions into a Python 3 file and run it. An assertion failure means the actual result differs from the expectation. Add the current topic's exercise and boundary tests to deepen your understanding.

assert top_k([3,1,5,2,4],2)==[5,4]
assert top_k([1],0)==[] and top_k([1],3)==[1]
m=Median()
for x in [3,1,5,2]: m.add(x)
assert m.value()==2.5
m.add(4); assert m.value()==3