Treat Proxy IPs Like Milk, Not Hardware: A Decay Model and Health Scorecard for Your Pool

Most scraping teams manage their proxy pool the way a bad datacenter manager manages servers: an IP goes in the pool, it works, and it's assumed to keep working until it visibly fails. When success rates drop, someone restarts things, swaps providers, and the cycle repeats.

This model is wrong because proxy IPs — especially residential ones — are perishable goods. An exit IP's value decays continuously as it accumulates blocks, CAPTCHAs, and 429s across every customer of your provider who happens to share it, plus the target's own per-IP aging heuristics. A residential IP that worked at 9am can be worthless by noon without anything changing on your side. Hardware fails suddenly; milk spoils gradually, and then all at once.

The core idea of this post: replace binary alive/dead thinking with a continuous health score per IP, maintained by synthetic probes, and drive your rotation and escalation off that score. Your pool has a half-life. Measure it.

Why naive pool management fails

The usual pattern: a worker grabs "some proxy" from a list, tries the request, and on failure marks the proxy bad. Three problems with this:

1. Confounded signals. A request can fail because the IP is bad, the target is having a bad minute, your headers are the problem, or the proxy gateway hiccuped. Attributing every failure to the IP poisons your pool data.
2. No shared state. Ten workers each discover independently that the same 200 IPs are dead — 10x the wasted requests and 10x the target-visible errors.
3. No recovery path. An IP marked dead is usually dead forever in that process, even though residential IPs recover — the block that killed them expires, the neighbor's abusive traffic moves on.

A health scorecard fixes all three: failures feed a decaying score, successes recover it, and the score is shared, persisted, and continuously refreshed by probes that separate "the IP is bad" from "the target is bad."

The scoring model

Use an EWMA (exponentially weighted moving average) with asymmetric update rates: failures drop the score fast, successes recover it slowly. That asymmetry matters — an IP that just got a hard block should not return to the hot pool because two cheap requests succeeded. Add a quarantine state for recently-failed IPs, and probe them on a slow cadence until they earn their way back.

Score semantics:

• 1.0: pristine, untested or fully recovered
• 0.6–1.0: healthy, in the hot rotation
• 0.3–0.6: degraded, use only for low-value requests or probe-only
• < 0.3: quarantined; only synthetic probes touch it

Here's the implementation:

import time
import random
import threading
from dataclasses import dataclass, field

@dataclass
class IPHealth:
    ip: str
    score: float = 1.0
    last_probe: float = 0.0
    quarantined_at: float | None = None
    failure_streak: int = 0

