如何建立竞争定价管道：每天跟踪8个竞争对手的5,000个SKU

Ethan Brown

Advanced Bot Mitigation Engineer

28-May-2026

主要收获：

竞争定价是一个篮子问题，而非产品问题。 定价团队在4个市场中追踪8个竞争对手的5000个SKU，每天进行160,000次读取。可扩展的架构是每个URL一个渲染调用，并在每个市场上固定出口，加上单一的标准输出模式——而不是160,000个临时获取。
**市场决定出口。**价格、货币和可用性因地区和IP信誉而异。将代理国家固定在测量市场上，可以保持每个记录价格的可比性；在同一个SKU上混合美国和欧盟的出口会产生毫无意义的价格历史。
在竞争对手之间使用一个规范化模式。 每个零售商的DOM都不同；仓库表格却相同。提取时规范化：{your_sku, competitor, market, price_value, price_currency, availability, promo_state, captured_at}。决策依赖于仓库，而非原始HTML。
反检测在服务器端处理。 每个请求在Scrapeless云中使用住宅出口、JavaScript执行和指纹随机化进行渲染。管道发送一个URL和一个国家；它会返回渲染后的HTML。没有浏览器二进制，没有代理轮换逻辑，也没有第三方CDP客户端在你的机器上。
管道的结束是差异，而不是HTML。 原始渲染页面是临时存储。定价团队依据的信号是每个市场、每个SKU的你和竞争对手之间的价格差——展示在重新定价规则、Slack提醒或分析仪表盘上。
免费开始。 新的Scrapeless账户包括免费运行时——在 app.scrapeless.com 注册。

引言：从网络数据到竞争定价决策

竞争定价团队面临多年的同一限制：价格变化的速度快于用于指导定价决策的数据源。零售商在一夜之间修改了标签价格；商务智能面板在48小时后更新；当分析师看到差距时，促销窗口已关闭。网络数据可以闭合这个循环，但前提是采集层能够跟上变化的速度，并提供一个可以与仓库连接的模式。

结构性挑战并不是“抓取产品页面”。而是在一个SKU篮子、一个竞争对手篮子和一个市场篮子中运营一个抓取系统——每天、每个市场、每个零售商，并确保相同的准确性保证。每个零售商的DOM会发生变化。每个市场的价格会本地化。每个请求都需要越过零售商的反机器人层并返回干净的、渲染的HTML。Octoparse OptiGroup案例研究在大规模中捕捉到了相同的模式：50个子公司、数十个竞争对手网站、区域价格和一个集中定价决策层。

本指南将介绍基于Scrapeless的定价智能管道的采集层的架构和Python代码。输出是一个归一化的NDJSON流，馈送到仓库表中；输入是分析师定义的篮子文件。可以阅读一次获得模式；通过改变每个零售商的提取器重复使用。

你能做什么

每日竞争篮子读取。 每天在一个固定的运行时间内跟踪4个市场中8个竞争对手的5000个SKU，并采用单一的标准模式。
市场特定的重新定价。 将出口国家固定在每个市场；获取反映本地购物者实际看到的、而不是地理回退价的本地化价格。
促销状态监控。 捕捉列出的价格和促销状态（打折、折扣百分比、时限徽章），以便仓库能够区分日常价格和清仓促销。
MAP合规审计。 将零售商列出的价格与您的最低广告价格政策进行比较，并向渠道管理团队报告违规情况。
新产品发布跟踪。 关注竞争对手在某个类别的SKU首次出现；管道同时作为“竞争对手是否即将推出X？”的信号。
价格弹性数据集。 在90天期间每天快照生成的时间序列，供收益管理计算SKU级别的弹性使用。

在Scrapeless，我们只访问公开可用的数据，同时严格遵守适用的法律、法规和网站隐私政策。本文中的内容仅用于演示目的。

为什么选择Scrapeless用于竞争定价

Scrapeless在反检测云浏览器中渲染每个目标URL，使用自主开发的Chromium，并通过单一API调用返回完成的HTML。特别是对于定价智能管道，它提供：

