Slotting Architecture · 10 min read

How to Detect Anomalies in Inventory Feeds

Your inventory feed lands on schedule, parses cleanly, and passes schema validation — but this run has 4,000 rows where every run this month had 10,000, and the null-rate on sku_id just jumped from zero to eight percent. Nothing errored; the data is simply wrong, and if you score it you demote half your fast-movers. This page builds one focused function that flags an anomalous feed run against its own recent history using a robust median/IQR test with an optional EWMA path for trending metrics. It is the detector behind the Data-Quality Monitoring for Inventory Feeds guide, which sits inside the wider Velocity Data Ingestion & WMS Sync Pipelines architecture.

Prerequisites

Confirm each before wiring the detector into the ingestion path:

  • Python 3.10+ — the code uses X | None unions, list[...] generics, and dataclass config.
  • A short metric history per feed — 20 to 60 prior runs’ summary metrics (row count, null rates, numeric ranges). The detector compares against this baseline; four runs is the hard floor below which it falls back to static thresholds.
  • Per-run summary metrics, not raw payloads — the function scores row_count, null_rate, and a numeric field’s value against history. Compute those once at ingestion; the detector never re-scans the payload.
  • A baseline that excludes quarantined runs — history must contain only good runs, or a prior bad run drags the band toward itself and hides the next anomaly.
  • A place to route flags — a quarantine queue and an alerting channel, so a flagged run holds the last-good aggregate rather than reaching the scoring engine.

Configuration Block

Every tunable lives in one profile. The two levers that decide sensitivity are iqr_k (the Tukey fence multiplier for the robust test) and ewma_z (the z-threshold for the trending test).

# feed_anomaly.yaml — one profile per feed
anomaly:
  history_window: 40            # prior runs kept for the baseline
  min_history: 4               # below this, fall back to static floors only
  iqr_k: 1.5                   # Tukey fence: value beyond Q1/Q3 +/- k*IQR is flagged
  ewma_alpha: 0.3              # EWMA smoothing for trending metrics (higher = faster)
  ewma_z: 3.0                  # z-score on EWMA residual that flags a trending metric
  static_floor:
    min_row_ratio: 0.6         # cold-start: rows below 60% of expected is a flag
    max_null_rate: 0.02        # cold-start: null rate above 2% is a flag
    expected_rows: 10000       # expected row count used only before history warms up
# Equivalent Python config dict consumed by the detector
FEED_ANOMALY = {
    "history_window": 40,
    "min_history": 4,
    "iqr_k": 1.5,
    "ewma_alpha": 0.3,
    "ewma_z": 3.0,
    "static_floor": {
        "min_row_ratio": 0.6,
        "max_null_rate": 0.02,
        "expected_rows": 10000,
    },
}

Implementation

The detector takes the current run’s metrics and the per-metric history, and returns a list of flags — one per metric that fell outside its robust band. It uses median/IQR for level metrics (row count, a value range) and an EWMA residual for anything you mark as trending. Below min_history it applies only the static floors so a cold start still catches a gross failure.

from __future__ import annotations

import logging
import statistics
from dataclasses import dataclass, field

logger = logging.getLogger("feed.anomaly")


@dataclass(frozen=True)
class Flag:
    """One anomaly finding for a single metric on a single feed run."""
    metric: str
    value: float
    low: float
    high: float
    method: str


