5 分钟搭建京东商品自动化监控系统，低成本搞定竞品价格、库存、销量数据分析（附完整可运行 Python 源码）

pip install requests schedule pandas

import requests
import json
import time

class JDProductCollector:
    def __init__(self, key, secret):
        # 身份凭证，替换为自己后台获取的密钥
        self.key = key
        self.secret = secret
        self.base_url = "https://api-gw.onebound.cn/jd/item_get_pro"
        self.session = requests.Session()  # 长连接优化请求速度

    def get_product_detail(self, item_id, real_time=False):
        """
        拉取京东商品完整结构化数据
        :param item_id: 京东商品数字ID
        :param real_time: True=关闭缓存，获取实时最新价格库存；False=开启缓存提速
        :return: dict 商品结构化数据 / None 采集失败
        """
        params = {
            "key": self.key,
            "secret": self.secret,
            "api_name": "jd.item_get_pro",
            "num_iid": item_id,
            "cache": "no" if real_time else "yes",
            "result_type": "json",
            "lang": "cn"
        }
        retry_times = 3  # 采集失败自动重试3次
        for attempt in range(retry_times):
            try:
                resp = self.session.get(self.base_url, params=params, timeout=15)
                resp.raise_for_status()
                result = resp.json()
                # 判断返回状态，0000代表采集成功
                if result.get("error_code") == "0000":
                    return result["item"]
                else:
                    print(f"商品{item_id}采集异常：{result.get('reason', '未知错误')}")
                    return None
            except Exception as e:
                print(f"第{attempt+1}次采集失败，商品ID:{item_id}，错误详情：{str(e)}")
                time.sleep(2)
        print(f"商品{item_id}已达到最大重试次数，采集跳过")
        return None

import schedule
import pandas as pd
from datetime import datetime
import os

class ProductMonitor:
    def __init__(self, collector, monitor_item_ids):
        self.collector = collector
        self.monitor_item_ids = monitor_item_ids  # 需要监控的商品ID列表
        self.history_file = "jd_monitor_history.json"
        self.history_data = self.load_history()  # 加载历史数据用于对比变动

    def load_history(self):
        """读取本地历史存储文件，无文件则初始化空列表"""
        if os.path.exists(self.history_file):
            with open(self.history_file, "r", encoding="utf-8") as f:
                return json.load(f)
        return []

    def save_history(self):
        """将本次采集全量数据写入本地文件持久化"""
        with open(self.history_file, "w", encoding="utf-8") as f:
            json.dump(self.history_data, f, ensure_ascii=False, indent=2)

    def compare_data_change(self, old_data, new_data):
        """对比新旧数据，识别价格、库存变动"""
        change_log = []
        old_price = float(old_data.get("price", 0))
        new_price = float(new_data.get("price", 0))
        old_stock = int(old_data.get("num", 0))
        new_stock = int(new_data.get("num", 0))

        if new_price < old_price:
            change_log.append(f"价格下跌：{old_price} → {new_price}")
        elif new_price > old_price:
            change_log.append(f"价格上调：{old_price} → {new_price}")
        if new_stock != old_stock:
            change_log.append(f"库存变动：{old_stock} → {new_stock}")
        if change_log:
            return " | ".join(change_log)
        return "无变动"

    def monitor_single_product(self, item_id):
        """单个商品采集、对比、记录逻辑"""
        now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        product_info = self.collector.get_product_detail(item_id, real_time=True)
        if not product_info:
            return
        # 组装本次采集记录
        record = {
            "crawl_time": now_time,
            "item_id": item_id,
            "title": product_info["title"],
            "price": product_info["price"],
            "original_price": product_info["orginal_price"],
            "stock_total": product_info["num"],
            "sales": product_info.get("sales", 0),
            "shop_name": product_info.get("seller_info", {}).get("shop_name", "未知店铺"),
            "sku_list": product_info.get("skus", {}).get("sku", []),
            "raw_data": product_info
        }
        # 查询历史记录，对比变动
        item_history = [i for i in self.history_data if i["item_id"] == item_id]
        if item_history:
            latest_old = item_history[-1]
            record["change_info"] = self.compare_data_change(latest_old, record)
            if record["change_info"] != "无变动":
                print(f"【变动提醒】{record['title']} | {record['change_info']}")
        else:
            record["change_info"] = "首次采集，无历史对比"
        self.history_data.append(record)
        self.save_history()

    def full_monitor_task(self):
        """批量监控全部商品的主任务"""
        print(f"\n===== 启动一轮全量监控 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} =====")
        for pid in self.monitor_item_ids:
            self.monitor_single_product(pid)
            time.sleep(1)  # 控制采集间隔，避免请求密集
        print("本轮监控全部完成，历史数据已保存")

    def start_timed_monitor(self, interval_hour=1):
        """启动定时循环监控，默认每小时执行一次"""
        schedule.every(interval_hour).hours.do(self.full_monitor_task)
        print(f"定时监控已启动，每{interval_hour}小时自动巡检一次商品")
        # 程序持续运行，执行定时任务
        while True:
            schedule.run_pending()
            time.sleep(60)

    def export_analysis_report(self):
        """导出Excel数据分析报表，用于趋势查看"""
        if not self.history_data:
            print("暂无监控历史数据，无法导出报表")
            return
        df = pd.DataFrame([{
            "采集时间": row["crawl_time"],
            "商品ID": row["item_id"],
            "商品标题": row["title"],
            "当前售价": row["price"],
            "原价": row["original_price"],
            "总库存": row["stock_total"],
            "总销量": row["sales"],
            "店铺名称": row["shop_name"],
            "数据变动": row["change_info"]
        } for row in self.history_data])
        save_path = f"京东商品监控报表_{datetime.now().strftime('%Y%m%d')}.xlsx"
        df.to_excel(save_path, index=False)
        print(f"数据分析报表已导出，文件路径：{save_path}")

if __name__ == "__main__":
    # ========== 配置区域（仅修改这里即可使用） ==========
    CLIENT_KEY = "你的后台Key"
    CLIENT_SECRET = "你的后台Secret"
    # 填入需要监控的京东商品ID列表
    MONITOR_GOODS_IDS = [
        "10335871600",
        "10057467958584"
    ]
    MONITOR_INTERVAL = 1  # 定时监控间隔，单位：小时
    # ==================================================

    # 初始化采集器与监控器
    collector = JDProductCollector(CLIENT_KEY, CLIENT_SECRET)
    monitor = ProductMonitor(collector, MONITOR_GOODS_IDS)

    # 执行一次全量采集测试
    monitor.full_monitor_task()
    # 导出当日数据分析报表
    monitor.export_analysis_report()
    # 开启定时循环监控（注释该行则只执行单次采集）
    monitor.start_timed_monitor(interval_hour=MONITOR_INTERVAL)