Build your own price tracker API with price history

A price tracker API answers one question for your app, your dashboard or your pricing team: what has this product cost, and where, over the last N days? Off-the-shelf price history is usually tied to one retailer. If you want the history across every seller on Google Shopping, you have to collect it yourself. That is less work than it sounds.

This guide builds the whole thing in Python:

1. A collector that fetches every Google Shopping seller for a list of products, once a day.
2. A price history table in SQLite, with a Postgres version for later.
3. A GET /price-history endpoint in FastAPI that returns the lowest, average and highest price per day.
4. A small chart that reads from it.

The data comes from the Google Shopping Sellers API, which returns every seller for a product in one request. All code below ran against the live API on 13 October 2026, and the example output is from that run.

The data model

Each daily check of a product returns a list of sellers. Here are three of the 30 sellers we got for a Logitech MX Master 3S mouse (product ID 7078668786207219211, United States), trimmed:

[
  { "position": 1, "merchant": "Walmart", "price": { "value": 89.99, "currency": "USD" },
    "shipping": { "value": 0, "currency": "USD" }, "total_price": { "value": 89.99, "currency": "USD" },
    "availability": "In Stock" },
  { "position": 4, "merchant": "Microless.com", "price": { "value": 92.88, "currency": "USD" },
    "shipping": { "value": 15, "currency": "USD" }, "total_price": { "value": 107.88, "currency": "USD" },
    "availability": "In Stock" },
  { "position": 8, "merchant": "watsinternational.com", "price": { "value": 119.99, "currency": "USD" },
    "shipping": null, "total_price": { "value": 119.99, "currency": "USD" },
    "availability": "In Stock" }
]

So the history needs two levels: one row per check (when, which product, which country, how many sellers, what it cost) and one row per offer within a check. Keeping every offer, not just the lowest price, means you can answer new questions later, such as “when did Walmart stop being cheapest?”, without having collected for it in advance.

A few details from the real response shape the schema:

• shipping can be null when Google shows no shipping figure. In our run, 5 of 30 sellers had none. Store it as NULL, not 0.
• Sellers are sorted by total_price (item plus shipping), and that is what a buyer compares, so the history endpoint uses total_price.
• Keep product_id as text. IDs can be 20 digits long.
• The job’s completedAt timestamp is in UTC. Store it as the check time.

Step 1: the schema and the collector

Put your products in products.csv. You get each product_id once, from a google-shopping-search request for the product name; product IDs explained covers why one model can have several.

product_id,country,label
7078668786207219211,us,Logitech MX Master 3S

The collector creates the tables on first run:

import csv
import os
import sqlite3
import time

import requests

API_URL = "https://api.crawlerdata.com/v1/jobs"
HEADERS = {"Authorization": f"Bearer {os.environ['CRAWLERDATA_API_KEY']}"}
DB_PATH = os.environ.get("PRICE_DB", "prices.db")

SCHEMA = """
CREATE TABLE IF NOT EXISTS products (
  product_id TEXT NOT NULL,
  country    TEXT NOT NULL,
  label      TEXT NOT NULL,
  PRIMARY KEY (product_id, country)
);
CREATE TABLE IF NOT EXISTS checks (
  job_id        TEXT PRIMARY KEY,
  product_id    TEXT NOT NULL,
  country       TEXT NOT NULL,
  checked_at    TEXT NOT NULL,          -- job.completedAt, UTC ISO 8601
  sellers_count INTEGER NOT NULL,
  credits       INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS checks_by_product ON checks (product_id, country, checked_at);
CREATE TABLE IF NOT EXISTS offers (
  job_id       TEXT NOT NULL REFERENCES checks (job_id),
  position     INTEGER NOT NULL,
  merchant     TEXT NOT NULL,
  offer_title  TEXT,                    -- the seller's own title: spot used or refurbished offers
  price        REAL,
  shipping     REAL,                    -- NULL when Google shows no shipping figure
  total_price  REAL,
  currency     TEXT,
  availability TEXT,
  PRIMARY KEY (job_id, position)
);
"""

