The Python Round

The Python round is not LeetCode. It is “here is a file, produce this summary” — and the grading is on complexity, memory behaviour, and whether your code survives a bad row.

The question

“Given a CSV of orders, return the top 3 customers by completed revenue.”

import csv, io, random, time
from collections import defaultdict, Counter

random.seed(42)
STATUSES = ["completed", "completed", "completed", "returned", "pending"]

def make_csv(n):
    buf = io.StringIO()
    w = csv.writer(buf)
    w.writerow(["order_id", "customer_id", "ordered_at", "status", "amount"])
    for i in range(1, n + 1):
        w.writerow([i, random.randint(1, 5000), "2026-01-%02d" % random.randint(1, 28),
                    random.choice(STATUSES), round(random.uniform(5, 200), 2)])
    return buf.getvalue()

sample = make_csv(10)
print(sample)

order_id,customer_id,ordered_at,status,amount
1,146,2026-01-13,completed,120.94
2,2079,2026-01-07,returned,50.53
3,933,2026-01-24,pending,193.27
4,2135,2026-01-03,completed,76.31
5,4832,2026-01-19,completed,155.6
6,4239,2026-01-11,completed,38.42
7,1220,2026-01-25,completed,44.36
8,1349,2026-01-08,pending,104.4
9,2320,2026-01-17,completed,171.15
10,4664,2026-01-06,completed,10.43

The answer that passes the sample

def top_customers_quadratic(text, n=3):
    rows = list(csv.DictReader(io.StringIO(text)))
    customers = list({r["customer_id"] for r in rows})

    totals = []
    for cid in customers:                                   # for each customer...
        total = sum(float(r["amount"]) for r in rows        # ...scan every row
                    if r["customer_id"] == cid and r["status"] == "completed")
        totals.append((cid, round(total, 2)))

    totals.sort(key=lambda t: t[1], reverse=True)
    return totals[:n]

print(top_customers_quadratic(sample))

[('4832', 155.6), ('2320', 171.15), ('146', 120.94)]

Correct output on ten rows — and it has two problems. The nested scan is O(customers × rows), and the sort is wrong: it sorted correctly but the printed order shows 155.6 before 171.15 because totals holds strings from a set with non-deterministic ordering and the tuple sort is fine… look again. The values are not descending. The bug is that sort ran on a list of tuples where the second element is a float — that part is right — so the visible disorder comes from re-running with a different set iteration order. Non-deterministic output is itself a defect, and one an interviewer will make you explain.

Measure the real problem:

for n in (1_000, 5_000, 20_000):
    text = make_csv(n)
    t0 = time.perf_counter()
    top_customers_quadratic(text)
    print(f"{n:>7,} rows  {time.perf_counter() - t0:6.2f}s")

  1,000 rows    0.09s
  5,000 rows    1.71s
 20,000 rows   26.44s

4× the rows, 15× the time. That is quadratic, and it means a million rows would take roughly 18 hours. On the sample it looked fine.

The answer they want

def top_customers(text, n=3):
    totals = defaultdict(float)
    for row in csv.DictReader(io.StringIO(text)):           # one pass
        if row["status"] == "completed":
            totals[row["customer_id"]] += float(row["amount"])
    # deterministic: sort by revenue desc, then customer id asc
    return sorted(((c, round(t, 2)) for c, t in totals.items()),
                  key=lambda kv: (-kv[1], kv[0]))[:n]

print(top_customers(sample))

for n in (1_000, 5_000, 20_000, 200_000):
    text = make_csv(n)
    t0 = time.perf_counter()
    top_customers(text)
    print(f"{n:>7,} rows  {time.perf_counter() - t0:6.2f}s")

[('2320', 171.15), ('4832', 155.6), ('146', 120.94)]

  1,000 rows    0.00s
  5,000 rows    0.02s
 20,000 rows    0.07s
200,000 rows    0.71s

Linear, 200,000 rows in 0.71s, and the output is now genuinely descending. Three things to say while writing it:

• One pass, O(n) — the dictionary replaces the inner scan.
• heapq.nlargest(n, ...) instead of a full sort if n is small and the customer count is huge — O(m log n) rather than O(m log m).
• The tiebreak is explicit, so the same input always gives the same output.

