一、为什么脚本第一天能跑，第三周就哑了

把教程里的脚本放进 crontab，前三天的日报通常很好看。问题集中在第三到第五周浮出水面，而且都不是报错的形式。

1.1 四个不报错的失效方式

四项的共同点是：监控系统不会叫。HTTP 状态码、异常数、任务退出码都正常，只有报表里的数字悄悄偏离。指望报警发现它们，等于让监控去猜业务语义。

1.2 教程停在「能拿到」，你要的是「能对齐」

一篇标准教程的路径是：装依赖 → 发请求 → 解析四个字段 → 存 CSV → 加一句 time.sleep(random.uniform(1, 3)) → 收工。这条路径解决的是「能不能拿到数据」。

生产环境的问题换了一组：这批数据覆盖了我昨天要的那批商品吗？同一件商品在两个市场取到的字段能对上吗？这次跑失败了，重跑会不会产生两行不同的价格？月底收到账单时，我能说出每一块钱换回多少条能用的记录吗？这类问题的答案不在请求里，在请求之外的那层结构里。

还有一笔更贵的成本：返工。笔记本电脑上的脚本错了，改一行再跑一次，代价是三十秒；pandas 里混了三周的重复行，代价是重新追回所有下游报表，而且你未必知道哪些报表用过这批数据。结构上省下的几个小时，会在数据事故里连本带利还回去。

本文的划分依据就是这个边界：中间四节是反爬的四道门，解决「请求能不能被当成正常流量」；后面八节是工程结构，解决「拿到的数据能不能每天对齐」。

二、五条 Python 路线，各自把多少工作留给你

选型不是挑 API，是挑「自己还要写多少代码」。同一份任务，五条路线留给你团队的工程量差别在十倍以上。

判断标准只有一条：哪一层的失败你打算自己负责。自建路线的隐性成本不在服务器账单上，在每次亚马逊调整风控之后那两三天的排查里。选路线时把「出事谁半夜起来修」这句话写进文档，比对比功能表有用。五条路线的授权边界、成本结构与适用场景，在亚马逊数据 API 全景里逐条展开。

三、第一道门：TLS 与 HTTP/2 指纹

多数组队在防护上栽的第一个跟头，是以为自己在和 HTTP 层打交道。握手在你发出第一行请求头之前已经结束，而那一瞬间就决定了你后面收到的页面是不是真的。

3.1 握手发生在请求之前

一次 HTTPS 连接建立时，客户端先发 ClientHello，里面包含 TLS 版本、密码套件列表、扩展列表与顺序、椭圆曲线与点格式、签名算法。这些信息在真实浏览器里是固定的，由浏览器版本与操作系统决定；在 Python 里由 ssl 模块与 OpenSSL 的编译选项决定。服务端不需要看你发什么 URL，光凭这一包就能判断「这是个 Python 客户端」。

所以你会遇到一种反直觉的情况：UA 换成了最新版 Chrome，代理换成了住宅 IP，请求间隔也降到了每 IP 每分钟三次，返回依然是验证页。因为 UA 是你在 HTTP 层自己写的字符串，而握手层的指纹你一行代码都没改。

3.2 JA3 与 JA4：从「顺序即身份」到「排序后哈希」

JA3 的做法是把 ClientHello 里的五个字段（TLS 版本、密码套件、扩展、椭圆曲线、曲线点格式）按出现顺序拼接后取 MD5。同一个软件栈的拼接结果稳定，于是这个哈希值就成了客户端的身份标识。厂商维护一份「已知浏览器指纹库」，不在库里的哈希按可疑处理。

JA4 由 FoxIO 提出，解决的问题是 Chrome 110 起引入的扩展顺序随机化——同一版本的 Chrome 每次连接会打乱部分扩展的顺序，JA3 因此会变，靠精确匹配 JA3 的检测会失效。JA4 在哈希前对扩展排序，因此不受置换影响，并且把 QUIC、ALPN、签名算法也纳入进来。今天主流风控用的是 JA4 一类的稳定指纹，这也是为什么「我明明固定了 JA3 还是被拦」。

