告别人工盯品，基于 Open Claw 快速搭建电商商品监控与数据分析实战方案

专注API从业者

14人浏览 · 2026-08-21 10:49:28

专注API从业者 · 2026-08-21 10:49:28 发布

前言

做电商运营、竞品调研的朋友，几乎都遇到过这样的困境。

想要盯紧竞品的售价变动、库存状态、规格上新，大促的时候更是需要时刻关注价格波动。如果靠人工打开网页一个个复制记录，不仅耗费大量时间精力，还很容易错过临时降价、库存告急这类关键信号。

如果自己从零写页面解析脚本，又会碰到一大堆麻烦：网页动态渲染、频繁访问触发风控限制、页面改版之后脚本直接失效，还要处理图片、多规格款式、店铺信息等复杂内容，调试成本非常高。

最近在实践中发现 Open Claw 这套工具可以很好解决这类痛点。不需要处理复杂的页面解析、反访问限制相关的底层工作，只需要简单编写少量代码，就可以拿到商品完整结构化信息，实现定时监控异动、存储历史快照，完成基础的数据分析工作。

本文会完整分享从数据获取、异动告警、本地存储到数据分析的完整实操流程，代码可以直接拿来修改运行，适合个人调研以及中小团队的竞品跟踪场景。

温馨提示：本方案仅用于学习研究，使用过程中请遵守对应平台相关使用规范，合理控制访问频次。

整体实现思路

整套工具分为 4 个模块，逻辑简单清晰，方便后续迭代扩展：

数据获取模块：输入商品编号，拿到标准化的商品全量信息，包含标题、售价、原价、多规格款式、库存、品牌、店铺、图片链接等内容。
定时监控模块：循环定时获取商品最新状态，和上一次保存的历史快照做对比。
异动告警模块：识别涨价、降价、库存清空、补货等变化，控制台输出提醒，后续可以拓展企业微信、邮件消息推送。
数据分析模块：把每次采集的数据保存本地，基于历史记录统计价格区间、高低价、规格分布，简单完成竞品分析。

环境准备

本地需要安装 Python3.7 以上版本，依赖基础网络请求库。

pip install requests pandas

提前准备好工具分配的凭证信息，包含两组密钥字符串，用于身份校验。我们只需要传入商品编号，就可以获取完整的商品结构化数据。

完整代码实战

1. 基础数据获取封装

先封装基础函数，输入商品编号，返回清洗之后的商品字典数据，过滤无效空字段。

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

# --------------------------配置区，请修改为自己的凭证-------------------------
ACCESS_KEY = "你的凭证key"
ACCESS_SECRET = "你的凭证secret"
MONITOR_GOODS = ["10335871600"]  # 需要监控的商品编号列表
MONITOR_INTERVAL = 300  # 监控间隔，单位秒，示例5分钟采集一次
SAVE_CSV_PATH = "./goods_monitor_record.csv"
# ---------------------------------------------------------------------------

