How to Detect When Your Scraper Is Blocked

Here is the check almost everyone ships first:

r = requests.get(url, proxies=proxies, timeout=30)
if r.status_code == 200:
    parse(r.text)          # this is where the bad data gets in

Status codes describe the transaction, not the payload. RFC 9110 section 15.3.1 defines 200 as "the request has succeeded", and a WAF that decides to serve you an interstitial instead of a product page has, technically, succeeded. Cloudflare's own documentation notes that managed challenges render a challenge page in the browser, and plenty of bot-management vendors return that page with a 200 so real browsers never see an error.

The asymmetry decides how you spend engineering time. A scraper that hard-fails tells you within one run cycle, costs a few hours of missing rows, and backfills cleanly. A scraper that returns wrong data tells you when a pricing analyst notices a competitor "dropped" 40% overnight, poisons every downstream aggregate, and forces a re-scrape of a window you can no longer reconstruct because the source pages have since changed. Verification is always cheaper than backfill.

So the design rule is simple: fail closed. If you cannot prove a response is real, do not write it to production.

Before building detectors, name the failures. These seven show up in production, all of them returning 200:

The last one is rare, and it is the reason canaries exist. If a target decides to serve deliberately wrong numbers rather than block you, no structural detector will save you. Only a value you already know is correct will.

Two of these overlap with problems covered elsewhere: loud status codes are dissected in proxy error codes explained, and what to do once you have confirmed a block belongs in retry and backoff strategies for web scraping. This guide stops at the detection boundary.

You cannot measure drift without a reference. Record a rolling window of successful responses per URL class, not per URL. A URL class is a template: product-detail, search-results, category-page. Pages inside a class have similar length and identical structure, which is exactly what makes drift measurable.

import statistics
from collections import deque

class Baseline:
    """Rolling reference distribution for one URL class."""

    def __init__(self, window=200):
        self.lengths = deque(maxlen=window)
        self.item_counts = deque(maxlen=window)
        self.skeletons = deque(maxlen=window)

    def record(self, html, item_count):
        # Only ever called for responses that passed the FULL contract.
        self.lengths.append(len(html))
        self.item_counts.append(item_count)
        self.skeletons.append(skeleton_hash(html))

    def ready(self):
        return len(self.lengths) >= 30

    @staticmethod
    def median_mad(series):
        med = statistics.median(series)
        mad = statistics.median([abs(x - med) for x in series]) or 1.0
        return med, mad

Two decisions in that class do most of the work.

Only record verified responses. If you feed every 200 into the baseline, then the moment a block starts, the baseline learns the block page. Within an hour your detector treats 4 KB interstitials as normal and goes quiet. The baseline must be trained exclusively on responses that passed every other detector.

Use median and MAD, not mean and standard deviation. Median absolute deviation is unmoved by a minority of outliers, while a mean shifts toward whatever anomaly sits in the window. When 15% of your responses turn into block pages, the mean length drops and the standard deviation explodes, so the z-score of the next block page looks unremarkable. The median barely moves.

What to store per class

Keep it small: content length, extracted item count, skeleton hash, HTTP status, duration_ms, exit IP or ASN, and the timestamp. That is enough for every detector below and cheap enough to hold 200 rows per class in Redis.

The naive version is if len(html) < 5000: alarm, and it breaks the first time a legitimate page is short. Score against the distribution instead. The modified z-score, described in the NIST/SEMATECH e-Handbook of Statistical Methods, scales MAD by 0.6745 so the result is comparable to a standard deviation, and flags at an absolute value above 3.5.

def modified_z(x, med, mad):
    return 0.6745 * (x - med) / mad

def length_drift(html, baseline):
    med, mad = baseline.median_mad(baseline.lengths)
    z = modified_z(len(html), med, mad)
    return {"len": len(html), "median": med, "z": round(z, 2), "flag": abs(z) > 3.5}

Length alone is a weak signal, so pair it with structure. A tag histogram compared by cosine similarity tells you whether the page you got is the page you asked for:

import re
from collections import Counter

TAG_RE = re.compile(rb"<([a-zA-Z][a-zA-Z0-9]*)")

