200 行 Python 实现代理 IP 质量评分系统（附完整代码）

原创

三三有猫

发布于 2026-07-17 15:55:53

很多人判断代理 IP 好不好用只看一件事——能不能连上。但能连上的 IP 里，相当一部分在请求头里回传了真实来源 IP，或者延迟高到拖垮批量采集。

IP 质量应该是连通性、响应延迟、来源可辨识度、稳定性四个维度加权的结果。

本文用 200 行 Python 把它量化成一个 0–100 的分数，一批 IP 跑一遍自动排序，低质量的直接沉底，比人工挨个试靠谱得多。

先看运行效果

把代理列表喂进脚本，跑完直接输出一份按总分降序的排名，死 IP 自动标“已剔除”：

http://12.34.56.78:8080      总分 92.3  延迟0.187s  匿名high        稳定1.0
http://23.45.67.89:8080      总分 78.5  延迟0.642s  匿名anonymous   稳定0.8
http://34.56.78.90:3128      总分 41.0  延迟2.310s  匿名transparent 稳定0.6
http://45.67.89.01:8888      不可用，已剔除

一眼就能看出该留谁、该扔谁。下面拆开讲每一维怎么测、分怎么算。

整体思路：四个维度，一条评分管线

连不通连通代理 IP 列表连通性 & 延迟测试判 0 分 · 直接剔除匿名度检测稳定性测试四维归一化 · 加权合成总分按总分降序排名输出

连通性是第一道闸——连不上的 IP 直接判 0，后面三项都不用测了，既省时间也避免对死 IP 发无意义的请求。

这套脚本解决什么问题，跑之前要准备什么

它解决的是“买了一堆代理 IP，肉眼分不清哪些能用”的问题——把主观感受换成一个可排序的分数。

前置条件很轻，Python 3.8+、装个 requests 就够，不依赖任何重型框架。核心要准备两样东西：一个待测的代理 IP 列表，一个能回显请求头和来源 IP 的接口。回显接口是整套评分的地基，因为判断来源是否泄露、延迟多少，都要靠一个“诚实告诉你它收到了什么”的服务端。

清单：

Python 3.8+ 与 requests：pip install requests
一批待测代理，格式统一为 http://ip:port 或 http://user:pass@ip:port
一个回显接口，返回 JSON 里带 origin（来源 IP）和 headers（请求头）。公开的 httpbin /get 就是这种结构；生产环境建议自建一个同结构接口，避免被限流
一个稳定的连通性测试目标 URL，选一个大站首页即可

手头暂时没有可测的代理？很多服务商提供限时免费测试，可以先领一批喂进脚本做首轮筛选——先看分数再决定要不要采购，比盲买套餐再后悔省事。（例如极安代理官网标注新注册可领 8 小时免费测试，仅作为“哪里搞测试样本”的思路举例，评分逻辑对任何厂商的代理都通用。）

该量化哪四个维度

单看任何一个维度都会误判：延迟极低但请求头把真实来源 IP 抖出去了，采集照样触发风控；来源藏得干净但时通时断，批量任务也跑不完。所以拆成四项，权重按业务可调：

维度	权重	好的信号	差的信号
连通性	30%	能稳定返回 200	超时、连接被拒
响应延迟	25%	平均 <0.5 秒	>3 秒或抖动大
来源可辨识度	25%	头信息不回传真实来源 IP	X-Forwarded-For 带真实来源 IP
稳定性	20%	连续请求成功率高	断断续续

延迟这一维没有绝对及格线，参照厂商公开口径来校准满分阈值更实际。比如有厂商官方口径给到平均响应＜0.1 秒、默认 5M 带宽，那么把 0.5 秒设为满分、5 秒记零分的线性坡度，就是照着“优质代理该有的响应水平”定的，而不是拍脑袋。

一、连通性 + 响应延迟

这两项一次测完最省事：连着发几次请求，能返回 200 就算连通，每次耗时取平均就是延迟。采样多轮而不是只发一次，是为了避开单次网络抖动带来的误差。

import time
import requests

TEST_URL = "https://www.baidu.com"   # 连通性/延迟测试目标
TIMEOUT = 8
LATENCY_ROUNDS = 3                   # 延迟采样次数

def test_latency(proxy):
    """返回 (是否连通, 平均延迟秒)；连不通返回 (False, None)。"""
    proxies = {"http": proxy, "https": proxy}
    costs = []
    for _ in range(LATENCY_ROUNDS):
        start = time.time()
        try:
            r = requests.get(TEST_URL, proxies=proxies, timeout=TIMEOUT)
            if r.status_code == 200:
                costs.append(time.time() - start)
        except Exception:
            continue
    if not costs:
        return False, None
    return True, sum(costs) / len(costs)

自测方法：拿一个已知能用的代理和一个瞎编的假地址各跑一次。真代理应返回 (True, 某个小数)，假地址应返回 (False, None)。两种都对，说明连通判定和异常捕获都没漏。

二、匿名度：怎么判断一个代理是不是真高匿

关键是拿代理去请求回显接口，看它回传的头信息里有没有把真实来源 IP 抖出去。代理按来源可辨识度分三档，判定完全靠两个头字段——X-Forwarded-For 和 Via：