3.3 Python 默认客户端的指纹，不存在于真实浏览器

公开的对比测试里，默认 requests 客户端在开启指纹检测的站点上成功率是个位数到低两位数，换成带 HTTP/2 的 httpx 略有改善，换成握手与浏览器一致的客户端可以进入 60%–90% 区间——同一个 IP、同样的频率，变量只有 TLS 栈。这类数字各家口径差异很大（目标站点、页面类型、IP 质量都不同），数量级可信，小数点不可信。

结论是：指纹是入场券，不是加分项。它在扫描器评估序列里排第一，因为它是唯一一个「错了后面全白搭」的变量。

3.4 换掉握手：curl_cffi 的三条注意事项

curl_cffi 是 curl-impersonate 的 Python 绑定，它把浏览器的 TLS 与 HTTP/2 栈打包进 wheel，用一个参数切换指纹档位。

from curl_cffi import requests

resp = requests.get(
    "https://www.amazon.com/dp/B08N5WRWNW",
    impersonate="chrome",                      # 用最新稳定档位
    headers={"Accept-Language": "en-US,en;q=0.9"},
    proxies={"https": "http://user:[email protected]:8000"},
    timeout=25,
)

三条容易踩的坑：

• 不要覆盖 User-Agent。impersonate="chrome" 会带出配套的 UA、Sec-Fetch-* 与 header 顺序。你手动写一个更高版本的 Chrome UA，等于把刚修好的一致性又拆开。
• 档位要么不锁版本，要么全局统一。impersonate="chrome" 跟随库升级；写死 chrome124 便于复现但会过期。同一个 Session 里不要混用档位， Session 级别的 cookie 与指纹应当同生命周期。
• Accept-Language 要跟着站点走。amazon.de 用 de-DE,de;q=0.9，amazon.co.jp 用 ja-JP,ja;q=0.9。语言与站点不匹配是最廉价也最常见的暴露点。

3.5 HTTP/2 的第二个指纹

握手过了还有一层：HTTP/2 的 SETTINGS 帧。客户端在连接前言之后发的第一帧里带着 HEADER_TABLE_SIZE、ENABLE_PUSH、INITIAL_WINDOW_SIZE、MAX_CONCURRENT_STREAMS 等参数，还有 WINDOW_UPDATE 的增量值与伪头顺序。Chrome、Firefox、Safari 各不相同，Go 标准库、Python 的 h2 实现又是一套。风控会把「TLS 说自己是 Chrome，HTTP/2 说自己是 Go 库」当成明显的矛盾信号。这也是为什么自己拼 httpx[http2] 常常不够。

如果目标还会执行 JS 挑战（传感器数据采集 navigator、screen、WebGL 等），纯 HTTP 客户端在那一层没有答案。这时候要么上真实浏览器内核，要么把这一层交给已经处理掉的服务。

3.6 怎么验证指纹改对了

别靠「能拿到数据」来判断。用公开指纹检测端点打一次，把返回的 ja3n / ja4 与你的目标浏览器版本比对；再拉一次目标页面，检查响应体里是否含 productTitle。两个都过，才叫改对了。

import json
from curl_cffi import requests

def fingerprint_report(profile: str) -> dict:
    r = requests.get("https://tls.browserleaks.com/json", impersonate=profile, timeout=20)
    d = r.json()
    return {"ja3n": d.get("ja3n_hash"), "ja4": d.get("ja4"), "ua": d.get("user_agent")}

print(fingerprint_report("chrome"))
print(fingerprint_report("safari"))   # 两个档位返回的哈希应当不同

提醒一句：curl_cffi 的 Windows 支持不完整（HTTP/3 相关依赖构建受限），Linux 与 macOS 上正常。生产环境用容器跑的话，把镜像基础系统与 wheel 的兼容性先验证一遍。

