DEV Community: Mox Loop

Building a Real-Time Amazon Sales Tracker with Scraper API: Complete Python Guide

Mox Loop — Thu, 07 May 2026 08:10:42 +0000

TL;DR

Amazon doesn't expose sales data via any official API. Every "sales tracker tool" estimates from BSR (Best Seller Rank) signals. This tutorial shows you how to:

Understand the BSR-to-sales estimation model
Query real-time ASIN data via Pangolinfo Scrape API
Build async batch queries for large ASIN sets
Set up a daily monitoring pipeline with change alerts

Why BSR-Based Estimation Works (And Where It Breaks)

Amazon updates BSR frequently — hourly in hot categories, daily in slower ones. Sales intelligence tools build regression models that map BSR position to approximate unit volume. The models are category-specific because a BSR of 500 in Electronics represents very different volume than BSR 500 in Collectibles.

Accuracy in practice: ±25-30% in major US categories, ±100%+ in niche subcategories. Freshness: SaaS tools typically snapshot every 1-7 days. If you need real-time data or more than a few thousand queries per day, you need direct API access.

Setup

pip install aiohttp python-dotenv schedule

# .env
PANGOLINFO_API_KEY=your_api_key_here

Core Module: Single ASIN Query

# amazon_tracker.py
import os
import requests
from datetime import datetime
from dataclasses import dataclass
from typing import Optional
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("PANGOLINFO_API_KEY")
API_URL = "https://api.pangolinfo.com/v1/amazon/product"

# BSR → monthly sales reference table (US, major categories)
# Source: industry benchmarks — calibrate per-category for production use
BSR_SALES_REF = {
    100: 12000,
    500: 4000,
    1000: 2200,
    3000: 900,
    5000: 600,
    10000: 300,
    30000: 80,
    100000: 20,
}

@dataclass
class ASINSnapshot:
    asin: str
    marketplace: str
    timestamp: str
    main_bsr: Optional[int]
    category: Optional[str]
    sub_bsr: Optional[int]
    sub_category: Optional[str]
    estimated_sales: Optional[int]
    price: Optional[float]
    review_count: Optional[int]
    availability: Optional[str]
    buybox_seller: Optional[str]
    error: Optional[str] = None

def bsr_to_sales(bsr: int) -> int:
    for threshold in sorted(BSR_SALES_REF):
        if bsr <= threshold:
            return BSR_SALES_REF[threshold]
    return 5

def query_asin(asin: str, marketplace: str = "US") -> ASINSnapshot:
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    payload = {"asin": asin, "marketplace": marketplace}

    snapshot = ASINSnapshot(
        asin=asin, marketplace=marketplace,
        timestamp=datetime.utcnow().isoformat(),
        main_bsr=None, category=None, sub_bsr=None, sub_category=None,
        estimated_sales=None, price=None, review_count=None,
        availability=None, buybox_seller=None
    )

    try:
        resp = requests.post(API_URL, headers=headers, json=payload, timeout=30)
        resp.raise_for_status()
        data = resp.json()

        bsr_list = data.get("best_sellers_rank", [])
        if bsr_list:
            snapshot.main_bsr = bsr_list[0].get("rank")
            snapshot.category = bsr_list[0].get("category")
            if len(bsr_list) > 1:
                snapshot.sub_bsr = bsr_list[1].get("rank")
                snapshot.sub_category = bsr_list[1].get("category")
            if snapshot.main_bsr:
                snapshot.estimated_sales = bsr_to_sales(snapshot.main_bsr)

        snapshot.price = data.get("price")
        snapshot.review_count = data.get("review_count")
        snapshot.availability = data.get("availability")
        snapshot.buybox_seller = data.get("buybox_winner", {}).get("seller_name")

    except Exception as e:
        snapshot.error = str(e)

    return snapshot

Async Batch Queries (10x faster for large ASIN sets)

# async_batch.py
import asyncio
import aiohttp
from typing import List
from amazon_tracker import BSR_SALES_REF, bsr_to_sales, ASINSnapshot
from datetime import datetime
import os

API_KEY = os.getenv("PANGOLINFO_API_KEY")
API_URL = "https://api.pangolinfo.com/v1/amazon/product"
CONCURRENCY = 10  # adjust based on your API rate limit

async def query_asin_async(
    session: aiohttp.ClientSession, asin: str, marketplace: str = "US"
) -> ASINSnapshot:
    payload = {"asin": asin, "marketplace": marketplace}
    snapshot = ASINSnapshot(
        asin=asin, marketplace=marketplace,
        timestamp=datetime.utcnow().isoformat(),
        main_bsr=None, category=None, sub_bsr=None, sub_category=None,
        estimated_sales=None, price=None, review_count=None,
        availability=None, buybox_seller=None
    )

    try:
        async with session.post(API_URL, json=payload) as resp:
            resp.raise_for_status()
            data = await resp.json()

            bsr_list = data.get("best_sellers_rank", [])
            if bsr_list:
                snapshot.main_bsr = bsr_list[0].get("rank")
                snapshot.category = bsr_list[0].get("category")
                if snapshot.main_bsr:
                    snapshot.estimated_sales = bsr_to_sales(snapshot.main_bsr)

            snapshot.price = data.get("price")
            snapshot.review_count = data.get("review_count")
            snapshot.availability = data.get("availability")
    except Exception as e:
        snapshot.error = str(e)

    return snapshot

async def batch_query(asins: List[str], marketplace: str = "US") -> List[ASINSnapshot]:
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    semaphore = asyncio.Semaphore(CONCURRENCY)

    async def bounded(asin):
        async with semaphore:
            return await query_asin_async(session, asin, marketplace)

    async with aiohttp.ClientSession(headers=headers) as session:
        return await asyncio.gather(*[bounded(asin) for asin in asins])

# Usage
if __name__ == "__main__":
    import json
    from dataclasses import asdict

    test_asins = ["B08N5WRWNW", "B07XJ8C8F5", "B09G9FPHY6"]
    results = asyncio.run(batch_query(test_asins))

    for r in results:
        if not r.error:
            print(f"{r.asin}: BSR={r.main_bsr}, est_sales={r.estimated_sales}, price={r.price}")

    with open("results.json", "w") as f:
        json.dump([asdict(r) for r in results], f, indent=2)

Daily Monitoring Pipeline with Change Detection

# monitor.py
import sqlite3
import schedule
import time
import asyncio
import requests
from dataclasses import asdict
from async_batch import batch_query
from datetime import datetime

DB_PATH = "asin_monitor.db"
WATCH_LIST = ["B08N5WRWNW", "B07XJ8C8F5", "B09G9FPHY6"]
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/YOUR_TOKEN"
BSR_ALERT_THRESHOLD = 0.30  # alert on >30% BSR change

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS snapshots (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                asin TEXT, main_bsr INTEGER, estimated_sales INTEGER,
                price REAL, review_count INTEGER, availability TEXT,
                recorded_at TEXT
            )
        """)

def save_snapshots(snapshots):
    with sqlite3.connect(DB_PATH) as conn:
        for s in snapshots:
            if not s.error:
                conn.execute(
                    "INSERT INTO snapshots VALUES (NULL,?,?,?,?,?,?,?)",
                    (s.asin, s.main_bsr, s.estimated_sales, s.price,
                     s.review_count, s.availability, s.timestamp)
                )

def check_alerts(snapshots):
    with sqlite3.connect(DB_PATH) as conn:
        for s in snapshots:
            if s.error or not s.main_bsr:
                continue
            prev = conn.execute(
                "SELECT main_bsr FROM snapshots WHERE asin=? ORDER BY recorded_at DESC LIMIT 1 OFFSET 1",
                (s.asin,)
            ).fetchone()
            if prev and prev[0]:
                change = (s.main_bsr - prev[0]) / prev[0]
                if abs(change) > BSR_ALERT_THRESHOLD:
                    direction = "improved" if change < 0 else "dropped"
                    msg = f"BSR Alert: {s.asin} rank {direction} {change:+.1%} | {prev[0]} -> {s.main_bsr}"
                    print(f"[ALERT] {msg}")
                    send_feishu(msg)

def send_feishu(message: str):
    try:
        requests.post(FEISHU_WEBHOOK, json={"msg_type": "text", "content": {"text": message}}, timeout=5)
    except Exception:
        pass

def daily_job():
    print(f"[{datetime.now().isoformat()}] Running daily BSR snapshot...")
    snapshots = asyncio.run(batch_query(WATCH_LIST, marketplace="US"))
    save_snapshots(snapshots)
    check_alerts(snapshots)
    success = sum(1 for s in snapshots if not s.error)
    print(f"Done: {success}/{len(snapshots)} succeeded")

init_db()
schedule.every().day.at("09:00").do(daily_job)

print("Monitor started. Running daily at 09:00.")
while True:
    schedule.run_pending()
    time.sleep(60)

Key Design Decisions

Why not scrape Amazon directly? Rate limiting, IP blocking, CAPTCHA challenges, and JavaScript rendering make direct scraping expensive to maintain reliably at scale. Pangolinfo Scrape API handles all of that transparently.

BSR-to-sales model accuracy? The reference table in this code is a simplified version for illustration. For production, calibrate per category using a sample of ASINs where you have verified sales data.

Rate limits? The CONCURRENCY = 10 semaphore in the async module keeps you well within typical API limits. Adjust based on your tier.

Resources

Pangolinfo Scrape API Documentation
Amazon BSR Official Help
GitHub: link to your own repo if you extend this

Questions or issues? Drop them in the comments.

Amazon Niche Research Data Analysis: Building a Six-Metric Pipeline with Python and Pangolinfo Scrape API

Mox Loop — Wed, 06 May 2026 04:27:50 +0000

TL;DR — Off-the-shelf Amazon research tools hide their decision logic behind a single "opportunity score." This post walks through building your own six-metric niche research pipeline in Python, using ASIN-level data from Pangolinfo Scrape API. Code is production-ready and runs async for cost efficiency.

Why this exists

I've worked on data tooling for several Amazon-native brands. The recurring frustration was always the same — the team would buy a SaaS tool, get an opportunity score for a category, and have no way to interrogate why the score was what it was. When the inevitable wrong sourcing call happened, there was no decision log to learn from.

The fix is to drop one layer below the SaaS tools, pull ASIN-level data directly, and build your own decision framework in code. Amazon niche research data analysis done at this layer is reproducible, transparent, and version-controllable. Once the pipeline exists, every subsequent category analysis costs near-zero engineering time.

The framework

Six independent metrics, each tied to a specific data source. Cross-reference rather than average them.

Metric	Question it answers	Data source
BSR concentration (CR10)	Is the market locked?	Category bestsellers
Review barrier (P25)	How many reviews to stand up?	Category bestsellers
New-listing velocity	Does the algorithm welcome newcomers?	Weekly category snapshots
Price-band coverage	Is the target price crowded?	Product detail
Sponsored placement density	How much organic traffic is left?	SERP with SP detection
Negative-review clustering	What's the unmet need?	Reviews + NLP

Picking the data source

I tested several Amazon data APIs. The key constraints are:

Full-category ASIN coverage — not just visible top 100
Reliable Sponsored placement detection — most generic scrapers miss 20–40% of SP slots
Complete review text returned — including full Customer Says keyword set, not truncated top 6

Pangolinfo Scrape API hit all three (98%+ SP detection rate is best-in-class), so the rest of this post uses it. The patterns transfer to any equivalent API.

Architecture

┌──────────────────────────────────────────────┐
│ Pangolinfo Scrape API                        │
│  /amazon/category/bestsellers                │
│  /amazon/serp                                │
│  /amazon/reviews                             │
└─────────────────┬────────────────────────────┘
                  │ JSON
                  ▼
┌──────────────────────────────────────────────┐
│ async data fetcher (aiohttp)                 │
│  with semaphore concurrency control          │
└─────────────────┬────────────────────────────┘
                  │
                  ▼
┌──────────────────────────────────────────────┐
│ pandas/numpy metric calculators              │
│  cr10, p25, velocity, price-band, sp density │
└─────────────────┬────────────────────────────┘
                  │
                  ▼
┌──────────────────────────────────────────────┐
│ NLP clustering (jieba/yake → BERT in prod)   │
└─────────────────┬────────────────────────────┘
                  │
                  ▼
┌──────────────────────────────────────────────┐
│ NicheReport dataclass + opportunity scoring  │
└──────────────────────────────────────────────┘

Code — async client

import asyncio
import aiohttp
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class NicheReport:
    node_id: str
    cr10: float
    review_p25: float
    price_band_competitors: int
    breakthrough_price: Optional[float]
    sp_density: float
    pain_points: List[tuple]
    opportunity_score: int

class PangolinAsyncClient:
    """
    Pangolinfo Amazon Scrape API async wrapper (aiohttp + semaphore).
    Targets the synchronous v1/scrape endpoint. Business type is
    selected via the parserName field — switch to /api/v1/scrape/async
    only when you scrape millions of pages per day.
    Full docs: https://docs.pangolinfo.com/en-api-reference/universalApi/universalApi

    parserName cheatsheet (the `content` field varies by parser):
      amzProductDetail      → ASIN (reviews are nested in this response)
      amzKeyword            → search keyword (SERP, includes SP placements)
      amzProductOfCategory  → category Browse Node ID
      amzBestSellers        → category keyword for best-sellers chart
      amzNewReleases        → category keyword for new-releases chart
      amzProductOfSeller    → seller/store ID
    """
    BASE_URL = "https://scrapeapi.pangolinfo.com/api/v1/scrape"
    SITE_MAP = {"US": "www.amazon.com", "DE": "www.amazon.de",
                "UK": "www.amazon.co.uk", "JP": "www.amazon.co.jp"}

    def __init__(self, api_key: str, concurrency: int = 8):
        self.api_key = api_key
        self.semaphore = asyncio.Semaphore(concurrency)

    async def _post(self, session, parser_name, content, marketplace="US", zipcode="10041"):
        payload = {
            "parserName": parser_name,
            "content": content,
            "site": self.SITE_MAP[marketplace],
            "format": "json",
            "bizContext": {"zipcode": zipcode},
        }
        async with self.semaphore:
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            }
            async with session.post(self.BASE_URL, json=payload, headers=headers) as resp:
                resp.raise_for_status()
                data = await resp.json()
                return data["data"]["json"]

    async def fetch_category_listings(self, session, node_id):
        return await self._post(session, "amzProductOfCategory", node_id)

    async def fetch_keyword_serp(self, session, keyword):
        return await self._post(session, "amzKeyword", keyword)

    async def fetch_product_detail(self, session, asin):
        """Returns the full product detail. The `reviews` array is nested in the
        response — filter client-side by `star` to get negative reviews only."""
        return await self._post(session, "amzProductDetail", asin)

Code — metric computation

def compute_cr10(top100):
    df = pd.DataFrame(top100).sort_values("estimated_monthly_sales", ascending=False)
    top10 = df.head(10)["estimated_monthly_sales"].sum()
    total = df["estimated_monthly_sales"].sum()
    return top10 / total if total else 0.0

def compute_review_p25(top50):
    reviews = [item["review_count"] for item in top50 if item.get("review_count")]
    return float(np.percentile(reviews, 25)) if reviews else 0.0

def compute_price_band(top200, target_price, band=5):
    df = pd.DataFrame(top200)
    in_band = df[(df["price"] >= target_price - band) & (df["price"] <= target_price + band)]
    low_review_winners = df[(df["review_count"] < 500)].head(100)
    breakthrough = float(low_review_winners["price"].median()) if len(low_review_winners) else None
    return len(in_band), breakthrough

def compute_sp_density(serp):
    top48 = serp[:48]
    sp = sum(1 for item in top48 if item.get("is_sponsored"))
    return sp / len(top48) if top48 else 0.0

def cluster_pain_points(reviews_text, top_k=5):
    """MVP: keyword extraction. Production: swap for BERT/sentence-transformers."""
    import yake
    extractor = yake.KeywordExtractor(top=top_k, n=2)
    text = " ".join(reviews_text)
    return extractor.extract_keywords(text)

Code — the orchestration

async def evaluate_niche(client, node_id, target_price, head_keyword):
    async with aiohttp.ClientSession() as session:
        listings, serp = await asyncio.gather(
            client.fetch_category_listings(session, node_id),
            client.fetch_keyword_serp(session, head_keyword),
        )
        top200 = listings[:200]

        # Reviews are nested inside amzProductDetail responses
        detail_tasks = [client.fetch_product_detail(session, item["asin"]) for item in top200[:30]]
        all_details = await asyncio.gather(*detail_tasks)

        flat_reviews = []
        for detail in all_details:
            results = detail[0].get("data", {}).get("results", [{}])[0]
            for review in results.get("reviews", []):
                star_value = float(review.get("star", "5").split()[0])  # "4.0 out of 5 stars"
                if star_value <= 3:
                    flat_reviews.append(review.get("content", ""))

    cr10 = compute_cr10(top200[:100])
    review_p25 = compute_review_p25(top200[:50])
    band_count, breakthrough = compute_price_band(top200, target_price)
    sp_density = compute_sp_density(serp)
    pain_points = cluster_pain_points(flat_reviews)
    score = score_opportunity(cr10, review_p25, sp_density, band_count)

    return NicheReport(
        node_id=node_id,
        cr10=cr10,
        review_p25=review_p25,
        price_band_competitors=band_count,
        breakthrough_price=breakthrough,
        sp_density=sp_density,
        pain_points=pain_points,
        opportunity_score=score,
    )

def score_opportunity(cr10, review_p25, sp_density, band_count):
    score = 100
    score -= 30 if cr10 > 0.70 else (10 if cr10 > 0.55 else 0)
    score -= 25 if review_p25 > 1500 else (10 if review_p25 > 800 else 0)
    score -= 20 if sp_density > 0.45 else (10 if sp_density > 0.35 else 0)
    score -= 15 if band_count > 30 else 0
    return max(0, score)

Running it

import asyncio

async def main():
    client = PangolinAsyncClient(api_key="your_key", concurrency=8)
    report = await evaluate_niche(
        client,
        node_id="2251606011",   # Coffee Filters
        target_price=18.99,
        head_keyword="coffee filters",
    )
    print(f"Opportunity score: {report.opportunity_score}/100")
    print(f"CR10: {report.cr10:.1%}")
    print(f"Review barrier (P25): {report.review_p25:.0f}")
    print(f"Competitors in target price band: {report.price_band_competitors}")
    print(f"Breakthrough price anchor: ${report.breakthrough_price}")
    print(f"SP density: {report.sp_density:.1%}")
    print(f"Top pain points: {report.pain_points}")

asyncio.run(main())

Sample output for a representative category:

Opportunity score: 65/100
CR10: 48.3%
Review barrier (P25): 412
Competitors in target price band: 14
Breakthrough price anchor: $16.50
SP density: 31.2%
Top pain points: [('bleach taste', 0.041), ('cup size fit', 0.038), ('packaging fragile', 0.029), ...]

Production hardening

A few things to add before this hits production:

Caching — BSR doesn't need minute-level refresh. Redis-cache the bestseller responses with a 24-hour TTL.

import redis, json
r = redis.Redis()

async def fetch_category_cached(client, session, node_id, ttl=86400):
    key = f"cat:{node_id}:{pd.Timestamp.now().date()}"
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    data = await client.fetch_category_listings(session, node_id)
    r.setex(key, ttl, json.dumps(data))
    return data

Rate limiting — token bucket on the client side to stay under API quota:

from aiolimiter import AsyncLimiter
limiter = AsyncLimiter(max_rate=10, time_period=1)  # 10 req/s

async def rate_limited_get(...):
    async with limiter:
        return await client._get(...)

NLP upgrade path — yake or jieba is fine for MVP but misses semantic clusters ("hard to clean" and "won't come off in dishwasher" should cluster together). For production, swap in sentence-transformers with HDBSCAN clustering, or call an LLM for thematic summarization.

New-listing velocity — needs historical snapshots, which means scheduling a weekly job that diffs current top-200 ASINs against a stored ledger. Use Celery + Redis for the scheduling.

Why this beats SaaS tools

Cross-metric filters work natively. "ASINs with under 500 reviews that still rank in top 100" is one DataFrame query, not a CSV export.
Refresh frequency under your control. Run it hourly during a launch monitoring window if you need to.
Decision logic is in code. Future-you (or your team) can always answer "why did we score this category at 65?"
Marginal cost per category approaches zero. Once the pipeline runs, scanning 100 candidates per week costs only API calls and compute.

If you don't want to self-build, AMZ Data Tracker ships the same six-metric framework as a hosted dashboard with daily refresh and built-in NLP. The trade-off is less customization but a 30-minute setup instead of 4–6 weeks.

Closing

The compounding payoff of this kind of pipeline is what makes it worth building. Every category you evaluate is logged, every decision is reproducible, every threshold can be tuned based on actual outcomes. After six months of running it you have a calibrated model that's specific to your supply chain and risk tolerance — something no off-the-shelf tool can give you.

Code repository structure if you want to extend this — break it into clients/, metrics/, scoring/, pipelines/, and tasks/. Drop the async client into clients/, the metric functions into metrics/, and use a Celery beat schedule in tasks/ for the recurring jobs.

Happy to answer questions in the comments — particularly interested in hearing how others have approached the new-listing velocity tracking, since the snapshot diffing is the trickiest part of the pipeline to operationalize cleanly.

Building an End-to-End Amazon Movers & Shakers Data Pipeline: Engineering Guide from Real-Time Crawling to Automated Alerting

Mox Loop — Thu, 23 Apr 2026 02:22:55 +0000

Background

I've been building data infrastructure for Amazon seller tools for the past three years. The most consistently requested feature from our seller clients is early trend detection — specifically, getting notified when a product starts showing unusual rank velocity before the broader market notices.

Amazon's Movers and Shakers (MnS) list is the best public data source for this. It tracks the largest BSR gainers in each category over a rolling 24-hour window, updating hourly. A product climbing from rank #100,000 to #2,000 represents a +4,900% gain — invisible in the Best Sellers list, immediately flagged in MnS.

The technical challenge: scraping MnS at scale (50+ categories, hourly) while maintaining data quality and managing Amazon's aggressive rate limiting.

Why Self-Built Scrapers Break at Scale

Our initial implementation used rotating proxies + Playwright for JavaScript rendering. It worked well for 5-10 categories but degraded rapidly at scale:

Scale	Success Rate	P95 Latency	Monthly Cost
10 categories/hour	94%	8.2s	~$45 (proxy)
50 categories/hour	71%	22.5s	~$180 (proxy)
200 categories/hour	43%	48.1s	~$620 (proxy)

The root causes:

IP reputation decay: Amazon's ML-based bot detection spots residential proxy patterns after sustained usage
Session state management: Price and availability fields require maintaining valid Amazon session cookies
HTML structure drift: MnS pages have A/B test variants; a parser built on one variant breaks on another

Solution: Pangolinfo Scrape API

Pangolinfo Scrape API provides pre-built parsing templates for Amazon page types including MnS. The managed infrastructure handles proxy rotation, session management, and parser maintenance — the output is clean structured JSON regardless of which A/B variant Amazon serves.

API response schema for MnS:

{
  "status": "success",
  "category_id": "1055398",
  "category_name": "Kitchen & Dining",
  "retrieved_at": "2026-04-22T09:00:00Z",
  "items": [
    {
      "rank": 1,
      "asin": "B09XK3K3T3",
      "title": "Electric Milk Frother Handheld",
      "current_rank": 47,
      "previous_rank": 1847,
      "rank_gain_pct": 3826.0,
      "rank_gain_absolute": 1800,
      "price": 24.99,
      "currency": "USD",
      "rating": 4.3,
      "review_count": 128,
      "is_prime": true,
      "badge": "Best Seller",
      "image_url": "https://m.media-amazon.com/images/...",
      "listing_url": "https://www.amazon.com/dp/B09XK3K3T3"
    }
  ]
}

Full Implementation

Async Multi-Category Collector

"""
mns_async_collector.py

