Out of Nowhere
A hard Python interview practice problem on DataDriven. Write the Python and run it against test cases, with instant feedback.
- Domain
- Python
- Difficulty
- hard
- Seniority
- L6
- Asked in
- Meta
Problem
A monitoring service emits readings as (timestamp, value) pairs in time order, and on-call wants the readings that jump away from their recent baseline. A reading is anomalous when it sits more than threshold population standard deviations from the mean of the window readings immediately before it, the current reading excluded. Skip the opening readings that lack a full window of history, and skip any reading whose prior window is perfectly flat.
Every stream has a rhythm. Find the readings that break it.
Example 1
Input:
data = [[0,10],[1,11],[2,10],[3,12],[4,11],[5,10],[6,50],[7,11]], window = 5, threshold = 2
Output:
[{"value":50,"z_score":52.38,"timestamp":6,"rolling_std":0.75,"rolling_mean":10.8}]Example 2
Input:
data = [[0,20],[1,20],[2,20],[3,80]], window = 3, threshold = 2
Output:
[]
Worked solution and explanation
What this really is
This is a rolling z-score wearing an anomaly-detection costume, and the whole problem turns on one boundary: which readings go into the baseline. Anyone can compute a mean and a standard deviation. What separates candidates is whether the current reading sneaks into its own window (it must not) and whether a flat stretch of identical readings detonates a divide-by-zero. Get the window boundary wrong and every spike quietly shrinks, because the spike is sitting inside the baseline it's supposed to stand out from.
How to get there
Step 1: Build the baseline from the PREVIOUS window only
For each index i at or beyond window, take the window readings ending just before i (values[i-window:i], current point excluded) and compute their mean and population standard deviation. The exclusion is the part interviewers actually check.
Step 2: Guard the divide, then apply the threshold
Before dividing, look at rolling_std. On a constant segment it is exactly 0 and the z-score is undefined, so skip the point rather than crash. Otherwise z = (value - mean) / std, and the reading is an anomaly when abs(z) > threshold (strict, not >=).
Step 3: Collect the rounded stats in scan order
Emit one dict per flagged reading with timestamp, value, rolling_mean, rolling_std, and z_score, numeric fields rounded to two decimals, in the order you scanned. The first window readings never produce output.
The solution
def detect_anomalies(data, window, threshold):
import math
times = [t for t, _ in data]
values = [v for _, v in data]
anomalies = []
for i in range(len(values)):
if i < window:
continue
win = values[i - window:i]
rolling_mean = sum(win) / len(win)
variance = sum((x - rolling_mean) ** 2 for x in win) / len(win)
rolling_std = math.sqrt(variance)
if rolling_std == 0:
continue
z_score = (values[i] - rolling_mean) / rolling_std
if abs(z_score) > threshold:
anomalies.append({
"value": values[i],
"z_score": round(z_score, 2),
"timestamp": times[i],
"rolling_std": round(rolling_std, 2),
"rolling_mean": round(rolling_mean, 2),
})
return anomaliesPopulation variance divides by len(win); the std == 0 guard runs before any division.
- Window includes the current point
- values[i-window:i+1] folds the spike into its own baseline. A 50 sitting among readings near 10 drags the mean up and inflates the std, so its z-score collapses toward the threshold and the alert never fires.
- Window excludes the current point
- values[i-window:i] measures the reading against history alone. The baseline stays small and tight, the 50 lands dozens of std away, and the anomaly is unmistakable.
Common follow-up questions
- How would you run this over an unbounded live stream instead of a batch list? (Tests maintaining a fixed-size deque with incrementally updated running sums.)
- What if the series has daily or weekly seasonality the rolling mean ignores? (Tests seasonal decomposition or comparing each point against the same phase in prior cycles.)
- How would you make the baseline adapt when the normal level slowly trends up? (Tests exponentially weighted moving averages so the baseline adapts to drift.)
- A single huge spike poisons the next few windows. How would you make detection robust to that? (Tests robustness of mean/std to outliers and a switch to median plus MAD.)