Then it submits every product on the fast queue, polls until each job finishes and stores the result:

def submit(product_id, country):
    body = {"endpoint": "google-shopping-product",
            "params": {"product_id": product_id, "country": country}, "queue": "fast"}
    resp = requests.post(API_URL, json=body, headers=HEADERS, timeout=30)
    resp.raise_for_status()
    return resp.json()["job"]["id"]


def wait_for(job_id, poll=15, max_wait=600):
    deadline = time.time() + max_wait
    while time.time() < deadline:
        time.sleep(poll)
        data = requests.get(f"{API_URL}/{job_id}", headers=HEADERS, timeout=30).json()
        if data["job"]["status"] in ("completed", "failed"):
            return data
    raise TimeoutError(job_id)


def amount(money):
    return money["value"] if money else None


def store(db, data):
    job, result = data["job"], data["result"]
    db.execute("INSERT OR IGNORE INTO checks VALUES (?, ?, ?, ?, ?, ?)",
               (job["id"], job["params"]["product_id"], job["params"]["country"],
                job["completedAt"], result["sellers_count"], job["credits"]))
    db.executemany(
        "INSERT OR IGNORE INTO offers VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
        [(job["id"], s["position"], s["merchant"], s["offer_title"], amount(s["price"]), amount(s["shipping"]),
          amount(s["total_price"]), (s["total_price"] or s["price"] or {}).get("currency"),
          s["availability"]) for s in result["sellers"]])
    db.commit()


def main():
    db = sqlite3.connect(DB_PATH)
    db.executescript(SCHEMA)
    with open("products.csv", newline="", encoding="utf-8") as f:
        products = list(csv.DictReader(f))
    db.executemany("INSERT OR REPLACE INTO products VALUES (:product_id, :country, :label)", products)
    db.commit()

    jobs = [(p, submit(p["product_id"], p["country"])) for p in products]  # submit all, then poll
    for p, job_id in jobs:
        try:
            data = wait_for(job_id)
        except TimeoutError:
            print(f"{p['label']}: still running, skipped")
            continue
        if data["job"]["status"] == "failed":
            print(f"{p['label']}: failed ({data['job']['error']}), not charged")
            continue
        if data["result"].get("error"):
            print(f"{p['label']}: {data['result']['error']}")
            continue
        store(db, data)
        print(f"{p['label']}: {data['result']['sellers_count']} sellers, {data['job']['credits']} credits")


if __name__ == "__main__":
    main()

Why it is built this way:

• Submit first, poll afterwards. A sellers request collects every seller on the product page in one go. On the fast queue the API answers straight away with a job ID, so all products run in parallel on our side. The fast queue costs the same as live. For large catalogs you can pass a postbackUrl and receive results by webhook instead of polling.
• Failed jobs are skipped, not retried in a loop. They are not charged, and tomorrow’s run fills the gap. A product ID Google no longer recognises completes with "error": "Product not found" and is charged the 1-credit minimum, so remove those from the CSV.
• INSERT OR IGNORE makes a re-run of the same job harmless.

Run it with your API key in the environment:

$ CRAWLERDATA_API_KEY=sk_... python collect.py
Logitech MX Master 3S: 30 sellers, 3 credits

Schedule it once a day with cron (0 6 * * * cd /opt/prices && python collect.py) or any scheduler. Every run adds one check and its offers.

Step 2: the GET /price-history endpoint

The endpoint turns raw offers into one point per day. For each UTC day it takes the last check of that day (so running the collector twice a day does not double-count), then returns the cheapest offer, the average and the highest total price, and the number of sellers.

import os
import sqlite3
from datetime import datetime, timedelta, timezone

from fastapi import FastAPI, HTTPException, Query

DB_PATH = os.environ.get("PRICE_DB", "prices.db")
app = FastAPI(title="Price history")