Async implementation for high-frequency, multi-category MnS data collection.
Reduces collection time for 50 categories from ~110s (sequential) to ~12s (async).

Requirements: pip install aiohttp python-dotenv
"""

import asyncio
import aiohttp
import os
import json
import logging
from datetime import datetime, timezone
from typing import List, Dict, Optional

logger = logging.getLogger(__name__)

API_KEY = os.environ["PANGOLINFO_API_KEY"]
API_ENDPOINT = "https://api.pangolinfo.com/scrape"

# Semaphore: limit concurrent requests to avoid overwhelming the API
MAX_CONCURRENT = 10


async def fetch_category(
    session: aiohttp.ClientSession,
    sem: asyncio.Semaphore,
    category_id: str,
    locale: str = "us"
) -> Optional[Dict]:
    """Async fetch for a single MnS category."""
    url = (
        f"https://www.amazon.{'co.uk' if locale == 'uk' else 'com'}"
        f"/gp/movers-and-shakers/{category_id}"
    )
    payload = {
        "url": url,
        "parse_type": "movers_shakers",
        "output_format": "json",
        "locale": locale
    }
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    async with sem:
        try:
            async with session.post(
                API_ENDPOINT, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=45)
            ) as resp:
                resp.raise_for_status()
                data = await resp.json()
                return {"category_id": category_id, "items": data.get("items", [])}
        except aiohttp.ClientResponseError as e:
            logger.error(f"Category {category_id}: HTTP {e.status}")
        except asyncio.TimeoutError:
            logger.error(f"Category {category_id}: Timeout")
        except Exception as e:
            logger.error(f"Category {category_id}: {e}")
        return None


async def collect_all_categories(
    category_ids: List[str],
    locale: str = "us"
) -> List[Dict]:
    """
    Collect MnS data for all specified categories concurrently.
    Returns list of category result dicts (may contain None for failed categories).
    """
    sem = asyncio.Semaphore(MAX_CONCURRENT)
    connector = aiohttp.TCPConnector(limit=MAX_CONCURRENT)

    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [
            fetch_category(session, sem, cat_id, locale)
            for cat_id in category_ids
        ]
        results = await asyncio.gather(*tasks, return_exceptions=False)

    return [r for r in results if r is not None]


# --- Spike Detection & Scoring ---

def compute_opportunity_score(item: Dict) -> float:
    """
    Composite opportunity score [0-100].
    Higher = stronger gain + lower competition barrier.
    """
    gain = item.get("rank_gain_pct", 0)
    reviews = item.get("review_count") or 9999
    rating = item.get("rating") or 0

    gain_score = min(50.0, gain / 200)

    if reviews < 50:
        comp_score = 30.0
    elif reviews < 200:
        comp_score = 20.0
    elif reviews < 500:
        comp_score = 8.0
    else:
        comp_score = 0.0

    rating_score = max(0.0, (rating - 3.0) / 2.0 * 20) if rating >= 3.0 else 0.0

    return round(gain_score + comp_score + rating_score, 2)


def extract_alerts(
    results: List[Dict],
    gain_threshold: float = 1000.0,
    min_score: float = 40.0
) -> List[Dict]:
    """
    Extract and score alert items across all collected categories.
    Returns sorted list of high-opportunity items.
    """
    alerts = []
    for result in results:
        cat_id = result["category_id"]
        for item in result["items"]:
            if item.get("rank_gain_pct", 0) < gain_threshold:
                continue
            score = compute_opportunity_score(item)
            if score < min_score:
                continue
            alerts.append({
                "score": score,
                "category_id": cat_id,
                "asin": item["asin"],
                "title": item.get("title", "")[:70],
                "gain_pct": item["rank_gain_pct"],
                "current_rank": item.get("current_rank"),
                "review_count": item.get("review_count"),
                "price": item.get("price"),
                "rating": item.get("rating"),
                "listing_url": item.get("listing_url"),
                "detected_at": datetime.now(timezone.utc).isoformat()
            })

    return sorted(alerts, key=lambda x: x["score"], reverse=True)


# --- Main Orchestration ---

WATCH_CATEGORIES = [
    "1055398",   # Kitchen & Dining
    "3375251",   # Sports & Outdoors
    "1063498",   # Home Storage
    "172282",    # Electronics
    "284507",    # Tools & Home Improvement
]

async def run_pipeline(interval_minutes: int = 30):
    print(f"[MnS Pipeline] Monitoring {len(WATCH_CATEGORIES)} categories "
          f"every {interval_minutes} min")

    while True:
        start = datetime.now(timezone.utc)
        print(f"\n[{start.strftime('%Y-%m-%d %H:%M UTC')}] Starting collection cycle...")

        results = await collect_all_categories(WATCH_CATEGORIES)
        total_items = sum(len(r["items"]) for r in results)
        print(f"  Collected: {total_items} items across {len(results)} categories")

        alerts = extract_alerts(results)
        print(f"  Alerts: {len(alerts)} high-opportunity items detected")

        for alert in alerts[:5]:
            print(
                f"  🎯 Score {alert['score']} | ASIN {alert['asin']} | "
                f"+{alert['gain_pct']:.0f}% | BSR #{alert['current_rank']} | "
                f"Reviews: {alert['review_count']} | {alert['title']}"
            )

        # TODO: Integrate downstream:
        # await send_slack_notification(alerts)
        # await write_to_postgres(results)
        # await trigger_erp_reorder_check(alerts)

        elapsed = (datetime.now(timezone.utc) - start).total_seconds()
        sleep_time = max(0, interval_minutes * 60 - elapsed)
        print(f"  Cycle completed in {elapsed:.1f}s. Next run in {sleep_time:.0f}s.")
        await asyncio.sleep(sleep_time)


if __name__ == "__main__":
    asyncio.run(run_pipeline(interval_minutes=30))

Production Deployment Notes

Environment setup:

# .env
PANGOLINFO_API_KEY=your_key_here

# Run with systemd or Docker
docker run -d \
  --env-file .env \
  --restart unless-stopped \
  python:3.11-slim \
  python main.py

Redis deduplication (optional but recommended):

import redis

r = redis.Redis(host="localhost", port=6379, db=0)

def is_new_alert(asin: str, category_id: str, ttl_seconds: int = 7200) -> bool:
    """Returns True if this alert hasn't been seen in the last `ttl_seconds`."""
    key = f"mns_alert:{category_id}:{asin}"
    if r.exists(key):
        return False
    r.setex(key, ttl_seconds, "1")
    return True

Rate limit handling:
The Pangolinfo API handles Amazon-side rate limits internally. The only rate limit to manage is the Pangolinfo API's own request quota, which is generous for standard tiers. For 50 categories × every 30 minutes = 2,400 requests/day — well within free tier limits for initial testing.

