×

api 数据挖掘

OpenClaw 实战｜搭建电商商品监控与数据分析系统，实现价格库存自动追踪

admin admin 发表于2026-08-25 16:25:04 浏览12 评论0

抢沙发发表评论

前言

做电商运营、选品或者竞品调研的时候，我们经常需要持续盯一批商品：价格什么时候降价、库存是否充足、不同规格的售价变化、店铺的商品动态。如果每天手动打开网页一个个复制记录，不仅耗费大量人力，还很容易漏掉关键的变动节点。

我之前踩过不少坑：直接解析网页源码会遇到页面动态渲染、反爬限制，频繁访问还会触发风控拦截，拿到的数据经常缺失规格、原价、店铺信息，后续分析很难开展。后来尝试 OpenClaw 工具来做数据获取，把重复的手工工作交给程序自动执行，轻松完成定时监控，再结合 Python 做清洗、统计分析，整套方案跑通之后，效率提升非常明显。

本篇就完整分享从获取商品原始数据、定时监控、数据清洗，到做简单竞品数据分析的完整流程，代码可以直接修改后投入使用。

整体实现思路

整套流程分成三个核心环节：

  2. 原始信息获取：输入商品编号，拿到商品标题、现价、原价、库存、多规格信息、店铺、图片等全套原始内容；

  4. 持久化存储与定时监控：定时循环采集，把每一次采集结果保存，对比前后数据，捕捉价格、库存变动；

  6. 数据分析处理：读取历史记录，做价格波动统计、规格对比、基础数据可视化，输出可以直接参考的分析结果。

小提示：实际使用时，做好请求间隔控制，避免短时间高频操作，兼顾稳定性。

环境准备

本地需要安装依赖库，打开终端执行：

pip install requests pandas matplotlib

第一部分：获取商品详情原始数据

我们封装一个简易工具类，填入自己的凭证和商品编号，就可以拿到完整商品原始报文，包含多规格、店铺、原价、库存等全部字段。

# product_monitor.py
import requests
import json
import time
from datetime import datetime
import pandas as pd
import matplotlib.pyplot as plt

plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False

class ProductCollector:
    def __init__(self, access_key, secret_key):
        self.access_key = access_key
        self.secret_key = secret_key
        self.base_address = "https://api‑gw.onebound.cn/jd/item_get_pro/"
        self.session = requests.Session()

    def get_product_raw(self, product_id):
        """获取单个商品完整原始数据"""
        params = {
            "key": self.access_key,
            "secret": self.secret_key,
            "num_iid": product_id,
            "result_type": "json",
            "cache": "no"
        }
        try:
            resp = self.session.get(
                self.base_address,
                params=params,
                headers={"Accept‑Encoding": "gzip", "Connection": "close"},
                timeout=15
            )
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            print(f"商品 {product_id} 获取异常：{str(e)}")
            return None

    def parse_product_info(self, raw_json):
        """解析原始报文，提取业务需要的核心字段"""
        if not raw_json or raw_json.get("error_code") != "0000":
            return None
        item = raw_json.get("item", {})
        sku_list = item.get("skus", {}).get("sku", [])
        sku_result = []
        for sku in sku_list:
            sku_result.append({
                "sku_id": sku.get("sku_id"),
                "spec_name": sku.get("properties_name"),
                "current_price": float(sku.get("price", 0)),
                "origin_price": float(sku.get("orginal_price", 0)),
                "stock": int(sku.get("quantity", 0)),
                "sku_url": sku.get("sku_url")
            })
        output = {
            "crawl_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
            "product_id": item.get("num_iid"),
            "title": item.get("title"),
            "brand": item.get("brand"),
            "shop_name": item.get("seller_info", {}).get("shop_name"),
            "current_price": float(item.get("price", 0)),
            "origin_price": float(item.get("orginal_price", 0)),
            "stock_total": item.get("num"),
            "detail_url": item.get("detail_url"),
            "main_image": item.get("pic_url"),
            "skus": sku_result
        }
        return output

if __name__ == "__main__":
    # 替换成你的凭证
    AK = "你的access_key"
    SK = "你的secret_key"
    collector = ProductCollector(AK, SK)
    target_product = "10335871600"
    raw_data = collector.get_product_raw(target_product)
    product_info = collector.parse_product_info(raw_data)
    print(json.dumps(product_info, ensure_ascii=False, indent=2))

运行代码后，就可以输出结构化后的商品信息，包含主商品基础信息，以及全部规格的价格、库存。

第二部分：实现定时监控，捕捉价格库存变动

很多场景下，我们需要每隔一段时间重新拉取，对比上一轮数据，如果价格下跌、库存发生变化，就记录变动事件。下面增加定时采集、变动检测、保存历史记录到 csv 文件的逻辑。

# 接上面的代码继续追加
def save_to_csv(data, file_name="product_history.csv"):
    """保存单条采集记录到本地csv"""
    base_row = {
        "crawl_time": data["crawl_time"],
        "product_id": data["product_id"],
        "title": data["title"],
        "shop_name": data["shop_name"],
        "current_price": data["current_price"],
        "origin_price": data["origin_price"],
        "stock_total": data["stock_total"]
    }
    df = pd.DataFrame([base_row])
    try:
        old = pd.read_csv(file_name)
        df = pd.concat([old, df], ignore_index=True)
    except FileNotFoundError:
        pass
    df.to_csv(file_name, index=False, encoding="utf‑8‑sig")

