告别反复手工查价：用 Open‑Claw 搭建商品监控与数据分析小工具

适用场景：竞品盯价、大促价格跟踪、库存异动监控、商品信息批量统计，Python 完整实战，可直接上手运行

前言

做电商运营、选品或者个人薅羊毛，经常会遇到一个头疼问题：需要持续关注一批商品的价格、库存、规格变化。如果每天手动点开页面一个个复制记录，不仅耗费大量时间，还很容易错过大促降价、库存清空这类关键变化。

直接写页面解析脚本又会碰到一堆麻烦：网页结构改版就全部失效、访问频繁触发风控限制、各种动态渲染内容拿不到完整规格信息，调试维护成本很高。

最近在实践中发现借助 Open‑Claw 的数据能力，只需要输入商品编号，就能够拿到结构化的商品全部信息，包含标题、原价、实时售价、多规格参数、库存、店铺信息、图集等完整内容，不用处理复杂网页解析，我们可以快速搭建一套本地监控 + 数据分析脚本，完成定时采集、异动提醒、历史数据留存与简单统计分析。

本文完整演示从数据获取、持久化存储、异动告警，再到简单数据分析的全流程，代码可以直接复制修改使用。

整体实现思路

整套工具分为 4 个核心模块：

数据获取模块：输入商品编号，拿到标准化结构化商品数据，包含全部规格 SKU、价格、库存、品牌、图片、店铺等字段；支持缓存加速，减少重复请求消耗。
数据存储模块：每次采集带上时间戳，保存到本地 csv 文件，留存时序历史记录，不需要部署数据库，普通 Excel 就可以打开查看历史变化。
异动检测模块：对比本次与上一轮采集记录，识别价格涨跌、库存变动，控制台输出告警，后续可以拓展企业微信、钉钉消息推送。
数据分析模块：基于本地历史文件，做价格波动统计、多商品横向对比，输出简单分析结果，辅助选品和竞品研判。

环境准备

依赖基础库，执行安装：

pip install requests pandas

完整代码实战

1. 基础数据获取封装

把网络请求逻辑封装，只传入商品编号，返回清洗之后的字典对象，同时处理异常情况（商品不存在、网络异常等）

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

class GoodsDataClient:
    def __init__(self, key, secret):
        self.key = key
        self.secret = secret
        self.base_url = "https://api‑gw.onebound.cn/jd/item_get_pro/"

    def get_goods_info(self, goods_id):
        """根据商品编号获取商品结构化数据"""
        params = {
            "key": self.key,
            "secret": self.secret,
            "num_iid": goods_id,
            "cache": "yes",
            "result_type": "json"
        }
        headers = {
            "Accept‑Encoding": "gzip",
            "Connection": "close"
        }
        try:
            resp = requests.get(self.base_url, params=params, headers=headers, timeout=15)
            json_data = resp.json()
        except Exception as e:
            print(f"网络请求异常：{str(e)}")
            return None

        # 判断返回状态
        if json_data.get("error_code") != "0000":
            err_msg = json_data.get("reason", "未知异常")
            print(f"获取数据失败：{err_msg}")
            return None

        item = json_data.get("item")
        if not item:
            print("未获取到商品主体信息")
            return None

        # 简单清洗整理核心字段
        clean_data = {
            "crawl_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
            "goods_id": item.get("num_iid"),
            "title": item.get("title"),
            "brand": item.get("brand"),
            "current_price": float(item.get("price", 0)) if item.get("price") else 0,
            "origin_price": float(item.get("orginal_price",0)) if item.get("orginal_price") else 0,
            "stock": item.get("num",0),
            "shop_name": item.get("seller_info",{}).get("shop_name",""),
            "detail_url": item.get("detail_url"),
            "sku_list": item.get("skus",{}).get("sku",[])
        }
        return clean_data

2. 本地持久化与异动告警逻辑

每次采集把数据追加写入 csv，同时对比上一条记录，发现价格或者库存发生明显变化，控制台打印告警提示。

class GoodsMonitor:
    def __init__(self, client, save_file="goods_monitor.csv"):
        self.client = client
        self.save_file = save_file

    def save_to_csv(self, data):
        """保存采集记录到本地csv"""
        df_new = pd.DataFrame([data])
        try:
            df_old = pd.read_csv(self.save_file, encoding="utf‑8")
            df_all = pd.concat([df_old, df_new], ignore_index=True)
        except FileNotFoundError:
            df_all = df_new
        df_all.to_csv(self.save_file, index=False, encoding="utf‑8")

    def check_change(self, new_data):
        """对比历史数据，检测价格库存异动"""
        goods_id = new_data["goods_id"]
        try:
            df = pd.read_csv(self.save_file, encoding="utf‑8")
            his = df[df["goods_id"] == goods_id].tail(1)
            if len(his) ==0:
                return []
            last = his.iloc[0]
            alert_list = []
            # 价格变动超过0.1元触发提醒
            price_diff = new_data["current_price"] - float(last["current_price"])
            if abs(price_diff) > 0.1:
                alert_list.append(f"【价格变动】原价{last['current_price']} → 现价 {new_data['current_price']}")
            # 库存发生变化
            if int(new_data["stock"]) != int(last["stock"]):
                alert_list.append(f"【库存变动】旧库存{last['stock']} → 当前库存 {new_data['stock']}")
            return alert_list
        except Exception:
            return []

    def monitor_one_goods(self, goods_id):
        print(f"\n====开始采集商品 {goods_id} ====")
        info = self.client.get_goods_info(goods_id)
        if not info:
            print("采集失败，跳过该商品")
            return None
        # 保存记录
        self.save_to_csv(info)
        # 检查异动
        alerts = self.check_change(info)
        if alerts:
            print("⚠️⚠️⚠️发现异动提醒：")
            for a in alerts:
                print(f"\t{a}")
        else:
            print("本次对比：无明显异动")
        print(f"商品标题：{info['title']}")
        print(f"当前售价：{info['current_price']} ｜ 原价：{info['origin_price']}｜库存：{info['stock']}")
        return info

