万邦api博客

Open‑Claw 实战｜无需逆向，快速搭建电商商品监控与数据分析系统

Wed, 02 Sep 2026 17:58:47 +0800

前言

做电商运营、选品、比价，很多时候我们需要持续盯紧商品的售价、库存、规格、店铺信息。手动复制粘贴做表格效率极低，大促期间价格频繁变动，人工几乎跟不上节奏。

很多同学尝试直接写网页解析，却遇到各种反爬、动态渲染、JS 加密，调试几天拿不到完整的 SKU 规格、原价、发货地等信息。今天分享一套落地思路，借助 Open‑Claw 工具能力，绕开逆向难题，快速实现商品的持续监控，把采集到的数据做清洗统计，完成从数据获取到分析的完整流程。

适合人群：电商运营、数据分析爱好者、Python 开发人员；环境要求 Python3.7+，会基础 Python 语法即可。

一、业务目标梳理

我们想要达成三件事：

单商品完整信息获取：拿到标题、当前售价、划线原价、库存、多规格 SKU、店铺名称、商品主图、发货地等全量资料。
定时持续监控：循环巡检，记录每一次价格、库存快照，捕捉调价、缺货、规格变更。
简单数据分析：保存历史记录，做价格波动统计、规格盘点，发现异常自动打印告警。

不做浏览器模拟、不处理复杂 JS 解密，直接拿到结构化 JSON 结果，减少大量排错成本。

二、环境准备

只需要安装网络请求库，用于发起网络访问，解析返回结果，同时引入时间处理、csv 存储模块用于持久化监控记录。

pip install requests

三、完整实战代码

说明：下面代码中your_key、your_secret替换成你自己的凭证，product_id替换成目标商品编号。逻辑分为三块：获取商品原始数据、解析提取业务字段、定时循环监控、写入本地 CSV 留存历史，异常捕获保证程序不会单次失败直接崩溃。

# coding:utf-8
import requests
import time
import csv
from datetime import datetime

class ProductMonitor:
    def __init__(self, key, secret):
        self.key = key
        self.secret = secret
        self.base_url = "https://api‑gw.onebound.cn/jd/item_get_pro/"
        self.csv_file = "product_monitor_record.csv"
        self.init_csv()

    def init_csv(self):
        """初始化本地存储文件，不存在则写入表头"""
        header = ["crawl_time", "product_id", "title", "current_price",
                  "original_price", "stock", "shop_name", "sku_count"]
        try:
            with open(self.csv_file, "r", encoding="utf‑8‑sig") as f:
                pass
        except FileNotFoundError:
            with open(self.csv_file, "w", newline="", encoding="utf‑8‑sig") as f:
                writer = csv.writer(f)
                writer.writerow(header)

    def fetch_product_raw(self, product_id):
        """获取商品原始结构化数据"""
        params = {
            "key": self.key,
            "secret": self.secret,
            "num_iid": product_id,
            "cache": "no"   # 关闭缓存，获取最新快照
        }
        headers = {
            "Accept‑Encoding": "gzip",
            "Connection": "close"
        }
        try:
            resp = requests.get(self.base_url, params=params, headers=headers, timeout=20)
            return resp.json()
        except Exception as e:
            print(f"网络请求异常：{str(e)}")
            return None

    def parse_product_data(self, raw_json):
        """从原始返回结果解析业务需要字段"""
        if not raw_json:
            return None
        error_code = raw_json.get("error_code")
        if error_code != "0000":
            err_msg = raw_json.get("error", "未知错误")
            print(f"数据获取失败，错误信息：{err_msg}")
            return None
        item = raw_json.get("item", {})
        if not item:
            print("未匹配到商品主体信息")
            return None

        sku_list = item.get("skus", {}).get("sku", [])
        parse_data = {
            "crawl_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
            "product_id": item.get("num_iid"),
            "title": item.get("title", ""),
            "current_price": float(item.get("price", 0)),
            "original_price": float(item.get("orginal_price", 0)),
            "stock": item.get("num", 0),
            "shop_name": item.get("seller_info", {}).get("shop_name", ""),
            "sku_count": len(sku_list),
            "skus_detail": sku_list
        }
        return parse_data

    def save_record(self, data):
        """把单次监控快照写入CSV做历史留存"""
        row = [
            data["crawl_time"],
            data["product_id"],
            data["title"],
            data["current_price"],
            data["original_price"],
            data["stock"],
            data["shop_name"],
            data["sku_count"]
        ]
        with open(self.csv_file, "a", newline="", encoding="utf‑8‑sig") as f:
            writer = csv.writer(f)
            writer.writerow(row)

    def alert_check(self, data, warn_price_threshold):
        """简单告警逻辑：价格低于阈值、库存为0触发提醒"""
        if data["current_price"] <= warn_price_threshold:
            print("‑" * 60)
            print(f"⚠️ 价格预警！商品：{data['title']}")
            print(f"当前售价 {data['current_price']}，低于设定阈值 {warn_price_threshold}")
            print("‑" * 60)
        if data["stock"] <= 0:
            print("‑" * 60)
            print(f"🔴 库存告警！商品 {data['title']} 当前已经无货")
            print("‑" * 60)

    def single_monitor_task(self, product_id, warn_price):
        """执行一次完整监控任务：拉取‑解析‑存储‑告警"""
        raw = self.fetch_product_raw(product_id)
        parsed = self.parse_product_data(raw)
        if parsed:
            self.save_record(parsed)
            self.alert_check(parsed, warn_price)
            print(f"[{parsed['crawl_time']}] 监控完成｜{parsed['title'][:35]}...｜现价:{parsed['current_price']}｜库存:{parsed['stock']}｜规格数:{parsed['sku_count']}")
        return parsed

if __name__ == "__main__":
    # ========== 配置区域，请修改为你的信息 ==========
    MY_KEY = "your_key"
    MY_SECRET = "your_secret"
    MONITOR_PRODUCT_ID = "10335871600"  # 商品编号
    PRICE_WARN_THRESHOLD = 120  # 低于该价格触发提醒
    MONITOR_INTERVAL = 60 * 30  # 监控间隔，单位秒，示例30分钟巡检一次
    # =============================================

    monitor = ProductMonitor(MY_KEY, MY_SECRET)
    print("===== 商品监控程序已启动 =====")
    print(f"监控商品ID:{MONITOR_PRODUCT_ID}，价格预警阈值:{PRICE_WARN_THRESHOLD}，轮询间隔:{MONITOR_INTERVAL}秒")
    while True:
        monitor.single_monitor_task(MONITOR_PRODUCT_ID, PRICE_WARN_THRESHOLD)
        time.sleep(MONITOR_INTERVAL)

四、代码功能说明

持久化存储：每次巡检把时间戳、价格、库存、店铺信息写入product_monitor_record.csv，Excel 可以直接打开查看历史轨迹。
异常容错：网络波动、商品下架会打印错误信息，不会直接让整个程序崩溃。
双重告警：价格低于预设阈值、库存归零，控制台输出醒目的提醒。
SKU 完整获取：返回结果中自带全部规格列表，可以扩展做规格级别的监控。

注意：轮询间隔不要设置过小，大促场景建议至少 10 分钟以上，避免高频访问造成限制。

五、拿到历史数据之后，怎么做数据分析

采集得到 CSV 历史记录文件之后，我们可以用 Pandas 做简单的统计分析，下面补充一小段数据分析脚本：

import pandas as pd

def analysis_history():
    df = pd.read_csv("product_monitor_record.csv", encoding="utf‑8‑sig")
    print("====历史数据概览====")
    print(f"总采集记录数：{len(df)}")
    print(f"价格最大值：{df['current_price'].max()}")
    print(f"价格最小值：{df['current_price'].min()}")
    print(f"平均售价：{round(df['current_price'].mean(),2)}")
    print("\n最近10条快照：")
    print(df.tail(10)[["crawl_time","current_price","stock"]])

if __name__ == "__main__":
    analysis_history()

执行后可以直接看到价格最高、最低、均值，快速判断商品的价格波动区间，辅助我们做采购、选品、竞品对标。如果需要更进一步，可以接入 Matplotlib 画出价格时序走势图。

六、常见踩坑点与优化方向

缓存干扰：代码中设置cache=no，保证拿到实时最新快照；不关闭缓存会返回历史旧数据，监控就失去意义。
商品下架情况：当商品已经删除、下架，返回结果会提示找不到商品，业务上可以增加逻辑，一旦连续多次返回不存在，记录到日志，停止监控。
多商品批量监控：可以维护一个商品 ID 列表，循环遍历列表，实现批量多款商品同时巡检。
告警渠道扩展：目前只是控制台打印，可扩展对接企业微信、邮件推送，做到无人值守通知。
规格粒度监控：当前示例取的商品主体价格，业务需要可以遍历skus_detail列表，针对每一个子规格单独记录价格库存。

七、小结

借助 Open‑Claw，我们不需要耗费大量精力处理网页逆向、JS 解密、验证码等麻烦工作，就能够稳定拿到完整结构化的商品信息。

整套流程分为：获取结构化原始数据 → 清洗解析业务字段 → 定时循环采集、落地存储快照 → 统计分析历史数据。对于商家、个人比价选品都非常实用。

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

