프록시를 활용한 아마존 키워드 순위 추적: 개발자 가이드

법적 고지: 본 가이드는 공개적으로 접근 가능한 검색 결과 데이터만 다룹니다. 미국에서는 CFAA(컴퓨터 사기 및 남용법)가 무단 접근을 제한하며, EU에서는 GDPR이 개인정보 처리를 규율합니다. 타인의 비공개 데이터를 수집하거나 서비스 약관을 위반하지 마세요. 가능하면 Amazon SP-API를 우선 사용하세요.

프록시를 활용한 아마존 키워드 순위 추적이 필요한 이유

아마존 검색은 Google이나 Bing과 완전히 다른 SERP 구조를 가집니다. 아마존의 랭킹 알고리즘은 관련성(relevance)과 판매 속도(sales velocity)를 핵심 신호로 사용하며, 이는 셀러가 키워드·마켓플레이스·ASIN별로 유기적 순위를 지속적으로 추적해야 함을 의미합니다. Google에서 1위인 키워드가 아마존에서는 3페이지에 있을 수 있습니다.

아마존 키워드 순위 추적기(amazon keyword rank tracker)를 직접 구축하려는 개발자와 셀러는 다음 문제에 직면합니다:

• 아마존은 마켓플레이스별로 현지화된 결과를 반환하므로 amazon.com과 amazon.de의 순위가 다릅니다.
• 동일 IP에서 반복 요청을 보내면 CAPTCHA가 빠르게 등장합니다.
• Sponsored 광고와 유기적 결과가 혼재하여 파싱 로직이 정교해야 합니다.
• 순위 변화 추이를 저장하지 않으면 최적화 효과를 측정할 수 없습니다.

이 가이드에서는 residential 프록시를 활용한 아마존 키워드 순위 추적 전체 파이프라인을 코드 중심으로 설명합니다. curl, curl_cffi, Playwright 예제와 ProxyHat 게이트웨이 설정을 단계별로 다룹니다.

아마존 SERP의 구조 이해하기

검색 결과 페이지의 DOM 구조

아마존 검색 결과 페이지의 각 상품 카드는 data-component-type="s-search-result" 속성을 가진 <div>로 렌더링됩니다. 각 카드에서 다음 정보를 추출할 수 있습니다:

• data-asin — 상품의 ASIN 식별자
• 제목 텍스트 — h2 > a 내부
• 가격 — .a-price 클래스
• 평점 — .a-icon-alt 텍스트 (예: "4.5 out of 5 stars")
• Sponsored 라벨 — .puis-label-popover-default 또는 "Sponsored" 텍스트 포함 요소

유기적 순위(organic position)는 Sponsored 결과를 제외한 카드의 순번입니다. Sponsored 결과는 페이지 상단이나 중간에 삽입되므로, 단순히 DOM 순서를 사용하면 유기적 순위가 왜곡됩니다.

순위 계산 로직

올바른 순위 계산을 위한 의사코드:

organic_position = 0
for card in search_results:
    if card has "Sponsored" label:
        continue  # 유기적 순위에서 제외
    organic_position += 1
    if card.data_asin == target_asin:
        return organic_position
return None  # 해당 페이지에 없음

프록시가 필수인 이유: 안티봇과 현지화

아마존은 IP 기반 rate limiting과 CAPTCHA를 적극 사용합니다. 단일 IP에서 짧은 시간에 여러 키워드를 검색하면, 아마존은 CAPTCHA 페이지로 리다이렉트하거나 일시적으로 IP를 차단합니다. 데이터센터 IP 대역은 특히 빠르게 차단됩니다.

또한 아마존은 요청 IP의 지리적 위치를 기반으로 결과를 현지화합니다. amazon.com에서 미국 외 IP로 접속하면 배송 가능 상품이 달라지고 순위가 왜곡됩니다. 정확한 순위를 얻으려면 대상 마켓플레이스와 일치하는 국가의 residential IP를 사용해야 합니다.