3.7 指纹对了，数据不一定对

指纹解决的是「被不被当作机器人」，不解决「拿到的是不是你要的东西」。风控还有一种更阴的处理方式：不封你，给你一份降质的数据——价格比真实慢一天、搜索结果的广告位被抽掉、库存状态固定为「有货」。这类响应状态码 200、字段齐全、解析无异常，只有跟真人对拍才能发现。所以反爬做到位之后，还得有第九节那道质量门。

四、第二道门：浏览器指纹不止 TLS

如果你的方案里有 Playwright 或 Puppeteer，攻击面会从协议层扩展到整个运行时。无头浏览器的破绽是系统性的，而且大部分在默认配置下就敞着。

4.1 无头浏览器的七个破绽

4.2 一致性比完美重要

新手常犯的错误是逐项优化：UA 用最新的，WebGL 渲染串挑个高端显卡，分辨率设成 4K。三项单看都合理，合起来是「用着 RTX 4090 的 MacBook 用户」，在风控模型里这种组合的存在概率接近零。风控抓的是矛盾，不是落后。

实操上维护一份「设备档案」，五件套一起切：UA 与平台、WebGL 渲染串、屏幕与像素比、时区与语言、Accept-Language。档案数量不用多，三到五套足够，关键是每套内部自洽、切换时整体切换。

4.3 哪些亚马逊数据真的需要执行 JS

这个问题决定你要不要上无头浏览器，成本差一个数量级。经验划分：

• 服务端渲染即可拿到：商品标题、品牌、BSR 主类目与子类目、评分与评论数、五点描述、变体的大部分属性。
• 需要渲染或交互才完整：Buy Box 的实时报价与卖家轮转、部分促销标签、搜索结果页的 sponsored 标记与广告位、评论区的懒加载与「Customer says」聚合块、指定邮区后的配送时效。

如果你的需求落在第一类，用 curl_cffi 就够了，没必要为一个页面扛一整套浏览器集群。落在第二类，尤其是搜索页广告位与邮区级价格，浏览器的成本就躲不掉——这也是很多团队最终改用数据 API 的转折点。

五、第三道门：什么时候必须换成住宅 IP

指纹改对之后，IP 是下一个变量。这一层的判断不需要试错，看 ASN 就能预判。

5.1 三类 IP 的信任来源不同

报价区间取自多家代理服务商 2026 年公开价目，不同套餐与地区差异很大，只用来判断数量级。关键在于：数据中心 IP 不是「质量差一点」，而是「在评估开始之前就已经被分类」，你后面的指纹做得再好也不进入同一套打分。

5.2 按页面类型看难度

把「采集亚马逊」当成一件事，是最大的误判。不同页面的防护强度差得很远，同一套 IP 与指纹配置在商品页和卖家页上的成功率可以差出几十个百分点。

公开的对比评测给出的量级大致是：数据中心 IP 在强保护页面上成功率常落在 10%–40%，住宅 IP 在 50%–95% 之间波动（取决于 IP 卫生与是否共享），移动 IP 多数场景在 85% 以上。区间宽是因为变量太多，但排序稳定：失败请求的钱是白花的，而且失败请求也一样计费。

5.3 五个信号：数据中心 IP 已经不够了

不必等到成功率崩了才换。出现下面任意两条，就该评估住宅 IP：

• 验证页占比超过 5%，且换指纹档位后没有下降；
• 成功率按小时衰减——早上 90%，下午 60%，第二天又回到 90%，典型的 IP 池被消耗；
• 同样的请求，从公司网络手动访问正常，从服务器访问必被拦；
• 粘性会话频繁中断，翻到第三页就丢会话；
• 返回数据的地域出现漂移——要的是 amazon.de 的价格，拿到的是其他站点的币种。

5.4 住宅 IP 的隐藏成本

住宅 IP 的账单不只有流量费。四笔容易被漏掉：

