DEV Community: Pangolinfo

Amazon Keyword Organic Ranking Tracking: Why "What Position" Is an Outdated Metric

Pangolinfo — Fri, 24 Jul 2026 09:30:53 +0000

Leo · Head of Engineering, Pangolinfo

Nearly every rank tracker sells the same promise: tell you "what position your product holds for keyword X." That metric is breaking down — not because the tools are inaccurate, but because "what position" answers the wrong question. Amazon's results swing by geo, device, Prime status, and personalization, so a single normalized "true rank" is a fiction. This post repeats the mainstream consensus, names five blind spots it leaves, and finishes with the alternative built for 2026.

The mainstream consensus (and the premise it gets wrong)

The industry's eight-point playbook: separate organic from sponsored; daily is enough, ignore hourly noise (A10 "shuffle" self-heals in 72–96h); rank is an estimate, watch trends not screenshots; verify manually with incognito + VPN; pair with Brand Analytics / Search Query Performance; use proxies to avoid blocks; view rank alongside conversion, inventory, price; pick 10–30 core keywords and commit to a cadence. Each point is correct — but they all anchor to one premise: that the goal is the most accurate possible "what position." Accept that, and the five blind spots below become inevitable.

The five blind spots

1. A normalized rank instead of segment SoV + traffic proxy. A "#7" hides the completely different Top 10 a New York mobile Prime member sees versus a Los Angeles desktop non-Prime user. What matters is segment share-of-voice (how often your ASIN appears in the head of a keyword cluster) plus a traffic proxy from the position CTR curve — position 1 can capture 10x the clicks of position 7.

2. Filtering noise throws away the only real-time dividend. "Ignore hourly swings" averages away competitor ad gaps, inventory / coupon windows, and temporal arbitrage. A cat-litter-box seller at $200/day and 35%–40% ACoS switched to real-time monitoring, found competitors paused ads 2–4 PM and weekends were weaker, moved 30% of budget into the afternoon gap — CTR +25%, CVR +18%, ACoS to 24%. Filter the noise and you throw away the most valuable signal.

3. Tracking only "known keywords," never a ranking gap map. Dashboards assume you already know which keywords to track. You never discover "I'm completely unranked for this term, yet it converts brutally well" because nobody continuously diffs your indexed terms against the category's query graph.

4. Rank decoupled from its levers. Mainstream advice stops at side-by-side observation. What's missing is closed-loop attribution: automatically pull price, coupon, inventory, competitor ad intensity, and your PPC as covariates and tell you "72% of this drop is competitor X's bid, 18% your stockout."

5. Fighting anti-bot with proxies, still human-stares-at-dashboard, not agent-native. The category teaches you how not to get blocked and ships a dashboard for a human. In 2026 rank data should be a primitive an agent calls directly — "what's my US mobile rank, is there a competitor window, adjust my bid" — with no human watching arrows.

The advanced answer: ranking as data infrastructure

Primary metric = segment SoV + traffic proxy, not a single normalized rank.
Keep the volatility distribution; mine it for windows.
Continuously draw the ranking gap map; proactively surface "should-rank-but-doesn't" terms.
Treat price / coupon / inventory / competitor ad intensity / PPC as rank covariates, auto-attributed.
Consume managed SERP data via Amazon Scraper API instead of self-hosted proxy scrapers.
Expose rank through Amazon Data MCP so an agent calls and acts on it.

Implementation: two ways

Option A, build it yourself. Pull from a managed SERP API, store, schedule, alert. Minimal Python skeleton:

import requests
api_key = "YOUR_KEY"
r = requests.post("https://api.pangolinfo.com/serp/amazon",
    json={"keyword":"wireless earbuds","marketplace":"US","page":1,"include_sponsored":True},
    headers={"Authorization": "Bearer " + api_key})
data = r.json()
organic = data["organic_results"]
my_rank = next((i+1 for i,p in enumerate(organic) if p["asin"]=="YOUR_ASIN"), None)
if my_rank is None or my_rank > 10:
    alert("wireless earbuds organic rank fell out of Top10, now " + str(my_rank))

Contract in the Universal Scrape API docs.

Option B, zero code. Use AMZ Data Tracker: configure rules in the UI, the system calls the API, stores, charts, and notifies.

Takeaways

Rank tracking isn't "make the number more accurate" — it's "turn rank into intelligence that triggers action."
Six things mainstream tools don't do: segment SoV, real-time windows, ranking gap maps, closed-loop attribution, managed data, agent-native.
For visual monitoring use AMZ Data Tracker; for agent-native callable rank data, use Amazon Data MCP.

Full case study: why real-time SERP data is the only way to analyze competitors.

Leo runs engineering at Pangolinfo, building real-time Amazon data infrastructure for developers and AI agents.

What Is Voice of Customer Analysis, and How Do You Build It in 2026

Pangolinfo — Tue, 21 Jul 2026 08:34:22 +0000

Leo · Head of Engineering, Pangolinfo

If you're a developer building anything customer-aware — a brand monitor, a churn early-warning system, an agent that briefs your PM — the trap isn't your model. It's whether you can actually hear what customers are saying, across the platforms where they talk. I run the real-time consumer-insight systems at Pangolinfo (30M+ calls/day, 99% success), and I keep watching teams ship features on data they never collected.

This post is practical: what VOC is, why 2026 changed it, and two runnable patterns — an MCP setup that puts multi-platform customer voice directly into your agent.

What Voice of Customer analysis actually is

Voice of Customer (VOC) analysis is the practice of turning raw customer feedback from social, reviews, and support channels into measurable share of voice, sentiment, and actions a brand can actually take. The key difference from surveys: it listens rather than asks, so there's no question-design bias, it's always-on, and it drills down to a single post.

What counts as VOC is broader than most think: comments and ratings on TikTok, Instagram, YouTube, X, Facebook, Pinterest, Trustpilot; e-commerce reviews; support transcripts; deep forum discussion on Reddit; plus surveys/NPS as a supplement. (Compliance note: we only use public data outside mainland China — not Xiaohongshu, Weibo, or in-region LinkedIn.)

The five-step method

Collect — pull raw expressions from multiple platforms.
Volume / SoV — compute your share of voice in the category.
Sentiment — label positive / negative / neutral, and net sentiment.
Drivers — attribute the reason behind the emotion (late shipping, weak packaging).
Act — route the list to product / ops / support / PR.

Worked example: a coffee-gear brand ran the loop for a month, pulled 12,000 posts, found its SoV at 18% vs a 34% leader, saw a Friday sentiment dip, traced it to mid-week shipping delays, shifted cutoff times — and the dip flattened. The loop paid for itself.

Wiring it into an AI agent with MCP (3-minute setup)

Before the MCP (Model Context Protocol) standard, getting this data into a model meant a copy-paste loop. We replaced that with VOC Insight MCP. One config, and the agent gains 26 VOC tools across 7 default platforms (TikTok, Instagram, YouTube, X, Facebook, Pinterest, Trustpilot) plus optional Threads and Reddit.

// .mcp.json — drop into any MCP-capable client (Claude Code, Cursor, Windsurf, Cline)
// exact install command per docs: https://docs.pangolinfo.com/en-help-center/mcp/voc-insight/overview
{
  "mcpServers": {
    "pangolinfo-voc": {
      "command": "npx",
      "args": ["-y", "@pangolinfo/voc-insight-mcp"],
      "env": { "PANGOLIN_API_KEY": "YOUR_API_KEY" }
    }
  }
}

After that you just say: "This week on TikTok, how do our share of voice and sentiment compare to competitor A?" — the agent discovers the right tools, calls them, and returns a report. No glue code. Enterprise environments can set PANGOLIN_READ_ONLY=true for a read-only agent.

Processing the payload (example)

def summarize_voc(payload):
    items = payload.get("mentions", [])
    total = len(items)
    if not total:
        return {"total": 0, "net_sentiment": 0, "share_of_voice": {}}
    pos = sum(1 for m in items if m.get("sentiment") == "positive")
    neg = sum(1 for m in items if m.get("sentiment") == "negative")
    sov = {}
    for m in items:
        b = m.get("brand", "unknown")
        sov[b] = sov.get(b, 0) + 1
    return {
        "total": total,
        "net_sentiment": round((pos - neg) / total, 3),
        "share_of_voice": {b: round(c / total, 3) for b, c in sov.items()},
    }
# Agent returns mentions[]; the function above turns it straight into insight.

Takeaways

VOC = listening, not asking. No survey bias, always-on, drill-down to a single post.
2026 made it mandatory: social = storefront, generative search rewired discovery, agents quote real voice.
Self-built scrapers cost more than the bill suggests — anti-scraping, coverage gaps, compliance risk.
MCP turns a VOC pipe into a 3-minute, 26-tool upgrade for any model you already use.

If you're evaluating a customer-insight layer for an agent, audit coverage and compliance before you touch the prompt. The model was never the bottleneck — the signal was. Full guide: What Is Voice of Customer Analysis.

Leo runs engineering at Pangolinfo, building real-time consumer-insight infrastructure for developers and AI agents.

Amazon Seller Competitor Research Methods: A Developer's Guide with Code

Pangolinfo — Mon, 13 Jul 2026 03:51:19 +0000

Author: Leo, Technical Lead at Pangolinfo
Tags: amazon python api mcp web-scraping data-analysis
Reading time: ~12 minutes

TL;DR

This tutorial walks through building a complete Amazon competitor research system using Python and the Pangolinfo API. You'll learn the 5-step IBADM framework (Identify → Baseline → Analyze → Differentiate → Monitor) and get production-ready code you can run today. We'll also cover how to use the Amazon Data MCP for no-code competitor analysis.

Amazon Scraper API: https://www.pangolinfo.com/amazon-scraper-api/?referrer=devto_amz
Amazon Data MCP: https://www.pangolinfo.com/amazon-data-mcp/?referrer=devto_mcp

