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

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

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

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

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

我们可以实现这些能力：

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

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

二、整体流程思路

整体流程非常清晰：

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

需要提前安装依赖包：

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()

四、代码讲解

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

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

1. save_to_csv 函数：将每一轮抓取的数据追加保存到本地 CSV 文件，utf‑8‑sig编码保证 Excel 打开不会乱码。后续可以直接用 Excel 打开，做数据透视、价格对比。

2. analysis_data 函数：简单的本地数据分析演示，统计总记录、平均价格，输出每个商品最新状态。实际业务可以在此扩展，例如检测价格波动，当本次价格和上一轮记录不一致，打印告警提示。

3. 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')}")

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

六、踩坑经验分享

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

七、落地应用场景总结

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

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

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

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