def tag_histogram(html_bytes):
    return Counter(t.lower() for t in TAG_RE.findall(html_bytes))

def cosine(a, b):
    keys = set(a) | set(b)
    dot = sum(a.get(k, 0) * b.get(k, 0) for k in keys)
    na = sum(v * v for v in a.values()) ** 0.5
    nb = sum(v * v for v in b.values()) ** 0.5
    return dot / (na * nb) if na and nb else 0.0

Measure your own thresholds rather than copying mine. On a fixed template, two real pages usually sit above 0.97 against the baseline centroid. A challenge interstitial shares almost none of the template's tags, so it falls far below. Anything under about 0.90 deserves a flag, and anything under 0.60 is a different page entirely.

A presence check such as if dom.select_one(".price") misses the most common partial-render block, where the node exists in the shell but never got populated. Declare a contract per URL class and assert against it:

from dataclasses import dataclass
from typing import Optional
import re

@dataclass
class FieldSpec:
    selector: str
    required: bool = True
    min_count: int = 1
    pattern: Optional[str] = None

CONTRACT = {
    "title":   FieldSpec("h1.product-title"),
    "price":   FieldSpec("[data-testid='price']", pattern=r"^\$\d[\d,]*\.\d{2}$"),
    "stock":   FieldSpec(".availability", required=False),
    "reviews": FieldSpec("li.review", min_count=3),
}

def check_contract(dom, contract):
    failures = []
    for name, spec in contract.items():
        nodes = dom.select(spec.selector)
        if len(nodes) < spec.min_count:
            if spec.required:
                failures.append(f"{name}: expected >={spec.min_count} node(s), got {len(nodes)}")
            continue
        text = nodes[0].get_text(strip=True)
        if not text:
            if spec.required:
                failures.append(f"{name}: selector matched but node is empty")
        elif spec.pattern and not re.match(spec.pattern, text):
            failures.append(f"{name}: {text!r} does not match {spec.pattern}")
    return failures

The pattern field earns its keep on geo swaps and currency mismatches. A price node containing 1.299,00 EUR matches the selector, is not empty, and is completely wrong for a US price feed. The regex catches it in one line. The same idea applies to dates, SKUs, and locale-specific separators, and it pairs with the normalisation work in how to clean scraped data.

A contract failure does not tell you why it failed. It could be a block or a redesign. Both need a human, which is the point: fail closed on either, then disambiguate.

Every detector so far is a negative control. It says "this does not look like a real page." A canary is the positive control: a URL whose correct answer you already know, so a pass proves the whole path works end to end.

CANARIES = [
    # (url, selector, value verified by hand)
    ("https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html",
     "h1", "A Light in the Attic"),
]

def canary_ok(fetch):
    for url, selector, expected in CANARIES:
        dom = BeautifulSoup(fetch(url), "lxml")
        node = dom.select_one(selector)
        got = node.get_text(strip=True) if node else None
        if got != expected:
            return False, f"canary {url}: expected {expected!r}, got {got!r}"
    return True, "ok"

Four rules make a canary useful rather than decorative.

• Same path. Fetch it through the same proxy cohort, headless settings, and headers as production traffic. A canary that goes out over a clean connection tests nothing.
• Same host where possible. A stable page on the target itself (a help page, a fixed category listing) proves this target still trusts you. A canary on a third-party host only proves your egress works. Run both, and label them differently.
• Head and tail of every batch. Blocks frequently start mid-run once volume crosses a threshold. A canary at the start alone will pass and lie to you.
• A value that does not change. If the canary value drifts legitimately, you will mute the alert within a week. Pick a title, an ISBN, a static heading, never a price.

The canary also answers a different question from raw connectivity, which is what how to test if your proxy is working covers. A proxy can be perfectly reachable and still be poisoned for one specific target.

Marker regexes such as "Access Denied" or "Pardon our interruption" only catch block pages you have already met. Skeleton hashing catches the ones you have not.

Strip every text node and attribute value, keep the tag structure, and hash it. All block pages of a given kind collapse to one hash, because they are one template:

import hashlib, re