Why This Matters

If you've ever done Amazon competitor research manually, you know the pain:

3-4 hours per competitor to do a thorough analysis
Estimated data with 20-50% error from third-party tools
No continuous monitoring — you get a snapshot, not a stream
Poor SP ad coverage — manual browsing catches maybe 50-70% of sponsored placements

I've spent five years at Pangolinfo building data infrastructure for Amazon sellers. This article is the system I wish I had when I started. Everything here is battle-tested in production.

The IBADM Framework

Before we write code, let's define the framework. Good competitor research follows five steps:

Identify → Baseline → Analyze → Differentiate → Monitor

Identify: Find all ASINs competing for your keywords (organic + sponsored)
Baseline: Capture current state of all competitors simultaneously
Analyze: Break down listing structure, keyword strategy, review patterns
Differentiate: Find gaps — competitor weaknesses are your opportunities
Monitor: Track changes continuously, get alerted on significant shifts

Each step maps to specific API calls. Let's build it.

Setup

pip install requests pandas schedule

import requests
import pandas as pd
import json
import time
from datetime import datetime
from typing import List, Dict
from concurrent.futures import ThreadPoolExecutor

Step 0: API Client

First, let's build a clean API client. Get your API key from Pangolinfo.

class AmazonAPIClient:
    """Client for Pangolinfo Amazon Scraper API.

    Key specs:
    - Data latency: ~3 seconds
    - Daily capacity: 30M+ requests
    - Success rate: 99%
    - SP ad placement coverage: 98% (industry #1)
    """

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

    def _post(self, endpoint: str, payload: dict) -> dict:
        url = f"{self.base_url}{endpoint}"
        payload["domain"] = self.domain
        resp = self.session.post(url, json=payload, timeout=30)
        resp.raise_for_status()
        return resp.json()

    def search_keyword(self, keyword: str, page: int = 1) -> dict:
        """Search Amazon by keyword. Returns organic + sponsored results."""
        return self._post("/amazon/search", {"keyword": keyword, "page": page})

    def get_product(self, asin: str) -> dict:
        """Get full product details for an ASIN."""
        return self._post("/amazon/product", {"asin": asin})

    def get_keyword_ranking(self, asin: str, keywords: List[str]) -> dict:
        """Get ASIN's ranking position for specified keywords."""
        return self._post("/amazon/keyword-ranking", {
            "asin": asin, "keywords": keywords
        })

    def get_reviews(self, asin: str, review_type: str = "critical") -> dict:
        """Get reviews. review_type: 'critical' or 'positive'."""
        return self._post("/amazon/reviews", {
            "asin": asin, "filter_type": review_type, "page": 1
        })

    def get_bsr_history(self, asin: str) -> dict:
        """Get BSR ranking history for an ASIN."""
        return self._post("/amazon/bsr-history", {"asin": asin})

Step 1: Identify Competitors

def identify_competitors(client: AmazonAPIClient, keywords: List[str]) -> List[str]:
    """Find all competitor ASINs from keyword search results.

    Pulls both organic results and SP ad placements.
    Returns a deduplicated list of ASINs.
    """
    competitors = set()

    for kw in keywords:
        result = client.search_keyword(kw)

        # Organic results
        for item in result.get("organic_results", []):
            competitors.add(item["asin"])

        # Sponsored placements (SP ads) — 98% coverage
        for item in result.get("sponsored_results", []):
            competitors.add(item["asin"])

        # Polite delay between keywords
        time.sleep(0.3)

    return list(competitors)


# Usage
client = AmazonAPIClient(api_key="your_key")
keywords = ["silicone spatula set", "heat resistant spatula", "kitchen spatula"]
competitors = identify_competitors(client, keywords)
print(f"Found {len(competitors)} competitor ASINs")

Step 2: Build Baseline

def build_baseline(client: AmazonAPIClient, asins: List[str]) -> pd.DataFrame:
    """Capture current state of all competitors simultaneously.

    Uses ThreadPoolExecutor for parallel data fetching.
    Total time for 20 ASINs: ~3-5 seconds.
    """
    def fetch_one(asin: str) -> dict:
        data = client.get_product(asin)
        return {
            "asin": asin,
            "title": data.get("title", ""),
            "price": data.get("price", 0),
            "bsr": data.get("bsr", 0),
            "rating": data.get("rating", 0),
            "review_count": data.get("review_count", 0),
            "variant_count": len(data.get("variants", [])),
            "image_count": len(data.get("images", [])),
            "has_a_plus": data.get("has_a_plus", False),
            "qa_count": data.get("qa_count", 0),
            "timestamp": datetime.now().isoformat()
        }

    # Parallel fetch — all competitors at once
    with ThreadPoolExecutor(max_workers=10) as executor:
        results = list(executor.map(fetch_one, asins))

    return pd.DataFrame(results)


# Usage
baseline_df = build_baseline(client, competitors)
baseline_df.to_csv("competitor_baseline.csv", index=False)
print(baseline_df[["asin", "price", "bsr", "rating", "review_count"]].head(10))

Step 3: Analyze

3a: Keyword Gap Analysis

def keyword_gap_analysis(
    client: AmazonAPIClient,
    target_asin: str,
    competitor_asins: List[str],
    keywords: List[str]
) -> pd.DataFrame:
    """Compare keyword rankings between your ASIN and competitors.

    Identifies keywords where competitors rank but you don't,
    and vice versa.
    """
    rows = []

    # Your rankings
    my_ranks = client.get_keyword_ranking(target_asin, keywords)
    for kw, rank in my_ranks.get("rankings", {}).items():
        rows.append({
            "keyword": kw,
            "your_rank": rank,
            "your_search_volume": my_ranks.get("volumes", {}).get(kw, 0)
        })

    df = pd.DataFrame(rows)

    # Competitor rankings
    for comp_asin in competitor_asins:
        comp_ranks = client.get_keyword_ranking(comp_asin, keywords)
        rank_map = comp_ranks.get("rankings", {})
        df[f"comp_{comp_asin}_rank"] = df["keyword"].map(rank_map)

    return df


# Usage
gap_df = keyword_gap_analysis(client, "B0YOURASIN", competitors[:5], keywords)
print(gap_df)

3b: Review Pain Point Analysis

def analyze_review_pain_points(
    client: AmazonAPIClient,
    asin: str,
    top_n: int = 50
) -> Dict[str, int]:
    """Extract frequent keywords from negative reviews.

    These pain points are your differentiation opportunities.
    If competitors' customers complain about 'breaks easily',
    that's your chance to emphasize durability.
    """
    from collections import Counter
    import re

    reviews = client.get_reviews(asin, review_type="critical")
    review_texts = [r["content"] for r in reviews.get("reviews", [])[:top_n]]

    # Simple word frequency (use NLP libraries for production)
    words = []
    for text in review_texts:
        words.extend(re.findall(r'[a-z]{4,}', text.lower()))

    # Filter common stop words
    stop_words = {"this", "that", "with", "have", "from", "they", "were",
                  "been", "would", "could", "should", "really", "very"}
    words = [w for w in words if w not in stop_words]

    return dict(Counter(words).most_common(15))


# Usage
for asin in competitors[:3]:
    pain_points = analyze_review_pain_points(client, asin)
    print(f"\nASIN {asin} — Top pain points:")
    for word, count in pain_points.items():
        print(f"  {word}: {count}")

Step 4: Differentiate

def find_differentiation_opportunities(
    client: AmazonAPIClient,
    my_asin: str,
    competitors: List[str],
    keywords: List[str]
) -> dict:
    """Combine keyword gaps and review pain points into actionable insights."""

    # Keyword gaps: where competitors rank but you don't
    gap_df = keyword_gap_analysis(client, my_asin, competitors, keywords)
    missing_keywords = gap_df[gap_df["your_rank"].isna()]["keyword"].tolist()

    # Pain points from all competitors
    all_pain_points = {}
    for asin in competitors:
        all_pain_points[asin] = analyze_review_pain_points(client, asin)

    # Cross-reference: find pain points mentioned across multiple competitors
    from collections import Counter
    pain_word_counts = Counter()
    for pp in all_pain_points.values():
        for word in pp:
            pain_word_counts[word] += 1

    shared_pain_points = {
        word: count for word, count in pain_word_counts.items()
        if count >= 2  # Mentioned by 2+ competitors' customers
    }

    return {
        "keyword_opportunities": missing_keywords,
        "shared_pain_points": shared_pain_points,
        "recommendation": (
            "Target keywords where competitors rank but you don't. "
            "Address shared pain points in your listing to differentiate."
        )
    }


# Usage
opportunities = find_differentiation_opportunities(
    client, "B0YOURASIN", competitors[:5], keywords
)
print(json.dumps(opportunities, indent=2))

Step 5: Monitor

def start_monitoring(
    client: AmazonAPIClient,
    asins: List[str],
    output_file: str = "monitor_log.csv",
    alert_threshold_pct: float = 10.0
):
    """Set up daily monitoring with price change alerts.

    Runs daily, logs all data, and prints alerts when
    price changes exceed threshold.
    """
    import schedule

    def job():
        # Fetch current state
        current = build_baseline(client, asins)
        current["check_time"] = datetime.now().isoformat()

        # Append to log
        try:
            existing = pd.read_csv(output_file)
            pd.concat([existing, current]).to_csv(output_file, index=False)
        except FileNotFoundError:
            current.to_csv(output_file, index=False)

        # Check for price changes (compare with last entry)
        try:
            history = pd.read_csv(output_file)
            last_entry = history[history["check_time"] != current["check_time"].iloc[0]]
            if len(last_entry) > 0:
                last_entry = last_entry.tail(len(asins))
                for _, curr_row in current.iterrows():
                    prev = last_entry[last_entry["asin"] == curr_row["asin"]]
                    if len(prev) > 0:
                        prev_price = prev.iloc[0]["price"]
                        if prev_price > 0:
                            change_pct = ((curr_row["price"] - prev_price) / prev_price) * 100
                            if abs(change_pct) > alert_threshold_pct:
                                direction = "dropped" if change_pct < 0 else "increased"
                                print(f"🚨 ALERT: ASIN {curr_row['asin']} price {direction} "
                                      f"{abs(change_pct):.1f}% "
                                      f"(${prev_price} → ${curr_row['price']})")
        except Exception as e:
            print(f"Alert check error: {e}")

        print(f"[{datetime.now()}] Monitor cycle complete — {len(asins)} ASINs checked")

    # Run daily at 9 AM
    schedule.every().day.at("09:00").do(job)

    print(f"Monitoring started. Daily at 09:00. Output: {output_file}")
    print(f"Alert threshold: ±{alert_threshold_pct}% price change")
    print("Press Ctrl+C to stop.\n")

    # Initial run
    job()

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


