如何使用 Shopping SERP 資料建立價格監控系統

了解如何使用 Shopping SERP 資料建立價格監控系統，透過 Python 收集商品標題、價格、商家、評分、運費訊號，並匯出價格快照為 CSV。

Marcus Bennett

最後更新於

2026-06-22

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 夠用嗎？

測試階段夠用。如果要做定期監控、提醒、看板和歷史趨勢，建議使用資料庫。

這套流程可以追蹤競品嗎？

可以。你可以按商家、網域、商品標題或品牌分組，觀察哪些競品經常出現，以及它們的可見價格如何變化。