• 失败请求照常计费。按 GB 计费的方案里，验证页也是流量。40% 成功率意味着 2.5 倍流量。
• IP 卫生。低价住宅池里的 IP 可能已经被别的采集任务用旧了。同样的报价，池子质量能把成功率拉开一倍。
• 粘性会话的溢价。需要保持同一 IP 一段时间（翻页、登录后操作）通常要额外付费或消耗更多额度。
• 运维。池子会退化，需要持续监控成功率分布、剔除坏 IP、调整每 IP 速率。这部分是人力成本，也是最容易被低估的一项。按「报价口径 vs 实际账单」比较多家方案的方法，见亚马逊数据 API 选型对比。

5.5 住宅 IP 还是移动 IP：什么时候值得加价

移动 IP 的信任度更高，原因是运营商的 CGNAT 让一个出口 IP 背后站着成百上千个真实手机用户，风控封它的代价太大。但移动出口更贵、延迟更高，不适合全量铺开。

合理的用法是分层：商品页与榜单页用住宅 IP，把移动 IP 留给两类任务——一是卖家与店铺信息这类防护最重的页面，二是住宅池成功率掉到阈值以下时的救急通道。判断依据不是单价，是「失败之后重跑一次的代价」：如果一次失败要等到第二天才能补，那笔账乘以失败率，通常高于移动 IP 的差价。

还有一条容易漏：别把整个任务的成功率当成决策依据。按页面类型分开统计，你会发现往往是某一个端点拖垮了平均值，而它只需要一小部分流量走高等级出口。

5.6 什么时候数据中心 IP 就够了

三种情况没必要上住宅 IP：目标是自建或授权的公开接口（无风控的 API）；采集对象是防护很轻的站点（独立站、小型电商）；你用的是已经把出口层收走的数据服务，出口由服务端负责。第三种情况正是第十六节要讲的。

六、第四道门：混淆——把节奏、会话与来源做成不像脚本

指纹与 IP 决定「你是谁」，节奏与行为决定「你像不像在逛」。这一层不需要高技术，需要的是放弃整齐。

6.1 抖动不是 random.uniform

固定间隔的 2 秒，和均匀分布在 1–3 秒之间的随机数，在时序特征上一样好认：两者的间隔分布都有明显边界，真实用户的访问间隔是长尾分布——大部分间隔短，偶尔出现几分钟的空档，还会成簇出现（翻了五页然后停一会儿）。

import random, asyncio

async def human_gap(base: float = 1.2) -> float:
    """长尾间隔：多数 0.6–1.8 秒，偶发 5–20 秒停顿。"""
    if random.random() < 0.08:              # 8% 概率来一次长停顿
        return random.uniform(5.0, 20.0)
    return random.expovariate(1.0 / base)   # 指数分布，无上界

async def paced_worker(work, limit_per_min: int = 20):
    """每 IP 每分钟硬上限，超出就等。"""
    gap = 60.0 / limit_per_min
    for item in work:
        await asyncio.sleep(await human_gap(gap))
        yield item

三个参数比间隔本身更重要：每 IP 每分钟请求数上限（搜索页建议个位数，商品页可放宽）、单会话连续请求数（超过二三十个就换出口）、任务的时间分布（别让所有任务都在整点启动）。

6.2 每请求换 IP，还是粘性会话

判断依据是目标页面要不要会话状态。商品页无状态，每请求换 IP 更省池子也更安全；搜索页翻页、指定邮区后的连续查询、带登录态的操作必须粘性，切换成本是丢会话和重新挑战。折中做法是按「会话」而不是按「请求」轮换：一个会话内固定出口，会话结束后换新出口。

6.3 Referer 链与来源可信度

直接深链到商品详情页，没有 referer、没有搜索历史、没有 cookie，是少数派行为。真实的访问多半来自搜索结果、类目页或站外广告。成本最低的改法是两步走：先请求一次搜索页或类目页，带上它的 URL 作为 referer 再进详情页。这一步能同时解决 cookie 初始化与 referer 缺失。

