适用场景：竞品跟踪、价格波动监控、库存变动记录、历史行情统计；Python 完整可运行示例，适合个人开发者、电商运营、数据分析同学参考。
前言
做电商运营或者竞品调研的时候，我们经常需要持续关注一批商品：价格有没有下调、库存是否清零、不同规格套餐的售价变化、店铺基础信息更新。如果依靠人工反复打开网页复制记录，不仅耗费大量时间，还很容易错过关键变动，历史数据也很难留存用于复盘分析。
很多同学会尝试直接解析网页页面，但平台页面结构经常改版，CSS 选择器隔一段时间就失效，调试成本非常高。Open‑Claw 这套工具集合，帮我们屏蔽了页面解析的繁琐工作，直接输出结构化的商品原始资料，我们只需要专注业务逻辑：定时拉取、存储历史记录、做对比分析，实现一套轻量化的商品监控系统。
本文会完整讲解整个实现思路，给出可直接调试的 Python 代码，实现：商品基础信息获取、定时循环监控、数据落盘保存、价格波动识别，简单的数据分析统计。
整体实现思路
整套程序分为 4 个核心环节：
资料拉取：输入商品编号，拿到标题、现价、原价、库存、多规格套餐、店铺名称、主图等完整结构化内容；
持久化存储：把每一次采集的结果保存，形成时间序列历史记录；
变动检测：对比上一轮记录，识别价格涨跌、库存变动；
简单分析：读取历史文件，计算最高价、最低价、价格变动频次，输出统计结果。
注意：使用前需要提前准备好你的访问凭证，包含两组密钥信息。
环境准备
依赖安装，执行下面命令：
完整代码实现
新建goods_monitor.py，复制下面代码，修改配置区域的凭证、待监控商品编号、监控周期即可运行。
# -*- coding:utf-8 -*-
import requests
import json
import time
import pandas as pd
import schedule
from datetime import datetime
# ======================【配置区，请修改这里】======================
# 你的两组访问凭证
ACCESS_KEY = "替换为你的key"
ACCESS_SECRET = "替换为你的secret"
# 需要监控的商品编号，可以填写多个
MONITOR_GOODS_ID_LIST = ["10335871600"]
# 监控间隔，单位秒，示例1800=30分钟跑一次
CHECK_INTERVAL = 1800
# 历史数据保存文件
HISTORY_CSV = "./goods_history.csv"
# =================================================================
def get_goods_raw_data(goods_id):
"""
获取商品原始结构化资料
goods_id:商品编号
返回解析后的字典，失败返回None
"""
try:
resp = requests.get(
url="https://api‑gw.onebound.cn/jd/item_get_pro",
params={
"key": ACCESS_KEY,
"secret": ACCESS_SECRET,
"num_iid": goods_id,
"result_type": "json"
},
headers={"Accept‑Encoding": "gzip", "Connection": "close"},
timeout=25
)
result = resp.json()
# 判断返回状态
if result.get("error_code") != "0000":
err_msg = result.get("reason", "未知异常")
print(f"【商品{goods_id}获取失败】{err_msg}")
return None
item_data = result.get("item")
if not item_data:
print(f"【商品{goods_id}返回内容为空】")
return None
return item_data
except Exception as e:
print(f"网络或解析异常 goods_id={goods_id}, error:{str(e)}")
return None
def parse_item_info(raw_item):
"""
从原始资料提取我们关心的业务字段
"""
sku_list = []
sku_raw = raw_item.get("skus", {}).get("sku", [])
for sku in sku_raw:
sku_list.append({
"sku_id": sku.get("sku_id"),
"sku_name": sku.get("properties_name"),
"price": float(sku.get("price", 0)),
"origin_price": float(sku.get("orginal_price", 0)),
"stock": int(sku.get("quantity", 0))
})
output = {
"capture_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", ""),
"shop_name": raw_item.get("seller_info", {}).get("shop_name", ""),
"current_price": float(raw_item.get("price", 0)),
"origin_price": float(raw_item.get("orginal_price", 0)),
"stock_total": raw_item.get("num", 0),
"detail_url": raw_item.get("detail_url", ""),
"main_image": raw_item.get("pic_url", ""),
"skus": json.dumps(sku_list, ensure_ascii=False)
}
return output
def save_to_csv(row_data):
"""追加写入历史记录csv"""
df_new = pd.DataFrame([row_data])
try:
df_old = pd.read_csv(HISTORY_CSV)
df_all = pd.concat([df_old, df_new], ignore_index=True)
except FileNotFoundError:
df_all = df_new
df_all.to_csv(HISTORY_CSV, index=False, encoding="utf‑8‑sig")
def detect_change(last_record, curr_record):
"""简单变动对比，识别价格、库存变化"""
msg_list = []
if last_record is None:
msg_list.append("首次采集，无历史对比")
return msg_list
last_price = float(last_record.get("current_price", 0))
curr_price = curr_record.get("current_price")
last_stock = int(last_record.get("stock_total", 0))
curr_stock = curr_record.get("stock_total")
if curr_price != last_price:
diff = curr_price - last_price
if diff < 0:
msg_list.append(f"⚠️价格下降：{last_price} → {curr_price}，降幅{abs(diff)}")
else:
msg_list.append(f"⚠️价格上涨：{last_price} → {curr_price}，涨幅{diff}")
if curr_stock != last_stock:
msg_list.append(f"⚠️库存变动：{last_stock} → {curr_stock}")
if len(msg_list) == 0:
msg_list.append("✅本次采集无明显变动")
return msg_list
def monitor_task():
"""单次监控任务，循环处理所有待监控商品"""
print(f"\n=========开始巡检 {datetime.now().strftime('%Y‑%m‑%d %H:%M:%S')} =========")
for goods_id in MONITOR_GOODS_ID_LIST:
raw = get_goods_raw_data(goods_id)
if raw is None:
continue
curr_info = parse_item_info(raw)
# 读取上一条同商品历史记录用于对比
last_one = None
try:
df_history = pd.read_csv(HISTORY_CSV)
goods_history = df_history[df_history["goods_id"] == str(goods_id)]
if not goods_history.empty:
last_one = goods_history.iloc[-1].to_dict()
except FileNotFoundError:
pass
change_msgs = detect_change(last_one, curr_info)
print(f"\n商品【{curr_info['title']}】")
for m in change_msgs:
print(m)
save_to_csv(curr_info)
print("=========本轮巡检结束=========\n")
def analysis_history():
"""读取本地历史文件做简单数据分析，输出统计结果"""
try:
df = pd.read_csv(HISTORY_CSV)
if df.empty:
print("暂无历史分析数据")
return
print("\n====历史数据分析报告====")
for gid in df["goods_id"].unique():
sub_df = df[df["goods_id"] == gid]
title = sub_df.iloc[0]["title"]
max_p = sub_df["current_price"].max()
min_p = sub_df["current_price"].min()
avg_p = round(sub_df["current_price"].mean(), 2)
count_row = len(sub_df)
print(f"商品：{title}")
print(f" 采样次数：{count_row} 次")
print(f" 历史最高：{max_p}，历史最低：{min_p}，平均价：{avg_p}")
print("-"*40)
except FileNotFoundError:
print("还没有生成历史记录文件，请先运行监控采集。")
if __name__ == "__main__":
# 立即执行一次
monitor_task()
# 设置定时调度
schedule.every(CHECK_INTERVAL).seconds.do(monitor_task)
print(f"已启动定时监控，每 {CHECK_INTERVAL} 秒执行一轮，按Ctrl+C终止程序")
print("输入 a 回车可以执行历史数据分析报告")
while True:
schedule.run_pending()
time.sleep(2)
# 简易交互，输入a执行分析报告
try:
import sys
if sys.stdin in select.select([sys.stdin], [], [], 0)[0]:
line = sys.stdin.readline()
if line.strip() == "a":
analysis_history()
except Exception:
pass补充说明：运行后，程序会按照设定周期自动拉取商品资料，每一轮全部写入goods_history.csv；控制台输入字母a回车，即可输出历史统计报告。
数据字段说明
保存到 csv 里面会包含这些核心信息：
依靠这份历史表格，我们后续还可以拓展：
使用 matplotlib 绘制价格走势折线图；
接入消息推送，发生价格变动的时候发送通知；
批量导入上百个商品编号做批量竞品巡检；
将 csv 迁移至 sqlite/mysql 数据库，支撑更大体量的数据。
踩坑记录与注意事项
凭证密钥保管：两组密钥不要硬编码直接上传到公开代码仓库，正式项目建议放到环境变量或者配置文件。
采集频次控制：不要设置过小的轮询间隔，频繁请求会触发访问限制，建议至少 10 分钟以上一轮。
规格价格注意：部分场景下返回的主条目价格不一定和子规格完全匹配，业务分析优先读取 skus 数组内各个套餐价格。
异常容错：代码里面已经增加捕获异常，遇到网络波动、商品下架不会直接让整个程序崩溃，控制台会打印错误提示。
商品下架情况：当商品已经被平台下架，会返回对应提示，脚本会打印日志，不会生成无效记录。
后续拓展方向
增加消息推送：企业微信、钉钉机器人，一旦监控到降价、库存变化直接推送提醒；
可视化：把历史 csv 数据绘制价格走势图，直观看到一段时间的价格波动；
持久化数据库：数据量大之后替换 csv，改用 sqlite 数据库存储；
多平台兼容：这套工具也可以拓展其他平台的商品资料获取，复用监控分析逻辑。
写在最后
对于运营、数据分析同学，不需要深度钻研页面逆向、页面反爬对抗，借助 Open‑Claw 直接拿到干净结构化的原始资料，把精力放在业务逻辑：定时巡检、历史留存、变动告警、行情分析，快速搭建一套属于自己的竞品监控工具。
