1Algorithms

Answer to Question 1.1.

Keep the highest price seen so far, and at each price update the largest gap below it, as in Listing 1.1. It takes $O(n)$ time and $O(1)$ memory, and the example returns 40: the fall from 120 to 80.

def max_drawdown(prices):
    peak = None
    worst = 0
    for p in prices:
        if peak is None or p > peak:
            peak = p
        worst = max(worst, peak - p)
    return worst

A later price can only be measured against a peak that came before it, and the running peak is the best such peak at every moment, so no earlier pair needs looking at again. A list with no fall, or an empty one, returns 0.

Answer to Question 1.2.

Walk the list once with a dictionary from value to position, and look up the partner before inserting the current element, as in Listing 1.2. It takes $O(n)$ average time and $O(n)$ memory. For [3, 4] and target 6 it reports no pair.

def two_sum(xs, target):
    seen = {}                     # value -> a position where it occurred
    for j, x in enumerate(xs):
        i = seen.get(target - x)
        if i is not None:         # looked up before inserting, so i != j
            return i, j
        seen[x] = j
    return None

Each pair is found when its later element arrives, because its earlier element is then in the dictionary, so one pass is enough. The order of the two lines inside the loop is the whole point: since the current element is not yet inserted, it cannot pair with itself, while [3, 3] with target 6 still gives positions 0 and 1.

Answer to Question 1.3.

The fills are 300 at 100, 200 at 101 and 100 at 103, for a total of 60 500 and an average price of about 100.83. Listing 1.3 walks the levels. It takes time proportional to the number of levels touched, not the number of shares.

def sweep(asks, qty):
    """asks: (price, size) from the lowest price up. Returns fills, unfilled size, cost."""
    fills, cost = [], 0
    for price, size in asks:
        if qty == 0:
            break
        take = min(size, qty)
        fills.append((price, take))
        cost += price * take
        qty -= take
    return fills, qty, cost

The cost is $300 \times 100 + 200 \times 101 + 100 \times 103 = 60\,500$, and $60\,500 / 600 \approx 100.83$, which is 0.83 above the best offer: that gap is the price of the order’s size. Prices and sizes stay integers, in ticks and lots, so the running cost is exact.

The edge the function must not hide is an order larger than the book. For 1200 shares it fills 1000 and returns 200 unfilled, at a cost of 101 700. The average price per share filled is 101.70, and dividing by the 1200 requested would give a false 84.75. So compute the average over what was filled, and report what was not. The function also leaves the book as it found it, which a caller may or may not want: say which you chose.

Answer to Question 1.4.

Keep a circular buffer of the last $k$ values and a running total, as in Listing 1.4: on each update subtract the value leaving and add the value arriving. With those floats the last call returns $1/3$, about 0.333, and the true mean is 1.

class RollingMean:
    def __init__(self, k):
        self.k = k
        self.buf = [0] * k        # the last k values, overwritten in a circle
        self.n = 0                # how many values have arrived
        self.total = 0

    def update(self, x):
        i = self.n % self.k
        if self.n >= self.k:
            self.total -= self.buf[i]
        self.buf[i] = x
        self.total += x
        self.n += 1
        return self.total / min(self.n, self.k)

The rounding is what breaks it. Doubles near $10^{16}$ are 2 apart, so 1e16 + 1.0 rounds back to 1e16, and both 1.0 values vanish from the total. When 1e16 leaves the window, the total falls to 0, and then the last 1.0 is added, so the mean comes out as $1/3$ over a window that holds three 1.0 values. The error stays in the total from then on.

The same drift happens on ordinary prices, only more slowly: after two million updates with prices near 100, the total in our test no longer equals the sum of its ten values, though it is off by far less than a millionth. Two cures: hold prices as integer ticks, for which the sum stays exact in Python, and in C++ too until it overflows. Or recompute the total from the buffer every so often.

Answer to Question 1.5.

Use a stack: push each opening bracket, and on each closing one pop and compare, as in Listing 1.5. It takes $O(n)$ time and up to $O(n)$ memory. ([)] returns false, and the empty string returns true.

def balanced(s):
    closes = {")": "(", "]": "[", "}": "{"}
    stack = []
    for c in s:
        if c in closes:
            if not stack or stack.pop() != closes[c]:
                return False
        else:
            stack.append(c)
    return not stack

