使用代理进行亚马逊关键词排名追踪：开发者完整指南

法律提示：本文仅讨论公开搜索数据的采集。在美国，未经授权访问受保护系统可能违反《计算机欺诈和滥用法》(CFAA)；在欧盟，个人数据处理受 GDPR 约束。请仅追踪你自己的商品列表或公开可见的搜索结果，遵守亚马逊服务条款，并在适用时优先使用 亚马逊SP-API。

为什么亚马逊关键词排名追踪需要代理

亚马逊搜索是一个独立的SERP生态系统。与Google不同，亚马逊的排名算法由相关性和销售速度驱动，而非链接权重。这意味着卖家必须针对每个关键词、每个市场（marketplace）和每个ASIN持续追踪自然排名位置，才能理解流量变化的原因。

当你直接用本地IP请求 amazon.com 搜索页时，通常在 50-200 次请求后就会触发反bot机制——返回CAPTCHA页面或HTTP 503。亚马逊的反爬系统会根据IP频率、TLS指纹和请求模式进行检测。这就是为什么亚马逊关键词排名追踪需要住宅代理来分散请求来源。

亚马逊SERP的特殊性

• 按市场本地化：amazon.com、amazon.de、amazon.co.jp 的搜索结果完全不同，需要按国家定位代理IP。
• 广告位混合：搜索结果页中自然结果和Sponsored广告交错出现，必须区分计算。
• 分页限制：通常最多显示 20 页结果，每页约 50-60 个商品。
• 位置依赖：同一关键词在不同邮编/城市返回不同结果。

解析亚马逊搜索结果页结构

亚马逊搜索结果页的核心结构是 [data-component-type="s-search-result"] 容器。每个商品卡片包含 data-asin 属性，可以直接读取ASIN。广告位通常带有 Sponsored 标签文本或 data-component-type="sp-sponsored-result"。

以下是亚马逊SERP抓取的解析逻辑：

from selectolax.parser import HTMLParser

def parse_amazon_serp(html: str, target_asin: str) -> dict:
    """解析亚马逊搜索结果页，返回目标ASIN的自然排名和广告位信息。"""
    tree = HTMLParser(html)
    results = tree.css('[data-component-type="s-search-result"]')

    organic_position = None
    sponsored_positions = []
    all_asins = []

    for idx, node in enumerate(results):
        asin = node.attributes.get('data-asin', '')
        if not asin:
            continue

        # 检测Sponsored标签
        sponsored_label = node.css_first('.a-color-secondary')
        is_sponsored = False
        if sponsored_label and 'Sponsored' in (sponsored_label.text() or ''):
            is_sponsored = True

        all_asins.append({'asin': asin, 'position': idx + 1, 'sponsored': is_sponsored})

        if is_sponsored:
            sponsored_positions.append(idx + 1)

        # 计算自然排名（排除广告位）
        if asin == target_asin and not is_sponsored:
            organic_count = sum(1 for r in all_asins if not r['sponsored'])
            organic_position = organic_count

    return {
        'target_asin': target_asin,
        'organic_position': organic_position,
        'sponsored_positions': sponsored_positions,
        'total_results': len(all_asins),
        'all_asins': all_asins,
    }

关键要点：自然排名是排除广告位后的序号，而不是页面上的原始位置。一个ASIN可能出现在第3位（广告）和第15位（自然），你需要分别记录。

为什么需要住宅代理与国家定位

数据中心IP几乎无法在亚马逊上持续工作。亚马逊维护着已知数据中心IP段黑名单，来自AWS、DigitalOcean等IP段的请求会被立即标记。住宅代理使用真实ISP分配的IP地址，具有更高的信任分数。

对于亚马逊排名追踪Python实现，你需要：

• 国家级地理定位：amazon.com 用美国IP（-country-US），amazon.de 用德国IP（-country-DE）。
• 粘性会话：分页抓取（第1-5页）时需要保持同一IP，否则不同页的结果可能来自不同地区，排名数据不一致。
• 请求轮换：不同关键词之间切换IP，避免单IP高频请求。

