万邦api博客

Python 实战｜搭建电商商品自动化监控与数据分析小工具，告别手工复制粘贴

Tue, 20 Oct 2026 14:26:03 +0800

适用场景：竞品跟踪、价格盯盘、库存监控、市场数据分析；适合商家、运营、数据分析同学直接上手跑通，代码可直接二次改造。

前言

做电商运营或者竞品调研的时候，很多人都有这样的痛点：想要盯一批商品的价格变动、库存余量、规格变化，每次都要手动打开网页一个个复制记录。商品一多，耗时巨大，还很容易错过促销降价、库存清空的关键节点。

如果直接写网页解析，还会遇到频繁的访问限制、页面结构改版、验证码等一堆麻烦，调试就要耗费大量时间。

今天分享一套轻量的实现方案，借助 Open Claw 能力，只需要商品编号，就能拿到完整结构化商品信息；再搭配 Python 脚本，实现定时巡检、异动提醒、历史数据存储、简单统计分析，快速拥有一套属于自己的监控小工具。

环境依赖：Python3.7+，需要安装 requests、pandas、schedule。

pip install 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
运行之后程序会：

立刻执行一次数据拉取；
按照设定间隔自动循环巡检；
本地生成商品监控历史记录.xlsx，保存全部历史采集数据；
控制台打印日志，出现降价、库存不足时输出警告；
每轮结束输出简单统计简报，看到该商品历史最高、最低、平均价格。

可以拓展的方向

消息通知：把告警信息接入企业微信、钉钉机器人，不用盯着控制台，异动直接推送消息。
多规格 SKU 深度监控：代码示例只取了商品主价格，可以进一步解析 skus 数组，对每一个子规格分别做价格、库存跟踪。
可视化图表：利用 matplotlib 或者 pyecharts 读取 Excel 历史数据，绘制价格趋势曲线图。
存储升级：数据量大之后，可以把存储从 Excel 替换成 SQLite 或者 MySQL，更适合长期大量数据。
批量商品导入：从 csv 读取一大批商品 ID，实现上百个商品批量监控。

踩坑小提示

不要把巡检间隔设置过小，过于频繁会触发访问限制，建议最少间隔 30 分钟以上。
部分商品会返回价格不准，这是平台数据特性，做监控时需要留意，不要完全依赖单一数值做决策。
遇到商品下架，脚本会打印异常日志，方便你及时清理监控列表。
服务器部署可以使用 nohup 或者 systemd 后台常驻，实现 7×24 小时不间断监控。

这套工具把我们从重复复制粘贴网页的工作中解放出来，把精力更多放在数据分析和业务决策上，而不是机械的数据收集。你可以基于这份基础代码，根据自己业务需求不断迭代改造。

告别人工盯品！Open‑Claw 快速搭建电商商品监控与数据分析完整方案（附可运行 Python 代码）

Tue, 20 Oct 2026 10:29:12 +0800

适用场景：竞品盯价、货源监控、价格走势分析、库存异动跟踪，本地电脑、云服务器都可以直接部署运行。

前言

做电商运营、选品、竞品分析的朋友，大多会遇到一个痛点：想要持续跟踪一批商品的售价变动、库存变化、规格调整，手动点开网页反复查看效率极低。尤其需要同时盯十几、几十款商品，每天反复截图、复制记录，耗费大量时间，还很容易漏掉调价、清库存这类关键信号。

自己手写页面解析方案，又会遇到不少麻烦：页面结构改版导致字段提取失效、访问频率过高被限制访问、动态页面渲染拿不到完整规格信息、SKU 多规格数据很难完整拆解。

最近在实践中发现 Open‑Claw 这套工具可以很好解决上述问题。只需要输入商品编号，就能拿到完整结构化的商品信息，包含标题、原价、实时售价、多规格 SKU、库存、品牌、店铺信息、图集等内容，内置缓存、异常处理机制，不用处理复杂的页面解析逻辑，能把开发工作量大幅降低。

本文分享一套完整可落地的监控方案，包含数据拉取、时序存储、异动告警、数据分析可视化完整流程，代码可以直接复制修改配置就运行。

整体方案设计

整套系统拆分为 4 个模块，模块之间低耦合，后续可以按需扩展功能：

数据获取模块：传入商品编号，拿到完整结构化商品信息，处理网络异常、商品不存在等各类异常情况。
数据持久化模块：每次采集带上时间戳，保存历史记录，这里选用 CSV 文件，不需要额外安装数据库，普通办公电脑直接打开表格查看历史数据。
异动监控告警模块：对比本次采集和上一轮历史数据，识别价格跳水、库存骤降，达到设定阈值就输出告警提示。
数据分析模块：基于留存的时序历史数据，完成价格走势绘图、多规格对比、竞品统计分析，输出可视化图表。

对比传统页面解析方式，这套方案优势：

稳定性更好，不用担心页面样式改版导致解析失效；
一次性返回全部规格、店铺、图片信息，不需要多次请求拼接；
自带缓存策略，重复查询可以加速返回，减少网络消耗；
代码量少，新手也能快速看懂、二次修改。

环境准备

需要安装依赖库，命令行执行：

pip install requests pandas matplotlib

完整代码实现

# -*- coding:utf-8 -*-
import requests
import json
import time
import pandas as pd
from datetime import datetime
import matplotlib.pyplot as plt

# ======================【请自行修改配置区域】======================
CLAW_KEY = "填入你的key"
CLAW_SECRET = "填入你的secret"

# 需要监控的商品编号列表
MONITOR_GOODS = ["10335871600", "10057467958584"]
# 循环监控间隔，单位秒，建议不要设置过小，300即5分钟
POLL_SECONDS = 300
# 降价告警阈值，0.1代表相比上一次，价格下跌超过10%触发提醒
PRICE_ALERT_RATIO = 0.1
# 历史数据存储文件
SAVE_CSV = "goods_monitor_history.csv"
# 是否开启缓存，开启后优先读取缓存，速度更快
USE_CACHE = "yes"
# =================================================================

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": CLAW_KEY,
        "secret": CLAW_SECRET,
        "num_iid": goods_id,
        "cache": USE_CACHE,
        "result_type": "json"
    }
    try:
        resp = requests.get(base_url, params=params, timeout=15, headers={"Accept‑Encoding": "gzip"})
        resp.raise_for_status()
        res_json = resp.json()
        # 异常判断
        if res_json.get("error_code") != "0000":
            err_msg = res_json.get("reason", res_json.get("error", "未知错误"))
            print(f"【采集异常】商品{goods_id}：{err_msg}")
            return None
        item_data = res_json.get("item")
        if not item_data:
            print(f"【采集异常】商品{goods_id}返回内容为空")
            return None
        return item_data
    except Exception as e:
        print(f"【网络异常】商品{goods_id}，错误信息：{str(e)}")
        return None


def save_history(data_list):
    """追加写入历史时序数据到csv文件"""
    df_new = pd.DataFrame(data_list)
    try:
        df_old = pd.read_csv(SAVE_CSV)
        df_merge = pd.concat([df_old, df_new], ignore_index=True)
    except FileNotFoundError:
        df_merge = df_new
    df_merge.to_csv(SAVE_CSV, index=False, encoding="utf‑8‑sig")


def check_alert(goods_id, current_price, last_price):
    """价格异动告警逻辑"""
    if last_price is None or current_price is None:
        return False
    try:
        curr = float(current_price)
        last = float(last_price)
        if last <= 0:
            return False
        drop_ratio = (last - curr) / last
        if drop_ratio >= PRICE_ALERT_RATIO:
            print("=" * 60)
            print(f"⚠️ 告警！商品 {goods_id} 发生明显降价！")
            print(f"上一轮价格：{last}，当前价格：{curr}，降幅 {round(drop_ratio*100,2)}%")
            print("=" * 60)
            return True
    except:
        pass
    return False


def analysis_draw(goods_id):
    """读取本地历史数据，绘制价格走势图表"""
    try:
        df = pd.read_csv(SAVE_CSV, encoding="utf‑8‑sig")
        df_target = df[df["goods_id"] == goods_id].copy()
        if len(df_target) < 2:
            print(f"商品{goods_id}历史记录不足，无法绘图")
            return
        df_target["crawl_time"] = pd.to_datetime(df_target["crawl_time"])
        plt.rcParams["font.sans‑serif"] = ["SimHei"]
        plt.figure(figsize=(10, 5))
        plt.plot(df_target["crawl_time"], df_target["price"], marker="o", label="实时售价")
        plt.title(f"商品 {goods_id} 价格走势")
        plt.xlabel("时间")
        plt.ylabel("价格")
        plt.legend()
        plt.grid(alpha=0.3)
        plt.tight_layout()
        plt.savefig(f"price_trend_{goods_id}.png", dpi=150)
        plt.close()
        print(f"✅ 商品{goods_id}价格走势图已保存到本地")
    except Exception as e:
        print(f"绘图失败：{str(e)}")


