电商竞品分析接口实战指南：从数据采集到决策洞察的全链路方案

┌─────────────────────────────────────────┐
│  第四层：决策应用层                        │
│  价格预警、选品推荐、动态定价、BI 报表       │
├─────────────────────────────────────────┤
│  第三层：分析引擎层                        │
│  价格趋势分析、评论情感分析、销量预测、异常检测 │
├─────────────────────────────────────────┤
│  第二层：数据标准化层                      │
│  多平台数据清洗、字段映射、统一模型、去重归一  │
├─────────────────────────────────────────┤
│  第一层：数据采集层                        │
│  官方 API、第三方 API、RPA、缓存队列        │
└─────────────────────────────────────────┘

from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List
@dataclass
class CompetitorProduct:
    """竞品统一数据模型"""
    # 基础信息
    platform: str                    # taobao / jd / 1688 / amazon / aliexpress
    product_id: str                  # 平台原始商品 ID
    title: str                       # 商品标题
    brand: Optional[str] = None      # 品牌
    
    # 价格体系（统一单位为"分"，避免浮点误差）
    current_price: int = 0           # 当前售价
    original_price: int = 0          # 划线价/原价
    promotion_price: int = 0         # 促销价
    currency: str = "CNY"            # 币种
    
    # 销售数据
    sales_count: Optional[int] = None      # 月销量/总销量
    review_count: Optional[int] = None     # 评价数
    review_score: Optional[float] = None   # 评分（1-5）
    rank: Optional[int] = None             # 类目排名/BSR
    
    # 库存与履约
    stock_status: str = "unknown"    # in_stock / out_of_stock / low_stock / unknown
    is_self_operated: bool = False   # 是否自营（京东/亚马逊）
    delivery_from: Optional[str] = None    # 发货地
    
    # 1688 特有
    wholesale_price: Optional[int] = None  # 批发价（分）
    min_order: Optional[int] = None        # 起订量
    
    # 亚马逊特有
    buy_box_price: Optional[int] = None    # Buy Box 价格
    fba: Optional[bool] = None             # 是否 FBA
    
    # 元数据
    fetch_time: datetime = datetime.now()
    url: Optional[str] = None        # 商品链接
class PlatformAdapter:
    """多平台数据适配器"""
    
    @staticmethod
    def from_taobao(raw: dict) -> CompetitorProduct:
        return CompetitorProduct(
            platform="taobao",
            product_id=str(raw.get("num_iid", "")),
            title=raw.get("title", ""),
            current_price=int(float(raw.get("price", 0)) * 100),
            original_price=int(float(raw.get("original_price", 0)) * 100),
            sales_count=raw.get("sold_quantity") or raw.get("sell_count"),
            review_count=raw.get("comment_count"),
            review_score=raw.get("score"),
            shop_name=raw.get("nick"),
            stock_status="in_stock" if raw.get("num") else "unknown",
            currency="CNY"
        )
    
    @staticmethod
    def from_jd(raw: dict) -> CompetitorProduct:
        price_info = raw.get("priceInfo", {})
        stock_info = raw.get("stockInfo", {})
        return CompetitorProduct(
            platform="jd",
            product_id=str(raw.get("skuId", "")),
            title=raw.get("title", ""),
            current_price=int(float(price_info.get("promotionPrice", 0)) * 100),
            original_price=int(float(price_info.get("jdPrice", 0)) * 100),
            sales_count=raw.get("salesInfo", {}).get("monthSales"),
            review_count=raw.get("salesInfo", {}).get("commentCount"),
            review_score=float(raw.get("salesInfo", {}).get("goodCommentRate", "0").replace("%", "")) / 20,
            is_self_operated=raw.get("shopInfo", {}).get("shopType") == "self",
            stock_status="in_stock" if stock_info.get("stockState") == 33 else "out_of_stock",
            currency="CNY"
        )
    
    @staticmethod
    def from_amazon_sp(raw: dict) -> CompetitorProduct:
        attributes = raw.get("attributes", {})
        title_list = attributes.get("title", [{}])
        title = title_list[0].get("value", "") if title_list else ""
        return CompetitorProduct(
            platform="amazon",
            product_id=raw.get("asin", ""),
            title=title,
            currency="USD"
            # 价格需从 Product Pricing API 单独获取
        )

