OpenClaw 实战：搭建京东商品监控与简易数据分析脚本

前言

做电商运营、竞品分析、选品调研的时候，我们经常需要持续盯一批商品：价格波动、库存变化、规格变动、销量走势。手动打开网页复制粘贴，效率低还容易遗漏关键变化。

最近折腾 OpenClaw 这套工具，发现可以快速把商品基础信息定时拉取到本地，做成轻量化监控脚本。不需要复杂爬虫框架，几行代码就可以完成定时采集、数据落盘、简单分析，适合个人小团队做竞品跟踪、选品复盘。

注意：本文仅用于技术学习，使用时请遵守平台规则，控制访问频率，请勿高频批量抓取，避免对网站造成压力。

工具准备

环境依赖：Python3，requests，pandas，sqlite3（内置）

pip install requests pandas

我们会实现这些功能：

根据商品编号获取完整商品元数据：标题、原价、售价、库存、SKU 规格、店铺信息、发货地等
定时采集，把历史价格、库存存入本地 SQLite 数据库，留存变化记录
简单数据分析：价格变动统计、库存波动、导出 CSV 报表
简易告警逻辑：价格大跌 / 库存清零打印提示

完整实现代码

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

# --------------------------配置区域---------------------------
KEY = "your_key_here"
SECRET = "your_secret_here"
# 需要监控的商品ID列表
MONITOR_ITEM_IDS = ["10335871600"]
# 采集间隔，单位秒，测试不要设置过小，建议至少300秒以上
FETCH_INTERVAL = 600
DB_PATH = "./jd_monitor.db"
# -------------------------------------------------------------

def init_db():
    """初始化数据库，存储每次采集快照"""
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()
    cur.execute('''
    CREATE TABLE IF NOT EXISTS item_history (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        num_iid TEXT,
        title TEXT,
        price REAL,
        original_price REAL,
        stock INTEGER,
        sales INTEGER,
        shop_name TEXT,
        location TEXT,
        fetch_time TIMESTAMP
    )
    ''')
    conn.commit()
    conn.close()


def fetch_item_detail(num_iid):
    """获取单款商品完整详情数据"""
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    base_url = "https://api-gw.onebound.cn/jd/item_get_pro/"
    params = {
        "key": KEY,
        "secret": SECRET,
        "num_iid": num_iid,
        "result_type": "json",
        "cache": "no"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=20)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        print(f"请求异常 num_iid:{num_iid}, err:{e}")
        return None


def save_snapshot(raw_data):
    """把单次采集结果写入数据库快照"""
    if not raw_data or "item" not in raw_data:
        return
    item = raw_data.get("item", {})
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()
    cur.execute('''
    INSERT INTO item_history
    (num_iid,title,price,original_price,stock,sales,shop_name,location,fetch_time)
    VALUES (?,?,?,?,?,?,?,?,?)
    ''', (
        item.get("num_iid"),
        item.get("title"),
        float(item.get("price", 0)),
        float(item.get("orginal_price", 0)),
        int(item.get("num", 0)),
        int(item.get("sales", 0)),
        item.get("nick", ""),
        item.get("location", ""),
        datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    ))
    conn.commit()
    conn.close()

    # 简单监控告警逻辑
    price = float(item.get("price", 0))
    stock = int(item.get("num", 0))
    print(f"【快照】{item.get('title')} | 当前价:{price} |库存:{stock} | {datetime.now()}")
    if stock <= 0:
        print(f"⚠️警告：商品 {item.get('num_iid')} 库存归零！")


def export_analysis_csv():
    """导出历史数据，做数据分析，输出csv"""
    conn = sqlite3.connect(DB_PATH)
    df = pd.read_sql("SELECT * FROM item_history ORDER BY fetch_time DESC", conn)
    conn.close()
    csv_name = f"./item_analysis_{datetime.now().strftime('%Y%m%d_%H%M')}.csv"
    df.to_csv(csv_name, index=False, encoding="utf_8_sig")
    print(f"✅分析报表导出完成：{csv_name}")
    return df