References & Further Reading

Pangolinfo Scrape API Documentation — Full field reference for MnS and other Amazon page types
AMZ Data Tracker — No-code visual dashboard for the same MnS monitoring use case
Amazon MnS base URL: https://www.amazon.com/gp/movers-and-shakers/

Found this useful? Drop a ❤️ or comment with questions. Happy to discuss async optimization strategies or downstream integration patterns.

Find Winning Products on Amazon via API: A Complete Technical Guide (2026)

Mox Loop — Mon, 20 Apr 2026 07:26:09 +0000

TL;DR

Stop relying on SaaS tools with 24-72 hour data lag to find winning Amazon products. Build a real-time discovery pipeline with the Pangolinfo Scrape API instead — faster data, fully customizable filters, no information homogenization problem. Here's the complete implementation.

The Problem With Standard Product Research Tools

Every conventional Amazon research tool — Helium 10, Jungle Scout, and their peers — runs on a crawl-and-cache model. Amazon data is scraped at intervals (typically once every 24-48 hours), processed, stored in a centralized database, and served to all subscribers through a shared interface.

That means:

Data you see is 24-72 hours old before you can act on it
Every subscriber is looking at the same data at the same time
There's no structural advantage to using the same tool as everyone else

In a 2026 Amazon marketplace where category competition compounds rapidly, "discovering" an opportunity that ten thousand other tool subscribers are also seeing is not a discovery. It's a race to second place.

The fix: find winning products on Amazon via API — query Amazon's public data directly, in real time, on your own cadence.

What You'll Build

A three-layer automated product discovery pipeline:

Layer 1: Category bestseller scanning (real-time BSR + review data)
    ↓
Layer 2: Keyword competition analysis (ad density, organic competition)
    ↓
Layer 3: Negative review mining (competitor product gap identification)
    ↓
Output: Ranked shortlist of blue ocean candidates with opportunity scores

Stack: Python 3.11+, Pangolinfo Scrape API, pandas, requests

Setup

pip install requests pandas python-dotenv

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("PANGOLINFO_API_KEY")
BASE_URL = "https://api.pangolinfo.com/v1/scrape"
REVIEWS_URL = "https://api.pangolinfo.com/v1/reviews"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

Get your API key at tool.pangolinfo.com.

Layer 1: Real-Time Bestseller Scanner

# scanner.py
import requests
import time
from config import BASE_URL, HEADERS


def scan_bestsellers(
    category_node: str,
    marketplace: str = "US",
    bsr_range: tuple = (20, 500),
    max_reviews: int = 500
) -> list[dict]:
    """
    Scan a category node's bestseller list in real time.
    Returns filtered candidates meeting BSR range and review ceiling.

    Args:
        category_node: Amazon category path, e.g. 'kitchen/coffee-makers'
        marketplace: Amazon marketplace code
        bsr_range: Tuple of (min_bsr, max_bsr) for filtering
        max_reviews: Maximum review count to include (lower = less competition)
    """
    payload = {
        "url": f"https://www.amazon.com/best-sellers/{category_node}",
        "parse_type": "bestsellers",
        "marketplace": marketplace,
        "include_sponsored": True,
        "output_format": "json"
    }

    resp = requests.post(BASE_URL, json=payload, headers=HEADERS, timeout=30)
    resp.raise_for_status()
    data = resp.json()

    candidates = []
    for p in data.get("products", []):
        bsr = p.get("bsr_rank", 99999)
        reviews = p.get("review_count", 99999)

        if bsr_range[0] <= bsr <= bsr_range[1] and reviews <= max_reviews:
            candidates.append({
                "asin": p["asin"],
                "title": p.get("title", ""),
                "bsr_rank": bsr,
                "review_count": reviews,
                "rating": p.get("rating", 0.0),
                "price": p.get("price", 0.0),
                "bsr_7d_change": p.get("bsr_7d_change"),  # positive = improving rank
                "source_category": category_node
            })

    return candidates


def scan_multiple_categories(
    categories: list[str],
    delay_between: float = 0.5,
    **filter_kwargs
) -> list[dict]:
    """Scan multiple category nodes with rate limiting."""
    all_candidates = []
    for cat in categories:
        try:
            results = scan_bestsellers(cat, **filter_kwargs)
            all_candidates.extend(results)
            print(f"  [{cat}] → {len(results)} candidates")
        except requests.RequestException as e:
            print(f"  [{cat}] → Error: {e}")
        time.sleep(delay_between)
    return all_candidates

Layer 2: Keyword Competition Analyzer

# competition.py
import requests
from config import BASE_URL, HEADERS


def analyze_keyword(keyword: str, marketplace: str = "US") -> dict:
    """
    Fetch keyword search results page and calculate competition metrics.

    Key metric: ad_density = sponsored_listings / total_results
    Lower density = lower cost-per-click competition = easier organic entry
    """
    payload = {
        "keyword": keyword,
        "parse_type": "search_results",
        "marketplace": marketplace,
        "include_ads": True,
        "output_format": "json"
    }

    resp = requests.post(BASE_URL, json=payload, headers=HEADERS, timeout=30)
    resp.raise_for_status()
    data = resp.json()

    total = data.get("total_results", 1)
    sponsored = data.get("sponsored_count", 0)
    results = data.get("results", [])

    return {
        "keyword": keyword,
        "total_listings": total,
        "sponsored_count": sponsored,
        "ad_density": round(sponsored / max(total, 1), 3),
        "avg_price_top10": round(
            sum(r.get("price", 0) for r in results[:10]) / max(len(results[:10]), 1), 2
        ),
        "top_asins": [r["asin"] for r in results[:5]]
    }


def batch_keyword_analysis(keywords: list[str]) -> dict[str, dict]:
    """Analyze multiple keywords and return aggregated competition data."""
    results = {}
    for kw in keywords:
        try:
            results[kw] = analyze_keyword(kw)
        except Exception as e:
            print(f"  Keyword '{kw}' failed: {e}")
    return results

Layer 3: Negative Review Miner

# review_miner.py
import requests
from collections import Counter
from config import REVIEWS_URL, HEADERS

# Complaint taxonomy for Amazon product categories
COMPLAINT_TAXONOMY = {
    "durability":     ["broke", "broken", "stopped working", "defective", "cracked", "damaged"],
    "ease_of_use":    ["hard to use", "complicated", "confusing", "difficult", "not intuitive"],
    "missing_feature":["wish it had", "should include", "missing", "lacks", "no option for", "would be better if"],
    "size_fit":       ["too small", "too large", "doesn't fit", "wrong size", "bulky"],
    "value":          ["overpriced", "not worth", "too expensive", "cheap quality", "waste of money"],
    "reliability":    ["inconsistent", "unreliable", "sometimes works", "random", "hit or miss"]
}


def mine_competitor_gaps(asin: str, max_pages: int = 5) -> dict:
    """
    Extract and categorize negative reviews for a competitor ASIN.
    Returns complaint category counts — higher count = bigger unmet need.
    """
    payload = {
        "asin": asin,
        "filter_star_rating": "three_star_and_below",
        "sort_by": "helpful",
        "max_pages": max_pages,
        "output_format": "json"
    }

    resp = requests.post(REVIEWS_URL, json=payload, headers=HEADERS, timeout=60)
    resp.raise_for_status()
    reviews = resp.json().get("reviews", [])

    complaint_counts = Counter()
    for review in reviews:
        text = (review.get("title", "") + " " + review.get("body", "")).lower()
        for category, keywords in COMPLAINT_TAXONOMY.items():
            if any(kw in text for kw in keywords):
                complaint_counts[category] += 1

    return {
        "asin": asin,
        "total_negative_reviews_analyzed": len(reviews),
        "complaint_breakdown": dict(complaint_counts.most_common()),
        "primary_gap": complaint_counts.most_common(1)[0][0] if complaint_counts else None
    }

Putting It All Together

# main.py
import pandas as pd
from scanner import scan_multiple_categories
from competition import batch_keyword_analysis
from review_miner import mine_competitor_gaps


def opportunity_score(product: dict, avg_ad_density: float) -> float:
    """
    Composite opportunity score (0-100):
      Market traction (BSR position): 40 pts
      Competition headroom (review count): 30 pts
      Growth momentum (BSR trend): 20 pts
      Ad market efficiency (density): 10 pts
    """
    bsr = product.get("bsr_rank", 9999)
    reviews = product.get("review_count", 9999)
    trend = product.get("bsr_7d_change") or 0

    market   = 40 * max(0, 1 - abs(bsr - 175) / 475)
    compete  = 30 * max(0, (500 - reviews) / 500)
    momentum = 20 * min(1.0, max(0, trend / 200))
    ad_eff   = 10 * (1 - avg_ad_density)

    return round(market + compete + momentum + ad_eff, 2)


def run_pipeline(categories: list[str], keywords: list[str]) -> pd.DataFrame:
    print("Phase 1 — Scanning categories...")
    candidates = scan_multiple_categories(
        categories, bsr_range=(30, 400), max_reviews=400
    )
    print(f"  Total candidates: {len(candidates)}")

    print("\nPhase 2 — Analyzing keyword competition...")
    kw_data = batch_keyword_analysis(keywords)
    avg_density = sum(v["ad_density"] for v in kw_data.values()) / len(kw_data)
    print(f"  Average ad density: {avg_density:.1%}")

    print("\nPhase 3 — Scoring and filtering...")
    for c in candidates:
        c["opportunity_score"] = opportunity_score(c, avg_density)

    shortlist = sorted(
        [c for c in candidates if c["opportunity_score"] >= 55],
        key=lambda x: x["opportunity_score"],
        reverse=True
    )

    # Phase 4: Review mining on top 5
    print(f"\nPhase 4 — Mining competitor reviews ({min(5, len(shortlist))} ASINs)...")
    for item in shortlist[:5]:
        gap = mine_competitor_gaps(item["asin"])
        item["primary_product_gap"] = gap["primary_gap"]
        item["gap_count"] = gap["total_negative_reviews_analyzed"]

    return pd.DataFrame(shortlist[:20])


if __name__ == "__main__":
    df = run_pipeline(
        categories=["kitchen/pour-over-coffee", "kitchen/coffee-makers"],
        keywords=["pour over coffee maker", "compact cold brew coffee maker"]
    )
    print("\n🏆 Top 10 Blue Ocean Candidates:")
    print(df[["asin", "bsr_rank", "review_count", "opportunity_score", "primary_product_gap"]].head(10).to_string())
    df.to_csv("discovery_results.csv", index=False)

Performance Tips

Rate limiting: Add time.sleep(0.5) between API calls in production
Caching: Cache keyword competition results daily — they change slowly
Retry logic: Wrap all API calls in exponential backoff (retry up to 3x)
Scheduling: Use APScheduler or Celery for automated daily runs
Storage: Write results to PostgreSQL with timestamps for trend tracking

Scaling Beyond the Script

Once you've validated the pipeline logic, AMZ Data Tracker provides a no-code visual layer on top of the same Pangolinfo data infrastructure — useful for sharing category monitoring views with non-technical teammates.

For AI-driven automation, Pangolinfo also provides an Amazon Scraper Skill compatible with MCP-protocol AI agents, enabling natural-language product research queries without manual pipeline management.

Resources

Questions or implementation issues? Drop them in the comments — happy to help debug.

Real-Time Amazon Product Research Without Jungle Scout or Helium 10: A Complete API Integration Guide

Mox Loop — Tue, 14 Apr 2026 07:53:44 +0000

TL;DR

Jungle Scout and Helium 10 serve as Amazon product research tools, but their subscription architectures introduce 24–72 hour data lag and zero AI Agent integration capability. If you need real-time Amazon price/BSR/keyword data that your AI Agent or automation pipeline can actually call programmatically — this guide shows you how.

What you'll build:

A Python client for Pangolinfo Scrape API (real-time Amazon data)
Batch ASIN monitoring with async collection
Feishu/Slack alert integration for price movement detection
OpenClaw AI Agent integration via Amazon Scraper Skill

Prerequisites: Basic Python, a free Pangolinfo API key, and optionally OpenClaw for the AI Agent section.

Why Subscription Tools Don't Work for Programmatic Use

If you've tried to build any kind of automated Amazon monitoring or AI workflow, you've already hit this wall: subscriptions tools like Jungle Scout and Helium 10 don't provide a usable API for programmatic access. Their "API" offerings (where they exist at all) give you preprocessed, cached data from their own database — not real-time Amazon data. And their data interfaces are designed for humans clicking through dashboards, not for machines making structured data requests.

The specific limitations that matter for developers and operators building automated systems:

Data freshness: Most subscription tool data is 24–72 hours stale. For price monitoring or competitive intelligence that needs to detect same-day movements, this is disqualifying.

No AI Agent support: You cannot configure an AI Agent (LangChain, OpenClaw, Dify) to call Jungle Scout or Helium 10 for live data. There's no function calling interface, no webhook support, no real-time data feed. The only workaround is manual CSV export → upload to AI → analysis on stale data.

Closed data model: You get the fields the tool company decided to expose. Cross-tabulating coupon activation frequency against BSR trajectory, or pulling SP advertising position distribution alongside organic keyword rankings for the same query, isn't possible through any subscription tool.

Pangolinfo Scrape API solves all three by taking a different architectural approach: every API call triggers a live real-time request to Amazon's public pages. You get current-state data, structured JSON output, full programmatic access, and optional Skill-based AI Agent integration.

Setup

# Install dependencies
pip install requests aiohttp python-dotenv

# Create .env file
echo "PANGOLINFO_API_KEY=your_key_here" > .env
echo "FEISHU_WEBHOOK_URL=https://open.feishu.cn/..." >> .env

Get your free API key from tool.pangolinfo.com — no credit card required for the trial tier.

1. Basic Sync Client

Start with the simplest working implementation:

import os
import requests
from dotenv import load_dotenv
from typing import Optional, Dict

load_dotenv()

class PangolinClient:
    """Minimal synchronous client for Pangolinfo Scrape API"""

    BASE_URL = "https://api.pangolinfo.com/v2"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.getenv("PANGOLINFO_API_KEY")
        if not self.api_key:
            raise ValueError("API key required: set PANGOLINFO_API_KEY env var")

        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def get_product(self, asin: str, marketplace: str = "amazon.com", 
                    zip_code: str = "90001") -> Dict:
        """
        Fetch real-time Amazon product data.
        Returns: price, BSR, rating, review count, coupon status, timestamp
        """
        response = requests.post(
            f"{self.BASE_URL}/amazon/product",
            json={"asin": asin, "marketplace": marketplace, "zip_code": zip_code},
            headers=self.headers,
            timeout=30
        )
        response.raise_for_status()
        return response.json()

    def get_keyword_results(self, keyword: str, marketplace: str = "amazon.com",
                            top_n: int = 20) -> Dict:
        """
        Fetch real-time keyword search results.
        Returns: ranked ASINs, organic positions, sponsored positions
        """
        response = requests.post(
            f"{self.BASE_URL}/amazon/keyword",
            json={"keyword": keyword, "marketplace": marketplace, "top_n": top_n},
            headers=self.headers,
            timeout=30
        )
        response.raise_for_status()
        return response.json()


