帮量化团队搭了几年采集系统，90% 的"解析 bug"其实是代理层的锅

做量化数据这行几年，接手过不少半路瘫痪的采集系统。有个规律很稳定：团队来找我的时候，十有八九认为是解析代码出了问题，排查下来，真正的病灶几乎都在代理接入层——没有验证环节，没有容错设计，IP 一失效整条链路就断。因子逻辑写得再漂亮，数据不连续，一切白搭。

这篇把我平时给团队做落地时的完整流程写出来：按调度、请求、解析、存储四层搭，代理配置每一步都给可复制的代码和验证方法。照着做，第一轮采集是能跑通的。

动手之前，先把三件事确认掉

很多人跳过这一步直接写代码，后面每一步都会卡。三个前置条件缺一不可：

1. 数据源合规且公开。 目标必须是公开页面或公开接口，不需要登录授权，采集行为遵守目标站点的 robots 协议。舆情、电商价格、招聘信息这类另类数据，业内确实已经广泛用于业绩预测和因子挖掘(21 财经有过相关报道)，但前提永远是公开、合规。数据源需要账号授权或者属于非公开数据的，先去解决授权问题，别带着不合规的源进采集环节。
2. Python 3.8+，pip 可用。 没什么好说的，基础中的基础。
3. 代理服务已开通，鉴权配好。 拿到接入地址或提取链接再开工。

环境准备：每个工具装它的理由

我一直反对"全家桶式"装依赖。最小可用环境就四类组件，每个对应一个明确职责：

代理服务的选型逻辑跟工具不一样。说句实在话：验证期最该看的不是资源池规模，而是能不能先测后买、接入改造成本低不低。 我现在给团队推荐的做法是找支持免费测试的服务商——比如极安代理对新注册用户开放 8 小时免费测试，API 快速接入，白名单和账密两种鉴权都支持。把下面的流程完整跑一遍，数据拿到手了再谈采购。采购决策放在验证之后，这个顺序别反。

系统分层：四层结构，各管各的

这套分层是我踩坑踩出来的通用划分，核心价值在于解耦：任何一层出问题，不污染其他层的逻辑，排查的时候能快速定位。

一个新手常犯的错：采集频率拍脑袋定。频率必须在调度层就跟数据源的更新节奏对齐，评估数据源要看频率、延迟、格式三要素。日更的电商价格数据按小时采，纯属浪费预算;分钟级更新的舆情数据按天采，信号早跑没了。

第一步：配代理接入，先验证再往下走

requests 的代理配置就一个 proxies 字典，两种鉴权方式两种写法。

白名单方式：先在代理服务后台把本机公网 IP 加进授权列表，代码里只写接入地址。适合出口 IP 固定的服务器环境：

proxies = {
    "http": "http://host:port",
    "https": "http://host:port",
}

账密方式：用户名密码直接写进接入地址。本地调试、出口 IP 不固定的环境用这个：

proxies = {
    "http": "http://user:pass@host:port",
    "https": "http://user:pass@host:port",
}

配完别急着写业务逻辑。我的习惯是任何代理配置必须先过回显验证，向 IP 回显接口发一次请求，对比返回的 IP:

import requests
r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=8)
print(r.json())

返回的 IP 和代理出口一致，链路才算生效。这个判定标准业内是共识，腾讯云开发者社区的 Requests 代理指南也是同样的做法。跳过这一步的人，后面排查问题时会分不清是代理没生效还是目标站封了你。

第二步：封装请求模块，裸调 requests.get 不能进生产

这是我 review 别人代码时说得最多的一句话。生产环境的请求模块必须内置三重容错：超时、重试、换 IP。我的参数经验值：超时 5-10 秒，重试 3 次，按 2 的幂次退避，每次重试前更换代理：

import time
import requests
​
def fetch(url, get_proxy, retries=3):
    for i in range(retries):
        try:
            r = requests.get(url, proxies=get_proxy(), timeout=8)
            r.raise_for_status()
            return r.text
        except requests.RequestException as e:
            print(f"第{i+1}次失败:{e}")
            time.sleep(2 ** i)
    return None

get_proxy 是代理提取函数：用短效类代理就在这里返回新提取的 IP，用隧道类代理返回固定接入地址即可。另外提醒一句，正式跑的时候把 print 换成 logging 落盘，捕获异常并记日志是排查代理问题的基础，没有日志的采集系统等于裸奔。

验证方法(我强烈建议做这个"破坏性测试")：故意填一个无效代理地址跑一遍，日志应该显示重试 3 次、间隔按 1 秒、2 秒、4 秒递增，最终返回 None 而不是程序崩溃。能优雅失败的模块才配上线。

第三步：调度和增量存储，各十行代码跑通验证版

APScheduler 按固定间隔触发采集函数，间隔取值参照前面说的数据源更新节奏：

from apscheduler.schedulers.blocking import BlockingScheduler
​
sched = BlockingScheduler()
sched.add_job(collect_task, "interval", minutes=30)
sched.start()

存储层的去重方案：用数据源主键加时间戳做去重键，SQLite 建表时把去重键设为唯一索引，写入用 INSERT OR IGNORE，重复数据自动跳过。

验证方法：连续手动触发两轮采集，第二轮结束后查总行数。行数持平，去重生效;行数翻倍，回去检查唯一索引建没建成功。这个测试两分钟，能省掉后面清洗重复数据的两小时。

报错排查表：按出现频率排序，先查前两条

这张表是我这几年积累的排错顺序，按高发程度从上往下排，大部分问题查前两条就能解决：

多说一句 ProxyError 为什么排第一：短效类代理的 IP 有固定存活期，到期自动失效，拿失效 IP 发请求只会得到连接错误。新手遇到这个错的第一反应往往是怀疑自己代码写错了，然后开始改代码。正确的排查顺序是先看提取时间，再看存活档位，最后才轮到代码。 记住这个顺序能少熬很多夜。

进阶：并发、成本、稳定性，按需选用别硬套

基础流程跑通之后再看这部分。三个方向我都标了适用场景，不匹配就不必上：

并发扩展：适用于分钟级更新的舆情、行情类数据，单线程采不完的情况。高并发下逐个提取 IP 会变成新瓶颈，我的经验是把换 IP 动作交给云端：隧道代理统一入口接入，云端毫秒级自动换 IP、异常 IP 自动切换，程序端并发逻辑一行不用改。极安代理的隧道产品默认 5M 带宽、每秒 5 个请求的基础配置，对多数中等规模采集够用。

成本控制：适用于策略验证期、临时性采集任务。验证期最怕为用不满的资源付费，计费粒度要能对上任务周期。短效代理按每日 IP 数计费的模式对这类场景很友好，像极安 1000 IP 低至 3.6 元/天起，存活 1-15 分钟五档可选，几个小时的临时任务不用背整月套餐。

可用率监控：适用于 7×24 小时持续采集。在请求模块里统计每小时的请求成功率和平均耗时，成功率连续两个周期低于 95% 就触发告警。告警的成本永远远低于事后补数据的成本，这句话建议贴在工位上。

以上就是全套流程。总结成一句话：采集系统的稳定性是设计出来的，不是修出来的。验证环节前置、容错三件套齐全、每层各管各的，系统才谈得上可维护。

评论区欢迎交流，尤其想听听大家在高并发场景下的换 IP 方案，以及 429 频控的其他实战处理经验。