Best Proxies for Amazon Scraping:
Complete 2026 Guide

import asyncio
from playwright.async_api import async_playwright
import json
import random

# Proxies.sx mobile proxy configuration
PROXY_HOST = "gate.proxies.sx"
PROXY_PORT = 10000
PROXY_USER = "your_username"
PROXY_PASS = "your_password"

# Realistic User-Agent rotation
USER_AGENTS = [
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_4 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Mobile/15E148 Safari/604.1",
    "Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Mobile Safari/537.36",
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.3.1 Mobile/15E148 Safari/604.1",
]

async def scrape_amazon_product(page, asin: str) -> dict:
    """Scrape a single Amazon product page by ASIN."""
    url = f"https://www.amazon.com/dp/{asin}"

    # Add random delay to mimic human behavior
    await asyncio.sleep(random.uniform(2, 5))

    try:
        response = await page.goto(url, wait_until="domcontentloaded", timeout=30000)

        if response and response.status == 200:
            # Extract product data
            title = await page.locator("#productTitle").inner_text()
            title = title.strip() if title else None

            # Price extraction (Amazon uses multiple price selectors)
            price = None
            for selector in [
                ".a-price .a-offscreen",
                "#priceblock_ourprice",
                "#priceblock_dealprice",
                ".a-price-whole",
            ]:
                try:
                    el = page.locator(selector).first
                    if await el.is_visible(timeout=2000):
                        price = await el.inner_text()
                        break
                except:
                    continue

            # Rating
            rating = None
            try:
                rating_el = page.locator("#acrPopover .a-icon-alt").first
                if await rating_el.is_visible(timeout=2000):
                    rating = await rating_el.inner_text()
            except:
                pass

            # Review count
            review_count = None
            try:
                review_el = page.locator("#acrCustomerReviewText").first
                if await review_el.is_visible(timeout=2000):
                    review_count = await review_el.inner_text()
            except:
                pass

            # Availability
            availability = None
            try:
                avail_el = page.locator("#availability span").first
                if await avail_el.is_visible(timeout=2000):
                    availability = (await avail_el.inner_text()).strip()
            except:
                pass

            return {
                "asin": asin,
                "title": title,
                "price": price,
                "rating": rating,
                "review_count": review_count,
                "availability": availability,
                "url": url,
                "status": "success",
            }
        else:
            return {"asin": asin, "status": "blocked", "code": response.status if response else None}

    except Exception as e:
        return {"asin": asin, "status": "error", "error": str(e)}

async def scrape_products(asins: list[str]) -> list[dict]:
    """Scrape multiple Amazon products using mobile proxy."""
    results = []

    async with async_playwright() as p:
        browser = await p.chromium.launch(
            proxy={
                "server": f"http://{PROXY_HOST}:{PROXY_PORT}",
                "username": PROXY_USER,
                "password": PROXY_PASS,
            },
            headless=True,
        )

        context = await browser.new_context(
            user_agent=random.choice(USER_AGENTS),
            viewport={"width": 390, "height": 844},
            locale="en-US",
            timezone_id="America/New_York",
        )

        page = await context.new_page()

        for asin in asins:
            result = await scrape_amazon_product(page, asin)
            results.append(result)
            print(f"[{result['status']}] {asin}: {result.get('title', 'N/A')[:60]}")

        await browser.close()

    return results

# Usage
if __name__ == "__main__":
    target_asins = [
        "B0CHX3QBCH",  # Example ASIN 1
        "B0D5CLQNL2",  # Example ASIN 2
        "B0BSHF7WHW",  # Example ASIN 3
    ]
    data = asyncio.run(scrape_products(target_asins))
    with open("amazon_products.json", "w") as f:
        json.dump(data, f, indent=2)
    print(f"\nScraped {len(data)} products. Success: {sum(1 for d in data if d['status'] == 'success')}")

import asyncio
import schedule
import time
from datetime import datetime
from dataclasses import dataclass
import json
import smtplib
from email.mime.text import MIMEText

@dataclass
class PriceAlert:
    asin: str
    product_name: str
    old_price: float
    new_price: float
    change_pct: float
    timestamp: str

class AmazonPriceMonitor:
    def __init__(self, proxy_config: dict, alert_email: str):
        self.proxy = proxy_config
        self.alert_email = alert_email
        self.price_history: dict[str, list] = {}
        self.alerts: list[PriceAlert] = []

    def parse_price(self, price_str: str) -> float | None:
        """Extract numeric price from Amazon price string."""
        if not price_str:
            return None
        import re
        match = re.search(r"[\d,]+\.\d{2}", price_str.replace(",", ""))
        return float(match.group()) if match else None

    async def check_prices(self, asins: list[str]):
        """Check current prices and detect changes."""
        # Uses the scrape_products function from above
        results = await scrape_products(asins)

        for result in results:
            if result["status"] != "success" or not result.get("price"):
                continue

            asin = result["asin"]
            current_price = self.parse_price(result["price"])
            if current_price is None:
                continue

            # Initialize history if needed
            if asin not in self.price_history:
                self.price_history[asin] = []

            # Check for price change
            if self.price_history[asin]:
                last_price = self.price_history[asin][-1]["price"]
                if current_price != last_price:
                    change_pct = ((current_price - last_price) / last_price) * 100
                    alert = PriceAlert(
                        asin=asin,
                        product_name=result.get("title", "Unknown")[:100],
                        old_price=last_price,
                        new_price=current_price,
                        change_pct=round(change_pct, 2),
                        timestamp=datetime.now().isoformat(),
                    )
                    self.alerts.append(alert)
                    self.send_alert(alert)

            # Record price point
            self.price_history[asin].append({
                "price": current_price,
                "timestamp": datetime.now().isoformat(),
            })

    def send_alert(self, alert: PriceAlert):
        """Send price change notification."""
        direction = "dropped" if alert.change_pct < 0 else "increased"
        subject = f"Price {direction}: {alert.product_name[:50]}"
        body = f"""
        Product: {alert.product_name}
        ASIN: {alert.asin}
        Old Price: ${alert.old_price:.2f}
        New Price: ${alert.new_price:.2f}
        Change: {alert.change_pct:+.2f}%
        Time: {alert.timestamp}
        """
        print(f"ALERT: {subject}")
        # Add your email/Slack/webhook notification logic here

    def save_history(self, filepath: str = "price_history.json"):
        """Persist price history to disk."""
        with open(filepath, "w") as f:
            json.dump(self.price_history, f, indent=2)

# Setup and run
monitor = AmazonPriceMonitor(
    proxy_config={
        "host": "gate.proxies.sx",
        "port": 10000,
        "user": "your_username",
        "pass": "your_password",
    },
    alert_email="alerts@yourcompany.com",
)

TRACKED_ASINS = [
    "B0CHX3QBCH",
    "B0D5CLQNL2",
    "B0BSHF7WHW",
    # Add up to 10,000 ASINs
]

# Run price checks every 30 minutes
def run_check():
    asyncio.run(monitor.check_prices(TRACKED_ASINS))
    monitor.save_history()

schedule.every(30).minutes.do(run_check)

print("Amazon Price Monitor started. Checking every 30 minutes...")
while True:
    schedule.run_pending()
    time.sleep(1)