Blog / Data engineering

Web Data Quality Checks for Scraped and API Data

11 Sept 2026Perceptlayer7 min read

Web data goes wrong quietly: a field disappears, a price turns into text, a page repeats. Seven checks catch most of it: the response envelope, schema drift, null rates, duplicate IDs, price outliers, geographic bounds and freshness. Each one is a short Python function you can run on every response. The examples below use the shape of a recorded Zillow search response, but the same checks work on any JSON endpoint.

Why does web data need its own quality checks?

Web data is collected from publicly available sources and returned as found at the time of the call, so the shape and completeness of a response can change without a version bump. Scrapers fail loudly when a selector breaks, but they also fail quietly, by returning a null where a number used to be. An API removes the parsing work, not the need to validate what comes back. For the maintenance side of that trade-off, see what maintaining a scraper really involves.

A useful rule: validate at the boundary where data enters your pipeline, and store the failures instead of dropping them.

What are the seven checks?

Seven checks cover most pipeline failures in web data: the response envelope, schema drift, null rates, duplicate IDs, price sanity, geographic bounds and freshness. Run them cheapest first on every response; each is a short Python function that returns a list of problems. Code for each follows in the next section.

1. Envelope: the call succeeded, cost what you expected, and the list is where you expect it.
2. Schema drift: required fields exist, and new, unexpected fields get logged.
3. Null rates: the share of missing values per field stays near its baseline.
4. Duplicate IDs: no zpid appears twice in a page or across pages.
5. Price sanity: prices agree with area and with the stated price per square foot, and extremes get flagged.
6. Geographic bounds: every listing sits inside the area you asked for.
7. Freshness: the response says when its source data was taken, and that stamp isn't older than your limit.

All seven read the response you already paid for, so the only cost is the fetch itself.

How do you fetch a sample to check?

Send one Zillow search request and keep the raw response, because every check below reads from it. Perceptlayer charges 1 credit for a successful call; a failed call costs nothing and returns X-Credits-Used: 0. Checking one query a day for 30 days is 30 calls × 1 credit = 30 credits. Pricing lists what credits cost.

import os, requests

URL = "https://api.perceptlayer.com/v1/zillow/search"
PARAMS = {"latitude": 47.6062, "longitude": -122.3321, "radius_mi": 5,
          "listing_category": "category1", "page_size": 20}

def fetch(page=1):
    return requests.get(URL, params={**PARAMS, "page": page},
                        headers={"X-API-Key": os.environ["API_KEY"]}, timeout=30)

How do you run each check in Python?

Each function takes the parsed listings and returns a list of problems, so you can log them, count them or fail a job on them. The listing fields (zpid, price.value, livingArea, location) come from the recorded response.

1. Envelope

def check_envelope(r):
    if not r.ok:
        return [f"http {r.status_code}"], []
    data = r.json().get("data") or {}
    listings = data.get("listings")
    if not isinstance(listings, list):
        return ["data.listings is not a list"], []
    problems = []
    if r.headers.get("X-Credits-Used") != "1":
        problems.append(f"credits used: {r.headers.get('X-Credits-Used')}")
    if data.get("count") != len(listings):
        problems.append("count differs from len(listings)")
    return problems, listings

A non-200 response carries a JSON error body with a code; the error reference lists them, and credits explains the headers.

2. How do you detect schema drift?

REQUIRED = {"zpid", "address", "location", "price", "livingArea",
            "bedrooms", "bathrooms", "propertyType", "listing"}

def check_schema(listings, known_keys):
    missing = [l.get("zpid") for l in listings if REQUIRED - set(l)]
    seen = set().union(*(set(l) for l in listings))
    return missing, sorted(seen - known_keys)

Set REQUIRED to the fields your code reads, not every field you've seen. For stricter type checks, JSON Schema describes the same rules declaratively. Save known_keys from your last good run, and alert on the second return value: a new key is how drift usually shows up first.

3. How do you track null rates?

def get(d, path):
    for key in path.split("."):
        d = d.get(key) if isinstance(d, dict) else None
    return d

def null_rates(listings, paths):
    return {p: sum(get(l, p) is None for l in listings) / len(listings)
            for p in paths}

Compare the result with a stored baseline, and alert on a jump rather than on any null. In the two listings excerpted from the recorded response, estimates.zestimate appears on the first and taxAssessment on the second, so both are optional fields. A rate that moves from 0.1 to 0.6 is a signal; a single null is not.

4. Duplicate IDs

from collections import Counter

def duplicate_ids(listings, seen_before=frozenset()):
    counts = Counter(l.get("zpid") for l in listings if l.get("zpid") is not None)
    return [z for z, c in counts.items() if c > 1 or z in seen_before]

Pass the IDs from earlier pages as seen_before. The recorded response notes that page * page_size must not exceed 1000. What happens past that limit isn't documented, so cap your loop below it; safe page loops covers it.

5. How do you detect price outliers?

Flag price per square foot with the modified z-score, 0.6745 × (x − median) / MAD, and review values above 3.5 (NIST e-Handbook). Skip it under five rows, and cross-check price.value against livingArea and pricePerSquareFoot.

import statistics

