价格战，你永远比对手慢几步
做亚马逊的人都知道一件事：价格是影响Buybox和转化率的核心杠杆。但有多少运营团队真正做到了对竞品价格的实时感知？大多数人的方式是——每天早上打开后台，翻翻竞品的Offer页，或者登录某个卖家工具看一眼昨天的价格快照。这意味着，当你的主要竞品在昨晚11点悄悄把价格从$89调到了$74的时候，你要等到今天早上才知道。而在这段时间里，你的广告一直在高价位跑着，转化率跌了你却还不明白为什么。
亚马逊竞品调价监控的本质是时效性问题。竞品的每一次降价，都是另一个卖家的运营动作的结果：可能是库存积压要清货，可能是敏感到了你的广告在吃流量准备用价格截杀，可能是他们要冲某一天的BSR榜单。任何一种情况，你都需要在第一时间得知，才能做出及时的应对——跟价、维持、还是主动发起应对策略。
问题不是”要不要监控”，而是”如何监控才能快到有意义”。日报级的工具告诉你昨天发生了什么，周报告诉你上周的趋势——这些都有价值，但对于价格博弈这种以小时计的动态场景来说，它们永远是马后炮。你需要的是一套能在竞品调价后10分钟内就把消息推到你眼前的系统。
这正是今天这篇文章的目的：用 OpenClaw AI Agent 与 Pangolinfo Scrape API，从零搭建一套可以在生产环境真实运行的亚马逊竞品调价监控脚本，配置完成后全程自动运行，竞品价格有变动就把告警推到你的飞书或Slack工作群。
传统价格监控方案的三大硬伤
在介绍具体实现之前，有必要理解为什么现有方案不够用，这样才能明白这套方案解决的是什么核心问题。
第一个硬伤是数据频率。市面上大多数卖家工具的价格监控模块以日级为主，部分工具能做到4-6小时级。对于竞品调价这种场景，4小时已经意味着对方能完成一轮Flash Sale并恢复原价，而你的告警才刚刚触发。这不是告警，是事后记录。真正有价值的竞品价格预警窗口是10-30分钟，超过这个时间告警就失去了即时响应的意义。
第二个硬伤是通知渠道。邮件告警在2026年基本等同于没告警，大多数运营人员查邮件的频率已经低于每小时一次。真正能穿透注意力的通知渠道是工作即时通讯工具——飞书或Slack的消息卡片，会以推送的形式出现在你手机上，像工作消息一样无法忽视。而现有工具的告警大多数还停留在邮件+APP推送的组合。
第三个硬伤是数据深度与分析能力的分离。工具发给你的告警是”B0D6BFMNN5的价格从$89变成了$74″，但它不会告诉你：这是不是Amazon直营在动价？这个价格是否低于你的成本线？历史上这个竞品有没有类似的调价规律？把这些分析留给运营人员自己做，等于把告警的价值折半。引入AI Agent层，就是为了让告警不只是事件通知，而是带有初步分析和建议的决策辅助。
系统架构：三层模型，各司其职
整套亚马逊竞品调价监控系统由三层构成，每一层职责清晰，互不耦合。
第一层：数据采集层（Pangolinfo Scrape API）。这是整个系统的数据基础。Pangolinfo Scrape API 负责每隔10分钟对目标ASIN的商品页面发起采集请求，返回结构化的商品数据，包括当前Buybox价格、所有Offer列表中各卖家的报价、库存状态、到货时间预估等字段。这一层解决了”数据从哪里来”的问题，同时屏蔽了直接爬取亚马逊页面所涉及的反爬对抗、IP轮换、代理管理等技术复杂性。
第二层：AI分析层（OpenClaw Agent）。OpenClaw 是一个轻量的 AI Agent 框架，在这套方案里承担的是”看数据、做分析、做决策”的角色。当新一轮价格数据到来时，OpenClaw Agent 会将本次数据与上一次快照进行比对，识别价格变化、计算降幅百分比、判断是否触及设定的告警阈值，并根据历史数据生成一段简短的背景分析（例如：”该竞品上次降价发生在2周前，当时持续了约48小时后恢复原价”）。这一层把原始数据变成有语境的信息。
第三层：通知层（飞书 + Slack Webhook）。当分析层判定告警需要触发时，通知层负责把格式化的告警消息卡片推送到指定的飞书群或Slack频道。卡片包含：触发告警的ASIN、竞品卖家、新旧价格对比、降幅百分比、AI生成的简短分析，以及跳转查看详细数据的按钮。这一层解决”告警怎么到人眼前”的问题。
三层之间通过 Python 脚本串联，整体运行在定时任务（Cron / APScheduler）框架下，每10分钟触发一次完整的采集→分析→告警周期。
完整代码实现：手把手搭建动态定价追踪系统
环境准备
开始之前，确认以下依赖已就绪：
- Python 3.9+
- Pangolinfo API Key（在 Pangolinfo 控制台申请）
- OpenClaw 安装（
pip install openclaw） - 飞书自定义机器人Webhook URL 或 Slack Webhook URL
pip install openclaw requests apscheduler python-dotenv第一步：Pangolinfo API 价格采集模块
# price_collector.py
import requests
import json
from datetime import datetime
from typing import Optional
PANGOLINFO_API_URL = "https://api.pangolinfo.com/v1/amazon/product"
def fetch_asin_prices(asin: str, api_key: str, marketplace: str = "US") -> Optional[dict]:
"""
通过 Pangolinfo Scrape API 采集指定 ASIN 的完整价格数据
Args:
asin: 目标 ASIN 编号
api_key: Pangolinfo API Key
marketplace: 站点（默认美国站）
Returns:
包含 Buybox 价格和所有 Offer 价格的结构化数据
"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"asin": asin,
"marketplace": marketplace,
"parse": True, # 返回结构化 JSON，非原始 HTML
"include_offers": True, # 获取所有卖家报价
"include_buybox": True # 获取 Buybox 归属和价格
}
try:
resp = requests.post(PANGOLINFO_API_URL, json=payload, headers=headers, timeout=30)
resp.raise_for_status()
data = resp.json()
# 提取价格监控所需的核心字段
return {
"asin": asin,
"collected_at": datetime.utcnow().isoformat() + "Z",
"marketplace": marketplace,
"buybox_price": data.get("buybox", {}).get("price"),
"buybox_seller": data.get("buybox", {}).get("seller_name"),
"buybox_seller_id": data.get("buybox", {}).get("seller_id"),
"all_offers": [
{
"seller_id": o.get("seller_id"),
"seller_name": o.get("seller_name"),
"price": o.get("price"),
"is_fba": o.get("is_fba"),
"seller_rating": o.get("seller_rating")
}
for o in data.get("offers", [])
],
"title": data.get("title", ""),
"in_stock": data.get("availability") == "In Stock"
}
except requests.RequestException as e:
print(f"[ERROR] 采集 ASIN {asin} 失败: {e}")
return None
def batch_collect(asin_list: list, api_key: str, marketplace: str = "US") -> list:
"""批量采集多个 ASIN 的当前价格快照"""
results = []
for asin in asin_list:
result = fetch_asin_prices(asin, api_key, marketplace)
if result:
results.append(result)
print(f"[INFO] {asin}: Buybox 价格 ${result['buybox_price']} | 卖家: {result['buybox_seller']}")
return results
第二步：OpenClaw Agent 价格变化分析模块
# price_analyzer.py
import json
from openclaw import Agent, Task
from typing import Optional
class PriceChangeAnalyzer:
"""
基于 OpenClaw Agent 的价格变化分析器
负责比对新旧价格快照，判断是否触发告警，并生成 AI 分析摘要
"""
def __init__(self, alert_threshold_pct: float = 5.0, my_seller_id: str = ""):
"""
Args:
alert_threshold_pct: 触发告警的降价百分比阈值（默认5%）
my_seller_id: 自己的卖家账号ID（用于排除自身价格变化）
"""
self.threshold = alert_threshold_pct
self.my_seller_id = my_seller_id
self.price_history = {} # 存储每个 ASIN 的历史价格
# 初始化 OpenClaw Agent（使用配置的 LLM 进行分析）
self.agent = Agent(
name="PriceMonitorAgent",
role="亚马逊竞品价格分析专家",
goal="分析竞品价格变化，识别潜在威胁并给出应对建议",
backstory="你是一位资深的亚马逊运营专家，深谙价格战策略和竞品分析方法"
)
def analyze(self, current_snapshot: dict) -> Optional[dict]:
"""
分析当前快照与历史数据的差异，生成告警事件（如有）
Returns:
告警事件字典，如无异常则返回 None
"""
asin = current_snapshot["asin"]
current_price = current_snapshot.get("buybox_price")
if current_price is None:
return None
# 首次采集，建立基准值
if asin not in self.price_history:
self.price_history[asin] = {
"price": current_price,
"seller": current_snapshot.get("buybox_seller"),
"collected_at": current_snapshot["collected_at"],
"change_log": []
}
print(f"[INIT] {asin}: 基准价格设定为 ${current_price}")
return None
previous = self.price_history[asin]
prev_price = previous["price"]
# 计算价格变化
if prev_price and prev_price > 0:
change_pct = (current_price - prev_price) / prev_price * 100
else:
return None
# 只在竞品降价超过阈值时触发
is_competitor = current_snapshot.get("buybox_seller_id") != self.my_seller_id
if change_pct <= -self.threshold and is_competitor:
# 调用 OpenClaw Agent 生成分析摘要
analysis_task = Task(
description=f"""
分析以下亚马逊竞品调价事件，给出简短的背景分析和应对建议（100字以内）：
ASIN: {asin}
商品名称: {current_snapshot.get('title', '未知')[:50]}
竞品卖家: {current_snapshot.get('buybox_seller', '未知')}
原价: ${prev_price:.2f}
现价: ${current_price:.2f}
降幅: {abs(change_pct):.1f}%
历史调价记录: {len(previous.get('change_log', []))} 次
""",
expected_output="一段简洁的中文分析，包含威胁等级评估和建议应对动作"
)
ai_analysis = self.agent.execute_task(analysis_task)
# 更新历史记录
self.price_history[asin]["change_log"].append({
"from": prev_price,
"to": current_price,
"change_pct": change_pct,
"timestamp": current_snapshot["collected_at"]
})
self.price_history[asin]["price"] = current_price
return {
"type": "PRICE_DROP",
"priority": "HIGH" if abs(change_pct) >= 10 else "MEDIUM",
"asin": asin,
"title": current_snapshot.get("title", "")[:60],
"competitor_seller": current_snapshot.get("buybox_seller"),
"price_from": prev_price,
"price_to": current_price,
"change_pct": change_pct,
"ai_analysis": ai_analysis,
"detected_at": current_snapshot["collected_at"]
}
# 价格回升也记录，但不告警
self.price_history[asin]["price"] = current_price
return None
第三步：飞书与Slack双渠道告警配置
# alert_dispatcher.py
import requests
import json
class AlertDispatcher:
"""负责将告警事件格式化并推送到飞书和/或Slack"""
def __init__(self, feishu_webhook: str = "", slack_webhook: str = ""):
self.feishu_webhook = feishu_webhook
self.slack_webhook = slack_webhook
def send_feishu_alert(self, event: dict) -> bool:
"""推送飞书消息卡片"""
if not self.feishu_webhook:
return False
priority_color = "red" if event["priority"] == "HIGH" else "orange"
change_pct = event["change_pct"]
card = {
"msg_type": "interactive",
"card": {
"header": {
"title": {
"content": f"🚨 竞品降价预警 | {event['asin']}",
"tag": "plain_text"
},
"template": priority_color
},
"elements": [
{
"tag": "div",
"text": {
"content": (
f"**商品**：{event['title']}\n"
f"**竞品卖家**：{event['competitor_seller']}\n"
f"**价格变化**：${event['price_from']:.2f} → **${event['price_to']:.2f}**\n"
f"**降幅**：{abs(change_pct):.1f}%\n"
f"**AI分析**：{event['ai_analysis']}"
),
"tag": "larkmd"
}
},
{
"tag": "action",
"actions": [
{
"tag": "button",
"text": {"content": "查看 ASIN 价格走势", "tag": "plain_text"},
"type": "primary",
"url": f"https://tool.pangolinfo.com/tracking/{event['asin']}"
},
{
"tag": "button",
"text": {"content": "查看竞品 Listing", "tag": "plain_text"},
"type": "default",
"url": f"https://www.amazon.com/dp/{event['asin']}"
}
]
}
]
}
}
resp = requests.post(self.feishu_webhook, json=card, timeout=10)
return resp.status_code == 200
def send_slack_alert(self, event: dict) -> bool:
"""推送 Slack 消息（Block Kit格式）"""
if not self.slack_webhook:
return False
change_pct = event["change_pct"]
emoji = "🔴" if event["priority"] == "HIGH" else "🟡"
payload = {
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{emoji} Amazon Competitor Price Drop Alert"
}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*ASIN:*\n{event['asin']}"},
{"type": "mrkdwn", "text": f"*Competitor:*\n{event['competitor_seller']}"},
{"type": "mrkdwn", "text": f"*Price Change:*\n${event['price_from']:.2f} → ${event['price_to']:.2f}"},
{"type": "mrkdwn", "text": f"*Drop:*\n{abs(change_pct):.1f}%"}
]
},
{
"type": "section",
"text": {"type": "mrkdwn", "text": f"*AI Analysis:*\n{event['ai_analysis']}"}
},
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {"type": "plain_text", "text": "View Price History"},
"url": f"https://tool.pangolinfo.com/tracking/{event['asin']}"
}
]
}
]
}
resp = requests.post(self.slack_webhook, json=payload, timeout=10)
return resp.status_code == 200
def dispatch(self, event: dict):
"""同时推送到飞书和Slack"""
feishu_ok = self.send_feishu_alert(event)
slack_ok = self.send_slack_alert(event)
print(f"[ALERT] {event['asin']} 降价 {abs(event['change_pct']):.1f}% | "
f"飞书: {'✅' if feishu_ok else '❌'} | "
f"Slack: {'✅' if slack_ok else '❌'}")
第四步：定时任务主程序（10分钟轮询）
# main.py
import os
from dotenv import load_dotenv
from apscheduler.schedulers.blocking import BlockingScheduler
from price_collector import batch_collect
from price_analyzer import PriceChangeAnalyzer
from alert_dispatcher import AlertDispatcher
load_dotenv()
# ========== 配置区域 ==========
PANGOLINFO_API_KEY = os.getenv("PANGOLINFO_API_KEY")
MY_SELLER_ID = os.getenv("MY_SELLER_ID", "")
# 要监控的竞品 ASIN 列表
MONITOR_ASINS = [
"B0D6BFMNN5",
"B08K2S9X7Q",
"B09XYZ1234",
# 添加更多目标 ASIN...
]
# 告警配置
ALERT_THRESHOLD_PCT = 5.0 # 降价超过5%触发告警
FEISHU_WEBHOOK = os.getenv("FEISHU_WEBHOOK_URL", "")
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL", "")
POLL_INTERVAL_MINUTES = 10 # 每10分钟采集一次
# ==============================
# 初始化各模块
analyzer = PriceChangeAnalyzer(
alert_threshold_pct=ALERT_THRESHOLD_PCT,
my_seller_id=MY_SELLER_ID
)
dispatcher = AlertDispatcher(
feishu_webhook=FEISHU_WEBHOOK,
slack_webhook=SLACK_WEBHOOK
)
def run_monitoring_cycle():
"""执行一轮完整的采集→分析→告警循环"""
print(f"\n{'='*50}")
print(f"[CYCLE] 开始采集 {len(MONITOR_ASINS)} 个 ASIN...")
# 采集当前价格快照
snapshots = batch_collect(MONITOR_ASINS, PANGOLINFO_API_KEY)
# 逐个分析，触发告警
alert_count = 0
for snapshot in snapshots:
alert_event = analyzer.analyze(snapshot)
if alert_event:
dispatcher.dispatch(alert_event)
alert_count += 1
print(f"[CYCLE] 完成。本轮采集 {len(snapshots)} 个，"
f"触发告警 {alert_count} 个。")
if __name__ == "__main__":
print("🚀 亚马逊竞品调价监控系统启动")
print(f"监控 ASIN 数量: {len(MONITOR_ASINS)}")
print(f"采集间隔: 每 {POLL_INTERVAL_MINUTES} 分钟")
print(f"告警阈值: 降价超过 {ALERT_THRESHOLD_PCT}%\n")
# 启动时先执行一次
run_monitoring_cycle()
# 设置定时任务
scheduler = BlockingScheduler()
scheduler.add_job(
run_monitoring_cycle,
'interval',
minutes=POLL_INTERVAL_MINUTES,
id='price_monitor'
)
try:
scheduler.start()
except KeyboardInterrupt:
print("\n监控系统已停止。")
在项目根目录创建 .env 文件，填入你的凭证：
PANGOLINFO_API_KEY=your_api_key_here
MY_SELLER_ID=your_amazon_seller_id
FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/your_token
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/your/path
生产部署：让动态定价追踪系统全天候运行
本地测试通过之后，建议采用以下方式将脚本部署到生产环境，确保7×24小时不间断运行。
方案一：服务器 + Systemd（推荐自建服务器）。在 Linux 服务器上创建 systemd service 文件，设置 Restart=always，脚本会在崩溃时自动重启，并在服务器重启时自动恢复运行。配合 journald 日志，所有告警和采集记录都有完整的审计轨迹。
方案二：GitHub Actions（无服务器，零成本）。利用 GitHub Actions 的 schedule 触发器，每10分钟触发一次 workflow，执行价格采集脚本。优点是完全免费且无需维护服务器；缺点是 GitHub Actions 的最小调度间隔是5分钟，且可能有小幅延迟。
方案三：云函数（AWS Lambda / 阿里云函数计算）。将采集逻辑打包为无服务器函数，通过 EventBridge 或云平台的定时触发器每10分钟调用一次。适合已有云基础设施的团队，几乎零运维成本，按调用次数计费，对于监控10-20个ASIN的场景月费用极低。
无论选择哪种部署方式，建议配套以下监控措施：设置脚本连续失败3次后发送紧急通知（防止采集静默失败）；定期将价格历史数据导出至数据库或电子表格，为后续趋势分析提供数据积累；在旺季（Q4、会员日、黑五）期间将采集频率提高到5分钟一次，确保最关键时期不遗漏任何价格动态。
如果你的团队不想维护自建脚本，AMZ Data Tracker 提供了同等能力的无代码配置界面，同样支持飞书告警和小时级数据采集，适合没有专职工程资源的运营团队。
把亚马逊竞品调价监控变成你的定价护城河
价格战的胜负，最终取决于谁掌握的信息更实时。当你能在竞品降价10分钟内就收到告警，你的决策窗口比对方慢半拍降幅就小了一个量级。亚马逊竞品调价监控不是一个”好有更好”的功能，而是在高竞争类目中保持价格主动权的基础设施。
这套 OpenClaw + Pangolinfo API 的方案给你的是：可以真正跑起来的代码、可以订制的告警逻辑、以及 AI 辅助分析带来的决策加速。代码是开放的、可扩展的，你可以在它的基础上加入更多维度的数据（评论变化、BSR波动、广告位变化），构建一套真正属于你的实时竞品智能系统。
开始行动：把你最核心的3-5个竞品ASIN加进监控列表，跑一周，看看你会发现多少曾经被你错过的竞品价格动作。那些被你错过的信号，正是你的竞对手一直在利用的优势。
📌 没有工程团队？直接试用 AMZ Data Tracker，零代码配置竞品价格监控和飞书告警，10分钟上线。需要 API 直接集成？访问 Pangolinfo Scrape API 获取你的 API Key，开始搭建属于你自己的调价预警系统。