3. 简单数据分析函数

读取本地全部历史记录，输出单商品价格统计，支持查看最高、最低、平均售价，也可以批量对比多款商品价格区间。

def analysis_goods(file_name, target_goods_id=None):
    """历史数据分析，打印价格统计信息"""
    df = pd.read_csv(file_name, encoding="utf‑8")
    if target_goods_id:
        df = df[df["goods_id"] == str(target_goods_id)]
        if len(df) ==0:
            print("没有找到该商品的历史采集记录")
            return
        print(f"\n========商品 {target_goods_id} 历史价格统计========")
        print(f"记录总条数：{len(df)}")
        print(f"历史最低售价：{df['current_price'].min():.2f}")
        print(f"历史最高售价：{df['current_price'].max():.2f}")
        print(f"历史平均售价：{df['current_price'].mean():.2f}")
    else:
        # 全部商品简单汇总
        group = df.groupby("goods_id").agg({
            "current_price":["min","max","mean","count"]
        }).round(2)
        print("\n====全部监控商品汇总统计====")
        print(group)
    return df

4. 主程序入口，定时循环监控

if __name__ == "__main__":
    # 填入自己的凭证
    MY_KEY = "你的密钥key"
    MY_SECRET = "你的密钥secret"

    client = GoodsDataClient(MY_KEY, MY_SECRET)
    monitor = GoodsMonitor(client, save_file="goods_monitor.csv")

    # 需要监控的商品编号列表，可以放多个商品
    watch_goods = [
        "10335871600",
        # "填入更多商品id"
    ]

    # 循环监控，设置间隔时间，单位秒，根据自身配额合理设置，不要请求过于频繁
    while True:
        for gid in watch_goods:
            monitor.monitor_one_goods(gid)
            time.sleep(4)
        print("\n本轮采集完成，等待下一轮循环......")
        time.sleep(60*10) # 10分钟执行一轮

        # 如果需要执行分析，可以取消下面注释
        # analysis_goods("goods_monitor.csv", watch_goods[0])

使用说明与注意事项

获取商品编号：商品详情网页链接中，item.jd.com/后面那串数字就是 goods_id，复制填入监控列表即可。
频率控制：循环间隔不要设置过小，频繁请求会消耗额度，也容易触发限制，建议至少数分钟以上一轮。
缓存开关：代码中 cache 参数设置 yes，会优先读取缓存数据，速度更快；如果需要实时最新数据，可以改为 no。
异常处理：脚本已经做了基础异常捕获，网络波动、商品下架不会让整个程序直接崩溃。当商品下架或者编号错误时控制台会打印提示。
拓展方向

告警：检测到价格变动之后，可以对接企业微信 / 钉钉 webhook 推送消息，不用盯着控制台。
可视化：基于 pandas+matplotlib，读取 csv 历史数据，画出商品价格走势图。
定时任务：Linux/macOS 用 crontab、Windows 使用任务计划程序，把脚本后台长期运行。
规格深度解析：返回结果中sku_list包含全部规格的编号、价格、库存，我们可以二次处理做多规格分别监控。

拿到数据之后可以做哪些分析？

有了持续沉淀的时序数据，就可以落地很多业务分析：

大促价格研判：对比历史最低价位，判断当前是不是好的入手时机，观察大促期间真实降价幅度，区分先涨价再打折的虚假活动。
竞品对标：持续跟踪同行多款竞品的调价节奏、库存状态，了解对手上新、清仓动作。
库存预警：监控爆款库存水位，当库存快速下跌，及时捕捉市场热度变化。
规格分析：解析多规格数据，看哪一个规格价格最有优势，哪些规格经常缺货。

踩坑小结

之前尝试直接页面解析的方案，经常遇到页面改版代码直接失效，动态 JS 渲染拿不全 SKU 规格，调试消耗大量时间。借助 Open‑Claw 拿到已经处理好的结构化数据之后，我们只需要专注业务逻辑：存储、告警、数据分析，不用再消耗精力对抗网页反爬和页面变更，小工具开发效率提升非常明显。

万邦api博客

Nice to meet you, too!

api 数据挖掘