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

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.

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

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.

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

Identify → Baseline → Analyze → Differentiate → Monitor

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

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

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})
php
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)

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

        for item in result.get("sponsored_results", []):
            competitors.add(item["asin"])

        time.sleep(0.3)

    return list(competitors)

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")
php
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()
        }

    with ThreadPoolExecutor(max_workers=10) as executor:
        results = list(executor.map(fetch_one, asins))

    return pd.DataFrame(results)

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))
python
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 = []

    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)

    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

gap_df = keyword_gap_analysis(client, "B0YOURASIN", competitors[:5], keywords)
print(gap_df)
python
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]]

    words = []
    for text in review_texts:
        words.extend(re.findall(r'[a-z]{4,}', text.lower()))

    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))

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}")
python
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."""

    gap_df = keyword_gap_analysis(client, my_asin, competitors, keywords)
    missing_keywords = gap_df[gap_df["your_rank"].isna()]["keyword"].tolist()

    all_pain_points = {}
    for asin in competitors:
        all_pain_points[asin] = analyze_review_pain_points(client, asin)

    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."
        )
    }

opportunities = find_differentiation_opportunities(
    client, "B0YOURASIN", competitors[:5], keywords
)
print(json.dumps(opportunities, indent=2))
python
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():
        current = build_baseline(client, asins)
        current["check_time"] = datetime.now().isoformat()

        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)

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

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

    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")

    job()

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

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

If you don't want to write code, or want your operations team to run analysis independently, use the Amazon Data 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.

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.

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:

Here's the real-world difference between approaches:

Tip 1: Store historical data

The monitor_log.csv

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

Tip 2: Focus on negative reviews

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

.

Tip 3: Set meaningful alert thresholds

A 5% price change might be noise. A 15% change is a strategy shift. Tune your alert_threshold_pct

based on your category's typical price volatility.

Tip 4: Use MCP for exploration, API for automation

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

Here's the full runnable script combining everything:

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

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


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

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

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

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

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

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

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

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

Resources:

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

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