6.4 Cookie 的生命周期

用 Session 复用 cookie jar，别每次新建连接；也不要把一份 cookie 用到天荒地老。亚马逊会轮换 session token，长期不变的 cookie 本身就是信号。合理策略是给会话设一个生命周期（比如 15–30 分钟或 20–50 个请求，先到先算），到期丢弃重建。

6.5 地理一致性：IP 在德国却说 en-US

出口国家、站点域名、Accept-Language、时区、币种、邮区，六项必须同源。采集德国站就用法兰克福出口 + amazon.de + de-DE + Europe/Berlin + EUR；做邮区级价格时，邮区也必须在出口国境内。这项检查应该写成断言进代码，而不是靠人记。

6.6 边界：合规与频率

亚马逊的服务条款明确限制自动化访问，公开数据的采集在多个司法辖区有判例支持，但两者不互相抵消。实操上守住三条：只取公开可见、无需登录即可访问的数据；不触碰个人信息与受版权保护的完整内容；频率以不对目标站点造成负担为准。商业用途上线前走一遍法务评估，比事后补救便宜。

6.7 上线前自检清单

下面十条按「出问题概率 × 排查成本」排序，每条都可以在十分钟内验证。跑通再上生产，能省掉大部分半夜告警。

1. 指纹自测：用公开指纹端点打一次，确认 ja3n / ja4 与目标浏览器版本匹配，而不是「能返回数据就算过」。
2. 拦截面识别：把验证页的特征串写进检测函数，确认返回 200 时也能判定失败。
3. 六项同源：出口国家、站点域名、Accept-Language、时区、币种、邮区，写成断言而不是靠人记。
4. 超时是元组：连接超时与读取超时分开设置，避免半开连接把 worker 占满。
5. 重试预算：只对瞬时失败与限流重试，契约类失败直接拦截；预算用完停任务。
6. 退避带抖动：重试时间加随机量，否则一批失败会在同一秒集体重发。
7. 速率双限：并发数 + 每秒请求数 + 每 IP 每分钟请求数，三个上限都要有。
8. 主键含日期：asin + marketplace + captured_at + contract_version，验证重跑不产生重复行。
9. 成本入日志：attempts、status、marketplace、contract_version 四个字段落盘。
10. 补数据路径：先想清楚失败之后怎么补——是重跑当天，还是等次日窗口。没有答案就别上量。

七、请求层：把超时、凭据、重试、指纹收进一个类

四道门都过了，才轮到代码结构。第一件事是把散落在各处的请求参数收进一个对象，让「一次调用」有唯一定义。

三条硬规则：超时必须显式给（连接超时与读取超时分开），凭据必须来自环境变量而不是源码，重试必须在类内部完成且带上次数。下面这个类同时对自建抓取与使用数据 API 两种后端适用，差别只在 transport。

# pip install curl_cffi tenacity
import os, random, logging
from dataclasses import dataclass
from curl_cffi import requests as creq

log = logging.getLogger("amz")

@dataclass
class FetchResult:
    ok: bool
    status: int
    payload: dict | None
    attempts: int
    error: str | None = None