def start_monitor():
    print("=====商品监控程序已启动=====")
    last_record = dict()  # 保存上一轮采集结果，用于比对告警
    while True:
        batch_save = []
        for gid in MONITOR_GOODS:
            item_info = fetch_goods_detail(gid)
            if not item_info:
                continue
            now_time = datetime.now().strftime("%Y‑%m‑%d %H:%M:%S")
            title = item_info.get("title", "")
            price = item_info.get("price", 0)
            origin_price = item_info.get("orginal_price", 0)
            stock = item_info.get("num", 0)
            shop_name = item_info.get("seller_info", {}).get("shop_name", "")
            # 简单提取第一条规格信息，完整SKU列表存在 item_info["skus"]中
            one_row = {
                "crawl_time": now_time,
                "goods_id": gid,
                "title": title,
                "price": price,
                "origin_price": origin_price,
                "stock": stock,
                "shop_name": shop_name
            }
            batch_save.append(one_row)
            # 告警判断
            last_p = last_record.get(gid, {}).get("price")
            check_alert(gid, price, last_p)
            last_record[gid] = one_row
            print(f"[{now_time}] 已采集：{gid} | {title[:25]} ｜现价:{price} 库存:{stock}")
        # 批量写入文件
        if len(batch_save) > 0:
            save_history(batch_save)
        print(f"本轮采集完成，休眠 {POLL_SECONDS} 秒等待下一轮...\n")
        time.sleep(POLL_SECONDS)


if __name__ == "__main__":
    # 启动持续监控
    start_monitor()
    # 如果只想做一次性数据分析，注释上面，调用下面函数传入商品编号
    # analysis_draw("10335871600")

代码说明与扩展方向

配置修改：只需要修改代码最上方配置区，填入自己的密钥、填写需要监控的商品编号，设置循环间隔就可以直接运行。

注意：轮询间隔不要设置过小，过于频繁会触发访问限制。

SKU 多规格处理：返回结果中skus字段包含全部规格，每一条规格带有规格名称、规格编号、对应价格、库存。可以在代码中循环解析item_info["skus"]["sku"]，把每一个子规格单独存一行记录，实现多规格分别监控。
告警能力拓展：目前告警仅控制台打印，实际项目可以拓展，触发告警后调用企业微信 / 钉钉机器人、发送邮件，实现消息推送。
数据分析拓展：

读取本地 CSV，可以做多款竞品横向价格对比；
统计原价、现价差值，计算折扣力度；
筛选库存低于阈值的商品做库存预警；
按店铺做批量统计，输出店铺商品大盘报表。

异常状态处理：返回结果会包含各类异常标记，比如商品下架、编号错误、访问受限，代码中已经做基础捕获，可以根据业务需求增加对应处理逻辑，比如商品下架打印提示。

踩坑经验总结

价格字段偶尔会出现数据延迟，不能完全单一依赖单次采集的价格做业务决策，建议结合历史多轮数据综合判断。
开启缓存模式可以加快返回速度，如果需要拿到实时最新状态，可以将USE_CACHE设置为no，但会消耗更多额度。
尽量不要设置很短的轮询周期，合理设置间隔，避免短时间大量请求。
存储 CSV 文件用utf‑8‑sig编码，Windows 系统 Excel 打开不会出现中文乱码。

小结

对于做竞品跟踪、货源调研的同学，这套方案可以把原本大量人工重复操作交给程序自动完成。不需要深入研究页面逆向解析，把精力放在业务数据分析上。程序持续后台运行，自动留存时序数据，价格变动第一时间收到提醒，还可以直接生成价格走势图，辅助选品、调价决策。

告别手工盯盘：用 Open‑Claw 搭建电商商品自动化监控与数据分析实战

Wed, 14 Oct 2026 10:07:37 +0800

适用场景：竞品跟踪、价格波动监控、库存变动记录、历史行情统计；Python 完整可运行示例，适合个人开发者、电商运营、数据分析同学参考。

前言

做电商运营或者竞品调研的时候，我们经常需要持续关注一批商品：价格有没有下调、库存是否清零、不同规格套餐的售价变化、店铺基础信息更新。如果依靠人工反复打开网页复制记录，不仅耗费大量时间，还很容易错过关键变动，历史数据也很难留存用于复盘分析。

很多同学会尝试直接解析网页页面，但平台页面结构经常改版，CSS 选择器隔一段时间就失效，调试成本非常高。Open‑Claw 这套工具集合，帮我们屏蔽了页面解析的繁琐工作，直接输出结构化的商品原始资料，我们只需要专注业务逻辑：定时拉取、存储历史记录、做对比分析，实现一套轻量化的商品监控系统。

本文会完整讲解整个实现思路，给出可直接调试的 Python 代码，实现：商品基础信息获取、定时循环监控、数据落盘保存、价格波动识别，简单的数据分析统计。

整体实现思路

整套程序分为 4 个核心环节：

资料拉取：输入商品编号，拿到标题、现价、原价、库存、多规格套餐、店铺名称、主图等完整结构化内容；
持久化存储：把每一次采集的结果保存，形成时间序列历史记录；
变动检测：对比上一轮记录，识别价格涨跌、库存变动；
简单分析：读取历史文件，计算最高价、最低价、价格变动频次，输出统计结果。

注意：使用前需要提前准备好你的访问凭证，包含两组密钥信息。

环境准备

依赖安装，执行下面命令：

pip install requests pandas schedule

requests：负责网络请求收发
pandas：用来做数据整理、存储 csv、简单统计分析
schedule：实现定时循环任务，周期性巡检商品状态

完整代码实现

新建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 里面会包含这些核心信息：

capture_time：采集时间戳
goods_id：商品编号
title：商品完整标题
brand：品牌名称
shop_name：店铺名称
current_price：当前售价
origin_price：划线原价
stock_total：总库存
detail_url：商品网页地址
main_image：主图地址
skus：JSON 字符串，存放各个规格套餐的编号、名称、价格、库存

依靠这份历史表格，我们后续还可以拓展：

使用 matplotlib 绘制价格走势折线图；
接入消息推送，发生价格变动的时候发送通知；
批量导入上百个商品编号做批量竞品巡检；
将 csv 迁移至 sqlite/mysql 数据库，支撑更大体量的数据。

踩坑记录与注意事项

凭证密钥保管：两组密钥不要硬编码直接上传到公开代码仓库，正式项目建议放到环境变量或者配置文件。
采集频次控制：不要设置过小的轮询间隔，频繁请求会触发访问限制，建议至少 10 分钟以上一轮。
规格价格注意：部分场景下返回的主条目价格不一定和子规格完全匹配，业务分析优先读取 skus 数组内各个套餐价格。
异常容错：代码里面已经增加捕获异常，遇到网络波动、商品下架不会直接让整个程序崩溃，控制台会打印错误提示。
商品下架情况：当商品已经被平台下架，会返回对应提示，脚本会打印日志，不会生成无效记录。

后续拓展方向

增加消息推送：企业微信、钉钉机器人，一旦监控到降价、库存变化直接推送提醒；
可视化：把历史 csv 数据绘制价格走势图，直观看到一段时间的价格波动；
持久化数据库：数据量大之后替换 csv，改用 sqlite 数据库存储；
多平台兼容：这套工具也可以拓展其他平台的商品资料获取，复用监控分析逻辑。

写在最后

对于运营、数据分析同学，不需要深度钻研页面逆向、页面反爬对抗，借助 Open‑Claw 直接拿到干净结构化的原始资料，把精力放在业务逻辑：定时巡检、历史留存、变动告警、行情分析，快速搭建一套属于自己的竞品监控工具。

OpenClaw 实战｜告别手工盯盘，搭建电商商品自动化监控与数据分析体系

Wed, 14 Oct 2026 09:18:47 +0800

适用场景：竞品盯价、库存异动跟踪、促销活动观察、商品数据归档复盘，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 result

2. 数据库初始化，存储历史快照

使用轻量 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_msg

4. 监控主循环 + 简单数据分析导出

定时循环监控，同时提供函数可以把历史数据导出 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 搭配成熟的数据获取能力，我们不需要深陷页面逆向、反爬对抗，把精力聚焦在业务本身：定时采集、历史归档、异动识别、数据分析。

告别反复手工查价：用 Open‑Claw 搭建商品监控与数据分析小工具

Mon, 12 Oct 2026 15:17:27 +0800

适用场景：竞品盯价、大促价格跟踪、库存异动监控、商品信息批量统计，Python 完整实战，可直接上手运行

前言

做电商运营、选品或者个人薅羊毛，经常会遇到一个头疼问题：需要持续关注一批商品的价格、库存、规格变化。如果每天手动点开页面一个个复制记录，不仅耗费大量时间，还很容易错过大促降价、库存清空这类关键变化。

直接写页面解析脚本又会碰到一堆麻烦：网页结构改版就全部失效、访问频繁触发风控限制、各种动态渲染内容拿不到完整规格信息，调试维护成本很高。

最近在实践中发现借助 Open‑Claw 的数据能力，只需要输入商品编号，就能够拿到结构化的商品全部信息，包含标题、原价、实时售价、多规格参数、库存、店铺信息、图集等完整内容，不用处理复杂网页解析，我们可以快速搭建一套本地监控 + 数据分析脚本，完成定时采集、异动提醒、历史数据留存与简单统计分析。

本文完整演示从数据获取、持久化存储、异动告警，再到简单数据分析的全流程，代码可以直接复制修改使用。

整体实现思路

整套工具分为 4 个核心模块：

数据获取模块：输入商品编号，拿到标准化结构化商品数据，包含全部规格 SKU、价格、库存、品牌、图片、店铺等字段；支持缓存加速，减少重复请求消耗。
数据存储模块：每次采集带上时间戳，保存到本地 csv 文件，留存时序历史记录，不需要部署数据库，普通 Excel 就可以打开查看历史变化。
异动检测模块：对比本次与上一轮采集记录，识别价格涨跌、库存变动，控制台输出告警，后续可以拓展企业微信、钉钉消息推送。
数据分析模块：基于本地历史文件，做价格波动统计、多商品横向对比，输出简单分析结果，辅助选品和竞品研判。