ProxyHat的网关地址是 gate.proxyhat.com，HTTP端口 8080，SOCKS5端口 1080。国家定位和会话ID通过用户名字段传递：

curl基础示例：单关键词单页抓取

先用最基础的curl验证代理连通性和页面结构：

# 抓取amazon.com搜索"bluetooth speaker"第1页
curl -s \
  --proxy "http://user-country-US-session-key001:pass@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=bluetooth+speaker&page=1" \
  -o amazon_serps.html

# 检查是否被CAPTCHA拦截
grep -c "captcha" amazon_serps.html

如果返回 0，说明请求成功；如果大于 0，需要更换IP或降低频率。

Python实现：curl_cffi + ProxyHat追踪关键词排名

curl_cffi 库能模拟浏览器的TLS指纹（JA3），显著降低被反bot系统拦截的概率。以下是完整的多页追踪示例：

import asyncio
import logging
from datetime import datetime, timezone
from selectolax.parser import HTMLParser
from curl_cffi.requests import AsyncSession

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

# ProxyHat配置
PROXYHAT_GATEWAY = "gate.proxyhat.com"
PROXYHAT_PORT = 8080
PROXYHAT_USER = "your_username"
PROXYHAT_PASS = "your_password"

def build_proxy_url(country: str, session_id: str) -> str:
    """构建ProxyHat代理URL，带国家定位和粘性会话。"""
    username = f"{PROXYHAT_USER}-country-{country}-session-{session_id}"
    return f"http://{username}:{PROXYHAT_PASS}@{PROXYHAT_GATEWAY}:{PROXYHAT_PORT}"