프록시 타입 비교

아마존 키워드 순위 추적에는 residential 프록시가 가장 균형 잡힌 선택입니다. 충분한 신뢰도를 제공하면서도 비용이 합리적입니다. ProxyHat의 글로벌 프록시 위치를 활용하면 190개 이상 국가에서 residential IP를 사용할 수 있습니다.

ProxyHat 게이트웨이 설정

ProxyHat은 사용자명에 국가 및 세션 플래그를 포함하는 방식으로 geo-targeting과 sticky session을 지원합니다. 아마존 순위 추적에 필요한 핵심 설정은 다음과 같습니다:

마켓플레이스별 추천 국가 코드:

• amazon.com → country-US
• amazon.de → country-DE
• amazon.co.uk → country-GB
• amazon.co.jp → country-JP

페이지네이션을 사용할 때는 동일한 IP를 유지해야 결과가 일관됩니다. -session- 플래그로 sticky session을 만들면 여러 페이지 요청이 동일한 residential IP에서 나갑니다.

코드 예제 1: curl로 단일 페이지 가져오기

# amazon.com에서 "wireless earbuds" 검색 결과 1페이지 가져오기
# 미국 residential IP + sticky session 사용

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

코드 예제 2: curl_cffi로 검색 결과 가져오기

curl_cffi는 브라우저 TLS 지문을 모방하여 Cloudflare 등 안티봇 시스템을 우회하는 Python HTTP 클라이언트입니다. 아마존 스크래핑에 효과적입니다.

import asyncio
from curl_cffi.requests import AsyncSession
from typing import Optional

PROXYHAT_GATEWAY = "gate.proxyhat.com"
PROXYHAT_PORT = 8080
PROXYHAT_USER = "YOUR_USERNAME"
PROXYHAT_PASS = "YOUR_PASSWORD"

def build_proxy_url(country: str, session_id: Optional[str] = None) -> str:
    """ProxyHat 프록시 URL 생성 (raw HTTP 클라이언트용)"""
    username_parts = [PROXYHAT_USER, f"country-{country}"]
    if session_id:
        username_parts.append(f"session-{session_id}")
    username = "-".join(username_parts)
    return f"http://{username}:{PROXYHAT_PASS}@{PROXYHAT_GATEWAY}:{PROXYHAT_PORT}"

async def fetch_amazon_search(
    keyword: str,
    page: int = 1,
    marketplace: str = "com",
    country: str = "US",
    session_id: Optional[str] = None,
) -> str:
    """아마존 검색 결과 페이지 HTML 반환"""
    base_urls = {
        "com": "https://www.amazon.com",
        "de": "https://www.amazon.de",
        "co.uk": "https://www.amazon.co.uk",
        "co.jp": "https://www.amazon.co.jp",
    }
    base = base_urls.get(marketplace, base_urls["com"])
    url = f"{base}/s?k={keyword}&page={page}"
    
    proxy = build_proxy_url(country, session_id)
    
    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": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.9" if country == "US" else "de-DE,de;q=0.9",
        "Referer": base,
    }
    
    async with AsyncSession(impersonate="chrome120") as session:
        try:
            resp = await session.get(
                url,
                headers=headers,
                proxies={"http": proxy, "https": proxy},
                timeout=30,
            )
            if resp.status_code == 200:
                return resp.text
            elif resp.status_code == 503:
                raise Exception(f"CAPTCHA 또는 차단 감지 (HTTP {resp.status_code})")
            else:
                raise Exception(f"예상치 못한 상태 코드: {resp.status_code}")
        except Exception as e:
            raise Exception(f"아마존 검색 실패: {e}")

# 실행 예제
async def main():
    html = await fetch_amazon_search(
        keyword="wireless earbuds",
        page=1,
        marketplace="com",
        country="US",
        session_id="track-earbuds-001",
    )
    print(f"가져온 HTML 길이: {len(html)} 문자")

