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_KEYyour_pangolinfo_api_key MY_SELLER_IDyour_amazon_seller_account_id FEISHU_WEBHOOK_URLhttps://open.feishu.cn/open-apis/bot/v2/hook/YOUR_TOKEN SLACK_WEBHOOK_URLhttps://hooks.slack.com/services/YOUR/SLACK/PATH TAOTOKEN_API_KEYyour_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_URLapi_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: fBearer {api_key}, Content-Type: application/json, }, json{ asin: asin, marketplace: marketplace, parse: True, include_offers: True, include_buybox: True, }, timeout30, ) 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三个参数值得单独说parseTrue让 API 直接返回结构化 JSON 而不是原始 HTML省掉解析工作include_offersTrue拿到所有卖家报价而不只是 Buybox 价格timeout30防止单个请求卡死拖垮整个采集周期。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( namePriceAnalyst, 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 fASIN: {snapshot[asin]}\n f当前竞品卖家: {snapshot.get(buybox_seller_name)}\n fFBA卖家: {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_output80字以内的竞争分析包含威胁等级和建议第一步应对动作, ) 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, jsonpayload, timeout10) 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, jsonpayload, timeout10) 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_idMY_SELLER_ID, alert_threshold_pctALERT_THRESHOLD) notifier MultiChannelNotifier(feishu_webhookFEISHU_WEBHOOK, slack_webhookSLACK_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, minutesPOLL_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. 本篇常见错排查QPangolinfo API 被限速了怎么办Pangolinfo 有并发限制批量采集时建议在请求之间加time.sleep(1)或者用异步并发但把最大并发数控制在 5 到 10 之间。别一上来就开 50 个线程容易被临时封禁。QOpenClaw 连接失败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 面板看价格走势。先把闭环跑起来再谈优化。