Build an Amazon price tracker by ASIN in Python

A price tracker needs three things: the current price of each product, a record of what it cost before, and something that runs the check on a schedule. This guide builds all three in one Python file, using Amazon's batch product endpoint and the SQLite module that ships with Python.

You will see what a batch call returns, why Amazon prices need careful parsing, the full tracker script, how to run it from cron and what that costs. The responses shown are real, trimmed with … where they run long.

Check 20 prices in one call

Every Amazon call is a POST to https://scrapingbot.io/api/v1/amazon with an endpoint name and its parameters. For price tracking, use /products: it takes up to 20 ASINs and returns the full product object for each one, the same object a single /product-details call returns.

curl -X POST https://scrapingbot.io/api/v1/amazon \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"endpoint": "/products", "params": {"asins": ["B07FDJMC9Q", "B0BSHF7WHW"]}}'

Here is the response for an air fryer and a laptop, with most product fields left out:

{
  "success": true,
  "status": "OK",
  "request_id": "e6c252d9-f2f4-4c90-99db-022c7084fa4f",
  "parameters": { "asins": ["B07FDJMC9Q", "B0BSHF7WHW"], "country": "US" },
  "data": {
    "items": [
      {
        "asin": "B07FDJMC9Q",
        "product_title": "Ninja AF101 Air Fryer, 4 Qt Capacity, 4-in-1 Crisp, Roast, Reheat…",
        "product_price": "79.99",
        "product_original_price": "$119.99",
        "currency": "USD",
        "product_availability": "In Stock",
        "sales_volume": "2K+ bought in past month",
        …
      },
      {
        "asin": "B0BSHF7WHW",
        "product_title": "Apple 2023 MacBook Pro Laptop with Apple M2 Pro chip…",
        "product_price": null,
        "product_availability": "Currently unavailable. We don't know when or if this item will be back in stock.",
        …
      }
    ],
    "errors": []
  },
  "duration": "2.75",
  "statusCode": 200,
  "creditsUsed": 20
}

Two products came back, so the call cost 20 credits. Billing works per product: when the call starts, 10 credits are held for every ASIN you send, and any ASIN that cannot be fetched is listed in data.errors instead of data.items and its 10 credits are refunded. creditsUsed is what you actually paid. If you do not have enough credits for the whole batch, the call fails with 402 before anything runs.

Amazon prices are strings, and sometimes missing

Look closely at the response above and you will find three different shapes for a price. Code that calls float() on whatever arrives will crash on the second product.

The tracker handles all of them with one small function: it returns None for a missing price, strips $ and thousands separators, and only then converts to a number. A missing price is stored as NULL, never as 0, so an out-of-stock day cannot show up as the lowest price ever seen. currency is USD because the API reads amazon.com only.

The full script

Save this as price_tracker.py. It splits your ASINs into batches of 20, records each product in a checks table, and prints one line per product comparing the new price with the last known price and the lowest one on record.

import sqlite3
import sys
import time
from datetime import datetime, timezone

import requests

API = "https://scrapingbot.io/api/v1/amazon"
HEADERS = {"x-api-key": "YOUR_API_KEY"}
ASINS = ["B07FDJMC9Q", "B0BSHF7WHW"]  # up to 20 per call; more are split into batches


def call(endpoint, params, tries=3):
    """One API call. Failed calls are refunded, so a retry costs nothing extra."""
    for attempt in range(tries):
        res = requests.post(API, headers=HEADERS, timeout=90,
                            json={"endpoint": endpoint, "params": params})
        body = res.json()
        if body.get("success"):
            return body
        if res.status_code in (400, 401, 402):  # bad input, bad key, not enough credits
            raise RuntimeError(body.get("error"))
        time.sleep(2 ** attempt)  # 408, 429 or 5xx: wait and try again
    raise RuntimeError(f"{endpoint} kept failing: {body.get('error')}")


def parse_price(value):
    """Prices are strings such as "79.99" or "$119.99", or null when there is no offer."""
    if value in (None, ""):
        return None
    try:
        return float(str(value).replace("$", "").replace(",", "").strip())
    except ValueError:
        return None


def fetch(asins):
    """Yield one product dict per ASIN that Amazon returned, 20 ASINs per call."""
    for i in range(0, len(asins), 20):
        body = call("/products", {"asins": asins[i:i + 20]})
        print(f"batch of {len(asins[i:i + 20])}: {body['creditsUsed']} credits")
        for err in body["data"].get("errors", []):
            print("  skipped (refunded):", err)
        yield from body["data"]["items"]


def open_db(path="prices.db"):
    db = sqlite3.connect(path)
    db.execute("""CREATE TABLE IF NOT EXISTS checks (
        asin TEXT, checked_at TEXT, price REAL, availability TEXT, title TEXT)""")
    return db


def record(db, item, now):
    asin, price = item["asin"], parse_price(item.get("product_price"))
    prev = db.execute("SELECT price FROM checks WHERE asin = ? AND price IS NOT NULL "
                      "ORDER BY checked_at DESC LIMIT 1", (asin,)).fetchone()
    low = db.execute("SELECT MIN(price) FROM checks WHERE asin = ?", (asin,)).fetchone()[0]
    db.execute("INSERT INTO checks VALUES (?, ?, ?, ?, ?)",
               (asin, now, price, item.get("product_availability"), item.get("product_title")))

    title = (item.get("product_title") or asin)[:32]
    if price is None:  # no offer right now: keep the row, report the availability text
        return f"{asin}  {'no price':>9}  {title}  [{(item.get('product_availability') or '')[:30]}]"
    line = f"{asin}  {f'${price:,.2f}':>9}  {title}"
    if prev is None:
        return line + "  first check"
    if price < prev[0]:
        line += f"  DROP from ${prev[0]:,.2f} (-{(prev[0] - price) / prev[0]:.0%})"
    elif price > prev[0]:
        line += f"  up from ${prev[0]:,.2f}"
    else:
        line += "  unchanged"
    if price < low:
        line += "  LOWEST SEEN"
    else:
        line += f"  low ${low:,.2f}"
    return line


if __name__ == "__main__":
    asins = sys.argv[1:] or ASINS
    db = open_db()
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    for item in fetch(asins):
        print(record(db, item, now))
    db.commit()

A few details in that code are deliberate:

• The previous price skips empty checks. The query asks for the latest row with a price, so a product that was unavailable yesterday is compared with the last day it had one.
• Every check is a new row. Nothing is updated in place, so the table is a complete history you can chart or query later.
• Errors are printed, not hidden. An ASIN listed in data.errors was refunded; if the same one fails every time, check that it still exists on amazon.com.
• Retries are limited to what can recover. 400, 401 and 402 will fail again the same way, so the script stops. Timeouts, 429 and server errors get up to three tries. Failed calls are refunded.

Run it

Pass ASINs on the command line, or edit the ASINS list at the top:

python price_tracker.py B07FDJMC9Q B0BSHF7WHW

The first run has nothing to compare with:

batch of 2: 20 credits
B07FDJMC9Q     $79.99  Ninja AF101 Air Fryer, 4 Qt Capa  first check
B0BSHF7WHW   no price  Apple 2023 MacBook Pro Laptop wi  [Currently unavailable. We don']

On the next run, a product whose price has not moved prints unchanged and the lowest price on record:

batch of 2: 20 credits
B07FDJMC9Q     $79.99  Ninja AF101 Air Fryer, 4 Qt Capa  unchanged  low $79.99
B0BSHF7WHW   no price  Apple 2023 MacBook Pro Laptop wi  [Currently unavailable. We don']

When a price falls, the line reads DROP from the previous price with the percentage, and adds LOWEST SEEN if the new price is below every earlier check. A rise prints up from. To send yourself an alert instead of reading the output, filter for lines that contain DROP and pass them to email or a chat webhook.

Query the history

Because prices.db is an ordinary SQLite file, you can answer questions about it without writing more Python. The sqlite3 command-line tool comes with macOS; on Linux, install your distribution's sqlite3 package:

sqlite3 -header -column prices.db \
  "SELECT asin, COUNT(*) AS checks, MIN(price) AS lowest, MAX(price) AS highest
   FROM checks GROUP BY asin;"

asin        checks  lowest  highest
----------  ------  ------  -------
B07FDJMC9Q  2       79.99   79.99
B0BSHF7WHW  2

MIN and MAX ignore NULL, which is why the unavailable laptop has two checks and no price range.

Schedule it with cron

Run crontab -e and add a line. This one checks every six hours and appends the output to a log. Use full paths, because cron starts with a minimal environment:

0 */6 * * * cd /home/you/tracker && /usr/bin/python3 price_tracker.py >> tracker.log 2>&1

Before you pick a schedule, do the arithmetic. Every product returned costs 10 credits, every time it is checked:

The Starter plan is $49.99 for 275,000 credits, so one batch of 20 costs about 3.6 cents and four checks a day on 20 products use under a tenth of the plan each month. Pick the slowest schedule that still catches the changes you care about: every check of every product is billed, whether or not the price moved.

Who is selling at that price

For a product you watch closely, /product-details adds the buy box: the offer Amazon shows next to the Add to Cart button, and which seller holds it. It costs the same 10 credits as one product in a batch.

import requests

res = requests.post("https://scrapingbot.io/api/v1/amazon", timeout=90,
                    headers={"x-api-key": "YOUR_API_KEY"},
                    json={"endpoint": "/product-details", "params": {"asin": "B07ZPKN6YR"}})
p = res.json()["data"]
box = p.get("main_buy_box") or {}
print(p["product_title"])
print("price:  ", p.get("product_price"))
print("buy box:", box.get("price"), "sold by", box.get("seller"))
print("offers: ", p.get("product_num_offers"))
print("stock:  ", p.get("product_availability"))
print("demand: ", p.get("sales_volume"))

Apple iPhone 11, 64GB, Black - Unlocked (Renewed)
price:   172.00
buy box: $172.00 sold by WirelessSource
offers:  56
stock:   In Stock
demand:  1K+ bought in past month

Note the two spellings of the same price: product_price without a dollar sign and main_buy_box.price with one. Run both through parse_price if you store them. sales_volume is Amazon's own rounded "bought in past month" label, a rough demand signal worth keeping next to the price.

What the API does not do

• Only amazon.com. Prices are US prices in USD; other Amazon stores are not supported.
• No reviews endpoint. /product-details includes a few top reviews from the product page, but there is no endpoint that pages through all reviews.
• No stored history. Each call is a fresh look at the page. The history is the table you build.

Where to go from here

• Find ASINs to track with /search, /best-sellers or /deals-v2; ranking pages often have no price, so send their ASINs to /products. The Amazon API page lists every endpoint.
• Every parameter and response field is in the Amazon API reference.
• For the bigger picture on collecting Amazon data, read the Amazon scraping guide.

Common questions