×

api 数据挖掘

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

admin admin 发表于2026-08-14 16:41:43 浏览10 评论0

抢沙发发表评论

一、前言：电商运营数据采集的痛点

做电商运营、竞品分析、货源采购的朋友几乎都有同一个困扰：

  2. 手动打开商品页记录价格、库存、促销活动，几十款竞品每天重复操作，耗时 2 小时以上；

  4. 平台反爬机制严格，普通页面抓取极易触发限制，频繁出现数据缺失、程序崩溃；

  6. 零散数据无法沉淀，没法做价格走势、销量波动、SKU 规格对比等深度分析；

  8. 竞品大促降价、库存清仓时无法及时感知，错过调价、活动跟进最佳时机。

过去尝试过浏览器自动化、页面解析抓取，要么运行卡顿、占用大量服务器资源，要么频繁被平台限制访问。后来接触到 Open Claw 这套标准化数据采集工具，无需复杂逆向、无需代理池，一行请求就能拉取商品全维度结构化数据，结合 Python 即可搭建7×24 小时自动监控 + 数据存储 + 波动预警 + 可视化分析一体化系统，中小企业、个人开发者都能快速落地。

本文全程实操，无晦涩理论，附完整可运行 Python 代码，零基础也能跟着搭建。

二、方案整体架构设计

整套系统分为四层，轻量化部署，本地电脑、云服务器均可运行：

  2. 数据采集层：依靠 Open Claw 稳定拉取商品结构化数据（标题、实时售价、原价、库存、SKU 规格、店铺信息、主图、发货地等全字段）；

  4. 数据持久层：使用 CSV / 本地轻量数据库存储每一次采集快照，留存历史时序数据；

  6. 监控预警层：对比前后两次采集数据，识别价格下调、库存骤减、新增促销活动等异常，打印预警信息（可扩展企业微信 / 短信推送）；

  8. 数据分析层：基于历史时序数据，用 Pandas、Matplotlib 完成价格走势、SKU 销量对比、类目均价分析，输出可视化图表。

对比传统页面抓取方案，这套方案三大核心优势：

  • 稳定性强：内置缓存与重试机制，规避平台访问限制，不会出现页面加载失败、字段漏取；

  • 字段完整：一次性获取商品所有规格、店铺资质、图文素材，无需多轮请求拼接；

  • 开发成本极低：无需处理 JS 渲染、Cookie、请求签名，代码量减少 70%。

三、前期准备工作

3.1 环境依赖安装

本地安装 Python3.8 及以上版本，执行命令安装依赖库：

pip install requests pandas matplotlib openpyxl

3.2 工具凭证配置

登录后台，获取两组专属身份凭证（后文代码替换对应参数即可）：

3.3 监控商品清单

整理需要监控的京东商品 ID（商品详情页链接末尾一串数字），支持批量添加多款商品持续轮询。

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

模块 1：基础数据采集工具类（核心拉取逻辑）

封装商品详情获取方法，内置失败重试、缓存控制、异常捕获，自动格式化返回干净结构化数据。

import requests
import json
import time
from datetime import datetime

# ===================== 配置区，自行修改 =====================
CLAW_KEY = "你的后台访问密钥"
CLAW_SECRET = "你的后台安全密钥"
# 需要监控的商品ID列表
MONITOR_GOODS_IDS = ["10335871600", "10057467958584"]
# 轮询间隔，单位秒；爆款建议60秒，普通商品300秒
POLL_INTERVAL = 300
# 价格预警阈值：降价超过20%触发提醒（0.2=20%）
PRICE_ALERT_RATIO = 0.2
# ==========================================================

