Алжир

Реальные домашние IP

Резидентные прокси — Алжир

Алжирские резидентные прокси — это реальные IP-адреса из алжирских домохозяйств, которые выдают интернет-провайдеры для домашних пользователей во главе с Algérie Télécom. С ними вы просматриваете, парсите и проверяете алжирские витрины, цены и контент как настоящий местный пользователь — в городах Алжир, Оран, Константина и Блида — а не с IP дата-центра, который Ouedkniss, Batolis и локальные геофильтры распознают и блокируют с первого запроса.

Попробовать за $1.99/GB Войти через Google

HTTP / SOCKS5Таргетинг по городу и ISPТрафик не сгораетЧистота проверяется ежемесячно

52 835IP в пуле

25+Охват городов

2Сети ISP

370 мсМедианная задержка

99.7%Строго чистые

Резидентные прокси для страны Алжир — кратко

Резидентные прокси для страны Алжир — это реальные IP-адреса из домохозяйств региона Алжир, выданные пользовательскими ISP. Они позволяют просматривать, скрапить и проверять сайты, цены и контент региона Алжир как настоящий местный пользователь — а не с дата-центрового IP, который локальные сайты помечают и блокируют.

Тип пула: Резидентный, несколько ISP
Охват городов: Algiers, Oran, Blida
Топ ISP: Algérie Télécom
Протоколы: HTTP, SOCKS5
Ротация: На каждый запрос или sticky
Геоточность: 99.97%
Чистота пула: 99.7% clean
Тарифы: from $2.00/GB
Последний замер: 2026-06-23
Поддержка: 24/7

Почему команды выбирают наш пул региона Алжир

Реальные резидентные IP, точный таргетинг и качество, которое не нужно принимать на веру.

Реальные резидентные IP

Настоящие домашние подключения от пользовательских ISP региона Алжир — а не дата-центровые диапазоны, которые блокируются с ходу.

Таргетинг по городу и ISP

Нацеливайтесь на отдельные города и операторов там, где это позволяет глубина пула — вплоть до провайдера.

Ротация или sticky

Ротация на каждый запрос для масштаба или sticky-сессии до ~72 ч для многошаговых сценариев.

HTTP и SOCKS5

Работает с любым клиентом, скрапером или стеком автоматизации из коробки.

Этичный источник

IP из резидентной сети с явным согласием участников — реальные пользователи, согласившиеся делиться трафиком.

Проверенная чистота, ежемесячно

Мы заново замеряем пул каждый месяц и публикуем цифры, чтобы они оставались актуальными.

Качество, которое можно проверить

Каждый месяц мы прогоняем один и тот же бенчмарк — случайная выборка IP оценивается по Scamalytics, задержка тестируется с нейтрального VPS, а геоточность сверяется по независимым базам. Цифры ниже отражают боевую среду, а не идеальный лабораторный прогон.

Чистота Scamalytics · 100 000 IP

Строго чистые99.7%

Доля дата-центровых0%

В любом чёрном списке0.22%

Низкий риск фрода (0–19)99.3%

Геоточность99.97%

Задержка и скорость VPS → Cloudflare

Медианная задержка370мс

Задержка p90 / p95480 / 622 мс

Успешность подключений98.8%

Пропускная способность (медиана / p95)0.51 / 1.84 MB/s

Загрузка крупных файлов82%

Состав сети

Тип подключения100% резидентные

Сети ISP (ASN)2

Режимы ротациина запрос · sticky

Топ операторовAlgérie Télécom

Замерено на 100 000 адресах с нейтрального VPS относительно edge-сети Cloudflare, актуально на момент последнего бенчмарка. Повторный прогон может отличаться на небольшую величину, но цифры полностью отражают порядок величины и служат надёжной основой, чтобы оценить, подходит ли пул под вашу задачу. Успешность скрапинга по конкретным целям не публикуется — она зависит от вашего отпечатка и настройки, а не только от IP.

benchmark.py — открытый бенчмарк пула

Открытый бенчмарк, стоящий за приведёнными выше цифрами. Он берёт случайную выборку IP пула и измеряет четыре вещи — чистоту (фрод-риск Scamalytics, дата-центр, чёрные списки и геоточность), задержку (круговой обход до edge Cloudflare и ip-api), ротацию (уникальные IP, ASN и подсети /24) и пропускную способность (скорость загрузки и время до первого байта) — после чего сообщает каждый показатель. Он опубликован целиком, чтобы методология за каждой цифрой оставалась открытой для проверки.

# -*- coding: utf-8 -*-