async def fetch_search_page(
    client: AsyncSession,
    keyword: str,
    page: int,
    marketplace: str,
    session_id: str,
) -> str | None:
    """抓取单个搜索结果页，带重试和退避。"""
    domain_map = {
        'US': 'www.amazon.com',
        'DE': 'www.amazon.de',
        'GB': 'www.amazon.co.uk',
        'JP': 'www.amazon.co.jp',
    }
    domain = domain_map.get(marketplace, 'www.amazon.com')
    url = f"https://{domain}/s?k={keyword}&page={page}"
    proxy = build_proxy_url(marketplace, 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-Language": "en-US,en;q=0.9" if marketplace == 'US' else "de-DE,de;q=0.9",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    }

    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = await client.get(url, proxy=proxy, headers=headers, timeout=30)
            if response.status_code == 200 and 'captcha' not in response.text.lower():
                logger.info(f"成功抓取 keyword={keyword} page={page} marketplace={marketplace}")
                return response.text
            elif 'captcha' in response.text.lower():
                logger.warning(f"CAPTCHA触发，尝试 {attempt+1}/{max_retries}")
                await asyncio.sleep(5 * (attempt + 1))  # 指数退避
            else:
                logger.warning(f"HTTP {response.status_code}，尝试 {attempt+1}/{max_retries}")
                await asyncio.sleep(2 * (attempt + 1))
        except Exception as e:
            logger.error(f"请求异常: {e}，尝试 {attempt+1}/{max_retries}")
            await asyncio.sleep(2 * (attempt + 1))

    return None

def find_asin_position(html: str, target_asin: str) -> dict:
    """在搜索结果页中定位目标ASIN的自然排名。"""
    tree = HTMLParser(html)
    results = tree.css('[data-component-type="s-search-result"]')

    organic_count = 0
    for node in results:
        asin = node.attributes.get('data-asin', '')
        if not asin:
            continue

        sponsored_el = node.css_first('.a-color-secondary')
        is_sponsored = sponsored_el and 'Sponsored' in (sponsored_el.text() or '')

        if not is_sponsored:
            organic_count += 1
            if asin == target_asin:
                return {
                    'found': True,
                    'organic_position': organic_count,
                    'page': None,  # 由调用方设置
                    'asin': asin,
                    'sponsored': False,
                }

    return {'found': False, 'organic_position': None, 'page': None, 'asin': target_asin, 'sponsored': False}

async def track_keyword_ranking(
    keyword: str,
    target_asin: str,
    marketplace: str = 'US',
    max_pages: int = 5,
) -> list[dict]:
    """追踪指定关键词下目标ASIN的排名，抓取多页。"""
    session_id = f"track-{keyword.replace(' ', '-')}-{datetime.now().strftime('%Y%m%d')}"
    history = []

    async with AsyncSession(impersonate="chrome120") as client:
        for page in range(1, max_pages + 1):
            html = await fetch_search_page(client, keyword, page, marketplace, session_id)
            if not html:
                logger.error(f"第{page}页抓取失败，跳过")
                continue

            result = find_asin_position(html, target_asin)
            result['page'] = page
            result['keyword'] = keyword
            result['marketplace'] = marketplace
            result['timestamp'] = datetime.now(timezone.utc).isoformat()
            history.append(result)

            if result['found']:
                logger.info(f"找到 ASIN={target_asin} 在第{page}页自然位置{result['organic_position']}")
                break  # 找到目标后停止翻页

            await asyncio.sleep(2)  # 页间延迟，降低频率

    return history

# 运行示例
if __name__ == "__main__":
    results = asyncio.run(track_keyword_ranking(
        keyword="bluetooth speaker",
        target_asin="B0D1234567",
        marketplace="US",
        max_pages=5,
    ))
    for r in results:
        print(r)

Playwright方案：处理动态加载和CAPTCHA

某些关键词页面使用动态加载，curl_cffi可能拿不到完整DOM。Playwright可以渲染JavaScript，但需要配合代理使用：

from playwright.async_api import async_playwright
import asyncio

async def track_with_playwright(keyword: str, target_asin: str, marketplace: str = 'US'):
    """使用Playwright渲染搜索页并提取ASIN排名。"""
    proxy_config = {
        'server': f'http://gate.proxyhat.com:8080',
        'username': f'your_username-country-{marketplace}-session-pw-{keyword.replace(" ", "-")}',
        'password': 'your_password',
    }

    domain_map = {'US': 'www.amazon.com', 'DE': 'www.amazon.de', 'JP': 'www.amazon.co.jp'}
    domain = domain_map.get(marketplace, 'www.amazon.com')

    async with async_playwright() as p:
        browser = await p.chromium.launch(
            proxy=proxy_config,
            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 marketplace == 'US' else 'de-DE',
        )
        page = await context.new_page()

        organic_position = None
        for pg in range(1, 6):
            url = f'https://{domain}/s?k={keyword}&page={pg}'
            try:
                await page.goto(url, wait_until='domcontentloaded', timeout=30000)
                await page.wait_for_selector('[data-component-type="s-search-result"]', timeout=10000)

                # 检测CAPTCHA
                content = await page.content()
                if 'captcha' in content.lower():
                    print(f'第{pg}页CAPTCHA触发，需要更换IP')
                    break

                # 提取ASIN和位置
                results = await page.query_selector_all('[data-component-type="s-search-result"]')
                count = 0
                for node in results:
                    asin = await node.get_attribute('data-asin')
                    if not asin:
                        continue

                    # 检查Sponsored
                    text = await node.inner_text()
                    is_sponsored = 'Sponsored' in text

                    if not is_sponsored:
                        count += 1
                        if asin == target_asin:
                            organic_position = count
                            print(f'找到: ASIN={target_asin} 第{pg}页 自然位置{count}')
                            await browser.close()
                            return {'page': pg, 'organic_position': count, 'found': True}

                await asyncio.sleep(2)
            except Exception as e:
                print(f'第{pg}页异常: {e}')
                continue

        await browser.close()
        return {'page': None, 'organic_position': None, 'found': False}

# 运行
if __name__ == '__main__':
    result = asyncio.run(track_with_playwright('wireless earbuds', 'B0CX123456', 'US'))
    print(result)

Node.js方案：批量关键词并发追踪

当需要同时追踪多个关键词时，Node.js的并发模型非常合适：

import { HttpsProxyAgent } from 'https-proxy-agent';
import fetch from 'node-fetch';
import { load } from 'cheerio';

const PROXYHAT_GATEWAY = 'gate.proxyhat.com';
const PROXYHAT_PORT = 8080;
const PROXYHAT_USER = 'your_username';
const PROXYHAT_PASS = 'your_password';

function buildProxyAgent(country, sessionId) {
  const username = `${PROXYHAT_USER}-country-${country}-session-${sessionId}`;
  const proxyUrl = `http://${username}:${PROXYHAT_PASS}@${PROXYHAT_GATEWAY}:${PROXYHAT_PORT}`;
  return new HttpsProxyAgent(proxyUrl);
}

async function fetchAmazonSerp(keyword, page, marketplace, sessionId) {
  const domainMap = { US: 'www.amazon.com', DE: 'www.amazon.de', GB: 'www.amazon.co.uk' };
  const domain = domainMap[marketplace] || 'www.amazon.com';
  const url = `https://${domain}/s?k=${encodeURIComponent(keyword)}&page=${page}`;
  const agent = buildProxyAgent(marketplace, sessionId);

  const 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': marketplace === 'US' ? 'en-US,en;q=0.9' : 'de-DE,de;q=0.9',
  };

  const maxRetries = 3;
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const res = await fetch(url, { agent, headers, timeout: 30000 });
      const html = await res.text();
      if (html.toLowerCase().includes('captcha')) {
        console.warn(`CAPTCHA on page ${page}, retry ${attempt + 1}`);
        await new Promise(r => setTimeout(r, 5000 * (attempt + 1)));
        continue;
      }
      return html;
    } catch (err) {
      console.error(`Fetch error page ${page}: ${err.message}`);
      await new Promise(r => setTimeout(r, 2000 * (attempt + 1)));
    }
  }
  return null;
}