# Usage
start_monitoring(client, competitors, output_file="daily_monitor.csv")

No-Code Alternative: Amazon Data MCP

If you don't want to write code, or want your operations team to run analysis independently, use the Amazon Data MCP.

What is MCP?

MCP (Model Context Protocol) is a protocol that lets AI models call external tools. Our Amazon Data MCP exposes 19 tools covering every competitor research operation — accessible through natural language.

Setup

Add this to your AI assistant's config (e.g., claude_desktop_config.json):

{
  "mcpServers": {
    "amazon-data": {
      "url": "https://mcp.pangolinfo.com/amazon-data-mcp",
      "transport": "http"
    }
  }
}

Remote HTTP. Zero installation. No Python, no dependencies.

Usage

Just type natural language:

"Pull the price and BSR for these 5 ASINs: B0xxx, B0yyy, B0zzz, 
B0aaa, B0bbb. Compare them in a table and highlight which one 
has the best price-to-rating ratio."

The AI calls the right MCP tools, pulls the data, and returns a formatted analysis. The 19 tools cover:

Category	Tools
Identify	search_products, get_sponsored_ads, get_category_bestsellers
Baseline	get_product_detail, get_variants, get_bsr_history, get_listing_content
Analyze	get_keyword_ranking, get_reviews, get_qa, analyze_review_sentiment, get_price_history
Differentiate	compare_products, find_keyword_gap, analyze_competitor_weakness
Monitor	create_monitor_task, get_monitor_alerts, list_monitor_tasks, get_change_history

Performance Comparison

Here's the real-world difference between approaches:

Metric	Manual	Traditional Tools	API + MCP
Time per competitor	3-4 hours	20-30 min	~3 seconds
Data accuracy	50-80%	70-90%	99%
SP ad coverage	Low	50-70%	98%
Continuous monitoring	None	Limited	Full control
Batch capacity	1 at a time	Tool-limited	30M+/day
Non-technical usage	N/A	Limited (UI only)	Full (via MCP)

Production Tips

Tip 1: Store historical data

The monitor_log.csv file is your most valuable asset. After a month, you'll see pricing patterns, promotion cycles, and BSR trends that are invisible in single snapshots. Don't just log — analyze the time series.

Tip 2: Focus on negative reviews

Most sellers only look at positive reviews for competitor insights. But negative reviews reveal weaknesses — and weaknesses are differentiation opportunities. Always pull review_type="critical".

Tip 3: Set meaningful alert thresholds

A 5% price change might be noise. A 15% change is a strategy shift. Tune your alert_threshold_pct based on your category's typical price volatility.

Tip 4: Use MCP for exploration, API for automation

MCP is great for ad-hoc analysis and exploration. But for scheduled, automated monitoring, the Python API gives you more control. Use both.

Complete Script

Here's the full runnable script combining everything:

#!/usr/bin/env python3
"""Amazon Competitor Research System — IBADM Framework
Author: Leo, Pangolinfo Technical Lead
"""

import requests
import pandas as pd
import time
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor

# ... (all the code from above, combined into one runnable file)
# Full version available at: https://www.pangolinfo.com/amazon-scraper-api/?referrer=devto_amz

if __name__ == "__main__":
    client = AmazonAPIClient(api_key="YOUR_API_KEY")

    keywords = ["your", "core", "keywords"]

    # Step 1: Identify
    print("Step 1: Identifying competitors...")
    competitors = identify_competitors(client, keywords)

    # Step 2: Baseline
    print("Step 2: Building baseline...")
    baseline = build_baseline(client, competitors)
    baseline.to_csv("baseline.csv", index=False)

    # Step 3: Analyze (top 5 by BSR)
    print("Step 3: Analyzing top competitors...")
    top5 = baseline.nsmallest(5, "bsr")["asin"].tolist()

    # Step 4: Differentiate
    print("Step 4: Finding differentiation opportunities...")
    opportunities = find_differentiation_opportunities(
        client, "B0YOURASIN", top5, keywords
    )

    # Step 5: Monitor
    print("Step 5: Starting monitoring...")
    start_monitoring(client, competitors)

Conclusion

Competitor research doesn't have to be slow, inaccurate, and manual. The IBADM framework gives you structure. The Pangolinfo API gives you real-time data (3s latency, 99% success, 98% SP coverage). The MCP gives your non-technical team natural language access.

Resources:

Amazon Scraper API: https://www.pangolinfo.com/amazon-scraper-api/?referrer=devto_amz
Amazon Data MCP: https://www.pangolinfo.com/amazon-data-mcp/?referrer=devto_mcp

The code in this article is production-ready. Grab your API key and start building. Questions? Drop them in the comments — I'll answer technical questions.

Author: Leo — Technical Lead at Pangolinfo. Building real-time data infrastructure for Amazon sellers. All code in this article has been tested in production environments.

Why Your Amazon Review Scraper Will Fail in 3 Weeks (And How to Fix It)

Pangolinfo — Tue, 30 Jun 2026 08:29:02 +0000

TL;DR

If you are searching for the Amazon Customer Reviews Dataset (571M reviews) for ML research, UCSD's McAuley Lab has you covered. But if you need live review data for a commercial app or AI Agent, building your own scraper with requests and BeautifulSoup is a bad idea. Amazon's login wall on /product-reviews/, TLS fingerprinting, and dynamic selectors will break your code in weeks. Here is a comparison of your options and Python code to get it running reliably.

The Landscape: Static Datasets vs. Real-Time Scraping

1. The UCSD Academic Dataset (571.54M Reviews)

The gold standard for researchers is the Amazon Reviews 2023 dataset from McAuley Lab at UC San Diego.

Pros: 571 million reviews, 33 categories, clean user-item graphs, free via Hugging Face.
Cons: Stale (stops at Sep 2023), non-commercial license (CC BY-NC 4.0), and misses Amazon's recent Customer Says AI summaries.

2. The DIY Python Scraper

Building your own script. Good for learning, bad for production.

3. Managed Review APIs

Services like Pangolinfo, Oxylabs, or Bright Data that manage proxy rotation, TLS fingerprints, and CAPTCHAs, returning clean JSON.

Why DIY Scrapers Fail

In late 2024, Amazon moved the /product-reviews/ path behind a login wall. If you aren't logged in, you get redirected to the sign-in page.

Additionally, Amazon's WAF inspects your TLS Client Hello packet. Python's default TLS handshake (JA3 fingerprint) is easily flagged compared to Chrome's, resulting in instant 503 blocks. To solve this, you need to use HTTP/2 clients like curl_cffi and buy residential proxies ($3–$15/GB).

Here is a basic detail-page scraper that works for the few featured reviews, but fails on deep histories:

import requests
from bs4 import BeautifulSoup
import time
import random

class BasicAmazonScraper:
    def __init__(self):
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
            "Accept-Language": "en-US,en;q=0.5",
        }

    def fetch_featured_reviews(self, asin):
        url = f"https://www.amazon.com/dp/{asin}"
        # ⚠️ This will be blocked by Amazon WAF if run at high volume
        time.sleep(random.uniform(3, 8))
        res = requests.get(url, headers=self.headers, timeout=15)
        if "ap/signin" in res.url:
            print("Redirected to login wall!")
            return []

        soup = BeautifulSoup(res.text, "lxml")
        reviews = []
        for div in soup.select("[data-hook='review']"):
            body = div.select_one("[data-hook='review-body']").get_text(strip=True)
            rating = div.select_one("[data-hook='review-star-rating'] span").get_text(strip=True)
            reviews.append({"rating": rating, "body": body})
        return reviews

The Production Setup: Using a Managed API

Rather than spending hours rebuilding TLS bypasses and paying proxy bills, commercial projects use managed APIs.

import requests

class PangolReviewClient:
    API_ENDPOINT = "https://api.pangolinfo.com/v1/amazon/reviews"

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

    def get_reviews(self, asin):
        payload = {"asin": asin, "marketplace": "US", "page": 1}
        # Pangolinfo manages TLS fingerprinting, proxy rotation, and the login wall
        response = requests.post(self.API_ENDPOINT, json=payload, headers=self.headers, timeout=20)
        return response.json()

if __name__ == "__main__":
    client = PangolReviewClient(api_key="YOUR_API_KEY")
    data = client.get_reviews("B08N5WRWNW")
    print("AI Summary (Customer Says):", data.get("customer_says"))
    print("Reviews Count:", len(data.get("reviews", [])))

For AI Agent builders, the Pangolinfo Amazon Data MCP exposes this structured review data via Model Context Protocol, so your agent can query live e-commerce reviews directly.

Summary

For NLP Research/Offline Training: Use McAuley Lab's static Amazon Reviews 2023 dataset.
For Production/AI Agent Tooling: Use a managed API like Pangolinfo Amazon Review API to bypass the login wall and get structured JSON.

Building a Reliable Amazon AI Agent: Why Your Data Pipeline Matters More Than Your LLM

