High Water

An easy Python interview practice problem on DataDriven. Write the Python and run it against test cases, with instant feedback.

Domain
Python
Difficulty
easy
Seniority
L4

Problem

A monitoring service records a stream of integer readings and, after each new reading, reports the largest value seen so far. Given the readings in arrival order, return the sequence it reports, one value per reading.

Every reading remembers the highest that came before.

Example 1

Input:

readings = [3,1,4,1,5,9,2]

Output:

[3,3,4,4,5,9,9]

Example 2

Input:

readings = [5,2,2,8]

Output:

[5,5,5,8]

Worked solution and explanation

What this really is

Strip off the monitoring costume and this is a running maximum: a scan that folds each reading against the peak already seen. Every output position depends on every reading before it, so there is no memoryless transform that produces it. The thing being probed is whether you thread one accumulated value across the stream instead of treating each element in isolation. Most candidates who miss it fall into one of two holes: they write an elementwise map that has no memory and lets the value dip back down, or they recompute max over a growing prefix slice at each position and quietly turn a linear pass into quadratic work.

The trap

The trap is reaching for the shape you already know. A comprehension is the reflex, and the only comprehension that produces the right answer is the one that recomputes max(readings[:i+1]) at every index. It looks clean and it passes the two visible cases, but each step rescans a prefix that grows with the stream, so the total work is 1 + 2 + ... + n. At ten thousand readings that is fifty million comparisons for a problem that needs ten thousand.

Per-slice recompute (rejected)
[max(readings[:i + 1]) for i in range(len(readings))]. Correct output, but every position rescans a longer prefix. O(n^2) time, and it degrades exactly when the stream gets long.
Single scan (accepted)
list(accumulate(readings, max)). One left-to-right pass, each reading compared once against the running peak. O(n) time, O(n) output, and the empty case falls out for free.
Running peak as a scan
from itertools import accumulate


def high_water_marks(readings):
    return list(accumulate(readings, max))

One linear pass; accumulate folds each reading against the peak with max.

If you would rather not lean on accumulate, the explicit fold is the same shape and just as fast. Both are a single pass carrying one variable, and both hand back a real list.

The explicit fold, for comparison
def high_water_marks(readings):
    result = []
    peak = None
    for r in readings:
        peak = r if peak is None else max(peak, r)
        result.append(peak)
    return result

Same O(n) pass written by hand; useful when you want to show the state you are threading.

Common follow-up questions