@dataclass
class AnomalyDetector:
    """Robust median/IQR + EWMA detector for feed-run metrics."""
    iqr_k: float = 1.5
    ewma_alpha: float = 0.3
    ewma_z: float = 3.0
    min_history: int = 4
    min_row_ratio: float = 0.6
    max_null_rate: float = 0.02
    expected_rows: int = 10000
    trending_metrics: set[str] = field(default_factory=set)

    def _iqr_band(self, history: list[float]) -> tuple[float, float]:
        ordered = sorted(history)
        n = len(ordered)
        q1, q3 = ordered[n // 4], ordered[(3 * n) // 4]
        iqr = q3 - q1
        return q1 - self.iqr_k * iqr, q3 + self.iqr_k * iqr

    def _ewma_band(self, history: list[float]) -> tuple[float, float]:
        ewma = history[0]
        for x in history[1:]:
            ewma = self.ewma_alpha * x + (1 - self.ewma_alpha) * ewma
        resid = [h - ewma for h in history]
        sd = statistics.pstdev(resid) or 1e-9
        return ewma - self.ewma_z * sd, ewma + self.ewma_z * sd

    def check(self, metric: str, value: float, history: list[float]) -> Flag | None:
        """Flag one metric if it falls outside its robust band; else return None."""
        if len(history) < self.min_history:
            return self._static_check(metric, value)
        low, high = (self._ewma_band(history) if metric in self.trending_metrics
                     else self._iqr_band(history))
        method = "ewma" if metric in self.trending_metrics else "iqr"
        if not (low <= value <= high):
            logger.warning("feed anomaly on %s: %.3f outside [%.3f, %.3f] (%s)",
                           metric, value, low, high, method)
            return Flag(metric, value, low, high, method)
        return None

    def _static_check(self, metric: str, value: float) -> Flag | None:
        if metric == "row_count" and value < self.min_row_ratio * self.expected_rows:
            return Flag(metric, value, self.min_row_ratio * self.expected_rows,
                        float(self.expected_rows), "static")
        if metric.startswith("null_rate") and value > self.max_null_rate:
            return Flag(metric, value, 0.0, self.max_null_rate, "static")
        return None

    def scan(self, run: dict[str, float],
             history: dict[str, list[float]]) -> list[Flag]:
        """Check every metric in the run; return all flags raised."""
        flags = [f for m, v in run.items()
                 if (f := self.check(m, v, history.get(m, []))) is not None]
        logger.info("feed scan complete: %d/%d metrics flagged", len(flags), len(run))
        return flags

Step-by-Step Walkthrough

  1. Pick the band per metric. check routes each metric to _ewma_band if it is in trending_metrics (a slowly-growing catalog count, for example) or _iqr_band otherwise. Level metrics get the robust IQR fence; trending metrics get a band that moves with the series so drift is not flagged as an anomaly.
  2. Build the IQR fence. _iqr_band sorts the history, takes the first and third quartiles, and sets the band at Q1 - k*IQR to Q3 + k*IQR. Because it uses quartiles, one prior outlier cannot widen the band the way a single spike inflates a standard deviation — this is why iqr_k is the sensitivity dial, not a raw sigma count.
  3. Build the EWMA residual band. _ewma_band folds the history into an exponentially-weighted mean, measures the residual spread, and bands at ewma_z residual standard deviations. A value is flagged only if it deviates from the current smoothed level, so a metric that grows 2% per week never trips.
  4. Fall back on cold start. When len(history) < min_history, check defers to _static_check, which applies the min_row_ratio and max_null_rate floors from config. A brand-new feed with no baseline still catches a half-empty payload or a null-rate spike.
  5. Scan the whole run. scan walks every metric in the run dict, collects the flags, and logs the flagged/total ratio. An empty list is a clean run; any flag routes the run to quarantine before it reaches scoring.

Verification

Drive the detector with a synthetic stable history and confirm it passes a normal run, flags a row-count drop, and flags a null-rate spike. The EWMA path is checked against a trending series that must not flag.

import logging


def test_detector() -> None:
    logging.basicConfig(level=logging.INFO)
    det = AnomalyDetector(iqr_k=1.5, expected_rows=10000)
    history = {
        "row_count": [10000, 10120, 9980, 10050, 9900, 10200] * 6,
        "null_rate.sku_id": [0.0, 0.001, 0.0, 0.002, 0.0] * 8,
    }

    clean = det.scan({"row_count": 10040, "null_rate.sku_id": 0.001}, history)
    assert clean == [], clean

    drop = det.scan({"row_count": 4000, "null_rate.sku_id": 0.0}, history)
    assert any(f.metric == "row_count" for f in drop)

    spike = det.scan({"row_count": 10010, "null_rate.sku_id": 0.08}, history)
    assert any(f.metric == "null_rate.sku_id" for f in spike)

    # EWMA path: a steadily growing catalog count must not flag.
    det.trending_metrics = {"catalog_size"}
    trend = [50000 + 200 * i for i in range(40)]
    assert det.check("catalog_size", 58020.0, trend) is None
    print(f"row-count flag: {drop[0].value:.0f} outside [{drop[0].low:.0f}, {drop[0].high:.0f}]")


test_detector()

Sample expected output:

WARNING:feed.anomaly:feed anomaly on row_count: 4000.000 outside [9540.000, 10560.000] (iqr)
WARNING:feed.anomaly:feed anomaly on null_rate.sku_id: 0.080 outside [0.000, 0.005] (iqr)
INFO:feed.anomaly:feed scan complete: 1/2 metrics flagged
row-count flag: 4000 outside [9540, 10560]

Common Pitfalls

  • Seasonality mistaken for an anomaly. A peak-day feed legitimately doubles every historical band, so a naive detector quarantines your busiest and most valuable run. Mark seasonal windows on a calendar and either widen iqr_k for those dates or baseline against the same period last year — never let a fixed band judge Black Friday against October.
  • Cold-start false confidence. With fewer than min_history runs the robust tests are meaningless, and a detector that silently returns “no flags” lets a bad first run through. Keep the static floors (min_row_ratio, max_null_rate) active until the baseline warms up, and treat any cold-start pass as provisional.
  • Threshold tuning by guesswork. Picking iqr_k or ewma_z from intuition produces either an alert storm or a detector that never fires. Backtest against a few weeks of labeled runs, measure the false-positive rate, and tighten only once seasonality is handled — a detector nobody trusts gets muted, which is worse than none.
  • Poisoned baseline. If flagged runs are written back into the history, the band drifts toward the bad values and the next identical failure scores as normal. Update history only with runs that passed, and log quarantined runs separately for forensics so the baseline stays clean.