Śledzenie rankingu słów kluczowych Amazon z proxy — przewodnik dla programistów

Nota prawna: Ten przewodnik dotyczy wyłącznie danych publicznie dostępnych na stronach wyników wyszukiwania Amazon. W Stanach Zjednoczonych Computer Fraud and Abuse Act (CFAA) reguluje nieautoryzowany dostęp do systemów komputerowych — trzymaj się publicznych stron i przestrzegaj robots.txt. W UE RODO (GDPR) ogranicza przetwarzanie danych osobowych — nie zbieraj danych użytkowników, tylko publiczne informacje o produktach. Jeśli jesteś sprzedawcą na Amazon, rozważ oficjalne Amazon SP-API jako pierwszą opcję.

Śledzenie rankingu słów kluczowych Amazon z proxy to kluczowa umiejętność dla sprzedawców i deweloperów marketplace SEO. W przeciwieństwie do Google, Amazon ma własny silnik wyszukiwania, w którym pozycja organiczna zależy od relewancji, sales velocity i historii konwersji. Aby skutecznie track amazon rankings python, potrzebujesz residential proxy z geo-targetingiem krajowym, stabilnych sesji sticky do paginacji i solidnego parsera HTML. Ten przewodnik przeprowadzi Cię przez pełną implementację — od pierwszego żądania HTTP po produkcyjny system z historią pozycji w SQLite.

Śledzenie rankingu słów kluczowych Amazon z proxy — dlaczego to inna gra niż Google SERP

Amazon nie jest Google. Jego algorytm rankingowy, znany jako A9, priorytetyzuje sales velocity — produkty, które sprzedają się częściej, wyżej rankują. To oznacza, że ten sam ASIN może mieć zupełnie inną pozycję dla różnych słów kluczowych, w różnych marketplace'ach (amazon.com, amazon.de, amazon.co.uk) i w różnych godzinach. Sprzedawcy muszą śledzić pozycję organiczną per keyword, per marketplace i per ASIN.

Amazon amazon serp scraping jest bardziej wymagający niż Google, ponieważ:

• Amazon serwuje różne wyniki w zależności od lokalizacji IP — niemieckie IP zobaczy inne wyniki na amazon.com niż amerykańskie.
• Strona wyników zawiera zarówno wyniki organiczne, jak i Sponsored placements (reklamy) — trzeba je rozróżniać.
• Amazon agresywnie blokuje boty — datacenter IP są często odrzucane już po kilku żądaniach.
• Paginacja wymaga stabilnej sesji — zmiana IP w trakcie przeglądania stron 1–5 powoduje niespójności w wynikach.

Dlatego amazon keyword rank tracker bez proxy residential jest w praktyce bezużyteczny po kilku godzinach — Twój IP zostanie zablokowany, a dane będą niekompletne.

Parsowanie strony wyników Amazon — struktura HTML i ekstrakcja ASIN

Strona wyników wyszukiwania Amazon (URL typu /s?k=KEYWORD&page=N) renderuje listę produktów w divach z atrybutem data-component-type="s-search-result". Każdy taki div ma atrybut data-asin z identyfikatorem produktu. Wyniki sponsorowane mają etykietę „Sponsored" widoczną w drzewie DOM.

Oto kluczowe kroki parsowania:

1. Znajdź wszystkie elementy [data-component-type="s-search-result"].
2. Odczytaj data-asin z każdego elementu.
3. Wykryj wyniki sponsorowane — szukaj tekstu „Sponsored" lub elementów z klasą s-label-popover-default.
4. Oblicz pozycję organiczną — inkrementuj licznik tylko dla wyników niesponsorowanych.
5. Zapisz parę (ASIN, pozycja_organiczna, strona, czy_sponsorowane).

Przykładowy parser w Python z BeautifulSoup:

from bs4 import BeautifulSoup
from typing import Optional

def parse_amazon_results(html: str) -> list[dict]:
    """Parse Amazon search results page and extract ASIN positions."""
    soup = BeautifulSoup(html, "html.parser")
    results = []
    organic_position = 0

    for div in soup.select('[data-component-type="s-search-result"]'):
        asin = div.get("data-asin", "")
        if not asin:
            continue

        # Detect sponsored listings
        sponsored_el = div.select_one(
            '.s-label-popover-default, '
            '[aria-label*="Sponsored"], '
            'a-color-secondary:contains("Sponsored")'
        )
        is_sponsored = sponsored_el is not None

        if not is_sponsored:
            organic_position += 1

        results.append({
            "asin": asin,
            "organic_position": organic_position if not is_sponsored else None,
            "sponsored": is_sponsored,
            "page_position": len(results) + 1,
        })

    return results

