ClawHub

amazon-competitor-keepa-monitor

MCP Tools

亚马逊竞品全维运营监控报告生成器 — 通过 SellerSprite Keepa MCP 拉取多ASIN历史数据，生成 Chart.js 交互式 HTML 报告 + Excel 明细表

Install

openclaw skills install amazon-competitor-keepa-monitor

亚马逊竞品全维运营监控报告生成器

概述

通过 SellerSprite Keepa MCP 批量拉取多个 ASIN 的历史数据（价格、BSR、评论、评分），生成带筛选器的交互式 HTML 报告和 Excel 数据表，通过 Dinzee 交付公网链接。

文件结构

amazon-competitor-keepa-monitor/
├── SKILL.md                          # 本文件
├── templates/
│   └── report.html                   # Chart.js 交互式 HTML 报告模版
└── scripts/
    └── generate_excel.py             # Excel 工作簿生成脚本

模版说明

HTML 报告模版 (templates/report.html)

Chart.js 交互式报告，配色 #0f3460 + #e94560。Python 生成时用 f-string 替换以下占位符：

占位符说明示例
{REPORT_TITLE}报告标题（品类名）iPhone 15 手机壳
{DATE_RANGE}数据周期2025-11-21 ~ 2026-05-20
{GENERATE_TIME}生成时间2026-05-21 22:03
{COUNT}竞品数量8
{ASIN_CARDS_HTML}ASIN 卡片 HTML（见下方生成逻辑）见 Step 5
{DATA_JSON}JS DATA 对象 JSON见 Step 5
{MONTHLY_JSON}JS MONTHLY 对象 JSON见 Step 5
{ASIN_LIST_JSON}ASIN 数组 JSON["B0XX1","B0XX2"]
{VARIANT_ANALYSIS_JSON}变体分析 HTML 字符串（JSON-encoded）见 Step 5
{COUNTER_STRATEGY_JSON}反制策略 HTML 字符串（JSON-encoded）见 Step 5

⚠️ 重要: {DATA_JSON} 和 {MONTHLY_JSON} 体积巨大（~100KB），需要用 json.dumps(data, separators=(',',':')) 紧凑格式，不要加空格。

Excel 生成脚本 (scripts/generate_excel.py)

python scripts/generate_excel.py \
  --keepa /tmp/keepa_data.json \
  --details /tmp/asin_details.json \
  --output /tmp/competitive_report.xlsx

输出工作簿包含：

  • 概览汇总 Sheet：每个 ASIN 一行（当前价格/最低价/最高价/BSR/评论/评分/Deal天数/估算月销）
  • 每个 ASIN 一个 Sheet：逐日数据（日期/价格/BSR/BuyBox/Deal价格）
  • 月度估算销量 Sheet：按月汇总估算销量

数据源

  • SellerSprite MCP (keepa_info): 历史价格、BSR、评论、评分、BuyBox、Deal价
  • SellerSprite MCP (asin_detail): 基础信息（标题、品牌、上架时间、类目）
  • SIF MCP (可选): 广告结构、关键词流量

MCP 调用方式

SellerSprite (必须走代理)

import json, urllib.request

PROXY = {'https': 'http://127.0.0.1:1087', 'http': 'http://127.0.0.1:1087'}
# ⚠️ Config 位置：/Users/dino/.openclaw/workspace/config/mcporter.json (不是 ~/.hermes/mcporter.json)
# 也可用 `mcporter config list` 直接查看 sellersprite-mcp 的 URL
SS_URL = 'https://mcp.sellersprite.com/mcp?secret-key=<KEY>'  # 从 mcporter config 获取
opener = urllib.request.build_opener(urllib.request.ProxyHandler(PROXY))

def ss_init():
    # 1. initialize
    payload = json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'initialize',
        'params': {'protocolVersion': '2024-11-05', 'capabilities': {}, 
                   'clientInfo': {'name': 'agent', 'version': '1.0'}}}).encode()
    req = urllib.request.Request(SS_URL, data=payload, headers={
        'Content-Type': 'application/json', 'Accept': 'application/json, text/event-stream'})
    opener.open(req, timeout=30)
    # 2. notifications/initialized (必须!)
    payload_n = json.dumps({'jsonrpc': '2.0', 'method': 'notifications/initialized', 'params': {}}).encode()
    req_n = urllib.request.Request(SS_URL, data=payload_n, headers={
        'Content-Type': 'application/json', 'Accept': 'application/json, text/event-stream'})
    opener.open(req_n, timeout=10)

def ss_call(tool, args, call_id=10, timeout=180):
    payload = json.dumps({'jsonrpc': '2.0', 'id': call_id, 'method': 'tools/call',
        'params': {'name': tool, 'arguments': args}}).encode()
    req = urllib.request.Request(SS_URL, data=payload, headers={
        'Content-Type': 'application/json', 'Accept': 'application/json, text/event-stream'})
    resp = opener.open(req, timeout=timeout)
    raw = json.loads(resp.read())
    text = raw['result']['content'][0]['text']
    parsed = json.loads(text)
    return parsed.get('data', parsed)