Mon, 31 Aug 2026 17:49:12 +0800

前言

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

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

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

Open Claw 实战｜快速搭建电商商品监控与数据分析能力

Sun, 30 Aug 2026 17:57:01 +0800

前言

做电商运营、市场调研或者竞品分析的时候，相信很多人都遇到这样的痛点：想要盯一批竞品商品，手动复制粘贴价格、库存、规格信息到 Excel，耗时又容易出错；大促期间价格频繁变动，等到人工发现调价，机会早就错过了；想要统计一批商品的价格区间、规格分布，手动统计效率极低。

自己写原生页面解析，会碰到层出不穷的反访问限制、页面结构改版、图片路径解析、规格多维度拆解等一堆麻烦问题。页面结构一改，之前写好的解析逻辑就要全部重写，维护成本很高。

最近在使用 Open Claw 这套工具，它封装好了商品信息解析能力，我们只需要传入商品编号，就可以直接拿到结构化的商品信息，把精力全部放在监控逻辑、数据存储、业务分析上，不用耗费大量时间处理页面解析细节。

本文会完整演示：单商品信息获取、定时监控价格库存变动、数据落地存储、简单数据分析，代码可以直接拿来二次改造，适合部署在本地或者服务器做轻量级业务监控。

提示：本教程仅用于学习研究，请遵守对应平台规则，不要高频大规模访问，避免对平台造成压力。

环境准备

运行代码依赖 requests 库，如果还没有安装，执行：

pip install requests pandas

一、获取商品结构化信息

我们只需要准备好自己的凭证密钥，传入商品编号，工具会返回整理完毕的数据：标题、售价、原价、库存、规格 SKU、店铺信息、图片地址、发货地等全部字段，不需要自己做 HTML 解析。

示例代码：

# -*- coding:utf-8 -*-
import requests
import json

class GoodsHelper:
    def __init__(self, access_key, access_secret):
        self.access_key = access_key
        self.access_secret = access_secret
        self.base_url = "https://api-gw.onebound.cn/jd/item_get_pro"

    def get_goods_detail(self, goods_id):
        """获取商品完整结构化信息"""
        params = {
            "key": self.access_key,
            "secret": self.access_secret,
            "num_iid": goods_id,
            "cache": "no",
            "result_type": "json"
        }
        try:
            resp = requests.get(self.base_url, params=params, timeout=20)
            result = resp.json()
            item_data = result.get("item", {})
            return item_data
        except Exception as e:
            print(f"获取商品异常：{str(e)}")
            return None


if __name__ == "__main__":
    # 替换成自己的凭证
    KEY = "你的key"
    SECRET = "你的secret"
    helper = GoodsHelper(KEY, SECRET)
    goods_info = helper.get_goods_detail("10335871600")
    if goods_info:
        print("商品标题：", goods_info.get("title"))
        print("当前售价：", goods_info.get("price"))
        print("原始标价：", goods_info.get("orginal_price"))
        print("库存：", goods_info.get("num"))
        print("店铺名称：", goods_info.get("seller_info", {}).get("shop_name"))
        print("商品链接：", goods_info.get("detail_url"))
        # 获取多规格信息
        sku_list = goods_info.get("skus", {}).get("sku", [])
        for sku in sku_list:
            print(f"规格:{sku.get('properties_name')} | 价格:{sku.get('price')} | sku库存:{sku.get('quantity')}")

运行之后直接输出结构化结果。我们拿到的数据包含：主图集合、多规格 SKU、品牌、发货地、店铺信息等，省去大量解析工作。

二、实现定时监控，捕捉价格、库存变动

最核心的场景：持续监控目标商品，对比上一轮采集结果，价格下调、库存骤变的时候输出告警提示。可以扩展：告警推送钉钉 / 企业微信、写入本地文件或者数据库持久化。

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

class GoodsHelper:
    def __init__(self, access_key, access_secret):
        self.access_key = access_key
        self.access_secret = access_secret
        self.base_url = "https://api-gw.onebound.cn/jd/item_get_pro"

    def get_goods_detail(self, goods_id):
        params = {
            "key": self.access_key,
            "secret": self.access_secret,
            "num_iid": goods_id,
            "cache": "no",
            "result_type": "json"
        }
        try:
            resp = requests.get(self.base_url, params=params, timeout=20)
            result = resp.json()
            return result.get("item", {})
        except Exception as e:
            print(f"请求出错:{e}")
            return None


class GoodsMonitor:
    def __init__(self, helper):
        self.helper = helper
        self.history = dict()  # 保存上一次采集的数据 {goods_id:data}

    def monitor_loop(self, goods_id_list, interval=300):
        """
        循环监控
        :param goods_id_list: 需要监控的商品id数组
        :param interval: 轮询间隔，单位秒，示例300=5分钟，不要设置过小
        """
        print(f"启动监控任务，监控商品列表：{goods_id_list}，轮询间隔 {interval}s")
        while True:
            now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            for gid in goods_id_list:
                data = self.helper.get_goods_detail(gid)
                if not data:
                    continue
                title = data.get("title")
                price = float(data.get("price", 0))
                stock = int(data.get("num", 0))
                print(f"\n[{now}] 商品:{title[:40]} 价格:{price} 库存:{stock}")

                # 判断是否历史存在
                if gid not in self.history:
                    self.history[gid] = {"price": price, "stock": stock}
                    continue

                old = self.history[gid]
                # 价格变动告警
                if price != old["price"]:
                    print(f"【价格变动告警】{title} 旧价:{old['price']} → 新价:{price}")
                # 库存大幅变动告警
                if abs(stock - old["stock"]) > 20:
                    print(f"【库存异动告警】{title} 旧库存:{old['stock']} → 新库存:{stock}")

                # 更新历史记录
                self.history[gid]["price"] = price
                self.history[gid]["stock"] = stock
            time.sleep(interval)


if __name__ == "__main__":
    KEY = "你的key"
    SECRET = "你的secret"
    helper = GoodsHelper(KEY, SECRET)
    monitor = GoodsMonitor(helper)
    # 设置需要监控的商品编号列表
    watch_list = ["10335871600"]
    # 启动监控，间隔5分钟
    monitor.monitor_loop(watch_list, interval=300)

⚠️注意：轮询间隔不要设置过小，建议最少 3‑5 分钟以上，避免频繁请求。

三、数据落地：保存历史记录到 CSV 做数据分析

采集到的数据不能只打印控制台，我们把每次采集结果写入 csv 文件，后续可以用 pandas 做统计分析，分析价格波动、价格分布区间。

def save_record_to_csv(goods_id, goods_data, csv_path="goods_record.csv"):
    """保存单条采集记录"""
    record = {
        "crawl_time": datetime.datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
        "goods_id": goods_id,
        "title": goods_data.get("title"),
        "price": goods_data.get("price"),
        "orginal_price": goods_data.get("orginal_price"),
        "stock": goods_data.get("num"),
        "shop_name": goods_data.get("seller_info", {}).get("shop_name"),
        "location": goods_data.get("location")
    }
    df_new = pd.DataFrame([record])
    try:
        df_old = pd.read_csv(csv_path)
        df_all = pd.concat([df_old, df_new], ignore_index=True)
    except FileNotFoundError:
        df_all = df_new
    df_all.to_csv(csv_path, index=False, encoding="utf‑8‑sig")


# 简单数据分析示例
def analysis_data(file_path="goods_record.csv"):
    df = pd.read_csv(file_path)
    print("====基础统计分析====")
    print(f"总采集记录条数：{len(df)}")
    print(f"价格均值：{df['price'].mean():.2f}")
    print(f"价格最高：{df['price'].max()}")
    print(f"价格最低：{df['price'].min()}")
    print("\n各商品平均价格：")
    print(df.groupby("goods_id")["price"].mean())


if __name__ == "__main__":
    # 测试写入与分析
    KEY = "你的key"
    SECRET = "你的secret"
    helper = GoodsHelper(KEY, SECRET)
    info = helper.get_goods_detail("10335871600")
    save_record_to_csv("10335871600", info)
    analysis_data()

运行完成之后，本地生成goods_record.csv，里面留存每一次采集快照，后续可以用 Excel 或者 Python 做：价格趋势、大促前后价格对比、竞品价格区间统计等分析工作。

四、业务拓展思路

拿到这套基础能力之后，可以衍生很多实用功能：

批量竞品监控：维护一份商品 id 清单，循环批量采集竞品，输出竞品周报；
消息告警推送：价格 / 库存变动时调用钉钉、企业微信机器人接口，把告警消息推送到工作群；
多规格 SKU 分析：解析返回的 SKU 数组，统计不同规格的定价差异，分析对手的规格定价策略；
持久化存储：把 CSV 替换成 MySQL、SQLite 数据库，支持长期海量历史数据存储；
简单可视化：借助 matplotlib 绘制价格波动曲线，直观看到一段时间价格走势。

踩坑总结与注意事项

