Special offer. See more information about Outbyte and uninstall instructions. Please review EULA and Privacy policy.

Reliable proxy rotation is an application reliability pattern, not just a list shuffle: select an eligible proxy, send an authorized request with finite timeouts, classify the result, update that proxy’s health record, and retry only within explicit safety limits. The example below implements that pattern with Python and Requests, including latency-aware scoring, cooldowns, failure classification, and bounded retries.

What health-scored rotation should do

A useful rotator keeps proxy selection separate from HTTP connection pooling. Each proxy record needs a stable identifier, endpoint or credential reference, recent measurements, consecutive-failure count, last success time, cooldown deadline, and eligibility state.

On every operation, the application should:

  1. Filter out proxies in cooldown or otherwise ineligible.
  2. Choose one using a defined policy such as round-robin, weighted random, or least recently used.
  3. Send the request with connect and read timeouts.
  4. Record success, latency, status category, and the failure class when applicable.
  5. Update the score and either keep the proxy eligible or apply a temporary cooldown.
  6. Retry only a bounded number of times and only when repeating the operation is safe.

Use this only for destinations and workloads you are authorized to access. Rotation does not override a site’s authentication, access policy, rate limits, or terms.

A complete Requests implementation

Requests accepts a per-request proxies mapping. The following implementation uses a 0–100 application-defined score. It blends recent successful outcomes, latency, recency, and a consecutive-failure penalty. These weights and thresholds are examples to calibrate with your own workload, not a universal standard.

Special offer. See more information about Outbyte and uninstall instructions. Please review EULA and Privacy policy.
from __future__ import annotations

from dataclasses import dataclass, field
from collections import deque
import random
import time
from typing import Optional

import requests


@dataclass
class Proxy:
    name: str
    url: str                 # Keep credentials out of logs.
    history: deque = field(default_factory=lambda: deque(maxlen=20))
    latencies: deque = field(default_factory=lambda: deque(maxlen=20))
    consecutive_failures: int = 0
    last_success: Optional[float] = None
    cooldown_until: float = 0.0

    def eligible(self, now: float) -> bool:
        return now >= self.cooldown_until

    def score(self, now: float) -> float:
        if not self.history:
            return 50.0  # Unknown proxies start neutral, not trusted.
        success_ratio = sum(self.history) / len(self.history)
        avg_latency = sum(self.latencies) / len(self.latencies) if self.latencies else 10.0
        # 0 ms is ideal; 2 seconds or more receives zero latency points.
        latency_score = max(0.0, 1.0 - min(avg_latency / 2.0, 1.0))
        age = (now - self.last_success) if self.last_success else 3600.0
        recency = max(0.0, 1.0 - min(age / 3600.0, 1.0))
        streak_penalty = min(self.consecutive_failures * 0.05, 0.35)
        raw = (0.55 * success_ratio +
               0.25 * latency_score +
               0.20 * recency)
        return round(max(0.0, min(100.0, (raw - streak_penalty) * 100)), 2)


class ProxyRotator:
    def __init__(self, proxies: list[Proxy], cooldown_base: float = 15.0):
        if not proxies:
            raise ValueError("At least one proxy is required")
        self.proxies = proxies
        self.cooldown_base = cooldown_base

    def choose(self) -> Proxy:
        now = time.monotonic()
        eligible = [p for p in self.proxies if p.eligible(now)]
        if not eligible:
            # Wait only until the soonest cooldown, rather than spinning.
            wait = min(p.cooldown_until for p in self.proxies) - now
            time.sleep(max(0.0, wait))
            now = time.monotonic()
            eligible = [p for p in self.proxies if p.eligible(now)]
        # Square scores to favor healthy proxies while retaining exploration.
        weights = [max(p.score(now), 1.0) ** 2 for p in eligible]
        return random.choices(eligible, weights=weights, k=1)[0]

    def record(self, proxy: Proxy, ok: bool, latency: float) -> None:
        now = time.monotonic()
        proxy.history.append(1 if ok else 0)
        proxy.latencies.append(latency)
        if ok:
            proxy.consecutive_failures = 0
            proxy.last_success = now
            proxy.cooldown_until = 0.0
        else:
            proxy.consecutive_failures += 1
            # Stepped exponential cooldown, capped at five minutes.
            delay = min(self.cooldown_base * (2 ** (proxy.consecutive_failures - 1)), 300.0)
            proxy.cooldown_until = now + delay

    def get(self, url: str, *, attempts: int = 3,
            timeout: tuple[float, float] = (5.0, 20.0), **kwargs):
        last_error = None
        for _ in range(attempts):
            proxy = self.choose()
            started = time.monotonic()
            try:
                response = requests.get(
                    url,
                    proxies={"http": proxy.url, "https": proxy.url},
                    timeout=timeout,
                    verify=True,
                    **kwargs,
                )
                elapsed = time.monotonic() - started
                # Transport worked. A destination 403/429 is not proof that
                # the proxy itself is dead, so count it as a policy outcome.
                transport_ok = response.status_code not in {502, 503, 504}
                self.record(proxy, transport_ok, elapsed)
                if transport_ok:
                    return response
                last_error = RuntimeError(f"upstream status {response.status_code}")
            except (requests.Timeout, requests.ConnectionError) as exc:
                self.record(proxy, False, time.monotonic() - started)
                last_error = exc
            except requests.RequestException as exc:
                self.record(proxy, False, time.monotonic() - started)
                last_error = exc
        raise RuntimeError(f"all {attempts} attempts failed") from last_error