# Quick test
if __name__ == "__main__":
    client = PangolinClient()
    data = client.get_product("B09Z8LSMSK")  # replace with real ASIN
    print(f"Price: ${data.get('current_price')}")
    print(f"BSR: #{data.get('bsr_rank')} in {data.get('bsr_category')}")
    print(f"Rating: {data.get('rating')} ({data.get('review_count'):,} reviews)")
    coupon = data.get('coupon_status', {})
    if coupon.get('is_active'):
        print(f"Coupon: {coupon.get('value')} ACTIVE")

2. Async Batch Collection (The Production Version)

For monitoring 20+ ASINs efficiently, async concurrent requests are essential:

import asyncio
import aiohttp
import time
from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class ProductSnapshot:
    asin: str
    price: Optional[float]
    bsr_rank: Optional[int]
    rating: Optional[float]
    review_count: Optional[int]
    coupon_active: bool
    coupon_value: Optional[str]
    collected_at: float
    error: Optional[str] = None


async def fetch_product_async(
    session: aiohttp.ClientSession,
    asin: str,
    api_key: str,
    semaphore: asyncio.Semaphore,
    marketplace: str = "amazon.com"
) -> ProductSnapshot:
    """Async single product fetch with concurrency control"""

    async with semaphore:  # Limits concurrent requests
        try:
            async with session.post(
                "https://api.pangolinfo.com/v2/amazon/product",
                json={"asin": asin, "marketplace": marketplace},
                headers={"Authorization": f"Bearer {api_key}"},
                timeout=aiohttp.ClientTimeout(total=45)
            ) as response:

                if response.status == 429:
                    await asyncio.sleep(5)
                    return await fetch_product_async(session, asin, api_key, semaphore)

                response.raise_for_status()
                data = await response.json()

                coupon = data.get("coupon_status", {})
                return ProductSnapshot(
                    asin=asin,
                    price=data.get("current_price"),
                    bsr_rank=data.get("bsr_rank"),
                    rating=data.get("rating"),
                    review_count=data.get("review_count"),
                    coupon_active=coupon.get("is_active", False),
                    coupon_value=coupon.get("value"),
                    collected_at=time.time()
                )

        except Exception as e:
            return ProductSnapshot(
                asin=asin, price=None, bsr_rank=None, rating=None,
                review_count=None, coupon_active=False, coupon_value=None,
                collected_at=time.time(), error=str(e)
            )


