Data-Quality Monitoring for Inventory Feeds

A feed that passes schema validation can still be catastrophically wrong. The JSON parses, every field has the right type, and the payload is half the size it should be because an upstream ETL job filtered a warehouse by accident. Schema checks answer “is this well-formed?”; data-quality monitoring answers the harder question “is this right?” — is it fresh, complete, in-distribution, and internally consistent with what you saw an hour ago. This guide is part of the Velocity Data Ingestion & WMS Sync Pipelines architecture, and it is the guard that stands between ingestion and every downstream slotting decision. It complements Schema Validation for Inventory Feeds: schema validation gates structure, this gates distribution, freshness, and completeness. A feed can be perfectly structured and still poison your ABC tiers, and only a quality gate catches that.

What Data-Quality Monitoring Is

Data-quality monitoring is the practice of computing measurable quality metrics on every feed run, comparing them against thresholds and recent history, and making a pass / quarantine / alert decision before the feed reaches the scoring engine. For velocity feeds specifically, four dimensions matter and each fails in its own characteristic way:

• Freshness. How stale is the newest record? A feed whose max event_ts is six hours old will tier SKUs on yesterday’s demand. Freshness is measured as the lag between the latest record timestamp and now, checked against an SLA.
• Completeness. How much of the expected data arrived? Measured as row count versus a recent baseline and as the null-rate on required fields. A 40% row-count drop or a null-rate spike on sku_id is a truncated or partially-failed extract.
• Validity. Are the values inside plausible ranges? Negative qty_picked, a weight_kg of 4000, or a pick_count ten times any historical value are individually valid types but invalid data.
• Consistency. Does this run agree with itself and with recent runs? A sudden shift in the SKU-to-zone distribution, or a total unit count that jumped 5× overnight, signals a join gone wrong or a unit-of-measure change upstream.

The output of the monitor is a quality scorecard per feed run — one score per dimension plus an overall verdict — and a policy decision: fail-open (accept the feed, alert, and let scoring proceed) or quarantine (reject the feed, hold the last-good aggregate, and page an operator). Which policy applies depends on the dimension and the blast radius, covered below.

Input Data Requirements

The monitor consumes a normalized feed run plus a short history of prior runs’ metrics. It does not need the raw payload beyond the fields it scores, but it does need enough recent history — 20 to 60 prior runs is the working range — to establish a robust baseline. A cold monitor with no history can only apply static thresholds; a warm monitor compares against its own past.

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime


@dataclass(frozen=True)
class FeedRun:
    """Summary metrics for one feed ingestion, the atomic input to scoring."""
    feed_id: str
    run_ts: datetime
    row_count: int
    max_record_ts: datetime
    null_rate: dict[str, float] = field(default_factory=dict)
    numeric_summary: dict[str, float] = field(default_factory=dict)

The most common preconditions violation is comparing across a schema change: if the upstream feed adds or renames a field, the null-rate and numeric baselines silently reset. Version the metric history by schema hash so a legitimate schema change starts a fresh baseline rather than firing a false anomaly on every field.

Step-by-Step Implementation

The monitor runs four scoring passes and one policy pass. Each pass is a pure function of the current FeedRun and the metric history, so the whole scorecard is deterministic and testable.

1. Score Freshness

Freshness is the lag between the newest record and the run time, compared against the SLA. It is a static threshold — you do not need history to know a six-hour-old feed is stale — so this gate is the cheapest and usually the first to fire.

import logging
from datetime import datetime, timedelta

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


def score_freshness(run: FeedRun, sla: timedelta) -> float:
    """Return a 0-1 freshness score; 1.0 is fresh, 0.0 is at or past the SLA."""
    lag = run.run_ts - run.max_record_ts
    score = max(0.0, 1.0 - lag / sla)
    if score < 1.0:
        logger.warning("feed %s freshness lag %.0fs (score %.2f)",
                       run.feed_id, lag.total_seconds(), score)
    return score

2. Score Completeness

Completeness combines a row-count check against the recent median and a null-rate check on required fields. A row count far below the historical median is a truncated extract; a null-rate spike on a key field is a partial failure. Using the median rather than the mean of recent counts keeps the baseline robust to a single prior outlier.