def simple_analysis(df):
    """简易数据分析：统计价格最大最小，价格变动次数，库存变化"""
    if df.empty:
        print("暂无采集数据")
        return
    print("\n========简易分析报告========")
    for item_id in df["num_iid"].unique():
        sub = df[df["num_iid"] == item_id]
        min_p = sub["price"].min()
        max_p = sub["price"].max()
        latest_price = sub.iloc[0]["price"]
        latest_stock = sub.iloc[0]["stock"]
        record_cnt = len(sub)
        print(f"商品ID:{item_id}")
        print(f"记录条数:{record_cnt} |历史最低:{min_p} |历史最高:{max_p} |最新价格:{latest_price} |最新库存:{latest_stock}")
        print("-"*40)


def monitor_loop():
    init_db()
    print("开始商品监控循环，按Ctrl+C停止程序")
    try:
        while True:
            for item in MONITOR_ITEM_IDS:
                res = fetch_item_detail(item)
                if res:
                    save_snapshot(res)
            # 每采集三轮导出一次报表，可自行修改
            export_analysis_csv()
            time.sleep(FETCH_INTERVAL)
    except KeyboardInterrupt:
        print("\n收到停止信号，准备导出最终报表...")
        df = export_analysis_csv()
        simple_analysis(df)
        print("程序退出")


if __name__ == "__main__":
    monitor_loop()

代码解读

数据库初始化：使用轻量 SQLite 保存每一轮的快照，每一次拉取都作为一条历史记录，后续用来回溯价格、库存变化。不需要部署 MySQL，开箱即用。
数据拉取函数：填入你的凭证和商品编号，就可以拿到商品标题、售价、原价、库存、店铺名称、发货地、SKU 规格全套信息。
快照存储：每次成功获取信息，就写入一条记录；同时内置简单告警，如果库存变为 0 控制台输出提醒。
报表导出与分析：定时把全部历史快照导出 CSV，直接可以用 Excel 打开做图表分析。脚本内置简单统计：最高最低价格、记录数量、最新状态。
循环监控：设置间隔时间，持续轮询。测试环境建议不要把间隔调太小，避免触发访问限制。

运行之后控制台输出示例：

开始商品监控循环，按Ctrl+C停止程序
【快照】京东鲜花爱丽丝泡泡鲜切花云南鲜花家庭插花爱丽丝泡泡10枝源头直发 | 当前价:30.9 |库存:99 | 2026‑07‑31 15:22:10
✅分析报表导出完成：./item_analysis_20260731_1522.csv

按下Ctrl+C终止脚本，程序会自动导出最后一次完整分析报告，打印各商品的价格区间统计。

拓展玩法

1. SKU 多规格监控

返回结果里面自带 skus 数组，可以扩展脚本把每一个子规格的价格、库存分开存入子表，监控不同规格的调价、缺货。

# 示例，解析sku片段
skus_data = raw_data["item"].get("skus",{})
sku_list = skus_data.get("sku",[])
for sku in sku_list:
    print(sku["properties_name"],sku["price"],sku["quantity"])

2. 邮件 / 企业微信告警

当检测到价格暴跌、库存售罄，可以增加推送逻辑，把告警消息推送到企业微信机器人、邮箱，不用盯着控制台。

3. Pandas 做可视化

导出 DataFrame 后，直接用 matplotlib 绘制价格走势图，把一段时间的价格波动生成图片保存，适合做竞品周报素材。

import matplotlib.pyplot as plt
plt.rcParams["font.sans-serif"]=["SimHei"]
df["fetch_time"] = pd.to_datetime(df["fetch_time"])
plt.plot(df["fetch_time"],df["price"])
plt.title("商品价格监控曲线")
plt.savefig("./price_trend.png")

踩坑和注意事项

凭证信息不要硬编码上传公开代码仓库，实际项目建议放到环境变量。
平台会提供缓存开关，调试阶段把 cache 设置为 no 拿到实时数据；日常监控打开缓存，可以减少消耗。
返回中的价格字段，部分场景下会和页面展示存在偏差，属于原始数据本身限制，做分析时要留意。
不要超大批量同时监控几百上千商品，控制轮询间隔，合理使用工具。

小结

借助 OpenClaw，不用手写复杂解析逻辑，只需要专注业务逻辑：存快照、写告警、做数据分析。对于个人做竞品跟踪、选品复盘，这套脚本足够轻便好用。

万邦api博客

Nice to meet you, too!

api 数据挖掘