pool = ProxyRotator([
    Proxy("proxy-a", "http://user:[email protected]:8080"),
    Proxy("proxy-b", "http://user:[email protected]:8080"),
])
response = pool.get("https://example.com", attempts=3)
print(response.status_code, len(response.content))
for proxy in pool.proxies:
    print(proxy.name, proxy.score(time.monotonic()), proxy.consecutive_failures)

Use environment variables, a secret manager, or another protected configuration source for credentials in production. Never print complete proxy URLs when they contain usernames or passwords.

Designing the health score

Measure several signals

  • Recent success ratio: a rolling window prevents an old outage from dominating forever.
  • Latency: track elapsed request time and use a ceiling appropriate to the workload.
  • Failure category: separate DNS errors, connection refusal, TLS failures, timeouts, and HTTP responses.
  • Recency: an old success should gradually contribute less than a recent one.
  • Failure streak: repeated transport failures justify temporary quarantine.

Keep raw observations beside the aggregate score. Operators need to explain whether a proxy was deprioritized for slow responses, repeated connection failures, or a destination policy response.

Do not confuse transport and destination policy

A 403 or 429 can be a destination decision, not a broken proxy. Record it separately from DNS, TCP, TLS, and timeout failures. Probe an endpoint you control or are authorized to query, and make the probe representative enough to expose connectivity problems without creating unnecessary load.

Choose eligibility and recovery rules

The example applies stepped exponential cooldown after transport failures and permits a later recheck. You can instead use fixed cooldowns, a circuit-breaker state, or a separate half-open state. Tune the window length, score cutoff, and cooldown against observed workload behavior; no library defines a canonical formula.

Special offer. See more information about Outbyte and uninstall instructions. Please review EULA and Privacy policy.

Requests, urllib3, and the standard library

Approach Proxy configuration Operational controls Best fit
Requests Per-request proxies mapping for HTTP and HTTPS timeout and verify; add your own pool and scoring policy Applications already using Requests
urllib3 ProxyManager for a proxy endpoint Separate connect/read Timeout, configurable retries, connection pools, and block=True to cap active connections Services needing lower-level pooling and concurrency control
urllib.request ProxyHandler with an explicit mapping, or environment-derived settings Standard-library deployment; rotation and scoring remain application code Minimal-dependency scripts

With urllib.request, an empty mapping disables autodetected proxies; otherwise settings such as http_proxy may be inherited from the environment. Make this choice explicit so a machine’s shell configuration does not silently change routing.

Protocols, TLS, and credentials

For an HTTPS destination through an HTTP proxy, CONNECT commonly creates a tunnel. An HTTPS proxy establishes TLS to the proxy first. SOCKS support in urllib3 requires its SOCKS extra and PySocks. Confirm which scheme your provider supports instead of assuming that every proxy URL behaves identically.

Leave certificate verification enabled. Requests exposes verify, and urllib3 verifies HTTPS certificates by default and documents CA-bundle configuration. Disabling verification is not a safe troubleshooting shortcut and can expose credentials or response data.

For HTTPS forwarding, use only a trusted proxy operator: forwarding can give the proxy visibility into requests. Store credentials outside source control, redact authorization headers and proxy URLs in logs, and rotate credentials independently of health state.

Special offer. See more information about Outbyte and uninstall instructions. Please review EULA and Privacy policy.

Retries, idempotency, and concurrency