def get_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,
        "cache": "no",  # 关闭缓存，尽量拿到实时数据
        "result_type": "json"
    }
    try:
        resp = requests.get(base_url, params=params, headers={"Accept-Encoding": "gzip"}, timeout=20)
        res_json = resp.json()
        if res_json.get("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"网络请求异常:{str(e)}")
        return None

2. 监控异动检测逻辑

缓存上一轮采集的数据，对比当前最新数据，捕捉价格、库存发生的变化，打印告警提示，同时把每一轮数据写入本地 CSV 文件留存历史记录。

# 内存缓存，保存上一轮采集结果
last_snapshot = {}


def monitor_one_goods(goods_id: str):
    """监控单个商品，检测价格库存异动"""
    current = get_goods_detail(goods_id)
    if not current:
        return

    title = current.get("title", "未知商品")
    now_price = float(current.get("price", 0))
    origin_price = float(current.get("orginal_price", 0))
    stock = int(current.get("num", 0))
    shop_name = current.get("seller_info", {}).get("shop_name", "")
    curr_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    # 写入本地历史记录文件
    record_row = {
        "capture_time": curr_time,
        "goods_id": goods_id,
        "title": title,
        "current_price": now_price,
        "origin_price": origin_price,
        "stock": stock,
        "shop_name": shop_name
    }
    df_row = pd.DataFrame([record_row])
    df_row.to_csv(SAVE_CSV_PATH, mode="a", header=not os.path.exists(SAVE_CSV_PATH), index=False, encoding="utf_8_sig")

    print(f"\n【{curr_time}】商品:{title}")
    print(f"店铺:{shop_name}｜现价:{now_price}｜原价:{origin_price}｜库存:{stock}")

    # 第一次采集，只做缓存，不做对比
    if goods_id not in last_snapshot:
        last_snapshot[goods_id] = {"price": now_price, "stock": stock}
        return

    old = last_snapshot[goods_id]
    # 降价提醒
    if now_price < old["price"]:
        drop = old["price"] - now_price
        print(f"🔔降价预警！由{old['price']}下降到{now_price}，降幅{drop:.2f}元")
    # 涨价提醒
    if now_price > old["price"]:
        rise = now_price - old["price"]
        print(f"📈涨价提醒！由{old['price']}上涨到{now_price}，涨幅{rise:.2f}元")
    # 库存告警
    if stock <= 0 and old["stock"] > 0:
        print(f"🚨库存告急，商品已经缺货！")
    if stock > 0 and old["stock"] <= 0:
        print(f"📦商品重新补货，当前库存{stock}")

    # 更新内存缓存快照
    last_snapshot[goods_id]["price"] = now_price
    last_snapshot[goods_id]["stock"] = stock


def run_monitor_loop():
    """循环执行全部监控任务"""
    import os
    if not os.path.exists(SAVE_CSV_PATH):
        pd.DataFrame().to_csv(SAVE_CSV_PATH, index=False, encoding="utf_8_sig")
    print("=====启动商品监控任务=====")
    while True:
        for gid in MONITOR_GOODS:
            try:
                monitor_one_goods(gid)
            except Exception as err:
                print(f"处理{gid}发生异常{err}")
        print(f"\n等待{MONITOR_INTERVAL}秒进入下一轮检测...")
        time.sleep(MONITOR_INTERVAL)

3. 历史数据分析函数

读取本地保存的 CSV 历史文件，做简单统计分析：计算最高价、最低价、平均售价，输出价格波动概况。

def analysis_history_data():
    """读取本地历史文件，做数据分析统计"""
    try:
        df = pd.read_csv(SAVE_CSV_PATH, encoding="utf_8_sig")
        if df.empty:
            print("暂无历史监控数据，请先运行监控采集")
            return
        print("\n========历史数据分析报告========")
        for goods_id in df["goods_id"].unique():
            sub_df = df[df["goods_id"] == goods_id]
            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)
            capture_count = len(sub_df)
            print(f"\n商品：{title}")
            print(f"采集记录条数：{capture_count}")
            print(f"历史最高售价：{max_p}")
            print(f"历史最低售价：{min_p}")
            print(f"平均售价：{avg_p}")
    except Exception as e:
        print(f"数据分析读取文件失败：{e}")


if __name__ == "__main__":
    import os
    # 启动循环监控
    run_monitor_loop()

    # 如果不需要循环监控，只做单次采集分析，可以注释上面run_monitor_loop()，打开下面两行
    # monitor_one_goods(MONITOR_GOODS[0])
    # analysis_history_data()

使用说明

修改代码最上方配置区，填入自己的两组凭证，填入需要监控的商品编号列表。
MONITOR_INTERVAL为采集轮询间隔，不建议设置过小，避免高频请求。
脚本运行之后，控制台会打印每一轮商品信息，出现价格库存异动输出提醒。
所有采集记录自动保存到goods_monitor_record.csv，可以用 Excel 直接打开查看。
如果不需要持续循环监控，可以注释run_monitor_loop()，调用单次采集函数，再执行analysis_history_data()输出统计报告。