适用场景：竞品跟踪、价格盯盘、库存监控、市场数据分析；适合商家、运营、数据分析同学直接上手跑通，代码可直接二次改造。
前言
做电商运营或者竞品调研的时候，很多人都有这样的痛点： 想要盯一批商品的价格变动、库存余量、规格变化，每次都要手动打开网页一个个复制记录。商品一多，耗时巨大，还很容易错过促销降价、库存清空的关键节点。
如果直接写网页解析，还会遇到频繁的访问限制、页面结构改版、验证码等一堆麻烦，调试就要耗费大量时间。
今天分享一套轻量的实现方案，借助 Open Claw 能力，只需要商品编号，就能拿到完整结构化商品信息；再搭配 Python 脚本，实现定时巡检、异动提醒、历史数据存储、简单统计分析，快速拥有一套属于自己的监控小工具。
环境依赖：Python3.7+，需要安装 requests、pandas、schedule。
整体实现思路
配置凭证与监控清单：填入自己的访问凭证，录入需要盯的商品编号列表。
定时拉取商品全量信息：获取标题、售价、原价、库存、多规格 SKU、店铺、图片等完整信息。
数据持久化保存：每次采集结果写入 Excel，留存历史记录，方便后续做对比分析。
异动检测逻辑：对比上一次记录，价格涨跌、库存低于阈值就在控制台输出告警提示。
简易数据分析：统计一段时间内最高价、最低价、价格波动幅度，输出简报。
完整可运行代码
# -*- coding: utf-8 -*-
import requests
import json
import time
import schedule
import pandas as pd
from datetime import datetime
# ======================【请修改这里的配置参数】======================
ACCESS_KEY = "你的访问key"
ACCESS_SECRET = "你的安全密钥"
# 需要监控的商品编号列表，可以放多个
MONITOR_GOODS = [
"10335871600",
"10057467958584"
]
# 预警阈值设置
PRICE_DROP_ALERT = 0.10 # 价格下跌超过10%提醒
STOCK_WARN_NUM = 50 # 库存低于50触发库存告警
CHECK_PERIOD = 60 * 60 # 巡检间隔，单位秒，这里1小时跑一次
EXPORT_FILE = "商品监控历史记录.xlsx"
# =================================================================
def fetch_goods_detail(goods_id):
"""获取单个商品结构化数据"""
base_url = "https://api-gw.onebound.cn/jd/item_get_pro/"
params = {
"key": ACCESS_KEY,
"secret": ACCESS_SECRET,
"num_iid": goods_id,
"cache": "no",
"result_type": "json"
}
headers = {
"Accept-Encoding": "gzip",
"Connection": "close"
}
try:
resp = requests.get(base_url, params=params, headers=headers, timeout=20)
res_json = resp.json()
error_code = res_json.get("error_code")
if error_code != "0000":
print(f"【异常】商品{goods_id}获取失败，原因：{res_json.get('reason','未知错误')}")
return None
item_data = res_json.get("item")
return item_data
except Exception as e:
print(f"网络请求异常 goods_id:{goods_id} ,err:{str(e)}")
return None
def parse_item(raw_item):
"""把原始返回数据，提取我们需要的核心字段"""
if not raw_item:
return None
data = {
"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", ""),
"shop_name": raw_item.get("seller_info", {}).get("shop_name", ""),
"current_price": float(raw_item.get("price", 0) or 0),
"origin_price": float(raw_item.get("orginal_price",0) or 0),
"stock": int(raw_item.get("num",0) or 0),
"sales": int(raw_item.get("sales",0) or 0),
"detail_url": raw_item.get("detail_url",""),
}
return data
def save_to_excel(data_list):
"""追加写入Excel保存历史记录"""
new_df = pd.DataFrame(data_list)
try:
old_df = pd.read_excel(EXPORT_FILE)
merge_df = pd.concat([old_df, new_df], ignore_index=True)
except FileNotFoundError:
merge_df = new_df
merge_df.to_excel(EXPORT_FILE, index=False)
def check_alert(new_data, history_df):
"""异动告警判断：价格、库存"""
goods_id = new_data["goods_id"]
same_goods = history_df[history_df["goods_id"] == goods_id]
if len(same_goods) <= 0:
return []
last_row = same_goods.iloc[-1]
alerts = []
# 价格下跌检测
old_price = last_row["current_price"]
new_price = new_data["current_price"]
if old_price > 0 and new_price > 0:
drop_rate = (old_price - new_price) / old_price
if drop_rate >= PRICE_DROP_ALERT:
alerts.append(f"⚠降价预警！商品【{new_data['title']}】 由{old_price} → {new_price}，跌幅{round(drop_rate*100,2)}%")
# 库存预警
if new_data["stock"] <= STOCK_WARN_NUM:
alerts.append(f"⚠库存预警！商品【{new_data['title']}】 当前库存仅剩 {new_data['stock']}")
return alerts
def simple_analysis(goods_id):
"""简单数据分析：读取本地历史，输出最高价、最低价、均值"""
try:
df = pd.read_excel(EXPORT_FILE)
target = df[df["goods_id"] == goods_id]
if len(target) == 0:
return None
info = {
"goods_id": goods_id,
"title": target.iloc[-1]["title"],
"max_price": round(target["current_price"].max(),2),
"min_price": round(target["current_price"].min(),2),
"avg_price": round(target["current_price"].mean(),2),
"record_count": len(target)
}
return info
except:
return None
def monitor_task():
"""单次巡检主任务"""
print("\n==========开始一轮巡检 {}==========".format(datetime.now()))
collect_result = []
for gid in MONITOR_GOODS:
raw = fetch_goods_detail(gid)
item = parse_item(raw)
if not item:
continue
collect_result.append(item)
print(f"采集成功｜{item['goods_id']}｜{item['title'][:25]}...｜现价：{item['current_price']}｜库存：{item['stock']}")
# 保存数据
if len(collect_result) >0:
save_to_excel(collect_result)
# 读取全部历史用于告警
try:
history_all = pd.read_excel(EXPORT_FILE)
except:
history_all = pd.DataFrame()
# 输出告警信息
for row in collect_result:
alert_list = check_alert(row, history_all)
for msg in alert_list:
print(msg)
# 输出统计简报
print("\n----------本轮数据分析简报----------")
for gid in MONITOR_GOODS:
ana = simple_analysis(gid)
if ana:
print(f"商品:{ana['title'][:22]} 最高{ana['max_price']} 最低{ana['min_price']} 均价{ana['avg_price']} 记录数:{ana['record_count']}")
print("==========本轮巡检结束==========\n")
if __name__ == "__main__":
print("监控程序已启动，按 Ctrl+C 停止程序")
monitor_task() # 先立刻执行一次
schedule.every(CHECK_PERIOD).seconds.do(monitor_task)
while True:
schedule.run_pending()
time.sleep(1)使用步骤说明
将代码中ACCESS_KEY、ACCESS_SECRET替换为自己的凭证，修改MONITOR_GOODS填入需要监控的商品编号。
按需修改预警阈值：降价提醒比例、库存告警数量、巡检时间间隔。
直接运行脚本：python goods_monitor.py
运行之后程序会：
可以拓展的方向
消息通知：把告警信息接入企业微信、钉钉机器人，不用盯着控制台，异动直接推送消息。
多规格 SKU 深度监控：代码示例只取了商品主价格，可以进一步解析 skus 数组，对每一个子规格分别做价格、库存跟踪。
可视化图表：利用 matplotlib 或者 pyecharts 读取 Excel 历史数据，绘制价格趋势曲线图。
存储升级：数据量大之后，可以把存储从 Excel 替换成 SQLite 或者 MySQL，更适合长期大量数据。
批量商品导入：从 csv 读取一大批商品 ID，实现上百个商品批量监控。
踩坑小提示
不要把巡检间隔设置过小，过于频繁会触发访问限制，建议最少间隔 30 分钟以上。
部分商品会返回价格不准，这是平台数据特性，做监控时需要留意，不要完全依赖单一数值做决策。
遇到商品下架，脚本会打印异常日志，方便你及时清理监控列表。
服务器部署可以使用 nohup 或者 systemd 后台常驻，实现 7×24 小时不间断监控。
这套工具把我们从重复复制粘贴网页的工作中解放出来，把精力更多放在数据分析和业务决策上，而不是机械的数据收集。你可以基于这份基础代码，根据自己业务需求不断迭代改造。
