TL;DR Stop reading your ACoS dashboard to understand competitor behavior. Build a real-time SP ad position monitor instead. This post covers: Async batch SERP capture via Pangolinfo API (98% SP coverage) Tiered keyword management (A/B/C) for signal vs. noise control Change detection: Top1 change, new Top3 entrant, Top3 exit, price drop LLM-enriched alerts via Open Claw + Claude Deduplication and Slack delivery Prerequisites: Open Claw deployed, Pangolinfo API key configured (see earlier posts in this series). We're jumping straight into the ad monitoring implementation. # Quick test: is your SERP API working? import requests headers = "Authorization": "Bearer YOUR_KEY" payload = "source": "amazon_search", "query": "wireless earbuds", "marketplace": "US", "include_sponsored": True, "include_organic": False, "output_format": "json" resp = requests.post("https://api.pangolinfo.com/v1/serp", headers=headers, json=payload) print(resp.json()["sponsored_results"][:3]) The Problem with Existing Monitoring Tools Your Amazon ad console shows internal metrics. Helium 10 / Jungle Scout show competitor data, but with 24-48h delays and no outbound API. When a new well-funded competitor enters your Top of Search overnight, you find out two days later when your ACoS report arrives. By then they've run 48 hours of aggressive ads at a discount price, accumulating early reviews. What we want: automated detection of competitor ad position changes within 2 hours of them occurring, categorized by severity, routed to Slack with business context. Step 1: Async SERP Capture 50 keywords × sequential requests = ~8 minutes per cycle. With asyncio: ~45 seconds. import asyncio import aiohttp from datetime import datetime, timezone from typing import List, Dict, Optional PANGOLINFO_API_KEY = "your_key" async def fetch_serp( keyword: str, marketplace: str, session: aiohttp.ClientSession, semaphore: asyncio.Semaphore ) -> Dict: """Capture Amazon SP ad positions for one keyword""" async with semaphore: headers = "Authorization": f"Bearer PANGOLINFO_API_KEY" payload = "source": "amazon_search", "query": keyword, "marketplace": marketplace, "page": 1, "include_sponsored": True, "include_organic": False, "output_format": "json" captured_at = datetime.now(timezone.utc).isoformat() try: async with session.post( "https://api.pangolinfo.com/v1/serp", headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=25) ) as resp: resp.raise_for_status() data = await resp.json() top_ads = sorted( [s for s in data.get("sponsored_results", []) if "top" in s.get("ad_placement", "").lower()], key=lambda x: x.get("ad_rank", 999) )[:5] return "keyword": keyword, "marketplace": marketplace, "captured_at": captured_at, "success": True, "top_of_search": [ "rank": a.get("ad_rank"), "asin": a.get("asin"), "brand": a.get("brand", ""), "price": a.get("price") for a in top_ads ] except Exception as e: return "keyword": keyword, "marketplace": marketplace, "captured_at": captured_at, "success": False, "error": str(e), "top_of_search": [] async def batch_monitor( keywords: List[str], marketplace: str = "US", max_concurrent: int = 8 ) -> List[Dict]: sem = asyncio.Semaphore(max_concurrent) async with aiohttp.ClientSession() as session: tasks = [fetch_serp(kw, marketplace, session, sem) for kw in keywords] return await asyncio.gather(*tasks) Step 2: Keyword Tier Configuration Don't monitor all keywords at the same frequency and alert threshold. It generates noise that gets ignored. KEYWORD_CONFIG = { "tiers": "A": # Core product keywords — monitor every 2h "keywords": ["wireless earbuds", "bluetooth speaker", "usb c hub"], "frequency_hours": 2, "alert_escalation": True # Upgrade HIGH → CRITICAL , "B": # Competitive category keywords — monitor every 6h "keywords": ["earbuds under 30 dollars", "tws earbuds 2026"], "frequency_hours": 6, "alert_escalation": False , "C": # Exploratory long-tail — monitor daily, log only "keywords": [], "frequency_hours": 24, "alert_escalation": False, "suppress_alerts": True # Record but don't notify , "price_drop_threshold_pct":