OpenClaw 实战｜从零搭建电商商品监控与数据分析系统（Python 完整可运行）

前言

做电商运营、比价囤货、竞品调研的时候，我们经常有这些痛点：

需要持续盯多款商品的售价、原价变动，大促期间手动刷页面效率极低；
多规格款式的库存、价格分散，很难统一汇总对比；
想要留存历史数据，分析价格波动规律，找出合适入手时机；
普通网页抓取容易遇到反访问、页面结构改版，脚本频繁失效。

最近体验了 OpenClaw 这套工具套件，可以稳定获取商品完整资料，包含标题、品牌、售价、吊牌原价、多规格 SKU、库存、图片、店铺信息等全套结构化内容。不用折腾浏览器模拟、不用写复杂正则解析页面，把精力聚焦在监控逻辑、数据存储、数据分析业务逻辑上。

本文手把手带大家实现一套完整方案：商品定时采集、变动告警、本地持久化存储、历史价格趋势分析，全部代码可以直接复制运行。

环境依赖：Python3.8+，requests、pandas、matplotlib、sqlite3(内置无需安装)

整体方案设计

整套程序分为 4 个模块：

数据获取模块：借助工具拿到标准化商品原始资料；
持久化存储模块：使用轻量 SQLite 保存每一次采集记录，留存历史快照；
监控告警模块：定时循环采集，对比上一轮数据，价格 / 库存发生变化输出提醒；
数据分析模块：读取本地历史记录，输出统计指标，绘制价格变化走势图。

支持能力： ✅ 获取商品标题、品牌、当前售价、原价、库存、多规格 SKU 明细 ✅ 定时循环监控，识别价格下调、库存变动 ✅ 本地数据库保存全部历史采集快照 ✅ 统计最高价、最低价、平均价格 ✅ 生成可视化价格趋势图表 ✅ 支持批量监控多款商品

完整代码实现

1. 基础工具类：数据获取与数据库初始化

import requests
import json
import sqlite3
import time
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime

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

# --------------------------配置区--------------------------
# 个人凭证，开通工具服务后获取
ACCESS_KEY = "替换成你的凭证key"
ACCESS_SECRET = "替换成你的凭证secret"
# 需要监控的商品ID列表
MONITOR_ITEM_IDS = ["10335871600", "10057467958584"]
# 采集间隔，单位秒，示例5分钟采集一次
MONITOR_INTERVAL = 300
# 本地数据库文件名
DB_FILE = "goods_monitor.db"
# -----------------------------------------------------------


def init_db():
    """初始化数据库，创建数据表"""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    # 商品基础信息表
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS goods_record (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        num_iid TEXT,
        title TEXT,
        brand TEXT,
        price REAL,
        orginal_price REAL,
        stock INTEGER,
        shop_name TEXT,
        collect_time TIMESTAMP
    )
    ''')
    # sku规格明细表
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS sku_record (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        num_iid TEXT,
        sku_id TEXT,
        sku_name TEXT,
        price REAL,
        orginal_price REAL,
        quantity INTEGER,
        collect_time TIMESTAMP
    )
    ''')
    conn.commit()
    conn.close()


def fetch_goods_detail(item_id):
    """
    获取商品详情原始结构化数据
    :param item_id: 商品ID
    :return: dict商品数据，失败返回None
    """
    base_url = "https://api-gw.onebound.cn/jd/item_get_pro/"
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "num_iid": item_id,
        "cache": "no"
    }
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=15)
        res_json = resp.json()
        if res_json.get("error_code") != "0000":
            print(f"【获取失败】商品:{item_id}, 原因:{res_json.get('reason','未知错误')}")
            return None
        item_data = res_json.get("item")
        if not item_data:
            print(f"【警告】商品{item_id}返回数据为空")
            return None
        return item_data
    except Exception as e:
        print(f"【网络异常】{item_id} , {str(e)}")
        return None


def save_data_to_db(item_data):
    """将采集到的商品以及sku数据存入sqlite数据库"""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    num_iid = item_data.get("num_iid")
    title = item_data.get("title", "")
    brand = item_data.get("brand", "")
    price = float(item_data.get("price", 0)) if item_data.get("price") else 0
    orginal_price = float(item_data.get("orginal_price", 0)) if item_data.get("orginal_price") else 0
    stock = item_data.get("num", 0)
    shop_name = item_data.get("seller_info", {}).get("shop_name", "")

    cursor.execute('''
    INSERT INTO goods_record(num_iid,title,brand,price,orginal_price,stock,shop_name,collect_time)
    VALUES (?,?,?,?,?,?,?,?)
    ''', (num_iid, title, brand, price, orginal_price, stock, shop_name, now))

    # 保存SKU多规格数据
    sku_list = item_data.get("skus", {}).get("sku", [])
    for sku in sku_list:
        sku_id = sku.get("sku_id", "")
        sku_name = sku.get("properties_name", "")
        sku_price = float(sku.get("price", 0)) if sku.get("price") else 0
        sku_origin_price = float(sku.get("orginal_price", 0)) if sku.get("orginal_price") else 0
        sku_qty = sku.get("quantity", 0)
        cursor.execute('''
        INSERT INTO sku_record(num_iid,sku_id,sku_name,price,orginal_price,quantity,collect_time)
        VALUES (?,?,?,?,?,?,?)
        ''', (num_iid, sku_id, sku_name, sku_price, sku_origin_price, sku_qty, now))
    conn.commit()
    conn.close()

2. 监控告警逻辑：定时采集，对比变化触发提醒

