Kenya

IP nhà dân thật

Proxy dân cư Kenya

Proxy dân cư Kenya là các địa chỉ IP thật từ hộ gia đình Kenya, do các ISP tiêu dùng như Safaricom, Airtel, Wananchi Group và Jamii Telecommunications cấp phát. Chúng cho phép bạn duyệt web, thu thập dữ liệu và kiểm tra các gian hàng, giá cả và nội dung của Kenya với tư cách một người dùng nội địa thực thụ — tại Nairobi, Mombasa, Nakuru và Westlands — thay vì từ một IP trung tâm dữ liệu mà Jumia, Kilimall và các cổng chặn theo vị trí địa lý lập tức phát hiện và chặn ngay.

Dùng thử với $1.99/GB Đăng nhập bằng Google

HTTP / SOCKS5Nhắm theo thành phố & ISPLưu lượng không hết hạnKiểm định sạch hàng tháng

22.663IP trong pool

25+Thành phố phủ sóng

71Mạng ISP

535 msĐộ trễ trung vị

98.8%Sạch tuyệt đối

Tổng quan nhanh về proxy dân cư Kenya

Proxy dân cư Kenya là địa chỉ IP thật từ các hộ gia đình Kenya, do các ISP tiêu dùng cấp phát. Chúng cho phép bạn duyệt web, thu thập dữ liệu và kiểm tra website, mức giá cùng nội dung Kenya với tư cách một người dùng nội địa thực thụ — thay vì từ một IP trung tâm dữ liệu bị các trang nội địa gắn cờ và chặn.

Loại pool: Dân cư, đa ISP
Thành phố phủ sóng: Nairobi, Nakuru, Bungoma
ISP hàng đầu: Safaricom · Airtel Kenya · Wananchi Group (Zuku)
Giao thức: HTTP, SOCKS5
Xoay IP: Theo từng yêu cầu hoặc cố định
Độ chính xác địa lý: 99.82%
Độ sạch của pool: 98.8% clean
Giá: from $2.00/GB
Đo lần cuối: 2026-06-23
Hỗ trợ: 24/7

Vì sao các đội ngũ chọn pool Kenya của chúng tôi

IP dân cư thật, nhắm mục tiêu chính xác và chất lượng bạn không phải tin suông.

IP dân cư thật

Kết nối nhà dân đích thực từ các ISP tiêu dùng Kenya — không phải dải IP trung tâm dữ liệu bị chặn ngay tức khắc.

Nhắm theo thành phố & ISP

Nhắm đến từng thành phố và nhà mạng cụ thể khi độ sâu của pool cho phép — chi tiết đến từng nhà cung cấp.

Xoay IP hoặc cố định

Xoay IP theo từng yêu cầu để mở rộng quy mô, hoặc phiên cố định lên đến ~72h cho các quy trình nhiều bước.

HTTP & SOCKS5

Hoạt động ngay với mọi ứng dụng, công cụ scraping hay nền tảng tự động hóa.

Nguồn gốc minh bạch

IP từ mạng dân cư theo cơ chế đăng ký tự nguyện — người dùng thật đã đồng ý chia sẻ băng thông.

Kiểm định sạch, hàng tháng

Chúng tôi đo lại pool mỗi tháng và công bố số liệu, nhờ vậy chúng luôn được cập nhật.

Chất lượng bạn có thể tự kiểm chứng

Mỗi tháng, chúng tôi chạy cùng một bộ benchmark — một mẫu IP ngẫu nhiên được chấm điểm trên Scamalytics, kiểm tra độ trễ từ một VPS trung lập và đối chiếu độ chính xác địa lý trên nhiều cơ sở dữ liệu độc lập. Các con số dưới đây phản ánh môi trường thực tế, không phải kết quả phòng thí nghiệm trong điều kiện lý tưởng.

Độ sạch Scamalytics · 100.000 IP

Sạch tuyệt đối98.8%

Lẫn IP trung tâm dữ liệu0%

Có trong danh sách đen bất kỳ0.41%

Rủi ro gian lận thấp (0–19)99.5%

Độ chính xác địa lý99.82%

Độ trễ & tốc độ VPS → Cloudflare

Độ trễ trung vị535ms

Độ trễ p90 / p95718 / 1012 ms

Tỷ lệ kết nối thành công98.5%

Thông lượng (trung vị / p95)0.4 / 1.55 MB/s

