Adjacent Pairs and Running Totals with pairwise and accumulate

You have a list of daily prices. You want each day's change from the day before, and the highest price so far on each day. The first compares neighbours. The second compares each price with everything before it. itertools has a tool for each.

from itertools import accumulate, pairwise

prices = [10.0, 100.0, 50.0, 200.0]
changes = [later - earlier for earlier, later in pairwise(prices)]
peaks = list(accumulate(prices, max))
assert changes == [90.0, -50.0, 150.0]
assert peaks == [10.0, 100.0, 100.0, 200.0]

Compare neighbours with pairwise

pairwise(items) yields each item together with the next one. For [a, b, c] it yields (a, b) and then (b, c). An input with fewer than two items yields nothing.

from itertools import pairwise

assert list(pairwise([1, 2, 3])) == [(1, 2), (2, 3)]
assert list(pairwise([])) == []
assert list(pairwise([42.0])) == []

For a list, indexing over range(len(prices) - 1) gives the same pairs. pairwise needs no index arithmetic, and it also works on a one-shot iterator, which has no length and no indexing.

Keep a running value with accumulate

from itertools import accumulate
from operator import mul

assert list(accumulate([1, 2, 3, 4])) == [1, 3, 6, 10]
assert list(accumulate([10, 100, 50, 200], max)) == [10, 100, 100, 200]
assert list(accumulate([2, 3, 4], mul)) == [2, 6, 24]
assert list(accumulate([1, 2, 3], initial=0)) == [0, 1, 3, 6]

accumulate(items, func) yields a running result. The first result is the first item. Each later result is func(previous_result, next_item). With no func, it adds, so you get running totals. With max, you get running peaks. An empty input yields nothing.

With initial=value, accumulate yields that value first and starts the calculation from it. The output is then one item longer than the input, and even an empty input yields the initial value. Passing initial=None is the same as leaving it out.

Pair each price with its peak

from itertools import accumulate

prices = [10.0, 100.0, 50.0, 200.0]
paired = list(zip(prices, accumulate(prices, max)))
assert paired == [(10.0, 10.0), (100.0, 100.0), (50.0, 100.0), (200.0, 200.0)]

This code loops over prices twice at the same time, once in zip and once inside accumulate. That works because each loop over a list gets its own iterator. Do not change the list while the loops run.

Unpack only fixed-size tuples

pairwise yields tuples of exactly two items, and its type says so. mypy rejects for a, b, c in pairwise(values). The batched function from the previous article yields tuples whose length the type leaves open, because the last batch can be short. So mypy accepts for a, b in batched(values, 2), and the code raises ValueError at runtime when the last batch has one item.

Measure the fall from a peak

The drawdown at a point is how far the price has fallen from its running peak, as a fraction of that peak. It equals 1 - price / peak. For [10, 100, 50, 200], the running peaks are [10, 100, 100, 200] and the drawdowns are [0, 0, 0.5, 0]. The largest one, the maximum drawdown, is 0.5. The exercise skips points whose peak is zero or negative.