"""
JETPROXY — POOL BENCHMARK

End-to-end audit of a residential proxy pool. A single run measures cleanliness,
latency, rotation and throughput, then prints every metric as one list.

  Cleanliness   fraud-risk, datacenter, blacklist and geo-accuracy (Scamalytics)
  Latency       Cloudflare edge (responsiveness) and ip-api origin (round-trip)
  Rotation      unique IPs / ASNs / subnets, success rate, country match
  Throughput    median & p95 download speed, time-to-first-byte, drop rate

Every stage size is a parameter, so the same script runs a quick check or a deep
audit:

    python benchmark.py --country ID
    python benchmark.py --quality 200 --downloads 50 --latency 1000 --rotation 5000 --country ID
    python benchmark.py --proxies proxies.txt --country ID --json result.json

Proxies   paste at the prompt (one per line) or pass --proxies FILE.
          host:port:user:pass  ·  host:port  ·  user:pass@host:port
The Scamalytics key for the cleanliness stage is read from the environment or
../.env (SCAMALYTICS_USER + SCAMALYTICS_KEY); without it that stage is skipped.
"""
from __future__ import annotations

import argparse
import asyncio
import csv
import ipaddress
import json
import os
import re
import statistics
import sys
import time
from collections import Counter
from datetime import datetime

try:
    sys.stdout.reconfigure(encoding="utf-8")
    sys.stderr.reconfigure(encoding="utf-8")
except Exception:
    pass

try:
    import aiohttp
    from rich.console import Console
    from rich.table import Table
    from rich import box
except ImportError:
    print("\n[!] Need: pip install aiohttp rich\n")
    sys.exit(1)

console = Console()

# ── endpoints ────────────────────────────────────────────────────────────────

IP_ENDPOINT = ("http://ip-api.com/json/?fields="
               "status,message,country,countryCode,regionName,city,isp,as,org,query")
CF_TRACE = "https://speed.cloudflare.com/cdn-cgi/trace"   # tiny edge response (~350 B)

FILE_MB = 10
DOWN_URL = f"https://speed.cloudflare.com/__down?bytes={FILE_MB * 1_000_000}"
SCAM_HOST = "https://api11.scamalytics.com/v3"            # US node (api12 = EU)


SCORE_CONCURRENCY = 10
DOWN_CONCURRENCY = 3   # downloads share the local uplink — keep low so the exit node is the bottleneck



# ── proxy parsing ──────────────────────────────────────────────────────────────

def parse_proxy(raw: str):
    """host:port | host:port:user:pass | user:pass@host:port | scheme://..."""
    raw = raw.strip()
    if not raw or raw.startswith("#"):
        return None
    scheme = "http"
    if "://" in raw:
        scheme, raw = raw.split("://", 1)
    user = pwd = host = port = None
    if "," in raw:
        parts = [p.strip() for p in raw.split(",")]
        if len(parts) == 4:
            host, port, user, pwd = parts
        elif len(parts) == 2:
            host, port = parts
        else:
            return None
    elif "@" in raw:
        cred, hostport = raw.rsplit("@", 1)
        if ":" in cred:
            user, pwd = cred.split(":", 1)
        host, port = hostport.rsplit(":", 1)
    else:
        parts = raw.split(":")
        if len(parts) == 2:
            host, port = parts
        elif len(parts) == 4:
            host, port, user, pwd = parts
        else:
            return None
    try:
        port = int(port)
    except (ValueError, TypeError):
        return None
    return {"scheme": scheme, "host": host, "port": port, "user": user, "pwd": pwd,
            "url": f"{scheme}://{host}:{port}", "label": f"{host}:{port}"}


def subnet24(ip: str) -> str:
    try:
        return str(ipaddress.ip_network(ip + "/24", strict=False))
    except ValueError:
        return ""


def pct(values, p):
    if not values:
        return 0.0
    s = sorted(values)
    k = min(len(s) - 1, max(0, int(round(p / 100 * (len(s) - 1)))))
    return s[k]


# ── ip-api request (identity + geo + round-trip) ───────────────────────────────

def _fail(idx, proxy, err, latency):
    return {"idx": idx, "ok": False, "proxy": proxy["label"], "error": err, "latency": latency}


async def _one_attempt(session, proxy, idx, timeout):
    t0 = time.perf_counter()
    auth = aiohttp.BasicAuth(proxy["user"], proxy["pwd"]) if proxy["user"] else None
    try:
        async with session.get(IP_ENDPOINT, proxy=proxy["url"], proxy_auth=auth,
                               timeout=aiohttp.ClientTimeout(total=timeout)) as r:
            latency = (time.perf_counter() - t0) * 1000
            if r.status != 200:
                return _fail(idx, proxy, f"HTTP {r.status}", latency)
            data = await r.json(content_type=None)
            if data.get("status") != "success":
                return _fail(idx, proxy, data.get("message", "api error"), latency)
            ip = data.get("query", "")
            asn = data.get("as", "").split(" ")[0] if data.get("as") else ""
            return {"idx": idx, "ok": True, "proxy": proxy["label"], "ip": ip,
                    "country": data.get("country", ""), "cc": data.get("countryCode", ""),
                    "region": data.get("regionName", ""), "city": data.get("city", ""),
                    "isp": data.get("isp") or data.get("org", ""), "asn": asn,
                    "org": data.get("org") or data.get("as", ""),
                    "subnet": subnet24(ip), "latency": latency, "error": ""}
    except asyncio.TimeoutError:
        return _fail(idx, proxy, "timeout", (time.perf_counter() - t0) * 1000)
    except Exception as e:
        return _fail(idx, proxy, type(e).__name__, (time.perf_counter() - t0) * 1000)