195多个国家的住宅代理，每次请求时与国家代码进行匹配——出口地理信息是每个市场的一个字段。
云端JavaScript渲染。 零售商的产品页面是React或Next.js应用；价格元素在初始加载后出现。js_render=True意味着您的管道读取的是后绘制的DOM，而不是SSR外壳。
服务器端反检测。 UA、时区、WebGL、画布和无头标志在云端根据请求随机化。无需本地隐匿插件维护，无需安装浏览器二进制文件。
无状态请求形状。 每个产品页面都是独立的读取：发送一个 URL 和一个国家，就可以得到渲染的 HTML。这可以清晰地映射到成千上万独立 SKU 读取的篮子上。
整个管道一个 API 密钥。 渲染、住宅代理和 SDK 都在同一个 Scrapeless 账户下计费；没有每个层级的集成。

在 app.scrapeless.com 的免费计划中获取您的 API 密钥。

前提条件

Python 3.10 或更新版本
一个 Scrapeless 账户和 API 密钥 — 在 app.scrapeless.com 注册
熟悉 requests 风格的 HTTP 以及 CSS 选择器库
一个目标竞争对手列表和一个 SKU 篮子文件

管道架构一瞥

Copy

basket.yaml                      (分析师定义的输入)
       │
       ▼
┌──────────────────┐
│   orchestrator   │  每个（市场、竞争对手、SKU）一个任务；有界并发
└──────┬───────────┘
       │
       ▼
┌──────────────────┐
│   Scrapeless     │  client.universal.scrape(url, country) — 住宅出站，
│  (云渲染)        │  JS 渲染，反检测，全部服务器端
└──────┬───────────┘
       │  渲染的 HTML
       ▼
┌──────────────────┐
│   normalizer     │  每个零售商提取器 → 标准模式
└──────┬───────────┘
       │
       ▼
prices.ndjson          (每个（产品、竞争对手、市场、日期）一行)
       │
       ▼
仓库加载 + 与您的价格对比 + 警报

每个阶段都是一个 Python 模块；下面的七个步骤从底部构建它。

第 1 步 — 安装 Scrapeless SDK

bash Copy

pip install scrapeless lxml pyyaml

scrapeless 是官方的 Python SDK；它在云端渲染页面并返回 HTML，因此无需安装浏览器二进制文件和第三方自动化库。lxml 是解析器；pyyaml 用于读取篮子配置。

第 2 步 — 定义篮子

定价团队拥有此文件。保持简单 — 市场、竞争对手、SKU 映射。每个（您的 SKU、竞争对手、竞争对手网址、市场）一行：

yaml Copy

# basket.yaml
markets:
  - US
  - GB
  - DE
  - JP

basket:
  - your_sku: SKU-1001
    name: "Acme Widget Pro"
    competitors:
      - retailer: target_competitor_a
        url:
          US: "https://competitor-a.com/p/widget-pro"
          GB: "https://competitor-a.co.uk/p/widget-pro"
          DE: "https://competitor-a.de/p/widget-pro"
          JP: "https://competitor-a.co.jp/p/widget-pro"
      - retailer: target_competitor_b
        url:
          US: "https://competitor-b.com/products/widget-pro"
          GB: "https://competitor-b.co.uk/products/widget-pro"

一个 5,000 SKU 的篮子保持相同的形状；仓库在 your_sku 上连接，以便与您自己的价格数据对齐。

第 3 步 — 通过 Scrapeless 渲染产品页面

每个（市场、SKU）一个渲染调用。国家代码确定住宅出站；js_render=True 返回后互作的 DOM：

python Copy

import os
from scrapeless import Scrapeless
from scrapeless.types.universal import (
    UniversalScrapingRequest, UniversalJsRenderInput, UniversalProxy,
)

client = Scrapeless()  # 从环境中读取 SCRAPELESS_API_KEY

def scrape_rendered(url: str, market: str) -> str:
    """在 Scrapeless 云中渲染一个产品页面并返回 HTML。"""
    request = UniversalScrapingRequest(
        actor="unlocker.webunlocker",
        input=UniversalJsRenderInput(url=url, js_render=True, headless=True),
        proxy=UniversalProxy(country=market),
    )
    return client.universal.scrape(request)  # 返回渲染的 HTML（str）

国家代码是承重字段。同一个产品 URL 在不同地区会渲染不同的价格、货币和可用状态，因此固定出站可以确保每个记录的价格属于同一个市场。js_render=True 等待页面绘制完再返回，因此 React/Vue/Next.js 的零售商返回价格元素，而不是空壳。

