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

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

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

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

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

Why This Matters

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

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

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

The IBADM Framework

Before we write code, let's define the framework. Good competitor research follows five steps: Identify → Baseline → Analyze → Differentiate → Monitor

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

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

Setup

pip install requests pandas schedule

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

Step 0: API Client

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

class AmazonAPIClient:
    """Client for Pangolinfo Amazon Scraper API.
    Key specs:
    - Data latency: ~3 seconds
    - Daily capacity: 30M+ requests
    - Success rate: 99%
    - SP ad placement coverage: 98% (industry #1)
    """
    def __init__(self, api_key: str, domain: str = "com"):
        self.api_key = api_key
        self.domain = domain
        self.base_url = "https://api.pangolinfo.com/v1"
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })

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

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

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

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

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

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

Step 1: Identify Competitors

def identify_competitors(client: AmazonAPIClient, keywords: List[str]) -> List[str]:
    """Find all competitor ASINs from keyword search results.
    Pulls both organic results and SP ad placements.
    Returns a deduplicated list of ASINs."""
    competitors = set()
    for kw in keywords:
        result = client.search_keyword(kw)
        # Organic results
        for item in result.get("organic_results", []):
            competitors.add(item["asin"])
        # Sponsored placements (SP ads) - 98% coverage
        for item in result.get("sponsored_results", []):
            competitors.add(item["asin"])
        # Polite delay between keywords
        time.sleep(0.3)
    return list(competitors)

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

Step 2: Build Baseline

def build_baseline(client: AmazonAPIClient, asins: List[str]) -> pd.DataFrame:
    """Capture current state of all competitors simultaneously.
    Uses ThreadPoolExecutor for parallel data fetching.
    Total time for 20 ASINs: ~3-5 seconds."""
    def fetch_one(asin: str) -> dict:
        data = client.get_product(asin)
        return {
            "asin": asin,
            "title": data.get("title", ""),
            "price": data.get("price", 0),
            "bsr": data.get("bsr", 0),
            "rating": data.get("rating", 0),
            "review_count": data.get("review_count", 0),
            "variant_count": len(data.get("variants", [])),
            "image_count": len(data.get("images", [])),
            "has_a_plus": data.get("has_a_plus", False),
            "qa_count": data.get("qa_count", 0),
            "timestamp": datetime.now().isoformat()
        }
    # Parallel fetch - all competitors at once
    with ThreadPoolExecutor(max_workers=10) as executor:
        results = list(executor.map(fetch_one, asins))
    return pd.DataFrame(results)

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

Step 3: Analyze

3a: Keyword Gap Analysis

def keyword_gap_analysis(client: AmazonAPIClient, target_asin: str, competitor_asins: List[str], keywords: List[str]) -> pd.DataFrame:
    """Compare keyword rankings between your ASIN and competitors.
    Identifies keywords where competitors rank but you don't, and vice versa."""
    rows = []
    # Your rankings
    my_ranks = client.get_keyword_ranking(target_asin, keywords)
    for kw, rank in my_ranks.get("rankings", {}).items():
        rows.append({
            "keyword": kw,
            "your_rank": rank,
            "your_search_volume": my_ranks.get("volumes", {}).get(kw, 0)
        })
    df = pd.DataFrame(rows)
    # Competitor rankings
    for comp_asin in competitor_asins:
        comp_ranks = client.get_keyword_ranking(comp_asin, keywords)
        rank_map = comp_ranks.get("rankings", {})
        df[f"comp_{comp_asin}_rank"] = df["keyword"].map(rank_map)
    return df

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

3b: Review Pain Point Analysis