There are two ways to fail, and both must be caught: a closing bracket with an empty stack or a wrong partner, and a stack left non-empty at the end, as with ((.

Answer to Question 1.6.

It returns 1 for $x = 2$ and 5 for $x = 4$. Listing 1.6 keeps the invariant that the answer lies between lo and hi, both ends included, which is why hi starts at $n$ rather than $n - 1$: $n$ itself is a possible answer.

def lower_bound(xs, target):
    lo, hi = 0, len(xs)               # the answer is always in [lo, hi]
    while lo < hi:
        mid = lo + (hi - lo) // 2
        if xs[mid] < target:
            lo = mid + 1
        else:
            hi = mid
    return lo

If the middle value is too small, the answer is further right, past the middle. If not, the middle itself may be the answer, so hi moves to it and not past it. The range shrinks at every step because mid is always below hi, so the loop ends.

In C++, lo + hi can exceed the largest int once the positions pass about a billion, and signed overflow is undefined behaviour. lo + (hi - lo) / 2 cannot overflow.

Answer to Question 1.7.

After A, B, A, C, B it returns C. Listing 1.7 keeps the symbols seen once in an OrderedDict, and those seen more often in a set.

from collections import OrderedDict


class FirstUnique:
    def __init__(self):
        self.once = OrderedDict()     # seen exactly once, in order of arrival
        self.repeated = set()

    def add(self, symbol):
        if symbol in self.repeated:
            return
        if symbol in self.once:
            del self.once[symbol]
            self.repeated.add(symbol)
        else:
            self.once[symbol] = True

    def first_unique(self):
        return next(iter(self.once), None)

A symbol moves at most once from the dictionary to the set, and never back. An OrderedDict is a hash table threaded on a linked list, so deleting from the middle and reading the first key, which is the answer, are both constant time on average. The memory is proportional to the number of distinct symbols.

In C++ the same structure is a std::list of candidates and an unordered_map from symbol to its place in the list, with a null place meaning repeated.

Answer to Question 1.8.

Walk the list with two pointers and turn each link around, as in Listing 1.8. It takes $O(n)$ time and $O(1)$ extra memory, and an empty list returns an empty list, because the loop does not run and prev is still None.

class Node:
    def __init__(self, val, next=None):
        self.val, self.next = val, next


def reverse(head):
    prev = None
    while head:
        head.next, prev, head = prev, head, head.next
    return prev

At each step everything before head is already reversed, with prev at its front. The assignment of three names is evaluated on the right first, so no pointer is lost. Written out in C++ you need a temporary for next, and forgetting it is the usual bug.

The recursive version is three lines, and it uses one stack frame per node. Python’s default recursion limit is 1000, so on 100 000 nodes it raises RecursionError. In C++ the same code can overflow the stack, and that crash has no exception to catch.

Answer to Question 1.9.

It clears the lowest set bit: 12 is binary 1100 and 12 & 11 is binary 1000, which is 8. Listing 1.9 uses it for all three tasks.

def is_power_of_two(x):
    return x > 0 and x & (x - 1) == 0


def count_set_bits(x):
    n = 0
    while x:
        x &= x - 1                    # clears the lowest set bit
        n += 1
    return n


def next_power_of_two(x):             # the smallest power of two >= x, for x >= 1
    return 1 << (x - 1).bit_length()

Subtracting 1 turns the lowest set bit into 0 and every zero below it into 1, and the & removes those. So x is a power of two exactly when the result is 0, since it then had one set bit. The test must rule out 0 separately, because 0 & -1 is 0 too. The count loops once per set bit, not once per bit.

The rounding works since x - 1 has the same bit length as the answer’s exponent: 1000 rounds to 1024, and 1024 stays 1024. A ring buffer of size $2^m$ turns the wrap of a growing counter i into i & (size - 1), which is one cheap instruction, where % on an unknown size is an integer division. The counter may also overflow an unsigned type safely, because the wrap of a power of two lines up with the mask.

Answer to Question 1.10.

XOR all of $1, \dots, n$ together with all the values present, as in Listing 1.10. Each number present cancels, and what is left is the missing one. For the example it returns 4. It takes $O(n)$ time and $O(1)$ memory.

def missing(seqs, n):
    x = 0
    for i in range(1, n + 1):
        x ^= i
    for s in seqs:
        x ^= s
    return x

The other classic is the sum: $n(n + 1)/2$ minus the total of what is there. It overflows a signed 64-bit integer once $n(n + 1)/2$ passes $2^{63} - 1$, which happens near $n = 4.3 \times 10^9$. Sequence numbers can reach that on a long-running feed, and XOR can never overflow, which is why it is the safer of the two.

If the file is in order, you need neither: compare each number with the last one plus one.

Answer to Question 1.11.

Keep a second stack of the minimum at each depth where it changed, as in Listing 1.11. All three operations are $O(1)$ in the worst case, and the extra memory is at most $O(n)$.

class MinStack:
    def __init__(self):
        self.items = []
        self.mins = []                # the minimum at each depth where it changed

    def push(self, x):
        self.items.append(x)
        if not self.mins or x <= self.mins[-1]:
            self.mins.append(x)

    def pop(self):
        x = self.items.pop()
        if x == self.mins[-1]:
            self.mins.pop()
        return x

    def min(self):
        return self.mins[-1]

A single variable fails on pop: when the current minimum leaves, the variable cannot tell you the previous one without a scan. The second stack holds exactly that history.

The comparison in push must be <= and not <. Push 1, 1 and pop once: the minimum is still 1, and with < the single entry on the second stack would have been popped, leaving nothing.

Answer to Question 1.12.

Push onto inbox. To pop, take from outbox, and when it is empty first move all of inbox across, which reverses it into arrival order, as in Listing 1.12. Each element is moved at most once, so $n$ operations do at most $n$ moves in all: amortised $O(1)$.

class Queue:
    def __init__(self):
        self.inbox, self.outbox = [], []
        self.moves = 0

    def push(self, x):
        self.inbox.append(x)

    def pop(self):
        if not self.outbox:
            while self.inbox:
                self.outbox.append(self.inbox.pop())
                self.moves += 1
        return self.outbox.pop()

Follow one element: pushed once, moved once, popped once. That is three operations, however long the queue gets. The expensive pop pays for the cheap operations that came before it, which is what amortised means. It is not an average over random inputs, and it holds for every sequence.

Order is preserved because outbox is never refilled until it is empty, so everything in it is older than everything in inbox.

On a path with a latency limit, I would not use it. Amortised $O(1)$ still allows one pop that moves a million elements, and that one message is the one that misses the limit. A ring buffer gives a true worst-case constant.

Answer to Question 1.13.

Listing 1.13 splits the text at the point and adds up integers. The float version gives 114 for the text 1.15, which is one tick wrong. The text -0.05 must give $-5$.

def to_ticks(s):
    """The text s, in integer hundredths. ValueError beyond 2 decimals."""
    negative = s.startswith("-")
    whole, _, frac = (s[1:] if negative else s).partition(".")
    if len(frac) > 2 or not (whole + frac).isdigit():
        raise ValueError(s)
    ticks = int(whole or "0") * 100 + int(frac.ljust(2, "0"))
    return -ticks if negative else ticks

1.15 has no exact binary form. The nearest double is slightly below it, the product comes out as 114.99999999999999, and int cuts it towards zero. Rounding first, with round(float(s) * 100), cures this example, and it still sends the value through a float, which goes wrong for prices large enough to lose the digit, and it cannot refuse text with too many decimals.

The sign has to be taken off first and put back at the end. A parser that reads the whole part as an integer sees -0 for -0.05. That is plain 0, so the sign is gone and the result is $+5$: a bid at the wrong side of zero. This matters on spreads, which can be negative.

Answer to Question 1.14.

It returns a sum of 6, from position 1 to position 3, for the first list, and $-1$, the single element at position 1, for the second. Listing 1.14 is Kadane’s algorithm: $O(n)$ time and $O(1)$ memory.

def best_stretch(xs):
    """Largest sum of a non-empty stretch, with its first and last positions."""
    best = (xs[0], 0, 0)
    here, start = xs[0], 0
    for i, x in enumerate(xs[1:], 1):
        if here < 0:                  # a negative past can only hurt
            here, start = x, i
        else:
            here += x
        best = max(best, (here, start, i), key=lambda t: t[0])
    return best

here is the best sum over stretches ending at the current day. If the best stretch ending at the previous day is negative, then dropping it does better, so the new stretch starts today. Otherwise extending it does better. Every stretch ends somewhere, so the largest here ever seen is the answer.

The stem says non-empty. That is why the code starts from the first element and not from 0: starting from 0 returns 0 on an all-negative list, which is the sum of no days. An empty list has no such stretch, so the listing raises IndexError: say that this is your choice.

Answer to Question 1.15.

It returns [0, 0, 24, 0]. Listing 1.15 makes two passes: one multiplies in everything to the left of each position, the other everything to the right. That takes $O(n)$ time, and the output list is the only array.

def product_except_self(xs):
    out = [1] * len(xs)
    left = 1
    for i, x in enumerate(xs):
        out[i] = left                 # the product of everything before i
        left *= x
    right = 1
    for i in range(len(xs) - 1, -1, -1):
        out[i] *= right               # and of everything after i
        right *= xs[i]
    return out

The total product of that list is 0, so dividing it by 0 at position 2 fails, and dividing by 2 at position 0 gives 0, which is right only by luck of the other positions. With division you would need to count the zeros: two or more make every output 0, and exactly one makes every output 0 except at that position.

The list [2, 3, 0, 0] shows the case of two zeros, and it returns all zeros.

Answer to Question 1.16.

Python 3 gives $-4$ and $1$, as // rounds down and the remainder takes the sign of the divisor. C++20 gives $-3$ and $-1$, as / truncates towards zero and the remainder takes the sign of the dividend. In C++ Listing 1.16 corrects for that.

std::int64_t snap_down(std::int64_t x, std::int64_t tick) {
    const std::int64_t r = x % tick;     // has the sign of x
    return r < 0 ? x - r - tick : x - r;
}

In both languages the quotient times the divisor plus the remainder gives back the value, which is why the two answers pair up: $-4 \times 2 + 1$ and $-3 \times 2 - 1$ are both $-7$.

In Python the function is one expression, p - p % tick, because the remainder is never negative for a positive tick. So $-7$ with tick 5 gives $-7 - 3 = -10$. The C++ remainder for those values is $-2$, and subtracting it alone gives $-5$, which is the wrong way: rounding towards zero, not down. So when the remainder is negative, the function takes off one more tick.

Answer to Question 1.17.

Keep a deque of positions whose values strictly decrease, as in Listing 1.17. The front is always the window’s maximum. The example returns [3, 5, 5, 5].

def window_max(xs, k):
    out = []
    q = deque()                       # positions, with their values strictly decreasing
    for i, x in enumerate(xs):
        while q and xs[q[-1]] <= x:   # x outlives and beats these: never a maximum again
            q.pop()
        q.append(i)
        if q[0] <= i - k:             # the front has slid out of the window
            q.popleft()
        if i >= k - 1:
            out.append(xs[q[0]])
    return out

When x arrives, any older value that is no larger can never be a maximum again: x is at least as big and leaves the window later. So those values are popped from the back. Then the front is dropped if it has slid out. What remains is decreasing, and its first value is the largest in the window.

A single step can pop up to $k$ entries, which is why the bound per step looks like $k$. The right count is over the whole run: each position is pushed once and popped at most once, so all the while loops together do at most $n$ pops. The total is $O(n)$, amortised, for any $k$, with $O(k)$ memory.

Answer to Question 1.18.

Use prefix sums and a hash map, as in Listing 1.18. A stretch from $i$ to $j$ sums to $T$ exactly when $\text{prefix}_j - \text{prefix}_{i} = T$, so at each $j$ you count how many earlier prefixes equal $\text{prefix}_j - T$. It takes $O(n)$ average time and $O(n)$ memory, and the example has 3 such stretches.

from collections import Counter


def count_sum(xs, target):
    seen = Counter({0: 1})        # the empty prefix
    prefix = count = 0
    for x in xs:
        prefix += x
        count += seen[prefix - target]
        seen[prefix] += 1         # after the lookup, so an empty span never counts
    return count

For [1, -1, 1], the prefixes are 0, 1, 0, 1. The three stretches are the first element, the last element and the whole list. The counter starts with one empty prefix, so that a stretch beginning at position 0 counts, and the current prefix goes in after the lookup, so that with $T = 0$ an empty stretch does not count.

Answer to Question 1.19.

Sort the opens and closes together as events and sweep, as in Listing 1.19. At equal times, closes must come before opens. It takes $O(n \log n)$ time and $O(n)$ memory. For the two orders, the answer is 1, first reached at time 1.

def peak_alive(orders):
    """orders: (open, close), alive for open <= t < close. Returns (peak, first time)."""
    events = []
    for start, end in orders:
        events.append((start, +1))
        events.append((end, -1))
    events.sort()                 # at equal times -1 sorts first: closes before opens
    alive = peak = 0
    when = None
    for t, delta in events:
        alive += delta
        if alive > peak:
            peak, when = alive, t
    return peak, when

Between events the count does not change, so its peak is reached at an event, and the first time it rises to the peak is an open. The tie-break is where the half-open rule lives. An order closing at 2 is no longer alive at 2, so it must leave before an order opening at 2 enters. Here that needs nothing extra: Python sorts tuples by their second field on a tie, and $-1$ sorts below $+1$.

Answer to Question 1.20.

Add up the notional per symbol in a hash map, then take the top $k$ with a heap of size $k$, as in Listing 1.20. It takes $O(n)$ average time to total and $O(m \log k)$ to select, with $O(m)$ memory.

import heapq
from collections import Counter


def top_k(trades, k):
    """trades: (symbol, price, qty). The k largest notionals, ties broken by symbol name."""
    notional = Counter()
    for symbol, price, qty in trades:
        notional[symbol] += price * qty
    return heapq.nsmallest(k, notional.items(), key=lambda item: (-item[1], item[0]))

heapq.nsmallest with key (-notional, symbol) keeps a heap of the $k$ best candidates so far. If you write the heap by hand, it is a min-heap of notional, whose root is the weakest candidate kept: each new symbol either replaces the root or is thrown away. The tie-break must then be reversed in that heap, because the root has to be the worst, which here means the later name. That inversion is the usual bug. For $k$ close to $m$, sorting is simply better, and quickselect finds the top $k$ set in $O(m)$ on average, if the order within them does not matter.

For trades AAA 100 3, BBB 50 10, CCC 300 1, AAA 100 2 and DDD 10 1, the top three are AAA 500, BBB 500 and CCC 300.

If trades keep arriving, the heap breaks, because a symbol’s total changes while it sits in the heap, or while it sits outside it, and a heap cannot find and raise an entry. One fix is an ordered set of (notional, symbol), plus the map: each trade removes and reinserts one entry, for $O(\log m)$ per trade, and reading the top $k$ costs $O(k)$. Another is to push a new entry per update and discard stale ones when they reach the root.

Answer to Question 1.21.

Keep a heap with one entry per feed, holding that feed’s next pair, as in Listing 1.21. It costs $O(N \log K)$ in time and $O(K)$ in memory. If one feed goes quiet, the generator stops yielding as soon as that feed’s turn comes, and waits: it cannot know whether an older message is still on its way.

import heapq


def merge_feeds(feeds):
    """feeds: iterators of (timestamp, message), each in time order. Yields the merge."""
    its = [iter(f) for f in feeds]
    heap = []
    for n, it in enumerate(its):
        head = next(it, None)
        if head is not None:
            heap.append((head[0], n, head[1]))   # n breaks ties: messages never compared
    heapq.heapify(heap)
    while heap:
        ts, n, message = heap[0]
        yield ts, message
        head = next(its[n], None)
        if head is None:
            heapq.heappop(heap)
        else:
            heapq.heapreplace(heap, (head[0], n, head[1]))

The smallest remaining timestamp is always at the head of some feed, so it is in the heap. After yielding it, you read that feed’s next pair and replace the root, which costs $\log K$. Since a feed has at most one entry in the heap, the pair (timestamp, feed number) is unique, so Python never has to compare two messages.

The generator reads only what it must, so it works on endless feeds: merging one feed ticking every 2 with another ticking every 3 gives 2, 3, 4, 6, 6, with the lower-numbered feed first at the tie.

The quiet feed is the real-world part. A merge in time order can emit a message only when every feed has shown it something later, so the slowest feed sets the pace. The usual fix is a heartbeat, which each feed sends with its current time so that the others can advance, or a timeout after which you mark a feed stale and go on without it, at the price of possible disorder when it comes back.

Answer to Question 1.22.

Keep two heaps, as in Listing 1.22: a max-heap for the smaller half and a min-heap for the larger half, with sizes that differ by at most one. Then add is $O(\log n)$ and median is $O(1)$, with $O(n)$ memory. With integer ticks in 0 to 1000, keep an array of 1001 counts instead.

import heapq


class StreamMedian:
    def __init__(self):
        self.low = []             # the smaller half, negated: a max-heap
        self.high = []            # the larger half: a min-heap

    def add(self, x):
        heapq.heappush(self.low, -heapq.heappushpop(self.high, x))
        if len(self.low) > len(self.high):
            heapq.heappush(self.high, -heapq.heappop(self.low))

    def median(self):
        if len(self.high) > len(self.low):
            return self.high[0]
        return (self.high[0] - self.low[0]) / 2

The median sits at the boundary between the two halves, so only the two roots matter. Each add pushes through high first, so that the smallest of the larger half moves across, then moves one back if low has grown too large. That single route keeps both rules true without a special case: every value in low is at most every value in high, and high holds the extra one when the count is odd. Python’s heapq has only a min-heap, so low stores negated values, and the even case reads $(\text{high}_0 - \text{low}_0) / 2$. For the values 5, 1, 9, 3 it gives 4.

With a bounded range, an add is one increment, $O(1)$. The median can then be found by walking the counts, or kept with a pointer that moves as values arrive; the memory is a constant 1001 counters, however long the stream runs. That is a real saving, since the heaps hold every value for ever.

Before the first add there is no median, and the listing raises IndexError: say that this is your choice.

Answer to Question 1.23.

Keep the first $k$ items. Then the item at 0-based position $i$ replaces a uniformly chosen slot with probability $k/(i+1)$, as in Listing 1.23. This is reservoir sampling: $O(n)$ time and $O(k)$ memory, and every $k$-subset is equally likely.

def reservoir(stream, k, rng):
    sample = []
    for i, x in enumerate(stream):        # i items came before x
        if i < k:
            sample.append(x)
        else:
            j = rng.randrange(i + 1)      # uniform on 0..i, so P(j < k) = k / (i + 1)
            if j < k:
                sample[j] = x
    return sample

The proof goes by induction on the number of items seen, $i$. Suppose after $i$ items each is held with probability $k/i$. Item $i+1$ enters with probability $k/(i+1)$, as the code says. An earlier item survives that step unless the newcomer enters and picks its slot, so it is held with probability

The base case is $i = k$, where all are held. The test checks this identity exactly, and goes further: for $n = 5$ and $k = 2$ it enumerates all 60 equally likely sequences of draws, and each of the 10 subsets appears exactly 6 times. A seeded simulation agrees: with 20 items and $k = 5$, each is kept about a quarter of the time. All 20 agree to within about one percentage point, over 40 000 runs.

Answer to Question 1.24.

Use a doubly linked list in recency order, and a hash map from key to the list iterator, as in Listing 1.24. splice relinks a node to the front in $O(1)$, without touching any iterator. So both operations are $O(1)$ on average, and memory is $O(\text{capacity})$.

template <class K, class V>
class Lru {
public:
    explicit Lru(std::size_t capacity) : capacity_(capacity) {}

    std::optional<V> get(const K& key) {
        auto it = where_.find(key);
        if (it == where_.end()) return std::nullopt;
        items_.splice(items_.begin(), items_, it->second);  // relinks: iterators stay valid
        return it->second->second;
    }

    void put(const K& key, V value) {
        if (capacity_ == 0) return;
        if (auto it = where_.find(key); it != where_.end()) {
            it->second->second = std::move(value);
            items_.splice(items_.begin(), items_, it->second);
            return;
        }
        if (items_.size() == capacity_) {
            where_.erase(items_.back().first);
            items_.pop_back();
        }
        items_.emplace_front(key, std::move(value));
        where_[key] = items_.begin();
    }

private:
    std::size_t capacity_;
    std::list<std::pair<K, V>> items_;  // most recently used at the front
    std::unordered_map<K, typename std::list<std::pair<K, V>>::iterator> where_;
};

Three things must all be cheap: finding a key, moving it to the front, and removing the oldest. The hash map gives the first, and the list gives the other two. The map stores iterators, which is safe only because std::list never invalidates an iterator to a surviving element, and splice does not either. A std::vector would shift elements on every move, costing $O(n)$. A std::deque can only push and pop at its ends, so moving from the middle costs $O(n)$ too, and inserting at the ends invalidates all its iterators.

The test runs random operations against a slow model in Python, under AddressSanitizer and UndefinedBehaviorSanitizer too, and they agree. The eviction removes the map entry before popping the node, because the key is read out of the node. Note that get writes: it changes the list.

Answer to Question 1.25.

Run Floyd’s two pointers, as in Listing 1.25: one moves one step at a time, the other two. If they meet, there is a cycle. Then put one back at the head and move both one step at a time: they meet at the start of the cycle. It takes $O(n)$ time and $O(1)$ memory.

def cycle_start(head):
    slow = fast = head
    while fast and fast.next:
        slow, fast = slow.next, fast.next.next
        if slow is fast:              # they met inside the cycle
            slow = head
            while slow is not fast:
                slow, fast = slow.next, fast.next
            return slow
    return None

Let the tail before the cycle have $a$ nodes, and let the cycle have length $c$. Say they meet after the slow pointer has walked $s$ steps. The fast one has walked $2s$, and the extra $s$ steps must be a whole number of laps, so $s$ is a multiple of $c$. The meeting point is $s$ steps from the head. Walking $a$ more steps from there reaches $s + a$ steps from the head, which is the cycle’s start and then a whole number of laps. A walker from the head reaches the same node after $a$ steps, so the two meet there, and they cannot meet sooner since the walker is not on the cycle before then.

Answer to Question 1.26.

Sort by start, then sweep and extend the last merged halt whenever the next one starts at or before its end, as in Listing 1.26. It returns $(1, 6)$ and $(8, 12)$, for 9 hours in all. It takes $O(n \log n)$ time.

def merge(intervals):
    out = []
    for start, end in sorted(intervals):
        if out and start <= out[-1][1]:
            out[-1][1] = max(out[-1][1], end)
        else:
            out.append([start, end])
    return [tuple(x) for x in out]


def covered(intervals):
    return sum(end - start for start, end in merge(intervals))

After sorting, a new halt cannot overlap anything before the last merged one, because every earlier merged halt ended before the last one began. So one comparison is enough. The max matters: a short halt inside a long one, such as $(2, 3)$ after $(1, 6)$, must not shrink the end. The test start <= end is where touching halts count as one, and changing it to < makes $(8, 10)$ and $(10, 12)$ stay apart.

The total is the sum of the merged lengths, 5 plus 4. Adding the raw lengths gives 2 + 4 + 2 + 2 = 10, which counts the overlap from 2 to 3 twice.

Answer to Question 1.27.

It returns [1, 1, 4, 2, 1, 1, 0, 0]. Listing 1.27 keeps a monotone stack of the days still waiting for a higher price, and each new price settles every waiting day it beats.

def days_until_higher(prices):
    out = [0] * len(prices)
    waiting = []                      # days with no higher price yet
    for i, p in enumerate(prices):
        while waiting and prices[waiting[-1]] < p:
            j = waiting.pop()
            out[j] = i - j
        waiting.append(i)
    return out

The prices on the stack never rise from bottom to top, as any price that beats the top pops it before going on. So if the new price does not beat the top, it beats nothing below either, and the loop can stop.

Take 103 on day 2: it waits while 99, 97 and 100 come in and pile up or pop one another, and it is settled at day 6 by 104, which gives 4.

A single price can pop many entries, so one step can cost a lot. Yet every day is pushed once and popped at most once, so across the whole list the pops number at most $n$: amortised $O(1)$ per day, $O(n)$ in total, and $O(n)$ memory in the worst case.

Answer to Question 1.28.

Largest-first gives 4, 1, 1, which is three orders, while the true answer is 3, 3, which is two. Listing 1.28 uses dynamic programming over quantities, in $O(ns)$ time and $O(n)$ memory for $s$ sizes.

def fewest_clips(sizes, n):
    inf = n + 1
    best = [0] + [inf] * n            # best[q]: fewest clips adding up to exactly q
    used = [0] * (n + 1)
    for q in range(1, n + 1):
        for s in sizes:
            if s <= q and best[q - s] + 1 < best[q]:
                best[q], used[q] = best[q - s] + 1, s
    if best[n] == inf:
        return None
    clips = []
    while n:
        clips.append(used[n])
        n -= used[n]
    return clips

An exact fill of $q$ ends with some last size $s$, and what comes before it is an exact fill of $q - s$. So best[q] is one more than the smallest best[q - s]. Recording the size chosen at each $q$ gives back the list by walking down. If no combination works, best[n] keeps its sentinel, and with sizes 2 and 4 the function returns nothing for $n = 5$.

Note that $ns$ is not polynomial in the input’s length, because $n$ is written in about $\log n$ bits. On the whiteboard the question is rarely large enough to care, but say so if $n$ can be $10^9$.

Answer to Question 1.29.

The answer is 9: buy at 10 and sell at 18 for 6, then buy at 14 and sell at 19 for 3. Listing 1.29 keeps two numbers, the best wealth while flat and while holding, updated once per day. With no fee it returns the sum of all the rises, here 14; with a fee of a million it returns 0.

def max_profit_with_fee(prices, fee):
    flat, holding = 0, float("-inf")  # the best wealth so far, not holding and holding
    for p in prices:
        holding = max(holding, flat - p)
        flat = max(flat, holding + p - fee)
    return flat

At the close of each day you are either flat or holding one share, and each state’s best wealth depends only on yesterday’s two values. To hold today, you held already or you buy from flat. To be flat today, you were flat already or you sell and pay the fee. No trade is ever forced, so flat starts at 0 and can only rise, which is why a large fee gives 0 and not a loss.

Update order is safe here: using today’s holding in today’s flat only allows a buy and a sell at the same price on one day, which loses the fee and so is never chosen.

Answer to Question 1.30.

The distance from ABCD to ACBD is 2: two substitutions, since there is no swap operation here. Listing 1.30 fills the classic table one row at a time, in $O(mn)$ time and $O(\min(m, n))$ memory.

def edit_distance(a, b):
    if len(a) < len(b):
        a, b = b, a                   # keep the shorter one as the row
    row = list(range(len(b) + 1))
    for i, x in enumerate(a, 1):
        diag, row[0] = row[0], i
        for j, y in enumerate(b, 1):
            cost = diag + (x != y)    # substitute, or keep
            diag = row[j]
            row[j] = min(cost, row[j] + 1, row[j - 1] + 1)
    return row[-1]

Let $D(i, j)$ be the distance from the first $i$ characters of one string to the first $j$ of the other. The last operation either keeps or substitutes the last characters, which costs 0 or 1 on top of $D(i - 1, j - 1)$, or it deletes one, or inserts one, each 1 on top of the cell above or to the left. The first row and column are the lengths: $D(i, 0) = i$.

Each cell needs only its left neighbour, the cell above, and the diagonal. A single row plus one saved variable, diag, holds all three.

The price of one row is that you get the number and not the edits. To recover which edits to make, keep the full table, or use Hirschberg’s divide-and-conquer method, which needs linear memory and about twice the time.

Answer to Question 1.31.

Use union-find, as in Listing 1.31: every group is a tree of parent pointers, and two accounts are in one group if they reach the same root. With union by size and path compression, $m$ operations cost $O(m\,\alpha(n))$, where $\alpha$ is the inverse Ackermann function, which is below 5 for any $n$ that fits in memory.

class Groups:
    def __init__(self, n):
        self.parent = list(range(n))
        self.size = [1] * n
        self.count = n

    def find(self, x):
        while self.parent[x] != x:
            self.parent[x] = self.parent[self.parent[x]]   # halve the path
            x = self.parent[x]
        return x

    def link(self, a, b):
        a, b = self.find(a), self.find(b)
        if a == b:
            return
        if self.size[a] < self.size[b]:
            a, b = b, a
        self.parent[b] = a            # the smaller tree goes under the larger
        self.size[a] += self.size[b]
        self.count -= 1

    def same(self, a, b):
        return self.find(a) == self.find(b)

Union by size keeps the trees shallow: an account’s depth grows only when its tree is merged under a larger one, so its tree at least doubles each time, and depth is at most $\log_2 n$. Path halving then flattens what it walks through. The number of groups is a counter that falls by one on each real merge.

What it cannot do is split a group. A link cannot be removed, because the structure forgets which pair made it. If removals are known in advance, process the whole sequence backwards, which turns them into merges.

Answer to Question 1.32.

Listing 1.32 is Kahn’s topological sort: start from tasks waiting on nothing, and whenever a task finishes, release those waiting only on it. It takes $O(n + e)$ time for $e$ pairs. If some task never gets released, there is a cycle. The batch finishes at hour 8.

from collections import deque


def run_order(n, edges):
    """edges: (a, b) means a before b. Returns an order, or None on a cycle."""
    after = [[] for _ in range(n)]
    waiting = [0] * n
    for a, b in edges:
        after[a].append(b)
        waiting[b] += 1
    ready = deque(i for i in range(n) if waiting[i] == 0)
    order = []
    while ready:
        a = ready.popleft()
        order.append(a)
        for b in after[a]:
            waiting[b] -= 1
            if waiting[b] == 0:
                ready.append(b)
    return order if len(order) == n else None

An order exists exactly when the pairs have no cycle. Each task on a cycle is waiting for another on that cycle, so none of them ever reaches a waiting count of zero. Conversely, if the graph has no cycle, some task always has no unfinished dependencies, so the loop finishes all of them. So comparing the length of order with $n$ is the whole cycle test.

With unlimited parallel workers, each task starts when the last of its dependencies ends, so walk the tasks in topological order and take the max. A ends at 3. B ends at 5 and C at 7. D starts at 7 and ends at 8. The chain A, C, D is the critical path, and speeding up B changes nothing.

Answer to Question 1.33.

The answer is 4 moves: down, down, right, right. Listing 1.33 uses breadth-first search, in $O(rc)$ time and memory.

from collections import deque


def shortest_path(grid, start, goal):
    """grid: strings, with # for a wall. Returns the number of moves, or None."""
    dist = {start: 0}
    queue = deque([start])
    while queue:
        r, c = queue.popleft()
        if (r, c) == goal:
            return dist[goal]
        for nr, nc in ((r + 1, c), (r - 1, c), (r, c + 1), (r, c - 1)):
            if (0 <= nr < len(grid) and 0 <= nc < len(grid[0])
                    and grid[nr][nc] != "#" and (nr, nc) not in dist):
                dist[nr, nc] = dist[r, c] + 1  # marked when pushed, not when popped
                queue.append((nr, nc))
    return None

The queue holds cells in order of distance: everything at distance $d$ comes out before anything at $d + 1$. So the first time a cell is reached is by a shortest route, and it never needs another look. Depth-first search has no such order. On an open grid it can wander along a long route first and mark cells as visited, so the first path it finds says nothing about the shortest, and fixing it means trying all paths.

Marking does matter, for cost rather than for correctness. Marked on pop, a cell can be pushed once by each of its up to four neighbours before it is first popped. It still gives the right distance, but the queue is larger, by up to four times, and each repeat scans its neighbours again.

Answer to Question 1.34.

The fastest route from A to D takes 4 microseconds, by A, C, B, D. Listing 1.34 is Dijkstra’s algorithm with a binary heap, in $O((V + E) \log V)$ time. The predecessors give the route, walked back from the target.

import heapq


def fastest(links, source):
    """links: {venue: [(next_venue, cost >= 0), ...]}. Returns distances, predecessors."""
    dist, before = {source: 0}, {}
    heap = [(0, source)]
    while heap:
        d, v = heapq.heappop(heap)
        if d > dist[v]:               # a stale entry: v was reached faster since
            continue
        for w, cost in links.get(v, ()):
            if d + cost < dist.get(w, float("inf")):
                dist[w], before[w] = d + cost, v
                heapq.heappush(heap, (d + cost, w))
    return dist, before

Python’s heap cannot lower a key, so each improvement pushes a new entry, and the entry left behind is skipped when it surfaces. That is the d > dist[v] line. The heap then holds up to $E$ entries, which changes nothing in the bound, since $\log E$ is at most $2 \log V$.

The proof rests on non-negative weights: when the smallest entry comes out, nothing still waiting can reach it more cheaply, as any other route passes through an entry at least as large. A negative link breaks this. With A to B of 2, A to C of 3 and C to B of $-2$, the version with a closed set finalises B at 2, while the true answer is 1. The lazy version above does find 1 when a node is revisited, but its running time can then explode, and if a negative cycle is reachable the loop never ends. For negative links use Bellman–Ford, at $O(VE)$.

Answer to Question 1.35.

The answer is 16 for $d = 2$, from the split 4, 9, 3 and 7, 8, and 13 for $d = 3$, from 4, 9 and 3, 7 and 8. Listing 1.35 uses binary search on the answer, in $O(n \log S)$ time, where $S$ is the sum.

def smallest_largest_day(sizes, days):
    def days_needed(cap):
        used, load = 1, 0
        for s in sizes:
            if load + s > cap:
                used, load = used + 1, 0
            load += s
        return used

    lo, hi = max(sizes), sum(sizes)   # no day can hold less than the largest slice
    while lo < hi:
        mid = (lo + hi) // 2
        if days_needed(mid) <= days:
            hi = mid
        else:
            lo = mid + 1
    return lo

Turn the question around. Given a cap on each day, can it be done within $d$ days? That has a simple answer: fill each day greedily until the next slice would pass the cap. Taking as much as possible today never hurts, since it leaves less for later. And the answer is monotone: if a cap works, every larger cap works. That is what lets you binary search the cap, between the largest slice and the sum.

Positive sizes are what makes the greedy test right. With a negative slice, adding one more may lower a day’s total, so greedy filling is no longer optimal, and its count can even rise with the cap: on [8, 2, -5, 8, -1, 4], caps 8, 9, 10 and 11 give 2, 2, 3 and 2 days. The binary search then has nothing sound to test. The largest slice is no longer a lower bound either, since a day can hold less than one of its slices.

Answer to Question 1.36.

Quickselect: partition around a random pivot, and keep only the side holding position $k$, as in Listing 1.36. On average that is $O(n)$, because the work falls roughly geometrically. Three groups (below, equal, above) and a random pivot avoid both of the bad cases.

import random


def kth_smallest(xs, k, rng=random):
    """The value with k elements below it in sorted order, k from 0."""
    xs = list(xs)
    lo, hi = 0, len(xs)
    while True:
        pivot = xs[rng.randrange(lo, hi)]
        below = [x for x in xs[lo:hi] if x < pivot]
        equal = sum(1 for x in xs[lo:hi] if x == pivot)
        if k < lo + len(below):
            xs[lo:hi] = below
            hi = lo + len(below)
        elif k < lo + len(below) + equal:
            return pivot
        else:
            above = [x for x in xs[lo:hi] if x > pivot]
            lo += len(below) + equal
            xs[lo:hi] = above
            hi = lo + len(above)

With the first element as pivot on a sorted list, each step removes one element only. Finding the largest of 100 then costs $99 + 98 + \dots + 1 = 4950$ comparisons, which is $n(n-1)/2$.

With all values equal and two groups, every value lands in “not below the pivot”, so again only the pivot leaves at each step, and 100 equal values looking for position 99 cost 4950 comparisons too. The three-way split ends at once, as the whole list is the “equal” group and $k$ falls within it, after one pass.

A random pivot makes the bad path unlikely, whatever the input is, so no adversary and no sorted input can force it. If you must have a worst-case guarantee, median of medians gives $O(n)$ in the worst case, with a larger constant.

Answer to Question 1.37.

It returns $(-2, 0, 2)$, $(-2, 1, 1)$, $(-1, -1, 2)$ and $(-1, 0, 1)$. Listing 1.37 sorts, fixes the smallest value of the triple, and walks two pointers inward on the rest. It takes $O(n^2)$ time and $O(1)$ memory beyond the sort and the output.

def three_sum(xs, target=0):
    xs = sorted(xs)
    out = []
    for i, a in enumerate(xs):
        if i and xs[i - 1] == a:
            continue                  # the same first value gives the same triples
        lo, hi = i + 1, len(xs) - 1
        while lo < hi:
            s = a + xs[lo] + xs[hi]
            if s < target:
                lo += 1
            elif s > target:
                hi -= 1
            else:
                out.append((a, xs[lo], xs[hi]))
                lo += 1
                while lo < hi and xs[lo] == xs[lo - 1]:
                    lo += 1
    return out

For a fixed first value, the pair must add up to its negative. On a sorted range, a sum that is too small can only be fixed by moving the left pointer up, and one too large by moving the right pointer down, so each pointer moves at most $n$ times.

Repeats get in at two places, and the code stops both. Equal first values give the same triples, so all but the first are skipped. And after a hit, equal second values give the same pair again, so lo steps past them. The right pointer then needs no extra care: once lo has changed value, the same hi gives a different sum and moves on.

Answer to Question 1.38.

The answer is 5, for the stretch C, C, D, D, D. Listing 1.38 grows a sliding window at the right and shrinks it at the left whenever it holds too many symbols, in $O(n)$ time and $O(k)$ memory.

from collections import Counter


def longest_k_distinct(tape, k):
    count = Counter()
    best = left = 0
    for right, s in enumerate(tape):
        count[s] += 1
        while len(count) > k:
            count[tape[left]] -= 1
            if not count[tape[left]]:
                del count[tape[left]]
            left += 1
        best = max(best, right - left + 1)
    return best

A stretch with at most $k$ distinct symbols contains only such stretches, so the rule is monotone: for each right end, there is a leftmost valid start, and it never moves left as the right end moves right. That is why the pointers only go forward, and each moves at most $n$ times. A symbol’s entry is deleted as its count hits 0, since len(count) is what is being compared with $k$.

The window is a valid stretch at every step, so the largest one seen is the answer, and it also gives you the leftmost start for each end: all later starts up to the end are valid too. This gives the follow-up.

Answer to Question 1.39.

Listing 1.39 is such a table, with open addressing. Clearing a slot on erase breaks get, as a later key that probed past that slot now looks absent. The fix used here is backward-shift deletion: move later entries back into the hole where that is allowed.

// Linear probing with backward-shift deletion. The key 0 marks an empty slot.
class FlatMap {
    std::vector<std::uint64_t> keys_, vals_;
    std::size_t mask_;

    std::size_t home(std::uint64_t k) const {
        return (k * 0x9E3779B97F4A7C15ULL >> 20) & mask_;
    }

public:
    explicit FlatMap(std::size_t power_of_two)
        : keys_(power_of_two), vals_(power_of_two), mask_(power_of_two - 1) {}

    void put(std::uint64_t k, std::uint64_t v) {       // k != 0, and the table never fills
        std::size_t i = home(k);
        while (keys_[i] != 0 && keys_[i] != k) i = (i + 1) & mask_;
        keys_[i] = k;
        vals_[i] = v;
    }
    std::optional<std::uint64_t> get(std::uint64_t k) const {
        for (std::size_t i = home(k); keys_[i] != 0; i = (i + 1) & mask_)
            if (keys_[i] == k) return vals_[i];
        return std::nullopt;
    }
    void erase(std::uint64_t k) {
        std::size_t i = home(k);
        while (keys_[i] != k) {
            if (keys_[i] == 0) return;
            i = (i + 1) & mask_;
        }
        for (std::size_t j = (i + 1) & mask_; keys_[j] != 0; j = (j + 1) & mask_) {
            if (((j - home(keys_[j])) & mask_) >= ((j - i) & mask_)) {  // j may move to i
                keys_[i] = keys_[j];
                vals_[i] = vals_[j];
                i = j;
            }
        }
        keys_[i] = 0;
    }
};

get stops at the first empty slot, which is safe only if no key lies beyond a gap in its probe sequence. After a plain clear, that rule breaks. The usual alternatives are a tombstone, a marker that get walks past and put may reuse, which needs periodic rehashing because tombstones pile up, or the shift used here. A key at slot j can move back to the hole at i only if its home slot is not cyclically between them, which is what the distance test checks. With an unsigned type and a mask, the wrap-around needs nothing more.

Being full is a matter of probe lengths. Near load $\alpha$, an unsuccessful lookup with linear probing inspects about $\frac{1}{2}(1 + 1/(1 - \alpha)^2)$ slots. In a simulation at $2^{16}$ slots that is about 2.5 at load 0.5, and about 50 at load 0.9, and at load 1 a search for an absent key never ends here: there is no empty slot to stop at. So keep the load near a half.

Answer to Question 1.40.

Listing 1.40 holds the tree in an array, with the children of $i$ at $2i + 1$ and $2i + 2$. push and pop take $O(\log n)$, the minimum is at the root in $O(1)$, and the constructor takes $O(n)$.

class MinHeap:
    def __init__(self, items=()):
        self.a = list(items)
        for i in range(len(self.a) // 2 - 1, -1, -1):
            self._down(i)             # from the last parent up to the root

    def push(self, x):
        self.a.append(x)
        i = len(self.a) - 1
        while i and self.a[i] < self.a[(i - 1) // 2]:
            self.a[i], self.a[(i - 1) // 2] = self.a[(i - 1) // 2], self.a[i]
            i = (i - 1) // 2

    def pop(self):
        top, last = self.a[0], self.a.pop()
        if self.a:
            self.a[0] = last
            self._down(0)
        return top

    def _down(self, i):
        while (child := 2 * i + 1) < len(self.a):
            if child + 1 < len(self.a) and self.a[child + 1] < self.a[child]:
                child += 1
            if self.a[i] <= self.a[child]:
                return
            self.a[i], self.a[child] = self.a[child], self.a[i]
            i = child

push puts the item at the end and swaps it upward. pop moves the last item to the root and swaps it downward with the smaller child. Comparing with the smaller child is the point: with the larger, that child could end up above the other one, breaking the order.

The build is linear because of where the work falls. Half the nodes are leaves and sift nothing. A node at height $h$ sifts at most $h$ levels, and there are at most $\lceil n/2^{h+1} \rceil$ nodes at height $h$, so the total is about $\sum_h h\,n/2^{h+1} = n$. Sifting up by pushes is the opposite: most nodes are near the bottom, and each may rise all the way. On a descending list of 1024 items, pushes make 8 204 swaps, while the build makes 1 015.

Answer to Question 1.41.

It returns 9 and 2. Listing 1.41 XORs everything to get $a \oplus b$, picks a bit on which they differ, and XORs again over just the IDs with that bit set. It takes $O(n)$ time and $O(1)$ memory.

def two_singles(ids):
    both = 0
    for x in ids:
        both ^= x                     # the pairs cancel, leaving a ^ b
    low = both & -both                # a bit on which a and b differ
    a = 0
    for x in ids:
        if x & low:
            a ^= x                    # the pairs cancel in this half too
    return a, both ^ a

XOR cancels pairs, so the first pass leaves $9 \oplus 2$, binary $1001 \oplus 0010 = 1011$, which is 11. That is not 0 because the two differ. A set bit of it, here the lowest, both & -both, which gives 1, splits the IDs into two groups. Each ID that appears twice falls in one group with its twin and cancels again, and the two singles fall in different groups, by the choice of the bit. So the XOR of one group is one of the singles, and $a \oplus b \oplus a$ gives the other.

The trick x & -x works because negation in two’s complement flips all bits above the lowest set bit and keeps that bit.

Answer to Question 1.42.

The answers are yes, yes, yes, no, yes. Listing 1.42 keeps the times of the accepted messages still inside the window. That costs $O(N)$ memory and amortised $O(1)$ time, and it never breaks the limit.

from collections import deque


class Limiter:
    """At most n accepted messages in any half-open time span of width t."""

    def __init__(self, n, t):
        self.n, self.t = n, t
        self.accepted = deque()

    def allow(self, now):             # times never go down
        while self.accepted and self.accepted[0] <= now - self.t:
            self.accepted.popleft()
        if len(self.accepted) == self.n:
            return False
        self.accepted.append(now)
        return True

At time 10, the message at 0 has left the window, so 1 and 10 fill it, and the second message at 10 is refused. At 11 the message at 1 has left, so there is room again. Refused messages are not stored, as they did not reach the exchange.

A counter reset every $T$ seconds looks right, and yet it can double the rate at a boundary. With a reset at time 10, messages at 9, 9, 10 and 10 all pass, which is 4 inside a window of 2 seconds, against a limit of 2. What breaking the limit costs you is the venue’s own rule, so read its specification.

A token bucket keeps only a balance and a time, so it uses $O(1)$ memory, and that suits a hot path. But the bucket allows a full burst of $N$ plus refill, so at a rate of $N/T$ it can send up to about $2N$ in a window of $T$, unless you size it down to stay under the exchange’s rule: $N/2$ of each is the usual way to make it safe.

Answer to Question 1.43.

Keep, for each key, a growing array of times and one of values. set appends, at amortised $O(1)$, and get is a binary search at $O(\log m)$ for $m$ entries under that key, as in Listing 1.43. The memory is proportional to all the writes.

from bisect import bisect_right
from collections import defaultdict


class AsOfStore:
    def __init__(self):
        self.times = defaultdict(list)
        self.values = defaultdict(list)

    def set(self, key, value, t):     # times increase for each key
        self.times[key].append(t)
        self.values[key].append(value)

    def get(self, key, t):
        i = bisect_right(self.times.get(key, ()), t)  # a read adds no key
        return self.values[key][i - 1] if i else None

bisect_right returns the position after the last time at or before $t$, so the entry just before it is the answer, and a position of 0 means there is none. A value set at exactly $t$ counts as “at or before”, and choosing bisect_left there is the usual bug: it returns the previous value. For example, with values at 1 and 3, get at 3 gives the second and get at 2 gives the first.

Out-of-order times break the cheap append. bisect.insort keeps the order, and each insert costs $O(m)$ for the shift, which is fine if late arrivals are rare and land near the end. If they are common, use an ordered map, at $O(\log m)$ for both. In either case, say what an insert does to an answer already given: a query for a past time can now return a different value, and research code that needs replays must also store when each value was learned.

Answer to Question 1.44.

Keep a list of the items plus a dictionary from item to position, as in Listing 1.44. pick takes a uniform position in the list. All three operations are $O(1)$ on average.

import random


class RandomSet:
    def __init__(self):
        self.items, self.where = [], {}

    def insert(self, x):
        if x in self.where:
            return False
        self.where[x] = len(self.items)
        self.items.append(x)
        return True

    def remove(self, x):
        if x not in self.where:
            return False
        i, last = self.where.pop(x), self.items.pop()
        if last != x:                 # move the last element into the hole
            self.items[i] = last
            self.where[last] = i
        return True

    def pick(self, rng=random):
        return self.items[rng.randrange(len(self.items))]

A hash set alone has no way to name its $i$-th element, so a uniform pick needs a scan, or else a random walk across buckets, which favours elements that are alone in a bucket. A list alone picks well, and it needs a scan to remove.

The trick for removal is that the list has no need to stay in order. Moving the last element into the hole costs one step and leaves no gap.

The classic bug is at the end of the list. If $x$ is the last element, a naive version pops the last and writes it back into position $i$, which is now past the end, and that either raises an error or leaves a stale entry behind. Here last != x guards it. The other classic bug is to forget to update the moved element’s position in the dictionary.

Uniformity is checked by simulation in the test: after a random sequence of inserts and removes, each of the items still in the set comes out with about the same frequency.

Answer to Question 1.45.

Listing 1.45 is the Fisher–Yates shuffle, which draws each position only from the part not yet fixed. The colleague’s version has $3^3 = 27$ equally likely outcomes, and 27 is not divisible by the $3! = 6$ orders, so it cannot be fair.

import random


def shuffle(xs, rng=random):
    for i in range(len(xs) - 1, 0, -1):
        j = rng.randrange(i + 1)      # from the part not yet fixed, i included
        xs[i], xs[j] = xs[j], xs[i]

Fisher–Yates makes $n$ choices, then $n - 1$, and so on down to 2: that is $n!$ equally likely paths in all, one for each order, so each order comes out with probability $1/n!$. The last position gets each item with probability $1/n$, and then the same is true of what remains.

For the colleague’s version, count by enumeration. Starting from 1, 2, 3, the order 1, 3, 2 comes up 5 times out of 27, as do 2, 1, 3 and 2, 3, 1. The orders 1, 2, 3 and 3, 1, 2 and 3, 2, 1 come up 4 times each. So the bias is $5/27$ against $4/27$. It does not fade for larger $n$, since $n^n$ is never divisible by $n!$ for $n \ge 3$: $n - 1$ divides $n!$ and shares no prime factor with $n^n$.

Answer to Question 1.46.

Walk the tree carrying the open range each key must lie in, as in Listing 1.46. Going left sets a new upper limit, and going right a new lower limit. It takes $O(n)$ time, and $O(h)$ memory for a tree of height $h$. A stack in place of recursion handles 100 000 levels.

def is_bst(node, low=None, high=None):
    """node: (key, left, right) or None. Every key must lie strictly between its limits."""
    stack = [(node, low, high)]
    while stack:
        n, lo, hi = stack.pop()
        if n is None:
            continue
        key, left, right = n
        if (lo is not None and key <= lo) or (hi is not None and key >= hi):
            return False
        stack.append((left, lo, key))
        stack.append((right, key, hi))
    return True

The condition is on whole subtrees, and a node’s direct children do not cover that. Take root 5, with left child 1, with right child 6, and 6 has left child 4. Each node is correct compared with its children, and yet 4 sits in the right subtree of 5, so the tree is invalid. In the listing, by the time you reach 4, its range is from 5 to 6, and 4 falls outside it.

The other classic route is an in-order walk, which must visit the keys in strictly increasing order. It needs only the previous key and is correct, and it takes the same time and memory.

Recursion in Python stops near a depth of 1000 by default, and in C++ a tree that deep can overflow the stack, depending on its size. That is why the listing keeps its own stack.

Answer to Question 1.47.

Listing 1.47 is Welford’s method: it updates the mean and the sum of squared gaps from it, in $O(1)$ per value. On the four values it gives 30. The count, sum and sum of squares method gives $-170.67$ in double precision, a negative variance.

class RunningStats:
    def __init__(self):
        self.n, self.mean, self.m2 = 0, 0.0, 0.0

    def add(self, x):
        self.n += 1
        d = x - self.mean
        self.mean += d / self.n
        self.m2 += d * (x - self.mean)

    def variance(self):               # the sample variance, for n >= 2
        return self.m2 / (self.n - 1)

The formula $(\sum x^2 - (\sum x)^2/n) / (n - 1)$ subtracts two numbers near $4 \times 10^{18}$, which is far above $2^{53}$, so doubles there are about 512 apart. The difference is near 90, so what is left is mostly rounding error. Welford’s works on gaps from the current mean, which are small, so little is lost. The product $d \times (x - \text{mean})$ uses the mean before and after the update, and it is never negative, so the sum stays non-negative.

A check at the extreme: if all values are equal, each gap is 0, so the variance is exactly 0, whatever their size.

Answer to Question 1.48.

Listing 1.48 encodes the differences 1000, 1, $-2$ and 301. They become 2000, 2, 3 and 602 by zigzag, which take 2, 1, 1 and 2 bytes: 6 bytes in all, against 32 for four plain 64-bit integers.

def encode(values):
    out, last = bytearray(), 0
    for v in values:
        d = v - last
        last = v
        z = 2 * d if d >= 0 else -2 * d - 1   # zigzag: 0, -1, 1, -2 become 0, 1, 2, 3
        while z >= 0x80:
            out.append(z & 0x7F | 0x80)
            z >>= 7
        out.append(z)
    return bytes(out)


def decode(data):
    out, last, z, shift = [], 0, 0, 0
    for b in data:
        z |= (b & 0x7F) << shift
        shift += 7
        if b < 0x80:
            last += z // 2 if z % 2 == 0 else -(z + 1) // 2
            out.append(last)
            z, shift = 0, 0
    return out

A varint spends one byte for values below 128, and one more for each further seven bits. So 2 and 3 fit in one byte each, while 602 needs two, because it is at or above 128 and below 16 384.

Zigzag exists for small negative values. Written as a 64-bit two’s complement number, $-2$ has its top bits set, so as an unsigned varint it takes ten bytes, the longest of all. Zigzag folds the number line, so that a small magnitude gives a small code whatever the sign, and $-2$ takes a single byte.

The decoder reverses each step: read seven bits at a time until a byte with a clear top bit, undo the zigzag, add to the running value. If a price is unchanged, the difference is 0, a single byte of 0. So the method is at its best on quiet prices.

Answer to Question 1.49.

Use one ordered map per side, from price to a level holding total quantity and order count, and a hash map from ID to the order and an iterator to its level, as in Listing 1.49. Adding costs $O(\log L)$, since it may need to find or create the level. Canceling is $O(1)$ on average, and the best price is $O(1)$.

enum class Side { Bid, Ask };
struct Level { long qty = 0; long orders = 0; };
struct Top { long price; long qty; };

class Book {
public:
    bool add(long id, Side side, long price, long qty) {
        if (qty <= 0 || orders_.count(id)) return false;
        auto level = levels_[idx(side)].try_emplace(price).first;
        level->second.qty += qty;
        level->second.orders += 1;
        orders_.emplace(id, Order{side, qty, level});
        return true;
    }
    bool cancel(long id) {
        auto it = orders_.find(id);
        if (it == orders_.end()) return false;
        const Order& o = it->second;
        o.level->second.qty -= o.qty;
        if (--o.level->second.orders == 0) levels_[idx(o.side)].erase(o.level);
        orders_.erase(it);
        return true;
    }
    std::optional<Top> best_bid() const {
        return top(levels_[0].crbegin(), levels_[0].crend());
    }
    std::optional<Top> best_ask() const {
        return top(levels_[1].cbegin(), levels_[1].cend());
    }
private:
    using Levels = std::map<long, Level>;
    struct Order { Side side; long qty; Levels::iterator level; };
    static int idx(Side s) { return s == Side::Bid ? 0 : 1; }
    template <class It> static std::optional<Top> top(It first, It last) {
        if (first == last) return std::nullopt;
        return Top{first->first, first->second.qty};
    }
    Levels levels_[2];                         // price -> level; iterators survive inserts
    std::unordered_map<long, Order> orders_;   // id -> where the order sits
};

The ID map is what makes cancels cheap. A cancel names only the ID, so without it you would search every level. Storing the map iterator goes one step further: there is no second lookup by price, since iterators into a std::map survive inserts and erases of other keys. Erasing through an iterator is amortised constant by the standard, and it is safe to remove the level when its last order leaves, so an empty level never shows up as the best. Reading the best uses begin on the ask side and rbegin on the bid side, each constant time.

Prices and quantities are integers, as a floating-point price is a poor map key. The tests run 500 random operations at a time against a brute-force model in Python, including repeated IDs and unknown cancels, and also under AddressSanitizer.

Answer to Question 1.50.

Take the graph with an edge $a \to b$ of weight $-\log r_{ab}$, and find a negative cycle with Bellman–Ford, as in Listing 1.50. A product of rates above 1 is a sum of $-\log$ below 0. It costs $O(n \cdot E)$, at most $O(n^3)$ for a full table. Checking triangles is not enough, because a profitable loop may need more hops.