Coding Fundamentals

What to expect

For staff-level full-stack DS, coding rounds skew toward "production-flavored" problems: write a function that another engineer could review, handle inputs cleanly, and be ready to discuss extensions. Staff-DS JDs name "production code and tests" explicitly.

You should expect:

• 30–60 minute live coding sessions.
• Problems framed in domain language (fraud, scoring, features) rather than pure algorithms.
• Follow-ups that add wrinkles — latency budget, streaming variant, scale requirement.
• An expectation that your code includes types, small functions, and at least one example test.

Classes & testability

When the function has state (a model, a threshold, a feature pipeline), wrap it in a class with explicit dependencies. Makes mocking and testing easy.

from typing import Protocol

class Model(Protocol):
    def predict_proba(self, X: list[list[float]]) -> list[list[float]]: ...

class Scorer:
    def __init__(self, model: Model, threshold: float):
        self.model = model
        self.threshold = threshold

    def score(self, features: list[float]) -> dict:
        prob = self.model.predict_proba([features])[0][1]
        return {"prob": prob, "decision": "flag" if prob >= self.threshold else "pass"}

# Test with a fake model — no need to load a real one
class FakeModel:
    def predict_proba(self, X):
        return [[0.4, 0.6] for _ in X]

def test_scorer_flags_above_threshold():
    s = Scorer(FakeModel(), threshold=0.5)
    assert s.score([1.0, 2.0])["decision"] == "flag"

The senior signal: writing the test alongside the class without being asked.

Generators & streaming

For data that doesn't fit in memory or arrives over time, use generators. They yield one item at a time and never load the whole stream.

from collections import defaultdict

def read_events(path):
    """Stream events from a large file one at a time."""
    with open(path) as f:
        for line in f:
            yield parse(line)

def aggregate_by_user(events):
    counts = defaultdict(int)
    for e in events:
        counts[e["user_id"]] += 1
    return counts

counts = aggregate_by_user(read_events("huge_log.jsonl"))

The interview tell: when the problem says "a large file," reach for generators. When the problem says "streaming," reach for generators or async iteration.

Vectorization vs loops

Loops over arrays / series are 10–100× slower than vectorized operations. For any data of meaningful size, vectorize.

import numpy as np

x = np.random.rand(1_000_000)
y = np.random.rand(1_000_000)

# Slow — Python loop
result_slow = [a * b for a, b in zip(x, y)]

# Fast — vectorized
result_fast = x * y

# Conditional logic: np.where, not Python if/else
flags = np.where(x > 0.95, "high", np.where(x > 0.5, "mid", "low"))

NumPy patterns

Broadcasting

Operations between arrays of different shapes follow rules — the smaller is broadcast over the larger. Master this and a lot of code shrinks.

import numpy as np

# Per-feature standardization without loops
X = np.random.randn(1000, 20)        # 1000 rows, 20 features
mean = X.mean(axis=0)                # shape (20,)
std = X.std(axis=0)                  # shape (20,)
X_normalized = (X - mean) / std      # broadcast (1000, 20) - (20,)

Indexing tricks

• Boolean masking: X[mask] selects rows where mask is True.
• Fancy indexing: X[indices] selects rows in arbitrary order.
• argsort / argpartition: get indices for sorting, no need to sort the actual array.

Pandas patterns

Group-by-aggregate

import pandas as pd

# Multiple aggregations per group
df.groupby('user_id').agg(
    n_events=('event_id', 'count'),
    total_spend=('amount', 'sum'),
    first_event=('event_at', 'min'),
)

# Window functions
df['cumulative_spend'] = df.groupby('user_id')['amount'].cumsum()
df['rank_within_user'] = df.groupby('user_id')['event_at'].rank('dense')
df['days_since_prev'] = df.groupby('user_id')['event_at'].diff().dt.days

Merge / join

Pandas defaults to inner join. Always specify how. Always check row counts before and after a merge — silent row duplication is a common bug.

The asof merge

For "what was the value of X at the time of event Y" — standard time-series alignment problem. Use pd.merge_asof with direction='backward' and a tolerance.

Testing & reproducibility