asyncio.run(main())

코드 예제 3: 검색 결과 파싱 및 순위 계산

from bs4 import BeautifulSoup
from dataclasses import dataclass
from typing import Optional

@dataclass
class SearchResult:
    asin: str
    title: str
    is_sponsored: bool
    organic_position: Optional[int]  # Sponsored이면 None
    page: int

def parse_amazon_search_results(
    html: str,
    target_asin: str,
    page: int = 1,
) -> tuple[list[SearchResult], Optional[int]]:
    """
    아마존 검색 결과 HTML을 파싱하여 결과 목록과 타겟 ASIN의 유기적 순위 반환.
    
    Returns:
        (전체 결과 목록, 타겟 ASIN의 유기적 순위 또는 None)
    """
    soup = BeautifulSoup(html, "html.parser")
    cards = soup.select('[data-component-type="s-search-result"]')
    
    results: list[SearchResult] = []
    organic_counter = 0
    target_position = None
    
    for card in cards:
        asin = card.get("data-asin", "")
        if not asin:
            continue
        
        # 제목 추출
        title_elem = card.select_one("h2 a")
        title = title_elem.get_text(strip=True) if title_elem else ""
        
        # Sponsored 감지
        sponsored_text = card.get_text()
        is_sponsored = "Sponsored" in sponsored_text or "sponsored" in sponsored_text.lower()
        
        # 유기적 순위 계산
        organic_pos = None
        if not is_sponsored:
            organic_counter += 1
            organic_pos = organic_counter
            
            if asin == target_asin:
                target_position = organic_pos
        
        results.append(SearchResult(
            asin=asin,
            title=title,
            is_sponsored=is_sponsored,
            organic_position=organic_pos,
            page=page,
        ))
    
    return results, target_position

# 사용 예제
# html = open("amazon_search_page1.html").read()
# results, position = parse_amazon_search_results(html, "B0CX23V2ZK")
# for r in results[:5]:
#     label = "[Sponsored]" if r.is_sponsored else f"[#{r.organic_position}]"
#     print(f"{label} {r.asin}: {r.title[:60]}")
# print(f"\n타겟 ASIN 순위: {position}")

코드 예제 4: Playwright로 전체 파이프라인 구축

JavaScript 렌더링이 필요한 경우 Playwright를 사용합니다. 아래는 1~5페이지를 순회하며 타겟 ASIN의 순위를 추적하는 완전한 예제입니다.

import asyncio
from datetime import datetime
from playwright.async_api import async_playwright
from bs4 import BeautifulSoup
from dataclasses import dataclass, asdict
import json
import logging

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

@dataclass
class RankRecord:
    asin: str
    keyword: str
    marketplace: str
    organic_position: Optional[int]
    page: Optional[int]
    found: bool
    timestamp: str

PROXYHAT_USER = "YOUR_USERNAME"
PROXYHAT_PASS = "YOUR_PASSWORD"
PROXYHAT_GATEWAY = "gate.proxyhat.com"
PROXYHAT_PORT = 8080

def build_proxy_auth(country: str, session_id: str) -> dict:
    """ProxyHat 인증 정보 생성 (Playwright용)"""
    username = f"{PROXYHAT_USER}-country-{country}-session-{session_id}"
    return {
        "server": f"http://{PROXYHAT_GATEWAY}:{PROXYHAT_PORT}",
        "username": username,
        "password": PROXYHAT_PASS,
    }

