如何使用 Shopping SERP 数据构建价格监控系统

了解如何使用 Shopping SERP 数据构建价格监控系统，通过 Python 采集商品标题、价格、商家、评分、运费信号，并导出价格快照为 CSV。

talor ai
最後更新於
7 分鐘閱讀

价格监控听起来很简单，真正做起来就会发现不是这样。

商品价格不只是一个数字。它可能会因国家、商家、库存、运费、促销、市场平台、设备，甚至搜索词写法而变化。

如果你在做电商，或者需要追踪竞品价格，手动查价格很快就会撑不住。几个商品还勉强可以，几百个商品、几个市场一起看，就不现实了。

这时候 Shopping SERP 数据就很有用。你不需要每天手动打开购物搜索结果，而是可以采集结构化商品结果，整理字段，并把每次查询保存成价格快照。

这篇文章会用 Python 做一个简单的价格监控流程。示例中使用 Talordata 风格的 SERP API 密钥和请求方式，因为 Talordata SERP API 支持结构化搜索数据、JSON / HTML 输出、地理定位，也覆盖商品结果、价格、评论和广告等电商情报场景。

我们要构建什么？

流程很直接：

商品关键词列表
→ Shopping SERP API 请求
→ 提取商品结果
→ 整理价格、商家、评分、运费
→ 导出价格快照到 CSV

第一版不需要做成完整 SaaS。先能回答几个实际问题就够了：

  • 这个商品查询下出现了哪些商家？

  • 可见的最低价是多少？

  • 我们的价格是高于还是低于市场区间？

  • 竞品是否频繁改价？

  • 哪些商品有折扣或运费信号？

  • 哪些品牌在 Shopping 结果中更常出现？

应该采集哪些数据？

一条有用的价格监控数据，不应该只有价格。

字段

为什么重要

query

被追踪的商品关键词

position

商品在 Shopping 结果中的位置

title

结果中显示的商品标题

seller

商店、商家或来源

price

显示价格

currency

从价格中提取的币种

rating

商品或商家评分，如果有

reviews

评论数，如果有

shipping

运费或配送信号

product_url

商品或报价 URL

thumbnail_url

商品缩略图

collected_at

采集时间

时间戳很重要。没有时间戳，你只有一份价格列表；有了时间戳，你才有价格历史。

查看完整的API文档>>

Step 1：设置环境变量

不要把 API 密钥直接写进代码。

export TALORDATA_API_KEY="your_api_key_here"
export TALORDATA_SERP_ENDPOINT="your_talordata_serp_endpoint_here"

Windows PowerShell：

$env:TALORDATA_API_KEY="your_api_key_here"
$env:TALORDATA_SERP_ENDPOINT="your_talordata_serp_endpoint_here"

安装 Python 依赖：

pip install requests

请使用后台或 API 文档中显示的端点和验证方式。

Step 2：准备商品查询词

先从五到十个商品查询词测试。

PRODUCT_QUERIES = [
    "wireless noise cancelling headphones",
    "standing desk 48 inch",
    "portable espresso maker",
]

MARKET = {
    "location": "United States",
    "country": "us",
    "language": "en",
    "device": "desktop",
}

正式使用时，可以把商品查询词放在数据库或表格中，也可以加上内部商品 ID、品牌名、SKU 和目标价格区间。

Step 3：建立请求辅助函数

把请求层和解析逻辑分开。这样端点、验证方式或请求格式变化时，不需要改整篇程序。

import os
import requests


def require_env(name):
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


TALORDATA_API_KEY = require_env("TALORDATA_API_KEY")
TALORDATA_SERP_ENDPOINT = require_env("TALORDATA_SERP_ENDPOINT")


def call_serp_api(payload):
    headers = {
        "Authorization": f"Bearer {TALORDATA_API_KEY}",
        "Content-Type": "application/json",
    }

    response = requests.post(
        TALORDATA_SERP_ENDPOINT,
        json=payload,
        headers=headers,
        timeout=30,
    )

    response.raise_for_status()
    return response.json()