def find_asin_position(
    results: list[dict], target_asin: str
) -> Optional[dict]:
    """Find the organic position of a target ASIN."""
    for r in results:
        if r["asin"] == target_asin and not r["sponsored"]:
            return r
    return None

Pamiętaj, że Amazon regularnie zmienia strukturę HTML. Selektory CSS mogą wymagać aktualizacji — to jeden z powodów, dla których web scraping wymaga ciągłego monitorowania.

Dlaczego residential proxy z geo-targetingiem są niezbędne

Amazon serwuje różne wyniki w zależności od kraju, z którego pochodzi żądanie. Jeśli śledzisz pozycje na amazon.de z amerykańskiego datacenter IP, otrzymasz zniekształcone wyniki. Residential proxy z geo-targetingiem krajowym rozwiązują ten problem — każde żądanie wychodzi z IP przypisanego do realnego dostawcy internetowego w docelowym kraju.

Dodatkowo Amazon stosuje agresywne anti-bot:

• Rate limiting per IP — typowo 100–200 żądań zanim pojawi się CAPTCHA.
• Fingerprinting TLS/SSL — datacenter IP mają charakterystyczne sygnatury TLS.
• Per-marketplace localization — amazon.com z niemieckiego IP zwraca inne wyniki niż z amerykańskiego.

Poniższa tabela porównuje typy proxy dla amazon serp scraping:

Dla śledzenia rankingu Amazon, residential proxy to optymalny wybór — wystarczająco niezawodne i tańsze niż mobile. Sprawdź dostępne lokalizacje ProxyHat, aby upewnić się, że docelowy marketplace jest obsługiwany.

Implementacja w Python z curl_cffi i ProxyHat

curl_cffi to biblioteka Python, która imituje TLS fingerprint przeglądarki — kluczowe dla omijania anti-bot Amazon. Połącz ją z ProxyHat residential proxy używając geo-targetingu w nazwie użytkownika.

Podstawowe żądanie z curl

Najpierw prosty test z curl:

curl -x http://user-country-US:pass@gate.proxyhat.com:8080 \
  "https://www.amazon.com/s?k=wireless+earbuds&page=1" \
  -H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" \
  -H "Accept-Language: en-US,en;q=0.9" \
  -s -o amazon_page1.html

echo "Pobrano: $(wc -c < amazon_page1.html) bajtów"

Pełny pipeline w Python z curl_cffi

from curl_cffi import requests
import time
import logging
from typing import Optional

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

PROXY_BASE = "http://gate.proxyhat.com:8080"
PROXY_USER = "user"
PROXY_PASS = "pass"

def build_proxy_url(country: str, session_id: Optional[str] = None) -> str:
    """Build ProxyHat URL with geo-targeting and optional sticky session."""
    username = f"{PROXY_USER}-country-{country}"
    if session_id:
        username += f"-session-{session_id}"
    return f"http://{username}:{PROXY_PASS}@gate.proxyhat.com:8080"

def fetch_amazon_search(
    keyword: str,
    page: int = 1,
    country: str = "US",
    session_id: Optional[str] = None,
) -> Optional[str]:
    """Fetch Amazon search results page HTML."""
    proxy_url = build_proxy_url(country, session_id)
    proxies = {"http": proxy_url, "https": proxy_url}

    url = f"https://www.amazon.{'com' if country == 'US' else 'de'}/s?k={keyword}&page={page}"
    headers = {
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/120.0.0.0 Safari/537.36"
        ),
        "Accept-Language": "en-US,en;q=0.9" if country == "US" else "de-DE,de;q=0.9",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    }

    try:
        response = requests.get(
            url,
            headers=headers,
            proxies=proxies,
            impersonate="chrome",
            timeout=30,
        )
        response.raise_for_status()
        html = response.text

        # CAPTCHA detection
        if "Type the characters you see in this image" in html:
            logger.warning(f"CAPTCHA detected on page {page} for keyword '{keyword}'")
            return None

        logger.info(f"Fetched page {page}: {len(html)} bytes for keyword '{keyword}'")
        return html
    except requests.RequestException as e:
        logger.error(f"Request failed for page {page}: {e}")
        return None

# Fetch pages 1-5 for a keyword and find target ASIN
keyword = "wireless earbuds"
target_asin = "B0DXYZ1234"
marketplace = "amazon.com"
country = "US"
session_id = f"track-{int(time.time())}"

all_positions = []

