用代理做电商价格监控：抓取策略、频率控制与合规边界

价格监控真正难在哪

技术上”抓一个价格数字”很简单。实际项目中难的是三件事：看到的价格是不是真实价格、抓取频率怎么定、以及哪些数据不能碰。

一、你看到的价格可能不是真实价格

电商价格高度个性化。同一个商品页，不同地区、不同设备、是否登录、有没有领券，展示的价格和库存都可能不同。如果你的监控出口在机房、IP 段固定，很可能拿到的是占位价格或者被抬高的价格。

验证方法：用目标地区的住宅 IP 和你的常规出口各请求一次，比对结果。如果价格不同，说明站点在做地域差异化，你的监控数据从源头上就是错的。

因此价格监控必须用住宅代理 + 城市级定向。城市定向不是可选项，因为很多平台的价格策略是按城市甚至按商圈定的。

# 每个监控点位绑定一个目标城市的粘性会话
TARGETS = [
    {"sku": "A123", "city": "shanghai",  "session": "sh-01"},
    {"sku": "A123", "city": "chengdu",   "session": "cd-01"},
    {"sku": "A123", "city": "guangzhou", "session": "gz-01"},
]

def build_proxy(session_id: str) -> str:
    # 城市定向参数以官方文档为准，这里演示结构
    return (f"http://用户名-country-cn-city-{session_id}"
            f":密码@gateway.example.com:8080")

这里用粘性会话是有意的：同一个城市点位始终从同一个出口请求，价格的波动才能归因于平台调价，而不是出口 IP 变化。

二、要监控哪些字段

只抓价格是不够的。下面这些字段决定了数据能不能用：

字段	为什么要
展示价	基础数据
划线价 / 原价	判断是否真降价，而不是营销话术
优惠券门槛	券后价才是用户实际支付价格
库存状态	有价无货的降价没有意义
促销标签	区分日常价与活动价
抓取时间戳	价格是时间序列，没有时间戳的数据无法分析

其中优惠券门槛最容易被漏掉，但它经常是价格差异的主要来源——两个城市展示价相同，到手价可能差几十元。

三、频率控制：不是越勤越好

监控频率需要和业务决策周期匹配。日常比价的决策周期是天级，那么每小时抓一次已经远超需求，纯属浪费流量和暴露风险。

合理的起点：

• 日常监控：每天 1–2 次，避开目标站点的流量高峰
• 大促期间：每小时 1 次，并按商品重要性分级，只对核心 SKU 提高频率
• 价格异常告警：发现波动超过阈值时才加密抓取

同时把这三个参数固定下来，避免形成可识别的流量模式：

import random, time

def polite_delay(base: float = 3.0):
    # 带长尾的随机间隔，比固定 sleep 更接近真人
    time.sleep(base + random.lognormvariate(0, 0.7))

# 单出口并发不用高，靠城市点位数量横向扩展
CONCURRENT_PER_SESSION = 2

扩展吞吐的正确方向是增加城市点位而不是提高单点并发。前者带来更多有价值的数据维度，后者只带来更高的被封概率。

四、合规边界

这部分必须写清楚，因为价格监控最容易越界。

可以做的：抓取公开的商品价格、公开的促销信息、公开的库存状态，用于自己的选品和定价决策。

不要做的：

• 抓取需要登录才能看到的内容，尤其是他人的账号数据
• 绕过明确的技术保护措施（比如破解验证码、绕过付费墙）
• 采集任何个人数据（买家信息、评价者信息、卖家联系方式）
• 用采集到的数据对目标平台进行攻击性操作（比如刷单、恶意比价干扰）
• 忽视目标站点的 robots.txt 与服务条款

不同国家和地区对数据抓取的法律规定差异很大，涉及商业决策时应当咨询专业法律意见。技术上的可行性不等于法律上的合规性——这是两件必须分开判断的事。

五、一个可用的监控骨架

import time, random, json
from datetime import datetime, timezone
import requests

def snapshot(sku: str, city: str, session_id: str) -> dict:
    proxies = {"http": build_proxy(session_id), "https": build_proxy(session_id)}
    url = f"https://shop.example.com/item/{sku}"
    r = requests.get(url, proxies=proxies, timeout=(10, 45),
                     headers={"Accept-Language": "zh-CN,zh;q=0.9"})
    r.raise_for_status()

    if "验证码" in r.text[:3000]:
        raise RuntimeError("challenged")   # 状态码 200 但被风控，必须区分

    return {
        "sku": sku,
        "city": city,
        "price": parse_price(r.text),
        "list_price": parse_list_price(r.text),
        "in_stock": parse_stock(r.text),
        "captured_at": datetime.now(timezone.utc).isoformat(),
    }

def run(skus, cities):
    rows = []
    for sku in skus:
        for city in cities:
            for attempt in range(3):
                try:
                    rows.append(snapshot(sku, city, f"{city}-01"))
                    break
                except Exception as e:
                    if attempt == 2:
                        rows.append({"sku": sku, "city": city, "error": str(e)})
                    time.sleep(2 ** attempt + random.uniform(0, 1))
            polite_delay()
    return rows

三个设计决策值得说明：失败重试带指数退避、challenged 单独识别而不是当成功、以及每次抓取都带 UTC 时间戳。

小结

价格监控的成败不取决于抓取代码，而取决于三件事：出口是否在正确的城市、字段是否完整到能算出真实到手价、频率是否与决策周期匹配。把城市点位当成横向扩展的单位，而不是把单 IP 并发往上堆，是这套系统能长期跑下去的关键。