告别人工盯品！Open Claw 轻量化搭建京东商品 7×24 小时监控与数据分析系统（完整可运行 Python 源码）

pip install requests pandas schedule openpyxl

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

# ====================== 【配置区，自行修改】 ======================
APP_KEY = "你的OpenClaw访问密钥"
APP_SECRET = "你的OpenClaw安全密钥"
# 待监控商品ID列表，可批量新增多个商品
MONITOR_GOODS_ID = [
    "10335871600",
    "10057467958584"
]
# 监控参数自定义
PRICE_ALERT_RATE = 0.15  # 价格跌幅超过15%触发降价预警
STOCK_ALERT_NUM = 30     # 库存低于30触发库存预警
SALE_ALERT_INCREMENT = 200  # 单次销量增长200触发爆款预警
CHECK_INTERVAL_MIN = 30  # 每30分钟自动巡检一次
# 本地存储文件名称
SAVE_EXCEL_NAME = "京东商品监控历史数据.xlsx"
# =================================================================

def get_goods_detail(goods_id):
    """
    调用Open Claw通道获取京东商品完整详情数据
    :param goods_id: 商品唯一ID
    :return: 结构化商品字典 / None（获取失败）
    """
    gateway_url = "https://api-gw.onebound.cn/jd/item_get_pro"
    params = {
        "key": APP_KEY,
        "secret": APP_SECRET,
        "num_iid": goods_id,
        "cache": "yes",
        "result_type": "json"
    }
    try:
        resp = requests.get(url=gateway_url, params=params, timeout=12)
        resp_data = resp.json()
        # 判断是否成功返回商品数据
        if resp_data.get("error_code") != "0000":
            print(f"【采集失败】商品{goods_id}，错误提示：{resp_data.get('reason')}")
            return None
        item_info = resp_data.get("item", {})
        # 提取核心字段，整理成标准化字典
        standard_data = {
            "crawl_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "goods_id": goods_id,
            "title": item_info.get("title", ""),
            "brand": item_info.get("brand", ""),
            "current_price": float(item_info.get("price", 0)),
            "original_price": float(item_info.get("orginal_price", 0)),
            "total_stock": int(item_info.get("num", 0)),
            "sales_volume": int(item_info.get("sales", 0)),
            "shop_name": item_info.get("seller_info", {}).get("shop_name", ""),
            "location": item_info.get("location", ""),
            "detail_link": item_info.get("detail_url", "")
        }
        return standard_data
    except Exception as e:
        print(f"【网络异常】商品{goods_id}，异常信息：{str(e)}")
        return None

def save_data_to_excel(data_list):
    """将采集到的数据追加保存至本地Excel"""
    new_df = pd.DataFrame(data_list)
    # 判断文件是否存在，不存在则新建，存在则追加写入
    try:
        old_df = pd.read_excel(SAVE_EXCEL_NAME)
        merge_df = pd.concat([old_df, new_df], ignore_index=True)
    except FileNotFoundError:
        merge_df = new_df
    merge_df.to_excel(SAVE_EXCEL_NAME, index=False)
    print(f"✅ 本次采集{len(data_list)}款商品数据，已保存至本地表格")

def check_data_warning(new_data, history_df):
    """对比历史数据，检测价格、库存、销量异动，输出预警"""
    goods_id = new_data["goods_id"]
    goods_title = new_data["title"]
    # 筛选该商品上一次采集的历史记录
    goods_history = history_df[history_df["goods_id"] == goods_id]
    if len(goods_history) < 1:
        print(f"ℹ️ 商品{goods_title}暂无历史数据，本次仅存档，不触发对比")
        return
    last_record = goods_history.iloc[-1]
    # 1. 价格异动检测
    price_drop_rate = (last_record["current_price"] - new_data["current_price"]) / last_record["current_price"]
    if price_drop_rate >= PRICE_ALERT_RATE:
        print(f"\033[31m🔴 降价预警 | {goods_title}\033[0m")
        print(f"历史售价：{last_record['current_price']}元 | 当前售价：{new_data['current_price']}元 | 降幅{round(price_drop_rate*100,2)}%")
    # 2. 库存预警
    if new_data["total_stock"] < STOCK_ALERT_NUM:
        print(f"\033[33m🟠 库存预警 | {goods_title}\033[0m")
        print(f"当前剩余库存：{new_data['total_stock']}件，低于警戒值{STOCK_ALERT_NUM}")
    # 3. 销量暴涨预警
    sale_grow = new_data["sales_volume"] - last_record["sales_volume"]
    if sale_grow >= SALE_ALERT_INCREMENT:
        print(f"\033[32m🟢 爆款预警 | {goods_title}\033[0m")
        print(f"新增销量：{sale_grow}，短期流量爆发，可重点跟进")

def goods_data_analysis():
    """简易数据分析函数：输出7日内价格波动、竞品对比简报"""
    try:
        df = pd.read_excel(SAVE_EXCEL_NAME)
        df["crawl_time"] = pd.to_datetime(df["crawl_time"])
        seven_day_data = df[df["crawl_time"] >= pd.Timestamp.now() - pd.Timedelta(days=7)]
        print("\n===================== 7日商品数据分析简报 =====================")
        # 按商品分组统计均价、最低价、总销量
        analysis_result = seven_day_data.groupby(["goods_id", "title"]).agg(
            avg_price=("current_price", "mean"),
            min_price=("current_price", "min"),
            latest_stock=("total_stock", "last"),
            total_sales=("sales_volume", "last")
        ).reset_index()
        print(analysis_result.to_string(index=False))
        print("================================================================\n")
    except FileNotFoundError:
        print("暂无历史监控数据，采集一轮后即可生成分析报表")

def monitor_task():
    """单次定时监控主任务"""
    print(f"\n===== {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 启动自动巡检 =====")
    collect_data = []
    # 遍历所有待监控商品，采集数据
    for gid in MONITOR_GOODS_ID:
        item_data = get_goods_detail(gid)
        if item_data:
            collect_data.append(item_data)
    if len(collect_data) == 0:
        print("本次未采集到有效商品数据，等待下一轮巡检")
        return
    # 保存数据到表格
    save_data_to_excel(collect_data)
    # 读取历史数据，触发异动预警
    history_data = pd.read_excel(SAVE_EXCEL_NAME)
    for item in collect_data:
        check_data_warning(item, history_data)
    # 执行数据分析，输出简报
    goods_data_analysis()
    print(f"===== 本轮巡检完成，{CHECK_INTERVAL_MIN}分钟后自动执行下一轮 =====\n")

if __name__ == "__main__":
    # 配置定时循环任务
    schedule.every(CHECK_INTERVAL_MIN).minutes.do(monitor_task)
    print(f"系统启动成功！监控商品总数：{len(MONITOR_GOODS_ID)}，巡检间隔：{CHECK_INTERVAL_MIN}分钟")
    print("等待首次巡检执行...\n")
    # 持续循环运行定时任务
    while True:
        schedule.run_pending()
        time.sleep(10)

python jd_goods_monitor.py