for page in range(1, 6):
    html = fetch_amazon_search(keyword, page=page, country=country, session_id=session_id)
    if html:
        results = parse_amazon_results(html)
        found = find_asin_position(results, target_asin)
        if found:
            all_positions.append({
                "page": page,
                "organic_position": found["organic_position"],
                "absolute_position": (page - 1) * 60 + found["organic_position"],
            })
            logger.info(
                f"Found ASIN {target_asin} on page {page}, "
                f"organic position {found['organic_position']}"
            )
            break
    time.sleep(2)  # Be polite — 2s between requests

if not all_positions:
    logger.info(f"ASIN {target_asin} not found in pages 1-5 for keyword '{keyword}'")

Kluczowe elementy tego kodu:

• Geo-targeting: -country-US w nazwie użytkownika wymusza amerykańskie IP. Dla amazon.de użyj -country-DE.
• Sticky session: -session-{id} utrzymuje to samo IP przez całą sesję paginacji — kluczowe dla spójności wyników.
• TLS impersonation: impersonate="chrome" w curl_cffi imituje fingerprint TLS Chrome.
• CAPTCHA detection: proste sprawdzenie tekstu w HTML.
• Throttle: 2 sekundy między żądaniami.

Alternatywa: Playwright z ProxyHat dla stron renderowanych JavaScriptem

Niektóre elementy Amazon mogą wymagać JavaScript. Playwright z proxy residential obsłuży te przypadki:

import asyncio
from playwright.async_api import async_playwright
from typing import Optional
import logging

logger = logging.getLogger(__name__)

async def fetch_with_playwright(
    keyword: str,
    page_num: int,
    country: str = "US",
    session_id: Optional[str] = None,
) -> Optional[str]:
    """Fetch Amazon search page using Playwright with ProxyHat."""
    username = f"user-country-{country}"
    if session_id:
        username += f"-session-{session_id}"
    proxy_server = f"http://{username}:pass@gate.proxyhat.com:8080"

    async with async_playwright() as p:
        browser = await p.chromium.launch(
            proxy={"server": proxy_server},
            headless=True,
            args=["--disable-blink-features=AutomationControlled"],
        )
        context = await browser.new_context(
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/120.0.0.0 Safari/537.36"
            ),
            locale="en-US" if country == "US" else "de-DE",
            viewport={"width": 1920, "height": 1080},
        )
        page = await context.new_page()

        domain = "amazon.com" if country == "US" else "amazon.de"
        url = f"https://www.{domain}/s?k={keyword}&page={page_num}"

        try:
            await page.goto(url, wait_until="domcontentloaded", timeout=30000)
            await page.wait_for_timeout(2000)  # Allow JS to render
            content = await page.content()

            if "Type the characters you see in this image" in content:
                logger.warning("CAPTCHA detected in Playwright session")
                return None

            return content
        except Exception as e:
            logger.error(f"Playwright error on page {page_num}: {e}")
            return None
        finally:
            await context.close()
            await browser.close()

# Usage
async def main():
    html = await fetch_with_playwright(
        "wireless earbuds", 1, country="US", session_id="pw-001"
    )
    if html:
        results = parse_amazon_results(html)
        print(f"Found {len(results)} results")

asyncio.run(main())

Przechowywanie historii pozycji w SQLite

Śledzenie rankingu ma sens tylko z historią. Prosta baza SQLite wystarczy na start:

import sqlite3
from datetime import datetime, timezone
from typing import Optional

