×

api 数据挖掘

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

admin admin 发表于2026-08-06 17:49:23 浏览7 评论0

抢沙发发表评论

一、前言：电商人都绕不开的数据痛点

不管是电商运营、无货源商家、选品分析师还是普通消费者，长期盯守商品动态是一件极度消耗时间的事：

  2. 竞品调价、大促降价只能靠手动刷新，错过最优定价 / 入手时机；

  4. 多款商品同时跟踪时，销量、库存、规格变动无法统一记录；

  6. 想要复盘历史定价、分析竞品营销节奏，缺少完整时序数据支撑；

  8. 原生页面抓取容易触发访问限制、页面改版后代码直接失效。

过去尝试过浏览器自动化、页面解析脚本，要么稳定性差，要么维护成本极高。后来借助 Open Claw 数据工具，仅几十行代码就能实现 7×24 小时不间断商品数据采集、异动告警、历史数据存储与趋势分析，无需复杂爬虫框架，新手零门槛落地。

本文完整拆解全流程方案：工具配置、数据拉取、定时监控、异动预警、本地存储、简单数据分析，附带可直接运行的 Python 完整源码，适配 CSDN、知乎技术内容发布。

二、方案整体架构说明

整套系统分为四大模块，逻辑简单清晰：

  2. 数据获取层：通过 Open Claw 内置数据通道，传入商品唯一编号，一键拉取全维度商品信息（标题、原价、现价、库存、各规格 SKU、店铺信息、主图、发货地、销量等）；

  4. 定时监控层：定时轮询目标商品，对比上一轮数据，识别价格、库存、销量异动；

  6. 持久化存储层：轻量 SQLite 数据库保存所有时序数据，用于后续数据分析；

  8. 数据分析层：读取历史库，计算价格波动幅度、库存变化、热销规格统计，输出分析报表。

优势亮点：

  • 自带缓存机制，高频监控不会频繁消耗额度，静态信息自动延时刷新；

  • 内置异常容错，商品下架、网络波动、平台维护均有对应返回标识，程序不会崩溃；

  • 全规格 SKU 完整返回，多尺寸多颜色商品可单独跟踪每一个规格价格库存；

  • 支持多语言输出，可按需切换简体 / 英文数据文本。

三、前期准备工作

3.1 工具权限开通

登录后台，完成账号注册后创建专属凭证两组字符串（下文代码中KEY、SECRET替换使用），开通京东商品详情数据能力，后台可查看每日可用调用额度。

3.2 本地环境依赖安装

本机安装 Python3.7 及以上版本，执行命令安装所需工具包：

pip install requests apscheduler sqlite3

  • requests：负责网络数据拉取；

  • apscheduler：实现定时循环监控；

  • sqlite3：内置轻量数据库，无需额外安装服务，本地存储历史数据。

3.3 获取商品唯一标识

打开京东商品详情页，地址栏数字串即为商品编号（例：https://item.jd.com/10335871600.html，编号为10335871600），可批量录入监控列表。

四、完整可运行 Python 代码实现

4.1 基础工具类：获取商品完整数据

创建jd_monitor.py，第一部分封装数据拉取函数，自动处理请求参数、异常捕获、数据解析：

import requests
import json
import sqlite3
import time
from apscheduler.schedulers.background import BackgroundScheduler

# ===================== 配置区（自行修改） =====================
# Open Claw后台获取的专属凭证
CLAW_KEY = "你的专属KEY"
CLAW_SECRET = "你的专属SECRET"
# 需要监控的商品编号列表，支持批量监控多个商品
MONITOR_GOODS_IDS = ["10335871600", "10057467958584"]
# 监控轮询间隔（单位：秒，建议普通商品300秒，爆款60秒）
POLL_SECONDS = 300
# 价格跌幅预警阈值（0.2代表降价超过20%触发提醒）
PRICE_ALERT_RATE = 0.2
# ==========================================================