HISTORY_SQL = """
WITH day_check AS (            -- the last check of each UTC day
  SELECT job_id, substr(checked_at, 1, 10) AS day,
         ROW_NUMBER() OVER (PARTITION BY substr(checked_at, 1, 10) ORDER BY checked_at DESC) AS rn
  FROM checks
  WHERE product_id = :product_id AND country = :country AND checked_at >= :since
),
ranked AS (
  SELECT d.day, o.merchant, o.total_price, o.currency,
         ROW_NUMBER() OVER (PARTITION BY d.day ORDER BY o.total_price, o.merchant) AS cheapest,
         COUNT(*)           OVER (PARTITION BY d.day) AS sellers,
         AVG(o.total_price) OVER (PARTITION BY d.day) AS average_total,
         MAX(o.total_price) OVER (PARTITION BY d.day) AS highest_total
  FROM day_check d
  JOIN offers o ON o.job_id = d.job_id
  WHERE d.rn = 1 AND o.total_price IS NOT NULL
    AND (:merchant IS NULL OR o.merchant = :merchant)
)
SELECT day, merchant, total_price, currency, sellers, average_total, highest_total
FROM ranked WHERE cheapest = 1 ORDER BY day
"""


def db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


@app.get("/products")
def products():
    with db() as conn:
        return [dict(r) for r in conn.execute("SELECT * FROM products ORDER BY label")]


@app.get("/price-history")
def price_history(product_id: str, country: str = "us",
                  days: int = Query(30, ge=1, le=365), merchant: str | None = None):
    since = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d")
    with db() as conn:
        rows = conn.execute(HISTORY_SQL, {"product_id": product_id, "country": country.lower(),
                                          "since": since, "merchant": merchant}).fetchall()
    if not rows:
        raise HTTPException(404, "no checks for this product, country and period")
    return {
        "product_id": product_id,
        "country": country.lower(),
        "merchant": merchant,
        "currency": rows[-1]["currency"],
        "points": [{
            "date": r["day"],
            "lowest_total": r["total_price"],
            "lowest_merchant": r["merchant"],
            "average_total": round(r["average_total"], 2),
            "highest_total": r["highest_total"],
            "sellers": r["sellers"],
        } for r in rows],
    }

Install and run it:

pip install fastapi uvicorn
uvicorn api:app --port 8000

The SQL uses window functions, which SQLite has supported since version 3.25, so any recent Python works. The str | None type hint needs Python 3.10 or newer.

What the endpoint returns

GET /price-history?product_id=7078668786207219211&country=us&days=30 returns:

{
  "product_id": "7078668786207219211",
  "country": "us",
  "merchant": null,
  "currency": "USD",
  "points": [
    {
      "date": "2026-10-13",
      "lowest_total": 89.99,
      "lowest_merchant": "Walmart",
      "average_total": 166.65,
      "highest_total": 305.32,
      "sellers": 30
    }
  ]
}

That is one real point, from our 13 October check. After a month of daily runs you get 30. We also tested the query with several checks on the same day, to confirm it keeps only the last one, and with the merchant filter.

Add &merchant=Walmart to follow one seller’s price instead of the whole market. The merchant name must match exactly what Google Shopping shows, including marketplace prefixes such as Newegg.com - TECH EDGE.

The spread between lowest and highest is large here, from $89.99 to $305.32 for the same mouse, which is why the average alone is a poor summary. Most uses want the lowest price over time; a market view wants lowest, average and highest together.

Keep the history clean

Raw offers include things you may not want in a “price of this product” line:

• Used, refurbished or bundled offers. These show up in the seller’s offer_title, which is why the collector stores it. Exclude matching titles in the history query (AND o.offer_title NOT LIKE '%refurbished%', and so on) or flag them in a column at insert time. In our 30-seller example no title said used or refurbished, but the second-cheapest offer, eBay at $93.77, was titled “Logitech MX Master 3S Wireless Mouse - White - Not Sealed”. Whether that belongs in your lowest price is your call.
• Sellers with no stock. availability was In Stock for 27 of the 30 sellers and empty for the other 3. Decide whether an out-of-stock offer should count towards the lowest price, and filter on it if not.
• Outliers. A single seller listing a multipack or a wrong item can drag the average. The lowest and highest values make this easy to spot; a median, computed in Python or with Postgres’s percentile_cont, resists it better than the average.