不要设置过于频繁的轮询间隔，高频采集不仅会触发限制，也没有实际业务意义；
返回的价格为商品展示价格，部分场景下多规格商品每个细分规格真实标价需要读取 sku 列表字段；
部分店铺不会返回完整店铺信息，这属于平台返回限制，代码做好空值判断；
工具自带缓存开关 cache 参数，如果需要实时最新数据设置cache:"no"；调试的时候可以打开缓存加快响应速度。

总结

Open Claw 最大价值就是帮我们屏蔽了复杂页面解析，不用去处理页面改版、反访问、图片路径、规格拆解等脏活累活。开发者把重心放在监控业务逻辑、数据存储、业务分析，几十行代码就搭建一套可用的商品监控体系。

告别人工盯品：借助 Open Claw 快速搭建商品自动监控与数据分析系统（附完整可运行源码）

Sun, 30 Aug 2026 16:34:35 +0800

前言

做电商运营、竞品跟踪、选品分析，很多人都遇到过这样的困境：

想要持续跟进多款商品的价格变动、库存余量、规格变化、店铺信息，尤其是大促期间，价格和优惠瞬息万变。如果依靠人工打开网页逐个复制记录，不仅每天耗费大量时间，还极易错过降价、库存清空等关键信号。

如果自己从零开发页面解析脚本，坑点会非常多：页面 JS 动态渲染、频繁访问触发平台风控封禁 IP、网页改版后原有解析逻辑直接失效；多规格款式、图片、店铺信息都需要单独编写解析逻辑，维护成本居高不下。

最近在实际项目中体验了 Open Claw 这套工具，很好地解决了以上痛点。不用处理复杂的网页解析、反访问限制等底层琐事，仅编写少量代码，就可以拿到标准化的商品全量信息，实现定时巡检、异动捕捉、历史数据存储，完成基础数据分析工作。

本文为完整实操教程，所有代码可以复制修改后直接运行，适合个人调研和中小团队竞品跟踪场景。

Open Claw 能解决什么问题

拿到完整标准化商品资料：商品标题、原价、实时售价、库存、多规格 SKU 信息、品牌、店铺名称、商品图、发货地等信息直接输出，不需要自己解析网页源码。
支持实时 / 缓存两种模式：做监控时开启实时模式，拿到最新价格库存；日常批量查询开启缓存，提升整体速度，减少资源消耗。
不受前端页面改版影响：网页样式更新不会打乱我们获取数据的逻辑，大幅降低脚本维护工作量。
方便做二次分析：输出结构化数据，可以直接存入本地数据库，做价格趋势、库存波动、竞品对标分析。

提示：使用前需要在后台获取两组身份凭证，用于身份校验。

整体实现思路

整个小系统拆分为 4 个简单模块：

数据获取模块：输入商品编号，获取商品结构化信息。
数据清洗模块：过滤无效字段，提取业务关心的核心字段。
定时监控模块：循环定时巡检商品，对比前后数据，捕捉价格、库存变动。
存储与分析模块：把每次采集快照存入本地数据库，后续做历史查询、简单统计分析。

环境准备

只需要基础 Python 环境，安装网络请求库即可，无需复杂框架。

pip install requests

完整实战代码

新建文件goods_monitor.py，复制下面全部代码，修改配置区凭证、需要监控的商品编号、巡检间隔，即可直接运行。

import requests
import json
import sqlite3
import time
from datetime import datetime

# ===================== 配置区域，请自行修改 =====================
CLAW_KEY = "填入你的凭证key"
CLAW_SECRET = "填入你的凭证secret"
# 需要监控的商品编号，支持多个商品批量监控
MONITOR_GOODS_ID = ["10335871600", "10057467958584"]
POLL_INTERVAL = 300  # 巡检间隔，单位秒，建议不要设置过小
PRICE_CHANGE_THRESHOLD = 0.1  # 价格变动超过10%输出告警
DB_NAME = "goods_monitor.db"
# =================================================================