async def fetch(session, proxy, sem, idx, timeout, max_retries=2):
    async with sem:
        res = None
        for attempt in range(1, max_retries + 2):
            res = await _one_attempt(session, proxy, idx, timeout)
            if res["ok"]:
                res["attempts"] = attempt
                res["first_try_ok"] = (attempt == 1)
                return res
        res["attempts"] = max_retries + 1
        res["first_try_ok"] = False
        return res


# ── Cloudflare edge probe (responsiveness + nearest datacenter) ────────────────

async def cf_probe(proxy, sem, timeout):
    """One warm-up request opens the tunnel + TLS, then the next request on the same
    kept-alive connection is timed — so the figure is the edge round-trip, not the
    one-off connection setup."""
    async with sem:
        auth = aiohttp.BasicAuth(proxy["user"], proxy["pwd"]) if proxy["user"] else None
        try:
            async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=2, ssl=False)) as sess:
                async with sess.get(CF_TRACE, proxy=proxy["url"], proxy_auth=auth,
                                    timeout=aiohttp.ClientTimeout(total=15)) as w:
                    await w.read()
                t0 = time.perf_counter()
                async with sess.get(CF_TRACE, proxy=proxy["url"], proxy_auth=auth,
                                    timeout=aiohttp.ClientTimeout(total=timeout)) as r:
                    txt = await r.text()
                    rtt = (time.perf_counter() - t0) * 1000
                    if r.status != 200:
                        return {"ok": False, "error": f"HTTP {r.status}"}
                    d = {}
                    for ln in txt.splitlines():
                        if "=" in ln:
                            k, v = ln.split("=", 1)
                            d[k] = v
                    return {"ok": True, "rtt": rtt, "ip": d.get("ip", ""),
                            "loc": d.get("loc", ""), "colo": d.get("colo", "")}
        except asyncio.TimeoutError:
            return {"ok": False, "error": "timeout"}
        except Exception as e:
            return {"ok": False, "error": type(e).__name__}


# ── download (throughput + TTFB + mid-transfer drops) ──────────────────────────

async def download(proxy, sem, idx, timeout, retries):
    async with sem:
        auth = aiohttp.BasicAuth(proxy["user"], proxy["pwd"]) if proxy["user"] else None
        last_err = ""
        dropped_at = []
        for attempt in range(1, retries + 2):
            try:
                async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=4, ssl=False)) as sess:
                    country = "?"
                    try:
                        async with sess.get(CF_TRACE, proxy=proxy["url"], proxy_auth=auth,
                                            timeout=aiohttp.ClientTimeout(total=15)) as tr:
                            for ln in (await tr.text()).splitlines():
                                if ln.startswith("loc="):
                                    country = ln[4:].strip() or "?"
                    except Exception:
                        pass
                    t0 = time.perf_counter()
                    async with sess.get(DOWN_URL, proxy=proxy["url"], proxy_auth=auth,
                                        timeout=aiohttp.ClientTimeout(total=timeout)) as r:
                        if r.status != 200:
                            last_err = f"HTTP {r.status}"
                            continue
                        first = None
                        nbytes = 0
                        try:
                            async for chunk in r.content.iter_chunked(65536):
                                if first is None:
                                    first = time.perf_counter()
                                nbytes += len(chunk)
                        except Exception as stream_err:
                            if nbytes > 0:
                                dropped_at.append(nbytes)
                                last_err = f"dropped@{nbytes // 1_000_000}MB"
                            else:
                                last_err = type(stream_err).__name__
                            continue
                        end = time.perf_counter()
                        if nbytes == 0:
                            last_err = "empty"
                            continue
                        ttfb = (first - t0) * 1000 if first else 0
                        dl_time = (end - first) if first and end > first else (end - t0)
                        mbps = (nbytes / 1_000_000) / dl_time if dl_time > 0 else 0
                        return {"idx": idx, "ok": True, "country": country, "bytes": nbytes,
                                "ttfb_ms": ttfb, "mbps": mbps, "dropped": bool(dropped_at), "error": ""}
            except asyncio.TimeoutError:
                last_err = "timeout"
            except Exception as e:
                last_err = type(e).__name__
        return {"idx": idx, "ok": False, "country": "?", "bytes": 0, "ttfb_ms": 0,
                "mbps": 0, "dropped": bool(dropped_at), "error": last_err}


# ── Scamalytics scoring ────────────────────────────────────────────────────────