async def batch_fetch(
    asin_list: List[str],
    api_key: str,
    max_concurrent: int = 8
) -> Dict[str, ProductSnapshot]:
    """
    Concurrent batch product data collection.
    100 ASINs typically completes in 20–60 seconds.
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_product_async(session, asin, api_key, semaphore)
            for asin in asin_list
        ]
        results = await asyncio.gather(*tasks)

    return {snapshot.asin: snapshot for snapshot in results}


# Example usage
async def main():
    api_key = os.getenv("PANGOLINFO_API_KEY")

    # Monitor 30 competitor ASINs
    competitor_asins = ["B0COMP001", "B0COMP002", "B0COMP003"]  # Your real ASINs

    print(f"Fetching {len(competitor_asins)} ASINs concurrently...")
    start = time.time()

    snapshots = await batch_fetch(competitor_asins, api_key)

    elapsed = time.time() - start
    success = sum(1 for s in snapshots.values() if s.error is None)
    print(f"Done: {success}/{len(competitor_asins)} succeeded in {elapsed:.1f}s")

    for asin, snap in list(snapshots.items())[:3]:
        if snap.error:
            print(f"  {asin}: ERROR - {snap.error}")
        else:
            print(f"  {asin}: ${snap.price} | BSR #{snap.bsr_rank} | {snap.rating}⭐")


asyncio.run(main())

3. Price Movement Detection + Feishu Alerts

import json
import requests
from datetime import datetime

FEISHU_WEBHOOK = os.getenv("FEISHU_WEBHOOK_URL")

def detect_price_changes(
    current_snapshots: Dict[str, ProductSnapshot],
    previous_snapshots: Dict[str, ProductSnapshot],
    threshold_pct: float = 5.0
) -> List[Dict]:
    """Compare two snapshots, return list of alert events"""

    alerts = []

    for asin, current in current_snapshots.items():
        previous = previous_snapshots.get(asin)
        if not previous or current.error:
            continue

        # Price change detection
        if current.price and previous.price and previous.price > 0:
            pct_change = ((current.price - previous.price) / previous.price) * 100
            if abs(pct_change) >= threshold_pct:
                alerts.append({
                    "type": "price_change",
                    "asin": asin,
                    "prev_price": previous.price,
                    "curr_price": current.price,
                    "pct_change": round(pct_change, 2),
                    "direction": "drop" if pct_change < 0 else "increase",
                    "curr_bsr": current.bsr_rank,
                    "severity": "high" if abs(pct_change) >= 15 else "medium"
                })

        # Coupon activation detection
        if current.coupon_active and not previous.coupon_active:
            alerts.append({
                "type": "coupon_activated",
                "asin": asin,
                "coupon_value": current.coupon_value,
                "curr_price": current.price,
                "curr_bsr": current.bsr_rank
            })

    return alerts


def send_feishu_alert(alerts: List[Dict]):
    """Push price alerts to Feishu bot"""
    if not alerts or not FEISHU_WEBHOOK:
        return

    text = f"🔔 **Amazon Competitor Alert** — {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"

    for alert in alerts:
        asin = alert['asin']
        link = f"https://www.amazon.com/dp/{asin}"

        if alert['type'] == 'price_change':
            direction = "⬇ Price Drop" if alert['direction'] == 'drop' else "⬆ Price Increase"
            text += (
                f"{'🔴' if alert['severity'] == 'high' else '🟡'} **{direction}** "
                f"[{asin}]({link})\n"
                f"${alert['prev_price']:.2f} → ${alert['curr_price']:.2f} "
                f"({alert['pct_change']:+.1f}%) | BSR #{alert.get('curr_bsr', 'N/A')}\n\n"
            )
        elif alert['type'] == 'coupon_activated':
            text += (
                f"🟡 **Coupon Activated** [{asin}]({link})\n"
                f"Value: {alert['coupon_value']} | "
                f"Price: ${alert.get('curr_price', 'N/A'):.2f} | "
                f"BSR #{alert.get('curr_bsr', 'N/A')}\n\n"
            )

    payload = {
        "msg_type": "text",
        "content": {"text": text}
    }
    requests.post(FEISHU_WEBHOOK, json=payload, timeout=10)
    print(f"✓ Sent {len(alerts)} alerts to Feishu")

4. AI Agent Integration via Amazon Scraper Skill

The fastest path to getting your AI Agent access to real-time Amazon data is the Pangolinfo Amazon Scraper Skill for OpenClaw.

Setup: Find in OpenClaw Skill library → Install → Enter API key → Done.

Example Agent conversations that now work:

You: Pull realtime data for keyword "portable espresso maker" — 
     top 15 ASINs, price range, rating distribution, BSR ranking.
     Which ASINs are rated below 4.0 and what are their main complaint themes?

Agent: [Calls Scraper Skill → Returns structured analysis with live data]

You: Monitor ASINs [B0XX001, B0XX002, B0XX003].
     Flag any that have activated coupons or changed price >5% since yesterday.

Agent: [Compares current vs cached data → Delivers change summary]

LangChain custom tool (for non-OpenClaw Agent frameworks):

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class ProductQueryInput(BaseModel):
    asin: str = Field(description="Amazon ASIN to fetch real-time data for")
    marketplace: str = Field(default="amazon.com")

class AmazonProductTool(BaseTool):
    name = "amazon_realtime_product_data"
    description = (
        "Fetch real-time Amazon product data: current price, BSR ranking, "
        "rating, review count, coupon status. Returns live data from Amazon, "
        "not cached. Use for competitor monitoring and product research."
    )
    args_schema = ProductQueryInput
    api_key: str

    def _run(self, asin: str, marketplace: str = "amazon.com") -> str:
        import requests
        resp = requests.post(
            "https://api.pangolinfo.com/v2/amazon/product",
            json={"asin": asin, "marketplace": marketplace},
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=30
        )
        d = resp.json()
        coupon = d.get("coupon_status", {})
        return (
            f"ASIN {asin} — Live data as of {d.get('timestamp', 'now')}:\n"
            f"Price: ${d.get('current_price', 'N/A')}\n"
            f"BSR: #{d.get('bsr_rank', 'N/A')} ({d.get('bsr_category', '')})\n"
            f"Rating: {d.get('rating')} ({d.get('review_count', 0):,} reviews)\n"
            f"Coupon: {'ACTIVE — ' + str(coupon.get('value','')) if coupon.get('is_active') else 'None'}"
        )

    async def _arun(self, **kwargs): raise NotImplementedError

Full Monitoring Script (Bring It All Together)

#!/usr/bin/env python3
"""
Complete Amazon competitor monitoring system
Runs on schedule, detects price/coupon changes, sends Feishu alerts
"""

import asyncio
import json
import os
import schedule
import time
from datetime import datetime

PANGOLINFO_API_KEY = os.getenv("PANGOLINFO_API_KEY")
FEISHU_WEBHOOK = os.getenv("FEISHU_WEBHOOK_URL")

MONITOR_ASINS = [
    "B0COMP001",  # Replace with your actual competitor ASINs
    "B0COMP002",
    "B0COMP003",
]

previous_snapshots: Dict[str, ProductSnapshot] = {}  # In-memory cache (use DB in production)


async def monitoring_cycle():
    print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Running monitoring cycle...")

    global previous_snapshots

    # Fetch current state
    current = await batch_fetch(MONITOR_ASINS, PANGOLINFO_API_KEY)

    # Detect changes if we have previous data
    if previous_snapshots:
        alerts = detect_price_changes(current, previous_snapshots, threshold_pct=5.0)
        if alerts:
            send_feishu_alert(alerts)
            print(f"  Detected {len(alerts)} changes — alerts sent")
        else:
            print(f"  No significant changes detected")
    else:
        print(f"  First run — establishing baseline snapshot")

    # Update cache
    previous_snapshots = current

    success = sum(1 for s in current.values() if s.error is None)
    print(f"  Snapshot complete: {success}/{len(MONITOR_ASINS)} ASINs collected")


def main():
    print("Amazon Competitor Monitor — Starting")
    print(f"Monitoring {len(MONITOR_ASINS)} ASINs every 2 hours")

    # Run immediately on start
    asyncio.run(monitoring_cycle())

    # Schedule subsequent runs
    schedule.every(2).hours.do(lambda: asyncio.run(monitoring_cycle()))

    while True:
        schedule.run_pending()
        time.sleep(60)


if __name__ == "__main__":
    main()

Deployment

# Local background process
nohup python monitor.py &

# Or Docker
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "monitor.py"]

Key Takeaways

Subscription tools (Jungle Scout, Helium 10) have 24–72 hour data lag by architectural design — not fixable by upgrading your plan
Real-time API calls via Pangolinfo return actual current Amazon data, enabling true competitive monitoring
The async batch pattern handles 100 ASINs in 20–60 seconds
AI Agent integration works natively via Amazon Scraper Skill (OpenClaw) or as a LangChain custom tool
Total monitoring stack cost at moderate volume: ~$30–50/month vs $100–200+/month for equivalent subscription coverage

Free trial (no credit card): tool.pangolinfo.com
API docs: docs.pangolinfo.com

Async vs Sync Amazon Data Scraping API: A Complete Developer Guide

Mox Loop — Mon, 13 Apr 2026 07:01:05 +0000

TL;DR

Sync API: ~5s per call, blocks until result returns. Use for real-time, user-triggered queries.
Async API: submits task in <200ms, result delivered via callback. Use for batch/scheduled work.
Same credit cost both ways (1 credit/JSON call)
Pangolinfo Scrape API supports both modes under the same token

The Problem with Defaulting to Sync

Amazon Scrape API calls take around 5 seconds per response—realistic given what the server has to do (load Amazon pages, bypass anti-bot systems, parse DOM, return structured JSON). For a single query, that's totally fine.

For a monitoring system tracking 5,000 ASINs daily? That's 6.9 hours of serial waiting. Add network variance and you're looking at 8–10 hours. Most business use cases can't absorb that delay.

Async Amazon data scraping breaks the dependency between request throughput and client-side wait time. You submit tasks in bulk, the platform processes them concurrently server-side, and results arrive at your callback endpoint. Your client never blocks.

API Endpoint Comparison

	Sync	Async
Endpoint	`POST /api/v1/scrape`	`POST /api/v1/scrape/async`
Client wait	~5 seconds	<200ms (returns taskId)
Result delivery	Inline response body	POST to your `callbackUrl`
Infrastructure req	None	Public callback endpoint
Credits (JSON)	1/call	1/call

Sync Mode: Complete Example

import requests

def sync_scrape(token: str, asin: str) -> dict | None:
    """
    Sync Amazon product detail scrape.
    Blocks ~5s. Use for real-time queries.
    """
    resp = requests.post(
        "https://scrapeapi.pangolinfo.com/api/v1/scrape",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        },
        json={
            "url": f"https://www.amazon.com/dp/{asin}",
            "parserName": "amzProductDetail",
            "site": "",
            "content": "",
            "format": "json",
            "bizContext": {"zipcode": "10041"}
        },
        timeout=30  # Must exceed API processing time (~5s)
    )
    data = resp.json()
    if data.get("code") == 0:
        return data["data"]["json"][0]["data"]["results"][0]
    raise RuntimeError(f"API error: {data.get('message')}")


# Usage
product = sync_scrape("your_token", "B0DYTF8L2W")
print(product["title"])
print(product["star"])        # e.g. "4.5"
print(product["bestSellersRank"])

Available Parser Types

PARSER_NAMES = {
    "amzProductDetail":    "Product detail by ASIN or URL",
    "amzKeyword":          "Keyword search results",
    "amzProductOfCategory": "Category product list (by Node ID)",
    "amzProductOfSeller":  "Seller product list",
    "amzBestSellers":      "Best sellers ranking",
    "amzNewReleases":      "New releases ranking"
}

Async Mode: Complete Implementation

1. Submit Tasks

import requests
import time
from typing import Optional

ASYNC_URL = "https://scrapeapi.pangolinfo.com/api/v1/scrape/async"

def async_submit(
    token: str,
    asin: str,
    callback_url: str,
    parser_name: str = "amzProductDetail",
    zipcode: str = "10041"
) -> Optional[str]:
    """Submit async scraping task. Returns task_id immediately."""
    resp = requests.post(
        ASYNC_URL,
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        json={
            "url": f"https://www.amazon.com/dp/{asin}",
            "callbackUrl": callback_url,
            "zipcode": zipcode,
            "format": "json",
            "parserName": parser_name
        },
        timeout=10  # Submission itself is fast
    )
    data = resp.json()
    if data.get("code") == 0:
        return data["data"]["data"]  # task_id
    print(f"Submit failed for {asin}: {data.get('message')}")
    return None


def bulk_submit(
    token: str,
    asins: list[str],
    callback_url: str,
    rate_delay: float = 0.1  # seconds between submissions
) -> dict[str, str]:
    """Bulk submit. Returns {asin: task_id}."""
    task_map = {}
    for i, asin in enumerate(asins):
        task_id = async_submit(token, asin, callback_url)
        if task_id:
            task_map[asin] = task_id
        if (i + 1) % 500 == 0:
            print(f"Submitted {i+1}/{len(asins)}")
        time.sleep(rate_delay)
    return task_map

2. Callback Receiver (FastAPI)

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()
processed_ids: set[str] = set()  # Use Redis in production

@app.post("/api/callback")
async def receive_amazon_callback(request: Request):
    payload = await request.json()
    task_id = payload.get("taskId")

    # Idempotency guard - critical for production
    if task_id in processed_ids:
        return JSONResponse({"code": 0, "message": "duplicate"})
    processed_ids.add(task_id)

    # Extract results
    results = (
        payload
        .get("data", {})
        .get("json", [{}])[0]
        .get("data", {})
        .get("results", [])
    )
    if results:
        product = results[0]
        print(f"Received: {product.get('title', 'N/A')[:60]}")
        # persist_to_db(task_id, product)

    return JSONResponse({"code": 0, "message": "ok"})

3. Timeout Compensation

def compensate_missing_callbacks(
    token: str,
    task_map: dict[str, str],
    received_ids: set[str]
) -> None:
    """
    For tasks that didn't callback within expected window,
    fetch results via sync API as fallback.
    """
    missing = {asin: tid for asin, tid in task_map.items()
               if tid not in received_ids}

    for asin, task_id in missing.items():
        print(f"Compensating {asin} (task {task_id})")
        result = sync_scrape(token, asin)  # fallback to sync
        if result:
            print(f"Compensated: {result.get('title', 'N/A')[:40]}")

Decision Framework

Daily task volume?
│
├── < 100 tasks/day
│   └── Need instant results?
│       ├── Yes → Sync API ✓
│       └── No  → Sync API ✓ (async overhead not worth it)
│
└── ≥ 100 tasks/day OR scheduled batches
    └── Can deploy public callback server?
        ├── Yes → Async API ✓ (10x+ throughput)
        └── No  → Sync + threading, or AMZ Data Tracker (no-code)

Local Development Setup

Testing async locally without deploying to a server:

# Terminal 1: Run your callback receiver
uvicorn main:app --port 5000

# Terminal 2: Expose it publicly via ngrok
ngrok http 5000
# → Forwarding: https://abc123.ngrok.io → http://localhost:5000

# Use the ngrok URL as your callbackUrl for testing
CALLBACK_URL = "https://abc123.ngrok.io/api/callback"

Performance Benchmarks

Task Count	Sync (single-thread)	Async
100	~8 min	~2 min
1,000	~83 min	~5-10 min
5,000	~7 hrs	~20-40 min
10,000	~14 hrs	~45-90 min

Credit cost is identical in both modes. Performance gap grows linearly with volume.

Resources

Pangolinfo Scrape API — supports both sync and async
API Documentation — complete parameter reference
AMZ Data Tracker — no-code scheduled data collection

Questions? Drop them in the comments.

From 1M Records/Month to 10M/Day: A Real Case Study in Amazon Data Infrastructure

Mox Loop — Wed, 08 Apr 2026 07:25:21 +0000

Who This Is For

If you're building an Amazon seller tool, a data analytics product, or any SaaS where web data is the core value proposition — this case study is probably directly relevant to your stack decisions. It's not theoretical. It's a documented infrastructure migration, including the POC numbers, migration approach, and six-month outcomes.

Background

SellerPulse (anonymized) is a leading Amazon seller tool platform: 32,000+ registered users, ~18% paid conversion, MRR in the $1M+ range. Core product: real-time competitor price monitoring and SP ad placement tracking.

By late 2024, their self-built scraping infrastructure had become the company's biggest operational liability:

200 scraper nodes + 7 dedicated maintenance engineers
$12,000/month in IP infrastructure spend
SP ad slot capture rate: 62% (38% of their most-valued data type was simply missing)
Average data latency: 52 hours (their marketing said "real-time")
Quarterly outages triggered by Amazon anti-scraping policy updates

The Infrastructure Problem, Technically

Amazon's anti-bot systems underwent a significant architectural shift in 2024. Before this, IP rotation + reasonable request throttling was largely sufficient. Post-2024, Amazon deployed behavioral fingerprinting and session continuity verification — meaning that even with rotating IPs, requests exhibiting non-human session patterns (wrong resource loading order, atypical scroll behavior, browser fingerprint anomalies) are systematically identified and served degraded content.

For data center IPs specifically, this produces a systemic SP ad capture problem: Amazon's ad auction system serves authentic ad results based on perceived request legitimacy. Requests from data center ranges get systematically stripped of ad content or receive placeholder responses. That's the architectural reason SellerPulse's capture rate was stuck at 62% — not bad engineering, but a fundamental infrastructure constraint.

The fix requires:

Residential-grade IP infrastructure (looks like real user traffic to Amazon's systems)
Full browser rendering (JS execution, realistic resource loading sequence, scroll simulation)
Session continuity (consistent browser fingerprint across a session lifecycle)

These three requirements together are expensive and complex to maintain at scale for a company whose core competency is building a seller tool product, not running residential proxy networks.

The Solution: Pangolinfo Scrape API

After a two-week POC on their actual production URL sets, the numbers:

Metric	Self-built	Pangolinfo
SP ad slot capture rate	62%	97.3%
Data latency (P95)	52 hours	18 minutes
Zip-code targeting	Not supported	Supported natively
Customer Says field	Not supported	91.2% coverage
Pricing model	Fixed high cost	Pay-per-use

The zip-code and Customer Says findings were surprises. SellerPulse hadn't been able to do region-specific ad collection at all — Pangolinfo's native geographic targeting support opened entirely new product feature directions. The Customer Says field (Amazon's AI-generated review sentiment summary) had been completely inaccessible to self-managed scrapers; at 91%+ coverage it immediately became a candidate for a new competitive intelligence feature.

The Implementation: 90-Day Phased Migration

Core Approach: Traffic Splitting + Parallel Comparison

No downtime. No big-bang cutover. Traffic splitting from day one.

# Simplified traffic routing logic
import random

def route_scrape_request(url: str, pangolinfo_ratio: float = 0.20) -> str:
    """
    Route scraping requests between legacy and Pangolinfo systems.
    pangolinfo_ratio increases from 0.20 → 1.00 over 60 days.
    """
    if random.random() < pangolinfo_ratio:
        return "pangolinfo"
    return "legacy"

# Week 1-2: 20% Pangolinfo
# Week 3-4: 40% Pangolinfo  
# Week 5-6: 60% Pangolinfo
# Week 7-8: 80% Pangolinfo
# Week 9+:  100% Pangolinfo (core pipelines)

Parallel Comparison Layer

Both systems run simultaneously during migration; a comparison service logs divergence for manual review:

import hashlib
import json
from dataclasses import dataclass
from typing import Optional

@dataclass
class ComparisonResult:
    url: str
    legacy_fingerprint: Optional[str]
    pangolinfo_fingerprint: Optional[str]
    match: bool
    divergent_fields: list

def compare_outputs(legacy_data: dict, pangolinfo_data: dict, key_fields: list) -> ComparisonResult:
    """Compare outputs from both systems on key fields."""

    def fingerprint(data, fields):
        subset = {k: data.get(k) for k in fields if data}
        return hashlib.md5(json.dumps(subset, sort_keys=True).encode()).hexdigest()

    lf = fingerprint(legacy_data, key_fields)
    pf = fingerprint(pangolinfo_data, key_fields)

    divergent = []
    if legacy_data and pangolinfo_data:
        for field in key_fields:
            if legacy_data.get(field) != pangolinfo_data.get(field):
                divergent.append(field)

    return ComparisonResult(
        url=legacy_data.get("url", "") if legacy_data else "",
        legacy_fingerprint=lf,
        pangolinfo_fingerprint=pf,
        match=(lf == pf),
        divergent_fields=divergent
    )

Phase 3 — New Feature Rollout (Days 61-90)

With the core pipeline complete, features that had been blocked by infrastructure constraints shipped in rapid succession:

# Customer Says extraction — previously impossible with self-built scrapers
def extract_customer_says(product_data: dict) -> dict:
    """
    Extract Amazon's AI-generated review summary (Customer Says field)
    Available in Pangolinfo structured JSON output by default
    """
    customer_says = product_data.get("customer_says", {})

    return {
        "positive_attributes": customer_says.get("positive", []),
        "negative_attributes": customer_says.get("negative", []),
        "common_mentions": customer_says.get("common_themes", []),
        "sentiment_score": customer_says.get("overall_sentiment"),
        "mention_count": customer_says.get("review_count", 0)
    }

# Full category New Releases monitoring — previously too expensive to run at scale
def monitor_new_releases(categories: list, api_key: str):
    """Now economically viable at commodity per-record rates"""
    from pangolin_client import PangolinScrapeClient, ScrapeConfig

    client = PangolinScrapeClient(api_key)
    results = []

    for category in categories:
        config = ScrapeConfig(
            url=f"https://www.amazon.com/gp/new-releases/{category}/",
            parse_template="amazon_new_releases",
            extract_fields=["rank", "asin", "title", "price", "rating", "launch_date", "badge"]
        )
        data = client.scrape(config)
        if data:
            results.append(data)

    return results

Results: 6 Months Post-Migration

BEFORE vs AFTER

Daily Records:      29,000          →    12,000,000   (+41,000%)
SP Ad Capture:      62%             →    98.1%         (+36.1ppt)
Latency (avg):      52 hours        →    13 minutes    (-99.6%)
Customer Says:      0%              →    91%           (new capability)
Monthly Cost:       ~$30,000        →    ~$13,500      (-55%)
                    (at 870K/mo)          (at 330M/mo)

Paid Conversion:    18.3%           →    22.7%         (+4.4ppt)
Monthly Churn:      11.8%           →    6.4%          (-5.4ppt)
Monthly ARR:        $1.1M           →    $1.57M        (+42.5%)

Engineering (crawler maintenance): -85%
New features shipped (30 days post-migration): 4
Annualized ROI: 14.3x

Dev Notes: What Actually Tripped Us Up

Zip code matters more than you'd expect. Different US zip codes can show materially different prices (shipping cost locality), different ad inventories (regional ad targeting), and different Prime badge availability. For any feature involving geographic competitive analysis, setting geo.zip_code correctly isn't optional.

Customer Says field isn't on every ASIN. Amazon's AI review summaries require a threshold of reviews before generating. Expect ~8-12% of ASINs to return empty for this field — handle gracefully, don't treat as an error.

Concurrency sweet spot. They settled on semaphore-controlled async at 20 concurrent requests for standard tasks, bumped to 50 during peak events with Pangolinfo's expanded allocation. Error rate at 20: <0.1%. Setting it higher without coordination with Pangolinfo's account team caused transient throttling.

Parse templates vs. manual field extraction. For common scenarios (bestsellers, search results, product detail pages), Pangolinfo's prebuilt parse templates save significant engineering time. For unusual ASINs or edge-case category structures, the extract_fields approach with explicit field targeting is more reliable.

Conclusion

The core engineering takeaway: past roughly 1-2M records/day, self-managed scraping infrastructure for Amazon data becomes increasingly difficult to maintain at production quality levels. The combination of behavioral fingerprinting, session continuity requirements, residential IP needs, and the ongoing anti-bot escalation pattern makes it a moving target that academic engineering quality alone can't consistently hit.

For teams considering a similar migration:

Run your own POC first. The numbers from vendor case studies (including this one) are real, but your URL distribution, category mix, and geo targeting requirements will produce your own baseline. Two weeks of POC on your actual production URLs tells you more than any case study.
Plan for a phased migration. Big-bang cutovers on live data pipelines are unnecessary and risky. Traffic splitting plus parallel comparison is the standard approach for good reason.
Budget for Phase 3. The features you'll unlock post-migration are real business value. Factor them into your ROI calculations.

API documentation: docs.pangolinfo.com
Free trial: pangolinfo.com/scrape-api

Tags: #api #python #ecommerce #webdev #scrapers #amazon #dataengineering #saasbuild

Amazon Auto Listing with SP-API: Complete Tutorial with Real-Time Pricing Integration

Mox Loop — Tue, 07 Apr 2026 07:44:57 +0000

Build an auto listing pipeline that doesn't just publish products — it publishes them at the right price, with the right keywords, informed by current market data.

TL;DR

Amazon auto listing via SP-API is approachable for developers who haven't done it before. The hard parts are:

Product Type field schemas vary wildly — always fetch the schema before building payloads
Use PATCH (not PUT) for price updates — it's faster, targeted, and lower risk
Static auto listing is a start, not a finish — wire a real-time data source in from day one

What We're Building

A complete Amazon auto listing system with:

SP-API OAuth authentication with automatic token refresh
Listings Items API for real-time single-SKU listing creation
Feeds API for async batch listing (hundreds of SKUs)
Pangolinfo Scrape API integration for live competitor pricing
30-minute repricing cycle that updates prices based on live market data

Prerequisites

Amazon Professional Seller account
SP-API app registration in Developer Central (with Listings and Feeds role permissions)
LWA Refresh Token from the authorization workflow
Pangolinfo API key (trial at tool.pangolinfo.com)
Python 3.9+

Install dependencies:

pip install requests aiohttp

Part 1: Authentication

"""
sp_auth.py — Amazon SP-API LWA authentication
Handles automatic Access Token refresh
"""
import requests
from datetime import datetime, timedelta
from threading import Lock


class SPAPIAuth:
    TOKEN_URL = "https://api.amazon.com/auth/o2/token"

    def __init__(self, client_id: str, client_secret: str, refresh_token: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self._token = None
        self._expires_at = datetime.min
        self._lock = Lock()

    @property
    def access_token(self) -> str:
        """Returns a valid access token, refreshing automatically if expired."""
        with self._lock:
            if datetime.now() >= self._expires_at:
                self._refresh()
            return self._token

    def _refresh(self):
        resp = requests.post(self.TOKEN_URL, data={
            "grant_type": "refresh_token",
            "refresh_token": self.refresh_token,
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        self._token = data["access_token"]
        self._expires_at = datetime.now() + timedelta(seconds=data["expires_in"] - 60)

Part 2: Listings Items API (Real-Time)

"""
listings_api.py — SP-API Listings Items API wrapper
"""
import requests
from typing import Optional, List
from datetime import datetime


class ListingsItemsAPI:
    BASE = "https://sellingpartnerapi-na.amazon.com"

    def __init__(self, auth: SPAPIAuth, seller_id: str, marketplace_id: str = "ATVPDKIKX0DER"):
        self.auth = auth
        self.seller_id = seller_id
        self.marketplace_id = marketplace_id

    def _headers(self) -> dict:
        return {
            "x-amz-access-token": self.auth.access_token,
            "x-amz-date": datetime.utcnow().strftime("%Y%m%dT%H%M%SZ"),
            "Content-Type": "application/json"
        }

    def get_product_type_schema(self, product_type: str) -> dict:
        """
        IMPORTANT: Always fetch the schema before building listing payloads.
        Each product type has unique required fields — don't skip this step.

        Returns the JSON Schema for the specified product type.
        """
        resp = requests.get(
            f"{self.BASE}/definitions/2020-09-01/productTypes/{product_type}",
            headers=self._headers(),
            params={
                "marketplaceIds": self.marketplace_id,
                "requirements": "LISTING",
                "locale": "en_US"
            }
        )
        resp.raise_for_status()
        schema = resp.json()

        required = schema.get("schema", {}).get("required", [])
        print(f"Product type '{product_type}' requires {len(required)} fields: {required[:8]}...")
        return schema

    def create_listing(
        self,
        seller_sku: str,
        product_type: str,
        attributes: dict
    ) -> dict:
        """
        Create or fully update a listing via PUT.
        Use for new listings or when updating many fields simultaneously.
        """
        url = f"{self.BASE}/listings/2021-08-01/items/{self.seller_id}/{seller_sku}"

        resp = requests.put(
            url,
            headers=self._headers(),
            params={"marketplaceIds": self.marketplace_id},
            json={
                "productType": product_type,
                "requirements": "LISTING",
                "attributes": attributes
            },
            timeout=30
        )

        result = resp.json()
        status = result.get("status", "UNKNOWN")

        if status == "ACCEPTED":
            print(f"✓ {seller_sku}: Created successfully")
        else:
            for issue in result.get("issues", []):
                print(f"✗ {seller_sku} [{issue.get('severity')}] {issue.get('attributeName', '')}: {issue.get('message')}")

        return result

    def update_price(
        self,
        seller_sku: str,
        product_type: str,
        new_price: float
    ) -> dict:
        """
        Update price only via PATCH.

        Why PATCH instead of PUT for pricing?
        - Faster processing (fewer fields to validate)
        - Minimal payload = lower risk of triggering listing re-review
        - Appropriate for high-frequency repricing (every 30 min)
        """
        url = f"{self.BASE}/listings/2021-08-01/items/{self.seller_id}/{seller_sku}"

        resp = requests.patch(
            url,
            headers=self._headers(),
            params={"marketplaceIds": self.marketplace_id},
            json={
                "productType": product_type,
                "patches": [{
                    "op": "replace",
                    "path": "/attributes/purchasable_offer",
                    "value": [{
                        "marketplace_id": self.marketplace_id,
                        "currency": "USD",
                        "our_price": [{"schedule": [{"value_with_tax": new_price}]}]
                    }]
                }]
            },
            timeout=15
        )

        result = resp.json()
        if result.get("status") == "ACCEPTED":
            print(f"✓ {seller_sku}: Price updated to ${new_price:.2f}")
        return result

    def build_standard_attributes(
        self,
        title: str,
        brand: str,
        bullet_points: List[str],
        description: str,
        keywords: List[str],
        price: float,
        quantity: int,
        image_url: str
    ) -> dict:
        """
        Build a listing attributes dict covering most common product types.
        Note: You'll need to add category-specific required fields based on
        the schema returned by get_product_type_schema().
        """
        return {
            "item_name": [{"value": title[:200], "language_tag": "en_US"}],
            "brand": [{"value": brand}],
            "bullet_point": [
                {"value": bp[:500], "language_tag": "en_US"}
                for bp in bullet_points[:5]
            ],
            "product_description": [{"value": description[:2000], "language_tag": "en_US"}],
            "generic_keyword": [{"value": " ".join(keywords[:50])}],
            "fulfillment_availability": [{
                "fulfillment_channel_code": "DEFAULT",
                "quantity": quantity
            }],
            "purchasable_offer": [{
                "marketplace_id": self.marketplace_id,
                "currency": "USD",
                "our_price": [{"schedule": [{"value_with_tax": price}]}]
            }],
            "main_product_image_locator": [{"media_location": image_url}]
        }

Part 3: Real-Time Pricing with Pangolinfo

"""
live_pricing.py — Real-time competitor price monitoring and repricing
Uses Pangolinfo Scrape API to fetch live Amazon product data
"""
import requests
from typing import Optional


class LivePricingEngine:
    PANGOLINFO_URL = "https://api.pangolinfo.com/v1/scrape"

    def __init__(self, pangolinfo_api_key: str, listings_api: ListingsItemsAPI):
        self.pangolinfo_headers = {
            "Authorization": f"Bearer {pangolinfo_api_key}",
            "Content-Type": "application/json"
        }
        self.listings = listings_api

    def fetch_live_price(self, asin: str) -> Optional[float]:
        """
        Fetch real-time price for an Amazon ASIN via Pangolinfo Scrape API.

        Returns the current Buybox price, or None if unavailable.
        Pangolinfo's SP ad position capture rate: 98% — useful for
        understanding how competitively a keyword is being bid on.
        """
        try:
            resp = requests.post(
                self.PANGOLINFO_URL,
                headers=self.pangolinfo_headers,
                json={
                    "url": f"https://www.amazon.com/dp/{asin}",
                    "platform": "amazon",
                    "data_type": "product_detail",
                    "marketplace": "US",
                    "render": True,
                    "extract_ads": True  # Also capture SP ad positions
                },
                timeout=30
            )
            resp.raise_for_status()
            result = resp.json().get("result", {})

            price_data = result.get("price", {})
            current_price = price_data.get("current") if isinstance(price_data, dict) else price_data

            if current_price:
                print(f"  Live price for {asin}: ${current_price:.2f} "
                      f"(Buybox: {result.get('buybox', {}).get('seller_name', 'N/A')})")

            return current_price
        except Exception as e:
            print(f"  Failed to fetch price for {asin}: {e}")
            return None

    def calculate_target_price(
        self,
        competitor_price: float,
        unit_cost: float,
        undercut: float = 0.02,   # Price this % below competitor
        min_margin: float = 0.15  # Never price below this margin
    ) -> float:
        """Calculate optimal price given competitor data and margin constraints."""
        floor = unit_cost / (1 - min_margin)
        target = round(competitor_price * (1 - undercut), 2)
        return max(target, floor)

    def reprice_product(
        self,
        seller_sku: str,
        product_type: str,
        competitor_asin: str,
        unit_cost: float,
        current_price: float
    ) -> bool:
        """
        Check competitor price and update our price if needed.
        Returns True if price was updated.
        """
        comp_price = self.fetch_live_price(competitor_asin)

        if not comp_price:
            return False

        new_price = self.calculate_target_price(comp_price, unit_cost)

        # Only update if price changes by more than $0.01 (avoid unnecessary API calls)
        if abs(new_price - current_price) <= 0.01:
            print(f"  {seller_sku}: No price change needed (${current_price:.2f})")
            return False

        result = self.listings.update_price(seller_sku, product_type, new_price)
        return result.get("status") == "ACCEPTED"

    def run_repricing_cycle(self, products: list) -> dict:
        """
        Full repricing cycle for a list of monitored products.
        Call this every 30 minutes for active price competition.

        products: [{"sku": str, "product_type": str, "competitor_asin": str, 
                    "cost": float, "current_price": float}, ...]
        """
        updated = 0
        skipped = 0
        failed = 0

        for product in products:
            success = self.reprice_product(
                seller_sku=product["sku"],
                product_type=product["product_type"],
                competitor_asin=product["competitor_asin"],
                unit_cost=product["cost"],
                current_price=product["current_price"]
            )

            if success:
                updated += 1
            elif success is False:
                skipped += 1
            else:
                failed += 1

        print(f"\nRepricing cycle complete: {updated} updated, {skipped} unchanged, {failed} failed")
        return {"updated": updated, "skipped": skipped, "failed": failed}

Complete Integration Example

# main.py — Full auto listing + repricing example

auth = SPAPIAuth(
    client_id="YOUR_CLIENT_ID",
    client_secret="YOUR_CLIENT_SECRET",
    refresh_token="YOUR_REFRESH_TOKEN"
)

listings = ListingsItemsAPI(auth, seller_id="YOUR_SELLER_ID")
pricing_engine = LivePricingEngine("YOUR_PANGOLINFO_KEY", listings)

# Step 1: Check what fields this product type actually requires
schema = listings.get_product_type_schema("WATER_BOTTLE")

# Step 2: Build listing attributes
attrs = listings.build_standard_attributes(
    title="Premium 32oz Stainless Steel Water Bottle — Vacuum Insulated, BPA Free",
    brand="YourBrand",
    bullet_points=[
        "STAYS COLD 24H / HOT 12H: Triple-wall vacuum insulation outperforms single-wall alternatives.",
        "BPA-FREE 18/8 STAINLESS STEEL: Zero plastic taste, LFGB and FDA certified.",
        "LEAK-PROOF LID: 360° silicone seal tested at all angles including inverted.",
        "FITS CUP HOLDERS: 2.95\" diameter fits standard car cup holders and most bag pockets.",
        "LIFETIME WARRANTY: No-questions replacement guarantee on every bottle."
    ],
    description="For people who refuse to compromise on hydration quality...",
    keywords=["insulated water bottle", "stainless steel water bottle", "bpa free water bottle",
              "32oz water bottle", "vacuum insulated bottle", "leak proof water bottle"],
    price=32.99,  # Will be overwritten by live competitor data below
    quantity=300,
    image_url="https://cdn.yourbrand.com/wb-32oz-hero.jpg"
)

# Step 3: Get live competitor price and override before submitting
live_comp_price = pricing_engine.fetch_live_price("B08MAINCOMPETITOR")
if live_comp_price:
    attrs["purchasable_offer"] = [{
        "marketplace_id": "ATVPDKIKX0DER",
        "currency": "USD",
        "our_price": [{"schedule": [{"value_with_tax": 
            pricing_engine.calculate_target_price(live_comp_price, unit_cost=9.20)
        }]}]
    }]

# Step 4: Submit to Amazon
result = listings.create_listing("BRAND-WB-32OZ", "WATER_BOTTLE", attrs)

# Step 5: Set up 30-minute repricing cycle (use cron or scheduler)
monitored = [{
    "sku": "BRAND-WB-32OZ",
    "product_type": "WATER_BOTTLE",
    "competitor_asin": "B08MAINCOMPETITOR",
    "cost": 9.20,
    "current_price": 31.65
}]

# Run this every 30 minutes:
pricing_engine.run_repricing_cycle(monitored)

Common Issues

INVALID on submission despite correct syntax:
Almost always a missing required field for the specific product type. Run get_product_type_schema() and check the required array.

Price update accepted but not reflected in Seller Central:
Price changes can take 10-30 minutes to propagate. Check the listing status via GET /listings/2021-08-01/items/{sellerId}/{sellerSku}.

Rate limit errors (429):
Listings Items API: 5 req/s burst, 0.5 req/s sustained. For bulk operations, use Feeds API instead of looping Listings Items PUT.

Feed stuck in IN_PROGRESS for hours:
Check GET /feeds/2021-06-30/feeds/{feedId} — if processingStatus stays IN_PROGRESS beyond 2 hours, there may be a malformed message in the feed causing processing to stall.

Resources

Pangolinfo Scrape API — Amazon real-time data
Pangolinfo Docs — API reference
SP-API Listings Items API Reference
SP-API Feeds API Reference
Amazon Developer Central — App registration

Questions about the SP-API integration or the Pangolinfo data layer? Drop them in the comments.

Tags: #python #api #amazon #ecommerce #automation #tutorial #webdev

Best Amazon Data Scraping Solution in 2026: A Developer's Honest Guide

Mox Loop — Fri, 03 Apr 2026 08:36:37 +0000

I've seen a lot of "how to scrape Amazon" tutorials. Most of them are outdated, incomplete, or both.

This is different. This is an honest breakdown of what actually works for Amazon data collection in 2026, including success rate data, real cost comparisons, and working code examples you can run today.

TL;DR

Approach	Success Rate	Monthly Cost (10K/day)	Maintenance
DIY scraper + free proxies	<15%	Low (but hidden costs)	20-40hrs/month
DIY + residential proxies	40-55%	$3,000-8,000	20-40hrs/month
General scraping API	70-85%	$500-1,500	2-5hrs/month
Dedicated e-commerce API	95%+	$300-800	< 2hrs/month

Skip to the code examples if you already know why DIY Amazon scrapers fail in 2026.

Why Amazon Scrapers Keep Failing in 2026

Amazon's bot detection has crossed a threshold. It's no longer running rule-based checks you can enumerate and bypass — it's running ML behavioral analysis trained on millions of bot interaction patterns.

The five-layer defense stack:

IP reputation scoring — Your proxy's IP might have been flagged before you even bought it
ML behavioral analysis — Session-level sequence analysis, not individual request flags
Browser fingerprinting — Canvas, WebGL, fonts, screen metrics — headless browsers have detectable signatures
Account risk propagation — Programmatic behavior contaminates associated infrastructure
Honeypot content delivery — The silent killer: serving wrong data to suspicious sessions without triggering error responses

The honeypot problem is particularly insidious. Your scraper runs fine, returns 200 OK, logs everything as successful — and you're collecting garbage data you won't notice until something downstream produces anomalous results.

Comparing Approaches

Approach 1: DIY Scrapy/Requests + Proxy Pool

# What everyone starts with
import requests
from scrapy import Selector

def scrape_amazon_diy(asin, proxy):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)...",
        "Accept-Language": "en-US,en;q=0.9",
    }

    try:
        resp = requests.get(
            f"https://www.amazon.com/dp/{asin}",
            headers=headers,
            proxies={"https": proxy},
            timeout=15
        )

        if resp.status_code == 200:
            sel = Selector(text=resp.text)
            title = sel.css("#productTitle::text").get("").strip()
            price = sel.css(".a-price-whole::text").get("").strip()
            return {"title": title, "price": price}
        elif resp.status_code == 503:
            # CAPTCHA or bot challenge page
            return None
    except Exception:
        return None

Reality check on this approach:

Public proxies: <15% success rate on Amazon in 2026
Quality residential proxies: 40-55% — costs $10-20/GB, Amazon pages average 1-2MB each
Parse breakage: Amazon updates page structure regularly; your CSS selectors will break without warning
Honeypot blindness: you won't know you're collecting wrong data

Verdict: Acceptable for experimentation. Not for production.

Approach 2: Dedicated E-Commerce Scraping API

import requests
from typing import Optional

class AmazonScraper:
    """
    Production-grade Amazon scraper using Pangolinfo Scrape API
    Docs: https://docs.pangolinfo.com/en-api-reference/universalApi/universalApi
    """

    BASE_URL = "https://api.pangolinfo.com/v1/scrape"

    def __init__(self, api_key: str):
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })

    def get_product(
        self,
        asin: str,
        marketplace: str = "US",
        zip_code: Optional[str] = None
    ) -> Optional[dict]:
        """
        Fetch structured Amazon product data.

        Returns clean JSON with:
        - title, brand, description
        - price (current, list_price, savings)
        - rating, review_count
        - buybox (seller_name, is_prime, is_amazon_fulfilled)
        - bsr (Best Seller Rank by category)
        - sponsored_products (SP ad positions, 98% capture rate)
        - customer_says (thematic summary)
        - images, variations, technical_specs
        """
        payload = {
            "url": f"https://www.amazon.com/dp/{asin}",
            "platform": "amazon",
            "data_type": "product_detail",
            "marketplace": marketplace,
            "render": True,           # JS rendering enabled
            "extract_ads": True,      # Capture SP ad positions
            "extract_customer_says": True
        }

        if zip_code:
            payload["zip_code"] = zip_code  # Geo-targeted pricing

        resp = self.session.post(self.BASE_URL, json=payload, timeout=30)
        resp.raise_for_status()
        return resp.json().get("result")

    def search(
        self,
        keyword: str,
        marketplace: str = "US",
        page: int = 1
    ) -> Optional[dict]:
        """
        Keyword search with sponsored position tagging.
        Returns organic_results + sponsored_results (with position metadata)
        """
        payload = {
            "url": f"https://www.amazon.com/s?k={keyword}&page={page}",
            "platform": "amazon",
            "data_type": "search",
            "marketplace": marketplace,
            "render": True,
            "extract_ads": True
        }

        resp = self.session.post(self.BASE_URL, json=payload, timeout=30)
        resp.raise_for_status()
        return resp.json().get("result")

    def get_best_sellers(
        self,
        category_url: str,
        marketplace: str = "US"
    ) -> list:
        """Pull complete Best Sellers / New Releases list for a category"""
        payload = {
            "url": category_url,
            "platform": "amazon",
            "data_type": "best_sellers",
            "marketplace": marketplace,
            "render": True,
            "extract_ads": True
        }

        resp = self.session.post(self.BASE_URL, json=payload, timeout=60)
        resp.raise_for_status()
        return resp.json().get("result", {}).get("products", [])


# Usage examples
scraper = AmazonScraper(api_key="your_pangolinfo_key")

# 1. Product detail with NYC pricing
product = scraper.get_product("B08N5WRWNW", zip_code="10001")
print(f"""
Title: {product['title']}
Price (NYC): ${product['price']['current']}
Rating: {product['rating']} ({product['review_count']} reviews)
Buybox: {product['buybox']['seller_name']}
SP Ads captured: {len(product['sponsored_products'])}
""")

# 2. Keyword competitive analysis
results = scraper.search("stainless steel water bottle")
sponsored = results.get("sponsored_results", [])
organic = results.get("organic_results", [])
print(f"Sponsored positions: {len(sponsored)}, Organic: {len(organic)}")

# 3. Category intelligence
best_sellers = scraper.get_best_sellers(
    "https://www.amazon.com/Best-Sellers-Kitchen-Dining/zgbs/kitchen/"
)
print(f"Retrieved {len(best_sellers)} Best Sellers")

# Find rising stars: high rating, low review count
rising = [p for p in best_sellers 
          if p.get('rating', 0) >= 4.5 and p.get('review_count', 999) < 500]
print(f"Rising star candidates: {len(rising)}")

Real-World Performance Comparison {#implementation}

I ran a 7-day test comparing approaches across three Amazon page types. Results:

Best Sellers category pages:

DIY + residential proxy: 48% success, $23 per 1,000 successful pages
Pangolinfo API: 96% success, ~$8 per 1,000 successful pages (estimate)

Product detail pages with advertising data:

DIY approach: SP ad positions captured correctly in ~12% of attempts
Pangolinfo API: SP ad positions captured in ~98% of attempts

Parse reliability over 30 days:

DIY approach: Required CSS selector fixes on 3 separate occasions after Amazon updates
Pangolinfo API: Zero maintenance required on calling code

AI Agent Integration

One underrated use case: using this as a live data layer for AI agents.

LLMs have knowledge cutoffs. "What's currently selling in the kitchen category?" answered from training data is increasingly fictional.

Pangolinfo packages its API as a standardized Amazon Scraper Skill compatible with AI agent frameworks. The agent calls the skill to retrieve live Amazon data before generating analysis — grounding responses in current market reality rather than stale training data.

# Conceptual: AI Agent with live Amazon data
agent_query = """
Analyze the top 20 products in Amazon US Kitchen Best Sellers.
Identify products with rating >= 4.5 and review count < 1000.
For each, estimate monthly sales and flag competitive gaps.
"""

# Agent automatically calls Pangolinfo Scraper Skill for live data
# Then applies LLM reasoning to the fresh dataset
response = agent.run(agent_query)  # Real data, not hallucinated

Quick Start

Sign up at tool.pangolinfo.com — free trial available
Check the API reference for full field documentation
Start with Best Sellers category pulls — lowest latency, clearest ROI demonstration

Summary

If you're building anything that depends on reliable Amazon data in 2026, the engineering decision is straightforward: the maintenance cost and success rate gap between DIY approaches and purpose-built APIs has grown large enough that specialized infrastructure is the rational default choice at any meaningful scale.

Save the DIY scraper for learning. Use the right tool for production.

Tags: #python #api #amazon #ecommerce #webscaping #dataengineering #tutorial

Questions about the implementation? Drop them in the comments — happy to dig into specifics.

Stop Drowning in Amazon Data: Build a Decision Framework with Scrape API

Mox Loop — Wed, 01 Apr 2026 05:14:19 +0000

TL;DR: Amazon seller data overload is a real problem, but it's fixable with the right infrastructure. This tutorial shows you how to unify BSR, keyword SERP, and ad position data using Pangolinfo's Scrape API — so your analytics dashboard answers questions instead of creating more of them.

The Problem

Amazon operators typically juggle three data sources that don't talk to each other:

BSR rankings — updates hourly, extracted from product pages
ABA keyword data — weekly CSV downloads from Seller Central (manual)
Advertising reports — 24-48 hour delay, different metrics entirely

When BSR drops 20 positions, diagnosing the root cause means manually correlating data from all three sources. Average time: 2-3 hours per incident. And the diagnosis still might be wrong.

The fix is a unified data pipeline that aligns these three streams on a shared time axis. Here's how to build one.

Prerequisites

pip install requests psycopg2-binary python-dotenv

PANGOLINFO_API_KEY=your_key_here
DATABASE_URL=postgresql://user:pass@localhost:5432/amazon_analytics

Step 1: Database Schema

-- Unified time-series schema for Amazon operational data
CREATE TABLE asin_snapshots (
    id              BIGSERIAL PRIMARY KEY,
    asin            VARCHAR(20) NOT NULL,
    marketplace     VARCHAR(5)  NOT NULL DEFAULT 'US',
    bsr_main        INTEGER,
    bsr_category    VARCHAR(100),
    price           DECIMAL(10,2),
    review_count    INTEGER,
    rating          DECIMAL(3,2),
    collected_at    TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE keyword_snapshots (
    id                  BIGSERIAL PRIMARY KEY,
    keyword             TEXT NOT NULL,
    target_asin         VARCHAR(20),
    organic_rank        SMALLINT,       -- NULL = not on first page
    sp_positions_count  SMALLINT NOT NULL DEFAULT 0,
    target_sp_rank      SMALLINT,       -- NULL = no SP ad running
    collected_at        TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX ON asin_snapshots (asin, collected_at DESC);
CREATE INDEX ON keyword_snapshots (keyword, collected_at DESC);

Step 2: Data Collector

"""
collector.py - Pangolinfo Scrape API wrapper
Handles ASIN snapshots and keyword SERP collection
"""

import os
import requests
from datetime import datetime
from dataclasses import dataclass
from typing import Optional

API_KEY = os.getenv("PANGOLINFO_API_KEY")
BASE_URL = "https://api.pangolinfo.com/v1/amazon"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}


@dataclass
class AsinSnapshot:
    asin: str
    marketplace: str
    bsr_main: Optional[int]
    bsr_category: Optional[str]
    price: Optional[float]
    review_count: Optional[int]
    rating: Optional[float]
    collected_at: str


@dataclass  
class KeywordSnapshot:
    keyword: str
    target_asin: Optional[str]
    organic_rank: Optional[int]
    sp_positions_count: int
    target_sp_rank: Optional[int]
    collected_at: str


def collect_asin(asin: str, marketplace: str = "US") -> AsinSnapshot:
    """Collect product snapshot: BSR, price, reviews, ratings."""
    resp = requests.post(
        f"{BASE_URL}/product",
        headers=HEADERS,
        json={
            "asin": asin,
            "marketplace": marketplace,
            "fields": ["bsr", "price", "reviews", "ratings"],
            "output_format": "json"
        }
    )
    resp.raise_for_status()
    d = resp.json()

    return AsinSnapshot(
        asin=asin,
        marketplace=marketplace,
        bsr_main=d.get("bsr", {}).get("main_category_rank"),
        bsr_category=d.get("bsr", {}).get("main_category_name"),
        price=d.get("price", {}).get("current"),
        review_count=d.get("reviews", {}).get("count"),
        rating=d.get("ratings", {}).get("average"),
        collected_at=datetime.utcnow().isoformat()
    )


def collect_keyword(
    keyword: str,
    target_asin: Optional[str] = None,
    marketplace: str = "US"
) -> KeywordSnapshot:
    """
    Collect keyword SERP: organic ranks + sponsored positions.

    Note: Sponsored ad position extraction is one of the more
    technically challenging aspects of Amazon scraping due to
    dynamic rendering. Pangolinfo handles this at the API level,
    so you get complete SP position data reliably.
    """
    resp = requests.post(
        f"{BASE_URL}/keyword-serp",
        headers=HEADERS,
        json={
            "keyword": keyword,
            "marketplace": marketplace,
            "include_ad_positions": True,
            "output_format": "json"
        }
    )
    resp.raise_for_status()
    d = resp.json()

    # Find target ASIN's organic position
    organic_rank = None
    if target_asin:
        for item in d.get("organic_results", []):
            if item.get("asin") == target_asin:
                organic_rank = item.get("position")
                break

    # Find target ASIN's sponsored position
    sp_positions = d.get("sponsored_positions", [])
    target_sp_rank = None
    if target_asin:
        for sp in sp_positions:
            if sp.get("asin") == target_asin:
                target_sp_rank = sp.get("position")
                break

    return KeywordSnapshot(
        keyword=keyword,
        target_asin=target_asin,
        organic_rank=organic_rank,
        sp_positions_count=len(sp_positions),
        target_sp_rank=target_sp_rank,
        collected_at=datetime.utcnow().isoformat()
    )

Step 3: Database Writer

"""
writer.py - Persist collected data to PostgreSQL
"""

import os
import psycopg2
from dataclasses import asdict
from collector import AsinSnapshot, KeywordSnapshot

conn = psycopg2.connect(os.getenv("DATABASE_URL"))


def write_asin_snapshot(snapshot: AsinSnapshot):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO asin_snapshots
                (asin, marketplace, bsr_main, bsr_category, price, review_count, rating, collected_at)
            VALUES
                (%(asin)s, %(marketplace)s, %(bsr_main)s, %(bsr_category)s,
                 %(price)s, %(review_count)s, %(rating)s, %(collected_at)s)
        """, asdict(snapshot))
    conn.commit()