def detect_price_anomaly(history_prices: List[int], threshold: float = 0.1) -> dict:
    """
    检测价格异常波动
    history_prices: 历史价格序列（按时间排序）
    threshold: 波动阈值，默认 10%
    """
    if len(history_prices) < 2:
        return {"is_anomaly": False}
    
    current = history_prices[-1]
    previous = history_prices[-2]
    avg_7d = sum(history_prices[-7:]) / len(history_prices[-7:])
    
    change_pct = (current - previous) / previous if previous else 0
    vs_avg_pct = (current - avg_7d) / avg_7d if avg_7d else 0
    
    return {
        "is_anomaly": abs(change_pct) > threshold,
        "change_pct": round(change_pct * 100, 2),
        "vs_7d_avg_pct": round(vs_avg_pct * 100, 2),
        "trend": "up" if change_pct > 0 else "down" if change_pct < 0 else "stable"
    }

from collections import Counter
import jieba
def extract_review_keywords(reviews: List[str], top_n: int = 20) -> dict:
    """
    从评价文本中提取高频痛点/卖点关键词
    """
    # 加载自定义词典（电商领域）
    jieba.load_userdict("ecommerce_dict.txt")
    
    # 停用词
    stopwords = set(["的", "了", "是", "我", "有", "和", "就", "都", "一个"])
    
    words = []
    for review in reviews:
        seg_list = jieba.cut(review)
        words.extend([w for w in seg_list if len(w) > 1 and w not in stopwords])
    
    word_freq = Counter(words).most_common(top_n)
    
    # 简单情感分类（基于预设词库）
    positive_words = {"好评", "满意", "推荐", "不错", "质量好", "物流快"}
    negative_words = {"差评", "失望", "退货", "质量问题", "假货", "慢"}
    
    sentiment = {"positive": [], "negative": [], "neutral": []}
    for word, count in word_freq:
        if word in positive_words:
            sentiment["positive"].append((word, count))
        elif word in negative_words:
            sentiment["negative"].append((word, count))
        else:
            sentiment["neutral"].append((word, count))
    
    return sentiment

ALERT_RULES = {
    "price_drop_5pct": lambda old, new: (old - new) / old > 0.05,
    "price_drop_10pct": lambda old, new: (old - new) / old > 0.10,
    "out_of_stock": lambda old_stock, new_stock: old_stock == "in_stock" and new_stock == "out_of_stock",
    "buy_box_lost": lambda old_bb, new_bb: old_bb is not None and new_bb is None  # 亚马逊特有
}

┌─────────────────────────────────────────────────────────────┐
│                        竞品监控中台                          │
├─────────────┬─────────────┬─────────────┬─────────────────┤
│   淘宝/天猫  │    1688     │    京东     │    亚马逊       │
│  API 采集   │  API 采集   │  API 采集   │  SP-API 采集    │
├─────────────┴─────────────┴─────────────┴─────────────────┤
│                      消息队列 (Kafka/RabbitMQ)              │
├─────────────────────────────────────────────────────────────┤
│  数据清洗 → 字段映射 → 统一模型 → 数据仓库 (ClickHouse/MySQL) │
├─────────────────────────────────────────────────────────────┤
│  实时计算 (Flink) │ 离线分析 (Spark) │ 时序数据库 (InfluxDB) │
├─────────────────────────────────────────────────────────────┤
│  价格预警引擎 │ 销量预测模型 │ 评论 NLP │ BI 可视化 (Grafana) │
├─────────────────────────────────────────────────────────────┤
│  企业微信/钉钉/邮件告警 │ 自动调价接口 │ 选品推荐报告        │
└─────────────────────────────────────────────────────────────┘