def load_creds():
    user = os.environ.get("SCAMALYTICS_USER")
    key = os.environ.get("SCAMALYTICS_KEY")
    if user and key:
        return user.strip(), key.strip()
    found = {}
    for path in ("../.env", "../.env.local", ".env"):
        p = os.path.join(os.path.dirname(__file__), path)
        if os.path.exists(p):
            with open(p, encoding="utf-8") as f:
                for line in f:
                    m = re.match(r'\s*(SCAMALYTICS_USER|SCAMALYTICS_KEY)\s*=\s*"?([^"\s]+)"?', line)
                    if m:
                        found[m.group(1)] = m.group(2)
    return found.get("SCAMALYTICS_USER"), found.get("SCAMALYTICS_KEY")


def parse_scam(ip, data):
    s = data.get("scamalytics", {})
    if s.get("status") != "ok":
        return {"ip": ip, "ok": False, "error": s.get("status") or "bad_response"}
    ext = data.get("external_datasources", {})
    score = s.get("scamalytics_score")
    risk = s.get("scamalytics_risk", "?")
    px = s.get("scamalytics_proxy", {})
    is_dc = bool(px.get("is_datacenter"))
    is_vpn = bool(px.get("is_vpn"))
    fh = bool(ext.get("firehol", {}).get("is_proxy"))
    ips_bl = bool(ext.get("ipsum", {}).get("ip_blacklisted"))
    sh = bool(ext.get("spamhaus_drop", {}).get("ip_blacklisted"))
    x4 = bool(ext.get("x4bnet", {}).get("is_blacklisted_spambot"))
    on_bl = bool(s.get("is_blacklisted_external")) or fh or ips_bl or sh or x4
    cc = (ext.get("dbip", {}).get("ip_country_code")
          or ext.get("ipinfo", {}).get("ip_country_code")
          or ext.get("maxmind_geolite2", {}).get("ip_country_code")
          or ext.get("ip2proxy_lite", {}).get("ip_country_code") or "?")
    isp = (ext.get("dbip", {}).get("isp_name")
           or ext.get("ip2proxy_lite", {}).get("isp_name")
           or s.get("scamalytics_isp") or "?")
    asn = ext.get("ipinfo", {}).get("asn") or ext.get("ip2proxy_lite", {}).get("asn") or "?"
    return {"ip": ip, "ok": True, "error": "", "score": score if isinstance(score, int) else 0,
            "risk": risk, "is_dc": is_dc, "is_vpn": is_vpn, "on_bl": on_bl,
            "country": cc, "isp": isp, "asn": asn,
            "credits": s.get("credits", {}).get("remaining"),
            "strict_clean": (risk == "low" and not is_dc and not on_bl)}


async def score_one(session, user, key, ip, sem):
    async with sem:
        try:
            async with session.get(f"{SCAM_HOST}/{user}/", params={"key": key, "ip": ip},
                                   timeout=aiohttp.ClientTimeout(total=20)) as r:
                return parse_scam(ip, await r.json(content_type=None))
        except Exception as e:
            return {"ip": ip, "ok": False, "error": type(e).__name__}


# ── stage runners ──────────────────────────────────────────────────────────────

def progress(tag, done, total):
    step = max(1, total // 10)
    if done % step == 0 or done == total:
        console.print(f"   [{tag}] {done}/{total}", highlight=False)


async def run_ipapi(proxies, n, concurrency, timeout, retries, tag):
    sem = asyncio.Semaphore(concurrency)
    conn = aiohttp.TCPConnector(limit=concurrency * 3, ssl=False)
    out = []
    async with aiohttp.ClientSession(connector=conn) as http:
        tasks = [asyncio.create_task(fetch(http, proxies[i % len(proxies)], sem, i, timeout, retries))
                 for i in range(n)]
        for done, coro in enumerate(asyncio.as_completed(tasks), 1):
            out.append(await coro)
            progress(tag, done, n)
    return out


async def run_cf(proxies, n, concurrency, timeout, tag):
    sem = asyncio.Semaphore(concurrency)
    tasks = [asyncio.create_task(cf_probe(proxies[i % len(proxies)], sem, timeout))
             for i in range(n)]
    out = []
    for done, coro in enumerate(asyncio.as_completed(tasks), 1):
        out.append(await coro)
        progress(tag, done, n)
    return out


async def run_download(proxies, n, timeout, retries, tag):
    sem = asyncio.Semaphore(DOWN_CONCURRENCY)
    tasks = [asyncio.create_task(download(proxies[i % len(proxies)], sem, i, timeout, retries))
             for i in range(n)]
    out = []
    for done, coro in enumerate(asyncio.as_completed(tasks), 1):
        out.append(await coro)
        progress(tag, done, n)
    return out


async def score_ips(ips, user, key, tag):
    sem = asyncio.Semaphore(SCORE_CONCURRENCY)
    out = []
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(score_one(session, user, key, ip, sem)) for ip in ips]
        for done, coro in enumerate(asyncio.as_completed(tasks), 1):
            out.append(await coro)
            progress(tag, done, len(ips))
    return out


# ── input ──────────────────────────────────────────────────────────────────────