class PoolHealthScorecard:
    """
    Continuous health scoring for a proxy pool.
    Failures decay the score fast; successes recover it slowly.
    """

    def __init__(self, failure_alpha=0.45, success_alpha=0.06,
                 quarantine_below=0.30, rejoin_above=0.60):
        self.failure_alpha = failure_alpha   # fast drop
        self.success_alpha = success_alpha   # slow recovery
        self.quarantine_below = quarantine_below
        self.rejoin_above = rejoin_above
        self._ips: dict[str, IPHealth] = {}
        self._lock = threading.Lock()

    def ensure(self, ip: str) -> IPHealth:
        with self._lock:
            if ip not in self._ips:
                self._ips[ip] = IPHealth(ip=ip)
            return self._ips[ip]

    def report_failure(self, ip: str, hard: bool = True):
        """hard=True: 403/CAPTCHA/timeout. hard=False: 429 or 5xx (weaker signal)."""
        with self._lock:
            h = self.ensure(ip)
            penalty = self.failure_alpha * (2.0 if hard else 1.0)
            h.score = max(0.0, h.score - penalty * h.score)
            h.failure_streak += 1
            if h.score < self.quarantine_below:
                h.quarantined_at = time.time()

    def report_success(self, ip: str):
        with self._lock:
            h = self.ensure(ip)
            h.score += self.success_alpha * (1.0 - h.score)
            h.failure_streak = 0
            if h.quarantined_at and h.score > self.rejoin_above:
                h.quarantined_at = None   # earned its way back

    def pick(self, exclude: set[str] = None) -> str | None:
        """Weighted pick favoring healthy IPs; never returns quarantined IPs."""
        exclude = exclude or set()
        with self._lock:
            candidates = [h for h in self._ips.values()
                          if h.quarantined_at is None and h.ip not in exclude]
        if not candidates:
            return None
        weights = [h.score ** 3 for h in candidates]   # cube => strongly prefer healthy
        return random.choices(candidates, weights=weights, k=1)[0].ip

    def stats(self) -> dict:
        with self._lock:
            ips = list(self._ips.values())
        q = [h for h in ips if h.quarantined_at]
        return {
            "total": len(ips),
            "quarantined": len(q),
            "median_score": sorted(h.score for h in ips)[len(ips)//2] if ips else 0,
            "quarantine_rate": round(len(q) / len(ips), 3) if ips else 0,
        }

Note weights = h.score ** 3: cubing the score makes a 0.9-health IP ~27x more likely to be picked than a 0.3-health one. That single exponent turns a scorecard into a traffic allocator.

Synthetic probes: separating IP health from target health

The scorecard above is only as good as its inputs. The trick is a probe that does not depend on the target you're scraping — otherwise you can't tell "this IP is burned" from "the target is having an incident." Probe each IP against a neutral, reliable endpoint (a status endpoint, httpbin, or your own health-check URL) on a slow cadence:

import requests

def probe_ip(proxy_ip: str, username: str, password: str,
             probe_url: str = "https://httpbin.org/status/200") -> bool:
    """Neutral probe: does this IP's tunnel work at all, independent of any target?"""
    proxies = {"http": f"http://{username}:{password}@{proxy_ip}:9000",
               "https": f"http://{username}:{password}@{proxy_ip}:9000"}
    try:
        r = requests.get(probe_url, proxies=proxies, timeout=15)
        return r.status_code == 200
    except Exception:
        return False

def probe_quarantined(scorecard: PoolHealthScorecard, username, password,
                      min_quarantine_seconds=900):
    """Give quarantined IPs a slow, lonely path back. Never probe hot."""
    now = time.time()
    for h in list(scorecard._ips.values()):
        if h.quarantined_at and now - h.last_probe > min_quarantine_seconds:
            h.last_probe = now
            if probe_ip(h.ip, username, password):
                scorecard.report_success(h.ip)
            else:
                scorecard.report_failure(h.ip, hard=False)

Quarantined IPs get probed at most once per 15 minutes. That costs almost nothing in traffic, keeps you from hammering dead tunnels, and — because residential IPs genuinely recover — recovers capacity you'd otherwise have written off.

Wiring it into a crawler

The full loop, with the scorecard as the shared brain between workers:

def crawl(url: str, scorecard: PoolHealthScorecard, username, password):
    tried: set[str] = set()
    for attempt in range(5):
        ip = scorecard.pick(exclude=tried)
        if ip is None:
            raise RuntimeError("pool exhausted — all IPs quarantined")
        tried.add(ip)
        proxies = {"http": f"http://{username}:{password}@{ip}:9000",
                   "https": f"http://{username}:{password}@{ip}:9000"}
        try:
            r = requests.get(url, proxies=proxies, timeout=25,
                             headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64)"})
            if r.status_code == 200:
                scorecard.report_success(ip)
                return r
            elif r.status_code in (403, 429):
                scorecard.report_failure(ip, hard=(r.status_code == 403))
            else:
                scorecard.report_failure(ip, hard=False)
        except requests.RequestException:
            scorecard.report_failure(ip, hard=True)
    raise RuntimeError(f"all attempts failed for {url}")

Two production details worth flagging. First, persist the scorecard (a JSON dump every few minutes is enough) — process restarts that reset all scores to 1.0 will happily send traffic to known-bad IPs. Second, chart stats()["median_score"] and quarantine_rate over time. When the median score trends down across days, your target has tightened per-IP defenses and no amount of pool hygiene saves you — that's your early warning to diversify geographies or rethink the fingerprint layer. The scorecard doesn't just allocate traffic; it measures your pool's half-life, which is the most honest KPI a scraping operation has.

What this replaces

Teams running this pattern typically report three outcomes: fewer wasted requests (dead IPs stop receiving real work), faster incident triage (the median-score chart points at IP-layer vs target-layer problems within minutes), and — the quiet win — recovered capacity, because IPs come back from quarantine instead of being condemned by a single bad afternoon. Your provider's pool is a shared, decaying resource. Treating your slice of it like a monitored system, rather than a static inventory, is the difference between debugging blocks at 2am and reading a dashboard.

---

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