Hoàn tất tải tệp lớn92%

Cấu thành mạng lưới

Loại kết nối100% dân cư

Mạng ISP (ASN)71

Chế độ xoay IPtheo từng yêu cầu · cố định

Nhà mạng hàng đầuSafaricom · Airtel Kenya · Wananchi Group (Zuku)

Đo trên 100.000 địa chỉ từ một VPS trung lập, đối chiếu với hạ tầng edge của Cloudflare, chính xác tại thời điểm benchmark gần nhất. Một lần chạy lại có thể chênh lệch đôi chút, nhưng các con số vẫn phản ánh đúng độ lớn thực tế và là cơ sở vững chắc để đánh giá liệu pool có phù hợp với nhu cầu của bạn hay không. Tỷ lệ scraping thành công trên từng mục tiêu không được công bố — điều đó phụ thuộc vào fingerprint và cấu hình của bạn, chứ không chỉ riêng IP.

benchmark.py — benchmark pool nguồn mở

Bộ benchmark mở đứng sau các con số phía trên. Nó lấy một mẫu IP ngẫu nhiên từ pool và đo bốn yếu tố — độ sạch (rủi ro gian lận Scamalytics, trung tâm dữ liệu, danh sách đen và độ chính xác địa lý), độ trễ (khứ hồi đến Cloudflare edge và ip-api), xoay IP (số IP, ASN và subnet /24 duy nhất) cùng thông lượng (tốc độ tải xuống và thời gian đến byte đầu tiên) — rồi báo cáo từng yếu tố. Mã nguồn được công bố đầy đủ để phương pháp luận đằng sau mỗi con số luôn rộng mở cho việc kiểm chứng.

# -*- 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.[/]")

Nhắm mục tiêu chi tiết đến từng thành phố

«Trung bình trực tuyến» là số lượng thiết bị riêng biệt trung bình đang kết nối và sẵn sàng tại cùng một thời điểm — không phải tổng tích lũy. Được phân nhóm thành Cao / Trung bình / Thấp.

VùngThành phốNhà cung cấp

Vùng	Độ trễ	Trung bình trực tuyến	Pool
Nairobi County	520 ms	~1.625	Cao
Nairobi	639 ms	~100	Cao
Nakuru County	489 ms	~64	Cao
Bungoma County	726 ms	~33	Cao
Kiambu County	530 ms	~31	Cao
Kajiado County	558 ms	~30	Cao
Mombasa County	555 ms	~23	Cao
Nyeri County	644 ms	~21	Cao
Machakos County	546 ms	~16	Trung bình
Marsabit County	579 ms	~9	Trung bình
Tharaka - Nithi	643 ms	~8	Trung bình
Meru County	505 ms	~7	Trung bình
Kericho County	586 ms	~7	Trung bình
Murang'A	547 ms	~6	Trung bình
Garissa County	588 ms	~6	Trung bình
Narok County	522 ms	~5	Thấp
Nyandarua County	444 ms	~5	Thấp
Uasin Gishu County	393 ms	~4	Thấp
Embu County	669 ms	~4	Thấp
Bomet County	577 ms	~4	Thấp
Kisumu County	941 ms	~4	Thấp
Laikipia	804 ms	~3	Thấp
Lusaka Province	887 ms	~2	Thấp
Nyamira county	498 ms	~2	Thấp

Thành phố	Độ trễ	Trung bình trực tuyến	Pool
Nairobi	529 ms	~1.723	Cao
Nakuru	487 ms	~54	Cao
Bungoma	726 ms	~34	Cao
Mombasa	555 ms	~23	Cao
Karatina	644 ms	~22	Cao
Ruiru	516 ms	~20	Cao
Kiserian	559 ms	~15	Cao
Imara Daima Estate	545 ms	~14	Cao
Ongata Rongai	558 ms	~11	Cao
Machakos	510 ms	~9	Trung bình
Westlands	400 ms	~9	Trung bình
Marsabit	579 ms	~9	Trung bình
Naivasha	476 ms	~8	Trung bình
Chogoria	643 ms	~8	Trung bình
Karuri	541 ms	~7	Trung bình
Meru	505 ms	~7	Trung bình
Murang'a	547 ms	~6	Thấp
Garissa	588 ms	~6	Thấp
Narok	522 ms	~5	Thấp
Kericho	580 ms	~5	Thấp
Ol Joro Orok	444 ms	~5	Thấp
Eldoret	393 ms	~4	Thấp
Ngong	554 ms	~4	Thấp
Mlolongo	550 ms	~4	Thấp
Embu	669 ms	~4	Thấp