def read_proxies(path):
    if path:
        with open(path, encoding="utf-8") as f:
            lines = f.readlines()
    else:
        console.print("Paste proxies (one per line), then an empty line:")
        lines = []
        while True:
            try:
                ln = input()
            except EOFError:
                break
            if ln.strip() == "":
                if lines:
                    break
                continue
            lines.append(ln)
    return [p for p in (parse_proxy(l) for l in lines) if p]


def fmt(v, suffix=""):
    return f"{v}{suffix}"


# ── main ─────────────────────────────────────────────────────────────────────

def build_report(M):
    t = Table(title="POOL BENCHMARK — RESULTS", box=box.DOUBLE_EDGE, header_style="bold cyan",
              title_style="bold green")
    t.add_column("Metric", style="cyan")
    t.add_column("Value", style="bold white", justify="right")
    for section, rows in M["sections"]:
        t.add_row(f"[bold yellow]{section}[/]", "")
        for k, v in rows:
            t.add_row(f"  {k}", str(v))
    console.print()
    console.print(t)


async def run(args, proxies):
    M = {"label": args.label, "country": args.country or None,
         "timestamp": datetime.now().isoformat(timespec="seconds"),
         "proxies_input": len(proxies), "stages": {}}
    user = key = None

    # 1 — cleanliness (Scamalytics)

    clean_rows = []
    if not args.no_quality and args.quality > 0:
        user, key = load_creds()
        if not user or not key:
            console.print("[yellow]Scamalytics key not found — skipping cleanliness stage.[/]")
        else:
            console.rule(f"[bold]Cleanliness — {args.quality} IPs[/]")
            collected = await run_ipapi(proxies, int(args.quality * 1.6) + 10,
                                        args.concurrency, 15, 1, "collect")
            uniq = {}
            for r in collected:
                if r.get("ok") and r.get("ip") and r["ip"] not in uniq:
                    uniq[r["ip"]] = r
            ips = list(uniq.keys())[:args.quality]
            scored = await score_ips(ips, user, key, "score")
            ok = [r for r in scored if r.get("ok")]
            n = len(ok)
            if n:
                strict = sum(r["strict_clean"] for r in ok) / n * 100
                low = sum(1 for r in ok if r["risk"] == "low") / n * 100
                dc = sum(r["is_dc"] for r in ok) / n * 100
                bl = sum(r["on_bl"] for r in ok) / n * 100
                vpn = sum(r["is_vpn"] for r in ok)
                geo = None
                if args.country:
                    geo = sum(1 for r in ok if (r["country"] or "").upper() == args.country.upper()) / n * 100
                M["stages"]["cleanliness"] = {
                    "n": n, "strict_clean_pct": round(strict, 1), "low_risk_pct": round(low, 1),
                    "datacenter_pct": round(dc, 1), "blacklist_pct": round(bl, 1),
                    "geo_accuracy_pct": round(geo, 1) if geo is not None else None, "vpn": vpn}
                clean_rows = [("Strictly clean", f"{strict:.1f}%"),
                              ("Low fraud-risk (0–19)", f"{low:.1f}%"),
                              ("Datacenter", f"{dc:.1f}%"),
                              ("Blacklisted", f"{bl:.1f}%")]
                if geo is not None:
                    clean_rows.append((f"Geo-accuracy ({args.country.upper()})", f"{geo:.1f}%"))
                # per-IP CSV

                ts = datetime.now().strftime("%Y%m%d_%H%M%S")
                with open(f"benchmark_ips_{ts}.csv", "w", newline="", encoding="utf-8") as f:
                    w = csv.writer(f)
                    w.writerow(["ip", "score", "risk", "is_datacenter", "is_vpn", "on_blacklist",
                                "country", "isp", "asn", "strict_clean"])
                    for r in scored:
                        if r.get("ok"):
                            w.writerow([r["ip"], r["score"], r["risk"], r["is_dc"], r["is_vpn"],
                                        r["on_bl"], r["country"], r["isp"], r["asn"], r["strict_clean"]])

    # 2 — throughput

    console.rule(f"[bold]Throughput — {args.downloads} downloads[/]")
    dl = await run_download(proxies, args.downloads, 60, 1, "download")
    dok = [r for r in dl if r["ok"]]
    dl_rows = []
    if dok:
        mbps = [r["mbps"] for r in dok]
        ttfb = [r["ttfb_ms"] for r in dok]
        nodrop = sum(1 for r in dl if r["ok"] and not r["dropped"]) / len(dl) * 100
        M["stages"]["throughput"] = {
            "n": len(dl), "ok": len(dok),
            "median_mbps": round(statistics.median(mbps), 2), "p95_mbps": round(pct(mbps, 95), 2),
            "median_ttfb_ms": round(statistics.median(ttfb)), "completed_no_drop_pct": round(nodrop, 1)}
        dl_rows = [("Median download", f"{statistics.median(mbps):.1f} MB/s ({statistics.median(mbps)*8:.0f} Mbps)"),
                   ("p95 download", f"{pct(mbps,95):.1f} MB/s ({pct(mbps,95)*8:.0f} Mbps)"),
                   ("Median TTFB", f"{statistics.median(ttfb):.0f} ms"),
                   ("Completed without drop", f"{nodrop:.0f}%")]

    # 3 — latency (Cloudflare edge + ip-api origin)

    console.rule(f"[bold]Latency — {args.latency} probes[/]")
    cf = await run_cf(proxies, args.latency, args.concurrency, 15, "cf")
    cfok = [r for r in cf if r["ok"]]
    lat_rows = []
    if cfok:
        rtts = [r["rtt"] for r in cfok]
        colo = Counter(r["colo"] for r in cfok if r["colo"]).most_common(1)
        M["stages"].setdefault("latency", {})
        M["stages"]["latency"]["cloudflare"] = {
            "n": len(cfok), "p50_ms": round(statistics.median(rtts)),
            "p90_ms": round(pct(rtts, 90)), "p95_ms": round(pct(rtts, 95)),
            "top_colo": colo[0][0] if colo else None}
        lat_rows += [("Cloudflare edge — p50", f"{statistics.median(rtts):.0f} ms"),
                     ("Cloudflare edge — p90", f"{pct(rtts,90):.0f} ms"),
                     ("Cloudflare edge — p95", f"{pct(rtts,95):.0f} ms")]
        if colo:
            lat_rows.append(("Nearest datacenter", colo[0][0]))

    # 4 — rotation (also yields ip-api round-trip latency)

    console.rule(f"[bold]Rotation — {args.rotation} requests[/]")
    rot = await run_ipapi(proxies, args.rotation, args.concurrency, args.timeout, 2, "rotation")
    rok = [r for r in rot if r["ok"]]
    rot_rows = []
    if rok:
        uips = {r["ip"] for r in rok if r.get("ip")}
        uasn = {r["asn"] for r in rok if r.get("asn")}
        usub = {r["subnet"] for r in rok if r.get("subnet")}
        success = len(rok) / len(rot) * 100
        first = sum(1 for r in rok if r.get("first_try_ok")) / len(rot) * 100
        cmatch = None
        if args.country:
            cmatch = sum(1 for r in rok if (r["cc"] or "").upper() == args.country.upper()) / len(rok) * 100
        ipapi_rtts = [r["latency"] for r in rok]
        M["stages"]["rotation"] = {
            "n": len(rot), "success_pct": round(success, 1), "first_try_pct": round(first, 1),
            "unique_ips": len(uips), "unique_asns": len(uasn), "unique_subnets": len(usub),
            "country_match_pct": round(cmatch, 2) if cmatch is not None else None}
        M["stages"].setdefault("latency", {})
        M["stages"]["latency"]["ip_api"] = {
            "n": len(ipapi_rtts), "p50_ms": round(statistics.median(ipapi_rtts)),
            "p90_ms": round(pct(ipapi_rtts, 90))}
        lat_rows += [("ip-api origin — p50", f"{statistics.median(ipapi_rtts):.0f} ms"),
                     ("ip-api origin — p90", f"{pct(ipapi_rtts,90):.0f} ms")]
        rot_rows = [("Success rate", f"{success:.1f}%"),
                    ("First-try success", f"{first:.1f}%"),
                    ("Unique exit IPs", f"{len(uips)}"),
                    ("Unique networks (ASN)", f"{len(uasn)}"),
                    ("Unique /24 subnets", f"{len(usub)}")]
        if cmatch is not None:
            rot_rows.append((f"Country match ({args.country.upper()})", f"{cmatch:.2f}%"))

    M["sections"] = [s for s in [
        ("CLEANLINESS", clean_rows), ("LATENCY", lat_rows),
        ("ROTATION", rot_rows), ("THROUGHPUT", dl_rows)] if s[1]]
    build_report(M)

    out_path = args.json or f"benchmark_{args.label}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    M.pop("sections", None)
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(M, f, indent=2, ensure_ascii=False)
    console.print(f"\n[green]✓[/] Saved: [bold]{out_path}[/]")


