هونغ كونغ الصينية (منطقة إدارية خاصة)

عناوين IP منزلية حقيقية

بروكسيات هونغ كونغ الصينية (منطقة إدارية خاصة) السكنية

البروكسيات السكنية في هونغ كونغ هي عناوين IP حقيقية لمنازل في هونغ كونغ، تُسندها شركات الإنترنت الاستهلاكية مثل HKT وHong Kong Broadband (HKBN) وHGC وSmarTone. وهي تتيح لك التصفح وكشط البيانات والتحقق من المتاجر والأسعار والمحتوى المحلي بصفتك مستخدمًا حقيقيًا داخل المدينة — عبر Central وCauseway Bay وShatin وTsuen Wan — بدلًا من عنوان IP تابع لمركز بيانات تكشفه وتحظره منصات مثل HKTVmall وFortress والبوابات الجغرافية المحلية فور رؤيته.

جرِّب مقابل $1.99/GB تسجيل الدخول عبر Google

HTTP / SOCKS5استهداف حسب المدينة ومزوّد ISPحركة المرور لا تنتهي صلاحيتها أبدًاتحقُّق شهري من النظافة

36,029عناوين IP في المجموعة

25+المدن المغطّاة

22شبكات ISP

150 msزمن الاستجابة الوسيط

99.3%نظيفة تمامًا

بروكسيات هونغ كونغ الصينية (منطقة إدارية خاصة) السكنية في لمحة

بروكسيات هونغ كونغ الصينية (منطقة إدارية خاصة) السكنية هي عناوين IP حقيقية من منازل هونغ كونغ الصينية (منطقة إدارية خاصة)، مُخصَّصة من مزوّدي خدمة الإنترنت للمستهلكين. تتيح لك تصفّح مواقع هونغ كونغ الصينية (منطقة إدارية خاصة) واستخراج بياناتها والتحقّق من أسعارها ومحتواها كمستخدم محلي حقيقي داخل البلد — بدلًا من عنوان IP تابع لمركز بيانات تكتشفه المواقع المحلية وتحظره.

نوع المجموعة: سكنية، متعددة مزوّدي ISP
المدن المغطّاة: Central, Kwai Chung, Tsing Yi
أبرز مزوّدي ISP: HKT · HKBN · HGC
البروتوكولات: HTTP، SOCKS5
التدوير: لكل طلب أو ثابت
الدقة الجغرافية: 99.87%
نظافة المجموعة: 99.3% clean
التسعير: from $2.00/GB
آخر قياس: 2026-06-23
الدعم: 24/7

لماذا تختار الفِرق مجموعتنا السكنية في هونغ كونغ الصينية (منطقة إدارية خاصة)

عناوين IP سكنية حقيقية، واستهداف دقيق، وجودة لا يلزمك تصديقها على عَواهنها.

عناوين IP سكنية حقيقية

اتصالات منزلية أصيلة من مزوّدي خدمة الإنترنت للمستهلكين في هونغ كونغ الصينية (منطقة إدارية خاصة) — وليست نطاقات مراكز بيانات تُحظر فور رصدها.

استهداف حسب المدينة ومزوّد ISP

استهدِف مدنًا ومشغّلين بعينهم حيثما يسمح عمق المجموعة — وصولًا إلى مستوى المزوّد.

متدوِّرة أو ثابتة

تدوير لكل طلب لتوسيع النطاق، أو جلسات ثابتة حتى نحو 72 ساعة للتدفّقات متعددة الخطوات.

HTTP و SOCKS5

تعمل مع أي عميل أو أداة استخراج أو منظومة أتمتة فور الاستخدام ودون إعداد إضافي.

مصدر أخلاقي

عناوين IP من شبكة سكنية قائمة على الاشتراك الطوعي — مستخدمون حقيقيون وافقوا على مشاركة عرض النطاق الترددي.

تحقُّق شهري من النظافة

نعيد قياس المجموعة كل شهر وننشر الأرقام، لتبقى محدَّثة دائمًا.

جودة يمكنك التحقّق منها

