告别人工盯品！Open Claw 搭建京东商品全自动监控与数据分析系统（附完整可运行代码）

前言：电商人都逃不开的数据痛点

做京东店铺运营、竞品调研、带货选品的朋友一定深有体会：每天手动打开几十款竞品页面，记录现价、原价、库存、销量、规格参数，一旦遇上大促、对手调价、库存清仓，信息差直接导致利润受损。人工统计不仅耗时 3 小时以上，还容易漏看、记错，无法追溯历史价格波动，更没法批量对比同类商品数据。

传统浏览器自动化抓取方案又存在大量短板：页面改版失效、频繁滑块验证、IP 限制封禁、数据不全拿不到多规格库存与历史售价，调试和维护成本极高。

今天分享一套轻量化落地方案，借助 Open Claw 工具，无需复杂逆向、不用搭建爬虫集群，本地电脑即可运行，实现7×24 小时无人值守商品监控 + 历史数据存储 + 自动异动预警 + 批量数据分析，新手复制代码就能直接跑通，适配个人卖家、中小运营、数据从业者。

一、方案整体逻辑与核心优势

整体流程

提前获取专属身份凭证，打通稳定商品数据通道；
Python 脚本结合 Open Claw 定时调度，循环采集目标商品全维度信息；
清洗结构化数据存入本地数据库，留存完整历史记录；
自定义价格、库存、销量波动阈值，触发异动预警；
基于历史数据做可视化分析，输出价格趋势、竞品对比报表。

这套方案的核心亮点

稳定无反爬干扰：区别于页面抓取，数据来源渠道成熟稳定，不会因页面改版、验证码中断采集；
全维度完整字段：一次性获取标题、品牌、原价、实时售价、多规格 SKU、库存、店铺信息、商品图、发货地等全部经营所需数据；
轻量化部署：仅需 Python 环境，无需服务器，Windows/Mac 本地运行；
自动化调度：Open Claw 支持自定义轮询间隔（10 分钟 / 1 小时 / 每日），后台静默执行；
数据可追溯：本地 SQLite 存储所有历史采集记录，随时导出 Excel 做深度分析；
低成本：按量消耗资源，小体量监控场景几乎无额外成本。

二、前期环境准备

1. 软件依赖安装

打开终端 / CMD 执行 pip 安装所需库：

pip install requests pandas sqlite3 matplotlib openclaw

2. 前置准备工作

登录工具后台，申请两组专属身份凭证（下文代码中KEY、SECRET替换为自己的凭证）；
提取需要监控的京东商品 ID（商品详情页链接item.jd.com/xxxx.html中间数字即为商品 ID）；
新建本地项目文件夹，新建monitor.py主程序、goods_monitor.db自动生成数据库。

三、完整实战代码（分四大模块）

模块 1：核心数据采集类（获取商品结构化数据）

import requests
import json
import time
import sqlite3
import pandas as pd
import matplotlib.pyplot as plt
from openclaw import TaskScheduler

# 替换为你后台获取的专属凭证
KEY = "你的专属凭证1"
SECRET = "你的专属凭证2"

class GoodsCollector:
    def __init__(self):
        self.session = requests.Session()
        # 固定数据请求地址
        self.base_url = "https://api-gw.onebound.cn/jd/item_get_pro"

    def get_goods_info(self, goods_id, real_time=True):
        """
        采集单款商品完整数据
        :param goods_id: 京东商品ID
        :param real_time: True=关闭缓存，获取实时价格库存（监控必开）
        :return: 清洗后的结构化商品字典
        """
        params = {
            "key": KEY,
            "secret": SECRET,
            "num_iid": goods_id,
            "cache": "no" if real_time else "yes",
            "result_type": "json"
        }
        retry_count = 3  # 失败自动重试3次
        for i in range(retry_count):
            try:
                resp = self.session.get(self.base_url, params=params, timeout=15)
                resp.raise_for_status()
                res_data = resp.json()
                # 判断数据是否正常返回
                if res_data.get("error_code") != "0000":
                    print(f"商品{goods_id}采集失败：{res_data.get('reason')}")
                    return None
                raw_item = res_data.get("item", {})
                sku_list = raw_item.get("skus", {}).get("sku", [])
                # 清洗核心字段
                clean_data = {
                    "goods_id": raw_item.get("num_iid", ""),
                    "title": raw_item.get("title", ""),
                    "brand": raw_item.get("brand", ""),
                    "current_price": float(raw_item.get("price", 0)),
                    "origin_price": float(raw_item.get("orginal_price", 0)),
                    "stock": int(raw_item.get("num", 0)),
                    "sales": int(raw_item.get("sales", 0)),
                    "shop_name": raw_item.get("seller_info", {}).get("shop_name", ""),
                    "location": raw_item.get("location", ""),
                    "main_img": raw_item.get("pic_url", ""),
                    "crawl_time": time.strftime("%Y-%m-%d %H:%M:%S"),
                    "sku_info": json.dumps(sku_list, ensure_ascii=False)
                }
                return clean_data
            except Exception as e:
                print(f"第{i+1}次采集异常：{str(e)}，30秒后重试")
                time.sleep(30)
        print(f"商品{goods_id}多次采集失败，跳过本条")
        return None