class AmazonClient:
    """一次调用 = 一次带指纹、带超时、带预算的请求。"""

    ENDPOINTS = {                       # 路径以官方接入文档为准
        "product":  "/api/v1/amazon/product",
        "search":   "/api/v1/amazon/search",
        "review":   "/api/v1/amazon/review",
        "bestseller": "/api/v1/amazon/bestsellers",
    }

    def __init__(self, base_url: str, token: str | None = None,
                 profile: str = "chrome", proxy: str | None = None,
                 max_attempts: int = 3):
        self.base = base_url.rstrip("/")
        self.token = token or os.environ["PANGOLINFO_TOKEN"]
        self.profile = profile
        self.proxies = {"https": proxy} if proxy else None
        self.max_attempts = max_attempts
        self.session = creq.Session(impersonate=profile)

    def fetch(self, kind: str, params: dict, marketplace: str) -> FetchResult:
        url = self.base + self.ENDPOINTS[kind]
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Accept-Language": self._lang(marketplace),
        }
        delay = 1.0
        for attempt in range(1, self.max_attempts + 1):
            try:
                r = self.session.get(url, params={**params, "marketplace": marketplace},
                                     headers=headers, proxies=self.proxies,
                                     timeout=(5, 30))          # (连接, 读取)
                if r.status_code == 200:
                    return FetchResult(True, 200, r.json(), attempt)
                if r.status_code in (429, 503):                # 可退避
                    self._sleep_backoff(attempt, delay, r.headers.get("Retry-After"))
                    continue
                return FetchResult(False, r.status_code, None, attempt, f"http {r.status_code}")
            except Exception as e:                             # 超时、连接重置、解析失败
                log.warning("attempt %s failed: %s", attempt, e)
                self._sleep_backoff(attempt, delay)
        return FetchResult(False, 0, None, self.max_attempts, "exhausted")

    @staticmethod
    def _sleep_backoff(attempt: int, delay: float, retry_after: str | None = None):
        import time
        wait = float(retry_after) if retry_after else delay * (2 ** (attempt - 1))
        time.sleep(min(wait + random.uniform(0, 0.4), 30))     # 抖动，避免重试同步

    @staticmethod
    def _lang(marketplace: str) -> str:
        return {"US": "en-US,en;q=0.9", "DE": "de-DE,de;q=0.9",
                "JP": "ja-JP,ja;q=0.9", "UK": "en-GB,en;q=0.8"}.get(marketplace, "en-US,en;q=0.9")

三个容易被忽略的细节：timeout 写成元组，连接 5 秒读取 30 秒，避免卡在半开连接上；退避带随机抖动，否则一批失败的请求会在同一秒集体重试；attempts 必须返回，它是第十四节成本公式里的变量。

如果你还在自建抓取，把 session.get 的 URL 换成商品页、把 impersonate 留着、把代理池接进来，其余结构不变。自建抓取与调用数据 API 的成本分界我们在另一篇里算过账。

八、字段契约：先约定「这个词到底叫什么」

数据事故里最常见的不是采不到，是采到了但含义变了：price 昨天是 Buy Box 价格，今天变成列表价；reviews 上周是评论数，这周变成评分。上游不会通知你，你的下游也不会问。

from dataclasses import dataclass, asdict
from typing import Optional

@dataclass(frozen=True)
class ProductSnapshot:
    asin: str
    marketplace: str
    captured_at: str          # ISO 日期，进主键
    title: str
    brand: Optional[str]
    price: Optional[float]    # Buy Box 价格，币种见 currency
    currency: str
    rating: Optional[float]   # 0–5
    review_count: Optional[int]
    bsr_main: Optional[int]
    bsr_category: Optional[str]
    contract_version: str = "2026-09-01"

P0_FIELDS = ("asin", "title", "price", "currency")   # 缺任一即视为不可用

契约的三条纪律：字段名一旦写下不再改语义，要改就升 contract_version；所有可空字段显式标 Optional，不允许隐式 None；captured_at 用采集日期而非写入时间，重跑同一天必须落到同一天。

九、质量门：覆盖率和填充率必须分开算

这两个数字一旦混成一个「成功率」，你就再也发现不了降质。

• 覆盖率：我要的 1,000 个 ASIN，这次拿回来几个。分母是任务清单。
• 填充率：拿回来的记录里，P0 字段有值的比例。分母是返回记录。

def is_robot_check(html_or_json) -> bool:
    """拦截面往往返回 200，必须从内容判断。"""
    if isinstance(html_or_json, str):
        markers = ("[email protected]", "Enter the characters you see",
                   "Robot Check", "automated access")
        return any(m in html_or_json for m in markers)
    return False

