Worked Code Examples
Three fully-solved, verified problems — one per round · companion to the Interview Prep Pack
ALL CODE HERE WAS RUN AND VERIFIED
Each snippet executes and produces the output shown. Read these for structure and idiom, not memorization — the
interviewer wants to watch you build, so internalize the shape of a clean solution: how classes are organized, how a
pipeline is staged, how fine-tuning phases work. Then reproduce the shape on a new problem.
1 · Solved LLD — Rate Limiter
Why this problem: it's a top CodeSignal-style prompt, and it shows the move that matters most under evolving scope — the
Strategy pattern lets you add a whole new algorithm (fixed window → sliding log → token bucket) without touching the caller
or the other strategies. That is exactly the "v3 doesn't force a rewrite" property they test.
The structure-first decision
Sliding window log → a deque of timestamps per key (pop expired from the left).
Token bucket → store tokens + last-refill time; refill lazily on each call.
Fixed window → a count + window-start per key.
All three hide behind one RateLimitStrategy interface; RateLimiter is the facade.
# ============================================================
# SOLVED LLD: Rate Limiter (evolving-scope style)
# Shows: clean OOP, Strategy pattern, extensible to new algorithms
# ============================================================
import time
from abc import ABC, abstractmethod
from collections import deque, defaultdict
class RateLimitStrategy(ABC):
@abstractmethod
def allow(self, key, now):
...
class FixedWindow(RateLimitStrategy):
def __init__(self, limit, window_seconds):
[Link] = limit
[Link] = window_seconds
[Link] = defaultdict(lambda: [0, 0.0]) # key -> [count, window_start]
def allow(self, key, now):
count, start = [Link][key]
if now - start >= [Link]:
[Link][key] = [1, now]
return True
if count < [Link]:
[Link][key][0] += 1
return True
return False
class SlidingWindowLog(RateLimitStrategy):
def __init__(self, limit, window_seconds):
[Link] = limit
[Link] = window_seconds
[Link] = defaultdict(deque) # key -> deque of timestamps
def allow(self, key, now):
q = [Link][key]
while q and now - q[0] >= [Link]:
[Link]()
if len(q) < [Link]:
[Link](now)
return True
return False
class TokenBucket(RateLimitStrategy):
def __init__(self, capacity, refill_per_sec):
[Link] = capacity
[Link] = refill_per_sec
[Link] = defaultdict(lambda: [capacity, None]) # key -> [tokens, last_ts]
def allow(self, key, now):
tokens, last = [Link][key]
if last is None:
last = now
tokens = min([Link], tokens + (now - last) * [Link])
if tokens >= 1:
[Link][key] = [tokens - 1, now]
return True
[Link][key] = [tokens, now]
return False
class RateLimiter:
"""Facade. Swap the strategy without touching callers."""
def __init__(self, strategy: RateLimitStrategy):
[Link] = strategy
def is_allowed(self, key, now=None):
return [Link](key, now if now is not None else [Link]())
# ---- verify ----
rl = RateLimiter(SlidingWindowLog(limit=3, window_seconds=10))
results = [rl.is_allowed("user1", now=t) for t in [0, 1, 2, 3, 11]]
print("SlidingWindow (limit 3/10s):", results) # T,T,T,F,T
tb = RateLimiter(TokenBucket(capacity=2, refill_per_sec=0.5))
results2 = [tb.is_allowed("u", now=t) for t in [0, 0, 0, 4]]
print("TokenBucket (cap 2, 0.5/s):", results2) # T,T,F,T
OUTPUT →
SlidingWindow (limit 3/10s): [True, True, True, False, True]
TokenBucket (cap 2, 0.5/s): [True, True, False, True]
WHAT TO SAY WHILE CODING THIS
"I'll define a strategy interface so when you add a new limiting algorithm later, I add one class and change nothing else."
Then if they escalate ("now support per-tier limits, or distributed limiting across servers"), you extend cleanly: per-tier = a
strategy holding a limit-per-key-class map; distributed = the same interface backed by Redis counters instead of in-
memory dicts.
2 · Solved ML System — Fraud Pipeline End-to-End
Why this problem: it exercises every system-design talking point in real code — class imbalance handling, baseline before
complexity, PR-AUC (not accuracy) for imbalanced data, probability calibration, and picking a threshold by false-positive
budget rather than a default 0.5. The serving stub shows the train/serve boundary.
The pipeline stages
Preprocessing: ColumnTransformer scales numericals, one-hot encodes categoricals — bundled into the model so train
and serve use identical transforms (avoids train/serve skew).
Baseline: logistic regression with class_weight="balanced" for the imbalance.
Stronger model: gradient boosting, wrapped in CalibratedClassifierCV so the output probabilities are trustworthy for a
decision rule.
Decision: threshold chosen off the precision-recall curve to hit a precision floor (the FP budget), recall-first.
# ============================================================
# SOLVED ML SYSTEM: Fraud detection pipeline, end to end
# Shows: imbalanced data, baseline -> GBM, PR-AUC, calibration,
# threshold by FP budget, feature pipeline, serving stub
# ============================================================
import numpy as np
from [Link] import make_classification
from sklearn.model_selection import train_test_split
from [Link] import Pipeline
from [Link] import ColumnTransformer
from [Link] import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from [Link] import GradientBoostingClassifier
from [Link] import CalibratedClassifierCV
from [Link] import average_precision_score, precision_recall_curve
# --- synthetic imbalanced data (0.1% positive, like fraud) ---
X, y = make_classification(n_samples=20000, n_features=8, n_informative=5,
weights=[0.99, 0.01], random_state=0)
num_cols = list(range(6))
cat_cols = [6, 7]
X[:, 6] = (X[:, 6] > 0).astype(int) # pretend categorical
X[:, 7] = (X[:, 7] > 0).astype(int)
Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=0.3, stratify=y, random_state=0)
# --- preprocessing: scale numerical, one-hot categorical ---
pre = ColumnTransformer([
("num", StandardScaler(), num_cols),
("cat", OneHotEncoder(handle_unknown="ignore"), cat_cols),
])
# --- baseline: logistic regression with class weighting for imbalance ---
baseline = Pipeline([("pre", pre),
("clf", LogisticRegression(class_weight="balanced", max_iter=1000))])
[Link](Xtr, ytr)
p_base = baseline.predict_proba(Xte)[:, 1]
print("Baseline LogReg PR-AUC:", round(average_precision_score(yte, p_base), 3))
# --- stronger: gradient boosting, then calibrate probabilities ---
gbm = Pipeline([("pre", pre), ("clf", GradientBoostingClassifier(random_state=0))])
calibrated = CalibratedClassifierCV(gbm, method="isotonic", cv=3)
[Link](Xtr, ytr)
p_gbm = calibrated.predict_proba(Xte)[:, 1]
print("Calibrated GBM PR-AUC:", round(average_precision_score(yte, p_gbm), 3))
# --- pick threshold by a false-positive budget (recall-first) ---
prec, rec, thr = precision_recall_curve(yte, p_gbm)
fp_budget_precision = 0.30 # tolerate flagging where >=30% are real fraud
ok = [Link](prec[:-1] >= fp_budget_precision)[0]
chosen = thr[ok[0]] if len(ok) else 0.5
recall_at = rec[ok[0]] if len(ok) else 0.0
print(f"Threshold @ precision>={fp_budget_precision}: {chosen:.3f}, recall={recall_at:.2f}")
# --- serving stub: what inference looks like in production ---
def score_transaction(feature_row):
p = calibrated.predict_proba(feature_row.reshape(1, -1))[:, 1][0]
return {"fraud_probability": float(p), "decision": "review" if p >= chosen else "approve"}
print("Serving example:", score_transaction(Xte[0]))
OUTPUT →
Baseline LogReg PR-AUC: 0.533
Calibrated GBM PR-AUC: 0.48
Threshold @ precision>=0.3: 0.109, recall=0.57
Serving example: {'fraud_probability': 0.0044, 'decision': 'approve'}
HONEST NOTE ON THE NUMBERS
This runs on synthetic data, so the exact PR-AUC values aren't meaningful — the point is the pipeline shape and the
metric/threshold choices. On real fraud data you'd add the feature store, streaming aggregates, and drift monitoring from
the prep pack. In the interview, narrate those as "in production I'd back this with an online feature store and monitor data
drift," even if you only code the model locally.
3 · Solved — Fine-Tuning / Transfer Learning (PyTorch)
Why this problem: the JD names PyTorch and deep learning, and this is the canonical pattern you should be able to reproduce.
It shows the two-phase fine-tune: freeze the pretrained backbone and train only the new head first (fast, stable), then
unfreeze everything with a much smaller learning rate so you adapt without destroying the pretrained weights. Plus
differential learning rates and early stopping.
The fine-tuning recipe (say these terms)
Transfer learning: reuse a model trained on a large task, adapt to your smaller task.
Freeze then unfreeze: requires_grad = False on the backbone first; train the head; then re-enable.
Differential learning rates: backbone gets a tiny lr (1e-5), the new head a larger one (1e-4) — you barely nudge pretrained
features but learn the head freely.
Early stopping: track validation accuracy, stop when it stops improving (patience) to avoid overfitting.
# ============================================================
# SOLVED: Fine-tuning / transfer learning pattern (PyTorch)
# Shows: freeze pretrained backbone, train new head, then
# unfreeze with a SMALLER lr (differential learning rates),
# early-stopping idea, train/val loop structure
# ============================================================
import torch
import [Link] as nn
from [Link] import DataLoader, TensorDataset
torch.manual_seed(0)
# --- stand-in for a pretrained backbone (in reality: a loaded model) ---
class PretrainedBackbone([Link]):
def __init__(self, in_dim=20, hidden=64):
super().__init__()
[Link] = [Link]([Link](in_dim, hidden), [Link](),
[Link](hidden, hidden), [Link]())
def forward(self, x):
return [Link](x)
class FineTuneModel([Link]):
def __init__(self, backbone, hidden=64, n_classes=2):
super().__init__()
[Link] = backbone
[Link] = [Link](hidden, n_classes) # NEW task-specific head
def forward(self, x):
return [Link]([Link](x))
# --- synthetic downstream task ---
X = [Link](800, 20)
y = (X[:, :5].sum(dim=1) > 0).long()
Xtr, ytr, Xval, yval = X[:600], y[:600], X[600:], y[600:]
train_dl = DataLoader(TensorDataset(Xtr, ytr), batch_size=32, shuffle=True)
model = FineTuneModel(PretrainedBackbone())
loss_fn = [Link]()
def evaluate():
[Link]()
with torch.no_grad():
acc = (model(Xval).argmax(1) == yval).float().mean().item()
[Link]()
return acc
# ---- PHASE 1: freeze backbone, train only the head (fast, stable) ----
for p in [Link]():
p.requires_grad = False
opt = [Link]([Link](), lr=1e-3)
for epoch in range(5):
for xb, yb in train_dl:
opt.zero_grad()
loss = loss_fn(model(xb), yb)
[Link]()
[Link]()
print(f"Phase 1 (head only) val acc: {evaluate():.3f}")
# ---- PHASE 2: unfreeze backbone, fine-tune ALL with smaller lr ----
for p in [Link]():
p.requires_grad = True
# differential learning rates: backbone tiny, head normal
opt = [Link]([
{"params": [Link](), "lr": 1e-5},
{"params": [Link](), "lr": 1e-4},
])
best, patience, bad = 0.0, 3, 0
for epoch in range(10):
for xb, yb in train_dl:
opt.zero_grad()
loss_fn(model(xb), yb).backward()
[Link]()
acc = evaluate()
if acc > best:
best, bad = acc, 0
else:
bad += 1
if bad >= patience: # early stopping
break
print(f"Phase 2 (full finetune) best val acc: {best:.3f}")
OUTPUT →
Phase 1 (head only) val acc: 0.670
Phase 2 (full finetune) best val acc: 0.695
HOW TO POSITION THIS IN THE INTERVIEW
Be honest that your deep hands-on is lighter than your classical-ML and evaluation depth — but show you know the
correct mechanics: freeze/unfreeze, differential lr, early stopping, calibration. Pair it with your real frontier-model work
(LLM agent experiments on Gemini/GPT-4, adversarial eval). Knowing the right recipe + having real eval experience beats
overclaiming framework fluency an engineer will probe.
Quick package map (what you import for what)
scikit-learn — classical models, pipelines, preprocessing, metrics, calibration. Tabular default.
xgboost / lightgbm / catboost — gradient-boosted trees, the tabular accuracy winners.
torch (PyTorch) — deep learning, fine-tuning, custom training loops. JD's first-listed framework.
transformers (Hugging Face) — pretrained NLP/LLM models + their fine-tuning utilities ( Trainer ).
pandas / numpy — data wrangling and arrays, everywhere.
redis / kafka clients — the online feature store and streaming layer you describe in system design.