Build Your Own Amazon Price Tracker in Python
Build an Amazon price tracker in Python: daily price, BSR and seller snapshots to SQLite, a Keepa-style chart, alerts, and robot-check detection that works.
An Amazon price tracker in Python is a small program with one awkward job: visit the same list of ASINs every day, write down the Buy Box price, Best Sellers Rank and seller, and never mistake a bot-check page for a product. This guide builds that tracker end to end, from a CSV watchlist to a SQLite history, a step chart that reads like Keepa's, and an alert that fires once when a price crosses your target. The proxy layer uses SparkProxy's rotating US gateway, with the Scraping API as a fallback for the ASINs that refuse to load.
What the Tracker Does, End to End
The whole system is one daily batch job. Nothing runs between batches, so there is no server to babysit.
| Stage | Input | Output | File |
|---|---|---|---|
| Watchlist | You, editing a CSV | `asin,label,target_cents` rows | `watchlist.csv` |
| Fetch | One ASIN | Raw HTML from `amazon.com/dp/ | `fetch.py`, `retry.py` |
| Classify | Status code + HTML | `ok`, `interstitial`, `throttled`, `gone`, `unknown` | `detect.py` |
| Parse | HTML that passed `ok` | Price in cents, BSR, seller, Buy Box flag | `parse.py` |
| Store | Parsed row | One row per ASIN per UTC day | `store.py` |
| Chart + alert | SQLite history | PNG step chart, email on threshold crossing | `chart.py`, `alerts.py` |
| Orchestrate | All of the above | One run per day | `run_daily.py` |
The watchlist is deliberately boring. Prices are stored in integer cents everywhere, so target_cents is too. A blank target means "track it, never alert":
asin,label,target_cents
B0EXAMPLE1,Espresso grinder,18999
B0EXAMPLE2,Standing desk frame,
B0EXAMPLE3,Replacement filter 3-pack,2450
Install three packages into a virtual environment on Python 3.11 or newer. The floor comes from matplotlib: version 3.11.2, released on PyPI on 2026-09-11, declares Requires-Python >=3.11.
python -m venv .venv
. .venv/bin/activate
pip install "requests>=2.34" "selectolax>=1.0" "matplotlib>=3.11"
If you sell on Amazon and only care about your own listings, the Selling Partner API is the cleaner source, and our breakdown of which proxies Amazon sellers actually need starts from that point. This tracker is for ASINs you don't control: competitors, suppliers, or things you want to buy.
Proxy Choices for amazon.com
Three decisions matter for this job, and each one has a right answer for a daily ASIN sweep.
Country. Amazon decides currency, default delivery region and sometimes the offer itself from where the request comes from. For amazon.com, exit in the US. SparkProxy's USA Rotating pool is listed on the homepage (read 2026-10-30) as "50,000+ premium US datacenter IPs", so every exit is already American and you don't need a country flag at all. On a Worldwide Rotating plan (1M+ IPs, 80+ countries), the dashboard's credential widget shows a country suffix in the username, displayed on the homepage as user-country-us. Copy the exact string your dashboard generates rather than typing it from memory.
Rotation mode. SparkProxy offers Random Rotation, a new exit IP on every request, on port 11000, and Sticky Sessions, which hold one IP for about five minutes, on port 11002. A product-page fetch is a single stateless GET, so Random Rotation is the right fit. It also gives you free retries: when one IP draws a challenge, the next attempt leaves from a different address. Sticky Sessions only earn their place if you set a delivery ZIP through the location picker and need the follow-up page view to come from the same IP. That flow is explained in our guide to how sticky session proxies hold an IP.
Concurrency. SparkProxy plans are sold by threads, with unlimited bandwidth on every plan. The Starter plan's 100 threads are far more than a tracker needs. A daily job gains nothing from running hot, and a burst of parallel hits is the easiest pattern for any anti-bot system to spot. Six workers is the default below, and you can raise it once your fetch_log shows a low challenge rate.
Be clear about what a proxy can and cannot fix here. This table saves a lot of misdirected debugging:
| Symptom in your data | Cause | Does a proxy fix it? |
|---|---|---|
| Interstitial or bot-check page, often HTTP 200 | IP reputation or request rate | Yes: rotate and retry, or move to a residential route |
| HTTP 503 after a burst | Rate limiting on one IP | Yes, together with lower concurrency |
| Price differs from what you see in your browser | Delivery ZIP or Prime/login state | No. Pin one ZIP, compare logged-out to logged-out |
| Empty price on a page that loaded fine | Selector drift, or no Buy Box winner | No. Fix the parser or record NULL |
| HTTP 404 "dogs" page | ASIN removed or not sold in this marketplace | No. Stop requesting it |
| Parent ASIN shows a price range | Variation family, not one product | No. Track the child ASIN |
ZIP pinning has its own write-up in our Amazon product data scraping guide, along with legal framing and search-page pagination. Neither is repeated here.
Scraping at scale? Skip the blocks.
Fast, unblockable datacentre proxies with unlimited bandwidth.
Fetching One ASIN Per Request
The fetcher does exactly one thing: make one request through the rotating gateway and return the status and body. It opens a fresh requests.Session per call so no Amazon cookie from a challenged attempt follows the next one.
# fetch.py: one GET per attempt through the rotating US gateway
import os
import requests
USER = os.environ["SPARK_USER"] # copy both from the SparkProxy dashboard
PASSWORD = os.environ["SPARK_PASS"]
ROTATING = f"http://{USER}:{PASSWORD}@gateway.sparkproxy.io:11000"
HEADERS = {
"User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
}
def fetch_via_proxy(asin: str) -> tuple[int, str]:
"""Return (http_status, html). A fresh session per call means no cookie
carry-over, so each attempt looks like a new visitor on a new IP."""
with requests.Session() as s:
r = s.get(
f"https://www.amazon.com/dp/{asin}",
headers=HEADERS,
proxies={"http": ROTATING, "https": ROTATING},
timeout=30,
)
return r.status_code, r.text
Request the bare /dp/ URL. The long, keyword-stuffed URLs you copy from a browser carry ref= and tracking parameters that change nothing about the price and make your request log harder to read. If you whitelist your server's IP in the dashboard instead of using user:pass, drop the credentials from ROTATING and keep the host and port.
Before trusting the gateway, confirm rotation from a shell. Two calls should print two different addresses:
curl -s -x "http://$SPARK_USER:$SPARK_PASS@gateway.sparkproxy.io:11000" https://api.ipify.org; echo
curl -s -x "http://$SPARK_USER:$SPARK_PASS@gateway.sparkproxy.io:11000" https://api.ipify.org; echo
Catching the Robot Check Before It Poisons Your Data
This is the part that quietly ruins most homemade trackers. When Amazon decides a visitor looks automated, it doesn't always send an error. It sends a small page with a 200 status, and a scraper that trusts the status code parses it, finds no price, and writes a NULL or, worse, a zero into your history.
We captured what that page looks like today. On 2026-10-30 we requested two product pages directly from one of our own servers, with no proxy, a desktop Chrome User-Agent and Accept-Language: en-US. Both requests got the same response: HTTP 200, 3,781 bytes, with these features:
is justAmazon.com, not "Robot Check".- The visible heading reads "Click the button below to continue shopping".
- The form posts to
/errors_page/validateCaptcha, not/errors/validateCaptcha. - An HTML comment begins "To discuss automated access to Amazon data please contact api-services-support@amazon.com."
- There is no image CAPTCHA and no "Enter the characters you see below" text.
That matters because the marker list most Amazon tutorials ship, including older code on this blog, checks for "Robot Check", "Enter the characters you see below", "/errors/validateCaptcha" and "Type the characters you see in this image". We ran that four-string check against the captured page and it matched none of them. A tracker using it would have stored this interstitial as a real product page. Two requests is a small sample, so treat this as one observed variant, not a census of every page Amazon serves.
The fix has two halves: match on the substring that survived the change (validateCaptcha without the path), and require positive proof that you got a detail page before parsing anything.
# detect.py: decide what Amazon actually sent back before anything is parsed
INTERSTITIAL_MARKERS = (
"validateCaptcha", # matches /errors/ and /errors_page/ forms
"Click the button below to continue shopping",
"Enter the characters you see below", # older image-CAPTCHA variant
"api-services-support@amazon.com", # comment on every bot page we have seen
)
def classify(status: int, html: str) -> str:
"""Return one of: ok, interstitial, throttled, gone, unknown."""
if status == 404:
return "gone" # dead or region-locked ASIN: do not retry
if status in (429, 503):
return "throttled"
if any(m in html for m in INTERSTITIAL_MARKERS):
return "interstitial" # often arrives with HTTP 200
if status == 200 and 'id="productTitle"' in html:
return "ok" # positive proof this is a detail page
return "unknown" # 200 but not a product page: treat as a miss
The unknown branch is the safety net. When Amazon ships a new interstitial that matches none of the markers, it still lacks productTitle, so it lands in unknown and gets retried instead of parsed. Blocklists age; a positive check ages much more slowly.
Retrying is where Random Rotation pays off. Every attempt exits from a new IP, so a challenge on attempt one says little about attempt two:
# retry.py: each attempt exits from a different IP, so retrying is cheap
import random
import time
from detect import classify
from fetch import fetch_via_proxy
RETRYABLE = {"interstitial", "throttled", "unknown", "neterror"}
def get_product_page(asin: str, attempts: int = 4) -> tuple[str, str | None, int]:
"""Return (outcome, html_or_None, attempts_used)."""
outcome = "neterror"
for n in range(1, attempts + 1):
try:
status, html = fetch_via_proxy(asin)
outcome = classify(status, html)
except Exception: # timeouts, proxy resets, TLS errors
outcome, html = "neterror", ""
if outcome == "ok":
return outcome, html, n
if outcome not in RETRYABLE:
return outcome, None, n # "gone" ends here
time.sleep(min(30, 2 ** n) + random.uniform(0, 1.5))
return outcome, None, attempts
A 404 ends the loop at once. Retrying a removed ASIN four times a day for months is pure waste, and the fetch_log table below lets you find and prune those rows. For a broader catalogue of block signals beyond Amazon, see how to detect when your scraper is blocked.
Parsing Price, BSR and the Buy Box Seller
One dependency change will bite anyone copying older Amazon code. selectolax 1.0.0 shipped on PyPI on 2026-10-03, and in 1.0.x the classic import now fails outright. We installed 1.0.2 (released 2026-10-22, the same day as the repository's latest commit) and from selectolax.parser import HTMLParser raised:
ImportError: Modest backend is deprecated since selectolax 1.0. The engine has not been updated since 2021, while lexbor has been available since 2021 and is actively developed. Modest is outdated, not maintained, contains bugs and does not follow modern HTML5 standards. Please use lexbor backend instead: `from selectolax.lexbor import LexborHTMLParser`. The APIs are nearly identical, so in most projects swapping the import is all that's needed.
The parser below uses the lexbor backend. Two more choices protect the data. Each field tries a list of selectors, because Amazon serves several page templates at once. And text(separator=" ", strip=True) keeps nested spans from gluing together, which matters for the BSR block, where "#1,234 in Kitchen" and "#12 in Electric Kettles" sit in sibling spans.
# parse.py: selectolax 1.x moved to the lexbor backend
import re
from decimal import Decimal, InvalidOperation
from selectolax.lexbor import LexborHTMLParser
PRICE_SELECTORS = (
"#corePrice_feature_div .a-offscreen",
"#corePriceDisplay_desktop_feature_div .a-offscreen",
"#apex_desktop .a-offscreen",
".a-price .a-offscreen",
)
SELLER_SELECTORS = (
"#sellerProfileTriggerId",
"#merchantInfoFeature_feature_div .offer-display-feature-text-message",
"#merchant-info a",
)
DETAIL_BLOCKS = (
"#productDetails_detailBullets_sections1",
"#detailBulletsWrapper_feature_div",
"#prodDetails",
)
BSR_RE = re.compile(r"#([\d,]+)\s+in\s+([^(#]+?)\s*(?:\(|#|$)")
def first_text(tree, selectors):
for sel in selectors:
node = tree.css_first(sel)
if node:
text = node.text(separator=" ", strip=True)
if text:
return text
return None
def to_cents(raw: str | None) -> int | None:
if not raw:
return None
m = re.search(r"\d[\d,]*(?:\.\d{2})?", raw)
if not m:
return None
try:
return int(Decimal(m.group(0).replace(",", "")) * 100)
except InvalidOperation:
return None
def parse_bsr(tree) -> tuple[int | None, str | None]:
for sel in DETAIL_BLOCKS:
node = tree.css_first(sel)
if not node:
continue
text = node.text(separator=" ", strip=True)
if "Best Sellers Rank" not in text:
continue
m = BSR_RE.search(text.split("Best Sellers Rank", 1)[1])
if m:
return int(m.group(1).replace(",", "")), m.group(2).strip()
return None, None
def parse_product(html: str) -> dict:
tree = LexborHTMLParser(html)
price_raw = first_text(tree, PRICE_SELECTORS)
rank, category = parse_bsr(tree)
avail = first_text(tree, ("#availability",))
return {
"title": first_text(tree, ("#productTitle",)),
"price_cents": to_cents(price_raw),
"price_raw": price_raw,
"bsr": rank,
"bsr_category": category,
"seller": first_text(tree, SELLER_SELECTORS),
"availability": avail,
"buy_box": tree.css_first("#add-to-cart-button") is not None,
}
Decimal avoids the float trap where 19.99 * 100 becomes 1998.9999999999998 and truncates to 1998 cents. The BSR regex keeps only the top-level rank, the first #N in Category after the "Best Sellers Rank" label; extend it if you also want the sub-category ranks.
A caveat on freshness: on the day we wrote this, our direct requests only ever reached the interstitial, so these selectors were tested against a fixture page built from Amazon's known markup, not against a live product page. They are the commonly used ones, but treat them as candidates. The ok gate in detect.py and the price_raw column mean a broken selector shows up as a run of NULL prices on pages that otherwise loaded, which is easy to spot and never mistaken for a price drop.
A SQLite Schema for Daily Snapshots
SQLite is the right database for this until you are tracking a very large catalogue: it's a single file, ships with Python, and handles one writer with ease. Two tables keep data and diagnostics apart.
# store.py: one row per ASIN per UTC day, plus a log of every attempt
import sqlite3
from datetime import datetime, timezone
SCHEMA = """
CREATE TABLE IF NOT EXISTS snapshot (
asin TEXT NOT NULL,
day TEXT NOT NULL, -- YYYY-MM-DD, UTC
price_cents INTEGER, -- NULL = no Buy Box price, never 0
bsr INTEGER,
bsr_category TEXT,
seller TEXT,
buy_box INTEGER NOT NULL,
availability TEXT,
fetched_at TEXT NOT NULL,
PRIMARY KEY (asin, day)
);
CREATE TABLE IF NOT EXISTS fetch_log (
asin TEXT NOT NULL,
at TEXT NOT NULL,
outcome TEXT NOT NULL, -- ok, interstitial, gone, ...
attempts INTEGER NOT NULL,
route TEXT NOT NULL -- proxy or api
);
"""
def connect(path: str = "prices.db") -> sqlite3.Connection:
db = sqlite3.connect(path, timeout=30)
db.execute("PRAGMA journal_mode=WAL")
db.executescript(SCHEMA)
return db
def save_snapshot(db, asin: str, row: dict) -> None:
now = datetime.now(timezone.utc)
db.execute(
"""INSERT INTO snapshot VALUES (?,?,?,?,?,?,?,?,?)
ON CONFLICT(asin, day) DO UPDATE SET
price_cents=excluded.price_cents, bsr=excluded.bsr,
bsr_category=excluded.bsr_category, seller=excluded.seller,
buy_box=excluded.buy_box, availability=excluded.availability,
fetched_at=excluded.fetched_at""",
(asin, now.strftime("%Y-%m-%d"), row["price_cents"], row["bsr"],
row["bsr_category"], row["seller"], int(row["buy_box"]),
row["availability"], now.isoformat(timespec="seconds")),
)
db.commit()
def log_fetch(db, asin: str, outcome: str, attempts: int, route: str) -> None:
db.execute("INSERT INTO fetch_log VALUES (?,?,?,?,?)",
(asin, datetime.now(timezone.utc).isoformat(timespec="seconds"),
outcome, attempts, route))
db.commit()
The primary key (asin, day) makes a rerun idempotent. If the job crashes halfway and you start it again, the second pass overwrites today's rows instead of doubling them. WAL mode lets you open the database in a notebook while the job is writing.
Three states need distinct representations, and conflating them is the most common data bug in DIY trackers:
| State | How it's stored | What the chart does |
|---|---|---|
| Page loaded, Buy Box price found | Row with `price_cents` set | Draws the step |
| Page loaded, no Buy Box (suppressed or out of stock) | Row with `price_cents` NULL, `buy_box` 0 | Breaks the line |
| Page never loaded (all attempts failed) | No row; a `fetch_log` entry instead | Carries the last price forward |
Once a week, this query tells you whether your proxy route is healthy, which ASINs are dead, and whether any are leaning on the paid fallback:
SELECT asin, outcome, route, COUNT(*) AS n, ROUND(AVG(attempts), 2) AS avg_attempts
FROM fetch_log
WHERE at >= date('now', '-7 day')
GROUP BY asin, outcome, route
ORDER BY n DESC;
For a multi-retailer system with raw-HTML archiving and Postgres, the automated price monitoring architecture post covers that bigger design. A single-marketplace ASIN tracker doesn't need it.
Charting Price History Like Keepa
A Keepa-style history works best as a step chart, and the reason is worth stating: a price is a state that holds until the seller changes it, not a sample that drifts smoothly between readings. Drawing straight lines between daily points invents prices nobody ever charged. Matplotlib's drawstyle="steps-post" holds each value flat until the next observation, which is the honest picture.
# chart.py: a step chart, because a price is a state that holds until it changes
import sqlite3
from datetime import date, timedelta
import matplotlib
matplotlib.use("Agg") # no display needed on a server
import matplotlib.pyplot as plt
def plot_history(db: sqlite3.Connection, asin: str, target_cents: int | None,
out: str | None = None, days: int = 180) -> str:
since = (date.today() - timedelta(days=days)).isoformat()
rows = db.execute(
"SELECT day, price_cents FROM snapshot WHERE asin=? AND day>=? ORDER BY day",
(asin, since),
).fetchall()
xs = [date.fromisoformat(d) for d, _ in rows]
ys = [p / 100 if p is not None else float("nan") for _, p in rows]
fig, ax = plt.subplots(figsize=(9, 3.5), layout="constrained")
ax.plot(xs, ys, drawstyle="steps-post", linewidth=1.8)
if target_cents:
ax.axhline(target_cents / 100, linestyle="--", linewidth=1)
ax.set_title(f"{asin}: Buy Box price, last {days} days")
ax.set_ylabel("USD")
ax.grid(alpha=0.3)
out = out or f"{asin}.png"
fig.savefig(out, dpi=120)
plt.close(fig)
return out
The NULL-to-nan conversion is what produces the gap for no-Buy-Box days, so a suppressed listing is visible at a glance instead of being painted as a continuous price. The Agg backend lets the script run under cron on a headless box. To overlay BSR, add ax2 = ax.twinx() and plot the bsr column on an inverted axis, since a smaller rank is better.
Threshold Alerts That Fire Once
A naive alert ("price is below target") emails you every single day the deal lasts. What you want is the crossing: today at or below target, the previous known price above it.
# alerts.py: fire on the crossing, not on every day the price stays low
import smtplib
from email.message import EmailMessage
def crossed_below(db, asin: str, target_cents: int) -> tuple[bool, int | None]:
rows = db.execute(
"""SELECT price_cents FROM snapshot
WHERE asin=? AND price_cents IS NOT NULL
ORDER BY day DESC LIMIT 2""",
(asin,),
).fetchall()
if not rows:
return False, None
today = rows[0][0]
previous = rows[1][0] if len(rows) > 1 else None
hit = today <= target_cents and (previous is None or previous > target_cents)
return hit, today
def send_alert(asin: str, label: str, price_cents: int, target_cents: int,
chart_path: str) -> None:
msg = EmailMessage()
msg["Subject"] = f"{label} hit ${price_cents / 100:.2f} (target ${target_cents / 100:.2f})"
msg["From"] = "tracker@sparkproxy.io"
msg["To"] = "you@sparkproxy.io"
msg.set_content(f"https://www.amazon.com/dp/{asin}")
with open(chart_path, "rb") as f:
msg.add_attachment(f.read(), maintype="image", subtype="png",
filename=f"{asin}.png")
with smtplib.SMTP("smtp.sparkproxy.io", 587) as smtp:
smtp.starttls()
smtp.login("tracker@sparkproxy.io", "APP_PASSWORD")
smtp.send_message(msg)
The query skips NULL prices on purpose. If a product goes out of stock for three days and returns below target, the comparison is against the last real price, so you still get the alert. A first-ever observation already under target also alerts, which is what you want when you add an item mid-sale. Swap the SMTP block for a Slack or Discord webhook if email isn't your channel; the crossing logic stays the same.
Scheduling the Daily Run
The orchestrator fetches in parallel, then does every database write on the main thread. sqlite3 connections shouldn't be shared across threads, and keeping writes serial removes that whole class of bug.
# run_daily.py: watchlist in, snapshots + charts + alerts out
import csv
import random
from concurrent.futures import ThreadPoolExecutor
from alerts import crossed_below, send_alert
from api_fallback import fetch_via_api
from chart import plot_history
from parse import parse_product
from retry import get_product_page
from store import connect, log_fetch, save_snapshot
WORKERS = 6 # well under the thread count on any SparkProxy plan
USE_API_FALLBACK = True
def collect(asin: str) -> tuple[str, str, int, str, dict | None]:
outcome, html, tries = get_product_page(asin)
route = "proxy"
if outcome != "ok" and outcome != "gone" and USE_API_FALLBACK:
outcome, html = fetch_via_api(asin)
tries, route = tries + 1, "api"
return asin, outcome, tries, route, parse_product(html) if html else None
def main() -> None:
with open("watchlist.csv", newline="") as f:
items = {r["asin"]: r for r in csv.DictReader(f)}
order = list(items)
random.shuffle(order) # no fixed crawl pattern day to day
db = connect()
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
results = list(pool.map(collect, order))
for asin, outcome, tries, route, row in results: # all writes on one thread
log_fetch(db, asin, outcome, tries, route)
if row is None or row["title"] is None:
continue
save_snapshot(db, asin, row)
target = int(items[asin]["target_cents"] or 0)
hit, price = crossed_below(db, asin, target) if target else (False, None)
if hit:
path = plot_history(db, asin, target)
send_alert(asin, items[asin]["label"], price, target, path)
if __name__ == "__main__":
main()
We ran this orchestrator end to end against stubbed fetches (one ASIN challenged then clean, one challenged on every attempt, one returning 404). The clean ASIN was stored after two attempts and alerted, the stubborn one fell through to the API route and was logged as a miss, and the 404 stopped after a single request.
On Linux, schedule it with cron at an off-round minute, add a random delay so the start time drifts, and wrap it in flock so a slow run never overlaps the next one:
# crontab -e (server clock in UTC)
SHELL=/bin/bash
17 6 * * * sleep $((RANDOM \% 1800)) && cd /opt/asin-tracker && flock -n run.lock .venv/bin/python run_daily.py >> run.log 2>&1
Two cron traps live in that line. The % must be escaped as \%, or cron cuts the command off at that character. And $RANDOM is a bash feature: cron's default /bin/sh is dash on Debian and Ubuntu, where it expands to nothing, hence the SHELL=/bin/bash line. On Windows, Task Scheduler does the same job:
schtasks /Create /TN "ASIN tracker" /SC DAILY /ST 06:17 /TR "C:\asin-tracker\.venv\Scripts\python.exe C:\asin-tracker\run_daily.py"
Daily is the right cadence for most watchlists. Hourly checks multiply requests by 24, and a chart with 24 points a day is rarely what you open when deciding whether to buy or reprice. If a handful of ASINs genuinely need intraday tracking, give them their own watchlist and cron line instead of speeding up the whole job. Our guide to scheduling and automating scrapers covers systemd timers and queue-based alternatives.
Sending Stubborn ASINs to the Scraping API
SparkProxy's rotating pools are datacenter IPs. The interstitial we captured came from a datacenter address with no proxy at all, on the very first request, so it says nothing about SparkProxy's pool specifically, but it is a fair warning about how Amazon can treat the category. The fallback route gives the ASINs that exhaust their four proxy attempts one more try through the Scraping API's residential pool.
# api_fallback.py: residential retry through the SparkProxy Scraping API
import base64
import os
import requests
from detect import classify
API = "https://scrape.sparkproxy.io/api/v1"
API_KEY = os.environ["SPARK_API_KEY"]
def fetch_via_api(asin: str, render: bool = False) -> tuple[str, str | None]:
params = {
"url": f"https://www.amazon.com/dp/{asin}",
"render_js": "true" if render else "false",
"proxy_type": "premium", # residential pool
"country_code": "US", # match the marketplace
"json_response": "true", # envelope with status_code + base64 body
}
r = requests.get(API, headers={"X-API-Key": API_KEY}, params=params, timeout=200)
if r.status_code != 200: # 429, 530 etc. come from the API itself
return f"api_{r.status_code}", None
env = r.json()
html = base64.b64decode(env.get("body") or "").decode("utf-8", "replace")
outcome = classify(env.get("status_code") or 0, html)
return outcome, html if outcome == "ok" else None
Four details here are easy to get wrong:
render_jsdefaults totruein the API. A tracker that forgets to set it pays for a headless browser on every call. Start withfalseand only passrender=Truefor ASINs whose price block is missing from the plain fetch.json_response=trueis what returns the envelope withjob_id,status_code,duration_ms,credits_usedandbody. Thebodyis base64. Decode it before runningclassify, or the marker search runs against base64 text, never matches, and a challenge page gets through asunknownat best.format=jsonis not the envelope.formatonly converts page content (html, md, mdx, json, screenshot, pdf), so it's the wrong parameter here.- The same
classifyfunction runs on the decoded body. A residential IP lowers the odds of an interstitial; it doesn't make detection optional.
The docs also list a 530 status for a failed scrape (with credits refunded) and 429 for rate or concurrency limits, which is why any non-200 from the API is logged as its own outcome. The full parameter list is in the Scraping API docs. For the wider decision between running proxies yourself and handing the fetch to an API, see web scraping API vs self-managed proxies.
What a Daily Run Costs
The proxy side is a flat fee, because SparkProxy bills by plan and thread count with unlimited bandwidth. Prices below were read from sparkproxy.io on 2026-10-30, in US dollars, excluding VAT/GST:
| Plan | Threads | Monthly price (USD) |
|---|---|---|
| Starter | 100 | $75 |
| Core | 250 | $140 |
| Boost | 500 | $240 |
| Plus | 1,000 | $440 |
Term discounts apply on top: 5% for 3 months, 10% for 6 months, 15% for 12 months. A tracker running six workers never comes near the Starter plan's 100 threads, so Starter covers it whether you watch 50 ASINs or several thousand. There's also a 24-hour free trial with no credit card, enough to run the job a couple of times and read your own fetch_log before paying.
The API fallback is metered in credits. From the credit table in the docs, read 2026-10-30:
| Fallback request | Credits |
|---|---|
| `proxy_type=premium`, `render_js=false` | 10 |
| ...plus `country_code=US` | 15 total |
| `proxy_type=premium`, `render_js=true`, `country_code=US` | 30 total |
New accounts get 1,000 free credits, which buys about 66 plain premium US fallbacks. The cost depends entirely on how many ASINs fail the proxy route, so measure that before budgeting. If 5% of a 1,000-ASIN list falls through each day, that's 50 × 15 = 750 credits a day; at 20% it's 3,000. Your weekly fetch_log query gives you the real rate.
Run time is the other budget. As a planning assumption, not a measurement: if an attempt averages 5 seconds and the average ASIN needs 1.5 attempts, 1,000 ASINs across 6 workers take about 1,000 × 1.5 × 5 / 6 ≈ 1,250 seconds, roughly 21 minutes. Replace the 5 and the 1.5 with numbers from your own logs after the first week.
Frequently asked questions
FAQ
Once a day suits most watchlists: it matches the resolution people actually read a price chart at, and hourly checks cost 24 times the requests. Put the few ASINs that need intraday checks on a separate watchlist with its own schedule instead of speeding up the whole job.
Usually because the 200 is an interstitial, not a product page. On 2026-10-30 we received one titled just "Amazon.com" with a form posting to /errors_page/validateCaptcha. Check for validateCaptcha and require id="productTitle" before parsing; if the page did load, the price selector has drifted or there is no Buy Box winner.
Not necessarily for every request. Start with rotating US datacenter proxies and retries, measure your challenge rate in fetch_log, and send only the ASINs that keep failing to a residential route such as the Scraping API with proxy_type=premium. That keeps metered spend tied to the hard cases.
Use Random Rotation (port 11000) for product-page checks, since each fetch is one stateless request and a fresh IP per retry helps. Use Sticky Sessions (port 11002) only when a flow spans several requests that must share an IP, such as setting a delivery ZIP before loading the product page.
Yes. The parser pulls the top-level BSR and its category from the product details block, and the bsr column stores it daily next to the price. Plot it on an inverted secondary axis, because rank 1 is the best position.
Change the domain in the fetch URL, exit from the matching country (a Worldwide Rotating plan with a GB or DE country username, or country_code=GB or DE on the API), and store the currency alongside the price. Keep one database per marketplace so prices in different currencies never share a chart.
Get 20% off your first month
Premium datacentre proxies with unlimited bandwidth. Use the code at checkout.
Save up to 15% more on quarterly, half-yearly and yearly plans
Related articles
FoxyProxy Setup With Username and Password
FoxyProxy setup for per-site routing: saved username and password, patterns that match, why Chrome rejects SOCKS5 auth, and how to stop the login popup.
Test Your GDPR Cookie Banner From the EU
Test cookie consent banner EU behaviour from outside Europe: load your site via EU exits with Playwright and catch cookies and pixels that fire before consent.
ScrapeBox Proxies: Load, Test, and Harvest Google
Set up ScrapeBox proxies that keep harvesting: the IP:PORT:USER:PASS format, loading a rotating gateway, what the Google test checks, and safe connections.