async def track_keyword_rank(
    keyword: str,
    target_asin: str,
    marketplace: str = "com",
    country: str = "US",
    max_pages: int = 5,
) -> list[RankRecord]:
    """키워드에 대한 타겟 ASIN의 순위를 1~max_pages 페이지에서 추적"""
    base_urls = {
        "com": "https://www.amazon.com",
        "de": "https://www.amazon.de",
        "co.uk": "https://www.amazon.co.uk",
    }
    base = base_urls.get(marketplace, base_urls["com"])
    session_id = f"track-{keyword.replace(' ', '-')}-{datetime.now().strftime('%Y%m%d')}"
    proxy_auth = build_proxy_auth(country, session_id)
    
    records: list[RankRecord] = []
    timestamp = datetime.now().isoformat()
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            proxy=proxy_auth,
            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()
        
        for page_num in range(1, max_pages + 1):
            url = f"{base}/s?k={keyword}&page={page_num}"
            logger.info(f"페이지 {page_num} 가져오는 중: {url}")
            
            try:
                await page.goto(url, wait_until="domcontentloaded", timeout=30000)
                await page.wait_for_selector(
                    '[data-component-type="s-search-result"]',
                    timeout=10000,
                )
                
                html = await page.content()
                results, position = parse_amazon_search_results(html, target_asin, page_num)
                
                found = position is not None
                records.append(RankRecord(
                    asin=target_asin,
                    keyword=keyword,
                    marketplace=marketplace,
                    organic_position=position,
                    page=page_num if found else None,
                    found=found,
                    timestamp=timestamp,
                ))
                
                if found:
                    logger.info(f"✓ ASIN {target_asin} 발견: 페이지 {page_num}, 순위 #{position}")
                    break  # 찾았으므로 중단
                else:
                    logger.info(f"페이지 {page_num}에 ASIN 없음, 다음 페이지로...")
                    await asyncio.sleep(2)  # 페이지 간 지연
                    
            except Exception as e:
                logger.error(f"페이지 {page_num} 실패: {e}")
                records.append(RankRecord(
                    asin=target_asin,
                    keyword=keyword,
                    marketplace=marketplace,
                    organic_position=None,
                    page=None,
                    found=False,
                    timestamp=timestamp,
                ))
                
                # CAPTCHA 감지 시 더 긴 대기
                if "503" in str(e) or "captcha" in str(e).lower():
                    logger.warning("CAPTCHA 감지, 30초 대기")
                    await asyncio.sleep(30)
        
        await browser.close()
    
    return records

async def main():
    records = await track_keyword_rank(
        keyword="wireless earbuds",
        target_asin="B0CX23V2ZK",
        marketplace="com",
        country="US",
        max_pages=5,
    )
    
    # JSON으로 저장
    output = [asdict(r) for r in records]
    with open("rank_history.json", "w") as f:
        json.dump(output, f, indent=2, ensure_ascii=False)
    
    print(f"{len(records)}개 레코드 저장 완료")

asyncio.run(main())

코드 예제 5: 재시도 및 백오프 로직

import asyncio
import random
from typing import Callable, TypeVar

T = TypeVar("T")

class CircuitBreakerOpen(Exception):
    """연속 실패 임계값 초과 시 발생"""
    pass

class RetryHandler:
    def __init__(
        self,
        max_retries: int = 3,
        base_delay: float = 2.0,
        max_delay: float = 60.0,
        circuit_threshold: int = 10,
    ):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.circuit_threshold = circuit_threshold
        self.consecutive_failures = 0
    
    async def execute(self, func: Callable[..., T], *args, **kwargs) -> T:
        """지수 백오프 + 지터를 사용한 재시도 래퍼"""
        if self.consecutive_failures >= self.circuit_threshold:
            raise CircuitBreakerOpen(
                f"연속 {self.circuit_threshold}회 실패 — 회로 차단기 활성화"
            )
        
        last_error = None
        for attempt in range(self.max_retries + 1):
            try:
                result = await func(*args, **kwargs)
                self.consecutive_failures = 0  # 성공 시 리셋
                return result
            except Exception as e:
                last_error = e
                self.consecutive_failures += 1
                
                if attempt == self.max_retries:
                    logger.error(f"최대 재시도 {self.max_retries}회 도달: {e}")
                    raise
                
                # 지수 백오프 + 지터
                delay = min(
                    self.base_delay * (2 ** attempt) + random.uniform(0, 1),
                    self.max_delay,
                )
                logger.warning(
                    f"시도 {attempt + 1}/{self.max_retries} 실패, "
                    f"{delay:.1f}초 후 재시도: {e}"
                )
                await asyncio.sleep(delay)
        
        raise last_error  # type: ignore