def price_problems(listings, limit=3.5):
    rows = [(l.get("zpid"), l["price"]["value"], l["livingArea"],
             l["price"].get("pricePerSquareFoot"))
            for l in listings
            if get(l, "price.value") and l.get("livingArea")]
    unit = [v / a for _, v, a, _ in rows]
    out = [(z, "stated $/sqft disagrees") for (z, v, a, s), u in zip(rows, unit)
           if s and abs(s - u) > 0.02 * u]
    if len(rows) >= 5:
        med = statistics.median(unit)
        mad = statistics.median(abs(u - med) for u in unit) or 1
        out += [(z, "price outlier") for (z, *_), u in zip(rows, unit)
                if abs(0.6745 * (u - med) / mad) > limit]
    return out

The NIST e-Handbook describes the modified z-score, attributes it to Iglewicz and Hoaglin and recommends a cutoff of 3.5: values with a larger absolute score are potential outliers. It needs enough rows to be stable, which is why the code skips it under five.

The first recorded listing has price.value 14000000 and livingArea 7870, which gives about 1,779 per square foot, matching its pricePerSquareFoot. A high price that agrees with its own area is flagged for review, not deleted. The Python statistics docs cover median, which stays stable when a few extreme values are present. For which fields matter in valuation work, see listing fields for valuation.

6. Geographic bounds

import math

def miles(lat1, lon1, lat2, lon2):
    p = math.pi / 180
    a = (math.sin((lat2 - lat1) * p / 2) ** 2 + math.cos(lat1 * p)
         * math.cos(lat2 * p) * math.sin((lon2 - lon1) * p / 2) ** 2)
    return 3958.8 * 2 * math.asin(math.sqrt(a))

def outside_radius(listings, lat, lon, radius_mi):
    out = []
    for l in listings:
        la, lo = get(l, "location.latitude"), get(l, "location.longitude")
        if la is not None and lo is not None \
                and miles(lat, lon, la, lo) > radius_mi * 1.05:
            out.append(l.get("zpid"))
    return out

The 5% margin absorbs rounding at the edge. Coordinates of 0, 0 or a different state are worth an alert.

7. Freshness

import re
from datetime import datetime, timezone
from zoneinfo import ZoneInfo

def age_hours(data):
    text = " ".join(data.get("freshness", []))
    m = re.search(r"as of (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", text)
    if not m:
        return None
    ts = datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S")
    ts = ts.replace(tzinfo=ZoneInfo("America/Los_Angeles"))
    return (datetime.now(timezone.utc) - ts).total_seconds() / 3600

The recorded freshness entry reads "Based on information submitted to the MLS GRID as of 2026-10-21 06:24:36 PDT…", so the code assumes Pacific time. Record the age and alert on a missing stamp, or one older than the limit you pass to the runner below (72 hours is a placeholder). The stamp tells you when the listing data was taken, not that the source is correct.

What do you do when a check fails?

Store the raw response with the failed check's name, keep it out of your clean tables, and look at it before you retry. A second call costs another credit, and a data-quality failure is often a property of the source rather than a glitch, so don't retry it by reflex. Retry transport problems and the retryable statuses: 429 and 503 carry a Retry-After header with the seconds to wait. Read retrying without paying twice for lost replies.

The sketch below wires the seven checks together. check_envelope returns a tuple while the others return lists, and age_hours needs the data object the envelope check discards, so the runner parses it again. The baseline file, the 0.3 null-rate jump and the 72-hour age limit are placeholders to tune.

import json, time

def run_checks(r, baseline, seen_ids=frozenset(), max_age_hours=72):
    env, listings = check_envelope(r)
    failed = {"envelope": env} if env else {}
    if listings:
        data = r.json()["data"]
        missing, new_keys = check_schema(listings, set(baseline["known_keys"]))
        rates = null_rates(listings, baseline["null_rates"])
        jumps = [p for p, v in rates.items()
                 if abs(v - baseline["null_rates"][p]) > 0.3]
        age = age_hours(data)
        found = {"schema": missing + new_keys, "null_rates": jumps,
                 "duplicates": duplicate_ids(listings, seen_ids),
                 "price": price_problems(listings),
                 "geo": outside_radius(listings, PARAMS["latitude"],
                                       PARAMS["longitude"], PARAMS["radius_mi"]),
                 "freshness": ["no timestamp"] if age is None
                              else [f"{age:.0f}h old"] if age > max_age_hours
                              else []}
        failed.update({k: v for k, v in found.items() if v})
    elif not env:
        failed["envelope"] = ["no listings"]
    if failed:
        with open(f"quarantine-{int(time.time())}.json", "w") as f:
            json.dump({"checks": failed, "raw": r.text}, f)
    return failed

baseline = json.load(open("baseline.json"))  # shape below
failed = run_checks(fetch(), baseline)

baseline.json holds what a known-good run looked like: the top-level listing keys you've seen, and the null rate you expect for each field you track. This shortened example lists only some keys. A real baseline lists every top-level key from your good run, or check_schema reports the rest as new, so write yours from that run.

{
  "known_keys": ["zpid", "location", "address", "price", "livingArea",
                 "bedrooms", "bathrooms", "propertyType", "listing",
                 "estimates", "taxAssessment"],
  "null_rates": {"estimates.zestimate": 0.5, "taxAssessment": 0.5,
                 "price.pricePerSquareFoot": 0.0}
}

One more trap sits in the response itself. The listing_category_counts_note field warns that a sale (category1) request reports the rent (category2) counts, so read the note before trusting those counts.

Next step: get a free API key, run the fetch above, and wire the seven checks into the job that already loads your data. More pipeline patterns live in the data engineering category.

data qualitymonitoringpythonzillowdata pipelines