DEV Community: Mox Loop

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

Mox Loop — 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

Mox Loop — 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

Mox Loop — 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

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

TL;DR

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

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

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

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

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

Setup

pip install aiohttp python-dotenv schedule

# .env
PANGOLINFO_API_KEY=your_api_key_here

Core Module: Single ASIN Query

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

load_dotenv()

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

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

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

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

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

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

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

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

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

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

    return snapshot

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

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

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

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

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

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

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

    return snapshot

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

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

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

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

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

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

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

Daily Monitoring Pipeline with Change Detection

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

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

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

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

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

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

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

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

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

Key Design Decisions

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

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

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

Resources

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

Questions or issues? Drop them in the comments.

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

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

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

Why this exists

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

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

The framework

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

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

Picking the data source

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

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

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

Architecture

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

Code — async client

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

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

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

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

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

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

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

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

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

Code — metric computation

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

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

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

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

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

Code — the orchestration

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

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

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

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

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

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

Running it

import asyncio

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

asyncio.run(main())

Sample output for a representative category:

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

Production hardening

A few things to add before this hits production:

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

import redis, json
r = redis.Redis()

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

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

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

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

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

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

Why this beats SaaS tools

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

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

Closing

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

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

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

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

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

Background

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

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

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

Why Self-Built Scrapers Break at Scale

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

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

The root causes:

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

Solution: Pangolinfo Scrape API

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

API response schema for MnS:

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

Full Implementation

Async Multi-Category Collector

"""
mns_async_collector.py

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

Requirements: pip install aiohttp python-dotenv
"""

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

logger = logging.getLogger(__name__)

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

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


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

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


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

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

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


# --- Spike Detection & Scoring ---

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

    gain_score = min(50.0, gain / 200)

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

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

    return round(gain_score + comp_score + rating_score, 2)


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

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


# --- Main Orchestration ---

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

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

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

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

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

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

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

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


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

Production Deployment Notes

Environment setup:

# .env
PANGOLINFO_API_KEY=your_key_here

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

Redis deduplication (optional but recommended):

import redis

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

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

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

References & Further Reading

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

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

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

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

TL;DR

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

The Problem With Standard Product Research Tools

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

That means:

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

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

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

What You'll Build

A three-layer automated product discovery pipeline:

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

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

Setup

pip install requests pandas python-dotenv

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

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

Get your API key at tool.pangolinfo.com.

Layer 1: Real-Time Bestseller Scanner

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


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

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

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

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

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

    return candidates


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

Layer 2: Keyword Competition Analyzer

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


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

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

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

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

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


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

Layer 3: Negative Review Miner

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

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


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

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

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

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

Putting It All Together

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


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

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

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


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

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

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

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

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

    return pd.DataFrame(shortlist[:20])


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

Performance Tips

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

Scaling Beyond the Script

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

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

Resources

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

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

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

TL;DR

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

What you'll build:

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

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

Why Subscription Tools Don't Work for Programmatic Use

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

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

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

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

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

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

Setup

# Install dependencies
pip install requests aiohttp python-dotenv

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

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

1. Basic Sync Client

Start with the simplest working implementation:

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

load_dotenv()

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

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

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

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

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

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


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

2. Async Batch Collection (The Production Version)

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

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

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


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

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

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

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

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

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


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

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

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


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

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

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

    snapshots = await batch_fetch(competitor_asins, api_key)

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

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


asyncio.run(main())

3. Price Movement Detection + Feishu Alerts

import json
import requests
from datetime import datetime

FEISHU_WEBHOOK = os.getenv("FEISHU_WEBHOOK_URL")

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

    alerts = []

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

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

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

    return alerts


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

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

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

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

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

4. AI Agent Integration via Amazon Scraper Skill

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

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

Example Agent conversations that now work:

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

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

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

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

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

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

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

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

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

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

Full Monitoring Script (Bring It All Together)

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

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

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

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

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


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

    global previous_snapshots

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

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

    # Update cache
    previous_snapshots = current

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


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

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

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

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


if __name__ == "__main__":
    main()

Deployment

# Local background process
nohup python monitor.py &

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

Key Takeaways

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

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

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

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

TL;DR

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

The Problem with Defaulting to Sync

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

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

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

API Endpoint Comparison

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

Sync Mode: Complete Example

import requests

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


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

Available Parser Types

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

Async Mode: Complete Implementation

1. Submit Tasks

import requests
import time
from typing import Optional

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

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


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

2. Callback Receiver (FastAPI)

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

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

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

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

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

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