Retries are safe only when repeating the operation is safe. GET requests are often repeatable, but application semantics still matter; never blindly replay a payment, account change, upload, or other non-idempotent write merely because a proxy failed. Use an idempotency key where the destination supports one, and cap attempts per logical operation.

Proxy rotation does not control destination load. Limit request rate separately. In urllib3, a pool configured with block=True can cap simultaneous connections rather than creating unlimited active connections. A large proxy list is not a substitute for per-host concurrency and rate controls.

Troubleshooting

Every proxy times out

Check DNS, firewall egress, proxy reachability, credentials, and the connect/read timeout split. Test one proxy at a time against an authorized endpoint and inspect the exception category before changing the score formula.

HTTPS requests fail while HTTP works

Verify the proxy scheme and CONNECT support. An HTTP proxy can tunnel HTTPS destinations, but provider policy or authentication may block CONNECT. Keep TLS verification enabled and inspect the underlying exception.

Special offer. See more information about Outbyte and uninstall instructions. Please review EULA and Privacy policy.

The score stays high after failures

Inspect the rolling-window size and failure classification. If only status codes are recorded while connection exceptions bypass record, the aggregate cannot reflect transport failures. Store raw events and test cooldown transitions.

Unexpected proxies are being used

Check environment variables and process startup configuration. In standard-library code, use an explicit ProxyHandler; in Requests, pass the mapping on the request rather than relying on ambient settings.

Retries create duplicate effects

Reduce attempts for writes, use idempotency controls, and classify failures before retrying. A new proxy does not make a non-idempotent operation safe to repeat.

Independent reader supportYour contribution helps us test, update, and keep practical guides available for everyone.Support on Ko-Fi

Or skip the browser setup

If your workflow also needs rendered website captures, ScreenshotNeo provides a single screenshot API call instead of maintaining browser automation. It accepts consent banners before capture and removes more than 60 known consent platforms, newsletter popups, and chat widgets. Bot checks, blank pages, timeouts, failed loads, and cache hits are not billed, and response headers identify the page verdict and billing result. Its MCP server exposes take_screenshot, get_page_info, and capture_pdf for Claude, Cursor, and other MCP clients.

Free tools Windows power users keep installed

One-click scans. No signup required.

Special offer. See more information about Outbyte and uninstall instructions. Please review EULA and Privacy policy.

For a direct capture, see the ScreenshotNeo API documentation:

curl -G "https://api.screenshotneo.com/v1/shot" -d access_key=YOUR_API_KEY --data-urlencode url=https://stripe.com -o shot.webp

Python:

import requests
r = requests.get("https://api.screenshotneo.com/v1/shot", params={"access_key": "YOUR_API_KEY", "url": "https://stripe.com"}, timeout=90)
open("shot.webp", "wb").write(r.content)

Node.js:

const q = new URLSearchParams({ access_key: 'YOUR_API_KEY', url: 'https://stripe.com' });
const res = await fetch(`https://api.screenshotneo.com/v1/shot?${q}`);

ScreenshotNeo includes full-page and element captures, device and viewport controls, custom JavaScript and CSS, request blocking, cookies and headers, PDF options, caching, signed links, asynchronous webhooks, bulk capture, and a usage API. Every feature is available on every plan. The Free plan includes 1,000 shots per month with no card; paid plans start at $5 for 3,000 shots. Create a free ScreenshotNeo account.

FAQ

Is there a standard proxy health-score formula?

No. The score is an application policy. Publish the inputs, weights, windows, and thresholds you selected so the result is auditable and tunable.

Should a 429 automatically disable a proxy?

No. A 429 describes destination policy or capacity. Treat it separately from transport failures and apply the destination’s documented pacing or authorization rules.

Special offer. See more information about Outbyte and uninstall instructions. Please review EULA and Privacy policy.

Can rotation replace rate limiting?

No. Control concurrency and request rate for each destination regardless of how many proxies are available.

Frequently Asked Questions

Is there a standard proxy health-score formula?

No. The score is an application policy. Publish the inputs, weights, windows, and thresholds you selected so the result is auditable and tunable.

Should a 429 automatically disable a proxy?

No. A 429 describes destination policy or capacity. Treat it separately from transport failures and apply the destination’s documented pacing or authorization rules.

Can rotation replace rate limiting?

No. Control concurrency and request rate for each destination regardless of how many proxies are available.

Special offer. See more information about Outbyte and uninstall instructions. Please review EULA and Privacy policy.

Product prices and availability are accurate as of the date/time indicated and are subject to change. Any price and availability information displayed on Amazon at the time of purchase will apply.