def main():
    ap = argparse.ArgumentParser(description="Jetproxy pool benchmark")
    ap.add_argument("--quality", type=int, default=100, help="IPs scored for cleanliness")
    ap.add_argument("--downloads", type=int, default=50, help="download samples")
    ap.add_argument("--latency", type=int, default=1000, help="latency probes")
    ap.add_argument("--rotation", type=int, default=2000, help="rotation requests")
    ap.add_argument("--concurrency", type=int, default=20)
    ap.add_argument("--timeout", type=float, default=15.0)
    ap.add_argument("--country", default="", help="expected ISO code for geo-accuracy")
    ap.add_argument("--proxies", default="", help="proxy list file (else paste)")
    ap.add_argument("--label", default="pool")
    ap.add_argument("--json", default="")
    ap.add_argument("--no-quality", action="store_true", help="skip the cleanliness stage")
    args = ap.parse_args()

    console.print("\n[bold cyan]═══  JETPROXY — POOL BENCHMARK  ═══[/]\n")
    proxies = read_proxies(args.proxies)
    if not proxies:
        console.print("[red]No valid proxies.[/]")
        return
    console.print(f"[green]✓[/] Proxies loaded: [bold]{len(proxies)}[/]")
    asyncio.run(run(args, proxies))


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        console.print("\n[yellow]Interrupted.[/]")

