0

Live top-k best-selling books leaderboard

by Frontier Interview· 1 day ago· ⑂ 0 remixes

Question: Live top-k best-selling books leaderboard

Companies: Snowflake

Difficulty: Medium

Problem statement

A bookstore wants a live leaderboard of its best-selling books. Sales arrive one at a time, and the leaderboard can be read at any moment.

Implement a BookSalesTracker class:

  • BookSalesTracker(): creates an empty tracker.
  • record_sale(book_id: str, quantity: int = 1) -> int: adds quantity copies to the book's running total and returns the new total. quantity is a positive integer.
  • get_sales(book_id: str) -> int: returns the book's total, or 0 if it has never sold.
  • top_k(k: int) -> List[str]: returns up to k book ids with the highest totals, best first. Break ties by the smaller book id. If fewer than k books have sales, return all of them. Return [] when k <= 0.

Constraints

  • Up to 10^5 distinct books and 10^6 calls in total.
  • top_k may be called as often as record_sale, so avoid sorting every book on each read.

Example

t = BookSalesTracker()
t.record_sale("dune", 5)      # 5
t.record_sale("emma", 3)      # 3
t.record_sale("beloved", 3)   # 3
t.top_k(2)                    # ["dune", "beloved"]  (tie at 3 goes to the smaller id)
t.record_sale("emma", 4)      # 7
t.top_k(3)                    # ["emma", "dune", "beloved"]
t.top_k(10)                   # ["emma", "dune", "beloved"]
t.get_sales("missing")        # 0

Interviewers often add parts: returns that lower a total, a leaderboard over only the last hour, per-category leaderboards, or a tracker spread across many machines.

Reconstructed from the reported question; details may differ from what interviewers ask.

Approach

Clarify first. Pin these down before writing code:

  • Order of the leaderboard. Most copies sold first. Ties go to the smaller book id, so the answer is deterministic.
  • What top_k returns. A list of book ids, best first. If fewer than k books have sales, return all of them. k <= 0 returns an empty list.
  • Quantities. A sale adds a positive quantity. Ask whether returns (negative quantities) exist; they are a follow-up below.
  • Read/write mix. Ask whether sales or leaderboard reads are more frequent. The answer picks the data structure.

Brute force. Keep sales: dict[book_id, int]. record_sale is O(1). For top_k, sort all books: O(n log n). Say this first; it is correct and shows the shape of the data.

Better read: a bounded heap. heapq.nlargest(k, ...) scans the dict with a heap of size k: O(n log k) per query. Good when sales are frequent and queries are rare, but every query still touches all n books.

Optimal for a live leaderboard: keep the books ordered at all times. Store one key per book, (-sales, book_id), in a sorted container. Python's smallest key is then the best seller, and ties break by id for free.

  1. record_sale(book, q): look up the old total. If the book is in the order, remove its old key. Add q, then insert the new key.
  2. top_k(k): read the first k keys.

Invariant. The sorted container holds exactly one key per book with sales, and that key matches sales[book]. Every update removes the stale key before inserting the fresh one, so the invariant holds after each call.

In an interview without libraries, a plain list with bisect works. Finding the key is O(log n); inserting and deleting shift memory in O(n), but that is a fast memmove. Name the real fix: a balanced BST, a skip list, or sortedcontainers.SortedList, which make each update O(log n).

Why not a plain max-heap of all books? A heap cannot change one entry's priority cheaply, and reading the top k means popping k items. Lazy deletion (push a new entry, skip stale ones on pop) works, but the heap grows with every sale, and top_k must pop and push back. The sorted order is simpler to get right.

Another option when quantities are small. Group books by total in buckets (total -> set of ids), and track the max total. Single-copy sales move a book up one bucket in O(1), like an LFU cache. It gets messy with large quantities, so mention it only as a trade-off.

Code

from __future__ import annotations

import bisect
import heapq
from typing import Dict, List, Tuple


class BookSalesTracker:
    """Cumulative sales per book with a live top-k leaderboard.

    Order: most copies sold first; ties go to the smaller book id.
    """

    def __init__(self) -> None:
        self._sales: Dict[str, int] = {}
        # Sorted keys (-sales, book_id); index 0 is the best seller.
        # Swap for sortedcontainers.SortedList to make updates O(log n).
        self._order: List[Tuple[int, str]] = []

    def record_sale(self, book_id: str, quantity: int = 1) -> int:
        """Adds quantity copies to book_id. Returns the new total."""
        if quantity <= 0:
            raise ValueError("quantity must be positive")
        old = self._sales.get(book_id, 0)
        if old:
            self._remove_key((-old, book_id))
        new = old + quantity
        self._sales[book_id] = new
        bisect.insort(self._order, (-new, book_id))
        return new

    def get_sales(self, book_id: str) -> int:
        return self._sales.get(book_id, 0)

    def top_k(self, k: int) -> List[str]:
        """Up to k book ids, best seller first."""
        if k <= 0:
            return []
        return [book for _, book in self._order[:k]]

    def top_k_with_sales(self, k: int) -> List[Tuple[str, int]]:
        if k <= 0:
            return []
        return [(book, -neg) for neg, book in self._order[:k]]

    def _remove_key(self, key: Tuple[int, str]) -> None:
        i = bisect.bisect_left(self._order, key)
        # The invariant guarantees the key is present.
        assert i < len(self._order) and self._order[i] == key
        self._order.pop(i)