Nhà cung cấp	Độ trễ	Trung bình trực tuyến	Pool
Safaricomdi động	487 ms	~697	Cao
Airtel Kenyadi động	620 ms	~236	Cao
Wananchi Group (Zuku)cố định	510 ms	~193	Cao
Vijiji Connectcố định	453 ms	~176	Cao
JTL (Faiba)cố định	374 ms	~171	Cao
Unwired Communicationscố định	568 ms	~94	Trung bình
Veenet Africacố định	501 ms	~80	Trung bình
Poa Internetcố định	585 ms	~74	Trung bình
Mymangacố định	642 ms	~56	Thấp
Hirani Telecommunicationscố định	611 ms	~56	Thấp
Savanna Fibrecố định	504 ms	~54	Thấp
Vos Africacố định	463 ms	~48	Thấp
Mawingucố định	726 ms	~46	Thấp
Surf Net Solutionscố định	556 ms	~34	Thấp

Độ chính xác địa lý — mọi IP bán dưới nhãn Kenya đều phân giải về Kenya trên ba cơ sở dữ liệu địa lý độc lập (db-ip, MaxMind, ipinfo); độ chính xác địa lý cao chính là tín hiệu mà các cổng kiểm tra địa lý nội địa dựa vào.

Proxy Kenya theo trường hợp sử dụng

Được sắp xếp theo công việc bạn đang làm, không phải theo nền tảng. Các dịch vụ nội địa được liệt kê ở đúng chỗ chúng phát huy tác dụng.

Thương mại điện tử & nghiên cứu thị trường

Kiểm tra gian hàng, mức giá và tin đăng nội địa từ ngay bên trong thị trường.

Giám sát gian hàng
Theo dõi giá nội địa
Nghiên cứu người bán

JumiaKilimallJijiCopiaMasoko

Xác minh & hiệu suất quảng cáo

Xem cách quảng cáo, mẫu sáng tạo và SERP hiển thị với người dùng nội địa — không cần chuyển đổi thủ công.

SERP nội địa
Kiểm tra mẫu sáng tạo
Trang đích

FacebookTikTokInstagramYouTubeX

Du lịch & phân tích giá

Thu thập giá vé máy bay và khách sạn đúng như cách một khách hàng nội địa nhìn thấy.

Giá theo khu vực
Giám sát ưu đãi
So sánh mức giá

Kenya AirwaysJambojetSafariBookingsBooking.comExpedia

Streaming & xác minh địa lý

Xác minh danh mục bị khóa theo khu vực và các cổng kiểm tra địa lý từ đúng vị trí.

Kiểm tra địa lý
Xác minh danh mục
Khả dụng theo khu vực

ShowmaxDStv StreamViusasaNetflix

Rao vặt & tin đăng nội địa

Giám sát hàng cũ, ô tô và bất động sản cho thuê trên các trang rao vặt nội địa.

Giám sát tin đăng
Theo dõi giá
Phủ sóng theo khu vực

JijiPigiaMeOLX KenyaChekiAutochek

Phát triển & tự động hóa

IP dân cư cho các pipeline, kiểm thử và thu thập dữ liệu hợp lệ qua API.

Tích hợp pipeline
Quản lý API
Phiên cố định

PythonScrapycURLGitHub

Ngoài các trang nội địa, cùng pool Kenya đó cũng hoạt động với các nền tảng toàn cầu lớn — Google, Meta, Amazon, YouTube, TikTok và Netflix — nhìn từ ngôn ngữ và khu vực Kenya: giá nội địa, quảng cáo, danh mục, SERP và nội dung.

Nhà phát triển API

Tạo các endpoint khoanh vùng theo Kenya trong bảng điều khiển, sao chép các đoạn mã đã kiểm thử, hoặc tự động hóa danh sách proxy qua API công khai. Thành phố, ISP, phiên cố định và các danh sách đã lưu đều nằm gọn trong một quy trình làm việc duy nhất.

Tài liệu API

