如果Wildberries或Ozon的解析器突然开始返回空页面、超时和403错误——问题不一定出在平台本身。通常，代理池中的一部分IP地址已经“过期”：它们被封禁、减速或被要求输入验证码。我们将分析如何通过哪些指标区分有效代理和无效代理，以及如何在没有人工监控的情况下自动化筛选。
为什么代理池会随着时间退化
任何代理池——即使是从可靠的提供商那里购买的——随着时间的推移都会失去一部分有效地址。原因各不相同：Wildberries、Ozon和Avito的反欺诈系统会在IP上积累声誉标记，一些地址在其他客户的批量请求后会被列入黑名单（如果代理是共享的），数据中心的IP范围会成批被封禁，而住宅IP有时会因为真实用户关闭设备而退出池。
如果不清理代理池，“死”代理的比例会非线性增长：最初是5%，经过一周的活跃解析后达到20-30%。解析器开始在无效IP上浪费资源进行重试，单个页面的平均收集时间增加，日志中积累的错误掩盖了实际问题（例如，网站布局的变化）。没有筛选系统，您要么保持代理池膨胀以备不时之需，要么定期遭遇数据收集的失败。
任务不仅是记录错误的事实，而是对其进行分类：服务器的临时过载、特定IP的封禁、区域性封锁或解析器端的问题（错误的头部、过期的cookie）。错误类型决定了是否需要永久性地将代理从池中移除、暂时将其置于“隔离”状态，还是仅仅重复请求。
筛选的关键错误指标
为了区分临时故障和真正的无效代理，需要为每个IP收集多个指标，而不是仅仅依赖单一的HTTP状态。以下是每次请求时应记录的基本指标集：
| 指标 | 信号 | 行动 |
|---|---|---|
| HTTP 403 / 429 | IP被封禁或限速 | 隔离30-60分钟，若重复则删除 |
| 连接超时 | 代理过载或无效 | 连续3次超时——从池中删除 |
| 响应中的CAPTCHA | IP受到反机器人系统的怀疑 | 降低请求频率或更换IP |
| 响应时间 > 5秒 | 代理速度退化 | 在池中降低优先级 |
| 响应大小低于平均值 | 返回了占位符/封锁页面 | 在接受之前进行额外内容检查 |
| 成功请求比例（成功率） | IP质量的综合评估 | 在50个请求中低于70%——删除 |
重要的是不要将网站错误代码与网络错误混淆。例如，404并不是代理的问题，而是Wildberries上缺少的商品。但是，如果同一个IP在其他IP返回200的地方经常收到403，这显然是该地址被单独封禁的迹象。
代理评分算法
与其采用二元的“有效/无效”方式，不如为每个IP计算一个从0到100的评分，该评分在出现错误时逐渐降低，在成功请求时恢复。这使得在一次偶然的故障后不必立即丢弃代理（市场平台的服务器可能只是暂时挂起了一秒），但可以快速移除那些系统性返回垃圾的IP。
逻辑很简单：成功请求返回200且内容有效时，评分增加+2（但不超过100）。超时——减去15。403/429代码——减去25。响应中有CAPTCHA——减去20。如果评分低于阈值（例如，30），代理将被隔离一段固定时间；如果在返回隔离后评分再次低于10——该IP将被永久删除并记录以供分析。
这种方法类似于反检测浏览器（如Dolphin Anty或AdsPower）中的声誉系统，其中账户的“健康”取决于活动。相同的机制也非常适用于解析中的IP地址。
代码：带有自动筛选的ProxyPool（Python）
以下是一个用于管理代理池的类的工作框架，考虑了错误指标。可以将其连接到任何基于requests、httpx或Scrapy的解析器，将获取代理的逻辑替换为调用pool.get_proxy()。
import time
import random
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class ProxyStats:
address: str
score: int = 100
quarantine_until: float = 0.0
total_requests: int = 0
total_success: int = 0
consecutive_timeouts: int = 0
is_dead: bool = False
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 1.0
return self.total_success / self.total_requests
class ProxyPool:
def __init__(self, proxies: list[str],
min_score: int = 30,
dead_score: int = 10,
quarantine_seconds: int = 1800):
self.stats = {p: ProxyStats(address=p) for p in proxies}
self.min_score = min_score
self.dead_score = dead_score
self.quarantine_seconds = quarantine_seconds
def _is_available(self, stats: ProxyStats) -> bool:
if stats.is_dead:
return False
if stats.quarantine_until > time.time():
return False
return True
def get_proxy(self) -> Optional[str]:
available = [s for s in self.stats.values() if self._is_available(s)]
if not available:
return None
# 按评分排序，从前30%中随机选择一个，以避免频繁使用同一个IP
available.sort(key=lambda s: s.score, reverse=True)
top_pool = available[:max(1, len(available) // 3)]
return random.choice(top_pool).address
def report_success(self, proxy: str):
s = self.stats[proxy]
s.total_requests += 1
s.total_success += 1
s.consecutive_timeouts = 0
s.score = min(100, s.score + 2)
def report_timeout(self, proxy: str):
s = self.stats[proxy]
s.total_requests += 1
s.consecutive_timeouts += 1
s.score -= 15
if s.consecutive_timeouts >= 3:
self._quarantine(s)
self._check_dead(s)
def report_blocked(self, proxy: str, captcha: bool = False):
s = self.stats[proxy]
s.total_requests += 1
s.score -= 20 if captcha else 25
self._quarantine(s)
self._check_dead(s)
def _quarantine(self, s: ProxyStats):
if s.score < self.min_score:
s.quarantine_until = time.time() + self.quarantine_seconds
def _check_dead(self, s: ProxyStats):
if s.score <= self.dead_score:
s.is_dead = True
def stats_report(self) -> list[dict]:
return [
{
"proxy": s.address,
"score": s.score,
"success_rate": round(s.success_rate, 2),
"dead": s.is_dead,
}
for s in self.stats.values()
]
在实际的解析器中，report_success、report_timeout和report_blocked的调用直接插入到每个HTTP请求后的响应处理程序中。与requests的集成示例：
import requests
def fetch(url: str, pool: ProxyPool):
proxy = pool.get_proxy()
if not proxy:
raise RuntimeError("代理池中没有可用的代理")
proxies = {"http": proxy, "https": proxy}
try:
resp = requests.get(url, proxies=proxies, timeout=8)
except requests.exceptions.Timeout:
pool.report_timeout(proxy)
return None
if resp.status_code in (403, 429):
pool.report_blocked(proxy)
return None
if "captcha" in resp.text.lower():
pool.report_blocked(proxy, captcha=True)
return None
pool.report_success(proxy)
return resp
这种方法减轻了人工监控的负担：您只需设置一次阈值，之后池子会自动“恢复”，排除无效地址并将暂时挂起的地址从隔离中恢复。
IP封禁的阈值
没有通用的数字——阈值取决于平台及其反机器人系统的激进程度。但有一些在Wildberries、Ozon和Avito解析中实践效果良好的参考：
- 成功率低于70%，在30-50个请求的样本中——即使当前评分尚未达到临界值，也应删除该IP。
- 连续3次超时——几乎总是意味着代理物理上不可用或过载，没有必要将其保留在活跃池中。
- 每20个请求中出现CAPTCHA超过1次——该IP已经受到反机器人系统的怀疑，最好降低请求频率或更换地址。
- 平均响应时间超过池的中位数的2倍——信号退化，虽然不严重，但应在选择时降低优先级。
对于数据中心和住宅代理，30-60分钟的隔离通常足够让反机器人系统“忘记”临时的可疑活动。对于移动代理情况则不同：在运营商更换基站时，IP会发生变化，因此隔离时间可以缩短——10-15分钟，因为在轮换后这实际上是另一个地址。
健康检查和后台轮换
除了根据解析主流中的错误进行反应式筛选外，定期进行后台健康检查也是有益的——一个单独的进程，每隔几分钟对代理进行轻量请求（例如，httpbin.org/ip），并在代理进入Wildberries或Ozon的实际请求之前更新池的统计数据。
import threading
def healthcheck_loop(pool: ProxyPool, interval: int = 300):
while True:
for proxy in list(pool.stats.keys()):
try:
r = requests.get(
"https://httpbin.org/ip",
proxies={"http": proxy, "https": proxy},
timeout=5
)
if r.status_code == 200:
pool.report_success(proxy)
else:
pool.report_blocked(proxy)
except requests.exceptions.RequestException:
pool.report_timeout(proxy)
time.sleep(interval)
threading.Thread(target=healthcheck_loop, args=(pool,), daemon=True).start()
这种后台进程尤其重要，如果代理池很大（数百个IP）且解析不是持续进行的，而是分批进行——没有健康检查，部分代理在启动之间会“过期”，而解析的前几分钟会花费在实时筛选上，而不是立即使用经过验证的地址。
选择哪种类型的代理进行市场解析
代理类型的选择直接影响到筛选的频率和阈值的激进程度。对于Wildberries和Ozon的稳定解析，反机器人系统分析行为模式和子网声誉，住宅代理表现最佳——它们的IP属于真实的家庭互联网提供商，反机器人系统通常不会成批封禁它们。
对于更简单和保护较少的来源（例如，从Avito收集公共数据或监控简单的价格展示），通常足够使用数据中心代理——它们速度更快且成本更低，这在大量请求时至关重要，但被封禁的频率略高，因此应对它们设置更严格的筛选阈值。
| 代理类型 | 筛选速度 | 何时使用 |
|---|---|---|
| 数据中心 | 高，子网成批被封禁 | 简单展示、大量请求、保护较低 |
| 住宅 | 中等，IP被单独封禁 | Wildberries、Ozon、反机器人网站 |
| 移动 | 低，IP由运营商自动更换 | 高度保护的来源、高频解析 |
在实践中，许多团队使用混合代理池：大部分请求通过住宅IP进行，而在大规模封禁时，移动代理作为“应急”备用，因为它们的IP更换是通过移动网络的轮换自然发生的，不需要人工干预。
结论
没有筛选系统的代理池不可避免地会退化，且它工作越久而不清理，解析器在无效IP上浪费的资源就越多。考虑HTTP代码、超时、CAPTCHA和成功率的评分模型可以自动识别问题地址，并保持池的有效状态，而无需人工监控。本文中的代码可以适配任何技术栈——requests、httpx、Scrapy或自定义异步解析器。
如果您刚开始为市场解析收集代理池，并希望在一开始就降低封禁频率，建议考虑使用住宅代理——它们在Wildberries和Ozon等受保护平台上提供明显更高的成功率，相比普通数据中心地址。