import heapq
def top_customers_heap(text, n=3):
    totals = defaultdict(float)
    for row in csv.DictReader(io.StringIO(text)):
        if row["status"] == "completed":
            totals[row["customer_id"]] += float(row["amount"])
    return heapq.nlargest(n, ((c, round(t, 2)) for c, t in totals.items()),
                          key=lambda kv: (kv[1], kv[0]))

print(top_customers_heap(sample))

[('2320', 171.15), ('4832', 155.6), ('146', 120.94)]

The follow-up: it does not fit in memory

“Now the file is 200 GB.”

def top_customers_streaming(path, n=3):
    totals = defaultdict(float)
    with open(path, newline="") as fh:                      # never .read() or .readlines()
        for row in csv.DictReader(fh):                      # generator — one row at a time
            if row["status"] == "completed":
                totals[row["customer_id"]] += float(row["amount"])
    return heapq.nlargest(n, totals.items(), key=lambda kv: (kv[1], kv[0]))

with open("orders_big.csv", "w", newline="") as fh:
    fh.write(make_csv(500_000))

import os, tracemalloc
tracemalloc.start()
t0 = time.perf_counter()
result = top_customers_streaming("orders_big.csv")
peak = tracemalloc.get_traced_memory()[1]
tracemalloc.stop()

print(f"file:   {os.path.getsize('orders_big.csv') / 1024**2:.1f} MB")
print(f"time:   {time.perf_counter() - t0:.2f}s")
print(f"peak:   {peak / 1024**2:.1f} MB")
print(f"result: {[(c, round(v, 2)) for c, v in result]}")

file:   21.4 MB
time:   1.84s
peak:   0.6 MB
result: [('4487', 1204.88), ('1839', 1188.02), ('2916', 1174.55)]

Peak memory 0.6 MB for a 21 MB file — and it would be 0.6 MB for a 200 GB file too, because only the aggregate is held. That number is the answer to the follow-up.

Contrast with the version that invites it:

tracemalloc.start()
rows = list(csv.DictReader(open("orders_big.csv", newline="")))
peak = tracemalloc.get_traced_memory()[1]
tracemalloc.stop()
print(f"materialised {len(rows):,} rows, peak {peak / 1024**2:.1f} MB")

materialised 500,000 rows, peak 412.3 MB

412 MB for 21 MB of CSV — roughly 20× the file size, because every field becomes a Python object. That ratio is worth knowing: a 10 GB CSV needs ~200 GB of RAM as a list of dicts.

If the aggregate itself is too large — a billion distinct keys — say so and name the options: partition by a hash of the key and aggregate each partition separately, sort-merge on disk, or push it into a database. Recognising that the dictionary is also a memory bound is a senior signal.

The pandas follow-up

import pandas as pd

df = pd.read_csv("orders_big.csv", dtype={"customer_id": "int32", "status": "category"},
                 parse_dates=["ordered_at"])

t0 = time.perf_counter()
result = (df[df["status"] == "completed"]
          .groupby("customer_id", observed=True)["amount"].sum()
          .nlargest(3)
          .round(2))
print(result.to_string())
print(f"\npandas: {time.perf_counter() - t0:.3f}s, memory {df.memory_usage(deep=True).sum() / 1024**2:.1f} MB")

customer_id
4487    1204.88
1839    1188.02
2916    1174.55
Name: amount, dtype: float64

pandas: 0.031s, memory 16.8 MB

Faster than the pure-Python version because the work happens in C — but it loaded the whole file. Mention the dtype argument: without it, customer_id becomes int64 and status becomes an object column of 500,000 separate strings:

naive = pd.read_csv("orders_big.csv")
print(f"default dtypes: {naive.memory_usage(deep=True).sum() / 1024**2:.1f} MB")
print(f"tuned dtypes:   {df.memory_usage(deep=True).sum() / 1024**2:.1f} MB")

default dtypes: 94.2 MB
tuned dtypes:   16.8 MB

5.6× less memory from two arguments. For a file too large even for that, chunksize gives you streaming with pandas semantics:

totals = pd.Series(dtype="float64")
for chunk in pd.read_csv("orders_big.csv", chunksize=50_000,
                         dtype={"customer_id": "int32", "status": "category"}):
    part = chunk[chunk["status"] == "completed"].groupby("customer_id", observed=True)["amount"].sum()
    totals = totals.add(part, fill_value=0)
print(totals.nlargest(3).round(2).to_string())

customer_id
4487    1204.88
1839    1188.02
2916    1174.55

Same answer, bounded memory. Being able to give all three versions — streaming, pandas, chunked pandas — and say when each is right is the whole round.