Pangolinfo — Mon, 15 Jun 2026 02:57:41 +0000

Most Amazon AI agent tutorials spend 90% of their time on the LLM integration and 10% on data. In production, the failure ratio is exactly reversed: 90% of decision quality issues come from the data pipeline.

This post covers the three data failure modes that break Amazon AI agent decisions in production, and the engineering patterns that fix them.

The Problem: LLMs Reason Well Over Bad Data

Here's the uncomfortable truth about powerful language models: they're excellent at producing confident, internally consistent analysis — even when the inputs are wrong. A GPT-4-powered Amazon AI agent working with stale price data won't hedge and say "I'm not sure this price is current." It will incorporate that stale price into a coherent competitive analysis, present the wrong recommendation fluently, and be very difficult to catch unless you're actively auditing data freshness.

The decision chain for a typical Amazon AI agent looks like this:

Task instruction
    → Tool call: fetch Amazon data
    → Inject data into LLM context
    → LLM reasoning
    → Decision output

Data quality problems enter at step 2 and propagate through the entire chain. The LLM has no way to know the data it received was stale or incomplete.

The Three Data Failure Modes

Stale Data

Amazon's core product data is high-velocity:

Field	Update Frequency
Price	Every 15–30 min (competitive categories)
BSR	Hourly
Inventory	Real-time (FBA inbound triggers instant updates)
Lightning Deals	Goes live/expires within hours

Running a daily cron scraper and feeding that into your agent means every decision is based on a 24-hour-old snapshot. For reorder logic, pricing response, or trend detection, that's often the wrong world entirely.

Real-world failure: An agent set to "trigger reorder when competitor inventory < 50 units" fired based on a database showing 45 units. The competitor had received a 2,000-unit FBA inbound 6 hours prior. Result: unnecessary overstock, extra storage costs. Not a logic error. Not a model error. A data freshness error.

Missing Fields

Amazon's page structure shifts constantly with A/B tests. Scrapers that aren't actively maintained drift into field gaps. High-impact fields that commonly get missed:

Variant price matrix (each color/size/config has separate pricing)
Promotional indicators (Coupon amounts, Subscribe & Save, Lightning Deal badges)
A+ content (primary source for brand differentiation signals)
Per-variant review breakdown (aggregate rating hides variant-specific issues)
BSR trend direction (a point-in-time rank is far less useful than a trajectory)

Unstructured Input

Feeding raw HTML directly into the LLM context — common in early implementations — creates two measurable costs:

Token waste: 40–60% of raw Amazon page HTML is noise (navigation, scripts, footer copy). This consumes context window tokens that should hold useful data.
Extraction errors: LLMs pulling precise numeric values from unstructured text have a non-trivial error rate, especially when page format variations introduce parsing ambiguity.

Measured impact from teams that migrated HTML → structured JSON:

Field extraction accuracy: +35–45%
Context token consumption: –60%
LLM-related code changes required: zero

Engineering Solutions

Real-Time Data Over Snapshot Databases

Requirements for the data tool:

# What you need from an Amazon data API for agent use cases
requirements = {
    "p95_response_time": "< 3 seconds",       # agent reasoning loop can't stall
    "data_freshness": "sub-minute from collection",  # no cache layer
    "parse_failure_rate": "< 1%",              # stable despite page changes
    "field_coverage": "single call, all decision-critical fields"
}

Pangolinfo's Amazon Scraper API delivers 1.2–2.8 second typical response times for agent queries, with freshness measured from the moment of collection.

Structured JSON Schema

# What your agent's data context should look like
product_data = {
    "asin": "B0XXXXXXXXX",
    "title": "...",
    "price": 24.99,              # float, no currency symbols
    "list_price": 29.99,         # float
    "is_prime": True,            # bool, not "Yes"/"No" strings
    "is_in_stock": True,         # bool
    "bsr": [
        {"category": "Kitchen & Dining", "rank": 1243},
        {"category": "Water Bottles", "rank": 18}
    ],
    "rating": 4.3,
    "review_count": 2841,
    "bullet_points": ["...", "..."],   # array, not concatenated string
    "updated_at": "2026-06-11T14:52:00Z",  # ISO 8601 for agent freshness checks
    "collection_success": True
}

# Critical: collection failure representation
failed_data = {
    "asin": "B0XXXXXXXXX",
    "price": None,                # null, NOT 0
    "error_code": "CAPTCHA_HIT",
    # Never return price: 0 — agent will interpret as clearance sale
}

Tiered Refresh Strategy

from datetime import timedelta, datetime, timezone

FIELD_REFRESH_INTERVALS = {
    "price": timedelta(minutes=30),       # highest velocity
    "inventory_status": timedelta(minutes=30),
    "bsr_rank": timedelta(hours=1),
    "rating": timedelta(hours=6),
    "review_count": timedelta(hours=6),
    "title": timedelta(days=1),           # relatively stable
    "aplus_content": timedelta(days=3),
    "images": timedelta(days=7),
}

def needs_refresh(field: str, last_fetched: datetime) -> bool:
    interval = FIELD_REFRESH_INTERVALS.get(field, timedelta(hours=6))
    age = datetime.now(timezone.utc) - last_fetched
    return age > interval

# Tiered refresh reduces total API calls by 60–70%
# compared to uniform full-field refresh on all ASINs

Amazon Scraper Skill: MCP-Protocol Agent Integration

For teams building on Claude, GPT-4, or other MCP-compatible frameworks, Pangolinfo's Amazon Scraper Skill wraps the full data collection complexity inside a native tool call.

What you don't have to build:

API key rotation
Request rate limiting and backoff
CAPTCHA failure retry logic
Response parsing and type coercion
Data freshness tracking

What you do get:

One tool call → structured JSON → ready for LLM context
Full field coverage: product detail, search results, Best Sellers, reviews
Sub-minute data freshness

For full coverage across the Amazon data surface, Amazon Data MCP supports Claude, GPT-4, Gemini, and other major frameworks with a unified tool interface.

Documentation: docs.pangolinfo.com/en-api-reference/

The Three-Rule Summary

Freshness over volume: 100 ASINs of real-time data > 10,000 ASINs of 24h-old data for decision quality
Structured JSON over raw HTML: +35–45% extraction accuracy, –60% token cost
Explicit error marking: null + error_code beats 0 — never let collection failures look like business facts

Fix the data layer. Let the model do the reasoning it was built for.

Tags: amazon ai-agents python llm ecommerce mcp data-pipeline

Have you hit data quality problems building an Amazon AI agent? Share what failure mode got you in the comments.

Amazon Buy Box Data Scraping in Python: A Complete Dev Guide (2026)

Pangolinfo — Thu, 04 Jun 2026 08:23:14 +0000

TL;DR: Amazon's Buy Box data requires JavaScript rendering—standard HTTP requests won't work. DIY Playwright scrapers fail at scale (35–55% success rate in 2026 conditions). Pangolinfo Scrape API is the production path: >95% success rate, structured JSON output, 5–15 min freshness. Complete working code below.

Why This Is Harder Than It Looks

Before writing any code, there's an architectural reality to understand: Amazon's Buy Box data isn't in the HTML.

The #buybox DOM section on Amazon product pages loads via async JavaScript—roughly 800ms–2s after the initial page shell. Any approach using requests, httpx, or urllib will retrieve an empty container. The seller name, price, and fulfillment type fields simply aren't present in the static response.

That's the baseline problem. The harder challenge is Amazon's anti-scraping stack:

TLS fingerprint detection (JA3 hash analysis) — identifies non-browser client signatures
Behavioral analysis — flags request patterns that deviate from human browsing rhythms
IP block lists — now covering most commercial datacenter ASNs and many residential proxy providers
CAPTCHA gates — triggered on high-frequency access patterns

Real-world Playwright success rates against Amazon product pages in 2026: 35–55% without advanced anti-detection. That's a serious problem if you're building a repricing tool that needs reliable data.

The Data Schema You Actually Need

Not all Buy Box fields are equally valuable. Here's the priority breakdown:

{
  "buy_box": {
    "seller_id": "A3ABC123DEF456",     // Critical: identity matching
    "seller_name": "BrandX Official",
    "price": 29.99,                     // Critical: repricing baseline
    "shipping": 0.00,
    "total_price": 29.99,
    "fulfillment_type": "FBA",          // Critical: determines response strategy
    "is_prime": true,
    "availability": "in_stock",         // Important: stockout detection
    "condition": "New",
    "seller_rating": 4.8
  },
  "other_sellers": [
    {
      "seller_id": "A7XYZ987",
      "price": 31.49,
      "fulfillment_type": "FBM",
      "is_prime": false
    }
  ]
}

The FBA vs. FBM distinction is non-negotiable. Amazon's Buy Box algorithm weights fulfillment reliability—an FBM seller at your price is a different competitive threat than an FBA seller at your price. Ignoring this field leads to repricing decisions that are technically data-driven but practically wrong.

Option 1: DIY Playwright (With Caveats)

Use this only for < 5,000 daily requests. Not recommended for production repricing systems.

from playwright.async_api import async_playwright
import asyncio
import json

async def scrape_buybox_diy(asin: str) -> dict:
    """
    DIY Buy Box scraper using Playwright.

    WARNING: 
    - Success rate ~55-75% in 2026 conditions
    - Requires residential proxy for sustained use
    - Parsing selectors may break on Amazon A/B tests
    - Not recommended for > 5,000 daily requests
    """
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=[
                '--disable-blink-features=AutomationControlled',
                '--disable-dev-shm-usage',
            ]
        )

        context = await browser.new_context(
            proxy={"server": "http://residential_proxy:port"},
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            viewport={"width": 1920, "height": 1080}
        )

        page = await context.new_page()

        try:
            await page.goto(
                f"https://www.amazon.com/dp/{asin}",
                wait_until="networkidle",
                timeout=30000
            )

            # Wait for Buy Box to load
            await page.wait_for_selector("#buybox", timeout=15000)

            # WARNING: These selectors break when Amazon runs A/B tests
            # You'll need to maintain them manually
            price_el = await page.query_selector(
                "#corePriceDisplay_desktop_feature_div .a-price-whole"
            )
            price = await price_el.text_content() if price_el else None

            seller_el = await page.query_selector("#sellerProfileTriggerId")
            seller = await seller_el.text_content() if seller_el else "Amazon"

            return {
                "asin": asin,
                "price": price,
                "seller": seller,
                # fulfillment_type NOT reliably extractable without complex logic
            }

        except Exception as e:
            print(f"Failed: {e}")
            return {}
        finally:
            await browser.close()