class GoodsDataCollector:
    def __init__(self):
        self.key = CLAW_KEY
        self.secret = CLAW_SECRET
        self.base_path = "https://api-gw.onebound.cn/jd/item_get_pro"
        self.session = requests.Session()
        # 历史数据缓存，用于对比波动
        self.history_data = {}

    def fetch_single_goods(self, goods_id: str, real_time: bool = False):
        """
        获取单款商品完整结构化数据
        :param goods_id: 商品数字ID
        :param real_time: True=关闭缓存，拉取实时最新价格库存（监控时开启）
        :return: 格式化商品字典，失败返回None
        """
        params = {
            "key": self.key,
            "secret": self.secret,
            "num_iid": goods_id,
            "result_type": "json",
            "cache": "no" if real_time else "yes",
            "lang": "cn"
        }
        headers = {
            "Accept-Encoding": "gzip",
            "Connection": "close"
        }
        retry_count = 3  # 失败自动重试3次
        for attempt in range(retry_count):
            try:
                resp = self.session.get(
                    url=self.base_path,
                    params=params,
                    headers=headers,
                    timeout=15
                )
                resp.raise_for_status()
                raw_data = resp.json()
                # 判断请求是否报错
                if raw_data.get("error_code") != "0000":
                    err_msg = raw_data.get("error", "未知采集失败")
                    print(f"【采集失败】商品{goods_id}，错误信息：{err_msg}")
                    return None
                # 提取商品核心数据
                item_info = raw_data.get("item", {})
                if not item_info:
                    print(f"【警告】商品{goods_id}无商品数据，可能已下架")
                    return None
                # 格式化基础字段
                clean_data = {
                    "crawl_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                    "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)),
                    "original_price": float(item_info.get("orginal_price", 0)),
                    "stock_num": int(item_info.get("num", 0)),
                    "shop_name": item_info.get("nick") or item_info.get("seller_info", {}).get("shop_name", ""),
                    "location": item_info.get("location", "无"),
                    "detail_url": item_info.get("detail_url"),
                    "sku_list": item_info.get("skus", {}).get("sku", []),
                    "sales": int(item_info.get("sales", 0))
                }
                return clean_data
            except Exception as e:
                print(f"【请求异常】第{attempt+1}次重试，商品{goods_id}，异常：{str(e)}")
                time.sleep(2)
        print(f"【采集彻底失败】商品{goods_id}已重试3次，跳过")
        return None

    def check_price_warning(self, goods_id: str, new_data: dict):
        """对比历史数据，检测价格大幅下跌预警"""
        if goods_id not in self.history_data:
            self.history_data[goods_id] = new_data
            return
        old_price = self.history_data[goods_id]["current_price"]
        new_price = new_data["current_price"]
        if old_price <= 0:
            self.history_data[goods_id] = new_data
            return
        # 计算降价幅度
        drop_ratio = (old_price - new_price) / old_price
        if drop_ratio >= PRICE_ALERT_RATIO:
            print("=" * 60)
            print(f"【价格预警！】商品：{new_data['title']}")
            print(f"原价格：{old_price} 元，现价：{new_price} 元，降幅{round(drop_ratio*100,2)}%")
            print(f"商品链接：{new_data['detail_url']}")
            print("=" * 60)
        # 更新历史缓存
        self.history_data[goods_id] = new_data

模块 2：数据持久化存储（写入 CSV 时序文件）

每次采集自动追加记录到本地 CSV，留存完整历史快照，作为后续数据分析数据源。

import csv
import os

def save_data_to_csv(goods_data: dict):
    """将单条采集数据存入csv文件，按日期生成文件"""
    file_name = f"商品监控数据_{datetime.now().strftime('%Y%m%d')}.csv"
    file_exists = os.path.exists(file_name)
    # 展开SKU简化存储（仅保留主商品价格，如需细分SKU可自行扩展）
    row = [
        goods_data["crawl_time"],
        goods_data["goods_id"],
        goods_data["title"],
        goods_data["brand"],
        goods_data["current_price"],
        goods_data["original_price"],
        goods_data["stock_num"],
        goods_data["sales"],
        goods_data["shop_name"],
        goods_data["location"],
        goods_data["detail_url"]
    ]
    headers = [
        "采集时间", "商品ID", "商品标题", "品牌", "实时售价", "原价", "库存", "累计销量", "店铺", "发货地", "商品链接"
    ]
    with open(file_name, "a", encoding="utf-8-sig", newline="") as f:
        writer = csv.writer(f)
        if not file_exists:
            writer.writerow(headers)
        writer.writerow(row)
    print(f"【数据已存储】{goods_data['title']} 记录写入当日CSV文件")

模块 3：定时轮询监控主程序

循环遍历所有监控商品，自动采集、存储、异常预警，持续后台运行。

def start_monitor_task():
    collector = GoodsDataCollector()
    print("===== 京东商品自动监控系统已启动 =====")
    print(f"监控商品清单：{MONITOR_GOODS_IDS}")
    print(f"轮询间隔：{POLL_INTERVAL}秒，降价{PRICE_ALERT_RATIO*100}%触发预警")
    print("======================================")
    while True:
        for gid in MONITOR_GOODS_IDS:
            goods_info = collector.fetch_single_goods(goods_id=gid, real_time=True)
            if goods_info:
                # 检测价格波动预警
                collector.check_price_warning(gid, goods_info)
                # 持久化存储
                save_data_to_csv(goods_info)
        print(f"\n本轮采集完成，等待{POLL_INTERVAL}秒后执行下一轮\n")
        time.sleep(POLL_INTERVAL)

