适用场景：竞品盯价、库存异动跟踪、促销活动观察、商品数据归档复盘，Python 完整可运行脚本，适合开发者、电商运营做二次落地。
前言
做电商运营或者市场调研的时候，很多人都有这样的痛点：需要持续关注一批竞品商品，手动打开网页复制价格、库存、规格信息，一天反复刷好几次，耗费大量时间，还很容易漏掉降价、断货、上新规格这类关键变化。
如果商品数量多，几十上百个商品，纯人工维护基本不现实。直接写原生页面抓取，又会遇到页面加密、动态渲染、频繁访问被限制等一堆麻烦，调试成本很高。
最近在实践 OpenClaw 这套工具链，搭配成熟的数据获取能力，可以快速搭建一套稳定的商品监控系统。不需要处理复杂页面解析逻辑，只需要关注业务逻辑：定时拉取商品信息、对比历史记录、识别异动、存储数据、简单统计分析。
本文就完整分享整套落地流程，从环境准备、单商品数据获取、定时监控、异动检测，再到简单数据分析，全部附带可直接运行的 Python 代码。
整体实现思路
整套系统分为 4 个核心环节：
- 数据获取：输入商品编号，拿到商品标题、售价、原价、库存、店铺、多规格信息、图片链接等结构化内容。
- 本地持久化：把每一轮采集的数据保存到本地数据库，留存历史快照，方便后续做对比和数据分析。
- 异动检测：对比上一轮记录，识别价格变动、库存清零、规格增减等变化，输出告警提示。
- 数据分析输出：基于历史数据，输出价格波动统计、库存变化记录，导出表格做报表。
环境依赖：Python3.8 及以上，需要安装 requests、pandas、sqlite3（python 自带）
pip install requests pandas完整代码实现
1. 基础封装：获取商品原始结构化数据
把基础请求逻辑封装，输入商品编号，返回整理后的商品字典。
# -*- coding:utf-8 -*- import requests import json import time import sqlite3 import pandas as pd # -------------------------- 配置区域，请自行修改 -------------------------- ACCESS_KEY = "你的访问凭证key" ACCESS_SECRET = "你的访问凭证secret" MONITOR_GOODS_ID_LIST = ["10335871600", "10057467958584"] # 需要监控的商品编号列表 MONITOR_INTERVAL = 60 * 30 # 监控轮询间隔，单位秒，示例30分钟跑一次 DB_FILE = "./goods_monitor.db" # ------------------------------------------------------------------------ def fetch_goods_detail(goods_id: str): """ 获取商品详情原始结构化数据 :param goods_id: 商品编号 :return: dict 整理后的商品信息，失败返回None """ base_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" } try: resp = requests.get(base_url, params=params, headers=headers, timeout=20) resp_json = resp.json() except Exception as e: print(f"【网络异常】商品 {goods_id} 请求失败：{str(e)}") return None # 判断返回状态 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 # 整理核心字段 result = { "num_iid": item_data.get("num_iid"), "title": item_data.get("title", ""), "price": float(item_data.get("price", 0)) if item_data.get("price") else 0, "orginal_price": float(item_data.get("orginal_price", 0)) if item_data.get("orginal_price") else 0, "stock": item_data.get("num", 0), "brand": item_data.get("brand", ""), "shop_name": item_data.get("seller_info", {}).get("shop_name", ""), "detail_url": item_data.get("detail_url", ""), "capture_ts": int(time.time()), "capture_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) } # 提取多规格SKU信息 sku_list = [] skus_raw = item_data.get("skus", {}).get("sku", []) for sku in skus_raw: sku_info = { "sku_id": sku.get("sku_id"), "sku_price": float(sku.get("price",0)) if sku.get("price") else 0, "sku_origin_price": float(sku.get("orginal_price",0)) if sku.get("orginal_price") else 0, "sku_prop": sku.get("properties_name", ""), "sku_stock": sku.get("quantity",0) } sku_list.append(sku_info) result["sku_list"] = sku_list return result2. 数据库初始化，存储历史快照
使用轻量 sqlite 保存每一轮采集的数据，一张表存商品快照，一张表存规格快照，方便后续对比分析。
def init_database(): """初始化数据库表""" conn = sqlite3.connect(DB_FILE) cur = conn.cursor() # 商品主表，每一次采集生成一条记录 cur.execute(''' CREATE TABLE IF NOT EXISTS goods_snapshot ( id INTEGER PRIMARY KEY AUTOINCREMENT, num_iid TEXT NOT NULL, title TEXT, price REAL, orginal_price REAL, stock INTEGER, brand TEXT, shop_name TEXT, detail_url TEXT, capture_ts INTEGER, capture_time TEXT ) ''') # SKU规格子表，保存每个规格的价格库存 cur.execute(''' CREATE TABLE IF NOT EXISTS goods_sku_snapshot ( id INTEGER PRIMARY KEY AUTOINCREMENT, parent_snapshot_id INTEGER, sku_id TEXT, sku_price REAL, sku_origin_price REAL, sku_prop TEXT, sku_stock INTEGER ) ''') conn.commit() conn.close() def save_to_db(goods_data: dict): """把单次采集的商品数据存入数据库""" conn = sqlite3.connect(DB_FILE) cur = conn.cursor() cur.execute(''' INSERT INTO goods_snapshot (num_iid,title,price,orginal_price,stock,brand,shop_name,detail_url,capture_ts,capture_time) VALUES (?,?,?,?,?,?,?,?,?,?) ''',( goods_data["num_iid"], goods_data["title"], goods_data["price"], goods_data["orginal_price"], goods_data["stock"], goods_data["brand"], goods_data["shop_name"], goods_data["detail_url"], goods_data["capture_ts"], goods_data["capture_time"] )) snapshot_id = cur.lastrowid # 写入sku子表 for sku in goods_data["sku_list"]: cur.execute(''' INSERT INTO goods_sku_snapshot (parent_snapshot_id,sku_id,sku_price,sku_origin_price,sku_prop,sku_stock) VALUES (?,?,?,?,?,?) ''',( snapshot_id, sku["sku_id"], sku["sku_price"], sku["sku_origin_price"], sku["sku_prop"], sku["sku_stock"] )) conn.commit() conn.close()3. 异动检测逻辑：对比上一轮数据，识别价格、库存变化
每次新拿到数据，查询数据库里该商品最近一条历史记录，做差值比对，输出变动提醒。
def get_last_snapshot(num_iid: str): """获取该商品上一次采集的快照""" conn = sqlite3.connect(DB_FILE) cur = conn.cursor() cur.execute(''' SELECT price,orginal_price,stock FROM goods_snapshot WHERE num_iid = ? ORDER BY capture_ts DESC LIMIT 1 ''',(num_iid,)) row = cur.fetchone() conn.close() if row is None: return None return { "price": row[0], "orginal_price": row[1], "stock": row[2] } def detect_change(last_data, new_data): """检测商品是否发生异动，返回变动信息列表""" change_msg = [] if last_data is None: change_msg.append("首次采集，无历史对比") return change_msg # 价格变化 if abs(new_data["price"] - last_data["price"]) > 0.01: change_msg.append(f"价格变动：{last_data['price']} → {new_data['price']}") if abs(new_data["orginal_price"] - last_data["orginal_price"]) >0.01: change_msg.append(f"原价变动：{last_data['orginal_price']} → {new_data['orginal_price']}") # 库存变化 if new_data["stock"] != last_data["stock"]: change_msg.append(f"库存变动：{last_data['stock']} → {new_data['stock']}") return change_msg4. 监控主循环 + 简单数据分析导出
定时循环监控，同时提供函数可以把历史数据导出 Excel，做后续数据分析。
def run_monitor_round(): """执行一轮完整监控任务""" print(f"\n========== 开始一轮监控 {time.strftime('%Y-%m-%d %H:%M:%S')} ==========") for goods_id in MONITOR_GOODS_ID_LIST: print(f"\n正在处理商品:{goods_id}") new_info = fetch_goods_detail(goods_id) if not new_info: continue last_info = get_last_snapshot(goods_id) changes = detect_change(last_info, new_info) if len(changes) > 0: print(f"⚠️【商品{goods_id} 检测到异动】") for m in changes: print(f" >> {m}") else: print(f"✅【商品{goods_id}无变化】当前价格:{new_info['price']},库存:{new_info['stock']}") save_to_db(new_info) def export_analysis_excel(out_file="./goods_monitor_report.xlsx"): """导出全部历史快照到Excel，用于数据分析""" conn = sqlite3.connect(DB_FILE) df_main = pd.read_sql("SELECT * FROM goods_snapshot", conn) df_sku = pd.read_sql("SELECT * FROM goods_sku_snapshot", conn) conn.close() with pd.ExcelWriter(out_file) as writer: df_main.to_excel(writer, sheet_name="商品主快照", index=False) df_sku.to_excel(writer, sheet_name="规格快照", index=False) print(f"✅数据分析报表已导出：{out_file}") if __name__ == "__main__": init_database() print("初始化完成，启动监控循环，按Ctrl+C停止程序") try: while True: run_monitor_round() # 可选：执行一次导出，注释掉就不会每次导出 # export_analysis_excel() time.sleep(MONITOR_INTERVAL) except KeyboardInterrupt: print("\n程序手动退出，结束监控") # 退出前导出一份报表 export_analysis_excel()使用说明
- 将代码中
ACCESS_KEY、ACCESS_SECRET替换成你自己的凭证； - 修改
MONITOR_GOODS_ID_LIST填入你需要监控的商品编号； MONITOR_INTERVAL代表轮询间隔，建议不要设置太短，避免频繁请求；- 运行脚本，程序会循环执行监控，按
Ctrl+C结束程序，结束时自动导出 Excel 报表。
运行之后，本地会生成goods_monitor.db数据库文件，保存全部历史快照，每次采集到价格、库存变动，控制台会打印告警信息。导出的 Excel 可以用来做价格趋势统计、库存波动复盘。
OpenClaw 如何赋能这套监控系统
很多人写监控脚本，难点不在于写请求代码，而在于工程化调度、异常容错、任务管理。
OpenClaw 可以把上面的业务逻辑封装成可复用的任务能力：
- 定时调度托管：不用自己写 while 循环 sleep，交给调度引擎，配置执行周期，程序崩溃自动重启重试；
- 异常容错处理：遇到网络超时、数据为空、返回异常，自动记录错误日志，不会因为单个商品失败导致整个监控任务挂掉；
- 告警扩展：检测到价格 / 库存异动后，可以对接消息推送，把变动消息直接发送到企业微信、钉钉，不用盯着控制台看输出；
- 数据分析能力增强：基于保存的历史快照，直接做趋势统计、批量报表生成，不用再手动写 pandas 处理逻辑；
- 横向扩展：后续增加更多平台商品监控，只需要新增数据获取逻辑，调度、存储、告警整套能力可以直接复用。
简单来说：上面的 Python 脚本是业务逻辑原型，OpenClaw 负责把原型变成稳定长期跑在服务器上的生产级监控任务。
拓展优化方向
- 告警推送：检测到 changes 非空的时候，增加钉钉 / 企业微信 webhook，实现消息推送；
- 阈值告警：设置心理价位，当商品价格低于目标价格才触发提醒；
- 增量过滤：历史数据库越来越大，可以增加定时清理旧快照逻辑，只保留最近 N 天记录；
- 多任务分组：把商品分成不同分组，不同分组设置不同监控频率，高优先级商品缩短轮询间隔；
- 可视化：读取 sqlite 数据，用 matplotlib 绘制价格波动曲线图，直观查看历史价格走势。
小结
对于电商从业者、市场调研人员，手动维护商品信息成本极高。借助 OpenClaw 搭配成熟的数据获取能力，我们不需要深陷页面逆向、反爬对抗，把精力聚焦在业务本身：定时采集、历史归档、异动识别、数据分析。
