前言
做电商运营、竞品分析或者个人薅优惠的时候，很多人都经历过这样的困境： 想要跟踪一批商品的价格变动、库存状态、规格信息，只能每天手动打开网页复制记录。商品一多，耗时耗力还容易出错；等到发现竞品降价、库存清空的时候，已经错过了最佳决策时机。
直接写脚本解析网页源码，又会遇到平台的访问限制、页面改版导致解析规则直接失效，经常今天跑通明天就报错。
最近我在实践中尝试用 OpenClaw 这套工具方案，快速搭建了一套轻量的商品监控 + 数据分析流程。不用纠缠复杂页面结构，就能稳定拿到商品标题、售价、原价、库存、规格、店铺信息等完整资料，配合简单 Python 脚本，实现定时监控、数据落盘、简单统计分析。本篇把完整实操流程分享出来。
⚠️提示：本文仅用于学习公开网页数据处理技术，请遵守平台规则与相关法律法规，仅用于个人学习分析，请勿高频大规模采集。
整体思路
整套流程分为 4 个环节：
准备工作：拿到访问凭证，确定要监控的商品编号；
数据拉取：借助工具能力获取商品结构化原始数据；
数据清洗存储：提取需要的字段，存入本地文件，留存历史记录；
监控告警与数据分析：定时循环执行，对比历史数据发现价格 / 库存变动，做简单统计。
商品编号就是商品网页链接里那一串数字，例如商品地址https://item.jd.com/10335871600.html，10335871600就是商品编号。
环境准备
本地 Python 环境，只需要安装网络请求库，执行下面命令：
完整实战代码
注意：代码中your_key、your_secret替换成你自己申请的凭证。
# coding:utf-8
import requests
import pandas as pd
import time
from datetime import datetime
class GoodsMonitor:
def __init__(self, key, secret):
self.key = key
self.secret = secret
# 目标服务地址
self.base_url = "https://api‑gw.onebound.cn/jd/item_get_pro/"
# 保存历史数据的文件
self.save_file = "goods_monitor.csv"
def get_goods_raw_data(self, goods_id):
"""获取单个商品原始结构化数据"""
params = {
"key": self.key,
"secret": self.secret,
"num_iid": goods_id,
"cache": "no", # 不读取缓存，获取最新状态
"result_type": "json"
}
headers = {
"Accept‑Encoding": "gzip",
"Connection": "close"
}
try:
resp = requests.get(self.base_url, params=params, headers=headers, timeout=20)
resp_json = resp.json()
except Exception as e:
print(f"网络异常：{e}")
return None
# 判断是否获取成功
if resp_json.get("error_code") != "0000":
print(f"获取失败，原因：{resp_json.get('reason','未知')}")
return None
return resp_json.get("item")
def parse_goods_info(self, raw_item):
"""从原始数据提取我们关心的核心字段"""
if not raw_item:
return None
# 多规格SKU处理，这里拿第一个规格作为示例
sku_list = raw_item.get("skus", {}).get("sku", [])
first_sku = sku_list[0] if sku_list else {}
info = {
"crawl_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
"goods_id": raw_item.get("num_iid"),
"title": raw_item.get("title"),
"brand": raw_item.get("brand"),
"current_price": float(raw_item.get("price", 0)) if raw_item.get("price") else 0,
"origin_price": float(raw_item.get("orginal_price", 0)) if raw_item.get("orginal_price") else 0,
"stock": raw_item.get("num", 0),
"shop_name": raw_item.get("seller_info", {}).get("shop_name", ""),
"detail_url": raw_item.get("detail_url"),
"sku_id": first_sku.get("sku_id", ""),
"sku_properties": first_sku.get("properties_name", "")
}
return info
def save_to_csv(self, item_info):
"""追加保存到csv，留存历史监控记录"""
df_new = pd.DataFrame([item_info])
try:
df_old = pd.read_csv(self.save_file)
df_all = pd.concat([df_old, df_new], ignore_index=True)
except FileNotFoundError:
df_all = df_new
df_all.to_csv(self.save_file, index=False, encoding="utf‑8‑sig")
def check_change_and_alert(self, new_info):
"""对比历史记录，检测价格、库存变动，简单告警打印"""
try:
df = pd.read_csv(self.save_file)
goods_history = df[df["goods_id"] == str(new_info["goods_id"])]
if len(goods_history) < 2:
return
# 获取上一次记录
last_record = goods_history.iloc[-2]
# 价格变动检测
if abs(new_info["current_price"] - last_record["current_price"]) > 0.01:
print(f"【变动提醒】商品 {new_info['title']} 价格发生变化！"
f"旧价：{last_record['current_price']} → 新价：{new_info['current_price']}")
# 库存变动检测
if new_info["stock"] != last_record["stock"]:
print(f"【变动提醒】商品 {new_info['title']} 库存变动！"
f"旧库存：{last_record['stock']} → 新库存：{new_info['stock']}")
except Exception:
pass
def run_single(self, goods_id):
"""单次采集一个商品完整流程"""
raw = self.get_goods_raw_data(goods_id)
parsed = self.parse_goods_info(raw)
if not parsed:
print("该商品数据获取失败")
return
print(f"采集成功：{parsed['title']}，现价：{parsed['current_price']}")
self.save_to_csv(parsed)
self.check_change_and_alert(parsed)
return parsed
def monitor_loop(self, goods_id_list, interval=1800):
"""循环监控多个商品，interval单位秒，示例30分钟跑一次"""
while True:
print(f"\n==== {datetime.now()} 开始一轮监控 ====")
for gid in goods_id_list:
self.run_single(gid)
time.sleep(2)
print(f"本轮结束，等待{interval}秒进入下一轮\n")
time.sleep(interval)
if __name__ == "__main__":
# 替换为自己的凭证
MY_KEY = "your_key"
MY_SECRET = "your_secret"
monitor = GoodsMonitor(MY_KEY, MY_SECRET)
# 需要监控的商品编号列表，可以填入多个
monitor_goods = ["10335871600"]
# 方式1：只执行一次，获取一次数据
# monitor.run_single("10335871600")
# 方式2：循环持续监控，每30分钟扫描一遍列表中的商品
monitor.monitor_loop(monitor_goods, interval=1800)代码逻辑拆解
get_goods_raw_data：向服务发起请求，拿到完整的商品原始结构化数据包。包含标题、品牌、原价现价、库存、多规格 SKU 集合、店铺资料等全部信息。
parse_goods_info：做数据清洗，从原始数据包提取业务需要字段。原始返回里包含完整规格集合，你可以修改代码遍历全部 SKU，实现多规格价格跟踪。
save_to_csv：把每一次采集结果追加写入本地 csv 文件。这样就保存了时间线，后续可以做回溯分析。
check_change_and_alert：读取历史文件，对比上一轮数据，如果价格、库存发生变化，控制台打印提醒。实际项目中可以扩展为邮件、企业微信推送消息。
monitor_loop：循环执行，定时批量扫描一批商品，完成持续监控。> 注意：间隔时间不要设置太短，避免频繁访问。
基于历史文件做简单数据分析示例
采集一段时间之后，本地goods_monitor.csv积累了时间序列数据，我们可以用 pandas 快速做简单分析。新建脚本或者在主程序追加下面代码：
运行之后，就可以直接看到商品历史价格波动，找出历史低价，筛选正在大促降价的商品。
遇到的坑和优化建议
数据延迟问题：工具返回的价格存在轻微延迟，如果需要最实时价格，关闭缓存参数（代码中已经设置cache:"no"）。
多规格商品处理：示例只取第一个 SKU，实际业务建议遍历skus下面全部的 sku 数组，把每个规格单独存一行记录，这样每个型号的价格变动都能监控到。
频率控制：循环监控间隔不要过小，建议最少 15 分钟以上，避免触发限制。
异常处理增强：生产场景可以增加日志文件，把报错信息写入日志；告警可以接入企业微信机器人、邮件，不需要盯着控制台。
商品下架：如果商品被删除，会返回对应的失败提示，代码会打印原因，业务侧可以做标记，停止监控该编号。
拓展方向
基于这套基础框架，还可以继续拓展：
批量导入 Excel 里面的商品清单，批量监控上百个商品；
将 csv 替换成 sqlite/mysql 数据库，存储更大体量数据；
使用 matplotlib 绘制价格波动曲线图，可视化展示商品价格走势；
增加竞品对标分析，批量对比多款同类商品的折扣率。
总结
借助 OpenClaw，我们不用再折腾复杂网页解析逻辑，把精力聚焦在业务层面：数据存储、变动检测、统计分析。几十行 Python 代码，就可以搭建一套属于自己的轻量商品监控系统。不管是个人蹲优惠，还是小规模竞品跟踪都很合适。
