amazon-competitor-keepa-monitor — 亚马逊竞争对手监控工具Keepa

亚马逊竞品全维运营监控报告生成器 概述 通过 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=' # 从 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}] review 评论数 [{timePoint: ms, value: int}] rating 评分 [{timePoint: ms, value: float}] buybox BuyBox 状态 [{timePoint: ms, value: bool}] deal Deal 价格 [{timePoint: ms, value: float}] sales 估算销量 [{timePoint: ms, value: int}] stock 库存状态 [{timePoint: ms, value: bool}] fba FBA 状态 [{timePoint: ms, value: bool}] merchant 供应商 [{timePoint: ms, value: str}] brand 品牌 [{timePoint: ms, value: str}] title 标题 [{timePoint: ms, value: str}] image 图片 [{timePoint: ms, value: str}] variation 变体 [{timePoint: ms, value: str}] weight 重量 [{timePoint: ms, value: float}] dimension 尺寸 [{timePoint: ms, value: str}] package 包装 [{timePoint: ms, value: str}] warranty 保修 [{timePoint: ms, value: str}] shipping 运输 [{timePoint: ms, value: str}] returns 退货 [{timePoint: ms, value: str}] questions 问题 [{timePoint: ms, value: str}] answers 答案 [{timePoint: ms, value: str}] reviews 评论 [{timePoint: ms, value: str}] ratings