Таргетинг вплоть до города

«Сред. онлайн» — это среднее число различных устройств, подключённых и доступных в один и тот же момент, а не суммарный итог. Сгруппировано по уровням Высокий / Средний / Низкий.

РегионыГородаПровайдеры

Регион	Задержка	Сред. онлайн	Пул
Algiers	364 ms	~1 203	Высокий
Oran	342 ms	~585	Высокий
Blida	370 ms	~454	Высокий
Constantine	391 ms	~350	Высокий
Tlemcen	364 ms	~285	Высокий
Sétif	386 ms	~250	Высокий
Tizi Ouzou	385 ms	~248	Высокий
Annaba	359 ms	~222	Высокий
Batna	403 ms	~187	Высокий
Ouargla	406 ms	~137	Высокий
Chlef	390 ms	~83	Средний
Sidi Bel Abbès	383 ms	~83	Средний
Mascara	339 ms	~73	Средний
Laghouat	364 ms	~68	Средний
Béjaïa	410 ms	~55	Средний
Béchar	350 ms	~41	Средний
Khenchela	479 ms	~41	Средний
Tipaza	364 ms	~40	Средний
Mila	372 ms	~33	Средний
Relizane	448 ms	~32	Средний
Tiaret	373 ms	~29	Низкий
Mostaganem	339 ms	~28	Низкий
Adrar	440 ms	~27	Низкий
El Oued	415 ms	~24	Низкий
Ghardaia	434 ms	~23	Низкий
Medea	355 ms	~21	Низкий
Biskra	380 ms	~21	Низкий
Tébessa	413 ms	~20	Низкий
Bouira	369 ms	~18	Низкий
Boumerdes	391 ms	~17	Низкий

Город	Задержка	Сред. онлайн	Пул
Algiers	364 ms	~1 127	Высокий
Oran	342 ms	~634	Высокий
Blida	371 ms	~489	Высокий
Constantine	391 ms	~380	Высокий
Sétif	386 ms	~271	Высокий
Tlemcen	358 ms	~270	Высокий
Tizi Ouzou	380 ms	~246	Высокий
Annaba	359 ms	~238	Высокий
Batna City	406 ms	~191	Высокий
Ouargla	404 ms	~138	Средний
Sidi Bel Abbes	383 ms	~90	Средний
Chlef	389 ms	~86	Средний
Laghouat	364 ms	~73	Средний
Bordj el Kiffan	361 ms	~72	Средний
Mascara	338 ms	~60	Средний
Kouba	357 ms	~52	Средний
Béchar	350 ms	~45	Низкий
Béjaïa	419 ms	~44	Низкий
Khenchela	480 ms	~41	Низкий
Tiaret	373 ms	~31	Низкий
Mostaganem	339 ms	~30	Низкий
Aoulef	434 ms	~24	Низкий
Medea	355 ms	~23	Низкий
Biskra	380 ms	~23	Низкий
Tébessa	413 ms	~22	Низкий

Провайдер	Задержка	Сред. онлайн	Пул
Algérie Télécomфиксированный	371 ms	~4 700	Высокий

Геоточность — каждый IP, продаваемый как Алжир, резолвится в страну Алжир по трём независимым гео-базам (db-ip, MaxMind, ipinfo); высокая геоточность — это сигнал, на который опираются локальные гео-фильтры.

Прокси для страны Алжир по сценариям использования

Организовано по тому, что вы делаете, а не по платформам. Локальные сервисы перечислены там, где они важны.

E-commerce и анализ рынка

Проверяйте витрины, цены и локальные листинги изнутри рынка.

Мониторинг витрин
Отслеживание локальных цен
Исследование продавцов

OuedknissBatolisWebStar ElectroMdinjdidaHamizOnline

Верификация и эффективность рекламы

Смотрите, как реклама, креативы и SERP отображаются локальному пользователю — без ручного переключения.

Локальная SERP
Проверка креативов
Лендинги

FacebookInstagramTikTokYouTube

Тревел и анализ цен

Снимайте тарифы на авиабилеты и отели так, как их видит местный клиент.

Гео-зависимые тарифы
Мониторинг предложений
Сравнение расценок

Air AlgérieTassili AirlinesNreserviH24VoyagesBooking.com

Стриминг и гео-верификация

Проверяйте региональные каталоги и гео-фильтры из нужной локации.

Гео-проверки
Верификация каталога
Региональная доступность

Yara (Ooredoo)Dzair Play (Mobilis)Shahid (MBC)

Объявления и локальные листинги

Отслеживайте товары б/у, авто и аренду на локальных досках объявлений.

