1. 亚马逊竞品调价预警，为什么值得自己搭一套
做亚马逊运营的朋友大概率都遇到过这种场景：早上打开后台，发现某个主力 ASIN 的 Buybox 被抢了，点进去一看，竞品昨晚悄悄降了 8%。等你手动翻完十几个竞品链接、对比完历史价格，大半天已经过去了，广告预算也烧掉了一截。市面上确实有现成的调价监控工具，但要么按 ASIN 数量收费贵得离谱，要么推送延迟到小时级，等你收到通知，黄花菜都凉了。
这篇要交付的，就是一套能自己掌控节奏的实时预警链路：用OpenClaw作为 AI Agent 调度入口，Pangolinfo API作为亚马逊 ASIN 数据源，APScheduler做 10 分钟级定时轮询，再通过Webhook把预警卡片推到飞书或 Slack 工作群。整套系统跑起来后，竞品一降价，你的运营群几分钟内就能收到带 AI 分析结论的告警，而不是干巴巴一个数字。
适合谁跟做：有 Python 3.9+ 基础、想给团队搭一套价格监控基础设施的亚马逊卖家或开发者。如果你完全没写过代码，也能照着把配置和命令复制跑通，只是排障时需要多一点耐心。下面从环境准备一路走到端到端验证，每一步都给可复制的代码和参数说明。
2. 前置准备：TaoToken 统一 Key 与 Pangolinfo 数据源
在写采集逻辑之前，先把两个关键凭证搞定：一个是 Pangolinfo 的 API Key（负责抓亚马逊数据），另一个是 TaoToken 的统一 Key（负责给 OpenClaw 里的 LLM 调用提供模型能力）。很多人卡在第一步就是因为把这两个 Key 搞混了，其实它们职责完全不同。
Pangolinfo API Key 的获取路径是：登录 Pangolinfo 控制台，进入 API 管理页面，创建一个新 Key，或者直接用 API 请求 Key。拿到后先别急着写代码，用 curl 测一下能不能通，避免后面把网络问题误判成代码 bug。
TaoToken 这边，你可以把它理解成一个统一的模型调用入口，OpenClaw 里 Agent 生成竞争态势分析时，底层走的就是它。注册后到控制台创建 API Key，地址是 https://taotoken.net/api-keys ，创建完记得复制保存，页面刷新后就看不到了。如果你后面打算长期跑编码类 Agent 任务，可以顺手看下 Coding Plan 页面 https://taotoken.net/coding-plan ，按量或包月都行，这里不展开。
环境依赖安装一条命令搞定：
pip install openclaw requests apscheduler python-dotenv然后在项目根目录建一个.env文件，把所有敏感信息集中管理，别硬编码进代码：
# .env PANGOLINFO_API_KEY=your_pangolinfo_api_key MY_SELLER_ID=your_amazon_seller_account_id FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/YOUR_TOKEN SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/SLACK/PATH TAOTOKEN_API_KEY=your_taotoken_api_key注意：
.env一定要加进.gitignore，别把 Key 推到公开仓库。我见过有人把 Webhook 地址提交到 GitHub，结果被陌生人往群里刷了几百条垃圾消息。
3. 可复制配置：config.toml 骨架与采集模块
OpenClaw 的 Agent 行为可以通过config.toml统一配置，把模型、阈值、监控列表都抽出来，改参数不用动代码。下面这份骨架可以直接复制，按注释替换成你自己的值：
# config.toml [openclaw] model = "claude-3-5-sonnet" api_base = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" max_tokens = 512 temperature = 0.3 [monitor] poll_interval_minutes = 10 alert_threshold_pct = 5.0 cooldown_hours = 2 [monitor.asins] watch_list = [ "B0D6BFMNN5", "B08K2S9X7Q", ] [notify] feishu_env = "FEISHU_WEBHOOK_URL" slack_env = "SLACK_WEBHOOK_URL"api_base指向 TaoToken 的 API 地址，OpenClaw 会读取TAOTOKEN_API_KEY环境变量完成鉴权。alert_threshold_pct是触发告警的最低降幅百分比，cooldown_hours用来防止同一个 ASIN 反复刷屏。
接着写数据采集模块price_collector.py，核心是调用 Pangolinfo Scrape API 拿结构化 JSON：
# price_collector.py import requests from datetime import datetime from typing import Optional, List PANGOLINFO_API_URL = "https://api.pangolinfo.com/v1/amazon/product" def fetch_asin_data(asin: str, api_key: str, marketplace: str = "US") -> Optional[dict]: resp = requests.post( PANGOLINFO_API_URL, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, json={ "asin": asin, "marketplace": marketplace, "parse": True, "include_offers": True, "include_buybox": True, }, timeout=30, ) resp.raise_for_status() raw = resp.json() return { "asin": asin, "collected_at": datetime.utcnow().isoformat() + "Z", "buybox_price": raw.get("buybox", {}).get("price"), "buybox_seller_name": raw.get("buybox", {}).get("seller_name"), "buybox_seller_id": raw.get("buybox", {}).get("seller_id"), "buybox_is_fba": raw.get("buybox", {}).get("is_fba"), "all_offers": raw.get("offers", []), "product_title": raw.get("title", ""), "in_stock": raw.get("availability") == "In Stock", } def batch_fetch(asin_list: List[str], api_key: str) -> List[dict]: results = [] for asin in asin_list: try: data = fetch_asin_data(asin, api_key) results.append(data) print(f"[OK] {asin}: ${data['buybox_price']} | {data['buybox_seller_name']}") except Exception as e: print(f"[FAIL] {asin}: {e}") return results三个参数值得单独说：parse=True让 API 直接返回结构化 JSON 而不是原始 HTML，省掉解析工作；include_offers=True拿到所有卖家报价，而不只是 Buybox 价格；timeout=30防止单个请求卡死拖垮整个采集周期。
4. OpenClaw Agent 分析层与 Webhook 推送
采集到数据只是原料，真正让预警有价值的是分析层。这里用 OpenClaw 的 Agent 来做差分对比和竞争态势判断，price_analyzer.py的核心逻辑如下：
# price_analyzer.py from openclaw import Agent, Task from typing import Optional class CompetitorPriceAnalyzer: DEFAULT_ALERT_THRESHOLD = 5.0 def __init__(self, my_seller_id: str = "", alert_threshold_pct: float = DEFAULT_ALERT_THRESHOLD): self.my_seller_id = my_seller_id self.threshold = alert_threshold_pct self._price_baseline: dict = {} self.agent = Agent( name="PriceAnalyst", role="亚马逊竞争定价分析专家", goal="快速识别竞品调价威胁，给出精准的竞争态势判断和应对建议", backstory="你是资深的亚马逊运营专家，擅长识别价格战信号并制定快速响应策略", ) def check(self, snapshot: dict) -> Optional[dict]: asin = snapshot["asin"] curr_price = snapshot.get("buybox_price") curr_seller_id = snapshot.get("buybox_seller_id") if curr_price is None: return None if asin not in self._price_baseline: self._price_baseline[asin] = {"price": curr_price, "seller_id": curr_seller_id, "history": []} print(f"[INIT] {asin} 基准价格: ${curr_price}") return None baseline = self._price_baseline[asin] prev_price = baseline["price"] if prev_price <= 0: return None change_pct = (curr_price - prev_price) / prev_price * 100 is_competitor_action = curr_seller_id != self.my_seller_id if change_pct <= -self.threshold and is_competitor_action: ai_analysis = self._generate_analysis(snapshot, prev_price, change_pct, baseline) baseline["history"].append({ "from": prev_price, "to": curr_price, "pct": change_pct, "ts": snapshot["collected_at"], }) alert_event = { "priority": "HIGH" if abs(change_pct) >= 10 else "MEDIUM", "asin": asin, "title": snapshot.get("product_title", "")[:60], "competitor_seller": snapshot.get("buybox_seller_name"), "price_from": prev_price, "price_to": curr_price, "change_pct": change_pct, "historical_changes": len(baseline["history"]), "ai_analysis": ai_analysis, "detected_at": snapshot["collected_at"], } baseline["price"] = curr_price baseline["seller_id"] = curr_seller_id return alert_event baseline["price"] = curr_price return None def _generate_analysis(self, snapshot, prev_price, change_pct, baseline) -> str: task = Task( description=( f"分析以下亚马逊竞品调价事件，输出简洁的中文竞争态势分析（不超过80字）：\n" f"ASIN: {snapshot['asin']}\n" f"当前竞品卖家: {snapshot.get('buybox_seller_name')}\n" f"FBA卖家: {snapshot.get('buybox_is_fba')}\n" f"降价幅度: {abs(change_pct):.1f}%（${prev_price:.2f} → ${snapshot['buybox_price']:.2f}）\n" f"历史调价次数: {len(baseline.get('history', []))}\n" f"市场库存状态: {snapshot.get('in_stock')}" ), expected_output="80字以内的竞争分析，包含威胁等级和建议第一步应对动作", ) return self.agent.execute_task(task)这里有两个设计细节：my_seller_id用来过滤自己账号的价格变化，只关注真实竞品动作；change_pct <= -threshold只在降价时告警，涨价不触发，如果你也关心涨价信号，把负号去掉即可。
推送模块notifier.py支持飞书和 Slack 双通道，飞书用交互式卡片，Slack 用 Block Kit：
# notifier.py import requests class MultiChannelNotifier: def __init__(self, feishu_webhook: str = "", slack_webhook: str = ""): self.feishu = feishu_webhook self.slack = slack_webhook def notify(self, event: dict): results = {} if self.feishu: results["feishu"] = self._send_feishu(event) if self.slack: results["slack"] = self._send_slack(event) status = " | ".join(f"{k}: {'OK' if v else 'FAIL'}" for k, v in results.items()) print(f"[ALERT] {event['asin']} 降价{abs(event['change_pct']):.1f}% → {status}") def _send_feishu(self, event: dict) -> bool: color = "red" if event["priority"] == "HIGH" else "orange" payload = { "msg_type": "interactive", "card": { "header": { "title": {"content": f"竞品降价 | ASIN: {event['asin']}", "tag": "plain_text"}, "template": color, }, "elements": [{ "tag": "div", "text": { "tag": "larkmd", "content": ( f"**商品**: {event['title']}\n" f"**竞品卖家**: {event['competitor_seller']}\n" f"**价格变化**: ${event['price_from']:.2f} → **${event['price_to']:.2f}**\n" f"**降幅**: {abs(event['change_pct']):.1f}% | 历史调价: {event['historical_changes']}次\n" f"**分析**: {event['ai_analysis']}" ), }, }], }, } resp = requests.post(self.feishu, json=payload, timeout=10) return resp.status_code == 200 def _send_slack(self, event: dict) -> bool: emoji = "🔴" if event["priority"] == "HIGH" else "🟡" payload = { "blocks": [ {"type": "header", "text": {"type": "plain_text", "text": f"{emoji} Competitor Price Drop | {event['asin']}"}}, {"type": "section", "fields": [ {"type": "mrkdwn", "text": f"*Competitor:*\n{event['competitor_seller']}"}, {"type": "mrkdwn", "text": f"*Price:*\n${event['price_from']:.2f} → ${event['price_to']:.2f} ({abs(event['change_pct']):.1f}%)"}, ]}, {"type": "section", "text": {"type": "mrkdwn", "text": f"*Analysis:* {event['ai_analysis']}"}}, ], } resp = requests.post(self.slack, json=payload, timeout=10) return resp.status_code == 200主程序main.py用 APScheduler 把采集、分析、推送串成闭环：
# main.py import os from dotenv import load_dotenv from apscheduler.schedulers.blocking import BlockingScheduler from price_collector import batch_fetch from price_analyzer import CompetitorPriceAnalyzer from notifier import MultiChannelNotifier load_dotenv() API_KEY = os.getenv("PANGOLINFO_API_KEY") MY_SELLER_ID = os.getenv("MY_SELLER_ID", "") FEISHU_WEBHOOK = os.getenv("FEISHU_WEBHOOK_URL", "") SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL", "") MONITOR_ASINS = ["B0D6BFMNN5", "B08K2S9X7Q"] POLL_INTERVAL = 10 ALERT_THRESHOLD = 5.0 analyzer = CompetitorPriceAnalyzer(my_seller_id=MY_SELLER_ID, alert_threshold_pct=ALERT_THRESHOLD) notifier = MultiChannelNotifier(feishu_webhook=FEISHU_WEBHOOK, slack_webhook=SLACK_WEBHOOK) def monitoring_cycle(): print("\n" + "-" * 40) snapshots = batch_fetch(MONITOR_ASINS, API_KEY) for snapshot in snapshots: alert = analyzer.check(snapshot) if alert: notifier.notify(alert) print(f"[DONE] {len(snapshots)} ASINs checked") if __name__ == "__main__": print(f"监控启动 | {len(MONITOR_ASINS)} ASINs | 每{POLL_INTERVAL}分钟") monitoring_cycle() scheduler = BlockingScheduler() scheduler.add_job(monitoring_cycle, "interval", minutes=POLL_INTERVAL) scheduler.start()5. 端到端验证：一次成功的告警推送
代码写完后别急着挂后台跑，先做一次端到端验证，确认整条链路是通的。验证分三步走。
第一步，单独测采集。在 Python 交互环境里跑：
from price_collector import fetch_asin_data import os data = fetch_asin_data("B0D6BFMNN5", os.getenv("PANGOLINFO_API_KEY")) print(data["buybox_price"], data["buybox_seller_name"])能打印出价格和卖家名，说明 Pangolinfo 这条链路没问题。如果报 401，检查 Key 有没有带Bearer前缀；如果超时，先确认网络能访问 API 域名。
第二步，手动触发一次告警。把ALERT_THRESHOLD临时改成0.1，然后跑python main.py。第一次运行只会初始化基准价格，不会告警。等 10 分钟后第二轮跑完，如果竞品价格有任何微小波动，飞书群就会收到卡片。收到卡片说明 OpenClaw 分析层和 Webhook 推送都正常。
第三步，验证 AI 分析内容。打开收到的飞书卡片，看分析字段是不是一段通顺的中文竞争态势判断，而不是报错信息或空字符串。如果这里是空的，大概率是 TaoToken 的 Key 没配对，或者config.toml里的api_base写错了。可以到模型对话页面 https://taotoken.net/chat 手动发一条消息，确认 Key 本身可用。
验证通过后，把ALERT_THRESHOLD改回5.0，用nohup python main.py &挂到服务器后台，或者用 systemd 做成服务，就正式上线了。
6. 本篇常见错排查
Q：Pangolinfo API 被限速了怎么办？Pangolinfo 有并发限制，批量采集时建议在请求之间加time.sleep(1)，或者用异步并发但把最大并发数控制在 5 到 10 之间。别一上来就开 50 个线程，容易被临时封禁。
Q：OpenClaw 连接失败，Agent 不返回分析？先检查~/.openclaw/config.yaml里的 LLM 配置，确认api_base指向https://taotoken.net/api，并且TAOTOKEN_API_KEY环境变量已经加载。可以用python -c "import os; print(os.getenv('TAOTOKEN_API_KEY'))"确认环境变量读到了。
Q：告警太多，群里被刷屏？两个办法：一是提高ALERT_THRESHOLD，比如从 5% 提到 8%；二是在_price_baseline里加冷却逻辑，同一个 ASIN 两次告警之间至少间隔 2 小时。冷却逻辑可以用baseline["last_alert_ts"]记录上次告警时间，判断时间差。
Q：想监控更多 ASIN，但担心 API 成本？10 分钟轮询是数据时效和 API 成本的最优平衡点。如果你有 100 个 ASIN，可以按优先级分组，核心竞品 10 分钟轮询，次要竞品 30 分钟轮询，用两个不同的 APScheduler job 分别调度。
Q：飞书卡片收到了，但 Slack 没收到？检查SLACK_WEBHOOK_URL是否完整，Slack 的 Webhook 地址很长，复制时容易漏掉末尾的路径段。另外确认 Slack 应用有chat:write权限。
整套系统跑通后，你手里就有了一条从数据采集到 AI 分析再到群内推送的完整预警链路。后续想扩展的话，可以把告警事件写进 SQLite 做历史回溯，或者接一个简单的 Web 面板看价格走势。先把闭环跑起来，再谈优化。