كل شهر نُجري الاختبار المرجعي نفسه — عيّنة عشوائية من عناوين IP تُقيَّم مقابل Scamalytics، ويُختبر زمن الاستجابة من VPS محايد، وتُفحص الدقة الجغرافية عبر قواعد بيانات مستقلة. تعكس الأرقام أدناه بيئة الإنتاج، لا تشغيلًا مخبريًا في أفضل الظروف.

النظافة Scamalytics · 100,000 عنوان IP

نظيفة تمامًا99.3%

تلوُّث بمراكز البيانات0%

على أي قائمة سوداء0.19%

مخاطر احتيال منخفضة (0–19)99.2%

الدقة الجغرافية99.87%

زمن الاستجابة والسرعة VPS ← Cloudflare

زمن الاستجابة الوسيط150ms

زمن الاستجابة p90 / p95194 / 245 ms

نجاح الاتصال98.8%

الإنتاجية (الوسيط / p95)2.59 / 7.22 MB/s

إتمام تنزيل الملفات الكبيرة100%

تركيبة الشبكة

نوع الاتصالسكني 100%

شبكات ISP (ASN)22

أوضاع التدويرلكل طلب · ثابت

أبرز المشغّلينHKT · HKBN · HGC

جرى القياس على 100,000 عنوان من VPS محايد مقابل حافة شبكة Cloudflare، وهي دقيقة حتى آخر اختبار مرجعي. قد يختلف أي تشغيل مكرَّر بهامش بسيط، لكن الأرقام تعكس رتبة المقدار تعكيسًا كاملًا وتُعدّ أساسًا سليمًا للحكم على مدى ملاءمة المجموعة لحالة استخدامك. لا يُنشر معدّل نجاح الاستخراج لكل هدف على حدة — فهو يعتمد على بصمتك وإعداداتك، لا على عنوان IP وحده.

benchmark.py — اختبار مرجعي مفتوح للمجموعة