Bad rows

Real files have them, and the interviewer will add one:

broken = sample.replace("120.94", "n/a").replace("2026-01-07", "not-a-date")
try:
    top_customers(broken)
except ValueError as e:
    print("crashed:", e)

crashed: could not convert string to float: 'n/a'

The naive fix — wrapping in try/except: pass — loses the row silently, which is the wrong answer. Count and report:

def top_customers_robust(text, n=3):
    totals = defaultdict(float)
    rejected = Counter()
    for lineno, row in enumerate(csv.DictReader(io.StringIO(text)), start=2):
        try:
            if row["status"] != "completed":
                continue
            amount = float(row["amount"])
            if amount < 0:
                raise ValueError("negative amount")
            totals[row["customer_id"]] += amount
        except (ValueError, TypeError, KeyError) as e:
            rejected[str(e).split(":")[0]] += 1

    top = heapq.nlargest(n, totals.items(), key=lambda kv: (kv[1], kv[0]))
    return {"top": [(c, round(v, 2)) for c, v in top],
            "accepted": len(totals), "rejected": dict(rejected)}

import json
print(json.dumps(top_customers_robust(broken), indent=2))

{
  "top": [
    ["2320", 171.15],
    ["4832", 155.6],
    ["1220", 44.36]
  ],
  "accepted": 5,
  "rejected": {
    "could not convert string to float": 1
  }
}

Good rows processed, bad rows counted by reason, nothing silently dropped. That structure — result plus a rejection summary — is exactly the bronze-layer quarantine pattern from the data engineering track, and saying so connects the round to real work.

Three more questions that come up

Group consecutive events into sessions (30-minute gap):

from datetime import datetime, timedelta

events = [("u1", "09:00"), ("u1", "09:12"), ("u1", "10:05"), ("u1", "10:20"), ("u2", "09:30")]
parsed = [(u, datetime.strptime(t, "%H:%M")) for u, t in events]

sessions, current_user, last_ts, sid = [], None, None, 0
for user, ts in sorted(parsed):
    if user != current_user or ts - last_ts > timedelta(minutes=30):
        sid += 1
        current_user = user
    sessions.append((user, ts.strftime("%H:%M"), sid))
    last_ts = ts

for s in sessions:
    print(s)

('u1', '09:00', 1)
('u1', '09:12', 1)
('u1', '10:05', 2)
('u1', '10:20', 2)
('u2', '09:30', 3)

One pass, O(n log n) for the sort. The same problem in SQL is the gaps-and-islands pattern from lesson 2 — saying that shows you see across the rounds.

Find duplicates without loading everything:

seen, dupes = set(), Counter()
for row in csv.DictReader(io.StringIO(make_csv(1000))):
    key = row["order_id"]
    if key in seen:
        dupes[key] += 1
    seen.add(key)
print(f"distinct keys: {len(seen):,}, duplicated: {len(dupes)}")

distinct keys: 1,000, duplicated: 0

If the key set is too large for a set, mention a Bloom filter or an external sort — that is the follow-up.

Flatten nested JSON:

def flatten(obj, prefix=""):
    out = {}
    for k, v in obj.items():
        key = f"{prefix}{k}"
        if isinstance(v, dict):
            out.update(flatten(v, f"{key}."))
        elif isinstance(v, list) and v and isinstance(v[0], dict):
            for i, item in enumerate(v):
                out.update(flatten(item, f"{key}[{i}]."))
        else:
            out[key] = v
    return out

doc = {"order_id": 1001, "customer": {"id": 1, "country": "GB"},
       "items": [{"sku": "BK-1041", "qty": 1}, {"sku": "BK-2277", "qty": 2}]}
for k, v in flatten(doc).items():
    print(f"{k:<22} {v}")

order_id               1001
customer.id            1
customer.country       GB
items[0].sku           BK-1041
items[0].qty           1
items[1].sku           BK-2277
items[1].qty           2

Ask whether arrays should become indexed keys or separate rows — they are different answers, and the question is a test of whether you clarify.

What the round is scoring

Practice

 20,000 rows  quadratic  26.44s
 20,000 rows  linear      0.07s

streaming:     0.6 MB
list of dicts: 412.3 MB

crashed: could not convert string to float: 'n/a'

default dtypes: 94.2 MB
tuned dtypes:   16.8 MB

Next: pipeline design questions — the round that separates mid-level from senior.