def detect_change(old_data, new_data):
    """对比两次采集结果，识别变动"""
    change_list = []
    if old_data["current_price"] != new_data["current_price"]:
        change_list.append(f"价格变动：{old_data['current_price']} → {new_data['current_price']}")
    if old_data["stock_total"] != new_data["stock_total"]:
        change_list.append(f"库存变动：{old_data['stock_total']} → {new_data['stock_total']}")
    # 对比sku规格价格
    old_sku_map = {s["sku_id"]: s for s in old_data["skus"]}
    for new_sku in new_data["skus"]:
        o_sku = old_sku_map.get(new_sku["sku_id"])
        if o_sku and o_sku["current_price"] != new_sku["current_price"]:
            change_list.append(f"规格[{new_sku['spec_name']}]价格：{o_sku['current_price']}→{new_sku['current_price']}")
    return change_list

def monitor_loop(product_id, interval=60*10):
    """循环监控，默认10分钟采集一次"""
    AK = "你的access_key"
    SK = "你的secret_key"
    coll = ProductCollector(AK, SK)
    last_record = None
    print("开始监控，按Ctrl+C终止程序")
    while True:
        raw = coll.get_product_raw(product_id)
        curr = coll.parse_product_info(raw)
        if curr:
            save_to_csv(curr)
            print(f"\n【{curr['crawl_time']}】采集完成｜商品：{curr['title']}")
            if last_record is not None:
                changes = detect_change(last_record, curr)
                if changes:
                    print("✅检测到变动事件：")
                    for item in changes:
                        print(f"‑ {item}")
            last_record = curr
        try:
            time.sleep(interval)
        except KeyboardInterrupt:
            print("\n监控任务已退出")
            break

# 启动监控
# monitor_loop("10335871600", interval=600)

说明：interval 单位是秒，示例 10 分钟 = 600 秒，正式运行不要设置过小，防止请求频率过高。运行后所有历史记录会持续写入product_history.csv，方便后续复盘。

第三部分：基于历史记录做数据分析

当我们积累一段时间的历史记录之后，就可以读取 csv 文件做价格走势、价格波动幅度分析，还可以生成简单可视化图表。

def analysis_history(csv_path="product_history.csv"):
    df = pd.read_csv(csv_path, encoding="utf‑8‑sig")
    df["crawl_time"] = pd.to_datetime(df["crawl_time"])
    print("====历史数据概览====")
    print(f"总采集记录数：{len(df)}")
    print(f"监控时间区间：{df['crawl_time'].min()} ~ {df['crawl_time'].max()}")
    print(f"价格统计：最高{df['current_price'].max()}｜最低{df['current_price'].min()}｜均值{round(df['current_price'].mean(),2)}")

    # 绘制价格走势
    plt.figure(figsize=(12, 5))
    plt.plot(df["crawl_time"], df["current_price"], marker="o", linestyle="-", color="#2E86AB")
    plt.title("商品价格监控走势")
    plt.xlabel("采集时间")
    plt.ylabel("当前售价")
    plt.xticks(rotation=30)
    plt.grid(alpha=0.3)
    plt.tight_layout()
    plt.savefig("price_trend.png", dpi=150)
    plt.show()

    # 统计降价幅度
    df["discount_rate"] = (df["origin_price"] - df["current_price"]) / df["origin_price"]
    print("\n====折扣统计====")
    print(df[["crawl_time","current_price","origin_price","discount_rate"]].round(3))

if __name__ == "__main__":
    analysis_history("product_history.csv")

运行之后，控制台输出统计信息，同时在本地生成price_trend.png价格走势图，可以直接放到报告里。

实际业务场景拓展

  2. 多商品批量监控：把商品 ID 放到列表循环处理，就可以同时监控一批竞品，分别保存记录；

  4. 消息提醒：检测到变动事件后，可以对接企业微信 / 钉钉机器人，一旦降价、库存清零，立刻推送通知；

  6. 规格维度分析：解析出来的 skus 列表，可以单独保存每个规格的历史，分析哪个规格经常调价；

  8. 数据导出报告：Pandas 可以直接输出 Excel，方便运营同学做周报复盘。

踩坑与注意事项

  2. 部分场景下返回的价格会存在偏差，这是数据来源本身特性，做分析时要结合业务校验，不要直接当作绝对唯一依据；

  4. 不要把轮询间隔设置太短，建议至少 5‑10 分钟以上，避免短时间大量请求；

  6. 如果返回找不到商品，优先核对商品编号是否正确，确认商品是否已经下架；

  8. 做好异常捕获，网络波动时程序不会直接崩溃，保证监控任务稳定跑。

写在最后

OpenClaw 这套工具可以帮助我们绕开复杂的网页逆向、动态渲染、字体加密等难题，快速拿到完整结构化商品信息。配合 Python 做定时任务、存储、统计分析，就能低成本搭建一套属于自己的商品监控系统，不管是个人选品，还是小团队做竞品跟踪，都很实用。

整套代码可以直接复制运行，替换凭证、商品编号即可上手，大家可以根据自己业务继续拓展告警、多商品管理等功能。


少长咸集

群贤毕至

访客