×

api 数据挖掘

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

admin admin 发表于2026-10-20 10:29:12 浏览7 评论0

抢沙发发表评论

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


前言

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

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

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

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

整体方案设计

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

  2. 数据获取模块：传入商品编号，拿到完整结构化商品信息，处理网络异常、商品不存在等各类异常情况。

  4. 数据持久化模块：每次采集带上时间戳，保存历史记录，这里选用 CSV 文件，不需要额外安装数据库，普通办公电脑直接打开表格查看历史数据。

  6. 异动监控告警模块：对比本次采集和上一轮历史数据，识别价格跳水、库存骤降，达到设定阈值就输出告警提示。

  8. 数据分析模块：基于留存的时序历史数据，完成价格走势绘图、多规格对比、竞品统计分析，输出可视化图表。

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

  • 稳定性更好，不用担心页面样式改版导致解析失效；

  • 一次性返回全部规格、店铺、图片信息，不需要多次请求拼接；

  • 自带缓存策略，重复查询可以加速返回，减少网络消耗；

  • 代码量少，新手也能快速看懂、二次修改。

环境准备

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

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")

代码说明与扩展方向

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

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

  2. SKU 多规格处理：返回结果中skus字段包含全部规格，每一条规格带有规格名称、规格编号、对应价格、库存。可以在代码中循环解析item_info["skus"]["sku"]，把每一个子规格单独存一行记录，实现多规格分别监控。

  4. 告警能力拓展：目前告警仅控制台打印，实际项目可以拓展，触发告警后调用企业微信 / 钉钉机器人、发送邮件，实现消息推送。

  6. 数据分析拓展：

    • 读取本地 CSV，可以做多款竞品横向价格对比；

    • 统计原价、现价差值，计算折扣力度；

    • 筛选库存低于阈值的商品做库存预警；

    • 按店铺做批量统计，输出店铺商品大盘报表。

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

踩坑经验总结

  2. 价格字段偶尔会出现数据延迟，不能完全单一依赖单次采集的价格做业务决策，建议结合历史多轮数据综合判断。

  4. 开启缓存模式可以加快返回速度，如果需要拿到实时最新状态，可以将USE_CACHE设置为no，但会消耗更多额度。

  6. 尽量不要设置很短的轮询周期，合理设置间隔，避免短时间大量请求。

  8. 存储 CSV 文件用utf‑8‑sig编码，Windows 系统 Excel 打开不会出现中文乱码。

小结

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


少长咸集

群贤毕至

访客