⚠️ 关键坑点：

  • 必须走代理 http://127.0.0.1:1087
  • 必须发送 notifications/initialized 通知
  • keepa_info 返回 ~150KB，timeout 设 180s
  • mcporter call 会截断 >64KB 的响应，必须用直接 HTTP 方式

数据处理流程

Step 1: 收集基础信息

details = {}
for asin in ASINS:
    details[asin] = ss_call('asin_detail', {'marketplace': 'US', 'asin': asin})

Step 1b: 变体结构分析 (asin_detail)

asin_detail 返回 variationList（子ASIN列表）和 variations（总数），可用于竞品SKU矩阵分析：

def analyze_variants(detail):
    """从 asin_detail 提取颜色/型号变体分布"""
    colors, sizes = set(), set()
    for v in detail.get('variationList', []):
        attr = v.get('attribute', '')
        for p in attr.split(' | '):
            if p.startswith('Color:'): colors.add(p.replace('Color: ', '').strip())
            elif p.startswith('Size:'): sizes.add(p.replace('Size: ', '').strip())
    return {
        'total_variants': detail.get('variations', 0),
        'colors': sorted(colors),
        'sizes': sorted(sizes),
        'color_count': len(colors),
        'size_count': len(sizes),
    }

Step 2: 收集历史数据

keepa = {}
for asin in ASINS:
    keepa[asin] = ss_call('keepa_info', {'marketplace': 'US', 'asin': asin}, timeout=180)

Step 3: 处理为每日时间序列

def process_timeseries(items, start_ms, end_ms):
    daily = {}
    for item in items:
        ts = item.get('timePoint', 0)
        val = item.get('value')
        if val is None or val == -1 or ts < start_ms or ts > end_ms:
            continue
        date_str = datetime.datetime.fromtimestamp(ts/1000, tz=datetime.timezone.utc).strftime('%Y-%m-%d')
        if date_str not in daily or ts > daily[date_str][0]:
            daily[date_str] = (ts, val)
    return {k: v[1] for k, v in daily.items()}

Keepa 数据字段说明

字段说明格式
price历史价格[{timePoint: ms, value: float}]
bsr大类排名[{timePoint: ms, value: int}]
subSalesRank小类排名[{nodeId, node, ranks: [{timePoint, value}]}]
reviews评论数[{timePoint: ms, value: int}]
rating评分[{timePoint: ms, value: float}]
buyBoxBuyBox价格[{timePoint: ms, value: float}]
dealPriceDeal价格[{timePoint: ms, value: float}] (value=-1=无Deal)

Step 3b: Deal 活动×价格联动分析

def find_deal_periods(deal_prices, gap_days=2):
    dates = sorted(deal_prices.keys())
    if not dates: return []
    periods = []
    start = end = dates[0]
    for i in range(1, len(dates)):
        d1 = datetime.datetime.strptime(dates[i-1], '%Y-%m-%d')
        d2 = datetime.datetime.strptime(dates[i], '%Y-%m-%d')
        if (d2 - d1).days <= gap_days:
            end = dates[i]
        else:
            periods.append({'start': start, 'end': end, 'deal_price': deal_prices[start]})
            start = end = dates[i]
    periods.append({'start': start, 'end': end, 'deal_price': deal_prices[start]})
    return periods

Step 4: 生成 Excel

使用内置脚本：

cd /Users/dino/.hermes/skills/amazon-competitor-keepa-monitor
python scripts/generate_excel.py --keepa /tmp/keepa_data.json --details /tmp/asin_details.json --output /tmp/competitive_report.xlsx

Step 5: 生成 HTML 报告

5a: 生成 DATA JSON（嵌入 JS 的核心数据）

import json

DATA = {}
for asin in ASINS:
    kd = keepa[asin]
    detail = details.get(asin, {})
    # 合并所有日期
    all_dates = set()
    for field in ['price', 'bsr', 'buyBox', 'dealPrice']:
        all_dates.update(kd.get(field, {}).keys())
    dates = sorted(all_dates)
    
    DATA[asin] = {
        'brand': detail.get('brand', '-'),
        'title': detail.get('title', '-')[:80],
        'image': detail.get('image', ''),
        'dates': dates,
        'prices': [kd.get('price', {}).get(d) for d in dates],
        'bsrs': [kd.get('bsr', {}).get(d) for d in dates],
        'sub_bsrs': [kd.get('subBsr', {}).get(d) for d in dates],  # 如果有的话
        'reviews': [kd.get('reviews', {}).get(d) for d in dates],  # 注意：keepa_info 可能没有 reviews
        'ratings': [kd.get('rating', {}).get(d) for d in dates],
    }

data_json = json.dumps(DATA, separators=(',',':'), ensure_ascii=False)

5b: 生成 MONTHLY JSON