检测项	高匿 (high)	普通匿名 (anonymous)	透明 (transparent)
返回的 origin	代理 IP	代理 IP	含真实来源 IP
X-Forwarded-For	无	代理 IP	真实来源 IP
Via	无	有	有

检测逻辑：先在不走代理的情况下拿到本机真实来源 IP，再让代理去请求回显接口，比对返回里有没有这个真实 IP，以及 Via、X-Forwarded-For 是否存在。

ECHO_URL = "https://httpbin.org/get"   # 回显请求头与来源 IP

def get_real_ip():
    try:
        r = requests.get(ECHO_URL, timeout=TIMEOUT)
        return r.json().get("origin", "").split(",")[0].strip()
    except Exception:
        return ""

def test_anonymity(proxy, real_ip):
    """返回 high / anonymous / transparent / unknown。"""
    proxies = {"http": proxy, "https": proxy}
    try:
        r = requests.get(ECHO_URL, proxies=proxies, timeout=TIMEOUT)
        data = r.json()
        headers = {k.lower(): v for k, v in data.get("headers", {}).items()}
        origin = data.get("origin", "")
    except Exception:
        return "unknown"
    xff = headers.get("x-forwarded-for", "")
    if real_ip and (real_ip in origin or real_ip in xff):
        return "transparent"
    if headers.get("via") or xff:
        return "anonymous"
    return "high"

自测方法：先单独跑 get_real_ip() 确认能拿到本机公网 IP。如果拿到的是空值或内网地址，说明本机自己挂了系统代理，要先关掉再取——否则后面所有匿名度判定都会偏。这是最容易踩的坑，务必先确认。

三、稳定性：连续打才暴露得出来

稳定性就是连续发 N 次请求，数成功几次，成功率就是这一维的原始分。一次成功不代表稳定，代理池里最坑的是那种“测的时候好好的、批量跑到一半就掉”的 IP，只有连续打才现形。

STABILITY_ROUNDS = 5   # 稳定性连续请求次数

def test_stability(proxy):
    """连续请求，返回成功率 0-1。"""
    proxies = {"http": proxy, "https": proxy}
    ok = 0
    for _ in range(STABILITY_ROUNDS):
        try:
            r = requests.get(TEST_URL, proxies=proxies, timeout=TIMEOUT)
            if r.status_code == 200:
                ok += 1
        except Exception:
            pass
    return ok / STABILITY_ROUNDS

稳定性本质是可用率的短周期采样。优质代理的连续成功率应该贴近满分，脚本里跑出 0.6、0.8 这种成绩的，基本可以判定是二手回收池而不是纯净资源。轮数越多越准，代价是耗时——日常筛选 5 次够用，进重点池前调到 10 次以上复筛。

自测方法：对同一个代理连跑两遍 test_stability，成绩应在小范围内浮动；如果两次差距很大（比如 1.0 和 0.2），说明这个 IP 本身就不稳，正好是要被扣分剔除的对象。

四、四维合成总分

把四个原始分各自归一到 0–100，再按权重加权求和，就是最终总分。归一化是为了让延迟的“秒”和稳定性的“成功率”能放在同一把尺子上比：延迟按 0.5 秒满分、5 秒零分线性衰减；匿名度按档位查表给分；连通和稳定直接乘 100。

WEIGHTS = {"conn": 0.30, "latency": 0.25, "anon": 0.25, "stability": 0.20}
ANON_SCORE = {"high": 100, "anonymous": 60, "transparent": 10, "unknown": 0}

def score_proxy(proxy, real_ip):
    alive, avg_latency = test_latency(proxy)
    if not alive:
        return {"proxy": proxy, "alive": False, "total": 0}
    latency_score = max(0, min(100, (5 - avg_latency) / 4.5 * 100))
    anon = test_anonymity(proxy, real_ip)
    stability = test_stability(proxy)
    total = (100 * WEIGHTS["conn"]
             + latency_score * WEIGHTS["latency"]
             + ANON_SCORE[anon] * WEIGHTS["anon"]
             + stability * 100 * WEIGHTS["stability"])
    return {"proxy": proxy, "alive": True, "latency": round(avg_latency, 3),
            "anon": anon, "stability": round(stability, 2), "total": round(total, 1)}

权重不是死的，跟着业务改才对：反爬严的目标站把 anon 调高，高频持续采集把 stability 调高，四个权重之和保持为 1 即可。

自测方法：手动构造一个全满分的假数据代入公式，总分应等于 100；把 anon 从 high 换成 transparent，总分应正好掉 (100-10)×0.25 = 22.5 分，对得上说明加权没写错。

完整代码

上面的函数拼起来，加上多线程批量调度就能直接跑。用线程池并发是因为检测是纯 IO 等待，串行测几百个 IP 会慢到没法用。

import time
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