def init_database():
    """初始化本地数据库，存储商品历史快照"""
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS goods_history (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        goods_id TEXT NOT NULL,
        title TEXT,
        brand TEXT,
        current_price REAL,
        origin_price REAL,
        stock INTEGER,
        shop_name TEXT,
        capture_time TIMESTAMP
    )
    ''')
    conn.commit()
    conn.close()


def get_goods_detail(goods_id: str, real_time: bool = False):
    """
    获取商品详情
    :param goods_id:商品编号
    :param real_time: True获取实时最新数据，False使用缓存提升速度
    :return:清洗后的商品字典，失败返回None
    """
    params = {
        "key": CLAW_KEY,
        "secret": CLAW_SECRET,
        "num_iid": goods_id,
        "cache": "no" if real_time else "yes",
        "result_type": "json",
        "lang": "cn"
    }
    retry_count = 3
    for attempt in range(retry_count):
        try:
            session = requests.Session()
            resp = session.get(
                "https://api-gw.onebound.cn/jd/item_get_pro",
                params=params,
                timeout=15
            )
            resp.raise_for_status()
            res_json = resp.json()
            item_data = res_json.get("item")
            if not item_data:
                print(f"商品 {goods_id} 未获取到有效数据")
                return None
            # 做简单字段清洗
            clean_data = {
                "goods_id": item_data.get("num_iid"),
                "title": item_data.get("title", ""),
                "brand": item_data.get("brand", ""),
                "current_price": float(item_data.get("price", 0)) if item_data.get("price") else 0,
                "origin_price": float(item_data.get("orginal_price", 0)) if item_data.get("orginal_price") else 0,
                "stock": int(item_data.get("num", 0)) if item_data.get("num") else 0,
                "shop_name": item_data.get("seller_info", {}).get("shop_name", "")
            }
            return clean_data
        except Exception as e:
            print(f"第{attempt+1}次请求异常：{str(e)}")
            time.sleep(2)
    return None


def save_to_db(goods_data):
    """将采集到的数据写入本地数据库"""
    if not goods_data:
        return
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute('''
    INSERT INTO goods_history(goods_id,title,brand,current_price,origin_price,stock,shop_name,capture_time)
    VALUES (?,?,?,?,?,?,?,?)
    ''', (
        goods_data["goods_id"],
        goods_data["title"],
        goods_data["brand"],
        goods_data["current_price"],
        goods_data["origin_price"],
        goods_data["stock"],
        goods_data["shop_name"],
        datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    ))
    conn.commit()
    conn.close()


def monitor_task():
    """监控主逻辑，对比历史价格，输出变动告警"""
    last_status = {}
    while True:
        print(f"\n===== 巡检开始 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} =====")
        for gid in MONITOR_GOODS_ID:
            current_info = get_goods_detail(gid, real_time=True)
            if not current_info:
                continue
            save_to_db(current_info)
            print(f"商品:{current_info['title']}")
            print(f"现价:{current_info['current_price']},原价:{current_info['origin_price']},库存:{current_info['stock']}")
            # 对比上一次状态，判断是否触发告警
            if gid in last_status:
                old_price = last_status[gid]["current_price"]
                new_price = current_info["current_price"]
                if old_price > 0 and abs(new_price - old_price)/old_price >= PRICE_CHANGE_THRESHOLD:
                    print(f"⚠️【价格异动提醒】商品{gid}，价格由{old_price}变动至{new_price}")
            last_status[gid] = current_info
        print(f"===== 本轮巡检结束，等待 {POLL_INTERVAL} 秒后下一轮 =====")
        time.sleep(POLL_INTERVAL)


def analysis_demo():
    """简单数据分析示例：读取数据库，打印某商品的价格历史"""
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    target_goods = MONITOR_GOODS_ID[0]
    cursor.execute('''
    SELECT capture_time,current_price,stock FROM goods_history WHERE goods_id=? ORDER BY capture_time
    ''', (target_goods,))
    rows = cursor.fetchall()
    print(f"\n=====商品 {target_goods} 历史价格库存记录 =====")
    for row in rows:
        print(f"时间:{row[0]} | 价格:{row[1]} | 库存:{row[2]}")
    conn.close()


if __name__ == "__main__":
    init_database()
    print("初始化完成，启动监控任务，Ctrl+C终止程序")
    try:
        monitor_task()
    except KeyboardInterrupt:
        print("\n程序手动终止，运行analysis_demo()可以查看历史数据分析结果")
        analysis_demo()

代码说明

初始化数据库：本地 SQLite 文件数据库，无需额外安装数据库服务，每次采集会写入一条快照记录。
数据获取函数：内置 3 次失败重试，网络抖动的时候提高稳定性；支持切换实时 / 缓存模式。
数据清洗：只保留业务上最核心字段，过滤大量无关内容。
监控主循环：定时循环遍历监控列表，对比前后价格，变动幅度超过阈值直接控制台输出告警。
简易分析示例：程序终止之后，可以调用analysis_demo()读取本地存储的历史记录，输出价格、库存随时间变化记录。

如何落地扩展

1、告警能力升级

现在告警只是打印控制台，实际项目可以扩展为：

对接企业微信 / 飞书机器人，价格变动推送消息到工作群
邮件通知、短信提醒

2、多规格 SKU 监控

原始返回内容中包含完整 SKU 数组，每一个子规格拥有独立编号、价格、库存。可以修改解析逻辑，单独跟踪每一个规格的价格与库存，适合服装、多型号数码产品。

3、导出报表

可以结合 pandas 库，直接读取 sqlite 数据库，导出 csv、Excel 报表，用于运营周报，做竞品对标分析。

import pandas as pd
conn = sqlite3.connect("goods_monitor.db")
df = pd.read_sql("select * from goods_history",conn)
df.to_excel("商品监控报表.xlsx",index=False)

4、部署后台持续运行

Linux 服务器可以使用 nohup、systemd 把脚本放在后台长期运行，实现 7×24 小时不间断监控。Windows 可以使用任务计划程序定时启动。

注意事项

巡检间隔不要设置过小，过高频率不仅消耗配额，也没有实际业务意义。普通商品建议 3‑10 分钟巡检一次。
身份凭证注意保管，不要随代码上传公开代码仓库。
原始返回中会携带大量字段，开发时可以根据业务需求，按需取用，不用全部保存入库。

总结

很多做电商、做调研的同学，要么人工手动记录，要么花大量时间维护网页抓取脚本。Open Claw 把底层复杂的网页处理做了封装，开发者只需要聚焦业务逻辑：监控哪些商品、什么条件触发提醒、如何分析历史数据。

告别人工盯品：借助 Open‑Claw 快速搭建电商商品监控与数据分析体系（附完整 Python 代码）

Thu, 27 Aug 2026 10:37:03 +0800

前言

做电商运营、选品、竞品调研的时候，相信很多朋友都遇到过这些痛点：

需要持续盯多款竞品，人工反复打开页面查看价格、库存、促销活动，耗费大量时间；
大促期间竞品调价、库存变动节奏很快，人很难做到 7×24 小时实时捕捉变化；
拿到一堆零散的商品信息，很难沉淀历史数据，做价格趋势、SKU 规格对比分析；
直接从网页解析页面，容易遇到访问限制、页面改版导致脚本直接失效，维护成本很高。

之前我尝试过写页面解析脚本，但是平台页面结构经常调整，隔一段时间代码就失效，调试反爬相关逻辑消耗大量精力。后来接触到 Open‑Claw，它可以直接输出结构化标准化的商品全量资料，不需要再折腾页面解析逻辑，我们只需要专注做监控逻辑、数据存储、业务分析，快速搭一套属于自己的商品监测工具。

本篇文章面向开发者、电商从业者，完整讲解从获取商品信息、变化监测、数据存储，再到简单数据分析的整套落地流程，代码可以直接修改配置后运行。

说明：本教程仅做技术学习演示，使用时请遵守对应平台使用规范，合理控制访问频次。

整体实现思路

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

商品信息获取模块：输入商品编号，返回结构化完整商品资料，包含标题、售价、原价、库存、规格、店铺信息、图片等内容；
变化检测模块：对比上一轮保存的数据，识别价格涨跌、库存变动、规格变更，输出告警提示；
本地持久化模块：把每一轮采集的数据保存到本地 JSON/CSV，沉淀历史时序数据；
简易数据分析模块：读取历史记录，做价格波动统计、SKU 规格整理，输出简单统计结果。

环境准备

依赖安装，执行下面命令：

pip install requests pandas

完整可运行代码实现

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

# ========== 配置区，请修改为你自己的凭证和监控商品 ==========
CLAW_KEY = "填入你的key"
CLAW_SECRET = "填入你的secret"
# 需要监控的商品编号列表
MONITOR_GOODS = ["10335871600", "10057467958584"]
# 循环检测间隔，单位秒，不建议设置过小，示例5分钟轮询一次
MONITOR_INTERVAL = 300
# 降价告警阈值，0.1代表价格下跌超过10%触发提醒
PRICE_DROP_ALERT = 0.1
# 历史数据存储文件
HISTORY_FILE = "goods_history.json"
CSV_EXPORT = "goods_record.csv"
# ==========================================================


def load_last_cache():
    """加载上一轮缓存的商品数据"""
    if os.path.exists(HISTORY_FILE):
        with open(HISTORY_FILE, "r", encoding="utf‑8") as f:
            return json.load(f)
    return {}


def save_cache(data):
    """保存当前采集结果到本地缓存文件"""
    with open(HISTORY_FILE, "w", encoding="utf‑8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def fetch_goods_data(goods_id):
    """
    获取单个商品结构化数据
    :param goods_id: 商品数字编号
    :return: 解析完成的商品字典，失败返回None
    """
    base_url = "https://api‑gw.onebound.cn/jd/item_get_pro"
    params = {
        "key": CLAW_KEY,
        "secret": CLAW_SECRET,
        "num_iid": goods_id,
        "result_type": "json",
        "cache": "yes",
        "lang": "cn"
    }
    headers = {
        "Accept‑Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=20)
        resp_json = resp.json()
        # 判断返回状态，非0000代表异常
        if resp_json.get("error_code") != "0000":
            err_msg = resp_json.get("reason", "未知异常")
            print(f"【获取失败】商品{goods_id}，原因：{err_msg}")
            return None
        item_data = resp_json.get("item", {})
        if not item_data:
            print(f"【获取失败】商品{goods_id}返回内容为空")
            return None

        # 提取核心业务字段，做标准化整理
        parse_result = {
            "goods_id": item_data.get("num_iid"),
            "title": item_data.get("title", ""),
            "price": float(item_data.get("price", 0)) if item_data.get("price") else 0.0,
            "original_price": float(item_data.get("orginal_price", 0)) if item_data.get("orginal_price") else 0.0,
            "stock": item_data.get("num", 0),
            "shop_name": item_data.get("seller_info", {}).get("shop_name", ""),
            "detail_url": item_data.get("detail_url", ""),
            "collect_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
            "skus": item_data.get("skus", {}).get("sku", [])
        }
        return parse_result
    except Exception as e:
        print(f"【请求异常】商品{goods_id}, 错误：{str(e)}")
        return None


def detect_change(old_data, new_data):
    """对比新旧两份数据，检测价格库存变化，输出告警信息"""
    alert_list = []
    if not old_data:
        alert_list.append("首次采集该商品，无历史对比")
        return alert_list

    old_price = old_data["price"]
    new_price = new_data["price"]
    old_stock = old_data["stock"]
    new_stock = new_data["stock"]

    # 价格变动检测
    if abs(new_price - old_price) > 0.01:
        change_rate = (new_price - old_price) / old_price if old_price > 0 else 0
        if new_price < old_price and abs(change_rate) >= PRICE_DROP_ALERT:
            alert_list.append(f"⚠️降价告警！{old_price} → {new_price}，降幅{abs(change_rate)*100:.1f}%")
        elif new_price > old_price:
            alert_list.append(f"📈涨价提醒！{old_price} → {new_price}")
        else:
            alert_list.append(f"💡价格微调：{old_price} → {new_price}")

    # 库存变动检测
    if new_stock != old_stock:
        if new_stock == 0:
            alert_list.append(f"🚨库存告急，商品变为无货！原库存：{old_stock}")
        elif old_stock == 0 and new_stock > 0:
            alert_list.append(f"✅商品补货成功！当前库存 {new_stock}")
        else:
            alert_list.append(f"📦库存变动 {old_stock} → {new_stock}")
    return alert_list


def export_to_csv(all_records):
    """把采集记录导出csv，方便后续做表格分析"""
    rows = []
    for goods_id, records in all_records.items():
        for rec in records:
            rows.append({
                "采集时间": rec["collect_time"],
                "商品ID": rec["goods_id"],
                "商品标题": rec["title"],
                "现价": rec["price"],
                "原价": rec["original_price"],
                "库存": rec["stock"],
                "店铺名称": rec["shop_name"]
            })
    df = pd.DataFrame(rows)
    df.to_csv(CSV_EXPORT, index=False, encoding="utf‑8‑sig")


def simple_analysis():
    """简易数据分析示例：读取csv，统计价格区间、输出基础统计信息"""
    if not os.path.exists(CSV_EXPORT):
        print("暂无导出记录，跳过分析")
        return
    df = pd.read_csv(CSV_EXPORT, encoding="utf‑8‑sig")
    print("\n=====简易数据分析报告=====")
    for gid in df["商品ID"].unique():
        sub_df = df[df["商品ID"] == gid]
        print(f"\n商品ID {gid}：{sub_df.iloc[0]['商品标题'][:30]}...")
        print(f"  采集记录条数：{len(sub_df)}")
        print(f"  历史最低价格：{sub_df['现价'].min()}")
        print(f"  历史最高价格：{sub_df['现价'].max()}")
        print(f"  平均价格：{round(sub_df['现价'].mean(),2)}")


def main_loop():
    print("====商品监控程序启动====")
    cache_data = load_last_cache()
    history_records = {}
    # 初始化历史记录容器
    for gid in MONITOR_GOODS:
        if gid not in history_records:
            history_records[gid] = []

    while True:
        now_time = datetime.now().strftime("%Y‑%m‑%d %H:%M:%S")
        print(f"\n【本轮采集时间】{now_time}")
        for goods_id in MONITOR_GOODS:
            new_item = fetch_goods_data(goods_id)
            if new_item is None:
                continue
            # 保存到全量历史，用于导出报表
            history_records[goods_id].append(new_item)
            old_item = cache_data.get(goods_id)
            alerts = detect_change(old_item, new_item)
            print(f"\n>>商品：{new_item['title'][:40]}")
            for msg in alerts:
                print(f"    {msg}")
            # 更新缓存为当前最新数据
            cache_data[goods_id] = new_item
        # 写入本地缓存文件
        save_cache(cache_data)
        export_to_csv(history_records)
        simple_analysis()
        print(f"\n本轮结束，等待 {MONITOR_INTERVAL} 秒进入下一轮")
        time.sleep(MONITOR_INTERVAL)


if __name__ == "__main__":
    try:
        main_loop()
    except KeyboardInterrupt:
        print("\n程序手动停止，已保存当前数据文件")

代码核心功能讲解

数据获取函数 fetch_goods_data 不需要解析网页 HTML，直接拿到整理好的结构化结果，提取标题、价格、原价、库存、SKU 规格、店铺、链接等关键字段。返回异常状态做捕获，商品下架、编号错误、权限问题都会打印提示。
变化检测函数 detect_change 拿上一轮缓存的数据和本轮对比，自动识别涨价、降价（设置阈值，小幅度波动不告警）、缺货、补货，输出告警文本。可以后续扩展，对接企业微信、钉钉消息推送，实现真正消息提醒。
持久化存储

goods_history.json：保存每个商品最新一轮数据，用于下一轮对比；
goods_record.csv：每一轮全部历史记录写入表格，用 Excel 打开就可以直接做透视、画图，看价格随时间变化走势。

简易数据分析 simple_analysis 读取 CSV 历史，对每个商品输出历史最高价、最低价、平均成交价，快速看商品的价格波动区间。

拓展优化方向

1. 消息告警推送

现在告警只打印控制台，可以增加钉钉 / 企业微信 webhook，当检测到降价、缺货的时候，直接推送消息到工作群，不用盯着控制台运行。

2. 数据库存储

小体量用 CSV 完全够用；如果监控几十上百款商品，建议换成 SQLite、MySQL，方便做复杂查询，比如筛选某时间段所有降价商品。

3.SKU 粒度监控

当前示例取商品整体信息，返回结果自带完整 SKU 数组，可以改造代码，针对每一个子规格单独记录价格库存，适合多规格复杂商品的竞品分析。

4. 可视化图表

结合 matplotlib、plotly，读取 CSV 历史数据，自动绘制每个商品价格趋势折线图，直观看到大促前后价格变动轨迹。

5. 任务调度分离

长时间运行可以把采集逻辑做成后台定时任务，不需要一直开着脚本；Linux 环境下可配置 systemd 服务或者 crontab 定时执行。

踩坑记录与注意事项

轮询间隔不要设置过小，短时间高频访问会触发限流，建议至少间隔 3‑5 分钟以上；
部分商品多规格场景，整体展示价格和子 SKU 实际售价存在差异，做精细分析时，优先读取 SKU 数组内部对应规格的价格；
部分自营、第三方店铺，店铺相关字段不一定全部返回，代码已经做了空值兼容；
遇到商品下架，返回结果会提示商品未找到，脚本会捕获异常不会直接崩溃；
凭证信息不要硬编码提交到公开代码仓库，生产环境建议放到环境变量读取。

业务场景可以怎么用

竞品监控：跟踪竞品多款商品，记录调价节奏、大促降价幅度，给自己的定价、促销活动做参考；
采购备货参考：持续看竞品库存变动，辅助判断市场热度，辅助备货决策；
个人好物蹲价：想要买的商品，设置降价阈值，一旦低于心理价位立刻收到提醒；
选品分析：批量收集一批候选商品历史价格、规格，筛选性价比高的目标单品。

写在最后

这套方案最大的优势，是不用投入精力维护页面解析逻辑，把重心放在业务逻辑上。Open‑Claw 帮我们处理了页面各种复杂解析工作，输出规整结构化数据，开发者可以快速把精力投入监控告警、历史统计、数据分析业务。

上面这份代码可以直接跑通，只需要填入自己的凭证，填写需要监控的商品编号，就可以立刻拥有一套属于自己的商品监测工具。后续可以基于这个基础，按需扩展推送、数据库、可视化等功能。

告别人工盯品：借助 OpenClaw 搭建商品自动监控与数据分析系统（完整实操）

Thu, 27 Aug 2026 10:03:16 +0800

前言

做电商运营，免不了要持续关注自家以及竞品商品状态：价格变动、库存变化、不同规格的售价波动、店铺基础信息更新。很多人最开始的做法就是打开网页，定时手动刷新页面复制数据，一旦要跟踪十几款商品，每天就要消耗大量人力。

如果自己写页面解析脚本，又会遇到不少麻烦：网页改版后解析规则直接失效，高频访问容易触发平台风控，多规格、图片、店铺信息的解析逻辑需要大量开发时间，遇到网络波动、商品下架，程序很容易直接报错终止运行。

最近在实际项目中尝试了 OpenClaw 这套工具，它可以直接输出标准化结构化商品资料，不用处理页面解析、代理池、加密校验，我们只需要聚焦业务逻辑：定时拉取信息、本地存储、异动识别、数据分析，几十行代码就可以完成一套完整的监控方案稀土掘金。

整套流程分为 4 个部分：

资料获取：输入商品编号，拿到标题、原价、现价、库存、全部规格、店铺、图片等完整内容；
定时轮询：周期性采集目标商品，对比历史记录，识别价格、库存异动；
本地持久化：把每一次采集结果保存到本地数据库，留存时序记录；
数据分析：读取历史记录，做价格波动统计、规格销量统计，输出简单报表。

准备工作

安装依赖库

pip install requests apscheduler sqlite3

2.拿到两组身份凭证，用于 OpenClaw 的身份校验；

3.准备需要跟踪的商品编号，就是商品网页链接中 item.jd.com/xxx.html 中间那串数字。

提示：工具自带缓存机制，做实时监控的时候建议关闭缓存，保证拿到最新的行情；做批量数据分析场景，开启缓存可以减少消耗，提升响应速度。工具内置异常识别，商品下架、网络故障、平台维护都会返回对应的标记，程序不会直接崩溃退出。

完整可运行代码实现

1. 基础数据获取、清洗模块

import requests
import json
import time
import sqlite3
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime

# ----------------配置区域----------------
KEY = "你的身份凭证key"
SECRET = "你的身份凭证secret"
MONITOR_ITEM_LIST = ["10335871600","10057467958584"] # 需要监控的商品编号列表
MONITOR_INTERVAL = 3600 # 监控周期，单位秒，这里设置1小时采集一次
CACHE_SWITCH = "no" # 实时监控设置no；批量分析可以设置yes开启缓存
# ----------------------------------------

def fetch_product_data(num_iid):
    """获取商品原始资料"""
    base_url = "https://api‑gw.onebound.cn/jd/item_get_pro"
    param = {
        "key": KEY,
        "secret": SECRET,
        "num_iid": num_iid,
        "cache": CACHE_SWITCH,
        "result_type":"json",
        "lang":"cn"
    }
    headers = {
        "Accept‑Encoding":"gzip",
        "Connection":"close"
    }
    try:
        resp = requests.get(base_url,params=param,headers=headers,timeout=20)
        res_json = resp.json()
        return res_json
    except Exception as e:
        print(f"商品{num_iid}获取失败，异常信息：{str(e)}")
        return None


def parse_raw_data(raw_json):
    """清洗原始返回内容，提取业务需要字段"""
    if not raw_json:
        return None
    error_code = raw_json.get("error_code")
    if error_code != "0000":
        err_msg = raw_json.get("reason","未知异常")
        print(f"业务异常：{err_msg}")
        return None
    item = raw_json.get("item")
    if not item:
        return None

    # 提取多规格sku列表
    sku_list = []
    sku_data = item.get("skus",{}).get("sku",[])
    for sku in sku_data:
        sku_info = {
            "sku_id":sku.get("sku_id"),
            "properties_name":sku.get("properties_name"),
            "price":float(sku.get("price",0)),
            "original_price":float(sku.get("orginal_price",0)),
            "stock":int(sku.get("quantity",0))
        }
        sku_list.append(sku_info)

    clean_result = {
        "num_iid":item.get("num_iid"),
        "title":item.get("title"),
        "brand":item.get("brand",""),
        "current_price":float(item.get("price",0)),
        "original_price":float(item.get("orginal_price",0)),
        "stock_total":item.get("num",0),
        "shop_name":item.get("seller_info",{}).get("shop_name",""),
        "main_image":item.get("pic_url",""),
        "location":item.get("location",""),
        "skus":json.dumps(sku_list,ensure_ascii=False),
        "crawl_time":datetime.now().strftime("%Y‑%m‑%d %H:%M:%S")
    }
    return clean_result

2. SQLite 本地存储模块

把每次采集的数据存入本地数据库，方便后续对比和数据分析。

def init_db():
    """初始化本地数据表"""
    conn = sqlite3.connect("./product_monitor.db")
    cursor = conn.cursor()
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS product_history(
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        num_iid TEXT,
        title TEXT,
        brand TEXT,
        current_price REAL,
        original_price REAL,
        stock_total INTEGER,
        shop_name TEXT,
        skus TEXT,
        crawl_time TEXT
    )
    ''')
    conn.commit()
    conn.close()