有些 API 使用 GET，也有些会把 API key 放在查询参数中。这些差异都放在这个辅助函数里处理。

Step 4：获取 Shopping SERP 结果

建立一个获取 Shopping 搜索结果的函数。

def fetch_shopping_results(query, market):
    payload = {
        "engine": "google_shopping",
        "q": query,
        "location": market["location"],
        "gl": market["country"],
        "hl": market["language"],
        "device": market["device"],
        "num": 20,
        "output": "json",
    }

    return call_serp_api(payload)

根据你的 API 设置，购物搜索类型可能使用 google_shopping、shopping 或其他参数格式。把这个函数独立出来，后续只需要改一处。

Step 5：整理商品结果

不同供应商的 Shopping 结果字段名称可能不同，所以解析器要保持弹性。

from datetime import datetime, timezone
from urllib.parse import urlparse
import re


def clean_text(value):
    if not value:
        return ""
    return " ".join(str(value).split())


def get_domain(url):
    if not url:
        return ""

    parsed = urlparse(url)
    return parsed.netloc.replace("www.", "")


def parse_price(value):
    if not value:
        return {"price_value": None, "currency": ""}

    text = str(value)
    currency = ""

    if "$" in text:
        currency = "USD"
    elif "€" in text:
        currency = "EUR"
    elif "£" in text:
        currency = "GBP"

    match = re.search(r"[\d,.]+", text)
    if not match:
        return {"price_value": None, "currency": currency}

    number = match.group(0).replace(",", "")

    try:
        return {
            "price_value": float(number),
            "currency": currency,
        }
    except ValueError:
        return {
            "price_value": None,
            "currency": currency,
        }


def get_shopping_results(serp_json):
    return (
        serp_json.get("shopping_results")
        or serp_json.get("shopping")
        or serp_json.get("product_results")
        or []
    )


def normalize_shopping_results(serp_json, query, market):
    results = get_shopping_results(serp_json)
    collected_at = datetime.now(timezone.utc).isoformat()

    rows = []

    for index, item in enumerate(results, start=1):
        raw_price = item.get("price") or item.get("extracted_price") or ""
        parsed_price = parse_price(raw_price)

        product_url = item.get("link") or item.get("product_link") or item.get("url") or ""

        rows.append({
            "query": query,
            "location": market["location"],
            "country": market["country"],
            "language": market["language"],
            "device": market["device"],
            "position": item.get("position") or item.get("rank") or index,
            "title": clean_text(item.get("title")),
            "seller": clean_text(item.get("source") or item.get("seller") or item.get("merchant")),
            "raw_price": clean_text(raw_price),
            "price_value": parsed_price["price_value"],
            "currency": parsed_price["currency"],
            "rating": item.get("rating") or "",
            "reviews": item.get("reviews") or item.get("reviews_count") or "",
            "shipping": clean_text(item.get("shipping") or item.get("delivery") or ""),
            "product_url": product_url,
            "product_domain": get_domain(product_url),
            "thumbnail_url": item.get("thumbnail") or item.get("thumbnail_url") or "",
            "collected_at": collected_at,
        })

    return rows

这里同时保存 raw_price 和 price_value。这是故意的。原始价格方便排查格式问题，数字价格方便做比较。

Step 6：导出价格快照到 CSV

import csv


CSV_COLUMNS = [
    "query",
    "location",
    "country",
    "language",
    "device",
    "position",
    "title",
    "seller",
    "raw_price",
    "price_value",
    "currency",
    "rating",
    "reviews",
    "shipping",
    "product_url",
    "product_domain",
    "thumbnail_url",
    "collected_at",
]


def export_to_csv(rows, filename="shopping_price_snapshots.csv"):
    if not rows:
        print("No shopping results found.")
        return

    with open(filename, mode="w", newline="", encoding="utf-8") as file:
        writer = csv.DictWriter(file, fieldnames=CSV_COLUMNS)
        writer.writeheader()
        writer.writerows(rows)

    print(f"Exported {len(rows)} price snapshots to {filename}")

