亚马逊关键词选品完全指南2026：热搜词榜单数据驱动选品方法与Pangolinfo API实战

月销量 ≈ (近30天新增评论数 / 评论率) × 30
评论率通常为2-5%

┌─────────────────────────────────────────────────┐
│           定时任务调度器 (Cron Job)               │
│         每天凌晨2点触发数据采集任务                │
└────────────────┬────────────────────────────────┘
                 │
        ┌────────▼────────┐
        │  Pangolinfo API │
        │   数据采集层     │
        └────────┬────────┘
                 │
        ┌────────▼────────┐
        │   数据存储层     │
        │  (PostgreSQL)   │
        └────────┬────────┘
                 │
        ┌────────▼────────┐
        │   数据分析层     │
        │  (Python脚本)   │
        └────────┬────────┘
                 │
        ┌────────▼────────┐
        │   告警通知层     │
        │ (Slack/Email)   │
        └─────────────────┘

import requests
import json
from datetime import datetime
import psycopg2

class KeywordMonitor:
    """
    亚马逊关键词自动化监控系统
    """
    def __init__(self, api_token, db_config):
        self.api_token = api_token
        self.base_url = "https://scrapeapi.pangolinfo.com/api/v1/scrape"
        self.db_conn = psycopg2.connect(**db_config)
    
    def fetch_keyword_data(self, keyword, marketplace="US"):
        """
        抓取单个关键词的搜索结果数据
        """
        payload = {
            "url": f"https://www.amazon.com/s?k={keyword}",
            "formats": ["json"],
            "parserName": "amzKeyword",
            "bizContext": {
                "marketplace": marketplace
            }
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_token}",
            "Content-Type": "application/json"
        }
        
        response = requests.post(self.base_url, json=payload, headers=headers)
        
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"API请求失败: {response.status_code}")
    
    def save_to_database(self, keyword, data):
        """
        将数据保存到数据库
        """
        cursor = self.db_conn.cursor()
        
        sql = """
        INSERT INTO keyword_tracking 
        (keyword, search_volume, rank_position, top_products, created_at)
        VALUES (%s, %s, %s, %s, %s)
        """
        
        cursor.execute(sql, (
            keyword,
            data.get('searchVolume', 0),
            data.get('rankPosition', 0),
            json.dumps(data.get('topProducts', [])),
            datetime.now()
        ))
        
        self.db_conn.commit()
        cursor.close()
    
    def analyze_trend(self, keyword, days=7):
        """
        分析关键词趋势变化
        """
        cursor = self.db_conn.cursor()
        
        sql = """
        SELECT search_volume, rank_position, created_at
        FROM keyword_tracking
        WHERE keyword = %s AND created_at >= NOW() - INTERVAL '%s days'
        ORDER BY created_at DESC
        """
        
        cursor.execute(sql, (keyword, days))
        results = cursor.fetchall()
        cursor.close()
        
        if len(results) < 2:
            return None
        
        # 计算涨跌幅
        latest_volume = results[0][0]
        previous_volume = results[-1][0]
        change_rate = ((latest_volume - previous_volume) / previous_volume) * 100
        
        return {
            'keyword': keyword,
            'current_volume': latest_volume,
            'previous_volume': previous_volume,
            'change_rate': change_rate,
            'trend': 'up' if change_rate > 0 else 'down'
        }
    
    def send_alert(self, keyword, trend_data):
        """
        发送告警通知(Slack示例)
        """
        if abs(trend_data['change_rate']) > 50:  # 涨跌幅超过50%
            webhook_url = "YOUR_SLACK_WEBHOOK_URL"
            
            message = {
                "text": f"⚠️ 关键词异常波动提醒",
                "blocks": [
                    {
                        "type": "section",
                        "text": {
                            "type": "mrkdwn",
                            "text": f"*关键词*: {keyword}\n*搜索量变化*: {trend_data['change_rate']:.2f}%\n*当前搜索量*: {trend_data['current_volume']}\n*趋势*: {'📈 上涨' if trend_data['trend'] == 'up' else '📉 下跌'}"
                        }
                    }
                ]
            }
            
            requests.post(webhook_url, json=message)

# 使用示例
if __name__ == "__main__":
    # 配置
    api_token = "YOUR_PANGOLINFO_API_TOKEN"
    db_config = {
        "host": "localhost",
        "database": "keyword_monitor",
        "user": "postgres",
        "password": "your_password"
    }
    
    # 初始化监控器
    monitor = KeywordMonitor(api_token, db_config)
    
    # 监控的关键词列表
    keywords = [
        "mechanical pencil",
        "yoga mat",
        "portable heater",
        "christmas decoration"
    ]
    
    # 执行监控
    for keyword in keywords:
        try:
            # 抓取数据
            data = monitor.fetch_keyword_data(keyword)
            
            # 保存数据
            monitor.save_to_database(keyword, data)
            
            # 分析趋势
            trend = monitor.analyze_trend(keyword)
            
            # 发送告警
            if trend:
                monitor.send_alert(keyword, trend)
            
            print(f"✅ {keyword} 监控完成")
        
        except Exception as e:
            print(f"❌ {keyword} 监控失败: {str(e)}")

核心产品词 + 场景词 + 人群词 + 功能词

示例:
"yoga mat" + "outdoor" + "for men" + "non-slip"
= "outdoor non-slip yoga mat for men"(户外防滑男士瑜伽垫)