# 사용 예제
retry = RetryHandler(max_retries=3, base_delay=2.0, circuit_threshold=10)

async def safe_fetch(keyword: str, page: int, country: str):
    return await retry.execute(
        fetch_amazon_search,
        keyword=keyword,
        page=page,
        country=country,
        session_id=f"retry-{keyword}-{page}",
    )

코드 예제 6: 일일 스케줄링 및 순위 이력 저장

import asyncio
import aiosqlite
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

DB_PATH = "rank_tracker.db"

async def init_db():
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS rank_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                asin TEXT NOT NULL,
                keyword TEXT NOT NULL,
                marketplace TEXT NOT NULL,
                organic_position INTEGER,
                page INTEGER,
                found INTEGER NOT NULL,
                timestamp TEXT NOT NULL
            )
        """")
        await db.execute("""
            CREATE INDEX IF NOT EXISTS idx_asin_keyword 
            ON rank_history(asin, keyword, marketplace)
        """")
        await db.commit()

async def save_rank_record(record: RankRecord):
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            """INSERT INTO rank_history 
               (asin, keyword, marketplace, organic_position, page, found, timestamp)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            (
                record.asin,
                record.keyword,
                record.marketplace,
                record.organic_position,
                record.page,
                int(record.found),
                record.timestamp,
            ),
        )
        await db.commit()

async def get_rank_trend(asin: str, keyword: str, marketplace: str, days: int = 30):
    """최근 N일간의 순위 변화 추이 조회"""
    async with aiosqlite.connect(DB_PATH) as db:
        cursor = await db.execute(
            """SELECT timestamp, organic_position, page 
               FROM rank_history 
               WHERE asin = ? AND keyword = ? AND marketplace = ?
                 AND found = 1
               ORDER BY timestamp DESC
               LIMIT ?""",
            (asin, keyword, marketplace, days),
        )
        rows = await cursor.fetchall()
        return [
            {"timestamp": r[0], "position": r[1], "page": r[2]}
            for r in rows
        ]

# 일일 추적 작업
TRACKING_CONFIG = [
    {"keyword": "wireless earbuds", "asin": "B0CX23V2ZK", "marketplace": "com", "country": "US"},
    {"keyword": "bluetooth speaker", "asin": "B0CX23V2ZK", "marketplace": "com", "country": "US"},
    {"keyword": "kabellose kopfhörer", "asin": "B0CX23V2ZK", "marketplace": "de", "country": "DE"},
]

async def daily_tracking_job():
    """매일 실행되는 순위 추적 작업"""
    await init_db()
    
    for item in TRACKING_CONFIG:
        logger.info(f"추적 중: {item['keyword']} ({item['marketplace']})")
        try:
            records = await track_keyword_rank(
                keyword=item["keyword"],
                target_asin=item["asin"],
                marketplace=item["marketplace"],
                country=item["country"],
                max_pages=5,
            )
            for record in records:
                await save_rank_record(record)
        except Exception as e:
            logger.error(f"추적 실패 ({item['keyword']}): {e}")
        
        await asyncio.sleep(5)  # 키워드 간 지연

# cron 또는 APScheduler로 매일 실행
# 예: 0 9 * * * python -c "import asyncio; from tracker import daily_tracking_job; asyncio.run(daily_tracking_job())"

프로덕션 팁과 엣지 케이스

CAPTCHA 감지

아마존은 의심스러운 요청에 대해 CAPTCHA 페이지를 반환합니다. 다음 신호로 CAPTCHA를 감지하세요:

• HTTP 503 상태 코드
• URL에 /errors/validateCaptcha 포함
• HTML에 captchacharacters 또는 robot 키워드 포함

