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

pip install requests pandas

# -*- 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

# 内存缓存，保存上一轮采集结果
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)

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