function findAsinRank(html, targetAsin) {
  const $ = load(html);
  let organicCount = 0;
  let found = false;
  let position = null;

  $('[data-component-type="s-search-result"]').each((_, el) => {
    const asin = $(el).attr('data-asin');
    if (!asin) return;
    const text = $(el).text();
    const isSponsored = text.includes('Sponsored');
    if (!isSponsored) {
      organicCount++;
      if (asin === targetAsin) {
        found = true;
        position = organicCount;
      }
    }
  });

  return { found, organicPosition: position };
}

async function trackKeyword(keyword, asin, marketplace = 'US', maxPages = 5) {
  const sessionId = `node-${keyword.replace(/\s/g, '-')}-${Date.now()}`;
  const results = [];

  for (let page = 1; page <= maxPages; page++) {
    const html = await fetchAmazonSerp(keyword, page, marketplace, sessionId);
    if (!html) continue;
    const rank = findAsinRank(html, asin);
    results.push({ page, ...rank, keyword, asin, marketplace, timestamp: new Date().toISOString() });
    if (rank.found) break;
    await new Promise(r => setTimeout(r, 2000));
  }

  return results;
}

// 并发追踪多个关键词
const keywords = [
  { keyword: 'bluetooth speaker', asin: 'B0D1234567' },
  { keyword: 'wireless earbuds', asin: 'B0CX7654321' },
  { keyword: 'usb c cable', asin: 'B0AB9876543' },
];

const results = await Promise.all(
  keywords.map(kw => trackKeyword(kw.keyword, kw.asin, 'US', 5))
);
console.log(JSON.stringify(results, null, 2));

存储排名历史数据

追踪的价值在于历史趋势。以下是SQLite存储方案：

import sqlite3
from datetime import datetime, timezone