环境准备

依赖基础库，执行安装：

pip install requests pandas

完整代码实战

1. 基础数据获取封装

把网络请求逻辑封装，只传入商品编号，返回清洗之后的字典对象，同时处理异常情况（商品不存在、网络异常等）

# -*- coding:utf-8 -*-
import requests
import time
import pandas as pd
from datetime import datetime

class GoodsDataClient:
    def __init__(self, key, secret):
        self.key = key
        self.secret = secret
        self.base_url = "https://api‑gw.onebound.cn/jd/item_get_pro/"

    def get_goods_info(self, goods_id):
        """根据商品编号获取商品结构化数据"""
        params = {
            "key": self.key,
            "secret": self.secret,
            "num_iid": goods_id,
            "cache": "yes",
            "result_type": "json"
        }
        headers = {
            "Accept‑Encoding": "gzip",
            "Connection": "close"
        }
        try:
            resp = requests.get(self.base_url, params=params, headers=headers, timeout=15)
            json_data = resp.json()
        except Exception as e:
            print(f"网络请求异常：{str(e)}")
            return None

        # 判断返回状态
        if json_data.get("error_code") != "0000":
            err_msg = json_data.get("reason", "未知异常")
            print(f"获取数据失败：{err_msg}")
            return None

        item = json_data.get("item")
        if not item:
            print("未获取到商品主体信息")
            return None

        # 简单清洗整理核心字段
        clean_data = {
            "crawl_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
            "goods_id": item.get("num_iid"),
            "title": item.get("title"),
            "brand": item.get("brand"),
            "current_price": float(item.get("price", 0)) if item.get("price") else 0,
            "origin_price": float(item.get("orginal_price",0)) if item.get("orginal_price") else 0,
            "stock": item.get("num",0),
            "shop_name": item.get("seller_info",{}).get("shop_name",""),
            "detail_url": item.get("detail_url"),
            "sku_list": item.get("skus",{}).get("sku",[])
        }
        return clean_data

2. 本地持久化与异动告警逻辑

每次采集把数据追加写入 csv，同时对比上一条记录，发现价格或者库存发生明显变化，控制台打印告警提示。

class GoodsMonitor:
    def __init__(self, client, save_file="goods_monitor.csv"):
        self.client = client
        self.save_file = save_file

    def save_to_csv(self, data):
        """保存采集记录到本地csv"""
        df_new = pd.DataFrame([data])
        try:
            df_old = pd.read_csv(self.save_file, encoding="utf‑8")
            df_all = pd.concat([df_old, df_new], ignore_index=True)
        except FileNotFoundError:
            df_all = df_new
        df_all.to_csv(self.save_file, index=False, encoding="utf‑8")

    def check_change(self, new_data):
        """对比历史数据，检测价格库存异动"""
        goods_id = new_data["goods_id"]
        try:
            df = pd.read_csv(self.save_file, encoding="utf‑8")
            his = df[df["goods_id"] == goods_id].tail(1)
            if len(his) ==0:
                return []
            last = his.iloc[0]
            alert_list = []
            # 价格变动超过0.1元触发提醒
            price_diff = new_data["current_price"] - float(last["current_price"])
            if abs(price_diff) > 0.1:
                alert_list.append(f"【价格变动】原价{last['current_price']} → 现价 {new_data['current_price']}")
            # 库存发生变化
            if int(new_data["stock"]) != int(last["stock"]):
                alert_list.append(f"【库存变动】旧库存{last['stock']} → 当前库存 {new_data['stock']}")
            return alert_list
        except Exception:
            return []

    def monitor_one_goods(self, goods_id):
        print(f"\n====开始采集商品 {goods_id} ====")
        info = self.client.get_goods_info(goods_id)
        if not info:
            print("采集失败，跳过该商品")
            return None
        # 保存记录
        self.save_to_csv(info)
        # 检查异动
        alerts = self.check_change(info)
        if alerts:
            print("⚠️⚠️⚠️发现异动提醒：")
            for a in alerts:
                print(f"\t{a}")
        else:
            print("本次对比：无明显异动")
        print(f"商品标题：{info['title']}")
        print(f"当前售价：{info['current_price']} ｜ 原价：{info['origin_price']}｜库存：{info['stock']}")
        return info

3. 简单数据分析函数

读取本地全部历史记录，输出单商品价格统计，支持查看最高、最低、平均售价，也可以批量对比多款商品价格区间。

def analysis_goods(file_name, target_goods_id=None):
    """历史数据分析，打印价格统计信息"""
    df = pd.read_csv(file_name, encoding="utf‑8")
    if target_goods_id:
        df = df[df["goods_id"] == str(target_goods_id)]
        if len(df) ==0:
            print("没有找到该商品的历史采集记录")
            return
        print(f"\n========商品 {target_goods_id} 历史价格统计========")
        print(f"记录总条数：{len(df)}")
        print(f"历史最低售价：{df['current_price'].min():.2f}")
        print(f"历史最高售价：{df['current_price'].max():.2f}")
        print(f"历史平均售价：{df['current_price'].mean():.2f}")
    else:
        # 全部商品简单汇总
        group = df.groupby("goods_id").agg({
            "current_price":["min","max","mean","count"]
        }).round(2)
        print("\n====全部监控商品汇总统计====")
        print(group)
    return df

4. 主程序入口，定时循环监控

if __name__ == "__main__":
    # 填入自己的凭证
    MY_KEY = "你的密钥key"
    MY_SECRET = "你的密钥secret"

    client = GoodsDataClient(MY_KEY, MY_SECRET)
    monitor = GoodsMonitor(client, save_file="goods_monitor.csv")

    # 需要监控的商品编号列表，可以放多个商品
    watch_goods = [
        "10335871600",
        # "填入更多商品id"
    ]

    # 循环监控，设置间隔时间，单位秒，根据自身配额合理设置，不要请求过于频繁
    while True:
        for gid in watch_goods:
            monitor.monitor_one_goods(gid)
            time.sleep(4)
        print("\n本轮采集完成，等待下一轮循环......")
        time.sleep(60*10) # 10分钟执行一轮

        # 如果需要执行分析，可以取消下面注释
        # analysis_goods("goods_monitor.csv", watch_goods[0])

使用说明与注意事项

获取商品编号：商品详情网页链接中，item.jd.com/后面那串数字就是 goods_id，复制填入监控列表即可。
频率控制：循环间隔不要设置过小，频繁请求会消耗额度，也容易触发限制，建议至少数分钟以上一轮。
缓存开关：代码中 cache 参数设置 yes，会优先读取缓存数据，速度更快；如果需要实时最新数据，可以改为 no。
异常处理：脚本已经做了基础异常捕获，网络波动、商品下架不会让整个程序直接崩溃。当商品下架或者编号错误时控制台会打印提示。
拓展方向

告警：检测到价格变动之后，可以对接企业微信 / 钉钉 webhook 推送消息，不用盯着控制台。
可视化：基于 pandas+matplotlib，读取 csv 历史数据，画出商品价格走势图。
定时任务：Linux/macOS 用 crontab、Windows 使用任务计划程序，把脚本后台长期运行。
规格深度解析：返回结果中sku_list包含全部规格的编号、价格、库存，我们可以二次处理做多规格分别监控。

拿到数据之后可以做哪些分析？

有了持续沉淀的时序数据，就可以落地很多业务分析：

大促价格研判：对比历史最低价位，判断当前是不是好的入手时机，观察大促期间真实降价幅度，区分先涨价再打折的虚假活动。
竞品对标：持续跟踪同行多款竞品的调价节奏、库存状态，了解对手上新、清仓动作。
库存预警：监控爆款库存水位，当库存快速下跌，及时捕捉市场热度变化。
规格分析：解析多规格数据，看哪一个规格价格最有优势，哪些规格经常缺货。

踩坑小结

之前尝试直接页面解析的方案，经常遇到页面改版代码直接失效，动态 JS 渲染拿不全 SKU 规格，调试消耗大量时间。借助 Open‑Claw 拿到已经处理好的结构化数据之后，我们只需要专注业务逻辑：存储、告警、数据分析，不用再消耗精力对抗网页反爬和页面变更，小工具开发效率提升非常明显。

Open‑Claw 实战｜告别页面逆向，快速搭建电商商品监控与数据分析完整方案

Fri, 09 Oct 2026 14:51:30 +0800

适用场景：竞品跟踪、价格盯盘、库存状态巡检、历史价格统计分析；语言 Python3，适合开发、数据分析从业者阅读，可直接复制代码运行调试。

前言

做电商相关数据分析的时候，最头疼的就是商品页面的数据获取。直接解析网页会遇到动态渲染、反访问、页面改版、规格页面逻辑复杂等一堆麻烦，每次页面改版，之前写好的解析脚本就要全部重写，大量时间消耗在页面适配上面，真正留给业务分析的时间反而很少。

很多同学会使用浏览器自动化工具去模拟访问，但又会碰到验证码、设备指纹、访问频率限制，部署到服务器运行的时候坑非常多。

今天分享一套基于 Open‑Claw 工具的实践方案，不需要处理网页逆向、不用维护页面选择器，直接拿到标准化的商品全量信息，完成定时监控、数据落盘、异常告警、历史数据分析整套流程。