def skeleton_hash(html: str) -> str:
    s = re.sub(r'\s+([\w:-]+)="[^"]*"', r" \1", html)   # keep attribute names, drop values
    s = re.sub(r">[^<]+<", "><", s)                      # drop text nodes
    s = re.sub(r"\s+", "", s)
    return hashlib.blake2b(s.encode("utf-8", "ignore"), digest_size=8).hexdigest()

def skeleton_report(hashes, known_good: set):
    counts = Counter(hashes)
    top, n = counts.most_common(1)[0]
    return {"top_hash": top, "share": n / len(hashes), "novel": top not in known_good}

The operational rule: any skeleton hash that is novel and takes more than 20% of a run is a block page you have not written a rule for yet. Save one sample body, read it once, add its marker to the fast path, and add the hash to known_good if it turns out to be a legitimate template variant. This single detector finds more block types in production than every hand-written regex combined, because it never requires you to predict anything.

If the target randomises class names or injects nonces, exact hashes fragment. Switch to a similarity hash then. Charikar's simhash construction (STOC 2002) clusters near-identical documents under a Hamming-distance threshold instead of demanding byte equality.

A cheaper cousin is compression ratio. Block pages are short and repetitive, so they compress harder than real content:

import zlib

def compression_ratio(body: bytes) -> float:
    return len(zlib.compress(body, 6)) / max(len(body), 1)

Treat it as a tiebreaker, never a primary trigger. A minified single-page-app shell compresses like a block page and will generate false positives on its own.

This is the detector nobody instruments, and it is where the slow poisoning happens. Suppose 8% of your pool has been flagged. Aggregate success sits at 92%, the dashboard is green, and every request through those exits returns a soft block at 200. You are quietly losing a biased slice of data, because burned IPs correlate with geography and with the targets that push back hardest. The rows you lose are not random, which is far worse for analysis than losing rows uniformly.

Attribute every outcome to its exit IP, then roll up. The trap is small samples: two failures out of two is not evidence. Use the Wilson score lower bound on the success proportion so the detector waits for enough data, the same interval statsmodels exposes as proportion_confint(method="wilson").

def wilson_lower(successes, trials, z=1.96):
    if trials == 0:
        return 0.0
    p = successes / trials
    denom = 1 + z * z / trials
    centre = p + z * z / (2 * trials)
    margin = z * ((p * (1 - p) / trials + z * z / (4 * trials * trials)) ** 0.5)
    return (centre - margin) / denom

def quarantine(stats, floor=0.75, min_trials=20):
    # stats: {exit_ip: (successes, trials)}
    return [ip for ip, (ok, n) in stats.items()
            if n >= min_trials and wilson_lower(ok, n) < floor]

18 successes out of 20 gives a Wilson lower bound near 0.70, while 180 out of 200 gives about 0.85. Same point estimate, very different confidence, and the bound stops you burning a good IP over a two-request fluke.

Roll the same counters up one level and the shape of the ban becomes readable:

Subnet-wide failure calls for different remediation than a single flagged IP, a distinction drawn out in what is IP blacklisting and how to avoid it.

Every contract failure triggers the same argument at 2 a.m.: did they redesign, or did they block us? Guessing wrong is expensive in both directions. Rotating your whole pool because a designer renamed a CSS class wastes a day. Patching selectors while you are actually banned bakes the block into your code.

The decisive test costs one extra request. Fetch the same URL twice in parallel over two very different egress profiles, then compare skeletons:

import requests

API = "https://scrape.sparkproxy.io/api/v1"
HEAD = {"X-API-Key": "YOUR_API_KEY"}

def differential(url):
    """Same URL, two very different egress profiles."""
    a = requests.get(API, headers=HEAD, params={
        "url": url, "render_js": "true",
    }).text
    b = requests.get(API, headers=HEAD, params={
        "url": url, "render_js": "true",
        "premium_proxy": "true", "stealth": "true", "country_code": "US",
    }).text
    same = skeleton_hash(a) == skeleton_hash(b)
    return {"verdict": "site_changed" if same else "blocked",
            "skeleton_default": skeleton_hash(a),
            "skeleton_premium": skeleton_hash(b)}

If both paths return the same structure, the target is showing everyone the same thing and your selectors are stale. If the clean residential path returns the real page and your normal path does not, you are blocked. Run this automatically on the first contract failure and put the verdict inside the alert body, so whoever gets paged starts with an answer instead of a question.