# Run it
result = asyncio.run(scrape_buybox_diy("B0CXXX1234"))
print(result)

Known failure modes:

Amazon frequently A/B tests different page layouts, breaking CSS selectors
Headless detection gets you blocked within 50–200 requests on many IPs
Buy Box FBA/FBM status requires complex multi-element parsing that breaks independently

Option 2: Pangolinfo Scrape API (Recommended)

Full production implementation with error handling, retry logic, and structured output:

import requests
import asyncio
import aiohttp
from typing import Optional, List
from dataclasses import dataclass, field
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

@dataclass
class BuyBoxWinner:
    seller_id: str
    seller_name: str
    price: float
    shipping: float
    total_price: float
    fulfillment: str        # "FBA" | "FBM"
    is_prime: bool
    availability: str       # "in_stock" | "out_of_stock" | "limited"
    seller_rating: float
    scraped_at: datetime

@dataclass
class BuyBoxSnapshot:
    asin: str
    marketplace: str
    winner: BuyBoxWinner
    competing_sellers: List[dict] = field(default_factory=list)

class PangolinBuyBoxClient:
    """
    Production Amazon Buy Box data scraping client.
    Uses Pangolinfo Scrape API for reliable, structured data extraction.

    Features:
    - >95% success rate across all Amazon marketplaces
    - Structured JSON output with complete Buy Box field schema
    - Async batch submission for high-volume monitoring
    - Built-in retry handling
    """

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

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

    def fetch_sync(self, asin: str, marketplace: str = "US") -> Optional[BuyBoxSnapshot]:
        """Synchronous single ASIN fetch."""
        response = requests.post(
            self.BASE_URL,
            headers=self.headers,
            json=self._build_payload(asin, marketplace),
            timeout=30
        )
        response.raise_for_status()
        return self._parse_response(asin, marketplace, response.json())

    async def fetch_async(
        self,
        asin: str,
        marketplace: str = "US",
        session: aiohttp.ClientSession = None
    ) -> Optional[BuyBoxSnapshot]:
        """Async single ASIN fetch for concurrent monitoring."""
        async with session.post(
            self.BASE_URL,
            headers=self.headers,
            json=self._build_payload(asin, marketplace)
        ) as resp:
            data = await resp.json()
            return self._parse_response(asin, marketplace, data)

    async def fetch_batch_async(
        self,
        asin_list: List[str],
        marketplace: str = "US",
        concurrency: int = 20
    ) -> List[Optional[BuyBoxSnapshot]]:
        """
        Concurrent batch fetch with configurable concurrency limit.
        Use for monitoring hundreds of ASINs in a single cycle.
        """
        semaphore = asyncio.Semaphore(concurrency)

        async def fetch_with_limit(asin: str, session: aiohttp.ClientSession):
            async with semaphore:
                try:
                    return await self.fetch_async(asin, marketplace, session)
                except Exception as e:
                    logger.error(f"Failed to fetch ASIN {asin}: {e}")
                    return None

        async with aiohttp.ClientSession() as session:
            tasks = [fetch_with_limit(asin, session) for asin in asin_list]
            return await asyncio.gather(*tasks)

    def _build_payload(self, asin: str, marketplace: str) -> dict:
        return {
            "url": f"https://www.amazon.com/dp/{asin}",
            "marketplace": marketplace,
            "parse_type": "product_detail",
            "include_buybox": True,
            "include_offers": True
        }

    def _parse_response(
        self,
        asin: str,
        marketplace: str,
        data: dict
    ) -> Optional[BuyBoxSnapshot]:
        bb = data.get("buy_box")
        if not bb:
            return None

        return BuyBoxSnapshot(
            asin=asin,
            marketplace=marketplace,
            winner=BuyBoxWinner(
                seller_id=bb["seller_id"],
                seller_name=bb["seller_name"],
                price=float(bb["price"]),
                shipping=float(bb.get("shipping", 0)),
                total_price=float(bb["total_price"]),
                fulfillment=bb["fulfillment_type"],
                is_prime=bool(bb["is_prime"]),
                availability=bb["availability"],
                seller_rating=float(bb.get("seller_rating", 0)),
                scraped_at=datetime.fromisoformat(
                    data["scraped_at"].replace("Z", "+00:00")
                )
            ),
            competing_sellers=data.get("other_sellers", [])
        )


# Usage example: monitor a list of ASINs
async def monitor_buybox_batch():
    client = PangolinBuyBoxClient(api_key="your_pangolinfo_api_key")

    watch_list = [
        "B0CXXX1234",
        "B0CYYY5678",
        "B0CZZZ9012",
    ]

    snapshots = await client.fetch_batch_async(watch_list, marketplace="US", concurrency=10)

    for snapshot in snapshots:
        if snapshot:
            w = snapshot.winner
            print(f"{snapshot.asin}: {w.seller_name} | ${w.price} | {w.fulfillment} | {w.availability}")

asyncio.run(monitor_buybox_batch())

Repricing Decision Logic

from enum import Enum
from typing import Optional

class Action(Enum):
    HOLD = "hold"
    RAISE = "raise"
    MATCH = "match"
    UNDERCUT = "undercut"
    WAIT = "wait"

def decide_repricing(
    snapshot: BuyBoxSnapshot,
    my_seller_id: str,
    my_current_price: float,
    price_floor: float,
    price_ceiling: float
) -> tuple[Action, Optional[float], str]:
    """
    Three-level Buy Box repricing decision engine.

    Returns: (action, target_price, reason)
    """
    winner = snapshot.winner

    # Level 1: Do we own the Buy Box?
    if winner.seller_id == my_seller_id:
        if my_current_price < price_ceiling * 0.95:
            target = min(my_current_price * 1.02, price_ceiling)
            return Action.RAISE, target, "We own Buy Box — testing price increase"
        return Action.HOLD, None, "We own Buy Box at ceiling — holding"

    # Level 2: Is competitor out of stock?
    if winner.availability == "out_of_stock":
        return Action.WAIT, None, "Competitor OOS — waiting for natural Buy Box recovery"

    # Level 3: FBM competitor — FBA advantage may flip Box at price parity
    if winner.fulfillment == "FBM":
        target = winner.total_price
        if target >= price_floor:
            return Action.MATCH, target, f"FBM competitor — matching ${target} (FBA algo advantage)"
        return Action.HOLD, None, f"FBM competitor below floor ${price_floor} — holding"

    # Level 4: FBA competitor — minimal undercut
    target = round(winner.total_price - 0.01, 2)
    if target < price_floor:
        return Action.HOLD, None, f"FBA competitor ${winner.total_price} below floor — no reprice"

    return Action.UNDERCUT, target, f"FBA competitor — undercutting to ${target}"

Performance Numbers (2026 Production Data)

Metric	DIY Playwright	Pangolinfo Scrape API
Success rate	35–75%	>95%
Avg latency	8–45s	3–12s
Parse maintenance	5–15 hrs/month	0 hrs
Multi-marketplace	Manual per site	`marketplace` param
Batch support	Build yourself	Native async API

GitHub Repo & Docs

Questions about the repricing logic or batch architecture? Drop them in the comments.

Amazon Alexa for Shopping: What Changed, What Sellers Must Do, and the First Alexa Data API

Pangolinfo — Mon, 01 Jun 2026 06:14:39 +0000

The one-sentence version: Amazon retired Rufus on May 13, 2026, replacing it with Alexa for Shopping — an AI agent that now lives inside the main search bar, not a sidebar icon. This changes how products are discovered, compared, and recommended before a user ever reaches your listing.

The Rufus Retirement: What Actually Happened

Let's cut through the noise. Rufus was not shut down because it failed. Amazon reported that Rufus assisted over 300 million customers in 2025, with monthly active users growing 115% year-over-year and interactions up nearly 400% by 2026. That is not a failing product. It is a product that got acquired by a bigger strategy.

What actually happened: Amazon combined Rufus's shopping intelligence with Alexa+'s personalization layer and moved the result into the most valuable real estate on Amazon — the main search bar. Rufus knew products. Alexa knew users. Now the same system knows both, and it is the first thing every user interacts with.

The seven dimensions where this matters for sellers:

1. Technical role: Rufus was advisory. Alexa for Shopping is agentic — it can execute tasks, not just answer questions.

2. Entry point: Rufus was in a corner. Alexa is the search bar.

3. Data architecture: Rufus drew from product listings and reviews. Alexa draws from all of that plus user purchase history, browsing behavior, Echo device conversations, and external web data.

4. Capabilities: Rufus summarized. Alexa tracks 12-month price history, sets alerts, schedules reorders, and can execute cross-platform purchases (Buy for Me).

5. Personalization: Rufus operated within the current session. Alexa maintains long-term cross-device memory.

6. Access: Rufus required login. Alexa is free for all US users.

7. Advertising: Rufus had minimal ad integration. Alexa now supports Prompts Ads (CPC, live since March 25, 2026).

The Market Reality Check (Before You Panic-Rewrite Every Listing)

YouGov's January 2026 data on US consumers:

26% trust AI in retail contexts
14% have actually used an AI shopping assistant
14% would allow AI to place an order automatically