Мониторинг объявлений
Отслеживание цен
Региональный охват

OuedknissEmploitic

Разработка и автоматизация

Резидентные IP для пайплайнов, тестов и разрешённого сбора данных через API.

Интеграция в пайплайн
Управление через API
Sticky-сессии

PythonScrapycURLGitHub

Помимо локальных сайтов, тот же пул региона Алжир работает с крупнейшими глобальными платформами — Google, Meta, Amazon, YouTube, TikTok и Netflix — в том виде, в каком они отображаются в локали региона Алжир: локальные цены, реклама, каталоги, SERP и контент.

Для разработчиков API

Генерируйте эндпоинты с привязкой к стране Алжир в панели управления, копируйте проверенные сниппеты или автоматизируйте списки прокси через публичный API. Город, ISP, sticky-сессии и сохранённые списки — всё в одном удобном рабочем процессе.

Документация API

1import requests
2 
3headers = {"x-api-key": "gpx_live_YOUR_KEY"}
4payload = {
5  "country_code": "DZ",
6  "city": "Algiers",
7  "session_type": "sticky",
8  "quantity": 5,
9  "protocol": "socks5"
10}
11 
12response = requests.post(
13  "https://jetproxy.io/api/v1/generate",
14  headers=headers,
15  json=payload,
16  timeout=30
17)
18print(response.json()["proxies"])

Пакеты без скрытых условий

Чем больше объём, тем ниже цена за гигабайт. Без подписок и скрытых платежей.

1 GB$3.70

$3.70 / GB

Выбрать

5 GB$18.00

$3.60 / GB

Выбрать

10 GB$34.50

$3.45 / GB

Выбрать

Популярный

25 GB$80.00

$3.20 / GB

Выбрать

50 GB$147.50

$2.95 / GB

Выбрать

100 GB$270.00

$2.70 / GB

Выбрать

250 GB$612.50

$2.45 / GB

Выбрать

500 GB$1,100.00

$2.20 / GB

Выбрать

1 TB$2,000.00

$2.00 / GB

Выбрать

Пакет

Цена за ГБ

Итого

1 GB

$3.70 / GB

$3.70Выбрать

5 GB

$3.60 / GB

$18.00Выбрать

10 GB

$3.45 / GB

$34.50Выбрать

Популярный

25 GB

$3.20 / GB

$80.00Выбрать

50 GB

$2.95 / GB

$147.50Выбрать

100 GB

$2.70 / GB

$270.00Выбрать

250 GB

$2.45 / GB

$612.50Выбрать

500 GB

$2.20 / GB

$1,100.00Выбрать

1 TB

$2.00 / GB

$2,000.00Выбрать

Доступные способы оплаты

Вопросы, которые покупатели задают на самом деле

Всё ещё взвешиваете, подойдёт ли пул региона Алжир под вашу задачу? Наша команда отвечает на реальные вопросы в Telegram, круглосуточно — до и после покупки.

Написать в Telegram

Подойдут ли они для парсинга Ouedkniss и Batolis?

Это основной сценарий использования. Пул полностью резидентный — именно это свойство ищут антибот-системы местных маркетплейсов — поэтому такие IP проходят там, где диапазоны дата-центров блокируются. Результат по-прежнему зависит от вашего отпечатка, заголовков и темпа запросов, так что сочетайте прокси с чистой конфигурацией и соответствующим алжирским часовым поясом (CET, UTC+1).

Можно ли нацелиться на конкретный город, например Алжир или Оран?

Да, насколько позволяет глубина пула — выберите город при генерации списка прокси. Наши таблицы покрытия показывают актуальное число одновременных подключений для городов Алжир, Оран, Блида, Константина, Сетиф и других, так что вы заранее увидите, в каких городах самая глубокая ротация, прежде чем сделать выбор.

Геолоцируются ли IP внутри Алжира?

Да — в нашем последнем замере пул определяется как Алжир с точностью геолокации 99,97% по данным Cloudflare и независимых геобаз, а выходные узлы подтверждены рядом с Аннабой (AAE) и Ораном (ORN). Высокая точность геолокации — это базовый сигнал, на который опираются алжирские геофильтры.

Можно ли удержать один алжирский IP на многошаговый сценарий?

Да. Используйте sticky-сессию, чтобы сохранить один и тот же алжирский выходной IP — средняя длительность удержания около 5 часов и до 72 часов по нашему 3-дневному замеру. Для масштаба ротируйте по запросу: выходной IP меняется примерно в 98% запросов, обычно на другую линию, а не просто на соседний адрес.

Какие алжирские операторы и города охвачены?

Пул возглавляет Algérie Télécom, охватывая города Алжир, Оран, Блида, Константина, Сетиф, Тлемсен, Тизи-Узу и Аннаба — всего 30 регионов и 25 городов, так что вы можете маршрутизировать трафик через крупнейшую в стране сеть фиксированной связи для домашних пользователей, причём из основных населённых центров.