amazon-competitor-keepa-monitor

SKILL.md

---
name: amazon-competitor-keepa-monitor
description: 亚马逊竞品全维运营监控报告生成器 — 通过 SellerSprite Keepa MCP 拉取多ASIN历史数据，生成 Chart.js 交互式 HTML 报告 + Excel 明细表
trigger: 用户要求分析多个 ASIN 的历史价格/BSR/评论趋势，或生成竞品监控报告
tags: [amazon, competitor, keepa, report, chart-js, sellersprite]
---

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

## 概述
通过 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`)
```bash
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 (必须走代理)
```python
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: 收集基础信息
```python
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矩阵分析：
```python
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: 收集历史数据
```python
keepa = {}
for asin in ASINS:
    keepa[asin] = ss_call('keepa_info', {'marketplace': 'US', 'asin': asin}, timeout=180)
```

### Step 3: 处理为每日时间序列
```python
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}]` |
| `buyBox` | BuyBox价格 | `[{timePoint: ms, value: float}]` |
| `dealPrice` | Deal价格 | `[{timePoint: ms, value: float}]` (value=-1=无Deal) |

### Step 3b: Deal 活动×价格联动分析
```python
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
使用内置脚本：
```bash
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 的核心数据）
```python
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
```python
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
```python
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
```python
# 读取模版
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: 上传交付
```bash
curl -sL -X POST "https://report.dinzee.ai/report/upload" \
  -H "Authorization: 538f4ad962266f5bc62dabda825e43021820988fdeeca85caaa1aca20e49a0eb" \
  -F "file=@/tmp/competitive_report.html;filename=report.html"
```

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

## BSR → 销量估算公式
```python
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 数据中提取月份，无需手动指定