1import requests
2 
3headers = {"x-api-key": "gpx_live_YOUR_KEY"}
4payload = {
5  "country_code": "KE",
6  "city": "Nairobi",
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"])

Các gói minh bạch, không điều khoản ẩn

Mua càng nhiều, giá mỗi gigabyte càng rẻ. Không thuê bao, không phí ẩn.

1 GB$3.70

$3.70 / GB

Chọn

5 GB$18.00

$3.60 / GB

Chọn

10 GB$34.50

$3.45 / GB

Chọn

Phổ biến

25 GB$80.00

$3.20 / GB

Chọn

50 GB$147.50

$2.95 / GB

Chọn

100 GB$270.00

$2.70 / GB

Chọn

250 GB$612.50

$2.45 / GB

Chọn

500 GB$1,100.00

$2.20 / GB

Chọn

1 TB$2,000.00

$2.00 / GB

Chọn

Gói

Giá mỗi GB

Tổng

1 GB

$3.70 / GB

$3.70Chọn

5 GB

$3.60 / GB

$18.00Chọn

10 GB

$3.45 / GB

$34.50Chọn

Phổ biến

25 GB

$3.20 / GB

$80.00Chọn

50 GB

$2.95 / GB

$147.50Chọn

100 GB

$2.70 / GB

$270.00Chọn

250 GB

$2.45 / GB

$612.50Chọn

500 GB

$2.20 / GB

$1,100.00Chọn

1 TB

$2.00 / GB

$2,000.00Chọn

Phương thức thanh toán khả dụng

Những câu hỏi mà người mua thực sự quan tâm

Vẫn đang cân nhắc liệu pool Kenya có phù hợp với nhu cầu của bạn? Đội ngũ của chúng tôi giải đáp những câu hỏi thực tế trên Telegram, suốt ngày đêm — cả trước và sau khi bạn mua.

Trò chuyện trên Telegram

Liệu chúng có hoạt động để thu thập dữ liệu Jumia và Kilimall không?

Đó chính là trường hợp sử dụng chính. Pool này là dân cư thực thụ — đặc điểm mà các hệ thống chống bot của sàn thương mại địa phương tìm kiếm — nên các IP này vượt qua được ở những nơi mà dải IP trung tâm dữ liệu bị chặn. Kết quả vẫn phụ thuộc vào dấu vân tay, header và nhịp độ yêu cầu của bạn, vì vậy hãy ghép proxy với một thiết lập sạch và múi giờ East Africa Time (EAT, UTC+3) tương ứng.

Tôi có thể nhắm tới một thành phố cụ thể như Nairobi hay Mombasa không?

Có, ở những nơi quy mô pool cho phép — chọn thành phố khi bạn tạo danh sách proxy. Các bảng vùng phủ sóng của chúng tôi hiển thị mức truy cập đồng thời theo thời gian thực cho Nairobi, Mombasa, Nakuru, Westlands, Ruiru và nhiều nơi khác, nhờ đó bạn có thể thấy thành phố nào có lượng xoay vòng dồi dào nhất trước khi quyết định.

Các IP có định vị địa lý bên trong Kenya không?

Có — pool phân giải về Kenya với độ chính xác địa lý 99.82% so với Cloudflare và trên các cơ sở dữ liệu địa lý độc lập trong lần benchmark gần nhất của chúng tôi, với các nút thoát được xác nhận tại Nairobi (NBO) và Mombasa (MBA). Độ chính xác địa lý cao là tín hiệu nền tảng mà các cổng chặn địa lý của Kenya dựa vào.

Tôi có thể giữ một IP Kenya cho một luồng nhiều bước không?

Có. Dùng phiên sticky để giữ nguyên một IP thoát Kenya — thời gian giữ trung bình khoảng 4.8 giờ và lên tới 72 giờ trong phép đo 3 ngày của chúng tôi. Để mở rộng quy mô, hãy xoay vòng theo từng yêu cầu: IP thoát thay đổi trong 96% số yêu cầu, thường là sang một nhà cung cấp khác, chứ không chỉ một địa chỉ lân cận.

Những nhà mạng và thành phố nào của Kenya được phủ?

Pool trải khắp 14 ISP Kenya, dẫn đầu là Safaricom, Airtel, Wananchi Group, Jamii Telecommunications và Poa Internet, qua 25 vùng và 25 thành phố — từ Nairobi, Mombasa và Nakuru đến Bungoma, Karatina, Ruiru, Machakos và Marsabit.