Đức

IP nhà dân thật

Proxy dân cư Đức

Proxy dân cư Đức là những địa chỉ IP thật từ các hộ gia đình Đức, được cấp bởi các ISP tiêu dùng như Deutsche Telekom, Vodafone, O2 (Telefónica) và NetCologne. Chúng cho phép bạn duyệt web, thu thập dữ liệu và kiểm chứng các cửa hàng trực tuyến, giá cả và nội dung của Đức với tư cách một người dùng thực thụ trong nước — trải khắp Berlin, Munich, Hamburg và Frankfurt am Main — thay vì từ một IP trung tâm dữ liệu mà Otto, Zalando cùng các rào chắn địa lý nội địa nhận diện và chặn ngay lập tức.

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

212.228IP trong pool

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

107Mạng ISP

368 msĐộ trễ trung vị

99.5%Sạch tuyệt đối

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

Proxy dân cư Đức là địa chỉ IP thật từ các hộ gia đình Đức, 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 Đức 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: Berlin, Stuttgart, Frankfurt am Main
ISP hàng đầu: Vodafone · Deutsche Telekom · O2 (Telefónica)
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.98%
Độ sạch của pool: 99.5% 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 Đức 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 Đức — 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 đối99.5%

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

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

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

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

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

Độ trễ trung vị368ms

Độ trễ p90 / p95415 / 468 ms

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

Thông lượng (trung vị / p95)0.68 / 2.34 MB/s

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

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

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

Mạng ISP (ASN)107

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

Nhà mạng hàng đầuVodafone · Deutsche Telekom · O2 (Telefónica)

Đ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
North Rhine-Westphalia	366 ms	~4.259	Cao
Baden-Wurttemberg	367 ms	~3.108	Cao
Bavaria	368 ms	~2.340	Cao
Hesse	362 ms	~2.311	Cao
Lower Saxony	369 ms	~1.466	Cao
State of Berlin	368 ms	~1.071	Cao
Rheinland-Pfalz	367 ms	~896	Trung bình
Free and Hanseatic City of Hamburg	360 ms	~822	Trung bình
Saxony	377 ms	~603	Trung bình
Schleswig-Holstein	368 ms	~408	Trung bình
Saxony-Anhalt	384 ms	~351	Trung bình
Brandenburg	381 ms	~245	Thấp
Saarland	371 ms	~223	Thấp
Mecklenburg-Vorpommern	379 ms	~220	Thấp
Thuringia	381 ms	~220	Thấp
Bremen	373 ms	~201	Thấp
Land Berlin	404 ms	~134	Thấp

Thành phố	Độ trễ	Trung bình trực tuyến	Pool
Berlin	370 ms	~2.448	Cao
Stuttgart	370 ms	~2.299	Cao
Frankfurt am Main	361 ms	~2.047	Cao
Hamburg	360 ms	~1.671	Cao
Munich	365 ms	~1.503	Cao
Essen	364 ms	~978	Cao
Cologne	350 ms	~946	Cao
Hagen	371 ms	~674	Cao
Freiburg im Breisgau	372 ms	~570	Cao
Leipzig	374 ms	~531	Trung bình
Nuremberg	371 ms	~479	Trung bình
Eschborn	360 ms	~473	Trung bình
Düsseldorf	362 ms	~460	Trung bình
Hanover	371 ms	~453	Trung bình
Mannheim	356 ms	~447	Trung bình
Hanau	362 ms	~434	Trung bình
Bremen	378 ms	~350	Thấp
Recklinghausen	369 ms	~330	Thấp
Dortmund	365 ms	~324	Thấp
Willich	361 ms	~317	Thấp
Karlsruhe	367 ms	~272	Thấp
Augsburg	358 ms	~253	Thấp
Bochum	373 ms	~227	Thấp
Dresden	379 ms	~201	Thấp
Halle	388 ms	~194	Thấp

Nhà cung cấp	Độ trễ	Trung bình trực tuyến	Pool
Vodafonedi động	367 ms	~7.864	Cao
Deutsche Telekomcố định	371 ms	~6.723	Cao
O2 (Telefónica)cố định	360 ms	~2.082	Cao
Osnatelcố định	361 ms	~412	Cao
PŸURcố định	380 ms	~405	Cao
NetColognecố định	344 ms	~360	Trung bình
M-netcố định	359 ms	~246	Trung bình
wilhelm.telcố định	345 ms	~242	Trung bình
PURtelcố định	358 ms	~152	Trung bình
inexiocố định	398 ms	~114	Thấp
DNS:NETcố định	368 ms	~86	Thấp
Deutsche Glasfasercố định	367 ms	~86	Thấp
TNG Stadtnetzcố định	371 ms	~55	Thấp
GoeTelcố định	335 ms	~52	Thấp

Độ chính xác địa lý — mọi IP bán dưới nhãn Đức đều phân giải về Đức 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 Đức 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

Amazon.deOttoZalandoMediaMarktSaturn

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

FacebookInstagramTikTokYouTubeX/Twitter

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á

Booking.comCheck24LufthansaEurowingsDeutsche Bahn

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

RTL+JoynWOWMagenta TVZDFmediathek

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

Kleinanzeigenmobile.deAutoScout24Vintedmarkt.de

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 Đức đó 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 Đức: 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 Đức 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": "DE",
6  "city": "Berlin",
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 Đức 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ó dùng được để thu thập dữ liệu Otto và Zalando không?

Đó chính là trường hợp sử dụng chủ đạo. Pool là dân cư thật — đúng dấu hiệu mà các hệ thống chống bot của chợ trực tuyến nội địa tìm kiếm — nên những IP này lọt qua được ở 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 fingerprint, header và nhịp độ gửi 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ờ Đức tương ứng (CET/CEST).

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

Có, ở những nơi độ sâu của pool cho phép — hãy chọn thành phố khi bạn tạo danh sách proxy. Các bảng vùng phủ của chúng tôi hiển thị mức đồng thời theo thời gian thực cho Berlin, Munich, Hamburg, Frankfurt am Main, Cologne, Stuttgart và nhiều nơi khác, để bạn thấy được thành phố nào có lượng IP 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 nước Đức không?

Có — pool phân giải về Đức với độ chính xác địa lý 99,98% khi đối chiếu 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 Frankfurt (FRA), Berlin (TXL) và Düsseldorf (DUS). Độ chính xác địa lý cao chính là tín hiệu nền tảng mà các rào chắn địa lý của Đức dựa vào.

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

Có. Hãy dùng phiên sticky để giữ nguyên một IP thoát Đức — thời gian giữ trung bình khoảng 4–5 giờ và lên tới 72 giờ trong đợt đo 3 ngày của chúng tôi. Để mở rộng quy mô, hãy xoay theo từng yêu cầu: IP thoát thay đổi ở khoảng 97% 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ỉ kề bên.

Những nhà mạng và thành phố Đức nào được phủ sóng?

Pool trải rộng trên 14 ISP Đức dẫn đầu bởi Deutsche Telekom, Vodafone, O2 (Telefónica), NetCologne, M-net và Tele Columbus, qua 17 vùng và 25 thành phố — từ Berlin, Munich và Hamburg đến Frankfurt am Main, Cologne, Stuttgart, Leipzig và Nuremberg.