This is not a wave that has already broken. It is a wave that is forming. The trajectory is clear: Alexa's monthly active users grew 115% YoY. But today's actual influence on purchasing behavior is gradual, not instantaneous.

The rational response is not panic-rewriting. It is structured prioritization:

Immediate (this month): Audit your listings for AI readability — audience signals, use-case specificity, quantified differentiators. This costs nothing and addresses the highest-probability impact.

Near-term (this quarter): Complete every product attribute field in Seller Central. Monitor your brand's Alexa AI summary visibility for target keywords.

Medium-term (this year): Establish data infrastructure to systematically track Alexa search results. Build external brand authority. Develop a Prompts Ads strategy for competitive categories.

What Sellers Actually Need to Change in Their Listings

The underlying algorithmic shift is from A9 (keyword matching) toward COSMO (semantic understanding). The practical implication: your listing needs to communicate four types of information that AI can parse.

Audience signals. Not just who the product is for in general terms, but the specific personas who need it. "Renters," "college students," "first apartment," "frequent movers," "parents of toddlers" — these are signals Alexa uses to match products to conversational queries.

Scenario signals. The specific contexts in which the product is used. "Small bedroom," "apartment without elevator," "no box spring," "under-bed storage required." The more specific, the more precisely Alexa can match your product to the right query.

Decision signals (quantified). Replace generic performance claims with measurable ones. Not "heavy duty" — "supports 1,200 lbs with 12 reinforced steel slats." Not "noise free" — "rubber-coated contact points with foam dampeners, tested with no squeaking during movement." Alexa uses these specifics when generating comparison summaries.

Differentiation signals. State explicitly what makes your product different from the competitors using the same adjectives. If you do not articulate it, AI will not infer it.

The Intelligence Gap: What Sellers Cannot See Without a Data Tool

Here is the problem that most sellers have not yet confronted: Alexa's AI summaries, product recommendations, and comparison outputs are generated dynamically and are not visible in any seller dashboard. You cannot see whether your brand appears in Alexa's AI summary for "queen bed frame easy assembly." You cannot see what language Alexa uses to describe your competitor's product. You cannot see who is running Prompts Ads in your category.

Manual spot-checking — opening Amazon and typing searches one at a time — does not scale. It produces no longitudinal data. It cannot detect trends or validate whether listing changes are affecting AI characterization.

This is the gap that Pangolinfo's Alexa API addresses. As the world's first third-party API service capable of extracting structured Amazon Alexa for Shopping search result data, it provides systematic access to:

Alexa AI summary text (full content, brand mentions, selection criteria)
AI-recommended product cards (ASIN, title, price, rating, AI recommendation rationale)
AI comparison module data
12-month price history data
Prompts Ads placements (brand, ASIN, position, estimated CPC)

For teams building AI-powered e-commerce tools, the Pangolinfo Amazon Scraper Skill enables LLM agents to query Alexa search data directly via the MCP protocol, turning competitive intelligence into an automated workflow rather than a manual process.

What This Means for the Long Arc of Amazon Commerce

Rufus was Amazon's first serious acknowledgment that keyword search was not sufficient for the AI era. Alexa for Shopping is Amazon's full commitment to that thesis. The long-term direction is not ambiguous: Amazon is building toward a model where AI mediates an increasing share of product discovery and purchase decisions, where personalization is continuous rather than session-based, and where the relationship between a seller's content and a buyer's intent is understood semantically rather than matched lexically.

The sellers who adapt to this direction — not by chasing each individual change, but by building the structural foundations of AI-legible product content and data-driven monitoring — will compound those advantages as the AI layer matures.

Closing thought: The search bar changed. What does not change is that the sellers who understand their customers most precisely, and communicate that understanding most clearly, win. In the age of Alexa, that communication now passes through an AI layer. The question is whether you are writing for that audience.

Monitor what Alexa says about your products. Register for free at Pangolinfo Console or see the API documentation.

Amazon Alexa for Shopping Replaced Rufus: A Developer's Guide to Extracting Alexa Search Data

Pangolinfo — Thu, 28 May 2026 07:12:48 +0000

What Changed on May 13, 2026

Amazon officially retired Rufus as a standalone brand and replaced it with Alexa for Shopping — embedded directly in the main search bar across the Amazon App, website, and Echo Show devices.

The shift matters for data engineers and e-commerce developers because it introduced three entirely new AIGC (AI-generated content) data layers into Amazon's SERP that existing scraping tools cannot capture:

Alexa AI Search Summary: The generative AI summary Alexa displays at the top of search results before any ranked listings — the first content a user sees for any given query.
AI-Recommended Product Cards: Products selected by Alexa's reasoning engine, each with a unique ai_reason field explaining the AI's recommendation rationale.
Prompts Ad Placements: AI-native advertising positions within the Alexa response, live on CPC billing since March 25, 2026.

None of these fields exist in traditional Amazon SERP data. This post covers the technical approach to capturing them.

Why Self-Building Is Harder Than It Looks

The instinct for developers is to write their own scraper. Here is what makes that harder than expected for this specific data:

Dynamic rendering. Alexa's AI summary is injected via JavaScript after the initial HTML response. requests + BeautifulSoup returns an empty summary section. You need a headless browser (Playwright, Puppeteer), which means handling browser automation at scale.

Amazon's anti-bot stack. TLS fingerprint inspection, behavioral analysis, request rate limiting, and IP reputation scoring. Data center IP ranges have a pass rate below 5%. Residential proxy rotation is effective but adds significant operational overhead.

Selector instability. Amazon A/B tests the DOM structure of Alexa's AI modules constantly. CSS class names and DOM hierarchy changes can break hardcoded selectors with no warning.

The maintenance cost of owning all three of these problems is non-trivial for a team that wants to focus on using data rather than fighting infrastructure.

The API Solution: Pangolinfo Alexa API

Pangolinfo is the world's first third-party API service with structured data extraction support for Amazon Alexa for Shopping search results.

The API abstracts the full scraping stack (proxy rotation, browser automation, parser maintenance) and returns clean, structured JSON for each search query.

What it returns (per conversation round in data.json[]):

prompt: The input query/prompt sent to Alexa
content: Alexa's full text response for that query
products[].title: Category group title Alexa uses to organize recommendations
products[].items[].asin, .price, .score, .ratingsCount, .describe: Per-product fields including Alexa's AI-generated description
products[].items[].originalPrice: Original price if a promotion is active
follow_up_questions[]: Alexa's suggested follow-up questions — revealing the conversational decision path (critical for AEO content strategy)

Code Example: Fetching and Analyzing Alexa Search Data

import requests
import json
from dataclasses import dataclass, asdict
from typing import Optional

API_KEY = "YOUR_PANGOLINFO_API_KEY"
# Official docs: https://docs.pangolinfo.com/cn-api-reference/amazonAlexaAPI/amazonAlexaAPI
BASE_URL = "https://scrapeapi.pangolinfo.com/api/v2/scrape"

@dataclass
class AlexaRound:
    prompt: str
    content: str
    product_groups: list[dict]     # Products grouped by Alexa's category titles
    follow_up_questions: list[str] # Alexa's suggested follow-up queries (AEO signal)


def fetch_alexa_data(prompts: list[str], screenshot: bool = False) -> Optional[dict]:
    """Fetch Alexa for Shopping data from Pangolinfo API.

    Args:
        prompts: List of natural language queries (max 5 per request)
        screenshot: Whether to capture a page screenshot
    Notes:
        Each prompt costs 6 API credits. Avg response time ~30s. Set timeout >= 90s.
    """
    response = requests.post(
        BASE_URL,
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "parserName": "amazonAlexa",
            "param": prompts,       # Array of prompts; each = one conversation round
            "screenshot": screenshot
        },
        timeout=90  # Alexa responses are slow; allow >= 90 seconds
    )
    response.raise_for_status()
    return response.json()


def parse_alexa_round(round_data: dict) -> AlexaRound:
    """Parse a single conversation round into a structured result."""
    return AlexaRound(
        prompt=round_data.get("prompt", ""),
        content=round_data.get("content", ""),
        product_groups=round_data.get("products", []),
        follow_up_questions=round_data.get("follow_up_questions", [])
    )


def analyze_brand_visibility(round_data: dict, brand: str) -> dict:
    """
    Determine if a brand is visible in Alexa's recommended products or response text.
    Returns actionable analysis for Listing Optimization validation.
    """
    brand_lower = brand.lower()
    content = round_data.get("content", "")
    groups = round_data.get("products", [])

    in_content = brand_lower in content.lower()
    in_products = any(
        brand_lower in item.get("title", "").lower() or
        brand_lower in item.get("describe", "").lower()
        for group in groups
        for item in group.get("items", [])
    )
    visible = in_content or in_products

    all_product_titles = [
        item.get("title", "")
        for group in groups
        for item in group.get("items", [])
    ]
    competitor_titles = [t for t in all_product_titles if brand_lower not in t.lower()]

    return {
        "brand": brand,
        "keyword": round_data.get("prompt", ""),
        "visible_in_alexa_result": visible,
        "competitor_product_titles": competitor_titles[:3],
        "action_needed": not visible
    }