def fetch_goods_detail(goods_id: str):
    """
    通过Open Claw通道获取京东商品完整数据
    :param goods_id: 商品数字编号
    :return: 格式化商品字典 / None（失败）
    """
    base_url = "https://api-gw.onebound.cn/jd/item_get_pro"
    params = {
        "key": CLAW_KEY,
        "secret": CLAW_SECRET,
        "num_iid": goods_id,
        "result_type": "json",
        "cache": "yes",
        "lang": "cn"
    }
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(url=base_url, params=params, headers=headers, timeout=15)
        resp_data = resp.json()
        # 判断异常状态
        if resp_data.get("error_code") != "0000":
            print(f"【商品{goods_id}拉取失败】错误信息：{resp_data.get('reason')}")
            return None
        item_info = resp_data.get("item", {})
        if not item_info:
            print(f"【商品{goods_id}】无商品数据，可能已下架")
            return None
        # 提取核心业务字段
        sku_list = []
        if "skus" in item_info and "sku" in item_info["skus"]:
            sku_list = item_info["skus"]["sku"]
        clean_data = {
            "goods_id": item_info.get("num_iid"),
            "title": item_info.get("title"),
            "brand": item_info.get("brand", ""),
            "current_price": float(item_info.get("price", 0)),
            "origin_price": float(item_info.get("orginal_price", 0)),
            "stock_total": int(item_info.get("num", 0)),
            "sales_total": int(item_info.get("sales", 0)),
            "shop_name": item_info.get("seller_info", {}).get("shop_name", ""),
            "location": item_info.get("location", ""),
            "main_img": item_info.get("pic_url", ""),
            "sku_detail": sku_list,
            "crawl_time": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        return clean_data
    except Exception as e:
        print(f"【网络异常】商品{goods_id} 请求报错：{str(e)}")
        return None

4.2 数据库模块：历史数据持久化存储

创建数据表，每次采集自动写入数据库，用于后续趋势分析：

def init_sqlite_db():
    """初始化本地数据库，创建商品监控数据表"""
    conn = sqlite3.connect("jd_goods_monitor.db")
    cursor = conn.cursor()
    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_total INTEGER,
        sales_total INTEGER,
        shop_name TEXT,
        crawl_time TEXT
    )
    """
    cursor.execute(create_sql)
    conn.commit()
    conn.close()

def save_data_to_db(data: dict):
    """将单次采集数据存入数据库"""
    if not data:
        return
    conn = sqlite3.connect("jd_goods_monitor.db")
    cursor = conn.cursor()
    insert_sql = """
    INSERT INTO goods_record 
    (goods_id, title, brand, current_price, origin_price, stock_total, sales_total, shop_name, crawl_time)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
    """
    params = (
        data["goods_id"], data["title"], data["brand"],
        data["current_price"], data["origin_price"], data["stock_total"],
        data["sales_total"], data["shop_name"], data["crawl_time"]
    )
    cursor.execute(insert_sql, params)
    conn.commit()
    conn.close()

4.3 监控核心逻辑：异动检测 + 告警提醒

对比上一轮采集数据，识别降价、库存清零、销量暴涨等变动并输出提醒：

# 全局缓存：存储上一轮采集的数据，用于对比异动
last_record_cache = {}

def monitor_single_goods(goods_id: str):
    """单商品监控逻辑：拉取数据、入库、对比异动"""
    current_data = fetch_goods_detail(goods_id)
    if not current_data:
        return
    # 写入数据库
    save_data_to_db(current_data)
    now_time = current_data["crawl_time"]
    title = current_data["title"]
    curr_price = current_data["current_price"]
    curr_stock = current_data["stock_total"]
    curr_sales = current_data["sales_total"]
    print(f"\n===== {now_time} 商品采集：{title} =====")
    print(f"现价：{curr_price} ｜ 原价：{current_data['origin_price']}")
    print(f"总库存：{curr_stock} ｜ 累计销量：{curr_sales}")

    # 对比历史数据，检测异动
    if goods_id in last_record_cache:
        old_data = last_record_cache[goods_id]
        old_price = old_data["current_price"]
        old_stock = old_data["stock_total"]
        old_sales = old_data["sales_total"]
        # 价格变动检测
        price_change_rate = abs(old_price - curr_price) / max(old_price, 0.01)
        if curr_price < old_price and price_change_rate >= PRICE_ALERT_RATE:
            print(f"⚠️【降价预警】降幅超过{PRICE_ALERT_RATE*100}%！原价{old_price} → 现价{curr_price}")
        elif curr_price > old_price:
            print(f"📈【涨价提醒】{old_price} → {curr_price}")
        # 库存变动
        if curr_stock < old_stock and curr_stock == 0:
            print(f"🚨【缺货预警】商品库存清零！原库存{old_stock}")
        elif curr_stock > old_stock:
            print(f"📦【补货通知】库存增加 {curr_stock - old_stock}")
        # 销量异动
        sales_grow = curr_sales - old_sales
        if sales_grow > 100:
            print(f"🔥【销量暴涨】本次周期新增销量{sales_grow}，疑似爆款活动")
    # 更新缓存
    last_record_cache[goods_id] = current_data

def run_all_monitor_task():
    """批量执行所有商品监控任务"""
    for gid in MONITOR_GOODS_IDS:
        monitor_single_goods(gid)

4.4 数据分析工具：读取历史库生成价格趋势统计

新增分析函数，读取本地数据库，输出商品历史价格区间、最低 / 最高价、平均售价：

def analyze_goods_history(goods_id: str):
    """针对单个商品做历史数据统计分析"""
    conn = sqlite3.connect("jd_goods_monitor.db")
    cursor = conn.cursor()
    query_sql = """
    SELECT current_price, crawl_time FROM goods_record WHERE goods_id = ? ORDER BY crawl_time ASC
    """
    cursor.execute(query_sql, (goods_id,))
    records = cursor.fetchall()
    conn.close()
    if not records:
        print(f"商品{goods_id}暂无历史采集数据，无法分析")
        return
    price_list = [row[0] for row in records]
    min_price = min(price_list)
    max_price = max(price_list)
    avg_price = round(sum(price_list)/len(price_list), 2)
    first_time = records[0][1]
    last_time = records[-1][1]
    print(f"\n===== 商品{goods_id}历史数据分析报告 =====")
    print(f"数据采集周期：{first_time} ~ {last_time}")
    print(f"历史最低价：{min_price}")
    print(f"历史最高价：{max_price}")
    print(f"周期平均售价：{avg_price}")
    print(f"总采集记录条数：{len(records)}")
    # 找出最低价出现时间
    min_time = [r[1] for r in records if r[0]==min_price][0]
    print(f"历史低价出现时间：{min_time}")

4.5 程序入口：启动定时监控

if __name__ == "__main__":
    # 初始化数据库
    init_sqlite_db()
    print("系统初始化完成，开始启动监控任务...")
    # 创建定时调度器
    scheduler = BackgroundScheduler()
    # 定时执行全商品采集任务
    scheduler.add_job(run_all_monitor_task, "interval", seconds=POLL_SECONDS)
    scheduler.start()
    print(f"定时监控已启动，每{POLL_SECONDS}秒采集一次数据，按Ctrl+C停止程序")
    print("输入 analyze 可手动执行数据分析，输入 exit 退出程序\n")
    # 控制台交互
    while True:
        cmd = input("请输入指令：").strip()
        if cmd.startswith("analyze"):
            target_gid = cmd.replace("analyze", "").strip()
            if target_gid:
                analyze_goods_history(target_gid)
            else:
                print("指令格式示例：analyze 10335871600")
        elif cmd == "exit":
            scheduler.shutdown()
            print("监控程序已关闭")
            break
        else:
            print("无效指令，支持指令：analyze 商品编号 / exit")

五、代码运行与功能实操说明

5.1 启动监控

  2. 将代码中KEY、SECRET替换为自己后台凭证；

  4. 修改MONITOR_GOODS_IDS填入需要跟踪的商品编号；

  6. 终端执行python jd_monitor.py，程序自动创建本地数据库并开始定时采集；

  8. 控制台实时打印每次采集的价格、库存、销量，出现异动自动弹出彩色提醒。

5.2 数据分析使用

程序运行中，在控制台输入指令：

analyze 10335871600

即可读取该商品全部历史时序数据，输出完整价格趋势分析报表，用来判断历史低价、大促调价规律。

5.3 核心功能拓展方向

  2. 消息通知：对接企业微信 / 邮件推送降价、缺货告警，无需盯着控制台；

  4. SKU 细分监控：循环遍历每个规格单独存储，跟踪不同尺寸 / 颜色定价差异；

  6. Excel 报表导出：读取 SQLite 数据，用 pandas 导出 CSV 表格用于线下复盘；

  8. 可视化图表：接入 Matplotlib 绘制价格波动折线图，直观展示定价周期；

  10. 多平台兼容：替换对应数据通道参数，同步监控淘宝、拼多多商品。

六、踩坑优化方案（实操避坑指南）

  2. 采集频繁被限制：开启代码中cache="yes"缓存参数，静态商品信息复用缓存，减少实时请求；爆款缩短轮询间隔，滞销商品拉长间隔；

  4. 商品返回空数据：商品编号输入错误、商品下架、当日额度耗尽均会返回提示，程序不会直接崩溃；

  6. 价格数据匹配不准确：工具返回独立 SKU 价格数组，如需精准单规格价格，循环读取sku_detail数组单独解析；

  8. 长时间运行内存堆积：本地 SQLite 分表按月存储，定期清理过期 7 天以上无用历史数据；

  10. 网络超时中断：代码内置 15 秒超时捕获，单次失败保留上一轮缓存数据，下一轮自动重试。

七、落地场景总结

这套基于 Open Claw 搭建的监控分析工具，适配三类人群高频需求：

  2. 电商运营 / 商家：批量监控竞品定价、库存、促销节奏，提前制定店铺调价策略；

  4. 无货源选品玩家：通过销量暴涨预警快速捕捉潜力爆款，分析类目定价区间；

  6. 普通消费者：长期盯守数码、家电、服饰，自动捕捉历史低价，避免错过大促；

  8. 数据分析师：自动沉淀时序商品数据库，快速完成市场定价、竞品营销数据分析。

全程无需搭建复杂爬虫、无需处理页面反爬、不用维护页面选择器，一套脚本即可实现全天候自动化数据采集与分析，低成本解决人工盯品低效痛点。


少长咸集

群贤毕至

访客