Benchmark
Proxy health-checking, latency measurement and benchmarking utilities.
All operations are fully async and parallelised with asyncio.gather.
Public API:
- :func:
ping_proxy— measure latency of a single proxy. - :func:
benchmark_proxies— benchmark a batch in parallel. - :func:
select_best_proxies— keep the fastest N proxies. - :func:
find_best_proxies— create, test, keep best, delete the rest.
CountryPingResult
Section titled “CountryPingResult”Aggregated ping statistics for a country/region.
FindBestResult
Section titled “FindBestResult”Result of the :func:find_best_proxies pipeline.
Attributes: best: Top proxies that passed the benchmark. total_created: Number of ports created. total_tested: Number of proxies tested. total_alive: Number of alive proxies. total_deleted: Number of discarded ports deleted. delete_errors: Port IDs that failed to delete. avg_latency_ms: Average latency of the best proxies. min_latency_ms: Best (lowest) latency. max_latency_ms: Worst latency among the best.
ProxyBenchmarkResult
Section titled “ProxyBenchmarkResult”Result of benchmarking a single proxy.
Attributes:
proxy_url: Full proxy URL.
port_id: ASocks port ID.
latency_ms: Round-trip latency in milliseconds (None on failure).
external_ip: External IP seen through the proxy.
is_alive: Whether the proxy responded successfully.
country: Country name from port metadata.
city: City name from port metadata.
error: Error message on failure.
Functions
Section titled “Functions”benchmark_proxies(proxy_urls: list[str], *, timeout: float = 5.0, concurrency: int = 200) -> list[asockslib.benchmark.ProxyBenchmarkResult]
Section titled “benchmark_proxies(proxy_urls: list[str], *, timeout: float = 5.0, concurrency: int = 200) -> list[asockslib.benchmark.ProxyBenchmarkResult]”Benchmark a batch of proxies in parallel.
Returns: Results sorted alive-first by ascending latency.
find_best_proxies(client: asockslib.client.ASocksClient, *, country_code: str = 'US', city: str = '', state: str = '', total: int = 100, keep: int = 10, batch_size: int = 1000, timeout: float = 5.0, concurrency: int = 200, type_id: int = 1, proxy_type_id: int = 1, server_port_type_id: int = 0, ttl: int = 1, traffic_limit: int = 1, delete_failures: bool = True, progress_callback: collections.abc.Callable[[str, int, int], None] | None = None) -> asockslib.benchmark.FindBestResult
Section titled “find_best_proxies(client: asockslib.client.ASocksClient, *, country_code: str = 'US', city: str = '', state: str = '', total: int = 100, keep: int = 10, batch_size: int = 1000, timeout: float = 5.0, concurrency: int = 200, type_id: int = 1, proxy_type_id: int = 1, server_port_type_id: int = 0, ttl: int = 1, traffic_limit: int = 1, delete_failures: bool = True, progress_callback: collections.abc.Callable[[str, int, int], None] | None = None) -> asockslib.benchmark.FindBestResult”Create total proxies, benchmark them all, keep the fastest keep, delete the rest.
The entire pipeline is parallelised: creation in batches, benchmark with up to concurrency simultaneous pings, deletion in parallel.
Args:
client: :class:ASocksClient instance.
country_code: ISO country code.
total: How many proxies to create.
keep: How many fastest proxies to keep.
timeout: Benchmark timeout per proxy.
concurrency: Maximum concurrent benchmark connections.
delete_failures: Delete discarded ports (default True).
progress_callback: (stage, current, total) -> None.
ping_proxy(proxy_url: str, *, timeout: float = 5.0, ping_urls: list[str] | None = None) -> asockslib.benchmark.ProxyBenchmarkResult
Section titled “ping_proxy(proxy_url: str, *, timeout: float = 5.0, ping_urls: list[str] | None = None) -> asockslib.benchmark.ProxyBenchmarkResult”Measure latency of a single proxy.
Sends requests to multiple lightweight endpoints in parallel and keeps the fastest successful result.
Args: proxy_url: Full proxy URL. timeout: Request timeout in seconds. ping_urls: Custom list of check URLs.
select_best_proxies(proxy_urls: list[str], *, keep: int = 10, timeout: float = 5.0, concurrency: int = 200) -> tuple[list[asockslib.benchmark.ProxyBenchmarkResult], list[asockslib.benchmark.ProxyBenchmarkResult]]
Section titled “select_best_proxies(proxy_urls: list[str], *, keep: int = 10, timeout: float = 5.0, concurrency: int = 200) -> tuple[list[asockslib.benchmark.ProxyBenchmarkResult], list[asockslib.benchmark.ProxyBenchmarkResult]]”Select the fastest proxies by latency.
Returns:
(best, discarded) tuple.