我们可以一次性拿到：商品标题、现价、原价、库存、全部规格信息、店铺信息、主图、发货地等完整内容，把精力聚焦业务逻辑，而不是和网页页面做对抗。

整体业务目标

商品全量信息快照：获取商品基础信息，包含多规格维度的数据。
持续定时巡检：周期性采集商品状态，记录每一次时间切片的数据，捕捉降价、涨价、缺货、规格变动。
本地持久化存储：将每次采集结果保存到 CSV，留存历史数据，用于后续统计分析。
异常提醒：价格低于预期阈值、库存清零，控制台输出告警。
简易数据分析：读取本地历史文件，做价格波动简单统计。

环境准备

依赖安装，执行下面命令：

pip install requests pandas

完整实现代码

整体分为：封装数据获取类、字段解析、存储、告警、定时循环、数据分析几个模块。

使用前需要替换凭证信息，以及填写你需要监控的商品编号。

# coding:utf-8
import requests
import time
import csv
import os
from datetime import datetime
import pandas as pd


class GoodsMonitor:
    def __init__(self, access_key, access_secret):
        self.access_key = access_key
        self.access_secret = access_secret
        self.base_address = "https://api‑gw.onebound.cn/jd/item_get_pro/"
        self.csv_file = "goods_monitor_record.csv"
        self.init_csv()

    def init_csv(self):
        """初始化存储文件，不存在则创建表头"""
        if not os.path.exists(self.csv_file):
            headers = [
                "crawl_time",
                "goods_id",
                "goods_title",
                "current_price",
                "origin_price",
                "stock_num",
                "brand",
                "shop_name",
                "sku_list"
            ]
            with open(self.csv_file, "w", encoding="utf‑8‑sig", newline="") as f:
                writer = csv.writer(f)
                writer.writerow(headers)

    def get_raw_data(self, goods_id):
        """获取商品原始结构化数据"""
        params = {
            "key": self.access_key,
            "secret": self.access_secret,
            "num_iid": goods_id,
            "cache": "no"
        }
        headers = {
            "Accept‑Encoding": "gzip",
            "Connection": "close"
        }
        try:
            resp = requests.get(
                self.base_address,
                params=params,
                headers=headers,
                timeout=25
            )
            return resp.json()
        except Exception as e:
            print(f"网络异常: {str(e)}")
            return None

    def parse_raw(self, raw_data):
        """清洗解析原始返回内容，提取业务需要字段"""
        if not raw_data:
            return None
        err_code = raw_data.get("error_code")
        if err_code != "0000":
            err_info = raw_data.get("error", "获取数据失败")
            print(f"数据异常：{err_info}")
            return None

        item_info = raw_data.get("item", {})
        if not item_info:
            return None

        sku_data = item_info.get("skus", {}).get("sku", [])
        sku_text = "||".join([f'{x.get("properties_name")}:{x.get("price")}' for x in sku_data])

        result = {
            "crawl_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
            "goods_id": item_info.get("num_iid"),
            "goods_title": item_info.get("title", ""),
            "current_price": float(item_info.get("price", 0)) if item_info.get("price") else 0,
            "origin_price": float(item_info.get("orginal_price", 0)) if item_info.get("orginal_price") else 0,
            "stock_num": int(item_info.get("num", 0)) if item_info.get("num") else 0,
            "brand": item_info.get("brand", ""),
            "shop_name": item_info.get("seller_info", {}).get("shop_name", ""),
            "sku_list": sku_text
        }
        return result

    def save_record(self, data):
        """写入本地csv历史记录"""
        with open(self.csv_file, "a", encoding="utf‑8‑sig", newline="") as f:
            writer = csv.writer(f)
            row = [
                data["crawl_time"],
                data["goods_id"],
                data["goods_title"],
                data["current_price"],
                data["origin_price"],
                data["stock_num"],
                data["brand"],
                data["shop_name"],
                data["sku_list"]
            ]
            writer.writerow(row)

    def check_warn(self, data, warn_price):
        """告警逻辑：价格低于阈值、库存为0触发提醒"""
        if data["current_price"] <= warn_price:
            print("="*70)
            print(f"⚠️【价格预警】商品：{data['goods_title']}")
            print(f"现价 {data['current_price']}，低于监控阈值 {warn_price}")
            print("="*70)
        if data["stock_num"] <= 0:
            print("="*70)
            print(f"🔴【库存告警】商品 {data['goods_title']} 当前缺货")
            print("="*70)

    def single_task(self, goods_id, warn_price):
        """单次监控任务：获取‑解析‑存储‑告警"""
        raw = self.get_raw_data(goods_id)
        parsed = self.parse_raw(raw)
        if parsed:
            self.save_record(parsed)
            self.check_warn(parsed, warn_price)
            print(f"✅ {parsed['crawl_time']} 完成记录：{parsed['goods_title']}")
        else:
            print(f"❌ 商品 {goods_id} 获取信息失败")

    def run_monitor_loop(self, watch_list, interval=60*30):
        """循环持续监控
        watch_list: [(商品ID,预警价格),...]
        interval：循环间隔，单位秒，示例30分钟=1800
        """
        print(">>>>>>启动商品循环监控<<<<<<")
        while True:
            for goods_id, warn_price in watch_list:
                self.single_task(goods_id, warn_price)
                time.sleep(3)
            print(f"\n等待下一轮采集，间隔 {interval//60} 分钟\n")
            time.sleep(interval)

    def analysis_history(self):
        """简单历史数据分析，读取本地csv做价格统计"""
        if not os.path.exists(self.csv_file):
            print("暂无历史记录文件")
            return
        df = pd.read_csv(self.csv_file, encoding="utf‑8‑sig")
        print("\n========历史数据统计分析========")
        goods_group = df.groupby("goods_title")["current_price"].agg(["min","max","mean","count"])
        print(goods_group)
        print("\n每一件商品最低、最高、平均价格，采集总次数如上")


if __name__ == "__main__":
    # =========配置区========
    MY_KEY = "替换为你的凭证key"
    MY_SECRET = "替换为你的凭证secret"

    # 待监控列表，格式(商品编号，价格预警线)
    WATCH_GOODS = [
        ("10335871600", 220),
        ("10057467958584", 600)
    ]
    LOOP_INTERVAL = 1800  # 循环监控间隔，单位秒，这里30分钟跑一轮
    # ======================

    monitor = GoodsMonitor(MY_KEY, MY_SECRET)

    # 1.执行循环监控，开启后会持续运行
    monitor.run_monitor_loop(WATCH_GOODS, interval=LOOP_INTERVAL)

    # 2.只做一次采集，测试用
    # monitor.single_task("10335871600",200)

    # 3.读取本地历史做数据分析
    # monitor.analysis_history()

代码说明

初始化文件：程序第一次运行自动生成goods_monitor_record.csv，所有采集快照全部保存在这里，Excel 可以直接打开查看。
数据获取与解析：工具返回已经结构化的商品信息，我们只需要从中筛选业务需要字段，不需要写 Xpath、CSS 选择器，页面改版不会影响业务代码。同时做异常捕获，单次获取失败不会让整个程序直接崩溃退出。
告警机制：当商品价格低于设置阈值，或者库存为 0，控制台打印提醒，可以在此基础上拓展推送能力，比如对接企业微信、邮件消息推送。
循环监控：run_monitor_loop会按照设定的间隔周期循环巡检多件商品，服务器后台运行就可以实现 7×24 小时盯盘。
数据分析模块：analysis_history读取本地历史记录，使用 pandas 统计每个商品历史最低价、最高价、平均价格，方便做价格趋势研判。

⚠️注意：不要把间隔设置过小，频繁请求会触发访问限制，业务场景建议至少 10‑30 分钟以上采集一轮。

拓展方向

1. 存储升级：替换 CSV 为数据库

如果监控商品数量很多，CSV 读写效率会下降，可以把存储逻辑修改为 SQLite / MySQL，增加批量分析、筛选查询能力。示例写入 SQL 片段：

# 伪代码示意
# insert_sql = """
# INSERT INTO goods_snapshot(goods_id,title,price,origin_price,stock,crawl_time)
# VALUES (%s,%s,%s,%s,%s,%s)
# """
# cursor.execute(insert_sql,(parsed["goods_id"],parsed["goods_title"],parsed["current_price"],parsed["origin_price"],parsed["stock_num"],parsed["crawl_time"]))
# conn.commit()

2. 消息推送扩展

现在告警只是控制台打印，可以接入企业微信机器人、邮件、钉钉，一旦触发价格 / 库存变化，直接推送消息到手机，不用一直盯着控制台。

3. 可视化

基于 pandas + matplotlib 读取历史记录，绘制商品价格变化曲线图，直观看历史价格波动。

4. 多平台扩展

这套思路可以迁移到其他电商品类，只需要调整解析字段，监控逻辑、存储、告警代码完全复用。

踩坑总结

价格字段说明：返回的基础价格为主商品价格，如果要看各个规格真实价格，需要读取 sku_list 字段，不同规格会存在价格差异。
缓存控制：代码中设置cache=no，目的是拿到最新的实时快照；测试调试的时候可以打开缓存，减少消耗。
异常处理：商品下架、编号错误、网络抖动，程序会捕获错误并打印日志，不会直接退出循环监控任务。
服务器部署：Linux 后台运行可以使用nohup python monitor.py &方式，注意配置好 Python 环境与依赖包。