def top_k_from_counts(sales: Dict[str, int], k: int) -> List[str]:
    """Read-rarely alternative: O(n log k) per query, no extra state."""
    if k <= 0:
        return []
    best = heapq.nsmallest(k, ((-total, book) for book, total in sales.items()))
    return [book for _, book in best]


if __name__ == "__main__":
    t = BookSalesTracker()
    assert t.top_k(3) == []
    t.record_sale("dune", 5)
    t.record_sale("emma", 3)
    t.record_sale("beloved", 3)
    assert t.top_k(2) == ["dune", "beloved"]
    t.record_sale("emma", 4)
    assert t.top_k(3) == ["emma", "dune", "beloved"]
    assert t.top_k_with_sales(1) == [("emma", 7)]
    assert t.top_k(10) == ["emma", "dune", "beloved"]
    assert t.top_k(0) == []
    assert t.get_sales("missing") == 0
    assert top_k_from_counts(t._sales, 2) == t.top_k(2)
    print("ok")

Complexity

With n distinct books:

  • record_sale: O(log n) to find the old key and the insert position, plus O(n) element shifts in the plain list. With a balanced tree, skip list, or SortedList, it is O(log n).
  • top_k(k): O(k).
  • get_sales: O(1).
  • top_k_from_counts (no ordered index): O(n log k) per query, O(1) per sale.
  • Space: O(n) for the dict and the ordered keys.

Test cases

  • Empty tracker: top_k(3) returns [].
  • record_sale("dune", 5), record_sale("emma", 3), record_sale("beloved", 3), then top_k(2) returns ["dune", "beloved"] (tie at 3 goes to the smaller id).
  • Then record_sale("emma", 4): emma has 7, so top_k(3) returns ["emma", "dune", "beloved"].
  • k larger than the number of books: top_k(10) returns all 3 books in order.
  • k = 0 or negative: returns [].
  • Repeated sales of one book: record_sale("a", 1) three times gives get_sales("a") == 3 and exactly one leaderboard entry for a.
  • Unknown book: get_sales("missing") returns 0.
  • Invalid quantity: record_sale("a", 0) raises ValueError.

Follow-ups

  1. Returns and refunds (negative quantity). The same remove-then-insert update works when a total goes down. Decide whether a total can go below zero, and drop a book from the order when it reaches zero.
  2. Top k over the last hour or day. Totals now expire. Keep a queue of (timestamp, book, quantity) and, as time moves on, pop old events and subtract them using the same update path. To save memory, store per-minute buckets per book instead of raw events.
  3. Top k by category or author. Keep one ordered index per category alongside the global one. Each sale updates the global index and the book's category indexes, so cost grows with the categories per book.
  4. Very many books, frequent reads of a small fixed k. Keep only a top-k cache with a "threshold" (the k-th total). A sale for a book below the threshold that does not cross it touches only the dict. This works only while totals never decrease; with refunds you need a full index or a rebuild.
  5. Distributed sales stream. Shard by book id so each shard holds exact totals and its own top-k. A query merges the shards' top-k lists, which is exact because each book lives on one shard. For approximate, low-memory counts across a huge catalog, mention Count-Min Sketch with a heap (heavy hitters) and say what accuracy is lost.

Interviewer expectations

Mid-level (L4)

  • Clarifies tie-breaking and what top_k returns.
  • Gets a correct dict + sort or dict + heap solution, then states its cost.
  • Writes tests for ties, k larger than the catalog, and repeated sales.

Senior (L5)

  • Asks about the read/write mix and picks the structure from it.
  • Keeps an ordered index with remove-then-insert and states the invariant.
  • Explains why a plain heap is awkward for priority updates and knows the lazy-deletion option.
  • Handles refunds and a sliding time window cleanly.

Staff (L6+)

  • Discusses true O(log n) options (balanced BST, skip list, SortedList) and the cost of a list insert.
  • Designs the distributed version: sharding by book, merging per-shard top-k, and freshness of reads.
  • Knows when approximate heavy-hitter methods are acceptable and what they give up.

Sign in to comment, vote or remix.