第 4 步 — 遍历篮子

每个 SKU 是一个独立的渲染调用，因此篮子遍历是一个普通循环（或有界线程池以实现并行性）。无需保持会话，无需预热主页 — 云渲染在每次请求时清除零售商的反机器人层：

python Copy

import yaml

def load_basket(path: str = "basket.yaml") -> dict:
    with open(path, encoding="utf-8") as f:
        return yaml.safe_load(f)

def walk_basket(basket: dict):
    """为每个篮子条目生成（your_sku、零售商、市场、url、html）。"""
    for item in basket["basket"]:
        for comp in item["competitors"]:
            for market, url in comp["url"].items():
                html = scrape_rendered(url, market)
                yield item["your_sku"], comp["retailer"], market, url, html

对于一个有5,000个SKU的购物篮，将scrape_rendered包装在concurrent.futures.ThreadPoolExecutor中，并将工作者数量限制在帐户计划允许的水平。每个调用都是无状态的，因此并行度通过添加工作者进行扩展——没有共享会话需要争用。

在免费计划中获取您的API密钥：app.scrapeless.com

第5步 — 提取到规范模式中

每个零售商的DOM不同；仓库表则相同。提取器的工作是将零售商呈现的内容每次转换为相同的形状。输出模式（每（your_sku，competitor，market，captured_at）一行）：

python Copy

from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Optional
from lxml import html as lxml_html

@dataclass
class PriceRecord:
    your_sku: str
    competitor: str
    market: str
    url: str
    price_value: Optional[float]
    price_currency: Optional[str]
    availability: Optional[str]       # "in_stock" | "out_of_stock" | "preorder" | None
    promo_state: Optional[str]        # "none" | "on_sale" | "clearance" | None
    promo_discount_pct: Optional[float]
    captured_at: str                  # ISO-8601 UTC

每个零售商的提取器插入相同的返回类型：

python Copy

def extract_competitor_a(html: str, your_sku: str, market: str, url: str) -> PriceRecord:
    doc = lxml_html.fromstring(html)

    price_el = doc.cssselect("[data-test='price'] .value")
    currency_el = doc.cssselect("[data-test='price'] .currency")
    availability_el = doc.cssselect("[data-test='availability']")
    promo_el = doc.cssselect("[data-test='promo-badge']")

    availability = (
        "in_stock" if availability_el and "In stock" in availability_el[0].text_content()
        else "out_of_stock" if availability_el
        else None
    )

    return PriceRecord(
        your_sku=your_sku,
        competitor="target_competitor_a",
        market=market,
        url=url,
        price_value=_to_float(price_el[0].text_content()) if price_el else None,
        price_currency=currency_el[0].text_content().strip() if currency_el else None,
        availability=availability,
        promo_state="on_sale" if promo_el else "none",
        promo_discount_pct=_to_float(promo_el[0].get("data-discount-pct")) if promo_el else None,
        captured_at=datetime.now(timezone.utc).isoformat(),
    )

def _to_float(text) -> Optional[float]:
    if not text:
        return None
    cleaned = "".join(c for c in text if c.isdigit() or c == ".")
    try:
        return float(cleaned)
    except (ValueError, TypeError):
        return None

每个零售商都有自己的extract_<name> 函数；每个函数返回相同的 PriceRecord。协调者不知道每个零售商使用的DOM——只知道要调用的函数名称。

选择器设计注意事项：

优先使用[data-test='...']属性，当零售商暴露这些属性时。 它们能适应外观类名的旋转；如.text-lg.font-semibold每次发布时都会变化。
将缺失字段视为可为空。 对于缺货产品，价格为None是数据，而不是失败。
捕获零售商呈现的货币字符串。 不要从市场推断货币——某些零售商在其.de域名上列出美元（USD）用于跨境产品。存储页面所显示的内容。

第6步 — 流式写入NDJSON以供仓库加载

流式写入NDJSON，以便管道在运行过程中遭遇中断时不会丢失记录。每一行是一个呈现的SKU；该文件是只附加的：

python Copy

import json
from pathlib import Path