def quality_gate(records: list[dict], wanted: set[str]) -> dict:
    got = {r["asin"] for r in records if r.get("asin")}
    usable = [r for r in records
              if not is_robot_check(r) and all(r.get(f) not in (None, "") for f in P0_FIELDS)]
    return {
        "coverage": len(got & wanted) / max(len(wanted), 1),
        "fill_rate": len(usable) / max(len(records), 1),
        "usable": len(usable),
        "blocked": len(records) - len(usable),
    }

阈值按业务定：价格监控建议覆盖率 ≥ 98%、P0 填充率 ≥ 95%，低于就拦截整批并告警，而不是让半批数据混进下游。半批数据比没有数据更贵，因为它看起来像成功了。

这一层在数据管道里是独立的一节，因为它是唯一一个「上游换了供应商也要重写」的模块。域名与出口换了，判定降质的标准不变。

十、翻页与去重：先定主键，再谈翻页

主键定错，翻页做得越漂亮，重复数据越多。主键四要素：asin + marketplace + captured_at + contract_version。前两项定位对象，第三项定位时点，第四项定位口径。

def dedupe(rows: list[dict]) -> list[dict]:
    seen, out = set(), []
    for r in rows:
        key = (r["asin"], r["marketplace"], r["captured_at"], r.get("contract_version", ""))
        if key in seen:
            continue
        seen.add(key)
        out.append(r)
    return out

翻页的两个坑：一是页数上限，亚马逊搜索结果通常只开放有限页数，超出的请求会静默返回首页或空页，表现为「数据在增长但增速不对」；二是翻页期间结果集会变动，同一页第二次取到的商品顺序可能不同。解决办法是把「本次任务的目标集合」先固化（用 ASIN 清单驱动），而不是靠翻页遍历。

十一、失败四类：不是所有失败都该重试

把失败分成四类，每类的处理动作不同。混为一谈的后果是：该重试的没重试，不该重试的把账单翻了几倍。

第四类最容易处理错：它不是请求失败，是上游变了。重试一百次得到的还是同样的空字段，只是把成本乘了一百。给每类失败单独的计数器，重试预算只在第一、二类上消耗。

十二、并发：信号量，而不是 random.sleep

并发不是越大越好。上限由三个数决定：上游给的速率额度、你的出口 IP 数量、以及目标能承受的每 IP 频率。用 asyncio.Semaphore 卡住同时在飞的请求数，再配一个令牌桶卡住每秒请求数。

import asyncio, time

class TokenBucket:
    def __init__(self, rate: float, burst: int):
        self.rate, self.tokens, self.burst, self.ts = rate, burst, burst, time.monotonic()

    async def take(self):
        while True:
            now = time.monotonic()
            self.tokens = min(self.burst, self.tokens + (now - self.ts) * self.rate)
            self.ts = now
            if self.tokens >= 1:
                self.tokens -= 1
                return
            await asyncio.sleep(1 / self.rate)

async def run_all(client, jobs, concurrency: int = 4, rps: float = 2.0):
    sem, bucket = asyncio.Semaphore(concurrency), TokenBucket(rps, burst=int(rps * 2) + 1)
    async def one(job):
        async with sem:
            await bucket.take()
            return await asyncio.to_thread(client.fetch, *job)
    return await asyncio.gather(*(one(j) for j in jobs))

调优顺序：并发从 4 起步，盯 429 占比与端到端时延两条曲线往上加；加到某个点，吞吐不再涨而排队时间变长，就退回上一档。天花板通常是上游额度，不是你的机器。

十三、落地：写快照，而不是写覆盖

价格监控、排名追踪、评论增长，这三件事的共同前提是历史留存。用 INSERT OR REPLACE 加上第十节的主键，重跑自动幂等，历史自动留下。

import sqlite3

DDL = """
CREATE TABLE IF NOT EXISTS product_snapshot (
  asin TEXT NOT NULL, marketplace TEXT NOT NULL,
  captured_at TEXT NOT NULL, contract_version TEXT NOT NULL,
  title TEXT, brand TEXT, price REAL, currency TEXT,
  rating REAL, review_count INTEGER, bsr_main INTEGER, bsr_category TEXT,
  PRIMARY KEY (asin, marketplace, captured_at, contract_version)
);"""

