电商竞品全自动监控 + 数据分析落地教程，半小时搞定全品类行情追踪

pip install requests pandas matplotlib openpyxl schedule

import requests
import json
import pandas as pd
import matplotlib.pyplot as plt
import schedule
import time
from datetime import datetime
import os

# 基础配置
# 工具后台获取的个人凭证（替换为自己账号生成的密钥）
TOKEN = "xxxxxxxxxxxxxxxxxxxx"
# 需要监控的商品ID列表，批量新增直接追加
MONITOR_GOODS = ["723456123987", "723456987654", "723456543210"]
# 数据存储文件
DATA_FILE = "goods_monitor_data.csv"
# 绘图字体适配中文
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei"]
plt.rcParams["axes.unicode_minus"] = False

def get_goods_info(item_id):
    """
    调用云端数据能力，获取单款商品完整信息
    :param item_id: 商品唯一标识
    :return: 结构化商品字典数据
    """
    headers = {
        "Content-Type": "application/json"
    }
    params = {
        "token": TOKEN,
        "item_id": item_id
    }
    # 工具数据请求地址
    resp = requests.get("https://data-claw.collector.work/shop/goods", headers=headers, params=params)
    res_data = resp.json()
    if res_data.get("code") != 200:
        print(f"商品{item_id}数据获取失败：{res_data.get('msg')}")
        return None
    info = res_data.get("data")
    # 提取核心业务字段
    result = {
        "crawl_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "item_id": item_id,
        "title": info.get("title"),
        "price": float(info.get("price", 0)),
        "prom_price": float(info.get("prom_price", 0)),
        "sales": int(info.get("sales", 0)),
        "stock": int(info.get("stock", 0)),
        "shop_name": info.get("shop_name"),
        "score": float(info.get("score", 0)),
        "review_count": int(info.get("review_count", 0))
    }
    return result

def save_and_check_change(goods_data):
    """
    写入本地CSV历史文件，对比历史数据识别价格/销量异动
    """
    if not goods_data:
        return
    # 初始化数据文件
    if not os.path.exists(DATA_FILE):
        df_init = pd.DataFrame([goods_data])
        df_init.to_csv(DATA_FILE, index=False, encoding="utf-8-sig")
        print(f"首次录入商品{goods_data['item_id']}监控数据")
        return
    # 读取历史数据
    df_history = pd.read_csv(DATA_FILE, encoding="utf-8-sig")
    item_id = goods_data["item_id"]
    # 筛选该商品历史最新一条记录
    history_item = df_history[df_history["item_id"] == item_id]
    # 写入新数据
    df_new = pd.concat([df_history, pd.DataFrame([goods_data])], ignore_index=True)
    df_new.to_csv(DATA_FILE, index=False, encoding="utf-8-sig")

    # 异动判断逻辑
    if len(history_item) > 0:
        last_record = history_item.iloc[-1]
        price_old = last_record["prom_price"]
        price_new = goods_data["prom_price"]
        sales_old = last_record["sales"]
        sales_new = goods_data["sales"]
        # 价格变动预警（差价大于5元触发提醒）
        if abs(price_new - price_old) > 5:
            print(f"【价格异动提醒】商品{goods_data['title']}，原价{price_old}→现价{price_new}")
        # 销量暴涨提醒（单次增量超200）
        if sales_new - sales_old > 200:
            print(f"【销量暴涨提醒】商品{goods_data['title']}，新增销量{sales_new - sales_old}")

def monitor_task():
    """单次批量监控任务，循环所有商品"""
    print(f"\n===== 开始执行批量监控 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} =====")
    for good_id in MONITOR_GOODS:
        data = get_goods_info(good_id)
        if data:
            save_and_check_change(data)
    print("===== 本轮监控执行完成 =====")

def data_analysis_export():
    """读取全部历史数据，统计分析并导出Excel+价格趋势图"""
    if not os.path.exists(DATA_FILE):
        print("暂无监控数据，无法分析")
        return
    df = pd.read_csv(DATA_FILE, encoding="utf-8-sig")
    # 1. 基础统计
    stat_df = df.groupby(["item_id", "title"]).agg(
        min_price=("prom_price", "min"),
        max_price=("prom_price", "max"),
        avg_price=("prom_price", "mean"),
        total_sales=("sales", "max"),
        latest_score=("score", "last")
    ).reset_index()
    # 导出统计报表
    stat_df.to_excel("竞品行情统计报表.xlsx", index=False)
    print("行情统计报表已生成：竞品行情统计报表.xlsx")

    # 2. 绘制价格趋势图（取第一个商品做示例）
    target_id = MONITOR_GOODS[0]
    target_data = df[df["item_id"] == target_id].sort_values("crawl_time")
    if len(target_data) > 1:
        fig, ax = plt.subplots(figsize=(12, 6))
        ax.plot(target_data["crawl_time"], target_data["prom_price"], marker="o", label="活动售价")
        ax.set_title("商品价格历史波动趋势")
        ax.set_xlabel("监控时间")
        ax.set_ylabel("价格（元）")
        plt.xticks(rotation=45)
        ax.legend()
        plt.tight_layout()
        plt.savefig("price_trend.png")
        print("价格趋势图表已保存：price_trend.png")

if __name__ == "__main__":
    # 设置每1小时自动执行一轮监控，可修改为30/60/120分钟
    schedule.every(1).hours.do(monitor_task)
    # 程序启动先执行一次监控+数据分析
    monitor_task()
    data_analysis_export()
    print("\n监控程序后台运行中，每小时自动同步数据，按Ctrl+C终止")
    # 持续循环调度
    while True:
        schedule.run_pending()
        time.sleep(60)