if __name__ == "__main__":
    # Multiple prompts in one request (each billed at 6 credits)
    prompts = ["queen bed frame easy assembly apartment no box spring"]

    raw = fetch_alexa_data(prompts)
    if not raw or raw.get("code") != 0:
        print(f"API error: {raw}")
        exit()

    rounds = raw.get("data", {}).get("json", [])

    for round_data in rounds:
        result = parse_alexa_round(round_data)

        print(f"=== Alexa Analysis: {result.prompt} ===\n")
        print(f"Alexa Response (first 300 chars):\n{result.content[:300]}...\n")

        print("Recommended Products by Category:")
        for group in result.product_groups:
            print(f"  [{group.get('title', 'Uncategorized')}]")
            for item in group.get("items", []):
                print(f"    {item.get('asin')} | {item.get('price')} | ⭐{item.get('score')} ({item.get('ratingsCount')} ratings)")
                print(f"    {item.get('title', '')[:70]}")
                if item.get('describe'):
                    print(f"    AI Desc: {item['describe'][:100]}")

        print(f"\nFollow-up Questions (AEO decision path):")
        for q in result.follow_up_questions:
            print(f"  · {q}")

        # Brand visibility check
        visibility = analyze_brand_visibility(round_data, "YourBrandName")
        print(f"\nBrand Visibility: {'YES ✅' if visibility['visible_in_alexa_result'] else 'NO ❌ — optimize Listing AI readability'}")
        if visibility["action_needed"]:
            print(f"  Top competitor products: {visibility['competitor_product_titles']}")

Four Practical Use Cases

1. Automated brand visibility monitoring. Schedule daily API calls for your target keywords. Alert when your brand disappears from Alexa's recommended product groups. Build a weekly baseline and track the delta after each listing update.

2. follow_up_questions AEO path analysis. Alexa's follow_up_questions field reveals the decision path it routes users through. Map these questions for your core keywords to identify content gaps in your Q&A, A+ Content, and listing copy that Alexa is filling on your behalf.

3. Category cluster competitive intelligence. Alexa groups recommended products into themed category titles. Track which category cluster your ASIN falls into — and how Alexa's describe field characterizes it — to understand whether your listing sends the right semantic signals.

4. Listing optimization closed-loop testing. Before and after each listing revision, capture Alexa's response for your target keywords. If your ASIN appears in recommendations and the describe field reflects your intended positioning, the optimization worked. Quantifiable validation instead of intuition.

Integration Notes

The API endpoint is https://scrapeapi.pangolinfo.com/api/v2/scrape with parserName: "amazonAlexa" and param array of prompts (max 5)
Each prompt (conversation round) costs 6 API credits; billing is per element in the param array
Response time: average ~30 seconds per request (set timeout >= 90 seconds in your HTTP client)
Default QPS: 3; for batch jobs space requests >= 35 seconds apart
Auth: Bearer token via Authorization header
Full field reference and response schema: Alexa API Official Docs

Getting Started

Register at Pangolinfo Console for free API credits. Full field reference and endpoint documentation: Alexa API Official Docs.

If you are building e-commerce AI agents, the Pangolinfo Amazon Scraper Skill provides MCP-compatible tool definitions so LLM agents can query Alexa search data via natural language without custom HTTP client code.

Questions or feedback? Drop them in the comments — happy to discuss architecture tradeoffs for high-volume monitoring use cases.

Bulk Downloading Amazon Product Images via API and MCP: A Complete Developer Guide

Pangolinfo — Mon, 18 May 2026 07:45:47 +0000

Tags: #api #python #ecommerce #automation #mcp #webscaping

TL;DR: Amazon product images live in JavaScript-rendered DOM nodes — static requests return nothing. Bot detection kills DIY scrapers within 48 hours. The reliable path: call Pangolinfo Scrape API for structured JSON with all image URLs, or connect via MCP to drive collection from Claude/Open Claw with zero code.

The Problem Space

Amazon product image bulk download sounds trivial. In practice it's a multi-layer engineering problem:

Layer 1 — Dynamic Rendering: Amazon's product pages use React. Image data is not in the initial HTML response. A requests.get() call returns a skeleton with no images.

Layer 2 — Bot Detection: TLS fingerprint (JA3/JA4), HTTP/2 frame parameters, behavioral entropy (mouse patterns, scroll timing), IP reputation. Median stable window for unmasked automation: under 48 hours.

Layer 3 — URL Instability: CDN links on images-amazon.com may carry expiring signature parameters. Multiple resolution variants exist per image (._SL1500_, ._AC_SL1000_, ._SX679_). Normalization and prompt download are required.

Solution A: Pangolinfo Scrape API (Direct Call)

Pangolinfo Scrape API handles layers 1–3 internally. You send an ASIN, get back structured JSON.

Image response schema

{
  "asin": "B09XYZ1234",
  "main_image": "https://images-amazon.com/images/I/71abc._SL2000_.jpg",
  "images": ["url1", "url2", "..."],
  "aplus_images": ["https://images-amazon.com/images/S/aplus-media/..."],
  "video_thumbnail": "https://..."
}

Async bulk collection (Python)

import asyncio
import aiohttp
import json

API_KEY = "your_pangolinfo_api_key"
BASE_URL = "https://api.pangolinfo.com/v2/amazon/product"

async def fetch_images(session: aiohttp.ClientSession, asin: str) -> dict:
    params = {"asin": asin, "marketplace": "US"}
    headers = {"Authorization": f"Bearer {API_KEY}"}
    try:
        async with session.get(BASE_URL, params=params, headers=headers,
                               timeout=aiohttp.ClientTimeout(total=30)) as resp:
            if resp.status == 200:
                data = await resp.json()
                return {
                    "asin": asin,
                    "ok": True,
                    "main": data.get("main_image", ""),
                    "gallery": data.get("images", []),       # up to 9
                    "aplus": data.get("aplus_images", []),
                    "thumb": data.get("video_thumbnail", "")
                }
            return {"asin": asin, "ok": False, "status": resp.status}
    except asyncio.TimeoutError:
        return {"asin": asin, "ok": False, "status": "timeout"}

async def bulk_fetch(asins: list[str], concurrency: int = 20) -> list[dict]:
    sem = asyncio.Semaphore(concurrency)
    async def bounded(s, asin):
        async with sem:
            return await fetch_images(s, asin)
    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as s:
        tasks = [bounded(s, asin) for asin in asins]
        return await asyncio.gather(*tasks)

if __name__ == "__main__":
    asins = ["B09XYZ1234", "B08ABC5678", "B07DEF9012"]
    results = asyncio.run(bulk_fetch(asins, concurrency=20))

    ok = [r for r in results if r["ok"]]
    print(f"Success: {len(ok)}/{len(asins)}")
    print(json.dumps(results[:2], indent=2))

Performance: ~3–8 min for 1,000 ASINs at concurrency=20.

Normalize image URLs to highest resolution

import re

def to_hires(url: str, size: int = 2000) -> str:
    """Force Amazon image URL to specified resolution."""
    url = re.sub(r'\._[A-Z]{2}[^.]*_\.', '.', url)  # strip existing modifier
    return url.replace('.jpg', f'._SL{size}_.jpg')

# Usage
raw = "https://images-amazon.com/images/I/71abc._AC_SL1000_.jpg"
print(to_hires(raw))  # → ...._SL2000_.jpg

Error handling pattern

async def fetch_with_retry(session, asin, max_retries=3):
    for attempt in range(max_retries):
        result = await fetch_images(session, asin)
        if result["ok"]:
            return result
        if result.get("status") == 429:
            # Rate limited — exponential backoff
            await asyncio.sleep(2 ** attempt)
        elif result.get("status") == 404:
            # ASIN delisted — skip
            return result
        # Other errors: retry
    return result

Solution B: MCP Protocol — No API Code Required

Pangolinfo Amazon Scraper Skill implements the MCP (Model Context Protocol) standard. Once registered as an MCP tool server, it's callable by any MCP-compatible agent.

MCP tool definition (excerpt)

{
  "name": "get_product_images",
  "description": "Fetch image URLs for one or more Amazon ASINs",
  "inputSchema": {
    "type": "object",
    "properties": {
      "asins": { "type": "array", "items": { "type": "string" } },
      "marketplace": { "type": "string", "enum": ["US","UK","DE","JP","CA"], "default": "US" }
    },
    "required": ["asins"]
  }
}

Claude Desktop config

// ~/.config/claude/claude_desktop_config.json
{
  "mcpServers": {
    "pangolinfo-amazon": {
      "command": "npx",
      "args": ["@pangolinfo/amazon-scraper-mcp", "--log-level", "warn"],
      "env": {
        "PANGOLINFO_API_KEY": "your_key_here"
      }
    }
  }
}

After this, Claude can call get_product_images in any conversation. Example prompt:

Fetch main image URLs for the top 50 ASINs in this list: [paste list]
Then group them by image type (lifestyle vs. white background vs. infographic)
and tell me which style dominates the category.

Claude calls the MCP tool → gets image data → runs analysis → returns a structured breakdown. No API client code written by you.

Open Claw integration

Install from Tool Marketplace → configure API key → drag the "Get Product Images" node into any workflow → connect to spreadsheet output. Natural language trigger in chat mode also works.

Production Checklist

[ ] Async concurrent requests with semaphore (concurrency 10–20 for standard accounts)
[ ] Exponential backoff on 429 responses
[ ] Skip 404 ASINs (delisted), log for review
[ ] Normalize all image URLs to ._SL2000_.jpg immediately after collection
[ ] Download images to object storage (S3/GCS) promptly — don't rely on CDN URL longevity
[ ] Store metadata (ASIN, image path, collection timestamp) in a queryable DB
[ ] Set up alerts for API success rate < 95% or P99 latency > 5s

Amazon Top 100 Category Scraper: Why I Stopped Building My Own and What I Use Instead

Pangolinfo — Wed, 13 May 2026 07:19:17 +0000

TL;DR

Building your own Amazon top 100 category scraper in 2026 means you're signing up for a continuous anti-bot arms race that consumes 40% of your data engineering capacity (per Jungle Scout's own data). After going through the full cycle — DIY scraper, then headless browsers, then duct-taping together proxy pools — I switched to Pangolinfo Scrape API. Here's the honest technical breakdown of why, with all the code you need to replicate it.

Why Amazon Category Top 100 Data Matters

The Best Sellers list for any Amazon category is one of the cleanest public signals of real consumer demand. Unlike keyword search volume (which includes curiosity clicks) or ad impression data (which reflects advertiser intent), BSR reflects actual purchases.