完整 Python 示例

import os
import csv
import re
import requests

from datetime import datetime, timezone
from urllib.parse import urlparse


PRODUCT_QUERIES = [
    "wireless noise cancelling headphones",
    "standing desk 48 inch",
    "portable espresso maker",
]

MARKET = {
    "location": "United States",
    "country": "us",
    "language": "en",
    "device": "desktop",
}

CSV_COLUMNS = [
    "query",
    "location",
    "country",
    "language",
    "device",
    "position",
    "title",
    "seller",
    "raw_price",
    "price_value",
    "currency",
    "rating",
    "reviews",
    "shipping",
    "product_url",
    "product_domain",
    "thumbnail_url",
    "collected_at",
]


def require_env(name):
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


TALORDATA_API_KEY = require_env("TALORDATA_API_KEY")
TALORDATA_SERP_ENDPOINT = require_env("TALORDATA_SERP_ENDPOINT")


def call_serp_api(payload):
    headers = {
        "Authorization": f"Bearer {TALORDATA_API_KEY}",
        "Content-Type": "application/json",
    }

    response = requests.post(
        TALORDATA_SERP_ENDPOINT,
        json=payload,
        headers=headers,
        timeout=30,
    )

    response.raise_for_status()
    return response.json()


def fetch_shopping_results(query, market):
    payload = {
        "engine": "google_shopping",
        "q": query,
        "location": market["location"],
        "gl": market["country"],
        "hl": market["language"],
        "device": market["device"],
        "num": 20,
        "output": "json",
    }

    return call_serp_api(payload)


def clean_text(value):
    if not value:
        return ""
    return " ".join(str(value).split())


def get_domain(url):
    if not url:
        return ""

    parsed = urlparse(url)
    return parsed.netloc.replace("www.", "")


def parse_price(value):
    if not value:
        return {"price_value": None, "currency": ""}

    text = str(value)
    currency = ""

    if "$" in text:
        currency = "USD"
    elif "€" in text:
        currency = "EUR"
    elif "£" in text:
        currency = "GBP"

    match = re.search(r"[\d,.]+", text)
    if not match:
        return {"price_value": None, "currency": currency}

    number = match.group(0).replace(",", "")

    try:
        return {
            "price_value": float(number),
            "currency": currency,
        }
    except ValueError:
        return {
            "price_value": None,
            "currency": currency,
        }


def get_shopping_results(serp_json):
    return (
        serp_json.get("shopping_results")
        or serp_json.get("shopping")
        or serp_json.get("product_results")
        or []
    )


def normalize_shopping_results(serp_json, query, market):
    results = get_shopping_results(serp_json)
    collected_at = datetime.now(timezone.utc).isoformat()

    rows = []

    for index, item in enumerate(results, start=1):
        raw_price = item.get("price") or item.get("extracted_price") or ""
        parsed_price = parse_price(raw_price)

        product_url = item.get("link") or item.get("product_link") or item.get("url") or ""

        rows.append({
            "query": query,
            "location": market["location"],
            "country": market["country"],
            "language": market["language"],
            "device": market["device"],
            "position": item.get("position") or item.get("rank") or index,
            "title": clean_text(item.get("title")),
            "seller": clean_text(item.get("source") or item.get("seller") or item.get("merchant")),
            "raw_price": clean_text(raw_price),
            "price_value": parsed_price["price_value"],
            "currency": parsed_price["currency"],
            "rating": item.get("rating") or "",
            "reviews": item.get("reviews") or item.get("reviews_count") or "",
            "shipping": clean_text(item.get("shipping") or item.get("delivery") or ""),
            "product_url": product_url,
            "product_domain": get_domain(product_url),
            "thumbnail_url": item.get("thumbnail") or item.get("thumbnail_url") or "",
            "collected_at": collected_at,
        })

    return rows