• Pytest for unit and integration tests.
• Fixtures for shared setup. Don't duplicate the model-loading boilerplate across tests.
• Parameterize tests over cases. One test, many cases, clean output on failure.
• Set random seeds wherever a model fit, sample, or train/test split happens. Reproducibility for free.

Big-O cheat sheet

Live-coding tips for full-stack DS rounds

1. Clarify the input: types, shape, edge cases. fraud-domain-flavored problems often have NULL/missing data buried in the spec.
2. State approach + complexity before coding.
3. Write tests first or alongside. Especially for transformation functions.
4. Use types: def f(x: list[int]) -> dict[str, float]: — interviewers notice.
5. Talk through tradeoffs out loud: "I'm using a dict for O(1) lookup; could also sort and binary-search."
6. Handle edge cases explicitly: empty input, single element, ties.
7. Anticipate the production follow-up: "What would change at 1B rows?" Have an answer.
8. When stuck, narrate — silence is the enemy.

✓ Check yourself

Answer from memory before opening each one.

1. A scoring class loads its model from disk inside its constructor. What does that cost you when you write tests, and how does the chapter’s version avoid it?

   Show answer

   Every test needs the real artifact, runs slowly and cannot control the model’s output, so you cannot test the threshold logic on its own. Passing the model in as a dependency typed by a protocol lets a tiny fake stand in. See classes and testability.

2. You count events per user from a 40 GB log file with a generator pipeline. What is held in memory, and what does still grow with the data?

   Show answer

   Only the current line plus the counter dictionary. Memory scales with the number of distinct users, not with file size. If the user count is itself huge, the counter becomes the bottleneck and you need an external store or a sort-based approach. See generators and streaming.

3. Run the broadcasting snippet and check the column means and standard deviations of the result. What should you see, and which rule makes the subtraction work without a loop?

   Show answer

   Each column mean is about 0 and each standard deviation is about 1. A shape (20,) array lines up with the trailing dimension of the (1000, 20) array and is stretched across all 1000 rows. If the shapes ended in different sizes you would get an error, not a silent result. See NumPy patterns.

4. After a left merge, a table of 100,000 rows has 130,000. What is the likeliest cause, and how do you guard against it?

   Show answer

   Duplicate keys in the right table, so each matching left row is repeated. Check key uniqueness and row counts before and after, and ask pandas to enforce the relationship with the merge validation option. See pandas patterns.

5. A feature job in pandas runs out of memory on 40 GB of events. What are your options, and how do you choose?

   Show answer

   Push the aggregation into SQL where the data lives, or switch to a columnar engine such as DuckDB or Polars that streams and parallelises. Choose by where the data already sits and whether the code must also serve real-time requests. Knowing when to leave pandas is part of the senior signal. See pandas patterns.

6. You need the 100 highest scores out of 10 million. Compare sorting everything with the alternatives, and state the cost of each.

   Show answer

   A full sort costs O(n log n). A heap that keeps the best 100 costs O(n log K), and NumPy’s partition-based indexing finds the top K in roughly linear time without sorting the rest. Sort only those K afterwards. See the Big-O cheat sheet.

Takeaway

• Inject dependencies, such as the model, so tests can use a fake and the logic is tested on its own.
• Use generators for files too large for memory; memory then grows with the aggregate state, not the input.
• Vectorise array and series work, and let broadcasting replace loops over rows and columns.
• Specify the join type, check row counts across every merge, and use as-of merges for point-in-time lookups.
• Pandas is for analysis and prototyping; move to SQL, DuckDB or Polars when data or latency outgrow it.
• State the approach and its complexity before coding, handle edge cases explicitly and write a test alongside.

The next chapter puts these patterns to work on practice problems. → Drillable Problems

Further reading

• Functional Programming HOWTO: Generators — Python documentation. How generators and iterators stream values without materialising a list
• typing: Protocol and support for type hints — Python documentation. Structural typing for dependency injection
• Broadcasting — NumPy documentation. The exact rules for combining arrays of different shapes
• pandas.DataFrame.merge — pandas documentation. Join types and the validate option for catching duplicate keys
• How to parametrize fixtures and test functions — pytest documentation. One test over many cases with clear failure output
• heapq: Heap queue algorithm — Python documentation. Top-K selection with nlargest and nsmallest