Store Amazon Product Data in a Database: Schema, Upserts and Price History

To store Amazon product data in a database, keep three tables: products keyed by ASIN and marketplace, price points keyed by product, series and day, and sales-rank points keyed by product and time. Store prices as integer minor units and write every row with an upsert, so re-importing the same history changes nothing. One call to /api/amazon-product-statistics backfills the full recorded history; the schema and loader below run on SQLite as-is and on PostgreSQL with a few type changes.

What tables do I need for Amazon product data?

Keep facts that change slowly (title, currency) apart from time series (prices, ranks). The price table has one row per product, series and day: the statistics series carry a date per point and several points can share one, so the day's last value wins (a dip and recovery within one day is not kept). series names the source - buybox, fba, fbm, amazon from statistics, lookup for your own live snapshots. buybox and fbm are landed prices (item plus shipping); fba, amazon and lookup are item prices without shipping, so compare within a series, never across. This is a schema for storing Amazon data you collect, not for building a store.

schema.sql - runs on SQLite 3.24+ and PostgreSQL

CREATE TABLE IF NOT EXISTS products (
    asin            TEXT NOT NULL,
    marketplace_id  TEXT NOT NULL,
    title           TEXT,
    currency        TEXT,
    updated_at      TEXT NOT NULL,              -- ISO 8601, UTC
    PRIMARY KEY (asin, marketplace_id)
);

CREATE TABLE IF NOT EXISTS price_points (
    asin            TEXT NOT NULL,
    marketplace_id  TEXT NOT NULL,
    series          TEXT NOT NULL,              -- buybox | fba | fbm | amazon | lookup
    day             TEXT NOT NULL,              -- YYYY-MM-DD, UTC
    price_minor     INTEGER NOT NULL,           -- cents, pence, whole yen
    PRIMARY KEY (asin, marketplace_id, series, day)
);

CREATE TABLE IF NOT EXISTS rank_points (
    asin            TEXT NOT NULL,
    marketplace_id  TEXT NOT NULL,
    observed_at     TEXT NOT NULL,              -- YYYY-MM-DD HH:MM:SS, UTC
    sales_rank      INTEGER NOT NULL,
    PRIMARY KEY (asin, marketplace_id, observed_at)
);

How do I load the price history without duplicates?

Fetch statistics once per product and upsert every point. Price series arrive as [["YYYY-MM-DD", 41800], ...] in minor units with no-offer periods already removed; the sales-rank series arrives as [["YYYY-MM-DD HH:MM:SS", rank], ...] and can contain -1. Because every write is an upsert, running the loader again - tomorrow or after a crash - leaves the tables exactly as a single run would. get() and new_session() are the helpers from the Python quickstart.

load.py - backfill products, prices and ranks with upserts

import sqlite3
from datetime import datetime, timezone

from sellermagnet import get, new_session

SERIES = {"buybox": "buyBoxPriceHistory", "fba": "lowestFBAPriceHistory",
          "fbm": "lowestFBMPriceHistory", "amazon": "amazonAsSellerPriceHistory"}
UPSERT_PRICE = """INSERT INTO price_points (asin, marketplace_id, series, day, price_minor)
                  VALUES (?, ?, ?, ?, ?)
                  ON CONFLICT (asin, marketplace_id, series, day)
                  DO UPDATE SET price_minor = excluded.price_minor"""
UPSERT_RANK = """INSERT INTO rank_points (asin, marketplace_id, observed_at, sales_rank)
                 VALUES (?, ?, ?, ?)
                 ON CONFLICT (asin, marketplace_id, observed_at)
                 DO UPDATE SET sales_rank = excluded.sales_rank"""
UPSERT_PRODUCT = """INSERT INTO products (asin, marketplace_id, title, currency, updated_at)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT (asin, marketplace_id)
                    DO UPDATE SET title = COALESCE(excluded.title, products.title),
                                  updated_at = excluded.updated_at"""


def backfill(db: sqlite3.Connection, asin: str, marketplace_id: str, currency: str) -> None:
    data = get(new_session(), "amazon-product-statistics", asin=asin, marketplaceId=marketplace_id)
    stats, now = data["stats"], datetime.now(timezone.utc).isoformat(timespec="seconds")
    with db:                                               # one transaction per product
        db.execute(UPSERT_PRODUCT, (asin, marketplace_id, data.get("productTitle"), currency, now))
        for series, key in SERIES.items():
            db.executemany(UPSERT_PRICE, [(asin, marketplace_id, series, day, int(price))
                                          for day, price in stats.get(key) or []])
        db.executemany(UPSERT_RANK, [(asin, marketplace_id, ts, int(rank))
                                     for ts, rank in stats.get("salesRankHistory") or [] if rank >= 1])


db = sqlite3.connect("amazon.db")
db.executescript(open("schema.sql").read())
backfill(db, "B0CL61F39H", "ATVPDKIKX0DER", "USD")

How do I query 30-day lows and price changes?

The series record changes, not days: a price holds until the next row, and periods without an offer are left out rather than stored as zero. So a 30-day low must include the price that was already in effect when the window opened. The window function LAG compares each change with the previous one and gives a change log without application code; that query runs unchanged on SQLite 3.25+ and PostgreSQL.

queries.sql - lowest Buy Box price in 30 days, and every change

-- Lowest Buy Box price per product in the last 30 days, including the price in effect on day one
SELECT p.asin, p.marketplace_id, MIN(p.price_minor) AS low_30d
FROM price_points AS p
WHERE p.series = 'buybox'
  AND p.day >= COALESCE(
        (SELECT MAX(q.day) FROM price_points AS q
         WHERE q.asin = p.asin AND q.marketplace_id = p.marketplace_id
           AND q.series = 'buybox' AND q.day <= date('now', '-30 days')),
        date('now', '-30 days'))
GROUP BY p.asin, p.marketplace_id;

-- Every Buy Box price change, with the previous price beside it
SELECT asin, marketplace_id, day, prev_price, price_minor
FROM (
    SELECT asin, marketplace_id, day, price_minor,
           LAG(price_minor) OVER (PARTITION BY asin, marketplace_id ORDER BY day) AS prev_price
    FROM price_points
    WHERE series = 'buybox'
) AS t
WHERE prev_price IS NOT NULL AND prev_price <> price_minor
ORDER BY asin, day;

On PostgreSQL, replace date('now', '-30 days') with CURRENT_DATE - 30 and store day as DATE. The other differences are listed below.

How often should I refresh the data?

It depends on what you need to see. A daily lookup snapshot records the featured offer's live item price (no shipping), 30 credits per ASIN a month, on all 23 marketplaces. Re-running statistics weekly costs about 4.3 credits a month and, thanks to the upserts, simply adds the points recorded since the last run - but it covers the 11 marketplaces with recorded history. The price history guide explains each series, the sales rank history guide the rank data, and the bulk lookup guide covers running thousands of ASINs.

Weekly refresh from cron, then a quick check with the sqlite3 shell

# crontab -e: cron has no login environment, so set the key there; times are the server's clock
# SELLERMAGNET_API_KEY=your-key-here
# 15 6 * * 1 cd /opt/amazon-db && .venv/bin/python load.py >> load.log 2>&1

sqlite3 -header -column amazon.db \
  "SELECT series, COUNT(*) AS points, MIN(day) AS first, MAX(day) AS last
   FROM price_points GROUP BY series;"

Bottom line: three tables keyed by ASIN and marketplace, integer prices, upserts everywhere, and one statistics call per product to backfill. The product statistics page lists every series, and a free account includes 150 credits.