الاختبار المرجعي المفتوح الكامن خلف الأرقام أعلاه. يأخذ عيّنة عشوائية من عناوين IP في المجموعة ويقيس أربعة أشياء — النظافة (مخاطر الاحتيال وفق Scamalytics، ومراكز البيانات، والقائمة السوداء، والدقة الجغرافية)، وزمن الاستجابة (الذهاب والإياب إلى حافة 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.[/]")

استهدِف وصولًا إلى مستوى المدينة

«متوسط المتصلين» هو متوسط عدد الأجهزة المتمايزة المتصلة والمتاحة في اللحظة نفسها — وليس إجماليًا تراكميًا. وهو مُصنَّف إلى مرتفع / متوسط / منخفض.

المناطقالمدنالمزوّدون

المنطقة	زمن الاستجابة	متوسط المتصلين	المجموعة
Kwai Tsing District	149 ms	~710	مرتفع
Central and Western	150 ms	~648	مرتفع
Kowloon City	149 ms	~203	مرتفع
Yuen Long District	151 ms	~182	مرتفع
Sha Tin	150 ms	~171	مرتفع
Eastern District	149 ms	~163	مرتفع
Kwun Tong District	151 ms	~139	مرتفع
Yau Tsim Mong District	153 ms	~129	متوسط
Tsuen Wan District	151 ms	~128	متوسط
North District	148 ms	~119	متوسط
Sham Shui Po District	149 ms	~101	متوسط
Wong Tai Sin District	150 ms	~94	متوسط
Sai Kung District	153 ms	~89	منخفض
Tai Po District	151 ms	~76	منخفض
Kowloon	153 ms	~73	منخفض
Tuen Mun	151 ms	~65	منخفض
Islands	154 ms	~45	منخفض
Southern District	151 ms	~42	منخفض
Wan Chai	154 ms	~27	منخفض

المدينة	زمن الاستجابة	متوسط المتصلين	المجموعة
Central	150 ms	~839	مرتفع
Kwai Chung	149 ms	~778	مرتفع
Tsing Yi	150 ms	~199	مرتفع
Hong Kong	153 ms	~136	مرتفع
Kwu Tung	150 ms	~126	مرتفع
Tin Shui Wai	151 ms	~93	مرتفع
Causeway Bay	148 ms	~79	مرتفع
Ho Man Tin	147 ms	~76	مرتفع
Tsuen Wan	153 ms	~69	مرتفع
Shatin	149 ms	~67	متوسط
Tai Po	151 ms	~64	متوسط
Sham Shui Po	149 ms	~64	متوسط
Lam Tin	185 ms	~61	متوسط
Tseung Kwan O	149 ms	~56	متوسط
Hang Hau	155 ms	~56	متوسط
Tai Kok Tsui	146 ms	~51	متوسط
Mong Kok	161 ms	~49	منخفض
Kowloon Bay	149 ms	~45	منخفض
Yuen Long	146 ms	~45	منخفض
Chai Wan Kok	152 ms	~44	منخفض
Tung Chung	154 ms	~44	منخفض
Hung Hom	153 ms	~42	منخفض
Tuenmen	147 ms	~42	منخفض
Kam Tin	180 ms	~40	منخفض
Jordan	153 ms	~39	منخفض

المزوّد	زمن الاستجابة	متوسط المتصلين	المجموعة
HKTثابت	150 ms	~1,433	مرتفع
HKBNثابت	137 ms	~1,075	مرتفع
HGCثابت	150 ms	~349	مرتفع
CommuniLinkمحمول	152 ms	~137	متوسط
SmarToneمحمول	201 ms	~110	منخفض
HKCثابت	153 ms	~96	منخفض
NOVAثابت	148 ms	~4	منخفض

الدقة الجغرافية — كل عنوان IP يُباع على أنه من هونغ كونغ الصينية (منطقة إدارية خاصة) يُحدَّد جغرافيًا في هونغ كونغ الصينية (منطقة إدارية خاصة) عبر ثلاث قواعد بيانات جغرافية مستقلة (db-ip و MaxMind و ipinfo)؛ والدقة الجغرافية العالية هي الإشارة التي تعتمد عليها بوابات التحقّق الجغرافي المحلية.

بروكسيات هونغ كونغ الصينية (منطقة إدارية خاصة) حسب حالة الاستخدام

مُنظَّمة حسب ما تفعله، لا حسب المنصّة. وتُدرَج الخدمات المحلية حيث يكون لها أهمية.

التجارة الإلكترونية وأبحاث السوق

افحص الواجهات والأسعار والقوائم المحلية من داخل السوق.

مراقبة الواجهات
تتبُّع الأسعار المحلية
أبحاث البائعين

HKTVmallYOHO (友和)Fortress (豐澤)Broadway (百老滙)ParknShop (百佳)

التحقّق من الإعلانات والأداء

شاهِد كيف تظهر الإعلانات والتصاميم ونتائج SERP لمستخدم محلي — دون تبديل يدوي.

نتائج SERP المحلية
فحوص التصاميم
صفحات الهبوط

WhatsAppFacebookInstagramWeChat

السفر وذكاء الأسعار

التقِط أسعار الرحلات والفنادق كما يراها العميل المحلي.

أسعار محدَّدة جغرافيًا
مراقبة العروض
مقارنة الأسعار

Trip.comKlookKKdayCathay PacificHK Express

البث والتحقّق الجغرافي

تحقّق من الكتالوجات المقيَّدة إقليميًا والبوابات الجغرافية من الموقع الصحيح.

الفحوص الجغرافية
التحقّق من الكتالوجات
التوافر الإقليمي

myTV SUPER (TVB)ViuNow E (PCCW)HBO GO AsiaNetflix

الإعلانات المبوَّبة والقوائم المحلية

راقِب السلع المستعملة والسيارات والإيجارات عبر الإعلانات المبوَّبة المحلية.

مراقبة القوائم
تتبُّع الأسعار
التغطية الإقليمية

CarousellDCFever28car (易發車網)Geo Expat

التطوير والأتمتة

عناوين IP سكنية لخطوط المعالجة والاختبارات وجمع البيانات المسموح به عبر الـ API.

تكامل خطوط المعالجة
إدارة الـ API
الجلسات الثابتة

PythonScrapycURLGitHub

إلى جانب المواقع المحلية، تعمل مجموعة هونغ كونغ الصينية (منطقة إدارية خاصة) نفسها مع كبرى المنصّات العالمية — Google و Meta و Amazon و YouTube و TikTok و Netflix — كما تظهر من النطاق المحلي لـ هونغ كونغ الصينية (منطقة إدارية خاصة): الأسعار والإعلانات والكتالوجات ونتائج SERP والمحتوى المحلية.

المطوّر API

أنشئ نقاط اتصال محدَّدة النطاق لـ هونغ كونغ الصينية (منطقة إدارية خاصة) من لوحة التحكم، وانسخ المقتطفات المختبَرة، أو أتمِت قوائم البروكسي عبر الـ API العام. تبقى المدينة ومزوّد ISP والجلسات الثابتة والقوائم المحفوظة ضمن سير عمل واحد نظيف.

وثائق الـ API

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

اختيار

الباقة

السعر لكل ⁦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

هل ستعمل هذه البروكسيات لكشط البيانات من HKTVmall وFortress؟

هذه هي حالة الاستخدام الأساسية. المجمّع سكني حقيقي — وهو ما تبحث عنه أنظمة مكافحة الروبوتات في المتاجر المحلية تحديدًا — لذا تمرّ عناوين IP هذه حيث تُحظر نطاقات مراكز البيانات. وتظل النتائج متوقفة على بصمتك وترويساتك وإيقاع طلباتك، لذا اقرن البروكسي بإعداد نظيف وبمنطقة زمنية مطابقة لهونغ كونغ (HKT، UTC+8).

هل يمكنني استهداف مدينة محددة مثل Central أو Causeway Bay؟

نعم، بحسب ما يسمح به عمق المجمّع — اختر المدينة عند توليد قائمة البروكسي. تعرض جداول التغطية لدينا التزامن اللحظي لكل من Central وKwai Chung وTsing Yi وCauseway Bay وShatin وTsuen Wan وغيرها، حتى ترى أي المدن يتوفر فيها أعمق تدوير قبل أن تلتزم.

هل تتموضع عناوين IP جغرافيًا داخل هونغ كونغ؟

نعم — يتموضع المجمّع في هونغ كونغ بدقة جغرافية 99.87% مقابل Cloudflare وعبر قواعد بيانات جغرافية مستقلة في آخر قياس مرجعي لنا، مع تأكيد عُقد الخروج عند مركز التوجيه المشترك (colo) في هونغ كونغ (HKG). والدقة الجغرافية العالية هي الإشارة الأساسية التي تعتمد عليها البوابات الجغرافية في هونغ كونغ.

هل يمكنني الاحتفاظ بعنوان IP واحد من هونغ كونغ لتدفق متعدد الخطوات؟

نعم. استخدم جلسة ثابتة للإبقاء على عنوان IP الخروج نفسه من هونغ كونغ — متوسط مدة الاحتفاظ نحو 5 ساعات وتصل إلى 72 ساعة في قياسنا على مدى 3 أيام. وعلى نطاق واسع، بدّل العنوان مع كل طلب: يتغير عنوان IP الخروج في نحو 98% من الطلبات، وعادةً ينتقل إلى مزوّد مختلف، لا إلى عنوان مجاور فحسب.

ما شركات الاتصالات والمدن المغطاة في هونغ كونغ؟

يمتد المجمّع عبر 9 شركات إنترنت في هونغ كونغ تتصدرها HKT وHong Kong Broadband (HKBN) وHGC وSmarTone وCommuniLink، عبر 19 منطقة و25 مدينة — من جزيرة هونغ كونغ (Central وCauseway Bay) إلى كولون (Sham Shui Po وHo Man Tin) والأقاليم الجديدة (Shatin وTsuen Wan وTai Po وKwai Chung).