3. Timeout Compensation

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

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

Decision Framework

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

Local Development Setup

Testing async locally without deploying to a server:

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

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

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

Performance Benchmarks

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

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

Resources

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

Questions? Drop them in the comments.

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

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

Who This Is For

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

Background

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

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

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

The Infrastructure Problem, Technically

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

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

The fix requires:

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

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

The Solution: Pangolinfo Scrape API

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

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

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

The Implementation: 90-Day Phased Migration

Core Approach: Traffic Splitting + Parallel Comparison

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

# Simplified traffic routing logic
import random

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

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

Parallel Comparison Layer

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

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

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

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

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

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

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

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

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

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

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

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

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

    client = PangolinScrapeClient(api_key)
    results = []

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

    return results

Results: 6 Months Post-Migration

BEFORE vs AFTER

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

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

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

Dev Notes: What Actually Tripped Us Up

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

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

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

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

Conclusion

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

For teams considering a similar migration:

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

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

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

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

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

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

TL;DR

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

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

What We're Building

A complete Amazon auto listing system with:

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

Prerequisites

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

Install dependencies:

pip install requests aiohttp

Part 1: Authentication

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


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

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

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

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

Part 2: Listings Items API (Real-Time)

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


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

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

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

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

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

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

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

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

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

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

        return result

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

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

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

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

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

Part 3: Real-Time Pricing with Pangolinfo

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


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

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

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

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

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

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

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

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

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

        if not comp_price:
            return False

        new_price = self.calculate_target_price(comp_price, unit_cost)

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

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

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

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

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

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

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

Complete Integration Example

# main.py — Full auto listing + repricing example

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

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

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

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

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

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

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

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

Common Issues

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

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

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

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

Resources

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

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

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

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

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

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

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

TL;DR

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

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

Why Amazon Scrapers Keep Failing in 2026

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

The five-layer defense stack:

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

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

Comparing Approaches

Approach 1: DIY Scrapy/Requests + Proxy Pool

# What everyone starts with
import requests
from scrapy import Selector

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

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

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

Reality check on this approach:

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

Verdict: Acceptable for experimentation. Not for production.

Approach 2: Dedicated E-Commerce Scraping API

import requests
from typing import Optional

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

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

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

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

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

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

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

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

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

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

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


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

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

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

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

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

Real-World Performance Comparison {#implementation}

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

Best Sellers category pages:

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

Product detail pages with advertising data:

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

Parse reliability over 30 days:

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

AI Agent Integration

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

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

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

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

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

Quick Start

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

Summary

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

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

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

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

DEV Community: Mox Loop

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

What Changed on May 13, 2026

Why Self-Building Is Harder Than It Looks

The API Solution: Pangolinfo Alexa API

Code Example: Fetching and Analyzing Alexa Search Data

Four Practical Use Cases

Integration Notes

Getting Started

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

The Problem Space

Solution A: Pangolinfo Scrape API (Direct Call)

Image response schema

Async bulk collection (Python)

Normalize image URLs to highest resolution

Error handling pattern

Solution B: MCP Protocol — No API Code Required

MCP tool definition (excerpt)

Claude Desktop config

Open Claw integration

Production Checklist

Links

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

TL;DR

Why Amazon Category Top 100 Data Matters

The Real Technical Challenges

Four Approaches Compared

Approach 1: DIY Python (requests + BeautifulSoup)

Approach 2: Playwright / Puppeteer

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

Approach 4: Pangolinfo Scrape API ✅

Complete Python Pipeline

Real Results: Before and After

Wrapping Up

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

TL;DR

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

Setup

Core Module: Single ASIN Query

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

Daily Monitoring Pipeline with Change Detection

Key Design Decisions

Resources

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

Why this exists

The framework

Picking the data source

Architecture

Code — async client

Code — metric computation

Code — the orchestration

Running it

Production hardening

Why this beats SaaS tools

Closing

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

Background

Why Self-Built Scrapers Break at Scale

Solution: Pangolinfo Scrape API

Full Implementation

Async Multi-Category Collector

Production Deployment Notes

References & Further Reading

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

TL;DR

The Problem With Standard Product Research Tools

What You'll Build

Setup

Layer 1: Real-Time Bestseller Scanner

Layer 2: Keyword Competition Analyzer

Layer 3: Negative Review Miner

Putting It All Together

Performance Tips

Scaling Beyond the Script

Resources

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

TL;DR

Why Subscription Tools Don't Work for Programmatic Use

Setup

1. Basic Sync Client