How to Build a Price Monitoring System with Shopping SERP Data

Price monitoring sounds simple until you actually try to do it.

A product price is not just a number. It can change by country, seller, availability, shipping fee, promotion, marketplace, device, and even the exact wording of the query.

If you sell products online, or you track competitors for an e-commerce brand, checking prices manually will break very quickly. A small catalog is already annoying. A few hundred products across several markets becomes impossible.

That is where Shopping SERP data helps. Instead of visiting shopping result pages by hand, you can collect structured product results, normalize the fields, and save price snapshots over time.

In this tutorial, we will build a small Python-based price monitoring workflow. The examples use Talordata-style SERP API credentials because Talordata’s SERP API covers structured search data, JSON / HTML output, geo-targeting, and e-commerce intelligence use cases such as product results, pricing, reviews, and ads.

What We Will Build

The workflow is simple:

Product keyword list
→ Shopping SERP API request
→ Extract product results
→ Normalize price, seller, rating, shipping
→ Export price snapshots to CSV

The first version will not try to be a full SaaS product. It will just collect enough data to answer practical questions like:

• Which sellers are showing up for this product query?

• What is the lowest visible price?

• Is our product above or below the market range?

• Are competitors changing prices often?

• Which products show discounts or shipping signals?

• Which brands dominate Shopping results?

What Data Should We Collect?

A useful price monitoring row should include more than the price.

The timestamp is important. Without it, you only have a list of prices. With it, you have a price history.

View the complete API documentation>>

Step 1: Set Environment Variables

Do not hardcode your API key inside the script.

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

On Windows PowerShell:

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

Install the Python package:

pip install requests

Use the endpoint and authentication method shown in your dashboard or API documentation.

Step 2: Prepare Product Queries

Start small. A good first test is five to ten product queries.

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

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

For production, store product queries in a database or a spreadsheet. You may also want to add internal product IDs, brand names, SKUs, and target price ranges.

Step 3: Create a Request Helper

Keep the request layer separate from parsing. This makes the script easier to fix when the endpoint, authentication method, or request format changes.

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()

Some API setups use GET instead of POST, or pass the API key as a query parameter. Keep those details inside this helper.

Step 4: Fetch Shopping SERP Results

Now create a function for Shopping search results.

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)

Depending on your API setup, the shopping search type may use google_shopping, shopping, or another parameter format. Keep this function isolated so you only need to adjust one place.

Step 5: Normalize Product Results

Shopping results can have different field names across providers. A flexible parser keeps your workflow from breaking too easily.

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

The parser keeps both raw_price and price_value. That is intentional. The raw price helps debug formatting issues, while the numeric price helps with comparisons.

Step 6: Export Price Snapshots to 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}")

Complete Python Example

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)

After running the script, you should get:

shopping_price_snapshots.csv

Example Output

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

How to Use the Data

A CSV file is enough for the first version. Once the workflow runs daily, you can start asking better questions:

For recurring monitoring, write each run to a database instead of overwriting the CSV. A simple table with query, seller, price_value, currency, position, and collected_at is already enough for trend charts.

Best Practices

Do not rely on title matching alone. Product titles in Shopping results can vary a lot. If you need accurate matching, combine title, seller, brand, product URL, and sometimes image signals.

Keep raw prices. Currency symbols, sale labels, and range prices can make parsing messy.

Track the same queries consistently. If the query changes every day, price trends become noisy.

Store country, language, device, and location. Shopping results can change by market.

Separate visibility from pricing. A cheap product in position 30 may matter less than a slightly more expensive product in position 1.

Respect marketplace and data usage rules. Shopping SERP data is useful for monitoring, but your own use should still follow applicable laws, platform terms, and internal compliance requirements. Talordata’s SERP API FAQ also reminds users to ensure their usage complies with applicable laws and rules.

FAQ

What is Shopping SERP data?

Shopping SERP data is structured information from shopping-style search result pages. It can include product titles, sellers, prices, ratings, reviews, thumbnails, shipping signals, and product links.

Can I build a price monitoring system with Python?

Yes. Python is enough for a first version. You can collect Shopping SERP data, normalize product fields, parse prices, and export price snapshots to CSV.

How often should I monitor prices?

It depends on the product category. For fast-moving categories, daily tracking may be useful. For slower categories, weekly tracking may be enough.

Is CSV enough for price monitoring?

CSV is fine for testing. For recurring monitoring, alerts, dashboards, and historical trends, use a database.

Can this workflow track competitors?

Yes. You can group results by seller, domain, product title, or brand to see which competitors appear often and how their visible prices change over time.