class GoodsMonitor:
    def __init__(self):
        self.last_data = dict()  # 缓存上一轮采集结果 {item_id: data}

    def check_change_and_alert(self, item_data):
        """对比新旧数据，如果价格/库存变动输出告警提示"""
        item_id = item_data.get("num_iid")
        current_price = float(item_data.get("price", 0))
        current_stock = int(item_data.get("num", 0))
        title = item_data.get("title", "")[:30]

        if item_id not in self.last_data:
            print(f"\n【首次采集】商品ID:{item_id} | {title}")
            print(f"当前售价:{current_price}元 | 库存:{current_stock}")
            self.last_data[item_id] = {"price": current_price, "stock": current_stock}
            return

        old = self.last_data[item_id]
        old_price = old["price"]
        old_stock = old["stock"]

        # 判断价格变动
        if abs(current_price - old_price) > 0.01:
            print(f"\n🔥【价格变动告警】{title}")
            print(f"旧价格:{old_price} → 新价格:{current_price}")
        # 判断库存变动
        if current_stock != old_stock:
            print(f"\n⚠️【库存变动告警】{title}")
            print(f"旧库存:{old_stock} → 新库存:{current_stock}")

        self.last_data[item_id]["price"] = current_price
        self.last_data[item_id]["stock"] = current_stock

    def run_monitor_loop(self):
        """循环执行监控任务"""
        print("=====商品监控服务启动=====")
        while True:
            for goods_id in MONITOR_ITEM_IDS:
                item = fetch_goods_detail(goods_id)
                if item is None:
                    continue
                self.check_change_and_alert(item)
                save_data_to_db(item)
            print(f"\n等待 {MONITOR_INTERVAL} 秒后进行下一轮采集...\n")
            time.sleep(MONITOR_INTERVAL)

3. 数据分析模块：读取历史记录，统计与绘图

def analysis_goods_history(item_id):
    """读取本地数据库，对单个商品做历史数据分析，输出统计+绘图"""
    conn = sqlite3.connect(DB_FILE)
    df = pd.read_sql('''
        select collect_time, price, orginal_price, stock from goods_record
        where num_iid = ? order by collect_time asc
    ''', conn, params=(item_id,))
    conn.close()

    if len(df) == 0:
        print(f"【数据分析】商品{item_id}没有历史采集记录")
        return None

    df["collect_time"] = pd.to_datetime(df["collect_time"])
    print(f"\n========商品 {item_id} 历史数据统计========")
    print(f"采集总次数：{len(df)}")
    print(f"最低售价：{df['price'].min():.2f}")
    print(f"最高售价：{df['price'].max():.2f}")
    print(f"平均售价：{df['price'].mean():.2f}")

    # 绘制价格趋势图
    plt.figure(figsize=(12, 6))
    plt.plot(df["collect_time"], df["price"], marker="o", color="#E53935", label="实时售价")
    plt.plot(df["collect_time"], df["orginal_price"], linestyle="--", color="#757575", label="商品原价")
    plt.title(f"商品ID:{item_id} 价格变化趋势")
    plt.xlabel("采集时间")
    plt.ylabel("价格（元）")
    plt.legend()
    plt.grid(alpha=0.3)
    plt.tight_layout()
    save_img_name = f"goods_{item_id}_trend.png"
    plt.savefig(save_img_name)
    plt.show()
    print(f"趋势图表已保存至：{save_img_name}")
    return df

4. 程序入口，两种运行模式

if __name__ == "__main__":
    init_db()
    print("""
请选择运行模式：
1 → 启动持续监控循环（定时采集+变动提醒）
2 → 读取本地历史数据做数据分析绘图
""")
    mode = input("输入模式数字：")
    monitor = GoodsMonitor()
    if mode == "1":
        monitor.run_monitor_loop()
    elif mode == "2":
        target_id = input("请输入要分析的商品ID：")
        analysis_goods_history(target_id)
    else:
        print("输入错误，程序退出")

使用说明

将代码中KEY、SECRET替换成自己开通服务之后拿到的凭证；
修改MONITOR_ITEM_IDS填入你想要监控的商品编号；
运行脚本：

选择模式 1：程序持续运行，每设定秒数自动采集，价格库存变动控制台输出告警；所有采集记录自动存入goods_monitor.db；
选择模式 2：输入商品编号，读取本地数据库历史记录，输出价格统计，自动生成价格走势图 png 图片。

业务拓展方向

消息推送：告警部分可以接入钉钉 / 企业微信机器人，一旦价格下降自动推送消息到工作群，不用盯着控制台；
阈值告警：增加心理价位判断，当售价低于我们预设目标价格，触发提醒，适合个人囤货比价；
SKU 粒度监控：目前保存全部规格数据，可以扩展针对某个具体规格单独监控；
批量导出报表：基于 pandas 把历史记录导出 Excel，用于周报、竞品复盘；
部署后台：放到服务器后台常驻运行，实现 7×24 小时不间断监控。

踩坑与实践经验

采集间隔不要设置过小，建议至少 3 分钟以上，合理控制请求频次；
工具支持关闭缓存模式cache=no拿到最新实时数据，适合监控场景；静态信息如标题图片可以开启缓存，减少消耗；
返回的多规格价格存在特殊场景，业务上建议以对应规格字段为准；
当商品下架、编号失效，脚本会捕获异常，打印错误信息，不会直接崩溃；
SQLite 适合个人小体量监控；如果后期监控几百个商品，建议迁移 MySQL。

小结

借助 OpenClaw 获取标准化商品资料，我们不用耗费大量精力处理页面反访问、页面改版维护解析规则。把重心放在业务逻辑：数据存储、监控告警、数据分析。上面整套代码可以直接落地，不管是个人囤货比价、小商家竞品监控、简单的数据分析调研，都可以快速满足需求。

万邦api博客

Nice to meet you, too!

api 数据挖掘