# ---------- 配置 ----------
ECHO_URL = "https://httpbin.org/get"     # 回显请求头与来源 IP
TEST_URL = "https://www.baidu.com"       # 连通性/延迟测试目标
TIMEOUT = 8
LATENCY_ROUNDS = 3
STABILITY_ROUNDS = 5
WEIGHTS = {"conn": 0.30, "latency": 0.25, "anon": 0.25, "stability": 0.20}
ANON_SCORE = {"high": 100, "anonymous": 60, "transparent": 10, "unknown": 0}


def get_real_ip():
    try:
        r = requests.get(ECHO_URL, timeout=TIMEOUT)
        return r.json().get("origin", "").split(",")[0].strip()
    except Exception:
        return ""


def test_latency(proxy):
    proxies = {"http": proxy, "https": proxy}
    costs = []
    for _ in range(LATENCY_ROUNDS):
        start = time.time()
        try:
            r = requests.get(TEST_URL, proxies=proxies, timeout=TIMEOUT)
            if r.status_code == 200:
                costs.append(time.time() - start)
        except Exception:
            continue
    if not costs:
        return False, None
    return True, sum(costs) / len(costs)


def test_anonymity(proxy, real_ip):
    proxies = {"http": proxy, "https": proxy}
    try:
        r = requests.get(ECHO_URL, proxies=proxies, timeout=TIMEOUT)
        data = r.json()
        headers = {k.lower(): v for k, v in data.get("headers", {}).items()}
        origin = data.get("origin", "")
    except Exception:
        return "unknown"
    xff = headers.get("x-forwarded-for", "")
    if real_ip and (real_ip in origin or real_ip in xff):
        return "transparent"
    if headers.get("via") or xff:
        return "anonymous"
    return "high"


def test_stability(proxy):
    proxies = {"http": proxy, "https": proxy}
    ok = 0
    for _ in range(STABILITY_ROUNDS):
        try:
            r = requests.get(TEST_URL, proxies=proxies, timeout=TIMEOUT)
            if r.status_code == 200:
                ok += 1
        except Exception:
            pass
    return ok / STABILITY_ROUNDS


def score_proxy(proxy, real_ip):
    alive, avg_latency = test_latency(proxy)
    if not alive:
        return {"proxy": proxy, "alive": False, "total": 0}
    latency_score = max(0, min(100, (5 - avg_latency) / 4.5 * 100))
    anon = test_anonymity(proxy, real_ip)
    stability = test_stability(proxy)
    total = (100 * WEIGHTS["conn"]
             + latency_score * WEIGHTS["latency"]
             + ANON_SCORE[anon] * WEIGHTS["anon"]
             + stability * 100 * WEIGHTS["stability"])
    return {"proxy": proxy, "alive": True, "latency": round(avg_latency, 3),
            "anon": anon, "stability": round(stability, 2), "total": round(total, 1)}


def batch_score(proxy_list, max_workers=20):
    real_ip = get_real_ip()
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {pool.submit(score_proxy, p, real_ip): p for p in proxy_list}
        for f in as_completed(futures):
            results.append(f.result())
    results.sort(key=lambda x: x["total"], reverse=True)
    return results


if __name__ == "__main__":
    proxies = [
        "http://127.0.0.1:8080",
        "http://127.0.0.1:8081",
    ]
    for item in batch_score(proxies):
        if item["alive"]:
            print(f"{item['proxy']:<28} 总分{item['total']:>5}  "
                  f"延迟{item['latency']}s  匿名{item['anon']}  稳定{item['stability']}")
        else:
            print(f"{item['proxy']:<28} 不可用，已剔除")

把 proxies 列表换成自己的真实代理，运行后就能看到一份按总分降序的排名，死 IP 直接标“已剔除”。

跑的时候常踩哪些坑

绝大多数报错不是脚本 bug，而是代理本身或环境的问题。对着下表排查，比重读代码快：

报错信号	原因	修复
ProxyError / Max retries exceeded	代理已失效或格式写错	检查 http:// 前缀和端口，剔除死 IP
SSLError	代理不支持 HTTPS 隧道	换成 http 测试目标，或只测 http 协议
请求全部超时	TIMEOUT 太短或并发太高	调大 TIMEOUT、降低 max_workers
匿名度全返回 unknown	回显接口被限流或不可达	换回显服务，优先自建同结构接口
real_ip 取到内网地址	本机自身走了系统代理	关闭系统代理再取真实来源 IP

其中最容易被忽略的是最后一条：如果 get_real_ip() 拿到的是本机内网 IP，后面所有代理的匿名度判定都会失真，务必先确认真实来源 IP 取得干净再批量跑。

还能怎么优化

基础版够日常筛选用，往生产方向走还有几个可选升级项，按需取用：

权重动态化：把 WEIGHTS 做成按目标站配置，反爬严的站调高来源可辨识度权重，长任务调高稳定性权重，一套脚本适配多种采集场景。
加协议维度：把 SOCKS5 代理纳入检测，proxies 里写 socks5://ip:port，requests 需额外装 requests[socks]。
落地成代理池：把分数写进 Redis 并定时复评，低于阈值的自动下线，配合业务侧按分数取用，替代一次性跑分。
并发调优：max_workers 不是越大越好，本机出口带宽和回显接口的限流才是天花板，从 20 起步按实测调。