import statistics


def score_completeness(run: FeedRun, recent_counts: list[int],
                       max_null_rate: float = 0.02) -> float:
    """Score row-count adequacy against recent median and required-field null rate."""
    if not recent_counts:
        return 1.0
    baseline = statistics.median(recent_counts)
    count_ratio = min(1.0, run.row_count / baseline) if baseline else 1.0
    worst_null = max(run.null_rate.values(), default=0.0)
    null_score = max(0.0, 1.0 - worst_null / max_null_rate) if max_null_rate else 1.0
    score = min(count_ratio, null_score)
    logger.info("feed %s completeness: rows %d vs median %.0f, worst-null %.3f, score %.2f",
                run.feed_id, run.row_count, baseline, worst_null, score)
    return score

3. Score Validity and Consistency With a Robust Baseline

Validity and consistency both ask “is this number where recent history says it should be?” — so they share one robust anomaly test. The detector uses the median and IQR (interquartile range) rather than mean and standard deviation, because a single prior bad run would inflate the standard deviation and hide the next anomaly. A value more than k IQRs outside the middle 50% is flagged.

def robust_range(history: list[float], k: float = 1.5) -> tuple[float, float]:
    """Return the (low, high) acceptable band from the median and IQR."""
    ordered = sorted(history)
    n = len(ordered)
    q1 = ordered[n // 4]
    q3 = ordered[(3 * n) // 4]
    iqr = q3 - q1
    return q1 - k * iqr, q3 + k * iqr


def score_consistency(value: float, history: list[float], k: float = 1.5) -> float:
    """Score a numeric metric against its robust band; 1.0 in-band, 0.0 far outside."""
    if len(history) < 4:
        return 1.0
    low, high = robust_range(history, k)
    if low <= value <= high:
        return 1.0
    span = max(high - low, 1e-9)
    excess = (low - value if value < low else value - high) / span
    score = max(0.0, 1.0 - excess)
    logger.warning("metric %.2f outside robust band [%.2f, %.2f] (score %.2f)",
                   value, low, high, score)
    return score

The IQR test is the simplest of three interchangeable detectors — z-score for roughly-normal metrics, IQR for skewed ones, and EWMA for trending series. The dedicated method that flags a whole feed run with these detectors is built in How to Detect Anomalies in Inventory Feeds.

4. Assemble the Scorecard and Apply Policy

The final pass combines the four dimension scores into a verdict and applies the quarantine-versus-fail-open policy. Freshness and completeness failures are hard — they quarantine, because scoring on a stale or truncated feed corrupts tiers. Validity and consistency failures are soft by default — they fail open with an alert, because a plausible-but-shifted distribution is often a real demand change, not a data error.

from dataclasses import dataclass


@dataclass
class Scorecard:
    feed_id: str
    freshness: float
    completeness: float
    validity: float
    consistency: float

    @property
    def verdict(self) -> str:
        if min(self.freshness, self.completeness) < 0.5:
            return "QUARANTINE"    # hard dimensions: hold last-good aggregate
        if min(self.validity, self.consistency) < 0.5:
            return "ALERT"         # soft dimensions: fail open, page an operator
        return "PASS"


def evaluate_feed(run: FeedRun, history: dict[str, list[float]],
                  sla: timedelta) -> Scorecard:
    """Run all four gates and return a scored, policy-ready scorecard."""
    card = Scorecard(
        feed_id=run.feed_id,
        freshness=score_freshness(run, sla),
        completeness=score_completeness(run, [int(c) for c in history.get("row_count", [])]),
        validity=score_consistency(run.numeric_summary.get("max_pick_count", 0.0),
                                   history.get("max_pick_count", [])),
        consistency=score_consistency(float(run.row_count), history.get("row_count", [])),
    )
    logger.info("feed %s verdict: %s", run.feed_id, card.verdict)
    return card

Tuning & Calibration

The two levers that move outcomes most are the freshness SLA and the IQR multiplier k. Set the SLA to just above your feed’s normal cadence plus its worst-case processing time — too tight and every slow run quarantines, too loose and stale data slips through. The k multiplier trades false positives against misses: k = 1.5 is the classic Tukey fence and a good default, k = 3.0 only catches gross outliers. Recompute the baselines on a rolling window so a permanent step-change in volume eventually becomes the new normal instead of alerting forever.

# feed_quality.yaml — one profile per feed
quality:
  freshness_sla_s: 5400          # 90 min: cadence + worst-case processing headroom
  history_window: 40             # prior runs kept for robust baselines
  completeness:
    max_null_rate: 0.02          # required-field null fraction that scores zero
    min_row_ratio: 0.6           # row count below 60% of median is a hard fail
  anomaly:
    iqr_k: 1.5                   # Tukey fence multiplier for validity/consistency
    quarantine_threshold: 0.5    # dimension score below this triggers policy action
  policy:
    hard_dimensions: ["freshness", "completeness"]   # quarantine on failure
    soft_dimensions: ["validity", "consistency"]     # fail-open + alert on failure

# Equivalent Python config dict consumed by the monitor
FEED_QUALITY = {
    "freshness_sla_s": 5400,
    "history_window": 40,
    "completeness": {"max_null_rate": 0.02, "min_row_ratio": 0.6},
    "anomaly": {"iqr_k": 1.5, "quarantine_threshold": 0.5},
    "policy": {
        "hard_dimensions": ["freshness", "completeness"],
        "soft_dimensions": ["validity", "consistency"],
    },
}

Parameter sensitivity is not uniform. freshness_sla_s and min_row_ratio are safety floors — set them from the feed’s operational envelope and rarely touch them. iqr_k is the tuning dial you actually turn: start conservative, watch the false-positive rate over a few weeks, and tighten only once seasonality is accounted for. history_window trades responsiveness against stability — shorter reacts to regime change faster but is noisier.

Validation & Testing

Never wire a monitor into the ingestion path without asserting its verdicts. Three properties must hold: a clean run passes, a stale run quarantines, and a truncated run quarantines regardless of how valid its values are. These pytest checks drive evaluate_feed with synthetic history.

from datetime import datetime, timedelta


def _run(rows: int, age_min: int) -> FeedRun:
    now = datetime(2025, 5, 28, 12, 0, 0)
    return FeedRun(
        feed_id="wms-main",
        run_ts=now,
        row_count=rows,
        max_record_ts=now - timedelta(minutes=age_min),
        null_rate={"sku_id": 0.0},
        numeric_summary={"max_pick_count": 120.0},
    )


HISTORY = {"row_count": [10000] * 40, "max_pick_count": [120.0] * 40}
SLA = timedelta(minutes=90)


def test_clean_run_passes() -> None:
    assert evaluate_feed(_run(rows=10000, age_min=5), HISTORY, SLA).verdict == "PASS"


def test_stale_run_quarantines() -> None:
    assert evaluate_feed(_run(rows=10000, age_min=240), HISTORY, SLA).verdict == "QUARANTINE"


def test_truncated_run_quarantines() -> None:
    # Half the expected rows is a hard completeness failure regardless of value validity.
    assert evaluate_feed(_run(rows=4000, age_min=5), HISTORY, SLA).verdict == "QUARANTINE"

A healthy run logs verdict: PASS and holds the four dimension scores at or near 1.0. When a gate fires, the scorecard tells you which dimension and by how much, so the operator page carries a cause, not just an alarm.

Integration Points

The monitor sits between ingestion and every scoring consumer, and its verdict is a contract each downstream layer relies on.

• ABC classification. A truncated or stale feed silently re-weights the Pareto curve, demoting fast-movers and promoting noise. The monitor must quarantine before the run reaches ABC Classification Tuning; a bad feed that slips through corrupts the tier map and, through it, every placement decision that consumes it.
• Re-slotting. A quarantined run must block re-slotting, not just warn — acting on a bad feed generates relocation labor that has to be reversed once the good feed lands. The break-even logic in Threshold Optimization for Re-slotting assumes its velocity input is trustworthy, so the quality gate is a precondition of firing any move.
• Schema validation. The monitor runs after Schema Validation for Inventory Feeds: structure first, distribution second. A payload that fails schema never reaches the quality gate, and a payload that passes schema still has to clear it.

Upstream, the same quality scorecard applies to the low-latency path: the streaming counters in Real-Time Streaming Velocity Updates should sample their event stream through the same freshness and distribution checks, so a bad producer batch is quarantined before it mutates a live counter.

Failure Modes & Edge Cases

• Seasonality read as an anomaly. Black Friday volume legitimately breaks every historical band, so a naive detector quarantines the busiest feed of the year. Remediation: baseline against the same period last year for known seasonal windows, or widen iqr_k on a scheduled calendar of peak days.
• Cold-start with no history. A new feed has no baseline, so consistency and validity default to passing and a genuinely bad first run slips through. Remediation: apply static floors (freshness SLA, min_row_ratio against an expected count) until enough runs accumulate for a robust baseline.
• Baseline poisoned by prior bad runs. If quarantined runs are still written into the metric history, the band drifts toward the bad values and stops catching them. Remediation: exclude quarantined runs from the baseline; only PASS/ALERT runs update history.
• Silent schema drift resetting baselines. An upstream field rename resets null-rate and numeric baselines and masks a real regression. Remediation: version metric history by schema hash so a change starts a fresh, clearly-labeled baseline instead of contaminating the old one.
• Fail-open on a hard dimension. Treating a freshness or completeness failure as a soft alert lets stale or truncated data into scoring. Remediation: keep freshness and completeness on the quarantine path and reserve fail-open for validity and consistency, where a shift is often real demand.

FAQ

How is data-quality monitoring different from schema validation?

Schema validation checks structure: field names, types, required-field presence, and enum membership. Data-quality monitoring checks content: is the feed fresh, complete, in-distribution, and internally consistent. A payload can pass every schema rule and still be half its expected size or six hours stale. Run schema validation first to reject malformed payloads cheaply, then run the quality monitor on the well-formed survivors to catch the errors structure cannot see.

Should a bad feed fail open or quarantine?

It depends on the dimension and the blast radius. Freshness and completeness failures should quarantine and hold the last-good aggregate, because scoring on stale or truncated data corrupts tiers in ways that are expensive to unwind. Validity and consistency failures usually fail open with an alert, because a plausible-but-shifted distribution is frequently a real demand change rather than a data error. The policy table encodes which dimension takes which path, and the safe default is to quarantine anything that would poison ABC tiers.

Which anomaly detector should I use for feed metrics?

Match the detector to the metric’s shape. Use a z-score for roughly-normal metrics like a stable daily row count. Use IQR (Tukey fences) for skewed metrics like pick-count distributions, because it is robust to a heavy upper tail. Use EWMA for metrics with a trend or drift, where a fixed band would alert constantly as the level moves. The IQR fence at k = 1.5 is the best single default when you are unsure, because it does not assume normality and is not thrown off by one prior outlier.

How much history do I need before the monitor is trustworthy?

Enough for a robust baseline — 20 to 60 prior runs is the working range, with 40 a reasonable default. Below four runs the IQR and z-score tests are unreliable and the monitor should fall back to static floors: a freshness SLA and an expected-row-count ratio. Never let a cold-start monitor default consistency to passing without those floors, or a genuinely bad first run walks straight through into scoring.

Why exclude quarantined runs from the baseline?

Because a baseline that absorbs its own failures stops detecting them. If a truncated run is written into the row-count history, the median drifts down toward the bad value, and the next equally-truncated run scores as normal. Only runs that passed or merely alerted should update the metric history; quarantined runs are logged for forensics but never feed the band that judges the next run.

Related

• How to Detect Anomalies in Inventory Feeds — the focused detector that flags a feed run with median/IQR or EWMA.
• Schema Validation for Inventory Feeds — the structural gate that runs before the quality monitor.
• ABC Classification Tuning — the tier map a bad feed would poison if the monitor let it through.
• Threshold Optimization for Re-slotting — the move-cost logic that a quarantined run must block.
• Velocity Data Ingestion & WMS Sync Pipelines — the parent architecture this quality gate guards.