def detect_captcha(html: str, status_code: int) -> bool:
    if status_code == 503:
        return True
    captcha_signals = [
        "captchacharacters",
        "validateCaptcha",
        "Type the characters you see in this image",
    ]
    html_lower = html.lower()
    return any(signal.lower() in html_lower for signal in captcha_signals)

인덱싱 확인

ASIN이 전혀 순위에 나오지 않는다면, 인덱싱 문제일 수 있습니다. 인덱싱 확인은 ASIN을 직접 검색하여 상품 페이지가 존재하는지 확인하는 것으로 시작합니다:

# ASIN 직접 검색으로 인덱싱 확인
curl -x "http://user-country-US-session-index01:PASSWORD@gate.proxyhat.com:8080" \
  -H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64)" \
  "https://www.amazon.com/dp/B0CX23V2ZK" \
  -o product_page.html \
  --max-time 30

상품 페이지가 정상적으로 반환되면 인덱싱은 되어 있고, 키워드 관련성이 낮아 순위가 낮은 것입니다. 5페이지까지 찾지 못했다면 키워드 최적화가 필요한 신호입니다.

동시성 관리

여러 키워드를 동시에 추적할 때는 프록시 세션을 분리하세요. 각 키워드에 고유한 session_id를 할당하면 서로 다른 residential IP에서 요청이 나가 차단 위험이 분산됩니다. 100개 동시 세션 정도까지 안정적으로 운영할 수 있지만, 키워드당 3~5초 간격을 유지하는 것이 안전합니다.

결과 일관성

아마존 검색 결과는 시간대, 사용자 위치, 검색 히스토리에 따라 달라집니다. 일관된 비교를 위해 매일 같은 시간에 추적하고, 동일한 국가 IP를 사용하세요. ProxyHat의 sticky session은 최대 30분간 동일 IP를 유지합니다.

윤리와 컴플라이언스

핵심 원칙: 본인의 상품과 공개된 리스팅만 추적하고, 적절한 속도 제한을 적용하며, 가능하면 Amazon SP-API를 우선 사용하세요.

• robots.txt 준수: 아마존의 robots.txt를 확인하고 허용된 경로만 스크랩하세요.
• 속도 제한: 키워드당 3~5초 간격, 동시 요청은 10개 이하로 유지하세요.
• SP-API 우선: Amazon SP-API에서 제공하는 Catalog Items API와 Product Pricing API는 공식적으로 지원되는 데이터 접근 방법입니다. 승인된 셀러라면 SP-API를 사용하는 것이 가장 안전합니다.
• 데이터 최소화: 순위 데이터만 수집하고 개인 식별 정보는 저장하지 마세요. GDPR이 적용되는 EU 마켓플레이스에서 특히 중요합니다.

ProxyHat의 웹 스크래핑 사용 사례와 SERP 추적 사용 사례에서 추가 모범 사례를 확인할 수 있습니다. 프록시 가격은 사용량에 따라 선택할 수 있습니다.

주요 요점

• 아마존 SERP는 Google과 다릅니다 — 관련성과 판매 속도가 순위를 결정하므로 키워드·마켓플레이스·ASIN별 추적이 필수입니다.
• data-component-type="s-search-result"와 data-asin으로 상품 카드를 식별하고, "Sponsored" 라벨로 유기적 순위와 광고를 분리하세요.
• Residential 프록시 + 국가 geo-targeting(-country-US, -country-DE)으로 정확한 현지화 결과를 얻고, sticky session(-session-)으로 페이지네이션 일관성을 유지하세요.
• 지수 백오프 재시도, CAPTCHA 감지, 회로 차단기로 프로덕션 안정성을 확보하세요.
• 가능하면 Amazon SP-API를 우선 사용하고, 스크래핑 시에는 속도 제한과 윤리 가이드라인을 준수하세요.

더 자세한 프록시 설정은 ProxyHat 문서를 참조하세요.