Supporting evidence, in the order worth checking:

The "ramps with volume" row is worth internalising. Rate-triggered blocks correlate with your own request curve, so plot failure rate against requests per minute on one chart. If the two lines track each other, you are pacing too hard, and the fix lives in how to avoid getting your proxy blocked rather than in your parser.

Detectors are worthless once the team mutes them. Three rules keep the signal clean.

Alert on rates over a window, never on a single request. One weird response is normal. A 30% contract-failure rate over 15 minutes is not.

Require two independent detectors before paging. Length drift alone is noise. Length drift plus a missing required selector is a block. Correlated detectors do not count as two, since length and compression ratio measure nearly the same thing.

Hold writes before you page. The alert is for humans. The write-hold is what protects the data.

def severity(run):
    if not run["canary_ok"] or run["contract_fail_rate"] > 0.40:
        return "P1"     # halt production writes, quarantine cohort, page on-call
    if run["contract_fail_rate"] > 0.10 or run["quarantined_ip_share"] > 0.15:
        return "P2"     # ticket within the hour, divert writes to staging
    if run["novel_skeleton_share"] > 0.05 or run["length_flag_rate"] > 0.05:
        return "P3"     # log, review at standup
    return "OK"

if severity(run) != "OK":
    write_target = STAGING_TABLE          # production history stays clean
    freeze(run["target"], minutes=30)
    run["differential"] = differential(run["sample_url"])

Routing suspect rows to a staging table instead of dropping them is the part people skip. You keep the evidence, production history stays clean, and once you confirm the run was fine after all you promote the batch with one query instead of re-scraping it.

For deciding what to build first, this is the cost-to-coverage picture:

Build the extraction contract first. It buys the most coverage per hour of work, and everything else refines it.

The SparkProxy Scraping API hands you several of these signals without extra parsing.

json_response=true returns an envelope with status_code, duration_ms, and a meta object carrying title and wordCount. That wordCount is a free length baseline, and duration is its own detector: challenge pages often resolve much faster than real ones, so a sudden drop in p50 duration is a block signature rather than a performance win.

A 530 response is an explicit block signal, not a generic timeout. The body carries reason, and when reason is captcha_blocked you also get captcha_type (cloudflare_turnstile, hcaptcha, recaptcha, or null for behavioural systems like DataDome and PerimeterX). Count those separately from network errors, because they call for different fixes.

extract_rules enforces the contract server-side. Any key that comes back null is a contract failure, and you never wrote a parser:

import requests

API = "https://scrape.sparkproxy.io/api/v1"
HEAD = {"X-API-Key": "YOUR_API_KEY", "Content-Type": "application/json"}

RULES = {
    "title": "h1.product-title",
    "price": "[data-testid='price']",
    "reviews": {"selector": "li.review", "type": "list"},
}

def guarded_scrape(url):
    r = requests.post(API, headers=HEAD, json={
        "url": url,
        "render_js": True,
        "wait_for": "[data-testid='price']",   # 30s ceiling, then it gives up
        "extract_rules": RULES,
        "tag": "price-monitor/target-a",       # echoed back, group failures by target
    })

    if r.status_code == 530:
        err = r.json()
        return {"ok": False, "signal": err.get("reason"),
                "captcha": err.get("captcha_type"), "job_id": err.get("job_id")}

    data = r.json()
    extracted = data.get("extracted", {})
    missing = [k for k, v in extracted.items() if v in (None, "", [])]
    if missing:
        return {"ok": False, "signal": "contract_failure", "missing": missing}

    return {"ok": True, "data": extracted, "duration_ms": data["duration_ms"]}

Two more parameters matter for detection. wait_for turns a partial render into an explicit failure instead of an empty node, since the browser holds until the selector appears and gives up after 30 seconds. And transparent_status_code=true (plain HTTP mode only, render_js=false) mirrors the target's real status into the API response, so a target 404 reaches your classifier as a 404 rather than a successful fetch of an error page.

Finally, the tag parameter is the cheapest observability you will ever add. Tag by target and by run, and per-target failure rates fall out of your own logs with no extra plumbing.