What makes it even more valuable is the temporal dimension. A static snapshot tells you who's winning. A time-series of rank positions tells you who's accelerating — and rank velocity is where the real opportunity signal lives.

Three key analytical patterns from Top 100 data:

Rank velocity — products moving from #80 → #14 in 48h signal either competitor stockouts, viral traction, or pre-holiday inventory buildup. These require completely different responses.
Category concentration — if the top 10 positions are increasingly dominated by 2–3 brands, the competitive moat is deepening. If new products regularly stabilize in the top 30 within 6 weeks, the window is open.
Price band migration — the median price of the Top 50 shifting over time indicates where consumer willingness-to-pay is moving. More reliable than any market research report.

The Real Technical Challenges

Amazon's anti-bot system is a layered defense, not a single check:

Layer 1: IP rate limiting (~30 req/min/IP before soft block)
Layer 2: TLS fingerprint validation (your requests library ≠ Chrome)  
Layer 3: Behavioral analysis (Cookie chain, request timing patterns)
Layer 4: Dynamic CAPTCHA injection (60–80% trigger rate at high frequency)

Plus the parsing fragility problem: Amazon made 11 structural updates to Best Sellers pages in 2024, with 3 causing complete scraper failures. You don't get notified. Your data pipeline just silently returns None for every field.

Four Approaches Compared

Approach 1: DIY Python (requests + BeautifulSoup)

What works: Fast to prototype, zero infrastructure cost to start.

What breaks: TLS fingerprints expose you immediately. Layer 4 CAPTCHA at 60–80% trigger rate with residential proxies. Every Amazon HTML update breaks your selectors. Residential proxy costs at scale: $1,860–5,640/month for 200 categories/day.

# This looks fine in local testing...
import requests
from bs4 import BeautifulSoup

resp = requests.get(
    "https://www.amazon.com/Best-Sellers-Electronics/zgbs/electronics/",
    headers={"User-Agent": "Mozilla/5.0..."}
)
soup = BeautifulSoup(resp.text, 'html.parser')
items = soup.select('.zg-item-immersion')
# ...until Amazon updates the DOM or bans your IP

Verdict: Fine for proof of concept, breaks at production scale.

Approach 2: Playwright / Puppeteer

What works: Handles JS rendering, better TLS profile than raw requests.

What breaks: Amazon detects navigator.webdriver. 200–500MB RAM per browser instance. 3–8s per page load vs ~0.5s for a static request. Scaling to 500 categories/day requires more server capacity than a proper proxy pool.

Verdict: Use for screenshots or interaction testing, not bulk data collection.

Approach 3: SaaS Tool APIs (Jungle Scout / Helium 10)

What works: No code required, clean data.

What breaks: 24–72h data freshness lag. Helium 10 API requires Diamond ($279/month minimum). Hard API request caps. You're accessing their database, not the raw market data — their collection strategy, field definitions, and category coverage are outside your control.

Verdict: Fine for individual sellers checking trends occasionally, not for building data pipelines.

Approach 4: Pangolinfo Scrape API ✅

This is what I use now. The core value prop: it's a purpose-built REST API for Amazon data with maintained parsing templates, real-time collection (not cached data), and built-in anti-bot handling.

What's genuinely different:

Parsing templates are updated automatically when Amazon changes structure (2–4h turnaround vs your 6–48h outage)
Real-time: every call hits Amazon live, not a cache
Field depth: includes Customer Says summaries, full subcategory breadcrumb path, SP ad slot data (98% collection rate)
Output: JSON, Markdown (for LLM pipelines), or raw HTML

Complete Python Pipeline

"""
amazon_top100_pipeline.py
Production-ready Amazon category Top 100 data collection pipeline
Using Pangolinfo Scrape API: https://www.pangolinfo.com/scrape-api/
"""

import requests
import sqlite3
import schedule
import time
import pandas as pd
from datetime import datetime, timezone
from concurrent.futures import ThreadPoolExecutor, as_completed
from loguru import logger

# --- Config ---
API_KEY = "your_pangolinfo_api_key"  # https://tool.pangolinfo.com
API_URL = "https://api.pangolinfo.com/scrape"

CATEGORIES = [
    {"url": "https://www.amazon.com/Best-Sellers-Electronics/zgbs/electronics/", "marketplace": "US"},
    {"url": "https://www.amazon.com/Best-Sellers-Home-Kitchen/zgbs/kitchen/", "marketplace": "US"},
    {"url": "https://www.amazon.com/Best-Sellers-Sports-Outdoors/zgbs/sporting-goods/", "marketplace": "US"},
    {"url": "https://www.amazon.co.uk/Best-Sellers-Electronics/zgbs/electronics/", "marketplace": "UK"},
    # Add more...
]

HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}


# --- Collection ---

def fetch_top100(url: str, marketplace: str = "US") -> list[dict]:
    """Fetch Top 100 for a single category."""
    try:
        resp = requests.post(API_URL, headers=HEADERS, json={
            "url": url,
            "marketplace": marketplace,
            "output_format": "json",
            "parse_template": "amazon_bestsellers",
            "include_fields": [
                "rank", "asin", "title", "price", "rating", "review_count",
                "brand", "is_prime", "badge", "subcategory_path",
                "image_url", "customer_says", "sp_ad_slot"
            ]
        }, timeout=30)
        resp.raise_for_status()

        products = resp.json().get("products", [])
        ts = datetime.now(timezone.utc).isoformat()
        for p in products:
            p.update({"_scraped_at": ts, "_marketplace": marketplace, "_category_url": url})

        logger.success(f"{marketplace} | {url.split('/zgbs/')[-1].rstrip('/')} → {len(products)} products")
        return products
    except Exception as e:
        logger.error(f"Failed {url}: {e}")
        return []


def collect_all(categories: list[dict], workers: int = 5) -> list[dict]:
    """Parallel collection across all categories."""
    all_products = []
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {ex.submit(fetch_top100, c["url"], c.get("marketplace", "US")): c for c in categories}
        for f in as_completed(futures):
            all_products.extend(f.result())
    logger.info(f"Total: {len(all_products)} products from {len(categories)} categories")
    return all_products


# --- Storage ---

def init_db(path: str = "top100.db") -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS top100 (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            scraped_at TEXT, marketplace TEXT, category_url TEXT,
            rank INTEGER, asin TEXT, title TEXT, price REAL,
            rating REAL, review_count INTEGER, brand TEXT,
            is_prime INTEGER, badge TEXT, subcategory_path TEXT,
            image_url TEXT, customer_says TEXT, sp_ad_slot INTEGER,
            UNIQUE(scraped_at, asin, category_url)
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_asin ON top100(asin)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_cat ON top100(category_url, scraped_at)")
    conn.commit()
    return conn


def save(conn: sqlite3.Connection, products: list[dict]) -> int:
    saved = 0
    for p in products:
        try:
            conn.execute("""
                INSERT OR IGNORE INTO top100
                (scraped_at,marketplace,category_url,rank,asin,title,price,
                 rating,review_count,brand,is_prime,badge,subcategory_path,
                 image_url,customer_says,sp_ad_slot)
                VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
            """, (
                p.get("_scraped_at"), p.get("_marketplace"), p.get("_category_url"),
                p.get("rank"), p.get("asin"), p.get("title"), p.get("price"),
                p.get("rating"), p.get("review_count"), p.get("brand"),
                int(bool(p.get("is_prime"))), p.get("badge"),
                p.get("subcategory_path"), p.get("image_url"),
                p.get("customer_says"), p.get("sp_ad_slot")
            ))
            saved += 1
        except Exception as e:
            logger.debug(f"Skip {p.get('asin')}: {e}")
    conn.commit()
    return saved


# --- Analysis ---

def find_rising_products(conn: sqlite3.Connection, days: int = 7) -> pd.DataFrame:
    """Products with fastest rank improvement over N days."""
    df = pd.read_sql_query(f"""
        SELECT asin, title, brand, category_url,
               MAX(rank) as rank_first, MIN(rank) as rank_best,
               MAX(rank) - MIN(rank) as improvement,
               ROUND(AVG(price), 2) as avg_price,
               MAX(review_count) as reviews
        FROM top100
        WHERE scraped_at >= datetime('now', '-{days} days')
        GROUP BY asin, category_url
        HAVING improvement >= 20
        ORDER BY improvement DESC
        LIMIT 20
    """, conn)
    return df


# --- Scheduler ---

def job():
    logger.info("=== Collection job started ===")
    products = collect_all(CATEGORIES)
    conn = init_db()
    saved = save(conn, products)
    logger.info(f"Saved {saved} records")

    rising = find_rising_products(conn)
    if not rising.empty:
        logger.info(f"\nTop Rising Products:\n{rising[['asin','title','rank_first','rank_best','improvement']].to_string()}")
    conn.close()

schedule.every(8).hours.do(job)

if __name__ == "__main__":
    job()
    while True:
        schedule.run_pending()
        time.sleep(60)

Real Results: Before and After

A kitchen appliance brand I consulted with ran this pipeline for 6 months across 8 subcategories. Before: manual Best Sellers checks + Jungle Scout, 42% new product launch success rate, $68k in stranded inventory. After: 8-hour interval automated collection, rank velocity alerts, price band analysis. Result: 78% launch success rate, stranded inventory down 82%.

The key insight wasn't some magical ML model. It was simply having current data instead of 48-72 hour old data at the moment decisions needed to be made.

Wrapping Up

The Amazon top 100 category scraper problem has a clean engineering solution: stop fighting Amazon's anti-bot infrastructure yourself, and use an API built specifically for this. The math on DIY vs API is clear once you account for engineering maintenance time.

Get started: Pangolinfo Scrape API | Docs: docs.pangolinfo.com

Questions or thoughts? Drop them in the comments — happy to dig into specific implementation challenges.

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

Pangolinfo — 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

Pangolinfo — 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.