Implement an AnomalyDetector that flags outliers in a transaction stream in real time. There is no randomness anywhere — this problem is fully deterministic. What matters is the ordering: judge "is this value anomalous" against the state before the update, then update; and handle the degenerate zero-variance state without raising.

model = AnomalyDetector(window=20, threshold=3.0)
model.observe(x)   # bool — True when anomalous, then x is folded into the state

Shared spec — the order of operations

Every decision is computed against the state as it stands before this step's x is folded in. Only after the decision is made does x enter the state (the window or the EWMA). Being judged anomalous never excludes a value from the state — "filter out the outliers and only accumulate the normal ones" is a common assumption, but this problem always folds the value in.

Mode 1 — sliding window (mode="window", the default)

Keep the most recent window observations (evicting the oldest once full). Compute the z-score with the sample standard deviation (ddof=1, dividing by n-1).

$z = \frac{x - \bar{w}}{s_w}, \qquad \text{anomalous} \iff |z| > \text{threshold}$

• With fewer than 2 values in the window there is nothing to judge — return False, without raising.
• When $s_w = 0$ (every value in a window of size ≥ 2 is identical) the z-score is undefined. In that case it is anomalous when x != the mean, and not anomalous when x == the mean.

Mode 2 — EWMA (mode="ewma")

Use an exponentially weighted moving mean and variance instead of a window. When mode="window", alpha is ignored entirely.

• The first observe call has no prior state, so it cannot judge — it always returns False. That call is the initialization: ewma_mean = x, ewma_var = 0.0.

• Later calls judge against the EWMA state before the update: when ewma_var > 0, $z = (x - \text{ewma\_mean}) / \sqrt{\text{ewma\_var}}$; otherwise apply the same boundary rule as the window mode's $s_w=0$ case (not anomalous when x == ewma_mean).

• After judging, update with exactly this formula and no other:

  textCopy

  diff = x - ewma_mean
  incr = alpha * diff
  ewma_mean += incr
  ewma_var  = (1 - alpha) * (ewma_var + diff * incr)
  

  ⚠️ This is the standard Welford-style EWMA variance recurrence. A plausible-looking variant such as ewma_var = (1-alpha)*ewma_var + alpha*diff*diff produces different values — it may coincide for a few steps but it eventually diverges. Implement the formula above exactly.

Auto-calibration — calibrate(historical, target_fp_rate)

model.calibrate(historical, target_fp_rate)   # only self.threshold changes

• Never touch self's live window/EWMA state. Instead build a new detector with the same window, mode, and alpha settings internally and stream historical through it in order.
• For every step where that fresh detector could actually compute a z-score, record |z| into a sample. Exclude steps that could not be judged (fewer than 2 values in the window) and steps where the variance was zero so z is undefined (the first is "not measurable yet", the second is "cannot divide" — neither produced a z).
• Sort the sample descending and use the value at index floor(target_fp_rate * len(sample)) as self.threshold. This guarantees the realized anomaly rate never exceeds the target.
• If the sample is empty (no step was scoreable at all), leave self.threshold unchanged.
• With target_fp_rate <= 0, floor(0 * len) = 0 naturally selects the sample's maximum — that falls out of the formula, it does not need its own branch.

Cooldown — check(x)

model.check(x)   # True = an alert should actually fire

This wraps observe. With cooldown=0 (the default) there is no suppression at all — it behaves identically to L1–L3 (a regression).

• When observe(x) is False, check is False too and no state changes (no side effect on the cooldown bookkeeping).
• When observe(x) is True (that call counts as an "anomaly event"), check is in principle True as well. But if the number of anomaly events counted from the last actual firing through this one is at most cooldown, it is suppressed (check is False, though it still counts as an anomaly event). A suppressed event does not update the "last firing" marker — only events that actually fired reset it.
• This counter measures calls where observe returned True, not the number of check calls — a check call judged normal has nothing to do with it.

Level structure

Each level must preserve the behavior of the previous ones. With cooldown=0, check agrees with observe exactly on whether something is anomalous.

At this level, implement only the sliding-window z-score.

AnomalyDetector(window: int, threshold: float)
observe(x: float) -> bool

• Keep the most recent window observations (evicting the oldest once full).
• observe(x) computes the z-score against the window state before this step's x is folded in (sample standard deviation, ddof=1). It is anomalous (True) when |z| > threshold.
• With fewer than 2 values in the window there is nothing to judge — return False, without raising.
• When every value in the window is identical so the standard deviation is 0: anomalous when x differs from that value, not anomalous when it matches.
• After judging, always append x to the window regardless of the verdict (evicting the oldest value once full).