if __name__ == "__main__":
    start_monitor_task()

模块 4：历史数据可视化分析脚本（独立运行）

读取 CSV 历史时序数据，绘制价格走势折线图，用于竞品价格周期分析、大促规律复盘。

import pandas as pd
import matplotlib.pyplot as plt

def analysis_price_trend(csv_file_path: str):
    # 读取本地存储的监控数据
    df = pd.read_csv(csv_file_path, encoding="utf-8-sig")
    # 时间格式转换
    df["采集时间"] = pd.to_datetime(df["采集时间"])
    plt.rcParams["font.sans-serif"] = ["SimHei"]
    plt.rcParams["axes.unicode_minus"] = False
    plt.figure(figsize=(14, 7))
    # 按商品分组绘制价格曲线
    for goods_id, group in df.groupby("商品ID"):
        title = group.iloc[0]["商品标题"]
        plt.plot(group["采集时间"], group["实时售价"], marker="o", label=title[:15]+"...")
    plt.title("商品历史价格走势监控图", fontsize=14)
    plt.xlabel("采集时间")
    plt.ylabel("商品售价（元）")
    plt.legend(loc="best")
    plt.grid(alpha=0.3)
    plt.xticks(rotation=30)
    plt.tight_layout()
    # 保存图表到本地
    plt.savefig("价格走势分析图.png", dpi=300)
    plt.show()
    # 输出基础统计指标
    analysis_result = df.groupby(["商品ID", "商品标题"]).agg(
        最低售价=("实时售价", "min"),
        最高售价=("实时售价", "max"),
        平均售价=("实时售价", "mean"),
        当前库存=("库存", "last")
    ).round(2)
    print("===== 商品价格统计分析报表 =====")
    print(analysis_result)

# 使用示例：传入当天生成的CSV文件名
if __name__ == "__main__":
    analysis_price_trend("商品监控数据_20260814.csv")

五、运行说明与效果展示

  2. 启动监控程序 修改代码顶部CLAW_KEY、CLAW_SECRET、MONITOR_GOODS_IDS三项配置，直接运行主程序，程序会持续循环采集所有商品，控制台实时输出采集日志，价格大幅下跌时弹出红色预警提示。 本地目录自动生成按日期命名的 CSV 文件，永久留存每一次采集的价格、库存、销量快照。

  4. 数据分析可视化 单独运行分析脚本，传入对应日期的 CSV 文件，自动生成高清价格走势图片，同时输出统计报表，直观看到近 7 天 / 30 天商品最低价、均价、库存变化。

  6. 拓展优化方向（生产级升级）

  • 预警推送：对接企业微信机器人、短信接口，价格波动无需盯电脑自动推送消息；

  • 数据库存储：替换 CSV 为 MySQL/InfluxDB，支持上万款商品长期监控；

  • 定时任务：Linux 服务器配置 crontab、Windows 配置计划任务，7×24 小时后台运行；

  • SKU 细分分析：拆分多规格价格，对比不同尺码 / 颜色销量、定价差异；

  • 竞品对比：批量采集多款同类商品，输出类目均价、竞品价差分析报表。

六、常见问题排查

  2. 采集返回商品未找到 核对商品 ID 是否正确，商品下架、链接失效会触发该提示，及时从监控列表移除失效商品。

  4. 采集频繁报错、权限异常 检查后台密钥是否填写正确，确认工具额度未耗尽、账号状态正常，降低轮询间隔减少请求频次。

  6. 图表中文乱码 代码内已配置中文字体，Windows 系统若仍乱码，自行替换SimHei为Microsoft YaHei。

  8. CSV 文件乱码 代码写入时使用utf-8-sig编码，Excel 打开可正常显示中文，无需转码。

七、落地场景总结

这套轻量化监控分析系统适配多类从业者：

  2. 电商运营：实时盯竞品调价、库存清仓，快速调整店铺活动、定价策略；

  4. 货源采购：跟踪商品历史低价，找准囤货、批发采购最佳时机；

  6. 数据分析师：自动沉淀时序商品数据，完成价格周期、类目竞争格局分析；

  8. 自媒体 / 比价工具：批量采集商品素材、售价，自动化生成比价图文素材。

无需复杂爬虫技术储备，依靠 Open Claw 标准化采集能力，十几分钟即可搭建完整监控分析链路，把人工重复盯品的时间释放出来做策略优化，大幅提升电商数据运营效率。


少长咸集

群贤毕至

访客