小结

借助 Open‑Claw 工具，我们绕开了复杂网页解析工作，把重心放在业务监控逻辑上。几十行代码就完成商品定时监控、历史数据留存、异常告警、数据分析整套工程，不用反复适配页面结构，极大降低维护成本。

对于个人爱好者、小团队做竞品盯盘、价格跟踪、库存巡检，这套方案足够满足大部分场景。在此基础上，还可以继续拓展报表输出、消息推送、多商品批量管理等能力。

告别复杂页面解析，OpenClaw 快速搭建电商商品监控与数据分析脚本

Fri, 09 Oct 2026 11:48:12 +0800

适用场景：个人比价、商家竞品跟踪、价格库存异动监测，无需处理页面渲染、验证码等棘手问题，Python 完整实战教程。

前言

做电商相关开发或者个人薅羊毛的时候，经常有这样的需求：持续跟踪一批商品的价格变动、库存状态、规格参数变化，做历史数据分析、降价提醒、竞品对比。

如果自己手写页面解析，会遇到一堆麻烦：网页结构经常改版、动态渲染、频繁弹出人机校验，IP 访问频率高还容易被限制。调试解析规则就要耗费大量时间，经常今天写好的脚本，过几天页面一改直接全部失效。

最近体验了 OpenClaw 这套工具，它把商品页面的各类信息做了标准化输出，我们只需要传入商品编号，就能拿到结构化的标题、售价、原价、SKU 规格、库存、店铺信息、图片等全部内容，把精力放在业务逻辑：监控、存储、分析，不用再折腾页面解析。

本文就完整演示如何基于 OpenClaw，实现定时商品监控、数据本地存储、简单异动分析，完整代码可以直接复制运行。

整体思路

准备好工具的身份凭证（密钥信息），拿到需要监控的商品 ID；
编写获取商品结构化信息的基础函数；
增加异常捕获，处理商品下架、网络波动等各类异常情况；
将每次采集到的数据存入本地文件，留存历史记录；
增加定时循环，实现持续监控；
简单数据分析：对比前后两次数据，识别降价、库存变动。

提示：运行脚本前，请先安装依赖库

pip install requests pandas schedule

完整 Python 实战代码

# -*- coding:utf-8 -*-
"""
OpenClaw 电商商品监控&数据分析脚本
功能：商品信息获取、异常容错、本地保存历史、价格库存异动检测
"""
import requests
import time
import pandas as pd
import schedule
import json
from datetime import datetime

# ======================== 配置区域，请自行修改 ========================
# 你的个人凭证信息
ACCESS_KEY = "替换成你的key"
ACCESS_SECRET = "替换成你的secret"

# 需要监控的商品编号列表，可以添加多个商品ID
MONITOR_GOODS_ID = [
    "10335871600",
    "10057467958584"
]

# 监控间隔，单位秒，不要设置过小，避免调用过于频繁
MONITOR_INTERVAL = 60 * 10

# 历史数据保存文件
SAVE_FILE = "goods_monitor_history.csv"
# ====================================================================


def get_goods_detail(goods_id: str):
    """
    获取单个商品结构化数据
    goods_id:商品编号
    return: 成功返回字典，失败返回None
    """
    base_url = "https://api-gw.onebound.cn/jd/item_get_pro/"
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "num_iid": goods_id,
        "cache": "no",   # no获取最新数据，yes读取缓存加快速度
        "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()
        error_code = resp_json.get("error_code")

        # 0000代表正常拿到数据
        if error_code == "0000":
            item_data = resp_json.get("item")
            return item_data
        else:
            err_msg = resp_json.get("error", "未知错误")
            print(f"【商品{goods_id}】获取失败，错误信息：{err_msg}")
            return None

    except Exception as e:
        print(f"【商品{goods_id}】网络或解析异常：{str(e)}")
        return None


def parse_item(raw_item: dict):
    """把原始返回数据，提取我们关心的核心字段"""
    if not raw_item:
        return None
    out = {}
    out["crawl_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    out["goods_id"] = raw_item.get("num_iid")
    out["title"] = raw_item.get("title", "")
    out["price"] = float(raw_item.get("price", 0)) if raw_item.get("price") else 0
    out["original_price"] = float(raw_item.get("orginal_price", 0)) if raw_item.get("orginal_price") else 0
    out["stock"] = raw_item.get("num", 0)
    out["shop_name"] = raw_item.get("seller_info", {}).get("shop_name", "")
    out["detail_url"] = raw_item.get("detail_url", "")
    # sku规格简单提取
    sku_list = raw_item.get("skus", {}).get("sku", [])
    sku_info = []
    for sku in sku_list:
        sku_info.append({
            "sku_id": sku.get("sku_id"),
            "spec_name": sku.get("properties_name"),
            "sku_price": sku.get("price"),
            "sku_stock": sku.get("quantity")
        })
    out["sku_info"] = json.dumps(sku_info, ensure_ascii=False)
    return out


def save_to_csv(data_row: dict):
    """单条记录追加保存到csv历史文件"""
    df_new = pd.DataFrame([data_row])
    try:
        # 文件存在就追加，不存在新建
        with open(SAVE_FILE, "r", encoding="utf-8-sig") as f:
            df_new.to_csv(SAVE_FILE, mode="a", header=False, index=False, encoding="utf-8-sig")
    except FileNotFoundError:
        df_new.to_csv(SAVE_FILE, mode="w", header=True, index=False, encoding="utf-8-sig")


last_status = {}  # 保存上一轮采集状态，用于异动对比


def monitor_task():
    """一轮完整监控任务，循环所有商品"""
    print(f"\n==========开始一轮监控 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ==========")
    global last_status
    for gid in MONITOR_GOODS_ID:
        item_raw = get_goods_detail(gid)
        parsed = parse_item(item_raw)
        if not parsed:
            continue
        save_to_csv(parsed)
        print(f"商品【{parsed['title'][0:25]}...】现价：{parsed['price']} 库存：{parsed['stock']}")

        # 和上一次数据对比，检测异动
        prev = last_status.get(gid)
        if prev:
            # 价格下跌
            if parsed["price"] < prev["price"]:
                print(f"⚠️ 价格异动提醒！商品{gid}发生降价，上次{prev['price']}，当前{parsed['price']}")
            # 库存变动
            if parsed["stock"] != prev["stock"]:
                print(f"⚠️ 库存变动提醒！商品{gid}库存由{prev['stock']}变更为{parsed['stock']}")
        # 更新记录
        last_status[gid] = parsed
        time.sleep(2)


def analysis_history():
    """简单数据分析示例，读取本地历史，打印每个商品价格区间"""
    try:
        df = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
        print("\n----------历史数据分析结果----------")
        for goods_id in df["goods_id"].unique():
            sub = df[df["goods_id"] == goods_id]
            title = sub.iloc[0]["title"]
            min_p = sub["price"].min()
            max_p = sub["price"].max()
            print(f"商品：{title[:30]}... 历史最低:{min_p} 历史最高:{max_p}，共采集{len(sub)}条记录")
    except Exception as e:
        print("暂无历史数据，等待采集完成", e)


if __name__ == "__main__":
    # 执行一次
    monitor_task()
    analysis_history()

    # 设置定时循环执行
    schedule.every(MONITOR_INTERVAL).seconds.do(monitor_task)
    print(f"\n已启动定时监控，每 {MONITOR_INTERVAL/60}分钟执行一次，Ctrl+C终止程序")
    while True:
        schedule.run_pending()
        time.sleep(1)

代码讲解

get_goods_detail：核心获取函数，传入商品编号，工具会返回完整的商品全部维度信息，我们不需要处理网页；同时做了网络超时、异常返回的捕获，防止程序直接崩溃。

cache 参数：设置no获取实时最新价格库存；设置yes会读取缓存，响应更快，适合非实时分析场景。

parse_item：原始返回字段比较多，这个函数做字段精简，挑选我们监控业务需要的：采集时间、商品标题、价格、原价、库存、店铺、SKU 规格，SKU 数组转字符串方便存入表格。
save_to_csv：每一轮采集结果都写入本地 CSV，相当于自建一份商品时间线数据库，后续可以用 Excel、Pandas 做趋势分析。
monitor_task：一轮监控主逻辑，遍历全部待监控商品；并且用last_status保存上一轮状态，对比发现降价、库存变化，控制台输出告警提示。
analysis_history：读取本地保存的历史记录，统计每个商品历史最高价、最低价，简单数据分析。
主程序部分：运行一次，然后开启定时任务，按设定周期自动轮询。

常见问题踩坑说明

数据拿不到？

确认密钥信息复制正确，不要有多余空格换行；
确认商品编号是正确有效，下架的商品会返回对应错误提示；
不要把监控间隔设置得太短，短时间大量请求会触发限制。

价格和页面显示不一致 部分大促优惠券、满减活动属于页面动态计算，拿到的是商品基础标价。多规格商品，不同规格各自有独立价格，需要看返回 SKU 列表。
数据存储 示例中用 CSV 文件存储，适合小批量监控；如果要监控几百上千个商品，可以改成 SQLite / MySQL 数据库存储。
告警扩展 当前示例只在控制台打印提醒，实际项目中可以拓展：邮件推送、企业微信、钉钉消息，实现真正的消息通知。

拓展玩法

拿到标准化结构化数据之后，可以衍生很多有意思的小项目：

个人比价助手：监控心仪家电、数码产品，降到心理价位就收到提醒；
商家竞品分析：批量跟踪同行多款竞品，记录调价节奏、库存变化；
数据报表输出：定时把历史数据生成 Excel 报表，做周度、月度价格统计；
筛选过滤 SKU：针对特定规格做监控，比如只盯某一个型号的价格变动。

写在最后

很多时候我们写脚本，大量时间消耗在网页页面适配、反爬对抗上。OpenClaw 这类工具，把网页解析这一层封装完成，直接输出干净的结构化数据。我们可以把重心放回业务逻辑：监控逻辑、数据存储、数据分析。

告别手工盯屏，用 Open‑Claw 快速搭建电商商品监控与数据分析体系

Thu, 08 Oct 2026 10:03:18 +0800

本文适合电商运营、数据爱好者、Python 开发，手把手教你实现商品价格、库存、规格自动跟踪，完成历史数据存储、简单可视化分析。

前言

做电商运营、竞品调研的时候，很多人都有这样的痛点：需要持续关注多款商品的价格变动、库存变化、促销活动。如果每天手动打开网页复制记录，不仅耗费大量时间，还很容易漏掉转瞬即逝的降价、缺货节点。

直接写脚本解析网页源码，又会遇到重重阻碍：网页动态渲染、频繁触发访问限制、页面改版之后全部代码失效，调试成本极高。

最近接触到 Open‑Claw 这套工具套件，可以省去页面解析、反访问限制处理这些繁琐工作，直接拿到结构化的商品全量信息。我们只需要专注业务逻辑：定时拉取数据、落地存储、做统计分析，快速搭建一套属于自己的商品监控系统。

本文会完整演示从数据获取、持久化存储，到监控告警、数据分析可视化完整流程，所有示例均可直接修改配置运行。

整体实现思路

整套系统分成 4 个核心环节：

数据获取：输入商品编号，拿到标题、现价、原价、库存、多规格信息、店铺信息等结构化内容。
持久化存储：把每一次采集的数据存入本地数据库，留存历史记录，用于后续对比。
监控逻辑：定时执行采集任务，对比本次和上一轮价格、库存，出现变动输出提醒。
数据分析：基于历史数据，做价格波动统计、规格对比，输出图表，辅助运营判断。

环境准备：Python3.7 以上，安装依赖库

pip install requests pandas matplotlib sqlite3

一、商品原始数据获取

首先实现基础读取函数，传入商品编号就可以拿到完整结构化资料。只需要配置好自己的凭证信息，不需要处理网页 DOM、JS 渲染等复杂逻辑。

# coding:utf‑8
import requests
import json
import time
import sqlite3
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime

# ========= 配置区，修改为自己的凭证与监控商品 =========
ACCESS_KEY = "你的访问凭证"
ACCESS_SECRET = "你的安全密钥"
MONITOR_GOODS_ID = "10335871600"
MONITOR_CYCLE = 1800  # 监控周期，单位秒，示例30分钟采集一次
# ==================================================

def get_goods_detail(goods_id):
    """
    获取商品结构化详情
    :param goods_id: 商品编号
    :return: dict商品数据，失败返回None
    """
    headers = {
        "Accept‑Encoding": "gzip",
        "Connection": "close"
    }
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "num_iid": goods_id
    }
    try:
        resp = requests.get(
            "https://api‑gw.onebound.cn/jd/item_get_pro",
            params=params,
            headers=headers,
            timeout=20
        )
        res_json = resp.json()
        # 判断返回状态
        if res_json.get("error_code") != "0000":
            print(f"获取失败：{res_json.get('reason','未知错误')}")
            return None
        item_data = res_json.get("item")
        if not item_data:
            print("未找到对应商品")
            return None
        return item_data
    except Exception as e:
        print(f"网络异常:{str(e)}")
        return None