def save_to_db(data):
    """将清洗完成的数据写入数据库"""
    if not data:
        return
    conn = sqlite3.connect("./product_monitor.db")
    cursor = conn.cursor()
    sql = '''
    INSERT INTO product_history
    (num_iid,title,brand,current_price,original_price,stock_total,shop_name,skus,crawl_time)
    VALUES (?,?,?,?,?,?,?,?,?)
    '''
    val = (
        data["num_iid"],data["title"],data["brand"],
        data["current_price"],data["original_price"],data["stock_total"],
        data["shop_name"],data["skus"],data["crawl_time"]
    )
    cursor.execute(sql,val)
    conn.commit()
    conn.close()
    print(f"✅成功保存记录：商品编号{data['num_iid']}，采集时间{data['crawl_time']}")

3. 异动检测 + 定时监控任务

对比上一轮记录，识别价格下调上涨、库存减少，简单打印预警，实际项目可以扩展为企业微信、钉钉消息推送。

def check_abnormal(num_iid,new_data):
    """对比历史数据，检测异动"""
    conn = sqlite3.connect("./product_monitor.db")
    cursor = conn.cursor()
    cursor.execute('''
    SELECT current_price,stock_total FROM product_history
    WHERE num_iid = ? ORDER BY id DESC LIMIT 1
    ''',(num_iid,))
    last_record = cursor.fetchone()
    conn.close()
    if not last_record:
        return
    last_price,last_stock = last_record
    new_price = new_data["current_price"]
    new_stock = new_data["stock_total"]

    # 价格异动
    if new_price != last_price:
        print(f"⚠️【价格异动提醒】商品{num_iid}，旧价{last_price} → 新价{new_price}")
    # 库存异动
    if new_stock != last_stock:
        print(f"⚠️【库存异动提醒】商品{num_iid}，旧库存{last_stock} → 新库存{new_stock}")


