Building an Amazon Price Tracker in Python

A price tracker is a small system with four parts: fetch, parse, store, alert. Each has a failure mode that only shows up after a few weeks of running, which is why the storage and validation parts matter more than they look.

This build targets a single marketplace at a time. The design assumes the per-marketplace rules in Collecting Amazon Product and Review Data.

What you need

• Python with httpx for requests and selectolax or beautifulsoup4 for parsing
• A database or a file store for price history (SQLite is enough to start)
• One proxy address per marketplace, held stable

For the proxy, a fixed endpoint per marketplace is simpler than a pool and avoids the address changing mid-run. The Proxy Checker confirms what you are actually exiting from.

Step 1: Fetch with a market address

Configure the proxy per marketplace rather than globally.

import os
import httpx

MARKETS = {
    "de": {
        "proxy": os.environ["PROXY_DE"],
        "accept_language": "de-DE,de;q=0.9",
        "currency": "EUR",
    },
    "pl": {
        "proxy": os.environ["PROXY_PL"],
        "accept_language": "pl-PL,pl;q=0.9",
        "currency": "PLN",
    },
}

def fetch(url: str, market: str) -> str:
    cfg = MARKETS[market]
    headers = {
        "Accept-Language": cfg["accept_language"],
        "User-Agent": os.environ["USER_AGENT"],
    }
    with httpx.Client(proxy=cfg["proxy"], timeout=20.0, follow_redirects=True) as client:
        response = client.get(url, headers=headers)
        response.raise_for_status()
        return response.text

Three details are deliberate. The language header matches the address, so locale and network agree. The user agent is fixed rather than randomised per request, because a device that changes identity between requests is a stronger signal than a consistent one. The timeout is finite, so one slow request cannot stall the run.

Step 2: Parse defensively

Parse into a value or an explicit failure. A tracker that writes None into a price column and continues is worse than one that stops.

import re
from selectolax.parser import HTMLParser

PRICE_RE = re.compile(r"(\d[\d\s.,]*)\s*([A-Z]{3})?")

def parse_price_and_availability(html: str) -> dict:
    tree = HTMLParser(html)
    price_el = tree.css_first("#corePrice_feature_div .a-offscreen")
    if price_el is None:
        raise ValueError("price element missing")

    raw = price_el.text(strip=True)
    match = PRICE_RE.search(raw)
    if match is None:
        raise ValueError(f"unparseable price: {raw!r}")

    digits = match.group(1).replace(" ", "").replace(",", "")
    if digits.count(".") > 1:
        digits = digits.replace(".", "", digits.count(".") - 1)

    availability = tree.css_first("#availability")
    return {
        "price": float(digits),
        "availability": availability.text(strip=True) if availability else "unknown",
    }

Three things to note. The selector targets the off-screen price element, which is stable and machine-readable. A missing selector raises rather than returning null, so a layout change is visible on the run it happens. And the number parsing handles both comma and dot decimal conventions, because marketplaces differ.

Selectors change. Expect to maintain them, and validate against a schema on every run so the change is caught immediately.

Step 3: Store history, not state

The table that makes the tracker useful:

CREATE TABLE price_history (
  asin       TEXT NOT NULL,
  market     TEXT NOT NULL,
  currency   TEXT NOT NULL,
  price      REAL NOT NULL,
  available  INTEGER NOT NULL,
  observed_at TEXT NOT NULL,
  PRIMARY KEY (asin, market, observed_at)
);

Storing every observation rather than only the latest value is what lets you answer the questions people actually ask: when did it change, how often, and how deep were the drops. It also detects the failure where a scraper keeps running but returns stale or unchanged data, which is one of the health metrics in Monitoring Scraper Health.

Step 4: Alert on change, not on value

Alerts should fire on transitions, and be rare enough that each one is read.

def detect_change(previous: dict | None, current: dict) -> list[str]:
    events = []
    if previous is None:
        return events
    if previous["price"] != current["price"]:
        delta = current["price"] - previous["price"]
        events.append(f"price {delta:+.2f} {current['currency']} to {current['price']:.2f}")
    if previous["available"] != current["available"]:
        state = "back in stock" if current["available"] else "out of stock"
        events.append(state)
    return events

A threshold is usually worth adding so noise does not train people to ignore the channel. Alerting on every small variation of a volatile product produces more messages than anyone reads.

Step 5: Schedule and guard

Run on a schedule matched to the category's volatility, not to convenience. A daily run is reasonable for most products; an hourly run is only justified where prices move hourly.

Guard the run with three checks:

• Block rate. If a run gets more 403s or challenge pages than usual, slow down rather than retrying harder. The pattern is in Rate Limiting vs Blocking.
• Record count. A run that returns fewer products than expected is a signal, and the missing products are usually the ones where the parse failed.
• Address. Confirm periodically that the exit address is still in the expected country, since a silently changed address means the prices are from the wrong market.

For scheduling and isolation, see Scheduling Scrapers with CI and Docker for Scrapers.

Terms

Amazon's conditions of use restrict automated collection. Keep request rates low, collect only what you have a basis for, and prefer the official Product Advertising API where it covers your use. The framework is in Data Collection Ethics for Engineering Teams.

For the tooling comparison, see Price Monitoring Tools Compared, and for the solution context, E-Commerce.