模块 2：本地数据库存储模块（留存历史数据）

class DataStorage:
    def __init__(self, db_name="goods_monitor.db"):
        self.conn = sqlite3.connect(db_name, check_same_thread=False)
        self.cursor = self.conn.cursor()
        self.init_table()

    def init_table(self):
        # 创建商品监控记录表
        create_sql = """
        CREATE TABLE IF NOT EXISTS goods_record (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            goods_id TEXT NOT NULL,
            title TEXT,
            brand TEXT,
            current_price REAL,
            origin_price REAL,
            stock INTEGER,
            sales INTEGER,
            shop_name TEXT,
            location TEXT,
            main_img TEXT,
            sku_info TEXT,
            crawl_time TEXT
        )
        """
        self.cursor.execute(create_sql)
        self.conn.commit()

    def insert_record(self, data):
        """单条数据入库"""
        if not data:
            return
        insert_sql = """
        INSERT INTO goods_record 
        (goods_id,title,brand,current_price,origin_price,stock,sales,shop_name,location,main_img,sku_info,crawl_time)
        VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
        """
        val = (
            data["goods_id"], data["title"], data["brand"], data["current_price"],
            data["origin_price"], data["stock"], data["sales"], data["shop_name"],
            data["location"], data["main_img"], data["sku_info"], data["crawl_time"]
        )
        self.cursor.execute(insert_sql, val)
        self.conn.commit()

    def get_history_by_goods(self, goods_id):
        """查询单商品全部历史数据，用于趋势分析"""
        sql = "SELECT crawl_time,current_price,stock,sales FROM goods_record WHERE goods_id=? ORDER BY crawl_time ASC"
        df = pd.read_sql(sql, self.conn, params=(goods_id,))
        return df

    def close(self):
        self.conn.close()

模块 3：异动预警 + 数据分析可视化模块

class GoodsAnalysis:
    def __init__(self, storage):
        self.storage = storage
        # 自定义预警阈值，可自行修改
        self.price_drop_rate = 0.1  # 降价超10%触发提醒
        self.stock_warn_num = 50   # 库存低于50触发预警

    def check_warning(self, new_data):
        """对比历史数据，检测价格/库存异动"""
        goods_id = new_data["goods_id"]
        history_df = self.storage.get_history_by_goods(goods_id)
        if len(history_df) < 1:
            print(f"商品{new_data['title']}暂无历史数据，无预警")
            return
        # 获取上一轮采集数据
        last_record = history_df.iloc[-1]
        old_price = last_record["current_price"]
        old_stock = last_record["stock"]
        new_price = new_data["current_price"]
        new_stock = new_data["stock"]

        # 降价预警
        if old_price > 0 and (old_price - new_price) / old_price >= self.price_drop_rate:
            print("="*50)
            print(f"【价格异动预警】商品：{new_data['title']}")
            print(f"原售价：{old_price} 现价：{new_price} 降幅超10%")
            print("="*50)
        # 库存预警
        if new_stock <= self.stock_warn_num and old_stock > self.stock_warn_num:
            print("="*50)
            print(f"【库存预警】商品：{new_data['title']} 当前库存仅{new_stock}件")
            print("="*50)

    def draw_price_trend(self, goods_id, save_path="price_trend.png"):
        """绘制商品历史价格趋势图"""
        df = self.storage.get_history_by_goods(goods_id)
        if len(df) < 2:
            print("历史数据不足，无法生成趋势图")
            return
        plt.rcParams["font.sans-serif"] = ["SimHei"]
        plt.figure(figsize=(12, 6))
        plt.plot(df["crawl_time"], df["current_price"], marker="o", color="#ff4444", label="实时售价")
        plt.title(f"商品{goods_id}历史价格走势", fontsize=14)
        plt.xlabel("采集时间")
        plt.ylabel("商品售价（元）")
        plt.xticks(rotation=45)
        plt.legend()
        plt.tight_layout()
        plt.savefig(save_path)
        print(f"价格趋势图已保存至：{save_path}")