def monitor_task():
    """单次监控任务，循环处理全部目标商品"""
    for item_id in MONITOR_ITEM_LIST:
        raw = fetch_product_data(item_id)
        clean_data = parse_raw_data(raw)
        if clean_data:
            check_abnormal(item_id,clean_data)
            save_to_db(clean_data)
        time.sleep(1) # 请求之间短暂休眠，避免请求过于密集

4. 简单数据分析示例

读取本地历史数据，输出价格变化统计，你可以基于此扩展，做 matplotlib 绘图导出图表，或者导出 csv 报表。

def analysis_product(num_iid):
    """针对单个商品，读取历史记录做简单数据分析"""
    conn = sqlite3.connect("./product_monitor.db")
    cursor = conn.cursor()
    cursor.execute('''
    SELECT crawl_time,current_price,stock_total FROM product_history
    WHERE num_iid = ? ORDER BY id
    ''',(num_iid,))
    rows = cursor.fetchall()
    conn.close()
    if len(rows) == 0:
        print(f"商品{num_iid}暂无历史记录")
        return
    price_list = [r[1] for r in rows]
    min_p = min(price_list)
    max_p = max(price_list)
    avg_p = sum(price_list)/len(price_list)
    print(f"\n📊商品{num_iid}历史数据分析结果")
    print(f"记录总条数：{len(rows)}")
    print(f"历史最低价格：{min_p}")
    print(f"历史最高价格：{max_p}")
    print(f"平均价格：{round(avg_p,2)}")
    print("时间‑价格轨迹：")
    for line in rows:
        print(f"  {line[0]} | 价格:{line[1]} |库存:{line[2]}")


if __name__ == "__main__":
    init_db()
    print("启动商品监控程序...")
    scheduler = BackgroundScheduler()
    scheduler.add_job(monitor_task,"interval",seconds=MONITOR_INTERVAL)
    scheduler.start()
    # 先执行一次采集
    monitor_task()
    # 演示数据分析，填入需要分析的商品编号
    analysis_product(MONITOR_ITEM_LIST[0])

    try:
        while True:
            time.sleep(60)
    except KeyboardInterrupt:
        print("\n程序退出")
        scheduler.shutdown()

关键说明与踩坑记录

关于多规格处理 返回的规格列表中，每一条都独立带有规格编号、售价、库存。实际业务不要只拿商品主体价格，主体价格有可能会和真实规格售价不一致，做竞品比价建议遍历全部规格分别跟踪记录。
缓存开关怎么选

7*24 小时实时盯价格、盯库存：CACHE_SWITCH="no"，每次都获取最新行情；
一次性批量拉取大量商品做选品分析：CACHE_SWITCH="yes"，复用缓存，提升速度、减少消耗。

异常状态处理 当商品下架、编号错误的时候，返回结果会带上异常标记，代码中已经做判断，不会直接崩溃。日常开发可以把异常日志写入文件，方便排查问题。
扩展方向

异动消息推送：检测到价格 / 库存变动后，调用钉钉 / 企业微信推送通知；
可视化：读取 sqlite 数据，用 matplotlib 绘制价格趋势图，导出图片；
导出报表：把历史记录直接导出为 CSV，交给运营做复盘；
多商品批量分析，批量导出全部监控商品统计报告。

小结

对比传统手写解析脚本，OpenClaw 最大的优势就是把繁杂的页面适配工作做了封装。开发者不用再关心网页改版、反拦截、JS 渲染这些问题，我们把精力聚焦业务：定时采集、异动告警、数据统计分析。几十行代码，就实现 7×24 小时不间断商品监控，对于中小团队和个人开发者是性价比很高的落地方案。

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

Tue, 25 Aug 2026 16:25:04 +0800

前言

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

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

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

整体实现思路

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

原始信息获取：输入商品编号，拿到商品标题、现价、原价、库存、多规格信息、店铺、图片等全套原始内容；
持久化存储与定时监控：定时循环采集，把每一次采集结果保存，对比前后数据，捕捉价格、库存变动；
数据分析处理：读取历史记录，做价格波动统计、规格对比、基础数据可视化，输出可以直接参考的分析结果。

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

环境准备

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

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价格走势图，可以直接放到报告里。

实际业务场景拓展

多商品批量监控：把商品 ID 放到列表循环处理，就可以同时监控一批竞品，分别保存记录；
消息提醒：检测到变动事件后，可以对接企业微信 / 钉钉机器人，一旦降价、库存清零，立刻推送通知；
规格维度分析：解析出来的 skus 列表，可以单独保存每个规格的历史，分析哪个规格经常调价；
数据导出报告：Pandas 可以直接输出 Excel，方便运营同学做周报复盘。

踩坑与注意事项

部分场景下返回的价格会存在偏差，这是数据来源本身特性，做分析时要结合业务校验，不要直接当作绝对唯一依据；
不要把轮询间隔设置太短，建议至少 5‑10 分钟以上，避免短时间大量请求；
如果返回找不到商品，优先核对商品编号是否正确，确认商品是否已经下架；
做好异常捕获，网络波动时程序不会直接崩溃，保证监控任务稳定跑。

写在最后

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

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

告别人工盯品！借助 Open Claw 搭建电商商品自动化监控与数据分析系统（完整可运行源码）

Mon, 24 Aug 2026 17:46:33 +0800

前言

做电商运营或者竞品调研，相信很多人都有这样的痛点：想要持续跟踪多款商品的价格变动、库存、规格信息，每天手动打开网页复制粘贴数据到 Excel 表格。不仅重复劳动耗时间，大促节点的临时降价、库存清仓往往转瞬即逝，人工很难第一时间捕捉变化。

如果自己手写页面解析脚本，又会遇到页面动态渲染、网页改版导致解析失效、访问频次受限等一系列棘手问题，调试维护成本居高不下。

最近在项目实践中使用了 Open Claw 工具，它可以直接输出结构化的商品信息，不用再折腾网页源码解析，我们可以把精力全部放在业务逻辑上，快速搭建一套本地商品监控与数据分析程序。

本篇教程会完整演示整套流程：包含本地数据持久化存储、价格库存异动提醒、历史时序数据分析、批量商品巡检，所有代码可以直接复制运行，Windows、Mac、Linux 都支持，部署到服务器就可以实现 7×24 小时不间断监控。

整体实现思路

获取商品结构化原始信息：商品标题、当前售价、原始标价、库存、多规格信息、店铺名称等完整内容；
使用 SQLite 本地数据库保存每一次采集记录，沉淀历史时序数据；
设置定时巡检任务，对比前后两次数据，识别降价、库存变动，输出告警提示；
读取本地历史记录做数据分析，统计历史最高、最低、平均售价，分析价格波动幅度；
支持批量加入多款商品，扩展可对接钉钉、企业微信实现消息推送。

环境准备

Python 版本建议 3.7 以上，安装需要的依赖包

pip install requests apscheduler

requests：负责网络交互
apscheduler：实现定时巡检任务
sqlite3：Python 内置轻量数据库，无需额外安装，用来存储监控历史记录

提前准备好 Open Claw 后台生成的两组身份凭证，用于程序访问。

