Build a Complete Competitor Price Monitor in Python

This is a runnable, end-to-end pattern for Amazon competitor price monitoring: fetch snapshots, diff against baselines, and fire alerts only when they matter — with polling tiers and cost math. Adapted from the full guide on the Pangolinfo blog.

Why this exists

Price monitoring is change detection, not data collection. The system's output is the diff, not the snapshot. Three things make it worth building: Buy Box share follows price (a rival $1.50 cheaper takes most orders while your listing looks unchanged); brand pricing erodes when resellers break MAP; and competitor stock-outs are short, lucrative windows — if you detect them in hours, not weeks.

The architecture

Six stages, same in a script or a pipeline:

1. Track list — ASINs (not URLs), tagged by tier: hero / competitors / watchlist / long-tail
2. Scheduler — what gets checked, how often
3. Fetch — structured API data
4. Normalize — "$19.99" → 19.99; variants handled; deduped
5. Store + diff — snapshots appended, compared to a rolling baseline
6. Alert + act — the only output anyone ever sees

Step 1: Fetch and normalize

import requests from datetime import date API_KEY = "pgl_xxx" # free key at tool.pangolinfo.com — first 60 requests free headers = {"Authorization": f"Bearer {API_KEY}"} def fetch_snapshot(asin): d = requests.get(DETAIL_URL, headers=headers, params={"asin": asin}).json() return { "asin": asin, # Normalize at ingest: "$19.99" (string) -> 19.99 (number).  # Comparing raw strings is the #1 bug in DIY monitors.  "buybox_price": float(d["buybox_price"].replace("$", "")), "list_price": float(d.get("list_price", "0").replace("$", "") or 0), "seller": d["buybox_seller"], "offers": d["offers"], # seller -> price -> fulfillment (attribution)  "in_stock": d["availability"] == "in_stock", "captured_at": date.today().isoformat(), # never skip the timestamp  } 

Track per ASIN: Buy Box price (what customers pay), offer list (who moved), list price (discount depth), seller + fulfillment (event severity — an FBM seller with 200 ratings dropping price is a different event than the brand owner doing it), availability, timestamp.

Step 2: Diff against the baseline

FLOOR = 18.50 # your floor price for this ASIN  def check(asin): baseline = load_baseline(asin) # latest snapshot from your store  today = fetch_snapshot(asin) if today["buybox_price"] < baseline["buybox_price"]: drop = (baseline["buybox_price"] - today["buybox_price"]) / baseline["buybox_price"] # Dual thresholds: absolute guards the floor, percentage catches drift  if today["buybox_price"] < FLOOR or drop > 0.05: maybe_alert(asin, today, baseline, drop) elif not today["in_stock"] and baseline["in_stock"]: maybe_alert(asin, today, baseline, 0, kind="restock_watch") elif today["in_stock"] and not baseline["in_stock"]: maybe_alert(asin, today, baseline, 0, kind="back_in_stock") save_baseline(asin, today) # roll the baseline forward 

Step 3: Alert like you mean it

def maybe_alert(asin, today, baseline, drop, kind="drop"): # Cooldown: same ASIN + direction, 24h quiet after firing.  # Without this, an oscillating rival pages you six times a night,  # and the channel gets muted by Friday.  if alerted_recently(asin, kind, hours=24): return # Attribution: attach WHO moved, not just that something moved.  # "Price dropped" is trivia; "Seller X cut $2 at 3pm and took  # the Buy Box" is intelligence.  alert(f"{kind.upper()} {asin}: ${baseline['buybox_price']} -> " f"${today['buybox_price']} by {today['seller']} ({drop:.1%})") record_alert(asin, kind) 

Severity tiers worth adding on top: critical (floor breached / Buy Box lost → page immediately), watch (moderate shift / new seller entered → daily digest), opportunity (rival out of stock / MAP violation → dashboard).

Step 4: Schedule by tier

Frequency is a cost function — tier it:

from apscheduler.schedulers.blocking import BlockingScheduler sched = BlockingScheduler() sched.add_job(lambda: poll_tier("hero"), "interval", hours=1) # revenue defense sched.add_job(lambda: poll_tier("competitors"), "interval", hours=6) # intraday moves sched.add_job(lambda: poll_tier("watchlist"), "interval", hours=24) # trends sched.add_job(lambda: poll_tier("longtail"), "interval", days=7) # structural shifts sched.start() 

Two practical notes: prices move in bursts around traffic peaks, so 6-hour polling often catches as much as 2-hour for a third of the cost. And some APIs cache hard targets ~10 minutes — polling faster than that pays twice for one snapshot.

The failure modes this design avoids

• Adversarial target: anti-bot, CAPTCHAs, rate limits — using a structured API instead of raw scraping moves this cost to the provider.
• Wrong price field: six tracked fields, Buy Box first.
• Noise: dual thresholds + 24h cooldown + severity tiers.
• No attribution: offer-list data attached to every alert.
• Stale data: timestamps on every record; cache behavior accounted for in scheduling.

What it costs

On Pangolinfo's Amazon Scraper API: 1 credit/page, one page per ASIN per check.

First 60 requests free — enough to wire this up against real ASINs. Tier your frequencies and most teams cut 60–70% off the naive number. Prefer zero infra? Their Product Research Tool runs scheduled tracking with alerts at 1.5 credits/page.

Checklist before you ship

• Snapshots carry timestamps; baseline rolls forward
• Prices normalized to numbers at ingest
• Dual thresholds + 24h cooldown + severity tiers
• Alerts carry seller attribution from the offer list
• Frequencies tiered; cache behavior accounted for
• Snapshot table supports replay at any historical point

Adapted from the complete guide — which also walks through MAP enforcement and Buy Box defense scenarios plus provider selection.