def write_keyword_snapshot(snapshot: KeywordSnapshot):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO keyword_snapshots
                (keyword, target_asin, organic_rank, sp_positions_count, target_sp_rank, collected_at)
            VALUES
                (%(keyword)s, %(target_asin)s, %(organic_rank)s,
                 %(sp_positions_count)s, %(target_sp_rank)s, %(collected_at)s)
        """, asdict(snapshot))
    conn.commit()

Step 4: Scheduled Monitoring Job

"""
monitor.py - Daily monitoring routine
Run via cron: 0 */6 * * * python monitor.py
"""

import json
from collector import collect_asin, collect_keyword
from writer import write_asin_snapshot, write_keyword_snapshot

# Configure your monitoring scope here
CONFIG = {
    "target_asin": "B09XXXXXXX",
    "competitor_asins": ["B08YYYYYYY", "B07ZZZZZZZ"],
    "core_keywords": [
        "camping chair lightweight",
        "portable outdoor chair",
        "folding camp chair"
    ],
    "marketplace": "US"
}


def run_monitoring():
    all_asins = [CONFIG["target_asin"]] + CONFIG["competitor_asins"]

    print("Collecting ASIN snapshots...")
    for asin in all_asins:
        snapshot = collect_asin(asin, CONFIG["marketplace"])
        write_asin_snapshot(snapshot)
        print(f"  [{asin}] BSR: {snapshot.bsr_main} | Reviews: {snapshot.review_count}")

    print("\nCollecting keyword SERP data...")
    for keyword in CONFIG["core_keywords"]:
        snapshot = collect_keyword(keyword, CONFIG["target_asin"], CONFIG["marketplace"])
        write_keyword_snapshot(snapshot)
        print(
            f"  [{keyword}] "
            f"Organic: {snapshot.organic_rank or 'not ranked'} | "
            f"SP slots: {snapshot.sp_positions_count} | "
            f"Your SP: {snapshot.target_sp_rank or 'not running'}"
        )


if __name__ == "__main__":
    run_monitoring()

Step 5: Anomaly Detection Query

Once you have a few days of data, this query detects BSR drops that may require immediate attention:

-- Detect BSR drops > 10% vs previous collection
WITH ranked AS (
    SELECT
        asin,
        bsr_main,
        LAG(bsr_main) OVER (PARTITION BY asin ORDER BY collected_at) AS prev_bsr,
        collected_at
    FROM asin_snapshots
    WHERE collected_at >= NOW() - INTERVAL '24 hours'
),
anomalies AS (
    SELECT
        asin,
        prev_bsr,
        bsr_main AS current_bsr,
        ROUND(((bsr_main - prev_bsr)::float / prev_bsr * 100)::numeric, 1) AS change_pct,
        collected_at
    FROM ranked
    WHERE prev_bsr IS NOT NULL
      AND bsr_main > prev_bsr * 1.10  -- BSR increased (rank worsened) by 10%+
)
SELECT * FROM anomalies ORDER BY change_pct DESC;

Cross-reference any anomaly result against keyword_snapshots on the same timestamp window to see if the BSR drop correlates with a sponsored position displacement.

What You Get

After running for a few days, you'll have a queryable time-series of:

ASIN BSR trajectory (visualize in Grafana or Metabase)
Keyword organic rank changes per collection window
SP position count fluctuations on your core keywords
Your ASIN's presence/absence in sponsored positions

When BSR drops, run the anomaly query. Cross-reference with keyword data. The root cause that used to take three hours manually is now visible in one joined query.

No-Code Alternative

Building and maintaining this pipeline requires ongoing engineering time. If you want the same structured, unified view without writing code, AMZ Data Tracker provides a configuration interface for the same monitoring logic — ASIN tracking, keyword ranking, competitor monitoring — with automatic collection and a built-in dashboard.

API documentation for the Scrape API is at docs.pangolinfo.com. Free trial credits available.

Have questions about the implementation or want to share how you've built your own Amazon analytics stack? Drop a comment — happy to discuss architecture tradeoffs.

Amazon Category Opportunity Analysis: Building a Smarter Launch and Ad Decision Workflow

Mox Loop — Tue, 31 Mar 2026 03:24:46 +0000

TL;DR

Most Amazon sellers waste money because they treat category selection and ad optimization as separate problems.

A better workflow:

Evaluate demand, competition, margin, traffic structure, and execution risk together
Calculate break-even ACoS and workable CPC before launch
Use AI Agents to reduce the API learning curve
Scale only when CTR, CVR, ACoS, and organic lift align

If you're non-technical, this matters even more now because Agent workflows make API-based category research much more accessible.

The Real Problem

The classic workflow looks like this:

def old_workflow():
    choose_category_by_instinct()
    launch_listing()
    run_ads()
    hope_data_proves_you_right()

The problem is obvious: ads are being used to validate a weak market decision.

Instead, category entry should be treated like a systems problem.

You need to evaluate:

Demand strength
Supply density
Profit structure
Ad cost elasticity
Traffic structure
Execution difficulty

A Simple Launch Math Layer

Before entering a category, calculate these three numbers:

def launch_math(price, contribution_margin, expected_cvr, daily_clicks, test_days):
    break_even_acos = contribution_margin / price
    target_cpc = price * expected_cvr * break_even_acos
    test_budget = daily_clicks * target_cpc * test_days
    return break_even_acos, target_cpc, test_budget

This is not about perfect forecasting. It is about making weak categories fail on paper before they fail in your account.

Why AI Agents Change the Game

This is the part many teams still underestimate.

A few years ago, API-based public page collection required technical implementation. Today, many non-technical operators can take a Scrape API endpoint, the relevant API docs, and an API key, then hand them to an agent such as openclaw to fetch and summarize Amazon public data.

That means sellers can prototype workflows like:

{
  "task": "Collect top 100 search results for a category",
  "fields": ["asin", "price", "rating", "review_count", "ad_placement"],
  "goal": "Estimate category competition and traffic structure"
}

The edge is no longer just coding ability. The edge is the ability to define useful data tasks.

Traffic Structure Should Drive Campaign Structure

Do not use the same campaign template for every stage.

Validation

SP auto
SP broad
SP phrase

Growth

SP exact
ASIN targeting

Expansion

SP exact
SB
SD retargeting

The point is simple: traffic structure first, campaign structure second.

When to Scale and When to Cut

Scale only if:

CTR stays above average
CVR holds as volume rises
ACoS remains acceptable
TACoS does not break
Organic rank improves

Cut if:

Clicks rise but conversion falls
ACoS and TACoS worsen together
Placement inflation kills efficiency
Inventory cannot support growth

Low ACoS is not automatically good. High ACoS is not automatically bad. Repeatability matters more than vanity efficiency.

A Practical Stack

If I were building this today, I would combine:

Scrape API for public Amazon page collection
AMZ Data Tracker for trend and competitor monitoring
An Agent workflow for quick API calling and summarization
A simple scorecard for category opportunity and launch risk

That gives you a workflow that is much closer to engineering discipline than guesswork.

Final Thought

Amazon category opportunity analysis is really about one thing:

Can you enter this market profitably and scale it without losing control?

If you answer that before launch, your ad account gets smarter.

If you wait for ads to answer it for you, your budget becomes the tuition fee.

Amazon Ad Position Monitoring with Open Claw + Pangolinfo SERP API

Mox Loop — Tue, 24 Mar 2026 07:04:59 +0000

TL;DR

Stop reading your ACoS dashboard to understand competitor behavior. Build a real-time SP ad position monitor instead.

This post covers:

Async batch SERP capture via Pangolinfo API (98% SP coverage)
Tiered keyword management (A/B/C) for signal vs. noise control
Change detection: Top1 change, new Top3 entrant, Top3 exit, price drop
LLM-enriched alerts via Open Claw + Claude
Deduplication and Slack delivery

Prerequisites: Open Claw deployed, Pangolinfo API key configured (see earlier posts in this series). We're jumping straight into the ad monitoring implementation.

# Quick test: is your SERP API working?
import requests

headers = {"Authorization": "Bearer YOUR_KEY"}
payload = {
    "source": "amazon_search",
    "query": "wireless earbuds",
    "marketplace": "US",
    "include_sponsored": True,
    "include_organic": False,
    "output_format": "json"
}

resp = requests.post("https://api.pangolinfo.com/v1/serp",
                     headers=headers, json=payload)
print(resp.json()["sponsored_results"][:3])

The Problem with Existing Monitoring Tools

Your Amazon ad console shows internal metrics. Helium 10 / Jungle Scout show competitor data, but with 24-48h delays and no outbound API. When a new well-funded competitor enters your Top of Search overnight, you find out two days later when your ACoS report arrives. By then they've run 48 hours of aggressive ads at a discount price, accumulating early reviews.

What we want: automated detection of competitor ad position changes within 2 hours of them occurring, categorized by severity, routed to Slack with business context.

Step 1: Async SERP Capture

50 keywords × sequential requests = ~8 minutes per cycle. With asyncio: ~45 seconds.

import asyncio
import aiohttp
from datetime import datetime, timezone
from typing import List, Dict, Optional

PANGOLINFO_API_KEY = "your_key"

async def fetch_serp(
    keyword: str,
    marketplace: str,
    session: aiohttp.ClientSession,
    semaphore: asyncio.Semaphore
) -> Dict:
    """Capture Amazon SP ad positions for one keyword"""
    async with semaphore:
        headers = {"Authorization": f"Bearer {PANGOLINFO_API_KEY}"}
        payload = {
            "source": "amazon_search",
            "query": keyword,
            "marketplace": marketplace,
            "page": 1,
            "include_sponsored": True,
            "include_organic": False,
            "output_format": "json"
        }
        captured_at = datetime.now(timezone.utc).isoformat()

        try:
            async with session.post(
                "https://api.pangolinfo.com/v1/serp",
                headers=headers, json=payload,
                timeout=aiohttp.ClientTimeout(total=25)
            ) as resp:
                resp.raise_for_status()
                data = await resp.json()

            top_ads = sorted(
                [s for s in data.get("sponsored_results", [])
                 if "top" in s.get("ad_placement", "").lower()],
                key=lambda x: x.get("ad_rank", 999)
            )[:5]

            return {
                "keyword": keyword, "marketplace": marketplace,
                "captured_at": captured_at, "success": True,
                "top_of_search": [
                    {"rank": a.get("ad_rank"), "asin": a.get("asin"),
                     "brand": a.get("brand", ""), "price": a.get("price")}
                    for a in top_ads
                ]
            }
        except Exception as e:
            return {"keyword": keyword, "marketplace": marketplace,
                    "captured_at": captured_at,
                    "success": False, "error": str(e), "top_of_search": []}


async def batch_monitor(
    keywords: List[str],
    marketplace: str = "US",
    max_concurrent: int = 8
) -> List[Dict]:
    sem = asyncio.Semaphore(max_concurrent)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_serp(kw, marketplace, session, sem) for kw in keywords]
        return await asyncio.gather(*tasks)

Step 2: Keyword Tier Configuration

Don't monitor all keywords at the same frequency and alert threshold. It generates noise that gets ignored.

KEYWORD_CONFIG = {
    "tiers": {
        "A": {  # Core product keywords — monitor every 2h
            "keywords": ["wireless earbuds", "bluetooth speaker", "usb c hub"],
            "frequency_hours": 2,
            "alert_escalation": True   # Upgrade HIGH → CRITICAL
        },
        "B": {  # Competitive category keywords — monitor every 6h
            "keywords": ["earbuds under 30 dollars", "tws earbuds 2026"],
            "frequency_hours": 6,
            "alert_escalation": False
        },
        "C": {  # Exploratory long-tail — monitor daily, log only
            "keywords": [],
            "frequency_hours": 24,
            "alert_escalation": False,
            "suppress_alerts": True    # Record but don't notify
        }
    },
    "price_drop_threshold_pct": 12,
    "dedup_window_hours": 6
}

Step 3: Change Detection Engine

from dataclasses import dataclass, field
from enum import Enum

class AlertLevel(Enum):
    CRITICAL = "CRITICAL"
    HIGH = "HIGH"
    MEDIUM = "MEDIUM"
    INFO = "INFO"

@dataclass
class AdAlert:
    keyword: str
    level: AlertLevel
    event: str          # top1_change / new_top3 / top3_exit / price_drop
    message: str
    asin: str
    timestamp: str
    llm_context: str = ""


def detect_changes(
    current: Dict,
    baseline: Optional[Dict],
    tier_config: Dict
) -> List[AdAlert]:
    """
    Compare SERP snapshots. Returns [] on first run (baseline only).
    """
    if baseline is None:
        return []

    kw = current["keyword"]
    ts = current["captured_at"]
    tier = tier_config.get("tier", "B")
    suppress = tier_config.get("suppress_alerts", False)
    escalate = tier_config.get("alert_escalation", False)
    threshold = KEYWORD_CONFIG["price_drop_threshold_pct"] / 100

    curr_top = [a["asin"] for a in current.get("top_of_search", [])[:3]]
    base_top = [a["asin"] for a in baseline.get("top_of_search", [])[:3]]
    curr_prices = {a["asin"]: a.get("price")
                   for a in current.get("top_of_search", [])[:3]}
    base_prices = {a["asin"]: a.get("price")
                   for a in baseline.get("top_of_search", [])[:3]}

    def resolve(base: AlertLevel) -> AlertLevel:
        if suppress:
            return AlertLevel.INFO
        if escalate and base == AlertLevel.HIGH:
            return AlertLevel.CRITICAL
        return base

    alerts = []

    # Top #1 changed
    if curr_top and base_top and curr_top[0] != base_top[0]:
        is_new = curr_top[0] not in base_top
        alerts.append(AdAlert(
            keyword=kw,
            level=resolve(AlertLevel.CRITICAL if is_new else AlertLevel.HIGH),
            event="top1_change",
            message=(f"Top of Search #1: {base_top[0]} → {curr_top[0]}"
                     + (" [NEW ENTRANT]" if is_new else "")),
            asin=curr_top[0], timestamp=ts
        ))

    # New Top 3 entrant (excluding #1 already handled)
    for asin in set(curr_top) - set(base_top) - {curr_top[0] if curr_top else ""}:
        rank = next((a["rank"] for a in current["top_of_search"]
                     if a["asin"] == asin), -1)
        alerts.append(AdAlert(
            keyword=kw, level=resolve(AlertLevel.HIGH),
            event="new_top3",
            message=f"New entrant at Top 3 position #{rank}: {asin}",
            asin=asin, timestamp=ts
        ))

    # Top 3 exit
    for asin in set(base_top) - set(curr_top):
        alerts.append(AdAlert(
            keyword=kw, level=resolve(AlertLevel.MEDIUM),
            event="top3_exit",
            message=f"{asin} exited Top 3 — possible budget reduction",
            asin=asin, timestamp=ts
        ))

    # Price drop in Top 3
    for asin in set(curr_top) & set(base_top):
        old, new = base_prices.get(asin), curr_prices.get(asin)
        if old and new and old > 0 and (old - new) / old >= threshold:
            drop = (old - new) / old * 100
            alerts.append(AdAlert(
                keyword=kw, level=resolve(AlertLevel.HIGH),
                event="price_drop",
                message=f"{asin} price drop {drop:.1f}%: ${old:.2f} → ${new:.2f}",
                asin=asin, timestamp=ts
            ))

    return alerts

Step 4: LLM Context Enrichment (Open Claw Layer)

from anthropic import Anthropic

_llm = Anthropic()

def add_llm_context(alert: AdAlert, snapshot: Dict) -> AdAlert:
    """Enrich CRITICAL/HIGH alerts with Claude business analysis"""
    if alert.level not in (AlertLevel.CRITICAL, AlertLevel.HIGH):
        return alert

    top3 = "\n".join([
        f"  #{a['rank']}: {a['asin']} | {a['brand']} | ${a['price'] or 'N/A'}"
        for a in snapshot.get("top_of_search", [])[:3]
    ])

    resp = _llm.messages.create(
        model="claude-3-7-sonnet-20250219",
        max_tokens=180,
        messages=[{"role": "user", "content":
            f"Amazon ad monitoring alert — keyword: '{alert.keyword}'\n"
            f"Event: {alert.message}\n"
            f"Current Top 3:\n{top3}\n\n"
            "In 70 words or fewer: likely cause + what the PPC team should check. "
            "Direct, specific, no preamble."
        }]
    )
    alert.llm_context = resp.content[0].text
    return alert

Step 5: Slack Delivery with Deduplication

import sqlite3
import aiohttp
from datetime import timedelta

DB_PATH = "/data/ad_monitor.db"
SLACK_WEBHOOK = "https://hooks.slack.com/services/YOUR/WEBHOOK"


def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.executescript("""
            CREATE TABLE IF NOT EXISTS snapshots (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                keyword TEXT, marketplace TEXT,
                captured_at TEXT, data_json TEXT
            );
            CREATE TABLE IF NOT EXISTS sent_alerts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                keyword TEXT, asin TEXT, event TEXT,
                triggered_at TEXT
            );
        """)


def is_duplicate(alert: AdAlert, window_hours: int = 6) -> bool:
    cutoff = (datetime.utcnow() - timedelta(hours=window_hours)).isoformat()
    with sqlite3.connect(DB_PATH) as conn:
        return bool(conn.execute(
            "SELECT 1 FROM sent_alerts WHERE keyword=? AND asin=? AND event=?"
            " AND triggered_at>? LIMIT 1",
            (alert.keyword, alert.asin, alert.event, cutoff)
        ).fetchone())


def record_alert(alert: AdAlert):
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "INSERT INTO sent_alerts (keyword, asin, event, triggered_at)"
            " VALUES (?,?,?,?)",
            (alert.keyword, alert.asin, alert.event, alert.timestamp)
        )


async def send_to_slack(alert: AdAlert):
    emoji = {"CRITICAL": "🚨", "HIGH": "⚠️", "MEDIUM": "📊"}.get(
        alert.level.value, "ℹ️"
    )
    text = (
        f"{emoji} *[Ad Monitor]* `{alert.level.value}` — "
        f"`{alert.keyword}`\n"
        f"{alert.message}\n"
        f"ASIN: `{alert.asin}` | {alert.timestamp[:16]} UTC"
        + (f"\n> _{alert.llm_context}_" if alert.llm_context else "")
    )
    async with aiohttp.ClientSession() as session:
        await session.post(SLACK_WEBHOOK, json={"text": text},
                           timeout=aiohttp.ClientTimeout(total=10))

Step 6: Wiring It All Together

import json

async def run_monitoring_cycle(tier: str, marketplace: str = "US"):
    """One complete monitoring run for a keyword tier"""
    tier_cfg = KEYWORD_CONFIG["tiers"][tier]
    keywords = tier_cfg["keywords"]
    if not keywords:
        return

    snapshots = await batch_monitor(keywords, marketplace)

    for snap in snapshots:
        if not snap.get("success"):
            continue

        # Load baseline (second-to-last snapshot)
        with sqlite3.connect(DB_PATH) as conn:
            row = conn.execute(
                "SELECT data_json FROM snapshots WHERE keyword=? AND marketplace=?"
                " ORDER BY captured_at DESC LIMIT 1 OFFSET 1",
                (snap["keyword"], marketplace)
            ).fetchone()
            baseline = json.loads(row[0]) if row else None

            # Save current snapshot
            conn.execute(
                "INSERT INTO snapshots (keyword, marketplace, captured_at, data_json)"
                " VALUES (?,?,?,?)",
                (snap["keyword"], marketplace, snap["captured_at"], json.dumps(snap))
            )

        # Detect + dispatch
        alerts = detect_changes(snap, baseline, tier_cfg)
        for alert in alerts:
            if not is_duplicate(alert, KEYWORD_CONFIG["dedup_window_hours"]):
                alert = add_llm_context(alert, snap)
                await send_to_slack(alert)
                record_alert(alert)


# Entry point — call from APScheduler or cron
async def main():
    init_db()
    await run_monitoring_cycle("A")   # Tier A every 2h
    # await run_monitoring_cycle("B") # Tier B every 6h

if __name__ == "__main__":
    asyncio.run(main())

Production Notes

Database scaling: SQLite works fine up to ~100 keywords at 2h frequency. Beyond that, switch to PostgreSQL — SQLite's write lock becomes a bottleneck under concurrent writes.

Rate limiting: If you hit HTTP 429, add exponential backoff (start 2s, max 30s) around the session.post call in fetch_serp.

Monitoring the monitor: Set up a heartbeat check — if no rows appear in snapshots for the expected time window, alert your on-call channel. Silent failures in monitoring systems are the worst kind.

Data retention: Keep 90 days of snapshots. At 30 days you see monthly patterns; at 90 days you start recognizing seasonal behaviors and pre-promotion signals.