def analyze_review_pain_points(client: AmazonAPIClient, asin: str, top_n: int = 50) -> Dict[str, int]:
    """Extract frequent keywords from negative reviews.
    These pain points are your differentiation opportunities.
    If competitors' customers complain about 'breaks easily', that's your chance to emphasize durability."""
    from collections import Counter
    import re
    reviews = client.get_reviews(asin, review_type="critical")
    review_texts = [r["content"] for r in reviews.get("reviews", [])[:top_n]]
    # Simple word frequency (use NLP libraries for production)
    words = []
    for text in review_texts:
        words.extend(re.findall(r'[a-z]{4,}', text.lower()))
    # Filter common stop words
    stop_words = {"this", "that", "with", "have", "from", "they", "were", "been", "would", "could", "should", "really", "very"}
    words = [w for w in words if w not in stop_words]
    return dict(Counter(words).most_common(15))

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

Step 4: Differentiate

def find_differentiation_opportunities(client: AmazonAPIClient, my_asin: str, competitors: List[str], keywords: List[str]) -> dict:
    """Combine keyword gaps and review pain points into actionable insights."""
    # Keyword gaps: where competitors rank but you don't
    gap_df = keyword_gap_analysis(client, my_asin, competitors, keywords)
    missing_keywords = gap_df[gap_df["your_rank"].isna()]["keyword"].tolist()
    # Pain points from all competitors
    all_pain_points = {}
    for asin in competitors:
        all_pain_points[asin] = analyze_review_pain_points(client, asin)
    # Cross-reference: find pain points mentioned across multiple competitors
    from collections import Counter
    pain_word_counts = Counter()
    for pp in all_pain_points.values():
        for word in pp:
            pain_word_counts[word] += 1
    shared_pain_points = {word: count for word, count in pain_word_counts.items() if count >= 2}  # Mentioned by 2+ competitors' customers
    return {
        "keyword_opportunities": missing_keywords,
        "shared_pain_points": shared_pain_points,
        "recommendation": (
            "Target keywords where competitors rank but you don't. "
            "Address shared pain points in your listing to differentiate."
        )
    }

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

Step 5: Monitor

def start_monitoring(client: AmazonAPIClient, asins: List[str], output_file: str = "monitor_log.csv", alert_threshold_pct: float = 10.0):
    """Set up daily monitoring with price change alerts.
    Runs daily, logs all data, and prints alerts when price changes exceed threshold."""
    import schedule
    def job():
        # Fetch current state
        current = build_baseline(client, asins)
        current["check_time"] = datetime.now().isoformat()
        # Append to log
        try:
            existing = pd.read_csv(output_file)
            pd.concat([existing, current]).to_csv(output_file, index=False)
        except FileNotFoundError:
            current.to_csv(output_file, index=False)
        # Check for price changes (compare with last entry)
        try:
            history = pd.read_csv(output_file)
            last_entry = history[history["check_time"] != current["check_time"].iloc[0]]
            if len(last_entry) > 0:
                last_entry = last_entry.tail(len(asins))
                for _, curr_row in current.iterrows():
                    prev = last_entry[last_entry["asin"] == curr_row["asin"]]
                    if len(prev) > 0:
                        prev_price = prev.iloc[0]["price"]
                        if prev_price > 0:
                            change_pct = ((curr_row["price"] - prev_price) / prev_price) * 100
                            if abs(change_pct) > alert_threshold_pct:
                                direction = "dropped" if change_pct < 0 else "increased"
                                print(f"🚨 ALERT: ASIN {curr_row['asin']} price {direction} {abs(change_pct):.1f}% (${prev_price} → ${curr_row['price']})")
        except Exception as e:
            print(f"Alert check error: {e}")
        print(f"[{datetime.now()}] Monitor cycle complete - {len(asins)} ASINs checked")
    # Run daily at 9 AM
    schedule.every().day.at("09:00").do(job)
    print(f"Monitoring started. Daily at 09:00. Output: {output_file}")
    print(f"Alert threshold: ±{alert_threshold_pct}% price change")
    print("Press Ctrl+C to stop.\n")
    # Initial run
    job()
    while True:
        schedule.run_pending()
        time.sleep(60)

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

No-Code Alternative: Amazon Data MCP

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

What is MCP?

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

Setup

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

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

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

Usage

Just type natural language:

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

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

The 19 tools cover:

Performance Comparison

Here's the real-world difference between approaches:

Production Tips

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

Tip 2: