Your CAPTCHA Rate Is a Metric: Telemetry-Driven Challenge Avoidance in Python

Our CAPTCHA solver bill grew 40% month over month for three months before anyone asked why. When we finally plotted challenge encounters over time, the answer was sitting there the whole time: the encounter rate on two of our target hosts had doubled three weeks before the first invoice bump. The signal was free, emitted on every single request, and nobody was recording it. We were watching the cost line and ignoring the leading indicator.

I've written before about adopting an avoidance-first mindset — the idea that the cheapest CAPTCHA is the one you never receive, and that fingerprinting, pacing, and identity hygiene come before any solving. This post is about the discipline that makes that mindset operational: measurement. A CAPTCHA is not an event to outsource to a solver farm. It is a telemetry data point, and the rate at which you collect those points is one of the most informative metrics a scraping pipeline can track.

Detecting challenges is harder than it sounds

Before you can measure an encounter rate, you need to define "encounter" precisely. In practice, challenges arrive in at least three disguises:

• 403 or 503 with challenge headers. The cleanest case. Some edge providers set an explicit header (for example cf-mitigated: challenge) that you can check before even reading the body.
• 200 with an interstitial body. The nastiest case. Many JS challenges are served with 200 OK, a tiny HTML payload, and the actual content nowhere in sight. If your extractor silently returns empty fields, your telemetry never fires.
• 429 leading into challenges. A burst of 429s is rate limiting, not a challenge — but it often precedes one. Track it, but as a separate signal, or you'll conflate "they're throttling me" with "they're challenging me."

For the 200-interstitial case, content-signature detection works well: check the first few kilobytes of the body for markers that only appear on challenge pages (script paths like recaptcha/api.js, hcaptcha.com/1/api.js, challenge-platform, vendor cookie names). Combine that with a size heuristic — real product pages are rarely under 8 KB — and detection becomes reliable without false-positiving on ordinary pages that merely mention "captcha" in a FAQ.

One more trap: retries. If your client retries a challenged request three times and you count each attempt, your encounter rate is inflated 4x and your policy engine will panic. Deduplicate by request ID before recording.

Designing the metric

The core metric is:

encounter_rate = challenges / total_requests

computed over a sliding window, not per-hour buckets. Bucketed counters have two failure modes: a burst at 10:59 and 11:01 gets split into two innocent-looking buckets, and the current bucket is half-empty for its entire first half, so an early spike looks diluted. A sliding window — "the last 10 minutes, continuously" — has neither problem, and its value degrades gracefully as events age out.

Segmentation is where the metric earns its keep. A single global rate is nearly useless because it averages across everything. Track the rate:

• per host — targets have independent sensitivity;
• per egress IP — this is the identity the target actually sees;
• per ASN — soft-blocking is often applied at network level;
• per session age bucket — a challenge on a 40-minute-old sticky session means something different from one on a fresh IP.

One more pool-health metric worth its weight: time-to-first-challenge per fresh IP. If freshly rotated IPs historically survive 25 minutes before their first challenge and that drops to 4 minutes, your pool's reputation is degrading before your encounter rate moves at all. It's the leading indicator of your leading indicator.

The build

Here is a compact, runnable module covering the whole loop: detection, sliding-window counters, segmented telemetry, dedupe, a threshold-driven policy engine, and Prometheus text exposition so your existing dashboards and alerting pick it up for free.

"""challenge_telemetry.py -- challenge encounter rate as a first-class metric."""
import time
import threading
from collections import deque, defaultdict

CHALLENGE_MARKERS = (
    b"challenge-platform", b"cf-chl-bypass", b"cdn-cgi/challenge-platform",
    b"hcaptcha.com/1/api.js", b"google.com/recaptcha/api.js",
    b"px-captcha", b"datadome", b"geo.captcha-delivery.com",
)

def is_challenge(status, body=b"", headers=None):
    """True if the response is an anti-bot interstitial, not our content."""
    headers = {k.lower(): v for k, v in (headers or {}).items()}
    if headers.get("cf-mitigated") == "challenge":
        return True
    if status in (403, 503):
        return any(m in body[:4096] for m in CHALLENGE_MARKERS)
    if status == 200 and len(body) < 8192:
        # 200 + tiny payload + challenge markup = JS interstitial
        return any(m in body for m in CHALLENGE_MARKERS)
    return False  # a bare 429 is throttling, not a challenge: track separately

class SlidingWindowCounter:
    """Thread-safe sliding-window counter: O(1) add, amortized O(1) prune."""
    def __init__(self, window_s):
        self._window = window_s
        self._events = deque()
        self._lock = threading.Lock()

    def add(self, ts=None):
        ts = time.monotonic() if ts is None else ts
        with self._lock:
            self._events.append(ts)
            self._prune(ts)

    def _prune(self, now):
        cutoff = now - self._window
        while self._events and self._events[0] <= cutoff:
            self._events.popleft()

    def count(self, now=None):
        now = time.monotonic() if now is None else now
        with self._lock:
            self._prune(now)
            return len(self._events)

class ChallengeTelemetry:
    """Records (total, challenge) event pairs per segment over sliding windows."""
    def __init__(self, window_s=600):
        self._total = defaultdict(lambda: SlidingWindowCounter(window_s))
        self._chall = defaultdict(lambda: SlidingWindowCounter(window_s))
        self._reqids = deque(maxlen=8192)
        self._reqid_set = set()
        self._lock = threading.Lock()

    def record(self, status, host, ip, asn, body=b"", headers=None, req_id=None):
        if req_id is not None:
            with self._lock:
                if req_id in self._reqid_set:
                    return                # retry of an already-counted request
                if len(self._reqids) == self._reqids.maxlen:
                    self._reqid_set.discard(self._reqids[0])
                self._reqids.append(req_id)
                self._reqid_set.add(req_id)
        now = time.monotonic()
        hit = is_challenge(status, body, headers)
        for key in ("all", f"host:{host}", f"ip:{ip}", f"asn:{asn}"):
            self._total[key].add(now)
            if hit:
                self._chall[key].add(now)

    def rate(self, key):
        total = self._total[key].count()
        return self._chall[key].count() / total if total else 0.0

    def prometheus(self):
        lines = [f'requests_total{{segment="all"}} {self._total["all"].count()}',
                 f'challenges_total{{segment="all"}} {self._chall["all"].count()}',
                 f'challenge_rate{{segment="all"}} {self.rate("all"):.4f}']
        for key in self._total:
            if key != "all":
                kind, _, value = key.partition(":")
                lines.append(f'challenge_rate{{segment="{value}",by="{kind}"}} '
                             f'{self.rate(key):.4f}')
        return "\n".join(lines) + "\n"

class PolicyEngine:
    """Threshold-driven avoidance: cool identities BEFORE the hard block."""
    COOLDOWN_S = 1800          # 30 min
    MIN_SAMPLES = 20           # don't panic over 1/3 = 33% on three requests

    def __init__(self, telemetry):
        self.t = telemetry
        self._cooled = {}       # ip -> monotonic deadline

    def cooled(self, ip):
        return self._cooled.get(ip, 0.0) > time.monotonic()

    def evaluate(self, ip, consecutive_challenges=0):
        actions = []
        ip_key = f"ip:{ip}"
        if self.t._total[ip_key].count() >= self.MIN_SAMPLES:
            if self.t.rate(ip_key) > 0.30:              # this IP is burning out
                self._cooled[ip] = time.monotonic() + self.COOLDOWN_S
                actions.append(("cool_ip", ip, self.COOLDOWN_S))
            elif self.t.rate(ip_key) > 0.10:            # early warning
                actions.append(("slow_ip", ip, 30))      # add 30s pacing delay
        if self.t.rate("all") > 0.10:                    # pipeline-wide problem
            actions.append(("halve_global_rate", None, None))
        if consecutive_challenges >= 2:                  # session is cooked
            actions.append(("retire_session", ip, None))
        return actions or [("ok", ip, None)]

if __name__ == "__main__":
    tel = ChallengeTelemetry(window_s=600)
    for i in range(120):   # 107 clean + 13 challenged -> rates ~0.108
        status, body = (200, b"<html>ok</html>") if i < 107 else (403, b"challenge-platform")
        tel.record(status, "api.example.com", "203.0.113.7", "AS64496",
                   body=body, req_id=f"r{i}")
    print(PolicyEngine(tel).evaluate("203.0.113.7", consecutive_challenges=2))
    print(tel.prometheus(), end="")

The numbers in PolicyEngine are the ones we run in production, roughly: per-IP rate above 0.10 gets a pacing delay, above 0.30 over at least 20 samples gets a 30-minute cooldown, a host-wide rate above 0.10 halves global throughput, and two consecutive challenges on one IP retires the session. The MIN_SAMPLES guard matters — a rate metric on tiny denominators will make your policy engine thrash. Run the module directly and you'll see it fire slow_ip, halve_global_rate, and retire_session on the simulated data, plus the exposition text you can paste into any Prometheus scrape config.

Reading the numbers

Once the metric exists, learn what its shapes mean:

• Healthy: under 1% on aged residential sessions, flat over hours. That's your baseline; alert on sustained deviation, not absolutes.
• Rising but flat-ish per IP, rising per ASN: the target is soft-blocking your network, not your identities. Rotating harder makes it worse — you're feeding them fresh IPs from the same blocked range. The fix is egress mix, not more volume.
• A step function: you tripped a rule. Something changed — concurrency, a new code path, a retry storm — and the target's response was immediate. Diff your deploy log against the step; the correlation is usually a one-liner.
• Slow uniform drift upward: session entropy. Cookies aging, fingerprints accumulating history. Time to rotate identities on schedule rather than on failure.

And tie the metric back to identity quality, because the numbers are not close. On the same target, same code, same pacing, we've seen fresh datacenter IPs run a 10–15% encounter rate while aged residential sessions on the same host run around 1%. That's an order of magnitude, which means egress quality is not an optimization you bolt on later — it's a dimension of the metric itself. Segment by it from day one, or your global rate is an average of two different universes.

Knowing when to stop

The metric also tells you when to quit. Block economics: if sessions that are cooled, well-paced, clean-fingerprinted, and on good residential egress still hit a wall of challenges on a given host, the target has decided that content is not worth serving to automation at any price you can pay. No solver farm fixes that; solving just converts a refusal into an arms race and your margin into their revenue. Watch the encounter rate on your best-behaved identities. If it stays pinned high after everything else is clean, that host is telling you something. Respect the signal and either negotiate API access, reduce scope, or walk away.

Wrapping up

Treat every challenge response as a datapoint, compute encounter rates over sliding windows, segment them by host, IP, ASN, and session age, and let thresholds drive cooldowns and pacing automatically. The solver bill is the lagging indicator; the encounter rate is the free, real-time one. Measure it, graph it, alert on it — and most CAPTCHAs disappear from your pipeline without ever being solved.

Disclosure: I use Thordata's residential proxies for this project. New users get 500MB free — code thor020 (10% off): https://www.thordata.com/?ls=uXcSHJzx&lk=02-tele