DSA Online Stock Span

The problem

Online Stock Span takes prices one day at a time and answers each new price at once. The span of a price is the count of days ending today whose prices are at or below today's price. The count stops at the first higher price before today.

Online means each answer is due before the next price arrives. The code may not look ahead. The streams below are the ones this article uses throughout.

streams = [
    ([100, 80, 60, 70, 60, 75, 85], [1, 1, 1, 2, 1, 4, 6]),
    ([50, 40, 60], [1, 1, 3]),
    ([90, 90], [1, 2]),
    ([30, 20, 10], [1, 1, 1]),
    ([42], [1]),
]
for prices, expected in streams:
    print("prices", prices, "-> spans:", expected)

Read one row at a time. In the first row, the closing 85 spans six days back. Only the opening 100 stands above it, and that price stops the count.

Intuition: count backwards from today

The first idea counts backwards from today, one day at a time. It counts each price at or below today's price. The first higher price stops the count.

prices = [90, 70, 75, 80]
span = 0
for p in reversed(prices):
    if p <= 80:
        span += 1
    else:
        break
print("span:", span)

The count gives 3, and the method is plainly correct. The trap is repetition. Every new day recounts days that earlier answers already walked over, and a long stream pays for the same past again and again.

Brute force: recount the past for every new price

The brute force answers each arriving price with a fresh backward scan over all earlier prices. A counter records every price check. Nothing is kept between days except the price list itself.

def spans_brute(prices):
    spans = []
    checks = 0
    for i, price in enumerate(prices):
        span = 0
        for j in range(i, -1, -1):
            checks += 1
            if prices[j] <= price:
                span += 1
            else:
                break
        spans.append(span)
    return spans, checks

spans, checks = spans_brute([100, 80, 60, 70, 60, 75, 85])
print("spans:", spans)
print("price checks:", checks)
spans, checks = spans_brute([30, 20, 10])
print("falling spans:", spans)
print("falling checks:", checks)

The main stream costs a backward walk on most days. A rising stream is the expensive shape, because each new price is the highest so far and its scan walks the whole past. The stack below keeps the past compressed so it is never recounted.

The optimal solution: pairs that carry their spans

A stack holds pairs of a price and its span, and a new price absorbs the spans of every pair it pops before pushing its own pair. That one sentence is the whole method.

A popped pair covers a run of days that the new price now spans in one addition. The runs never need recounting, because each pair already carries its count. Popped pairs are never pushed back.

The rule counts a price equal to today's price as part of the span. The pop test therefore removes pairs whose price is at or below the new price.

class Spanner:
    def __init__(self):
        self.stack = []

    def next(self, price):
        span = 1
        while self.stack and self.stack[-1][0] <= price:
            span += self.stack.pop()[1]
        self.stack.append((price, span))
        return span

streams = [
    ([100, 80, 60, 70, 60, 75, 85], [1, 1, 1, 2, 1, 4, 6]),
    ([50, 40, 60], [1, 1, 3]),
    ([90, 90], [1, 2]),
    ([30, 20, 10], [1, 1, 1]),
    ([42], [1]),
]
for prices, expected in streams:
    spanner = Spanner()
    result = [spanner.next(p) for p in prices]
    print("prices", prices, "->", result, "expected:", expected, "match:", result == expected)

All five streams pass. The first day of any stream always answers 1. Its span holds only itself, because no earlier day exists.

Walkthrough: [100, 80, 60, 70, 75] price by price

The walk feeds five prices to one spanner. The stack prints after each price is answered. Each pair shows the price and the span it carries.

class Spanner:
    def __init__(self):
        self.stack = []

    def next(self, price):
        span = 1
        while self.stack and self.stack[-1][0] <= price:
            span += self.stack.pop()[1]
        self.stack.append((price, span))
        return span

spanner = Spanner()
for price in [100, 80, 60, 70, 75]:
    span = spanner.next(price)
    print("price", price, "span", span, "stack:", spanner.stack)

The price 70 pops the pair for 60 and absorbs its span of 1, so its own span is 2. The price 75 then pops the pair for 70 and absorbs its span of 2, so its span is 3. The pair for 80 stays on the stack, because 80 is higher than 75.

Complexity

The brute force checks up to n prices per day across n days, so its time is O(n^2). Each pair is pushed once and popped at most once, so the spanner's total work across a whole stream is O(n), an average of O(1) per day.

Amortised means the average cost per day stays small across a long run, even when one single day pops many pairs. The stack holds at most one pair per day, so its space is O(n).

class Spanner:
    def __init__(self):
        self.stack = []
        self.pops = 0

    def next(self, price):
        span = 1
        while self.stack and self.stack[-1][0] <= price:
            span += self.stack.pop()[1]
            self.pops += 1
        self.stack.append((price, span))
        return span

for n in (8, 16, 32):
    prices = list(range(1, n + 1))
    checks = n * (n - 1) // 2
    spanner = Spanner()
    for p in prices:
        spanner.next(p)
    print("days", n, "- brute checks:", checks, " spanner pops:", spanner.pops)

The rising stream is the worst shape for the recount and the best demonstration for the stack. At 32 days the recount makes 496 checks. The spanner makes 31 pops in total, fewer than one per day.

Edge cases and pitfalls

Treating an equal price as a stopper is the classic mistake. The span rule counts equal prices, so an equal pair must pop and be absorbed. A strict pop test would undercount every repeated price.

Pushing the new pair before absorbing is a second mistake. The new pair would sit on top of the pairs it must absorb, and the pops would take the wrong entries.

class Spanner:
    def __init__(self):
        self.stack = []

    def next(self, price):
        span = 1
        while self.stack and self.stack[-1][0] <= price:
            span += self.stack.pop()[1]
        self.stack.append((price, span))
        return span

def run_stream(prices):
    spanner = Spanner()
    return [spanner.next(p) for p in prices]

print("empty stream:", run_stream([]))
print("equal prices:", run_stream([90, 90]))
print("falling prices:", run_stream([30, 20, 10]))
print("rising prices:", run_stream([10, 20, 30]))
print("one price:", run_stream([42]))

The rising row answers 1, 2, 3, because each new highest price spans the whole past. The falling row answers 1 every day, because each price is stopped by its higher neighbour at once.