前言
做电商运营、选品、竞品调研的时候，相信很多朋友都遇到过这些痛点：
需要持续盯多款竞品，人工反复打开页面查看价格、库存、促销活动，耗费大量时间；
大促期间竞品调价、库存变动节奏很快，人很难做到 7×24 小时实时捕捉变化；
拿到一堆零散的商品信息，很难沉淀历史数据，做价格趋势、SKU 规格对比分析；
直接从网页解析页面，容易遇到访问限制、页面改版导致脚本直接失效，维护成本很高。
之前我尝试过写页面解析脚本，但是平台页面结构经常调整，隔一段时间代码就失效，调试反爬相关逻辑消耗大量精力。后来接触到 Open‑Claw，它可以直接输出结构化标准化的商品全量资料，不需要再折腾页面解析逻辑，我们只需要专注做监控逻辑、数据存储、业务分析，快速搭一套属于自己的商品监测工具。
本篇文章面向开发者、电商从业者，完整讲解从获取商品信息、变化监测、数据存储，再到简单数据分析的整套落地流程，代码可以直接修改配置后运行。
说明：本教程仅做技术学习演示，使用时请遵守对应平台使用规范，合理控制访问频次。
整体实现思路
整套工具分成 4 个核心模块：
商品信息获取模块：输入商品编号，返回结构化完整商品资料，包含标题、售价、原价、库存、规格、店铺信息、图片等内容；
变化检测模块：对比上一轮保存的数据，识别价格涨跌、库存变动、规格变更，输出告警提示；
本地持久化模块：把每一轮采集的数据保存到本地 JSON/CSV，沉淀历史时序数据；
简易数据分析模块：读取历史记录，做价格波动统计、SKU 规格整理，输出简单统计结果。
环境准备
依赖安装，执行下面命令：
完整可运行代码实现
# coding:utf-8
import requests
import json
import time
import os
import pandas as pd
from datetime import datetime
# ========== 配置区，请修改为你自己的凭证和监控商品 ==========
CLAW_KEY = "填入你的key"
CLAW_SECRET = "填入你的secret"
# 需要监控的商品编号列表
MONITOR_GOODS = ["10335871600", "10057467958584"]
# 循环检测间隔，单位秒，不建议设置过小，示例5分钟轮询一次
MONITOR_INTERVAL = 300
# 降价告警阈值，0.1代表价格下跌超过10%触发提醒
PRICE_DROP_ALERT = 0.1
# 历史数据存储文件
HISTORY_FILE = "goods_history.json"
CSV_EXPORT = "goods_record.csv"
# ==========================================================
def load_last_cache():
"""加载上一轮缓存的商品数据"""
if os.path.exists(HISTORY_FILE):
with open(HISTORY_FILE, "r", encoding="utf‑8") as f:
return json.load(f)
return {}
def save_cache(data):
"""保存当前采集结果到本地缓存文件"""
with open(HISTORY_FILE, "w", encoding="utf‑8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def fetch_goods_data(goods_id):
"""
获取单个商品结构化数据
:param goods_id: 商品数字编号
:return: 解析完成的商品字典，失败返回None
"""
base_url = "https://api‑gw.onebound.cn/jd/item_get_pro"
params = {
"key": CLAW_KEY,
"secret": CLAW_SECRET,
"num_iid": goods_id,
"result_type": "json",
"cache": "yes",
"lang": "cn"
}
headers = {
"Accept‑Encoding": "gzip",
"Connection": "close"
}
try:
resp = requests.get(base_url, params=params, headers=headers, timeout=20)
resp_json = resp.json()
# 判断返回状态，非0000代表异常
if resp_json.get("error_code") != "0000":
err_msg = resp_json.get("reason", "未知异常")
print(f"【获取失败】商品{goods_id}，原因：{err_msg}")
return None
item_data = resp_json.get("item", {})
if not item_data:
print(f"【获取失败】商品{goods_id}返回内容为空")
return None
# 提取核心业务字段，做标准化整理
parse_result = {
"goods_id": item_data.get("num_iid"),
"title": item_data.get("title", ""),
"price": float(item_data.get("price", 0)) if item_data.get("price") else 0.0,
"original_price": float(item_data.get("orginal_price", 0)) if item_data.get("orginal_price") else 0.0,
"stock": item_data.get("num", 0),
"shop_name": item_data.get("seller_info", {}).get("shop_name", ""),
"detail_url": item_data.get("detail_url", ""),
"collect_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
"skus": item_data.get("skus", {}).get("sku", [])
}
return parse_result
except Exception as e:
print(f"【请求异常】商品{goods_id}, 错误：{str(e)}")
return None
def detect_change(old_data, new_data):
"""对比新旧两份数据，检测价格库存变化，输出告警信息"""
alert_list = []
if not old_data:
alert_list.append("首次采集该商品，无历史对比")
return alert_list
old_price = old_data["price"]
new_price = new_data["price"]
old_stock = old_data["stock"]
new_stock = new_data["stock"]
# 价格变动检测
if abs(new_price - old_price) > 0.01:
change_rate = (new_price - old_price) / old_price if old_price > 0 else 0
if new_price < old_price and abs(change_rate) >= PRICE_DROP_ALERT:
alert_list.append(f"⚠️降价告警！{old_price} → {new_price}，降幅{abs(change_rate)*100:.1f}%")
elif new_price > old_price:
alert_list.append(f"📈涨价提醒！{old_price} → {new_price}")
else:
alert_list.append(f"💡价格微调：{old_price} → {new_price}")
# 库存变动检测
if new_stock != old_stock:
if new_stock == 0:
alert_list.append(f"🚨库存告急，商品变为无货！原库存：{old_stock}")
elif old_stock == 0 and new_stock > 0:
alert_list.append(f"✅商品补货成功！当前库存 {new_stock}")
else:
alert_list.append(f"📦库存变动 {old_stock} → {new_stock}")
return alert_list
def export_to_csv(all_records):
"""把采集记录导出csv，方便后续做表格分析"""
rows = []
for goods_id, records in all_records.items():
for rec in records:
rows.append({
"采集时间": rec["collect_time"],
"商品ID": rec["goods_id"],
"商品标题": rec["title"],
"现价": rec["price"],
"原价": rec["original_price"],
"库存": rec["stock"],
"店铺名称": rec["shop_name"]
})
df = pd.DataFrame(rows)
df.to_csv(CSV_EXPORT, index=False, encoding="utf‑8‑sig")
def simple_analysis():
"""简易数据分析示例：读取csv，统计价格区间、输出基础统计信息"""
if not os.path.exists(CSV_EXPORT):
print("暂无导出记录，跳过分析")
return
df = pd.read_csv(CSV_EXPORT, encoding="utf‑8‑sig")
print("\n=====简易数据分析报告=====")
for gid in df["商品ID"].unique():
sub_df = df[df["商品ID"] == gid]
print(f"\n商品ID {gid}：{sub_df.iloc[0]['商品标题'][:30]}...")
print(f" 采集记录条数：{len(sub_df)}")
print(f" 历史最低价格：{sub_df['现价'].min()}")
print(f" 历史最高价格：{sub_df['现价'].max()}")
print(f" 平均价格：{round(sub_df['现价'].mean(),2)}")
def main_loop():
print("====商品监控程序启动====")
cache_data = load_last_cache()
history_records = {}
# 初始化历史记录容器
for gid in MONITOR_GOODS:
if gid not in history_records:
history_records[gid] = []
while True:
now_time = datetime.now().strftime("%Y‑%m‑%d %H:%M:%S")
print(f"\n【本轮采集时间】{now_time}")
for goods_id in MONITOR_GOODS:
new_item = fetch_goods_data(goods_id)
if new_item is None:
continue
# 保存到全量历史，用于导出报表
history_records[goods_id].append(new_item)
old_item = cache_data.get(goods_id)
alerts = detect_change(old_item, new_item)
print(f"\n>>商品：{new_item['title'][:40]}")
for msg in alerts:
print(f" {msg}")
# 更新缓存为当前最新数据
cache_data[goods_id] = new_item
# 写入本地缓存文件
save_cache(cache_data)
export_to_csv(history_records)
simple_analysis()
print(f"\n本轮结束，等待 {MONITOR_INTERVAL} 秒进入下一轮")
time.sleep(MONITOR_INTERVAL)
if __name__ == "__main__":
try:
main_loop()
except KeyboardInterrupt:
print("\n程序手动停止，已保存当前数据文件")代码核心功能讲解
数据获取函数 fetch_goods_data 不需要解析网页 HTML，直接拿到整理好的结构化结果，提取标题、价格、原价、库存、SKU 规格、店铺、链接等关键字段。返回异常状态做捕获，商品下架、编号错误、权限问题都会打印提示。
变化检测函数 detect_change 拿上一轮缓存的数据和本轮对比，自动识别涨价、降价（设置阈值，小幅度波动不告警）、缺货、补货，输出告警文本。可以后续扩展，对接企业微信、钉钉消息推送，实现真正消息提醒。
持久化存储
简易数据分析 simple_analysis 读取 CSV 历史，对每个商品输出历史最高价、最低价、平均成交价，快速看商品的价格波动区间。
拓展优化方向
1. 消息告警推送
现在告警只打印控制台，可以增加钉钉 / 企业微信 webhook，当检测到降价、缺货的时候，直接推送消息到工作群，不用盯着控制台运行。
2. 数据库存储
小体量用 CSV 完全够用；如果监控几十上百款商品，建议换成 SQLite、MySQL，方便做复杂查询，比如筛选某时间段所有降价商品。
3.SKU 粒度监控
当前示例取商品整体信息，返回结果自带完整 SKU 数组，可以改造代码，针对每一个子规格单独记录价格库存，适合多规格复杂商品的竞品分析。
4. 可视化图表
结合 matplotlib、plotly，读取 CSV 历史数据，自动绘制每个商品价格趋势折线图，直观看到大促前后价格变动轨迹。
5. 任务调度分离
长时间运行可以把采集逻辑做成后台定时任务，不需要一直开着脚本；Linux 环境下可配置 systemd 服务或者 crontab 定时执行。
踩坑记录与注意事项
轮询间隔不要设置过小，短时间高频访问会触发限流，建议至少间隔 3‑5 分钟以上；
部分商品多规格场景，整体展示价格和子 SKU 实际售价存在差异，做精细分析时，优先读取 SKU 数组内部对应规格的价格；
部分自营、第三方店铺，店铺相关字段不一定全部返回，代码已经做了空值兼容；
遇到商品下架，返回结果会提示商品未找到，脚本会捕获异常不会直接崩溃；
凭证信息不要硬编码提交到公开代码仓库，生产环境建议放到环境变量读取。
业务场景可以怎么用
竞品监控：跟踪竞品多款商品，记录调价节奏、大促降价幅度，给自己的定价、促销活动做参考；
采购备货参考：持续看竞品库存变动，辅助判断市场热度，辅助备货决策；
个人好物蹲价：想要买的商品，设置降价阈值，一旦低于心理价位立刻收到提醒；
选品分析：批量收集一批候选商品历史价格、规格，筛选性价比高的目标单品。
写在最后
这套方案最大的优势，是不用投入精力维护页面解析逻辑，把重心放在业务逻辑上。Open‑Claw 帮我们处理了页面各种复杂解析工作，输出规整结构化数据，开发者可以快速把精力投入监控告警、历史统计、数据分析业务。
上面这份代码可以直接跑通，只需要填入自己的凭证，填写需要监控的商品编号，就可以立刻拥有一套属于自己的商品监测工具。后续可以基于这个基础，按需扩展推送、数据库、可视化等功能。