调用示例：

if __name__ == "__main__":
    info = get_goods_detail(MONITOR_GOODS_ID)
    if info:
        print(json.dumps(info, ensure_ascii=False, indent=2))

运行之后，即可拿到商品标题、当前售价、划线原价、库存、全部规格 sku、店铺名称、商品大图等完整结构化字典，不用自己写复杂页面解析逻辑。

二、本地数据库存储历史数据

要做监控和分析，必须留存每一次采集快照。这里选用轻量的 SQLite，无需额外安装数据库服务。新建数据表保存每次采集时间、商品基础信息。

def init_db():
    """初始化数据库，创建商品监控记录表"""
    conn = sqlite3.connect("goods_monitor.db")
    cursor = conn.cursor()
    create_sql = """
    CREATE TABLE IF NOT EXISTS goods_record (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        goods_id TEXT,
        title TEXT,
        price REAL,
        original_price REAL,
        stock INTEGER,
        shop_name TEXT,
        capture_time TEXT
    )
    """
    cursor.execute(create_sql)
    conn.commit()
    conn.close()

def save_to_db(item):
    """单条商品记录入库"""
    conn = sqlite3.connect("goods_monitor.db")
    cursor = conn.cursor()
    insert_sql = """
    INSERT INTO goods_record
    (goods_id, title, price, original_price, stock, shop_name, capture_time)
    VALUES (?,?,?,?,?,?,?)
    """
    cursor.execute(insert_sql, (
        item.get("num_iid"),
        item.get("title"),
        float(item.get("price",0)),
        float(item.get("orginal_price",0)),
        int(item.get("num",0)),
        item.get("seller_info",{}).get("shop_name",""),
        datetime.now().strftime("%Y‑%m‑%d %H:%M:%S")
    ))
    conn.commit()
    conn.close()

初始化数据库：

init_db()

三、实现监控告警逻辑

核心逻辑：每次拿到最新数据，读取数据库中上一条记录，对比价格、库存，如果发生变化打印告警提示，你可以扩展对接企业微信、邮件实现消息推送。

def get_last_record(goods_id):
    """读取该商品最近一条历史记录"""
    conn = sqlite3.connect("goods_monitor.db")
    df = pd.read_sql(
        f"SELECT * FROM goods_record WHERE goods_id='{goods_id}' ORDER BY id DESC LIMIT 1",
        conn
    )
    conn.close()
    if len(df) > 0:
        return df.iloc[0].to_dict()
    return None

def check_change(new_item):
    """对比新旧数据，检测变动，输出告警"""
    goods_id = new_item.get("num_iid")
    last = get_last_record(goods_id)
    if not last:
        print("首次采集，无历史对比数据")
        return

    new_price = float(new_item.get("price",0))
    old_price = last["price"]
    new_stock = int(new_item.get("num",0))
    old_stock = last["stock"]

    print(f"\n=====变动检测【{new_item.get('title')}】=====")
    if new_price != old_price:
        print(f"⚠️价格变动！旧价：{old_price} → 新价：{new_price}")
    if new_stock != old_stock:
        print(f"⚠️库存变动！旧库存：{old_stock} → 新库存：{new_stock}")
    if new_price == old_price and new_stock == old_stock:
        print("✅价格库存均无变化")

def monitor_loop():
    """循环监控主逻辑"""
    while True:
        print(f"\n==== {datetime.now().strftime('%Y‑%m‑%d %H:%M:%S')} 执行采集 ====")
        item_info = get_goods_detail(MONITOR_GOODS_ID)
        if item_info:
            check_change(item_info)
            save_to_db(item_info)
        print(f"等待{MONITOR_CYCLE/60}分钟，进入下一轮采集...")
        time.sleep(MONITOR_CYCLE)

启动监控循环：

# 启动持续监控
if __name__ == "__main__":
    init_db()
    monitor_loop()

提示：实际部署可以使用 nohup 或者 systemd 把脚本放后台运行，实现 7*24 小时不间断盯商品。当检测到降价、库存清空，控制台直接输出提醒，业务场景可以继续扩展消息推送。

四、基于历史记录做数据分析与可视化

积累一段时间历史数据后，就可以做数据分析，观察商品价格波动趋势，判断促销节点。下面从数据库读取全部历史记录，绘制价格走势图表。

def analysis_and_plot(goods_id):
    """读取历史数据，数据分析+绘制价格走势图"""
    conn = sqlite3.connect("goods_monitor.db")
    df = pd.read_sql(
        f"SELECT * FROM goods_record WHERE goods_id='{goods_id}' ORDER BY capture_time",
        conn
    )
    conn.close()
    if len(df) == 0:
        print("暂无历史监控数据")
        return

    # 基础统计
    print("\n====历史数据统计摘要====")
    print(f"总采集记录数：{len(df)}")
    print(f"历史最低售价：{df['price'].min()}")
    print(f"历史最高售价：{df['price'].max()}")
    print(f"平均售价：{round(df['price'].mean(),2)}")

    # 绘图设置中文
    plt.rcParams["font.sans‑serif"] = ["SimHei"]
    plt.rcParams["axes.unicode_minus"] = False

    df["capture_time"] = pd.to_datetime(df["capture_time"])
    plt.figure(figsize=(12,6))
    plt.plot(df["capture_time"], df["price"], marker="o", label="实时售价")
    plt.plot(df["capture_time"], df["original_price"], linestyle="‑‑", label="原价")
    plt.title("商品价格历史监控走势")
    plt.xlabel("采集时间")
    plt.ylabel("价格")
    plt.legend()
    plt.xticks(rotation=30)
    plt.tight_layout()
    plt.savefig("price_trend.png", dpi=150)
    print("图表已保存 price_trend.png")
    plt.show()