def append_records(records: list[PriceRecord], out_path: str = "prices.ndjson"):
    Path(out_path).parent.mkdir(parents=True, exist_ok=True)
    with open(out_path, "a", encoding="utf-8") as f:
        for r in records:
            f.write(json.dumps(asdict(r)) + "\n")

NDJSON可以直接加载到Snowflake（COPY INTO ... FILE_FORMAT = (TYPE = JSON)）、BigQuery（bq load --source_format=NEWLINE_DELIMITED_JSON）、Redshift、ClickHouse和DuckDB。选择BI堆栈已经使用的那个；模式是相同的。

第7步 — 计算差异并路线定价决策

定价团队依据的信号不是原始价格——而是对手价格与您价格之间的差异，按市场、按SKU分类。差异存在于仓库中，而不在抓取器中：

sql Copy

-- 每SKU每竞争对手每市场的日价格差
WITH yours AS (
  SELECT sku, market, list_price, currency, captured_date
  FROM your_internal_prices
  WHERE captured_date = CURRENT_DATE
),
theirs AS (
  SELECT your_sku, competitor, market, price_value, price_currency,
         availability, promo_state, CAST(captured_at AS DATE) AS captured_date
  FROM competitor_prices
  WHERE CAST(captured_at AS DATE) = CURRENT_DATE
)
SELECT
  t.your_sku,
  t.competitor,
  t.market,
  y.list_price                                  AS our_price,
  t.price_value                                 AS their_price,

ROUND(100.0 * (y.list_price - t.price_value) / NULLIF(t.price_value, 0), 2) AS price_gap_pct,
t.availability,
t.promo_state
FROM theirs t
LEFT JOIN yours y
ON y.sku = t.your_sku AND y.market = t.market
WHERE y.list_price IS NOT NULL
AND t.price_value IS NOT NULL
ORDER BY price_gap_pct DESC;

根据定价规则定义的阈值对price_gap_pct超出情况进行路由：

高于你的价格阈值（例如：你比领先者贵5%+）→ 重新定价审核。
低于最低广告价格（MAP）阈值 → 向渠道管理发送MAP违规警报。
自昨天以来促销状态发生变化 → 向类别管理者发送竞争促销通知。

差异查询是集合与决策之间的合同。只要仓库架构保持稳定，定价团队的下游BI图块、警报和定价规则在零售商轮换其DOM时从不改变——只有步骤5中的每个零售商提取器会有所更改。

你将得到的结果

每个（your_sku，competitor，market，day）一个NDJSON行，格式如下：

json Copy

{
  "your_sku": "SKU-1001",
  "competitor": "target_competitor_a",
  "market": "US",
  "url": "https://competitor-a.com/p/widget-pro",
  "price_value": 79.99,
  "price_currency": "USD",
  "availability": "in_stock",
  "promo_state": "on_sale",
  "promo_discount_pct": 15.0,
  "captured_at": "<ISO-8601 UTC timestamp written at read time>"
}

运行该模式的诚实观察：

渲染时机比DOM的具体性更重要。 在SSR外壳上运行的选择器在价格元素绘制之前返回空字符串。js_render=True返回后期氢化的DOM，这是使价格选择器解析的原因。
货币与市场并不冗余。 跨境SKU有时在本地域名上列出非本地货币。存储渲染的字符串；让仓库层进行规范化。
促销状态至少有三个值，而不是两个。 none、on_sale和clearance在重新定价规则中表现不同——清仓降价表示生命周期结束，而不是促销推动。
可用性是第二个可操作性字段。 在缺货SKU上的20%价格差距与在有货SKU上的相同差距所传递的竞争信号不同。将两者都展示给决策层。
一个规范架构是承载决策的基础。 每个零售商的字段、货币约定和促销格式各不相同；但仓库表不变。将变动推入提取函数，保持架构平坦。

结论：扩展你的竞争定价管道

该管道简化为六个步骤：定义篮子 → 经由Scrapeless渲染每个SKU并按市场固定出口 → 提取到规范架构 → 流式传输到NDJSON → 加载仓库 → 与你自己的价格进行差异比较。每个步骤都足够小，便于阅读；该组合处理在单个每日计划下跨8个竞争对手和4个市场的5000个SKU。

