前言
做电商运营、选品分析或者做比价工具，相信大家都遇到过这样的痛点：
需要持续关注多款竞品，价格、库存、规格一变，只能手动刷新页面，人力成本高，很容易错过调价节点；
自己写页面解析脚本，平台防护多，经常遇到访问限制；平台页面稍微改版，整个解析逻辑直接失效，维护成本很高；
拿到零散数据之后，缺少统一存储，无法回溯历史价格，很难做价格波动、库存变化的趋势分析；
多规格商品，不同型号的价格、库存分散在页面各处，很难一次性把全部规格信息完整收集。
之前我尝试过浏览器自动化、页面解析脚本，都要花大量精力处理反限制、页面改版适配。后来使用 Open‑Claw 这套工具，只需要传入商品编号，就可以拿到标准化结构化商品数据，几十行代码就搭建起 7×24 小时自动监控、本地存储、异动识别、简单数据分析的完整体系，不用再关心页面布局变化。
本篇文章完整分享整套落地流程：工具准备、数据获取、定时轮询监控、异常判断、本地持久化存储、基础数据分析，附带可直接运行 Python 源码。
提示：本文仅用于学习公开商品公开信息，业务使用请遵守平台相关规则。
整体方案思路
整套系统拆分为 4 个简单模块：
数据获取模块：输入商品编号，获取商品标题、现价、原价、库存、全部规格、店铺信息等结构化数据；
定时监控模块：循环定时执行，周期性采集商品快照；
异动检测模块：对比上一次记录，识别价格下跌、库存清零、规格变动等异常；
数据存储 & 分析模块：把每一次采集结果保存到本地文件，基于历史记录做简单统计分析。
不需要去解析网页 DOM，不用处理各种页面加密、动态渲染的问题，工具已经帮我们处理好底层的页面访问与内容提取，我们只需要专注业务逻辑。
前期准备
安装依赖库
requests：网络请求
pandas：数据存储与简单统计分析
apscheduler：实现定时任务
2.准备好工具的凭证信息，获取key与secret两个凭证字符串，拿到需要监控的商品编号。
完整 Python 代码实现
import requests
import json
import time
import csv
from datetime import datetime
import pandas as pd
from apscheduler.schedulers.background import BackgroundScheduler
# --------------------------配置区，请修改为你自己的信息--------------------------
KEY = "你的key字符串"
SECRET = "你的secret字符串"
MONITOR_GOODS_ID_LIST = ["10335871600"] # 需要监控的商品编号列表
MONITOR_INTERVAL_MINUTE = 30 # 监控间隔，单位分钟
WARNING_PRICE_THRESHOLD = 50 # 价格低于该数值触发简单告警
SAVE_CSV_PATH = "./goods_monitor_record.csv"
# --------------------------------------------------------------------------------
class GoodsMonitor:
def __init__(self):
self.history_data = {}
def fetch_goods_info(self, goods_id):
"""获取商品结构化数据"""
base_url = "https://api-gw.onebound.cn/jd/item_get_pro"
params = {
"key": KEY,
"secret": SECRET,
"num_iid": goods_id,
"cache": "no",
"result_type": "json"
}
try:
resp = requests.get(base_url, params=params, timeout=15)
raw_json = resp.json()
except Exception as e:
print(f"【网络异常】商品 {goods_id} 获取失败，异常：{str(e)}")
return None
error_code = raw_json.get("error_code", "")
if error_code != "0000":
err_msg = raw_json.get("error", "未知错误")
print(f"【业务异常】商品 {goods_id} 读取失败，信息：{err_msg}")
return None
item = raw_json.get("item", {})
if not item:
print(f"【警告】商品 {goods_id} 未返回有效主体数据")
return None
sku_list = item.get("skus", {}).get("sku", [])
parse_result = {
"snapshot_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"goods_id": item.get("num_iid", ""),
"title": item.get("title", ""),
"current_price": float(item.get("price", 0)) if item.get("price") else 0.0,
"origin_price": float(item.get("orginal_price", 0)) if item.get("orginal_price") else 0.0,
"stock_total": item.get("num", 0),
"shop_name": item.get("seller_info", {}).get("shop_name", ""),
"sku_count": len(sku_list),
"sku_detail": sku_list
}
return parse_result
def detect_change(self, new_data):
"""对比历史快照，识别价格、库存异动"""
goods_id = new_data["goods_id"]
if goods_id not in self.history_data:
self.history_data[goods_id] = new_data
return []
old = self.history_data[goods_id]
alert_list = []
# 价格下跌告警
if new_data["current_price"] < old["current_price"]:
alert_list.append(f"价格变动：由{old['current_price']} 降到 {new_data['current_price']}")
# 库存归零告警
if old["stock_total"] > 0 and new_data["stock_total"] == 0:
alert_list.append("库存告警：商品变为无货状态")
# 自定义阈值告警
if new_data["current_price"] < WARNING_PRICE_THRESHOLD:
alert_list.append(f"阈值告警：当前价格 {new_data['current_price']} 低于预设阈值 {WARNING_PRICE_THRESHOLD}")
self.history_data[goods_id] = new_data
return alert_list
def save_to_csv(self, data):
"""保存本次快照到csv文件，留存历史记录"""
headers = ["snapshot_time", "goods_id", "title", "current_price", "origin_price", "stock_total", "shop_name", "sku_count"]
row = [
data["snapshot_time"],
data["goods_id"],
data["title"],
data["current_price"],
data["origin_price"],
data["stock_total"],
data["shop_name"],
data["sku_count"]
]
file_exist = False
try:
with open(SAVE_CSV_PATH, "r", encoding="utf‑8") as f:
file_exist = True
except FileNotFoundError:
pass
with open(SAVE_CSV_PATH, "a", newline="", encoding="utf‑8") as f:
writer = csv.writer(f)
if not file_exist:
writer.writerow(headers)
writer.writerow(row)
def run_single_task(self):
"""单次监控任务执行入口"""
print(f"\n====== 开始执行一轮监控 {datetime.now().strftime('%Y‑%m‑%d %H:%M:%S')} ======")
for gid in MONITOR_GOODS_ID_LIST:
info = self.fetch_goods_info(gid)
if not info:
continue
print(f"商品：{info['title']}｜现价：{info['current_price']}｜库存：{info['stock_total']}")
alerts = self.detect_change(info)
if alerts:
print("‼️ 检测到异动：")
for a in alerts:
print(f"\t‑ {a}")
self.save_to_csv(info)
time.sleep(1)
def analysis_history():
"""简单数据分析：读取本地csv，输出统计信息"""
try:
df = pd.read_csv(SAVE_CSV_PATH, encoding="utf‑8")
if df.empty:
print("暂无历史监控记录")
return
print("\n========历史数据简单分析结果========")
for goods_id in df["goods_id"].unique():
sub_df = df[df["goods_id"] == goods_id]
print(f"\n商品编号：{goods_id}")
print(f"记录采样次数：{len(sub_df)}")
print(f"历史最低价格：{sub_df['current_price'].min()}")
print(f"历史最高价格：{sub_df['current_price'].max()}")
print(f"价格均值：{round(sub_df['current_price'].mean(),2)}")
except FileNotFoundError:
print("还没有生成监控记录文件，请先运行监控任务")
if __name__ == "__main__":
monitor = GoodsMonitor()
# 执行一次
monitor.run_single_task()
# 开启后台定时任务
scheduler = BackgroundScheduler()
scheduler.add_job(monitor.run_single_task, "interval", minutes=MONITOR_INTERVAL_MINUTE)
scheduler.start()
print(f"定时监控已启动，每 {MONITOR_INTERVAL_MINUTE} 分钟执行一次，按 Ctrl+C 停止")
try:
while True:
time.sleep(60)
except KeyboardInterrupt:
scheduler.shutdown()
print("\n定时任务已停止，正在输出历史数据分析")
analysis_history()代码说明与运行效果
修改配置区：填入你的凭证，填写需要监控的商品编号，设置监控间隔、告警价格阈值。
运行脚本，程序会先执行一次采集，之后后台定时循环采集；
每一轮采集完成，自动把快照写入本地goods_monitor_record.csv；
如果检测到降价、库存清零、低于预设阈值，控制台打印告警信息；
Ctrl+C 终止程序后，自动读取本地历史文件，输出简单统计（最高、最低、平均价格）。
返回的数据里面还包含完整 SKU 规格数组，你可以扩展代码，把每个规格的价格、库存一并写入文件，实现多规格维度监控。
常见踩坑点
返回价格不准：部分场景下返回的主商品价格存在偏差，优先读取 skus 数组内各个子规格的价格作为基准；
商品查无结果：确认商品编号正确，商品没有下架；下架商品会返回对应异常提示；
请求频繁报错：不要设置过小的轮询间隔，业务场景建议至少 10‑30 分钟以上采集一次；
凭证报错：核对 key、secret 字符串，不要带多余空格换行。
业务拓展方向
基于这套基础脚本，还可以继续扩展能力：
消息推送：检测异动之后，对接企业微信、钉钉机器人，实现消息推送，不用盯着控制台；
数据库存储：把 csv 替换为 sqlite/mysql，适合长期大批量商品监控；
可视化：使用 matplotlib 或者 pyecharts 读取历史记录，画出商品价格走势图；
批量监控：从 txt/excel 读取上百个商品编号，批量循环监控竞品池；
规格级监控：遍历 skus 列表，针对每一个子规格独立做价格库存变化判断。
总结
对比传统自己写页面解析的方案，Open‑Claw 工具把页面访问、解析、适配全部封装完成，开发者只需要关注业务逻辑。几十行代码就可以实现 7×24 小时的商品监控，不用再反复调试页面选择器，平台页面改版也几乎不需要修改业务代码，对于中小规模竞品监控、选品数据分析场景非常友好。