调用分析函数：

if __name__ == "__main__":
    analysis_and_plot(MONITOR_GOODS_ID)

运行完成后，会输出统计信息，同时在本地生成一张价格趋势图片。运营可以通过这张图看什么时候降价、什么时候恢复原价，辅助判断采购、上架时机。

五、扩展方向

上面的例子只监控单个商品，还可以继续扩展能力：

批量监控：维护一份商品编号列表，循环遍历批量获取多款竞品数据，一次性入库。
SKU 维度监控：解析返回内容中的规格列表，对每一个不同规格单独记录价格库存，监控不同配置的价差。
消息推送：检测到变动时调用企业微信 / 钉钉机器人接口，手机实时接收提醒，不用守着控制台。
导出报表：用 pandas 把数据库记录导出 csv，直接给到运营同事做周报分析。
异常容错：增加重试逻辑，网络波动自动重试，避免单次网络问题中断整个监控任务。

踩坑经验总结

不要把采集周期设置过短，过于频繁请求容易触发限流，业务场景一般 15‑60 分钟一次足够。
拿到的数据里面，部分规格价格存在偏差，业务做决策建议多轮采集交叉核对，不要只看单次快照。
商品下架、编号失效的时候会返回空结果，代码要做好判空，防止程序直接崩溃。
如果长期后台运行，需要注意磁盘占用，定期清理过久的历史快照，或者做分表。

写在最后

很多人做电商数据相关开发，大部分时间消耗在页面解析、处理各类访问限制。Open‑Claw 把这些底层脏活全部封装起来，开发者可以把精力全部放在业务逻辑上。

十几行代码完成基础读取，再搭配上 Python 的数据存储、绘图能力，快速拥有一套自用的商品监控分析工具。不管是个人玩家做薅羊毛盯价，还是中小企业做竞品跟踪，这套思路都可以直接复用。

告别手工复制粘贴：搭建电商商品自动化监控与数据分析实践

Mon, 28 Sep 2026 10:52:16 +0800

适用场景：竞品盯价、库存巡检、商品信息跟踪、运营数据分析；语言 Python；适合电商运营、数据分析、后端开发阅读

前言

做电商运营或者竞品分析，很多人还在重复一件枯燥工作：打开商品页面，复制标题、现价、原价、库存、规格信息，粘贴到 Excel 表格，每天定时手动记录一遍。

商品一旦多起来，几十个上百个商品，人工处理不仅耗时巨大，还很容易出错，价格变动、库存清零、规格更新常常错过最佳的处理时机。促销大促期间，商品信息变化飞快，依靠人工盯守几乎不可能做到实时跟进。

最近接触到 OpenClaw 这套工具，它可以把商品页面的各类信息直接转为结构化数据，省去复杂页面解析、反爬处理，我们只需要聚焦业务逻辑，快速搭建一套属于自己的数据监控脚本，实现定时采集、本地存储、简单统计分析，把重复工作交给程序完成。

本文从实际落地角度完整演示，从环境准备、编写采集脚本、数据持久化存储，再到简单数据分析，完整跑通一套可用方案。

一、工具与环境准备

Python3.7 及以上版本
依赖库安装

pip install requests pandas sqlite3

requests：负责网络请求获取结构化商品数据
pandas：做数据清洗、统计、导出报表
sqlite3：轻量数据库，用来存放历史监控记录，无需额外安装服务

拿到 OpenClaw 凭证信息，包含两组密钥，用于身份校验，替换脚本内对应位置即可。

提示：不要把密钥硬编码直接提交到公开代码仓库，生产环境建议放到环境变量中。

二、核心采集脚本编写

我们的目标是拿到商品：商品编号、标题、现价、原价、库存、规格 SKU 列表、商品链接、图片地址、店铺名称等关键业务字段，做结构化解析。

# -*- coding:utf-8 -*-
import requests
import time
import json
import sqlite3
import pandas as pd

# --------------------------配置区域，请修改--------------------------
KEY = "你的凭证key"
SECRET = "你的凭证secret"
TARGET_GOODS_ID = "10335871600"  # 需要监控的商品编号
MONITOR_INTERVAL = 1800  # 监控周期，单位秒，这里设置30分钟采集一次
# ----------------------------------------------------------------

def fetch_goods_data(goods_id):
    """获取商品结构化原始数据"""
    base_url = "https://api-gw.onebound.cn/jd/item_get_pro"
    params = {
        "key": KEY,
        "secret": SECRET,
        "num_iid": goods_id
    }
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=20)
        resp.raise_for_status()
        result = resp.json()
        return result
    except Exception as e:
        print(f"获取商品数据异常：{str(e)}")
        return None


def parse_goods_info(raw_json):
    """解析原始返回内容，提取业务需要字段"""
    if not raw_json or "item" not in raw_json:
        return None
    item = raw_json.get("item", {})
    if not item:
        return None

    data = {
        "num_iid": item.get("num_iid"),
        "title": item.get("title"),
        "price": float(item.get("price", 0)) if item.get("price") else 0,
        "orginal_price": float(item.get("orginal_price",0)) if item.get("orginal_price") else 0,
        "stock": item.get("num", 0),
        "detail_url": item.get("detail_url"),
        "pic_url": item.get("pic_url"),
        "brand": item.get("brand", ""),
        "shop_name": item.get("seller_info", {}).get("shop_name", ""),
        "capture_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    }

    # 提取多规格SKU信息
    sku_list = []
    skus_data = item.get("skus", {}).get("sku", [])
    for sku in skus_data:
        sku_item = {
            "sku_id": sku.get("sku_id"),
            "sku_price": sku.get("price"),
            "sku_original_price": sku.get("orginal_price"),
            "sku_properties": sku.get("properties_name"),
            "sku_stock": sku.get("quantity")
        }
        sku_list.append(sku_item)
    data["sku_info"] = sku_list
    return data

三、本地数据库存储：保存历史监控记录

想要做价格趋势、库存变化分析，必须把每一次采集结果落地保存，这里选用 sqlite，开箱即用。

def init_db():
    """初始化数据库表"""
    conn = sqlite3.connect("goods_monitor.db")
    cursor = conn.cursor()
    # 商品主记录表
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS goods_record (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        num_iid TEXT,
        title TEXT,
        price REAL,
        orginal_price REAL,
        stock INTEGER,
        brand TEXT,
        shop_name TEXT,
        capture_time TEXT
    )
    ''')
    # SKU规格子表，存储多规格数据
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS goods_sku (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        goods_record_id INTEGER,
        sku_id TEXT,
        sku_price REAL,
        sku_original_price REAL,
        sku_properties TEXT,
        sku_stock INTEGER
    )
    ''')
    conn.commit()
    conn.close()


def save_to_db(parsed_data):
    """将解析后的数据写入数据库"""
    if not parsed_data:
        return
    conn = sqlite3.connect("goods_monitor.db")
    cursor = conn.cursor()
    # 写入主表
    insert_sql = '''
    INSERT INTO goods_record(num_iid, title, price, orginal_price, stock, brand, shop_name, capture_time)
    VALUES (?,?,?,?,?,?,?,?)
    '''
    cursor.execute(insert_sql, (
        parsed_data["num_iid"],
        parsed_data["title"],
        parsed_data["price"],
        parsed_data["orginal_price"],
        parsed_data["stock"],
        parsed_data["brand"],
        parsed_data["shop_name"],
        parsed_data["capture_time"]
    ))
    record_id = cursor.lastrowid

    # 写入SKU子表
    for sku in parsed_data.get("sku_info", []):
        sku_sql = '''
        INSERT INTO goods_sku(goods_record_id,sku_id,sku_price,sku_original_price,sku_properties,sku_stock)
        VALUES (?,?,?,?,?,?)
        '''
        cursor.execute(sku_sql, (
            record_id,
            sku["sku_id"],
            float(sku["sku_price"]) if sku.get("sku_price") else 0,
            float(sku["sku_original_price"]) if sku.get("sku_original_price") else 0,
            sku["sku_properties"],
            sku["sku_stock"]
        ))
    conn.commit()
    conn.close()
    print(f"【{parsed_data['capture_time']}】商品 {parsed_data['num_iid']} 数据入库完成")

四、简单预警逻辑 + 定时循环监控

增加简单预警逻辑：当价格下跌超过阈值，或者库存变为 0 的时候打印提醒，实际项目可以扩展为邮件、企业微信消息推送。

def check_warning(data):
    """简单业务预警逻辑"""
    current_price = data.get("price", 0)
    stock = data.get("stock",0)
    threshold_rate = 0.8  # 低于原价80%触发降价提醒
    original = data.get("orginal_price", 0)

    if original > 0 and current_price <= original * threshold_rate:
        print(f"⚠️降价预警！商品：{data['title']}，现价{current_price}，原价{original}")
    if stock <= 0:
        print(f"⚠️库存预警！商品 {data['title']} 当前库存为0！")


def monitor_loop():
    """持续监控主循环"""
    init_db()
    print("商品监控程序已启动...")
    while True:
        raw = fetch_goods_data(TARGET_GOODS_ID)
        parsed = parse_goods_info(raw)
        if parsed:
            save_to_db(parsed)
            check_warning(parsed)
        time.sleep(MONITOR_INTERVAL)


if __name__ == "__main__":
    monitor_loop()

运行脚本之后，程序就会按照设定时间周期，自动拉取商品全部结构化信息，存入本地 sqlite 数据库，一旦触发降价、无库存条件就会打印告警信息。