完整代码实现

新建文件 goods_monitor.py，复制下面全部代码，修改配置区域的凭证、监控商品编号、告警阈值就可以直接运行。

import requests
import json
import sqlite3
import time
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler

# ===================== 【配置区域，请自行修改】 =====================
CLAW_KEY = "填入你的凭证key"
CLAW_SECRET = "填入你的凭证secret"
# 需要监控的商品编号，可以填入多个实现批量监控
MONITOR_GOODS = ["10335871600", "10057467958584"]
POLL_INTERVAL = 300  # 巡检间隔，单位秒，普通商品建议300秒，爆款可缩短
PRICE_ALERT_RATIO = 0.15  # 降价超过15%就触发告警
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,
        goods_id TEXT NOT NULL,
        title TEXT,
        price REAL,
        original_price REAL,
        stock INTEGER,
        shop_name TEXT,
        record_time TIMESTAMP
    )
    ''')
    conn.commit()
    conn.close()


def get_goods_info(goods_id: str):
    """
    获取商品结构化数据
    :param goods_id: 商品数字编号
    :return: 字典格式商品信息，失败返回None
    """
    params = {
        "key": CLAW_KEY,
        "secret": CLAW_SECRET,
        "num_iid": goods_id,
        "cache": "no",
        "result_type": "json"
    }
    retry = 3
    while retry > 0:
        try:
            resp = requests.get(
                "https://api-gw.onebound.cn/jd/item_get_pro",
                params=params,
                timeout=15
            )
            data = resp.json()
            if data.get("error_code") != "0000":
                print(f"获取商品{goods_id}异常：{data.get('reason','未知错误')}")
                return None
            item_data = data.get("item", {})
            result = {
                "goods_id": goods_id,
                "title": item_data.get("title", ""),
                "price": float(item_data.get("price", 0)) if item_data.get("price") else 0,
                "original_price": float(item_data.get("orginal_price", 0)) if item_data.get("orginal_price") else 0,
                "stock": int(item_data.get("num", 0)) if item_data.get("num") else 0,
                "shop_name": item_data.get("seller_info", {}).get("shop_name", "")
            }
            return result
        except Exception as e:
            retry -= 1
            print(f"网络异常，剩余重试次数 {retry}, 错误：{str(e)}")
            time.sleep(2)
    return None


def save_to_db(goods_data):
    """将采集到的数据存入本地数据库"""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    cursor.execute('''
    INSERT INTO goods_record
    (goods_id, title, price, original_price, stock, shop_name, record_time)
    VALUES (?, ?, ?, ?, ?, ?, ?)
    ''', (
        goods_data["goods_id"],
        goods_data["title"],
        goods_data["price"],
        goods_data["original_price"],
        goods_data["stock"],
        goods_data["shop_name"],
        datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    ))
    conn.commit()
    conn.close()


def get_last_record(goods_id):
    """读取该商品上一次记录，用于对比变动"""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    cursor.execute('''
    SELECT price, stock FROM goods_record WHERE goods_id=?
    ORDER BY record_time DESC LIMIT 1
    ''', (goods_id,))
    row = cursor.fetchone()
    conn.close()
    if row:
        return {"price": row[0], "stock": row[1]}
    return None


def monitor_task():
    """定时执行的监控任务"""
    print(f"\n==== 巡检开始 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ====")
    for gid in MONITOR_GOODS:
        info = get_goods_info(gid)
        if not info:
            continue
        save_to_db(info)
        last = get_last_record(gid)
        print(f"商品【{info['title']}】 当前价格：{info['price']} 库存：{info['stock']}")
        if last:
            # 降价告警判断
            if last["price"] > 0 and info["price"] < last["price"]:
                drop_rate = (last["price"] - info["price"]) / last["price"]
                if drop_rate >= PRICE_ALERT_RATIO:
                    print(f"【⚠️价格告警】商品{gid}发生大幅降价！上次:{last['price']} 本次:{info['price']} 降幅{drop_rate:.1%}")
            # 库存变动提醒
            if info["stock"] != last["stock"]:
                print(f"【ℹ库存变动】商品{gid}库存变更，上一次:{last['stock']}，当前:{info['stock']}")
        time.sleep(1)


def analysis_goods(goods_id):
    """针对单个商品做历史数据分析"""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    cursor.execute('''
    SELECT price, record_time FROM goods_record WHERE goods_id=? ORDER BY record_time
    ''', (goods_id,))
    rows = cursor.fetchall()
    conn.close()
    if not rows:
        print("暂无历史数据，请先运行监控采集数据！")
        return
    price_list = [r[0] for r in rows]
    max_p = max(price_list)
    min_p = min(price_list)
    avg_p = sum(price_list)/len(price_list)
    print("\n========商品历史数据分析========")
    print(f"历史最高售价：{max_p:.2f}")
    print(f"历史最低售价：{min_p:.2f}")
    print(f"历史平均售价：{avg_p:.2f}")
    print(f"数据记录总条数：{len(price_list)}")


if __name__ == "__main__":
    init_db()
    scheduler = BackgroundScheduler()
    scheduler.add_job(monitor_task, "interval", seconds=POLL_INTERVAL)
    scheduler.start()
    print("监控程序已经启动，按 Ctrl+C 终止程序")
    try:
        while True:
            cmd = input("\n输入指令：\n1:执行一次巡检  2:分析商品历史  q:退出\n>")
            if cmd == "1":
                monitor_task()
            elif cmd == "2":
                gid = input("输入要分析的商品编号：")
                analysis_goods(gid)
            elif cmd.lower() == "q":
                break
    except KeyboardInterrupt:
        pass
    finally:
        scheduler.shutdown()
        print("程序已退出")

程序使用说明

修改代码顶部配置，填入自己的两组凭证，把需要跟踪的商品编号写进MONITOR_GOODS列表；
直接运行脚本 python goods_monitor.py，程序自动初始化数据库，开启后台定时巡检；
控制台交互指令：

输入1立刻手动执行一轮巡检，不用等待定时周期；
输入2，填入商品编号，程序会统计该商品全部历史记录，输出最高价、最低价、平均价格；
输入q关闭程序。

采集的全部历史会保存在goods_monitor.db文件，你可以使用数据库工具打开，导出为 Excel 表格做进一步处理。

功能拓展方向

消息推送：出现降价、库存变动告警的时候，可以增加钉钉机器人、企业微信 webhook，把告警信息推送到工作群，不用守着控制台；
多规格 SKU 分析：源码中只提取了基础价格，还可以读取多规格的数据，跟踪不同规格的价格、库存变化；
数据可视化：结合 matplotlib 绘制价格变化曲线图，直观观察一段时间的价格走势；
导出报表：定时把每日数据导出 csv 文件，方便周报复盘；
部署服务器：Linux 服务器后台运行，实现全天候无人值守监控。

实际业务场景

竞品跟踪：电商运营持续盯对手商品调价、库存情况，把握促销时机；
选品比价：跟踪多款目标商品，等待合适价位再入手；
店铺自检：跟踪自家店铺多款商品，监控异常改价、库存异常；
大促备战：618、双 11 大促期间高频巡检，捕捉短期促销活动。

写在最后

这套方案避开了复杂的页面解析工作，把重点放在业务逻辑开发，新手也可以快速落地。实际使用的时候注意控制巡检间隔，不要设置过于频繁，避免超出额度。如果后续需要更大规模的批量任务，可以在这个基础上做任务队列优化。

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

Fri, 21 Aug 2026 10:47:18 +0800

前言

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

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

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

最近在实践中发现 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()输出统计报告。

拓展方向

上面是最小可用版本，基于这个脚本还可以继续拓展很多实用功能：

消息推送：捕捉到降价缺货之后，对接钉钉 / 企业微信机器人，把告警消息推送到工作群。
多规格分析：解析返回的多规格列表，跟踪每一个不同款式的价格与库存变化。
可视化图表：使用 matplotlib 或者 plotly，读取 csv 历史数据绘制价格趋势折线图，直观看到一段时间价格走势。
多商品批量管理：把监控商品列表放到配置文件或者轻量数据库，不用修改代码就可以增删监控商品。
阈值告警：设置心理价位，当商品价格低于目标价格的时候才触发提醒。

踩坑小结

工具返回的主价格字段有时候会存在偏差，如果业务对价格精度要求高，优先读取规格列表里面对应款式的价格。
部分自营和非自营店铺返回字段会有差异，部分字段为空属于正常现象，代码要做好空值容错，避免脚本直接崩溃。
如果出现获取数据失败，优先检查密钥凭证是否填写正确，网络是否正常连通。
不要把轮询间隔设置得很短，合理控制采集频次，兼顾稳定性与实际业务需求。

写在最后

对于中小运营团队或者个人开发者来说，直接从零做商品数据跟踪，要投入不少精力去处理页面变化、访问限制。Open Claw 把底层繁杂的工作封装完成，我们只需要聚焦业务逻辑，几行代码就能搭建一套完整监控分析系统。

竞品监控与商品数据分析实战：用 Open‑Claw 搭建自动化商品跟踪系统

Thu, 20 Aug 2026 09:33:03 +0800

摘要：做电商运营、选品、竞品调研，最头疼的就是手动刷页面记录价格、库存、规格变化。反复复制粘贴，效率低下还容易漏掉关键变动。本文分享一套可直接落地的实战方案，借助 Open‑Claw 工具快速完成商品信息自动化抓取，实现价格、库存、SKU 规格持续监控，搭配 Python 完成数据清洗、本地存储与简单统计分析，完整代码可直接运行，适合个人开发者、中小商家做市场情报收集。

一、业务痛点：为什么需要自动化商品监控

很多做电商调研的朋友日常会遇到这些问题：

跟踪几十款竞品，每天手动打开页面记录价格、库存，耗费大量时间；
错过竞品调价、上新规格、库存清零的节点，错失市场决策时机；
拿到一堆零散复制的文本，没有结构化数据，很难做价格带统计、规格对比；
直接写浏览器模拟采集，经常遇到访问限制、页面改版、签名校验，调试成本极高。

传统的页面模拟采集，要处理 Cookie、设备指纹、页面结构变动，维护成本很高。Open‑Claw 封装好了底层的访问适配能力，我们只需要传入商品编号，就能拿到结构化完整商品信息，省去大量逆向、调试的时间，把精力聚焦在数据分析业务逻辑上。

我们可以实现这些能力：

批量输入商品编号，批量拉取商品标题、售价、原价、店铺信息、SKU 规格、库存、主图、发货地等全套信息；
定时循环执行，持续监控价格变动、SKU 增减、库存状态；
将数据保存本地 CSV，方便 Excel 做透视分析；
简单逻辑判断，检测到价格变动、缺货时输出提醒日志。

温馨提示：本教程仅用于个人学习研究，请遵守平台相关使用规范，控制请求频率，不要高频大规模批量调用。

二、整体流程思路

整体流程非常清晰：

准备工具凭证，获取身份密钥；
Python 脚本组装请求，传入商品编号获取完整结构化数据；
解析返回结果，提取业务关心字段；
数据写入本地 CSV 文件持久化；
增加循环延时，实现定时监控；
简单统计，做价格、库存维度的数据分析。

需要提前安装依赖包：

pip install requests pandas

三、完整可运行 Python 代码

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

# ========== 配置区域，请修改为自己的凭证和目标商品 ==========
KEY = "你的身份key"
SECRET = "你的身份密钥"
# 需要监控的商品ID列表，可以填入多个商品编号
MONITOR_ITEM_IDS = ["10335871600", "10057467958584"]
# 监控间隔，单位秒，示例设置30分钟扫描一次，请勿设置过小
MONITOR_INTERVAL = 1800
# 本地存储文件
SAVE_CSV = "goods_monitor_data.csv"
# =========================================================

def fetch_goods_data(item_id):
    """获取单款商品结构化数据"""
    base_url = "https://api‑gw.onebound.cn/jd/item_get_pro"
    params = {
        "key": KEY,
        "secret": SECRET,
        "num_iid": item_id,
        "result_type": "json",
        "cache": "no"  # 设置no获取最新数据；yes读取缓存速度更快
    }
    headers = {
        "Accept‑Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=20)
        result = resp.json()
    except Exception as e:
        print(f"商品 {item_id} 请求异常：{str(e)}")
        return None

    error_code = result.get("error_code")
    if error_code != "0000":
        err_msg = result.get("error", "未知错误")
        print(f"商品 {item_id} 获取失败，错误：{err_msg}")
        return None

    item_data = result.get("item", {})
    if not item_data:
        print(f"商品 {item_id} 返回内容为空")
        return None

    # 提取核心字段
    output = {
        "crawl_time": time.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": item_data.get("price"),
        "orginal_price": item_data.get("orginal_price"),
        "stock": item_data.get("num"),
        "shop_name": item_data.get("seller_info", {}).get("shop_name", ""),
        "detail_url": item_data.get("detail_url"),
        "pic_url": item_data.get("pic_url")
    }
    return output


def save_to_csv(data_list):
    """把采集到的数据追加写入CSV"""
    df_new = pd.DataFrame(data_list)
    try:
        df_old = pd.read_csv(SAVE_CSV, encoding="utf‑8‑sig")
        df_total = pd.concat([df_old, df_new], ignore_index=True)
    except FileNotFoundError:
        df_total = df_new
    df_total.to_csv(SAVE_CSV, index=False, encoding="utf‑8‑sig")
    print(f"✅ {len(data_list)} 条数据已写入本地文件 {SAVE_CSV}")


def analysis_data():
    """简单本地数据分析示例，读取CSV做基础统计"""
    try:
        df = pd.read_csv(SAVE_CSV, encoding="utf‑8‑sig")
        df["price"] = pd.to_numeric(df["price"], errors="coerce")
        print("\n==== 简单数据分析报告 ====")
        print(f"总采集记录数：{len(df)}")
        print(f"监控商品数量：{df['num_iid'].nunique()}")
        print(f"商品平均售价：{df['price'].mean():.2f}")
        print("\n各商品最新价格：")
        latest = df.sort_values("crawl_time").groupby("num_iid").last()
        print(latest[["title", "price", "stock"]])
    except Exception as e:
        print(f"分析数据异常：{str(e)}，可能还没有采集到数据")


def main_loop():
    print("🚀 启动商品监控任务")
    while True:
        collect_result = []
        for goods_id in MONITOR_ITEM_IDS:
            item_info = fetch_goods_data(goods_id)
            if item_info:
                collect_result.append(item_info)
                print(f"已获取：{item_info.get('title')} | 当前价：{item_info.get('price')}")
            time.sleep(2)  # 每个请求之间间隔，避免访问过于密集
        if collect_result:
            save_to_csv(collect_result)
            analysis_data()
        print(f"\n⏳ 等待 {MONITOR_INTERVAL} 秒后进行下一轮扫描...\n")
        time.sleep(MONITOR_INTERVAL)


if __name__ == "__main__":
    main_loop()

四、代码讲解

fetch_goods_data 函数：负责传入商品编号，拉取完整商品信息，处理网络异常、返回状态判断，提取我们关心的标题、价格、库存、店铺等核心字段。

cache 参数：设置yes优先读取缓存，响应速度更快；设置no拉取实时最新数据，适合监控价格变动场景。

save_to_csv 函数：将每一轮抓取的数据追加保存到本地 CSV 文件，utf‑8‑sig编码保证 Excel 打开不会乱码。后续可以直接用 Excel 打开，做数据透视、价格对比。
analysis_data 函数：简单的本地数据分析演示，统计总记录、平均价格，输出每个商品最新状态。实际业务可以在此扩展，例如检测价格波动，当本次价格和上一轮记录不一致，打印告警提示。
main_loop 主循环：循环遍历所有待监控商品，间隔一定时间重复执行，实现持续监控。不要把 MONITOR_INTERVAL 设置过小，避免短时间大量请求。

运行脚本后，项目目录下会生成goods_monitor_data.csv文件，每一轮采集都会追加新的一行，附带采集时间戳，可以完整记录商品随时间变化的轨迹。

五、拓展：SKU 规格拆解与深度分析

上面示例只提取了商品基础信息，如果需要分析多规格变体，返回数据中自带 SKU 列表，我们可以简单改造解析 SKU。示例片段（放在fetch_goods_data函数内部）：

# 解析SKU规格示例
sku_list = item_data.get("skus", {}).get("sku", [])
for sku in sku_list:
    print(f"规格名称：{sku.get('properties_name')}")
    print(f"规格价格：{sku.get('price')}，库存：{sku.get('quantity')}")

拿到每个规格的价格、库存后，就可以做更细粒度的分析：比如统计一款商品不同规格的价格梯度，观察哪些规格长期缺货，辅助选品和竞品定价分析。

六、踩坑经验分享

价格偶尔出现偏差：部分场景返回的价格和页面展示会有出入，属于数据源本身的情况，适合做趋势监控，不要直接当做 100% 精准成交价格。
商品下架：当商品已经被删除、下架，程序会捕获错误并打印日志，可以据此做商品状态跟踪。
凭证密钥保护：不要直接硬编码提交到代码仓库，生产环境建议放到环境变量中读取。
频率控制：不要把轮询间隔设置很短，高频请求不仅消耗额度，也容易触发限制，做监控一般 15‑60 分钟一轮足够业务使用。

七、落地应用场景总结

这套方案可以服务于很多业务场景：

竞品跟踪：持续跟踪同行多款商品的调价、库存、规格迭代，及时捕捉对手动作；
选品调研：批量收集一批候选商品，导出到表格分析价格区间、库存水平，筛选潜力款；
自家商品巡检：监控自己上架商品，快速发现异常缺货、价格异常；
个人比价：跟踪心仪商品，观察历史价格变化。

写在最后：工具只是手段，真正价值在于结构化之后的数据。当我们拥有时间序列的商品记录，不再靠肉眼刷网页，很多市场趋势、竞品策略，都可以从数据中直观看到。

如果你需要进一步扩展，可以加上告警逻辑，检测价格变动后发送消息通知；也可以把 CSV 替换成 SQLite、MySQL 数据库，支持更大规模的数据存储。