Keep the raw rows unchanged and apply these rules in the query. If you change your mind about a rule later, the whole history is recomputed from the same data.

Step 3: chart it

Any charting library can read the endpoint. With Chart.js in a plain HTML page served from the same origin:

<canvas id="history" height="120"></canvas>
<script src="https://cdn.jsdelivr.net/npm/chart.js@4.4.1/dist/chart.umd.min.js"></script>
<script>
  fetch('/price-history?product_id=7078668786207219211&country=us&days=90')
    .then((r) => r.json())
    .then(({ points, currency }) => new Chart(document.getElementById('history'), {
      type: 'line',
      data: {
        labels: points.map((p) => p.date),
        datasets: [
          { label: `Lowest (${currency})`, data: points.map((p) => p.lowest_total) },
          { label: `Average (${currency})`, data: points.map((p) => p.average_total) },
        ],
      },
    }));
</script>

A step line for the lowest price (stepped: true on that dataset) reads more honestly than a smooth one, because prices change in jumps, and you only know the value at each check. Show lowest_merchant in the tooltip and you have the “who was cheapest when” view that most pricing teams ask for first.

Moving to Postgres

When more than one service reads or writes the history, move to Postgres. The schema is the same with proper types:

CREATE TABLE checks (
  job_id        text PRIMARY KEY,
  product_id    text        NOT NULL,
  country       char(2)     NOT NULL,
  checked_at    timestamptz NOT NULL,
  sellers_count integer     NOT NULL,
  credits       integer     NOT NULL
);
CREATE INDEX checks_by_product ON checks (product_id, country, checked_at);

CREATE TABLE offers (
  job_id       text    NOT NULL REFERENCES checks (job_id),
  position     integer NOT NULL,
  merchant     text    NOT NULL,
  offer_title  text,
  price        numeric(12, 2),
  shipping     numeric(12, 2),
  total_price  numeric(12, 2),
  currency     char(3),
  availability text,
  PRIMARY KEY (job_id, position)
);

In the history query, replace substr(checked_at, 1, 10) with (checked_at AT TIME ZONE 'UTC')::date, use ON CONFLICT DO NOTHING instead of INSERT OR IGNORE, and use %(name)s placeholders with psycopg. Use numeric rather than floating point for money. At large volumes, partition offers by month, or keep a daily summary table that the endpoint reads instead of the raw offers.

What it costs

Each check returns every seller in one request and costs 1 credit per 10 sellers returned, rounded up, minimum 1. The mouse above, with 30 sellers, costs 3 credits a check. Polling is free, failed jobs are free, and the queue does not change the price.

Assuming about 21 to 30 sellers per product (3 credits per check):

• 100 products, once a day: 100 × 30 × 3 = 9,000 credits a month. One $50 Starter pack (50,000 credits, valid 6 months) lasts about five and a half months.
• 1,000 products, once a day: 90,000 credits a month, $90 at the Starter rate of $1.00 per 1,000. Over 6 months that is 540,000 credits, which one Standard pack (500,000 credits, $375) plus one Starter pack ($50) covers: about $71 a month.
• 1,000 products, twice a day: double that. Fresher daily points, the same endpoint output.

Products with 10 or fewer sellers cost 1 credit per check, so niche products are cheaper. All four packs and their per-1,000 rates are on the pricing page. Storage is small: in our run one check was 30 offer rows.

Running it in production

• Start collecting early. Every daily run adds a complete point for every seller, so the history grows from your first check.
• Add markets as rows. Track each country as its own row in products.csv. country takes one of 31 two-letter codes, and prices come back in that country’s currency.
• Key sellers by merchant. It is the name Google Shopping shows for the seller. If you want one line per retailer across marketplaces, keep a small alias table next to the history.

Where to go next

For alerting on price drops rather than serving history, see tracking competitor prices in Node.js, which diffs each run against the last. The competitor price monitoring page covers the business use case, and Google Shopping sellers and offers explains what the seller list includes. If your goal is enforcing a minimum advertised price, MAP monitoring software: build or buy? compares the options.