五、数据分析：读取历史记录，导出报表

积累一段时间数据之后，我们可以读取数据库，做简单统计分析，导出 Excel 报表，方便运营复盘、看价格波动走势。新建 analysis.py 文件：

# analysis.py
import sqlite3
import pandas as pd

def export_history(goods_id):
    conn = sqlite3.connect("goods_monitor.db")
    df = pd.read_sql('''
        SELECT * FROM goods_record WHERE num_iid = ? ORDER BY capture_time
    ''', conn, params=(goods_id,))
    conn.close()
    if len(df) > 0:
        out_file = f"goods_{goods_id}_history.xlsx"
        df.to_excel(out_file, index=False)
        print(f"历史数据导出完成，文件:{out_file}")
        print("价格简单统计：")
        print(df[["price","orginal_price","stock"]].describe())
    else:
        print("暂无监控记录")

if __name__ == "__main__":
    export_history("10335871600")

执行后，会生成一份 Excel 文件，里面保存全部时间点采集的价格、库存、品牌店铺等记录，我们可以直接用 Excel 绘制价格折线图，直观看到一段时间商品价格波动变化。

六、实际业务拓展思路

批量监控多个商品：把商品 ID 放到列表或者配置文件，循环遍历采集，实现批量竞品巡检；
告警升级：把 print 打印告警改为企业微信、钉钉机器人推送，不用守着控制台；
数据清洗优化：SKU 多规格价格做单独监控，跟踪每个规格的价格、库存变化；
部署运行：Linux 服务器使用 nohup 后台运行脚本，实现 7×24 小时不间断监控；
可视化看板：读取 sqlite 数据接入 Grafana，直接做网页端可视化监控大盘。

七、踩坑经验总结

价格字段偶尔会存在偏差，属于平台页面动态渲染问题，做分析的时候建议多对比连续多次采集结果，不要单次结果直接判定；
密钥一定做好保管，公开代码不要明文上传；
设置合理采集间隔，不要高频请求，避免触发限制；
部分非自营店铺，部分店铺相关字段不会返回，代码做好判空，防止脚本报错中断；
商品下架时，返回内容会缺少主体信息，脚本做好空判断，保证循环不会崩溃。

写在最后

很多人一听到商品监控，就会觉得写页面解析、处理反爬会非常麻烦。借助 OpenClaw 把页面转为结构化数据，我们就不用再纠缠网页 DOM 结构、验证码、签名等复杂底层问题，开发者只需要把精力聚焦业务：存储、统计、预警、报表输出。

对于中小团队或者个人开发者，这套方案成本很低，几十行代码快速落地商品监控，把运营人员从复制粘贴表格这种重复劳动解放出来，把时间真正留给业务决策。

用 OpenClaw 把京东商品数据做成自动监控报表，全程不到一小时

Mon, 28 Sep 2026 09:57:23 +0800

做电商运营或者盯竞品的朋友应该都有体会：想搞清楚某个商品今天什么价、库存还剩多少、月销涨了没，最笨的办法就是收藏夹里存几十个链接，隔三差五挨个点开看。以前我也这么干，直到有天发现自己一天要重复打开同一个商品页七八次，实在绷不住了，决定搞个脚本把这些活全自动掉。

这篇文章记录一下我是怎么用 OpenClaw 这个数据工具把京东商品信息拿回来，再做成一个能每天自动跑、自动记录变化的监控小系统的。全程不引什么复杂框架，一个 Python 脚本为主，能跑就行。

先解决数据从哪来的问题

自己写爬虫去抓京东商品页，不是不行，但谁弄谁知道：各种加密参数、验证、反爬策略，光调试就得耗掉一两天，而且页面结构一改就得跟着改，维护成本很高。我的诉求很简单——拿商品ID换回结构化的商品信息（标题、价格、库存、销量、SKU、店铺、图片这些），越省事越好。

后来换用 OpenClaw 这种现成的数据通道，思路就完全不一样了：拿 key 和 secret 换一个请求地址，把商品ID传过去，返回的就是整理好的 JSON，字段清清楚楚。省下的时间全花在正事上——做监控和分析。

先看看一次请求能拿到什么

OpenClaw 里京东商品详情这块，入口参数就一个核心的：商品ID（就是链接里那段数字，比如 100126351037）。请求地址大概是这个套路（地址和密钥开通之后会拿到，下面先用占位符表示）：

BASE_URL = "https://你的服务地址/jd/item_get_pro"
KEY = "你的key"
SECRET = "你的secret"

带上参数发个 GET 请求，返回的 JSON 里关键的字段大概是这些：

title 商品标题，一眼看出是什么东西

price / orginal_price 当前价和原价（注意文档里也写了，价格字段存在对不上的情况，后面会讲怎么处理）

num 库存数量

sales 销量，很多商品有这个字段，可以拿来做趋势

skus 规格列表，每个规格的 id、价格、库存都在里面，多规格商品要重点看这个

item_imgs 商品图片

seller_info 店铺信息

location 发货地，选品的时候有用

这些字段基本覆盖了日常监控 90% 的需求。下面直接上代码。

写一个最朴素的取数脚本

依赖就一个 requests，装好就能跑：

import requests

BASE_URL = "https://你的服务地址/jd/item_get_pro"
KEY = "你的key"
SECRET = "你的secret"

def fetch_item(num_iid):
    params = {
        "key": KEY,
        "secret": SECRET,
        "num_iid": num_iid,
    }
    resp = requests.get(BASE_URL, params=params, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    return data.get("item") or {}

第一次跑通的时候还挺有成就感的，一个商品ID换回一坨结构化数据，直接能用。接下来就顺手把它存下来。

把快照存下来

监控的前提是有历史。我的做法很简单：每次取数就把当天的"快照"追加到 SQLite 里，一张表搞定，不用上什么重型数据库。

import sqlite3
from datetime import datetime

def init_db():
    conn = sqlite3.connect("jd_monitor.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS snapshots (
            num_iid  TEXT,
            ts       TEXT,
            title    TEXT,
            price    REAL,
            orginal_price REAL,
            stock    INTEGER,
            sales    INTEGER
        )
    """)
    conn.commit()
    return conn

def save_snapshot(conn, num_iid, item):
    conn.execute(
        "INSERT INTO snapshots (num_iid, ts, title, price, orginal_price, stock, sales) VALUES (?,?,?,?,?,?,?)",
        (
            num_iid,
            datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            item.get("title"),
            item.get("price"),
            item.get("orginal_price"),
            item.get("num"),
            item.get("sales"),
        ),
    )
    conn.commit()

定时自动跑

我自己是每天 9 点、14 点、21 点各跑一次，用系统自带的定时任务就够了，没必要为了这个专门引一个调度框架：

# crontab 里加三行即可
# 0 9 * * * cd /path/to/script && python3 monitor.py
# 0 14 * * * cd /path/to/script && python3 monitor.py
# 0 21 * * * cd /path/to/script && python3 monitor.py

monitor.py 的主逻辑就是遍历我收藏的那批商品ID，逐个取数、落库：

WATCH_LIST = ["10335871600", "100126351037"]  # 换成你要盯的商品ID

def main():
    conn = init_db()
    for num_iid in WATCH_LIST:
        try:
            item = fetch_item(num_iid)
            save_snapshot(conn, num_iid, item)
            print(f"[OK] {num_iid} {item.get('title', '')[:20]} {item.get('price')}")
        except Exception as e:
            print(f"[FAIL] {num_iid}: {e}")
    conn.close()

if __name__ == "__main__":
    main()

让数据自己说话

跑了几天之后，数据开始有点意思了。比如我能直接回答这种问题：这个商品最近 7 天最低价是多少？什么时候降的？SKU 里的某个规格是不是一直缺货？

def price_trend(conn, num_iid):
    rows = conn.execute(
        """SELECT ts, price, stock, sales FROM snapshots
           WHERE num_iid=? ORDER BY ts""",
        (num_iid,),
    ).fetchall()
    for ts, price, stock, sales in rows:
        print(ts, "价格:", price, "库存:", stock, "销量:", sales)

想更直观一点，用 matplotlib 画个价格曲线，几行代码的事，这里就不展开贴图了。降价提醒也可以做：如果最新价格比历史最低价还低，就往群里发个通知，抢券补货都来得及。

踩过的坑，帮你提前避雷

1.价格字段别全信。文档里原话是"当前商品id价格【价格会对不上】"，尤其多规格商品，price 这个字段可能是某个规格的价，真正准的是 skus 里逐条对。拿不准的时候就遍历 skus，按你自己的规则取价。

2.有缓存。默认走的是缓存数据，速度快但可能不是最新，想要更实时的数据记得把 cache 参数关掉（cache=no），代价是慢一点、调用量上去了，看你的场景取舍。

3.别高频猛打。我一开始图省事写了个 5 分钟一轮的循环，结果跑一会儿就报错被限了。监控类需求几分钟和几小时的差别其实没那么大，改成每天固定几次，稳稳的。

4.商品下架/改ID。有些商品链接会失效，脚本里记得 try/except，单条失败别让整个任务崩掉，日志打出来就行。

最后说两句

其实整个过程拆开看没什么高深的：一个现成的数据通道解决取数，SQLite 解决存储，cron 解决定时，剩下的就是根据你自己的业务去定义"什么变化值得关注"。这套东西我跑了快一个月，省下来的时间早就够写这篇文章了。