def init_db(db_path: str = 'amazon_ranks.db') -> sqlite3.Connection:
    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,
            page INTEGER,
            organic_position INTEGER,
            found INTEGER DEFAULT 0,
            timestamp TEXT NOT NULL
        )
    ''')
    conn.execute('CREATE INDEX IF NOT EXISTS idx_keyword_asin ON rank_history(keyword, asin, marketplace)')
    conn.commit()
    return conn

def save_rank(conn: sqlite3.Connection, keyword: str, asin: str, marketplace: str,
               page: int, organic_position: int | None, found: bool):
    conn.execute(
        'INSERT INTO rank_history (keyword, asin, marketplace, page, organic_position, found, timestamp) VALUES (?,?,?,?,?,?,?)',
        (keyword, asin, marketplace, page, organic_position, 1 if found else 0, datetime.now(timezone.utc).isoformat())
    )
    conn.commit()

def get_rank_trend(conn: sqlite3.Connection, keyword: str, asin: str, marketplace: str, days: int = 30):
    """获取最近N天的排名趋势。"""
    cursor = conn.execute(
        '''SELECT timestamp, organic_position, found FROM rank_history
           WHERE keyword=? AND asin=? AND marketplace=?
           ORDER BY timestamp DESC LIMIT ?''',
        (keyword, asin, marketplace, days)
    )
    return [{'timestamp': row[0], 'position': row[1], 'found': bool(row[2])} for row in cursor]

生产级最佳实践

每日调度

使用 APScheduler 或 cron 定时运行追踪任务。建议在非高峰时段（如UTC 03:00-06:00）执行，减少被检测概率：

from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def daily_rank_job():
    keywords = load_keywords_from_config()  # 你的关键词列表
    tasks = [track_keyword_ranking(kw['keyword'], kw['asin'], kw['marketplace']) for kw in keywords]
    await asyncio.gather(*tasks, return_exceptions=True)

scheduler = AsyncIOScheduler()
scheduler.add_job(daily_rank_job, 'cron', hour=4, minute=0, timezone='UTC')
scheduler.start()

重试与退避策略

• HTTP 503或CAPTCHA：指数退避 5s → 10s → 20s，最多 3 次重试。
• 连接超时：线性退避 2s → 4s → 6s。
• 连续失败超过 5 次：熔断该关键词 30 分钟。

CAPTCHA检测

在响应HTML中检测以下标志：

def detect_captcha(html: str) -> bool:
    """检测亚马逊CAPTCHA页面。"""
    captcha_markers = [
        'captcha',
        'robot check',
        'Type the characters you see in this image',
        'api-services-support@amazon.com',
    ]
    html_lower = html.lower()
    return any(marker.lower() in html_lower for marker in captcha_markers)

索引化检查

如果ASIN在前 5 页（约 250 个结果）中完全找不到，可能意味着：

• 该ASIN未被该关键词索引（listing未优化该关键词）。
• 该ASIN已下架或库存为零。
• 搜索结果被个性化过滤（需要清除cookie和使用新IP）。

伦理与合规

核心原则：仅追踪你自己的商品列表或公开可见的搜索结果数据。不要采集个人数据，不要绕过付费墙，不要以损害亚马逊基础设施的频率发送请求。

• 请求频率：单IP每分钟不超过 10-15 个请求，关键词之间随机延迟 2-5 秒。
• 遵守robots.txt：检查 amazon.com/robots.txt 了解允许的爬取路径。
• 优先使用SP-API：如果你是亚马逊卖家，SP-API 提供官方授权的目录和报告接口，无需爬取。
• 数据最小化：只存储排名位置和ASIN，不存储用户评价或个人数据。

了解更多代理配置选项，请访问 ProxyHat代理位置 和 定价页面。如需更多Web抓取场景参考，请阅读 Web抓取用例 和 SERP追踪用例。完整API文档请参考 ProxyHat官方文档。

关键要点

• 亚马逊SERP独立于Google：排名由相关性和销售速度驱动，必须按关键词×ASIN×市场维度追踪。
• 分离自然排名与广告位：用 Sponsored 标签过滤，只计算自然位置。
• 住宅代理是必需品：数据中心IP在 50 次请求内就会被拦截，住宅IP配合国家定位和粘性会话才能稳定分页抓取。
• curl_cffi优于requests：TLS指纹模拟显著降低CAPTCHA触发率。
• 存储历史数据：排名趋势比单次快照更有价值，用SQLite或PostgreSQL持久化。
• 合规优先：优先考虑SP-API，控制请求频率，仅采集公开数据。