def save(con: sqlite3.Connection, rows: list[ProductSnapshot]) -> int:
    con.executemany(
        "INSERT OR REPLACE INTO product_snapshot "
        "(asin, marketplace, captured_at, contract_version, title, brand, price,"
        " currency, rating, review_count, bsr_main, bsr_category) "
        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
        [(r.asin, r.marketplace, r.captured_at, r.contract_version, r.title, r.brand,
          r.price, r.currency, r.rating, r.review_count, r.bsr_main, r.bsr_category)
         for r in rows])
    con.commit()
    return len(rows)

变更检测因此退化成一条窗口函数：同一 ASIN 按日期排序，LAG(price) 与当前值比较即可，不需要额外的比对任务。

十四、成本埋点：把钱算进每一次调用

自建路线的成本公式：

每千条可用记录成本 = 1000 × 单价 × 平均尝试次数 × (1 + 渲染倍率)
                     ────────────────────────────────────────────
                              成功率 × 填充率 × (1 - 拦截率)
                     + 每千条的代理流量费 + 每千条的工程维护摊销

分母里的三个数（成功率、填充率、拦截率）必须来自你自己的日志，不能用供应商给的平均值。平均尝试次数来自第七节的 attempts。代理流量费按 GB 计，失败请求也算。工程维护摊销最容易被忽略：一次风控升级引发的排查，通常是两三个人日，按月摊进每条记录里。

算出来之后做一次对比实验：同样 1,000 个 ASIN、同样频率，自建路线与数据 API 各跑一周，比三个数——每千条可用记录成本、数据延迟中位数、凌晨被叫起来的次数。第三个数字往往最有说服力。

十五、观测：四个指标，两条告警

指标不用多，四个够用：覆盖率、P0 填充率、p95 端到端时延、每千条可用记录成本。前两个来自第九节，第三个按请求打点算分位数，第四个来自上一节。

两条告警就够：

• 静默告警：连续两个周期覆盖率或填充率低于阈值 → 大概率是上游降质或改版，不要等用户发现。
• 成本告警：每千条可用记录成本环比上涨超过 30% → 通常是拦截率上升或重试失控，此时成功率指标可能还很好看。

把 attempts、status、marketplace、contract_version 四个字段打进每条日志。出事时你要回答的第一个问题永远是「从什么时候开始的、影响哪些市场」，没有这四个字段就得重新跑一遍历史。

十六、Pangolinfo 在这套代码里的位置

把前面的结构摊开看，四道门加八个环节里，只有三件事是业务侧必须自己做的：定义字段契约、定义质量阈值、定义数据怎么用。其余都是「为了拿到干净数据而不得不做的基础设施」。

Pangolinfo 的 Amazon Scraper API 承接的是后者。你发出去的那一个请求里，已经包含：

换句话说是这个分工：你发一个请求，收一份结构化 JSON，中间那七层由服务端负责。计费口径是单一的记录口径，不存在「住宅 IP 加价」「渲染加倍」「端点难度倍率」这类拆分项；具体费率与套餐见定价页，接入方式见接入文档。

三件我们不做，也会明确告诉你不做：不替你定义字段契约——「价格」指 Buy Box 还是列表价是你的业务决定；不替你判断数据保留多久、能不能用于某个用途——合规边界在你和你的法务手里；不采集需要登录才能访问的私有数据——卖家后台、订单与广告报表不在公开数据范围内。

这套分工的价值不在省下多少代理费，在于把「风控升级 → 排查 → 改代码 → 补数据」这条循环从你的季度计划里删掉。

十七、常见问题

下面十一个问题按来源分成三组：前三问来自「跑不通」，中间三问来自「会不会出事」，最后五问来自「值不值这个钱」。这也是我们在接入支持里被问得最多的顺序。