前言
做电商运营或者竞品调研，相信很多人都有这样的痛点：想要持续跟踪多款商品的价格变动、库存、规格信息，每天手动打开网页复制粘贴数据到 Excel 表格。不仅重复劳动耗时间，大促节点的临时降价、库存清仓往往转瞬即逝，人工很难第一时间捕捉变化。
如果自己手写页面解析脚本，又会遇到页面动态渲染、网页改版导致解析失效、访问频次受限等一系列棘手问题，调试维护成本居高不下。
最近在项目实践中使用了 Open Claw 工具，它可以直接输出结构化的商品信息，不用再折腾网页源码解析，我们可以把精力全部放在业务逻辑上，快速搭建一套本地商品监控与数据分析程序。
本篇教程会完整演示整套流程：包含本地数据持久化存储、价格库存异动提醒、历史时序数据分析、批量商品巡检，所有代码可以直接复制运行，Windows、Mac、Linux 都支持，部署到服务器就可以实现 7×24 小时不间断监控。
整体实现思路
获取商品结构化原始信息：商品标题、当前售价、原始标价、库存、多规格信息、店铺名称等完整内容；
使用 SQLite 本地数据库保存每一次采集记录，沉淀历史时序数据；
设置定时巡检任务，对比前后两次数据，识别降价、库存变动，输出告警提示；
读取本地历史记录做数据分析，统计历史最高、最低、平均售价，分析价格波动幅度；
支持批量加入多款商品，扩展可对接钉钉、企业微信实现消息推送。
环境准备
Python 版本建议 3.7 以上，安装需要的依赖包
提前准备好 Open Claw 后台生成的两组身份凭证，用于程序访问。
完整代码实现
新建文件 goods_monitor.py，复制下面全部代码，修改配置区域的凭证、监控商品编号、告警阈值就可以直接运行。
import requests
import json
import sqlite3
import time
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
# ===================== 【配置区域，请自行修改】 =====================
CLAW_KEY = "填入你的凭证key"
CLAW_SECRET = "填入你的凭证secret"
# 需要监控的商品编号，可以填入多个实现批量监控
MONITOR_GOODS = ["10335871600", "10057467958584"]
POLL_INTERVAL = 300 # 巡检间隔，单位秒，普通商品建议300秒，爆款可缩短
PRICE_ALERT_RATIO = 0.15 # 降价超过15%就触发告警
DB_FILE = "goods_monitor.db"
# =================================================================
def init_db():
"""初始化本地数据库，创建数据表"""
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS goods_record (
id INTEGER PRIMARY KEY AUTOINCREMENT,
goods_id TEXT NOT NULL,
title TEXT,
price REAL,
original_price REAL,
stock INTEGER,
shop_name TEXT,
record_time TIMESTAMP
)
''')
conn.commit()
conn.close()
def get_goods_info(goods_id: str):
"""
获取商品结构化数据
:param goods_id: 商品数字编号
:return: 字典格式商品信息，失败返回None
"""
params = {
"key": CLAW_KEY,
"secret": CLAW_SECRET,
"num_iid": goods_id,
"cache": "no",
"result_type": "json"
}
retry = 3
while retry > 0:
try:
resp = requests.get(
"https://api-gw.onebound.cn/jd/item_get_pro",
params=params,
timeout=15
)
data = resp.json()
if data.get("error_code") != "0000":
print(f"获取商品{goods_id}异常：{data.get('reason','未知错误')}")
return None
item_data = data.get("item", {})
result = {
"goods_id": goods_id,
"title": item_data.get("title", ""),
"price": float(item_data.get("price", 0)) if item_data.get("price") else 0,
"original_price": float(item_data.get("orginal_price", 0)) if item_data.get("orginal_price") else 0,
"stock": int(item_data.get("num", 0)) if item_data.get("num") else 0,
"shop_name": item_data.get("seller_info", {}).get("shop_name", "")
}
return result
except Exception as e:
retry -= 1
print(f"网络异常，剩余重试次数 {retry}, 错误：{str(e)}")
time.sleep(2)
return None
def save_to_db(goods_data):
"""将采集到的数据存入本地数据库"""
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO goods_record
(goods_id, title, price, original_price, stock, shop_name, record_time)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
goods_data["goods_id"],
goods_data["title"],
goods_data["price"],
goods_data["original_price"],
goods_data["stock"],
goods_data["shop_name"],
datetime.now().strftime("%Y-%m-%d %H:%M:%S")
))
conn.commit()
conn.close()
def get_last_record(goods_id):
"""读取该商品上一次记录，用于对比变动"""
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
cursor.execute('''
SELECT price, stock FROM goods_record WHERE goods_id=?
ORDER BY record_time DESC LIMIT 1
''', (goods_id,))
row = cursor.fetchone()
conn.close()
if row:
return {"price": row[0], "stock": row[1]}
return None
def monitor_task():
"""定时执行的监控任务"""
print(f"\n==== 巡检开始 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ====")
for gid in MONITOR_GOODS:
info = get_goods_info(gid)
if not info:
continue
save_to_db(info)
last = get_last_record(gid)
print(f"商品【{info['title']}】 当前价格：{info['price']} 库存：{info['stock']}")
if last:
# 降价告警判断
if last["price"] > 0 and info["price"] < last["price"]:
drop_rate = (last["price"] - info["price"]) / last["price"]
if drop_rate >= PRICE_ALERT_RATIO:
print(f"【⚠️价格告警】商品{gid}发生大幅降价！上次:{last['price']} 本次:{info['price']} 降幅{drop_rate:.1%}")
# 库存变动提醒
if info["stock"] != last["stock"]:
print(f"【ℹ库存变动】商品{gid}库存变更，上一次:{last['stock']}，当前:{info['stock']}")
time.sleep(1)
def analysis_goods(goods_id):
"""针对单个商品做历史数据分析"""
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
cursor.execute('''
SELECT price, record_time FROM goods_record WHERE goods_id=? ORDER BY record_time
''', (goods_id,))
rows = cursor.fetchall()
conn.close()
if not rows:
print("暂无历史数据，请先运行监控采集数据！")
return
price_list = [r[0] for r in rows]
max_p = max(price_list)
min_p = min(price_list)
avg_p = sum(price_list)/len(price_list)
print("\n========商品历史数据分析========")
print(f"历史最高售价：{max_p:.2f}")
print(f"历史最低售价：{min_p:.2f}")
print(f"历史平均售价：{avg_p:.2f}")
print(f"数据记录总条数：{len(price_list)}")
if __name__ == "__main__":
init_db()
scheduler = BackgroundScheduler()
scheduler.add_job(monitor_task, "interval", seconds=POLL_INTERVAL)
scheduler.start()
print("监控程序已经启动，按 Ctrl+C 终止程序")
try:
while True:
cmd = input("\n输入指令：\n1:执行一次巡检 2:分析商品历史 q:退出\n>")
if cmd == "1":
monitor_task()
elif cmd == "2":
gid = input("输入要分析的商品编号：")
analysis_goods(gid)
elif cmd.lower() == "q":
break
except KeyboardInterrupt:
pass
finally:
scheduler.shutdown()
print("程序已退出")程序使用说明
修改代码顶部配置，填入自己的两组凭证，把需要跟踪的商品编号写进MONITOR_GOODS列表；
直接运行脚本 python goods_monitor.py，程序自动初始化数据库，开启后台定时巡检；
控制台交互指令：
采集的全部历史会保存在goods_monitor.db文件，你可以使用数据库工具打开，导出为 Excel 表格做进一步处理。
功能拓展方向
消息推送：出现降价、库存变动告警的时候，可以增加钉钉机器人、企业微信 webhook，把告警信息推送到工作群，不用守着控制台；
多规格 SKU 分析：源码中只提取了基础价格，还可以读取多规格的数据，跟踪不同规格的价格、库存变化；
数据可视化：结合 matplotlib 绘制价格变化曲线图，直观观察一段时间的价格走势；
导出报表：定时把每日数据导出 csv 文件，方便周报复盘；
部署服务器：Linux 服务器后台运行，实现全天候无人值守监控。
实际业务场景
竞品跟踪：电商运营持续盯对手商品调价、库存情况，把握促销时机；
选品比价：跟踪多款目标商品，等待合适价位再入手；
店铺自检：跟踪自家店铺多款商品，监控异常改价、库存异常；
大促备战：618、双 11 大促期间高频巡检，捕捉短期促销活动。
写在最后
这套方案避开了复杂的页面解析工作，把重点放在业务逻辑开发，新手也可以快速落地。实际使用的时候注意控制巡检间隔，不要设置过于频繁，避免超出额度。如果后续需要更大规模的批量任务，可以在这个基础上做任务队列优化。