MONTHLY = {}
for asin in ASINS:
    kd = keepa[asin]
    month_data = {}
    for field_name in ['price', 'bsr']:
        for date_str, val in kd.get(field_name, {}).items():
            month = date_str[:7]
            if month not in month_data:
                month_data[month] = {'prices': [], 'bsrs': []}
            if field_name == 'price':
                month_data[month]['prices'].append(val)
            else:
                month_data[month]['bsrs'].append(val)
    # 加上评论首尾
    reviews = kd.get('reviews', {})
    for month in month_data:
        month_dates = [d for d in sorted(reviews.keys()) if d.startswith(month)]
        if month_dates:
            month_data[month]['reviews_start'] = reviews[month_dates[0]]
            month_data[month]['reviews_end'] = reviews[month_dates[-1]]
    MONTHLY[asin] = month_data

monthly_json = json.dumps(MONTHLY, separators=(',',':'), ensure_ascii=False)

5c: 生成 ASIN 卡片 HTML

def generate_asin_cards(DATA, ASINS):
    cards = []
    for i, asin in enumerate(ASINS):
        d = DATA[asin]
        prices = [p for p in d['prices'] if p is not None]
        bsrs = [b for b in d['bsrs'] if b is not None]
        reviews = [r for r in d['reviews'] if r is not None]
        
        current_price = f"${prices[-1]:.2f}" if prices else '-'
        current_bsr = f"BSR {bsrs[-1]:,}" if bsrs else '-'
        review_growth = f"+{reviews[-1] - reviews[0]:,}" if len(reviews) > 1 else '-'
        img_url = d.get('image', '')
        
        active = ' active' if i == 0 else ''
        card = f'''<div class="asin-card{active}" onclick="selectAsin('{asin}', this)" id="card-{asin}">
<img src="{img_url}" alt="{asin}" onerror="this.src='data:image/svg+xml,<svg xmlns=%22http://www.w3.org/2000/svg%22 viewBox=%220 0 60 60%22><rect fill=%22%231f2937%22 width=%2260%22 height=%2260%22/></svg>'">
<div class="info">
<div class="brand">{d['brand']} | {asin}</div>
<div class="title">{d['title'][:60]}</div>
<div style="margin-top:8px;font-size:.8em">
<span class="tag tag-green">{current_price}</span>
<span class="tag tag-blue">{current_bsr}</span>
<span class="tag tag-yellow">评论 {review_growth}</span>
</div>
</div>
</div>'''
        cards.append(card)
    return '\n'.join(cards)

asin_cards_html = generate_asin_cards(DATA, ASINS)

5d: 组装最终 HTML

# 读取模版
skill_dir = '/Users/dino/.hermes/skills/amazon-competitor-keepa-monitor'
with open(f'{skill_dir}/templates/report.html') as f:
    template = f.read()

# 替换占位符
html = template
html = html.replace('{REPORT_TITLE}', 'iPhone 15 手机壳')
html = html.replace('{DATE_RANGE}', '2025-11-21 ~ 2026-05-20')
html = html.replace('{GENERATE_TIME}', '2026-05-21 22:03')
html = html.replace('{COUNT}', str(len(ASINS)))
html = html.replace('{ASIN_CARDS_HTML}', asin_cards_html)
html = html.replace('{DATA_JSON}', data_json)
html = html.replace('{MONTHLY_JSON}', monthly_json)
html = html.replace('{ASIN_LIST_JSON}', json.dumps(ASINS))
html = html.replace('{VARIANT_ANALYSIS_JSON}', '""')  # 可选
html = html.replace('{COUNTER_STRATEGY_JSON}', '""')  # 可选

# 写入文件
with open('/tmp/competitive_report.html', 'w') as f:
    f.write(html)

Step 6: 上传交付

curl -sL -X POST "https://report.dinzee.ai/report/upload" \
  -H "Authorization: 538f4ad962266f5bc62dabda825e43021820988fdeeca85caaa1aca20e49a0eb" \
  -F "file=@/tmp/competitive_report.html;filename=report.html"

Excel 同理：

curl -sL -X POST "https://report.dinzee.ai/report/upload" \
  -H "Authorization: 538f4ad962266f5bc62dabda825e43021820988fdeeca85caaa1aca20e49a0eb" \
  -F "file=@/tmp/competitive_report.xlsx;filename=competitive_report.xlsx"

BSR → 销量估算公式

def bsr_to_daily_sales(bsr):
    if bsr <= 10: return 50
    if bsr <= 50: return 30
    if bsr <= 100: return 20
    if bsr <= 500: return 10
    if bsr <= 1000: return 5
    if bsr <= 5000: return 2
    return 1

注意事项

  • 数据时间范围：通过 timePoint 过滤，单位毫秒
  • mcporter call 会截断 >64KB 响应，必须用直接 HTTP
  • mcporter config 位置：/Users/dino/.openclaw/workspace/config/mcporter.json（不是 ~/.hermes/mcporter.json）
  • Excel Sheet 名称限制 31 字符
  • 报告配色：深蓝 #0f3460 + 红 #e94560
  • 8 个 ASIN 的 keepa 全量拉取约需 25 分钟，建议后台执行 + notify_on_complete
  • HTML 模版中 JS 的 allMonths 变量会自动从 MONTHLY 数据中提取月份，无需手动指定
More in MCP Tools