def export_to_csv(rows, filename="shopping_price_snapshots.csv"):
    if not rows:
        print("No shopping results found.")
        return

    with open(filename, mode="w", newline="", encoding="utf-8") as file:
        writer = csv.DictWriter(file, fieldnames=CSV_COLUMNS)
        writer.writeheader()
        writer.writerows(rows)

    print(f"Exported {len(rows)} price snapshots to {filename}")


if __name__ == "__main__":
    all_rows = []

    for query in PRODUCT_QUERIES:
        serp_json = fetch_shopping_results(query, MARKET)
        rows = normalize_shopping_results(serp_json, query, MARKET)
        all_rows.extend(rows)

    export_to_csv(all_rows)

执行后会得到：

shopping_price_snapshots.csv

输出示例

query,location,country,language,device,position,title,seller,raw_price,price_value,currency,rating,reviews,shipping,product_url,product_domain,thumbnail_url,collected_at
wireless noise cancelling headphones,United States,us,en,desktop,1,Example Wireless Headphones,Example Store,$129.99,129.99,USD,4.6,2300,Free shipping,https://example.com/product,example.com,https://example.com/thumb.jpg,2026-06-22T00:00:00Z
standing desk 48 inch,United States,us,en,desktop,2,Adjustable Standing Desk,Desk Shop,$219.00,219.0,USD,4.4,810,Delivery available,https://deskshop.example/item,deskshop.example,https://deskshop.example/thumb.jpg,2026-06-22T00:00:00Z

如何使用这些数据？

第一版用 CSV 就够了。当这个流程每天跑起来后，你就可以开始回答更实际的问题：

问题

怎么看

谁的价格最低？

按 query 分组，再按 price_value 排序

价格是否在变？

比较今天和昨天的快照

哪些商家最常出现？

统计 seller 或 product_domain

我们是否高于市场价？

比较自身价格和最低价、中位数、最高价

竞品是否提供免运？

筛选并分组 shipping

如果要定期监控，建议把每次结果写入数据库，而不是覆盖 CSV。一张包含 query、seller、price_value、currency、position 和 collected_at 的表，就已经足够做趋势图。

Best Practices

不要只靠商品标题匹配。Shopping 结果中的商品标题差异很大。要做准确匹配，可以结合标题、商家、品牌、商品 URL，必要时再加入图片信号。

保留原始价格。币种符号、促销标签和价格区间都可能让解析变复杂。

固定追踪同一组查询词。如果每天换查询词，价格趋势会变得很乱。

保存国家、语言、设备和地区。Shopping 结果会因市场不同而变化。

分开看可见度和价格。第 30 名的低价商品，可能不如第 1 名稍贵一点的商品重要。

注意平台规则和数据使用合规。Shopping SERP 数据很适合监控，但你的使用方式仍应符合适用法律、平台条款和内部合规要求。Talordata SERP API FAQ 也提醒用户，需要确保自身使用方式符合适用法律和规则。

FAQ

什么是 Shopping SERP 数据？

Shopping SERP 数据是来自购物型搜索结果页的结构化信息，可能包括商品标题、商家、价格、评分、评论数、缩略图、运费信号和商品链接。

可以用 Python 构建价格监控系统吗？

可以。第一版用 Python 就足够了。你可以采集 Shopping SERP 数据，整理商品字段，解析价格，再导出价格快照到 CSV。

多久监控一次价格比较合适？

取决于商品类目。变化快的类目可以每天追踪；变化慢的类目，每周追踪可能就够了。

CSV 够用吗？

测试阶段够用。如果要做定期监控、提醒、看板和历史趋势，建议使用数据库。

这套流程可以追踪竞品吗？

可以。你可以按商家、域名、商品标题或品牌分组，观察哪些竞品经常出现，以及它们的可见价格如何变化。

立即開展您的數據業務

加入全球最強大的代理網絡

免費試用