模块 4：Open Claw 定时调度主程序

def main_task():
    """单次采集任务，由Open Claw定时循环执行"""
    collector = GoodsCollector()
    storage = DataStorage()
    analysis = GoodsAnalysis(storage)
    # 填入你需要监控的商品ID列表，可批量添加
    monitor_goods_list = ["10335871600", "10057467958584"]

    for goods_id in monitor_goods_list:
        print(f"\n=====正在采集商品{goods_id}=====")
        goods_data = collector.get_goods_info(goods_id)
        if goods_data:
            storage.insert_record(goods_data)
            analysis.check_warning(goods_data)
    storage.close()
    print("\n本轮采集任务执行完成，等待下一轮调度")

if __name__ == "__main__":
    # 初始化Open Claw定时任务调度器
    scheduler = TaskScheduler()
    # 设置每30分钟执行一轮监控任务，可修改interval单位：秒
    scheduler.add_task(task=main_task, interval=1800)
    print("商品监控系统已启动，每30分钟自动采集一轮数据...")
    # 后台持续运行调度
    scheduler.run_forever()

四、代码运行与实操步骤

将代码全部保存为monitor.py，替换代码内KEY、SECRET为自己后台获取的凭证；
修改monitor_goods_list数组，填入所有需要监控的竞品 / 自有商品 ID；
自定义预警规则：调整price_drop_rate降价阈值、stock_warn_num库存预警线；
终端执行启动命令：

python monitor.py

5.程序启动后后台静默循环采集，每一轮自动入库、打印异动提醒；

6.如需查看某款商品价格走势，在主程序新增调用代码：

# 查看指定商品趋势图
storage = DataStorage()
analysis = GoodsAnalysis(storage)
analysis.draw_price_trend("10335871600")

7.数据库goods_monitor.db可使用 DB Browser 打开，一键导出全部历史数据为 Excel，用于月度竞品复盘。

五、常见异常问题排查

采集返回商品未找到：核对商品 ID 是否正确，确认商品未下架；
凭证校验失败：检查 KEY、SECRET 是否复制完整，无多余空格；
请求超时：本地网络波动，程序自带 3 次自动重试，等待即可；
图表中文乱码：Windows 自带宋体，Mac 需安装中文字体并修改plt.rcParams字体名称；
采集数据缓存未更新：代码默认开启real_time=True强制拉取实时数据，无需额外修改。

六、延伸拓展玩法

多规格 SKU 深度分析：解析sku_info字段，拆分不同尺码、颜色的独立售价与库存，对比竞品规格定价策略；
批量导出报表：基于 pandas 将数据库数据输出 Excel，自动计算 7 天均价、最大降幅、库存波动区间；
消息推送预警：对接企业微信 / 钉钉机器人，价格异动时自动推送消息到工作群，无需盯着程序终端；
多平台扩展：同一套工具可适配其他主流电商平台，仅需替换采集地址与字段清洗逻辑；
长期选品库搭建：新增关键词批量采集逻辑，自动筛选高销量、高优惠潜力新品，持续扩充选品池。

结语

对于中小电商运营、数据从业者来说，不需要投入高昂的商用数据工具，借助 Open Claw 搭配这套轻量化采集分析代码，就能搭建一套完全自主可控的商品监控体系。全程规避传统爬虫的各类封禁、改版风险，数据完整稳定，自动化流程解放大量人工统计时间，历史数据趋势分析也能为定价、备货、竞品策略提供客观数据支撑。

代码可直接复制运行，根据自身监控商品数量、预警需求微调参数，本地长期稳定运行，是低成本电商数据化运营的最优落地方案之一。

万邦api博客

Nice to meet you, too!

api 数据挖掘