def init_db(db_path: str = "rank_history.db") -> sqlite3.Connection:
    """Initialize SQLite database for rank tracking history."""
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS rank_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            keyword TEXT NOT NULL,
            asin TEXT NOT NULL,
            marketplace TEXT NOT NULL,
            organic_position INTEGER,
            page INTEGER,
            sponsored INTEGER DEFAULT 0,
            fetched_at TEXT NOT NULL
        )
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_keyword_asin_market
        ON rank_history(keyword, asin, marketplace, fetched_at)
    """)
    conn.commit()
    return conn

def save_position(
    conn: sqlite3.Connection,
    keyword: str,
    asin: str,
    marketplace: str,
    organic_pos: Optional[int],
    page: int,
):
    """Save a rank tracking data point."""
    conn.execute(
        """INSERT INTO rank_history
           (keyword, asin, marketplace, organic_position, page, sponsored, fetched_at)
           VALUES (?, ?, ?, ?, ?, ?, ?)""",
        (keyword, asin, marketplace, organic_pos, page, 0,
         datetime.now(timezone.utc).isoformat())
    )
    conn.commit()

def get_position_trend(
    conn: sqlite3.Connection,
    keyword: str,
    asin: str,
    marketplace: str,
    days: int = 30,
) -> list[dict]:
    """Get position history for a keyword/ASIN/marketplace combo."""
    rows = conn.execute("""
        SELECT organic_position, fetched_at
        FROM rank_history
        WHERE keyword = ? AND asin = ? AND marketplace = ?
          AND organic_position IS NOT NULL
          AND fetched_at >= datetime('now', ?)
        ORDER BY fetched_at ASC
    """, (keyword, asin, marketplace, f"-{days} days")).fetchall()

    return [
        {"position": r[0], "fetched_at": r[1]}
        for r in rows
    ]

# Full pipeline: fetch, parse, save
conn = init_db()

for page in range(1, 6):
    html = fetch_amazon_search(
        keyword, page=page, country="US", session_id=session_id
    )
    if html:
        results = parse_amazon_results(html)
        found = find_asin_position(results, target_asin)
        if found:
            save_position(conn, keyword, target_asin, "amazon.com",
                          found["organic_position"], page)
            logger.info(f"Saved: position {found['organic_position']} on page {page}")
            break
    time.sleep(2)

# Check trend
trend = get_position_trend(conn, keyword, target_asin, "amazon.com", days=7)
for t in trend:
    print(f"{t['fetched_at']}: pozycja {t['position']}")

Wskazówki produkcyjne — harmonogram, retry, CAPTCHA i indexation checks

Retry z exponential backoff i circuit breaker

import time
import logging
from functools import wraps
from typing import Callable, Optional, Any

logger = logging.getLogger(__name__)

class CircuitBreaker:
    """Simple circuit breaker for Amazon requests."""
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 300):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure_time: Optional[float] = None
        self.is_open = False

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.is_open = True
            logger.warning(f"Circuit breaker OPEN after {self.failure_count} failures")

    def record_success(self):
        self.failure_count = 0
        self.is_open = False

    def can_execute(self) -> bool:
        if not self.is_open:
            return True
        if self.last_failure_time and \
           time.time() - self.last_failure_time > self.reset_timeout:
            self.is_open = False
            self.failure_count = 0
            logger.info("Circuit breaker reset — retrying")
            return True
        return False

cb = CircuitBreaker(failure_threshold=5, reset_timeout=300)

def retry_with_backoff(
    max_retries: int = 3,
    base_delay: float = 2.0,
    backoff_factor: float = 2.0,
):
    """Decorator with exponential backoff and circuit breaker."""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Optional[Any]:
            if not cb.can_execute():
                logger.error("Circuit breaker is OPEN — skipping request")
                return None

            delay = base_delay
            for attempt in range(max_retries + 1):
                try:
                    result = func(*args, **kwargs)
                    if result and "Type the characters" in result:
                        raise RuntimeError("CAPTCHA challenge encountered")
                    if result:
                        cb.record_success()
                        return result
                except Exception as e:
                    if attempt == max_retries:
                        logger.error(f"Max retries ({max_retries}) reached: {e}")
                        cb.record_failure()
                        return None
                    wait = delay + (attempt * 0.5)  # Jitter
                    logger.warning(
                        f"Attempt {attempt + 1}/{max_retries} failed: {e}. "
                        f"Retrying in {wait:.1f}s..."
                    )
                    time.sleep(wait)
                    delay *= backoff_factor
            return None
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3, base_delay=2.0, backoff_factor=2.0)
def fetch_amazon_search_safe(keyword: str, page: int = 1,
                              country: str = "US",
                              session_id: Optional[str] = None) -> Optional[str]:
    """Wrapper with retry/backoff for Amazon search fetch."""
    return fetch_amazon_search(keyword, page, country, session_id)

Codzienne harmonogramy z APScheduler

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import logging

logger = logging.getLogger(__name__)

KEYWORDS_TO_TRACK = [
    {"keyword": "wireless earbuds", "asin": "B0DXYZ1234", "country": "US"},
    {"keyword": "bluetooth speaker", "asin": "B0ABC45678", "country": "US"},
    {"keyword": "kabel usb c", "asin": "B0DEF90123", "country": "DE"},
]

def track_all_keywords():
    """Daily rank tracking job for all configured keywords."""
    conn = init_db()
    for item in KEYWORDS_TO_TRACK:
        session_id = f"daily-{int(time.time())}-{item['asin']}"
        domain = "amazon.com" if item["country"] == "US" else "amazon.de"

        for page in range(1, 6):
            html = fetch_amazon_search_safe(
                item["keyword"], page=page,
                country=item["country"],
                session_id=session_id,
            )
            if html:
                results = parse_amazon_results(html)
                found = find_asin_position(results, item["asin"])
                if found:
                    save_position(
                        conn, item["keyword"], item["asin"],
                        domain, found["organic_position"], page,
                    )
                    logger.info(
                        f"{item['keyword']} / {item['asin']}: "
                        f"pozycja {found['organic_position']} (strona {page})"
                    )
                    break
            time.sleep(3)  # 3s between pages — be polite

scheduler = BlockingScheduler()
scheduler.add_job(
    track_all_keywords,
    CronTrigger(hour=6, minute=0),  # Daily at 6:00 AM UTC
    id="daily_rank_tracking",
)

if __name__ == "__main__":
    logger.info("Starting Amazon rank tracker scheduler...")
    scheduler.start()

Indexation check — czy ASIN w ogóle rankuje?

Zanim zaczniesz śledzić pozycję, sprawdź czy ASIN jest w ogóle indeksowany dla danego słowa kluczowego. Jeśli nie znajdziesz go na stronach 1–20, prawdopodobnie nie jest indeksowany lub ma zbyt niską relewancję. To osobny sygnał dla sprzedawcy — może być potrzebna optymalizacja listing (tytuł, bullet points, backend keywords).

def check_indexation(
    keyword: str, asin: str, country: str = "US", max_pages: int = 20
) -> dict:
    """Check if an ASIN is indexed for a keyword (appears in search results)."""
    session_id = f"index-{int(time.time())}"

    for page in range(1, max_pages + 1):
        html = fetch_amazon_search_safe(
            keyword, page=page, country=country, session_id=session_id
        )
        if not html:
            return {"indexed": False, "reason": "fetch_failed", "last_page": page}

        results = parse_amazon_results(html)
        if not results:
            return {"indexed": False, "reason": "no_results", "last_page": page}

        found = find_asin_position(results, asin)
        if found:
            return {
                "indexed": True,
                "page": page,
                "organic_position": found["organic_position"],
            }
        time.sleep(2)

    return {"indexed": False, "reason": "not_found_in_pages", "last_page": max_pages}

# Usage
result = check_indexation("wireless earbuds", "B0DXYZ1234", country="US")
print(result)
# {'indexed': True, 'page': 3, 'organic_position': 42}

Etyka i zgodność z przepisami

Śledzenie rankingu Amazon to operacja na publicznie dostępnych danych, ale wymaga odpowiedzialnego podejścia:

• Throttle: utrzymuj co najmniej 2–3 sekundy między żądaniami. Nie wysyłaj setek żądań na minutę.
• Track własne i publiczne listingi: nie zbieraj danych osobowych użytkowników ani recenzji z identyfikatorami klientów.
• Przestrzegaj robots.txt: sprawdź https://www.amazon.com/robots.txt przed rozpoczęciem.
• Rozważ Amazon SP-API: jeśli jesteś zarejestrowanym sprzedawcą, oficjalne SP-API oferuje endpointy do raportów katalogowych i analitycznych bez ryzyka blokady.
• GDPR/CCPA: jeśli śledzisz listingi na marketplace'ach europejskich, upewnij się że nie przechowujesz danych osobowych sprzedawców (nazwisk, adresów).

Sprawdź cennik ProxyHat, aby dobrać plan proxy odpowiedni do Twojego wolumenu śledzenia. Więcej o zastosowaniach proxy w analityce SERP znajdziesz na stronie SERP tracking. Pełną dokumentację techniczną znajdziesz na docs.proxyhat.com.

Kluczowe wnioski

• Amazon SERP ≠ Google SERP — ranking zależy od sales velocity i relewancji, nie tylko linków. Śledź per keyword, per marketplace, per ASIN.
• Residential proxy z geo-targetingiem są obowiązkowe — datacenter IP są blokowane po ~100–200 żądaniach. Użyj -country-US dla amazon.com, -country-DE dla amazon.de.
• Sticky sessions (-session-{id}) zapewniają spójną paginację — zmiana IP w trakcie przeglądania stron daje niespójne wyniki.
• Rozróżniaj organic vs Sponsored — etykieta „Sponsored" w DOM pozwala oddzielić płatne od organicznych pozycji.
• Retries z backoff i circuit breaker chronią przed kaskadowymi błędami, gdy Amazon tymczasowo blokuje Twój IP.
• Indexation check jako pierwszy krok — jeśli ASIN nie rankuje na stronach 1–20, problem jest w listing, nie w tracking.
• Etyka: throttle, przestrzegaj robots.txt, rozważ SP-API jeśli jesteś zarejestrowanym sprzedawcą.