对于定价相关抓取的供应商比较视图（特别是房地产定价），2026最佳Zillow抓取工具的列表将八种工具与同类本地价格提取挑战进行排名。有关将NDJSON输出加载到云仓库的信息，Scrapeless + Snowflake数据摄取指南介绍了COPY INTO和流式传输路径。

为每个市场固定出口国家，独立渲染每个SKU，在提取时规范化，存储每个SKU/竞争对手/市场/日期的一行规范行，并在仓库中进行差异比较——而不是在抓取器中。

准备好建立您的人工智能驱动数据管道吗？

加入我们的社区以申请免费计划，并与构建竞争定价管道的开发者联系：Discord · Telegram。

在app.scrapeless.com注册以获得免费的运行时，并根据定价管道所需的市场、竞争对手和SKU篮子调整上述模式。有关定价详情，请查看 scrapeless.com/en/pricing；代理解决方案产品页面见 scrapeless.com/en/product/proxy-solutions；完整SDK参考见 docs.scrapeless.com。

常见问题

Q1: 抓取竞争对手的价格是否合法？

定价在零售商的产品页面上是公开信息，价格比较是一种成熟的商业实践。合法性取决于你抓取的内容、来源及其条款。公开可见的数据通常是可以访问的；网站使用条款、地区隐私法（如GDPR、CCPA）和版权法适用。对于高风险的使用案例，请咨询法律顾问。Scrapeless仅访问公开可用的数据。

Q2: 我需要代理进行竞争定价吗？
是的，国家PIN比IP轮换更重要。零售商根据市场本地化定价；向.co.uk域的美国出口请求可能返回备用价格、重定向或地理封锁。通过UniversalProxy(country=...)将国家固定到测量市场。无卡式住宅代理在195多个国家覆盖典型的定价篮子，而不需要将单独的代理提供商引入堆栈。

Q3：我该如何处理反机器人挑战和机器人检测？

渲染在Scrapeless云的服务器端运行，使用住宅出口、真实的JavaScript执行和随机指纹识别，因此到达零售商的请求看起来像是来自目标市场的普通浏览器的住宅IP。设置js_render=True，以便响应是经过水合的DOM，而不是预渲染的外壳，并将国家固定到您测量的市场。

Q4：管道应该多久运行一次？

每天是重新定价决策的标准频率；每小时对于价格在一天内变化的促销窗口监控是现实的。每个SKU的成本由单次渲染调用限制，因此5,000个SKU每天的频率完全在单周期商店预算之内。更高的频率线性增加成本——选择定价决策实际消耗的频率。

Q5：当零售商更换其DOM时会发生什么？

步骤5中的每个零售商提取器是唯一会更改的文件。标准模式、仓库表、商业智能模块、差异查询和警报规则都不受影响。当零售商发布新版本时，再次检查选择器；在可能的情况下更喜欢[data-test='...']属性；将提取器视为不稳定层，将模式视为稳定层。

Q6：我可以并行运行多个零售商吗？

可以。每个渲染调用是无状态的，因此调度程序会在线程池中扩展（市场、竞争对手、SKU）任务，并将工人数量限制在账户计划允许的级别。并行性通过增加工人而不是共享会话来扩展——没有需要争用的连接。

Q7：我该如何捕捉促销状态和折扣百分比？

步骤5提取器直接从渲染的DOM读取促销徽章，并将promo_state（"on_sale"、"clearance"、"none"）和promo_discount_pct存储为单独字段。仓库将两者合并到差异查询中，以便定价规则可以基于“竞争对手是否在促销？”与“竞争对手的常规价格是多少？”进行分支。

Q8：国际货币和外汇呢？

为每个记录存储渲染的货币字符串（USD、EUR、JPY、GBP）。货币转换属于仓库层，而不是抓取器——将原始价格 + 原始货币 + 市场保留在NDJSON中，并在商业智能方面每天运行外汇交叉连接。这样，即使一个糟糕的外汇汇率也不会影响整个历史记录。

在Scrapeless，我们仅访问公开可用的数据，并严格遵循适用的法律、法规和网站隐私政策。本博客中的内容仅供演示之用，不涉及任何非法或侵权活动。我们对使用本博客或第三方链接中的信息不做任何保证，并免除所有责任。在进行任何抓取活动之前，请咨询您的法律顾问，并审查目标网站的服务条款或获取必要的许可。