你想要用 Python 结合京东 API 搭建一套「竞品分析系统」，核心目标是批量抓取竞品商品数据、进行多维度对比分析（价格、销量、库存等）、生成可视化分析报告。我会提供完整可落地的方案，涵盖「数据采集 → 数据存储 → 多维度分析 → 可视化输出」全流程，兼顾实用性和易扩展性。
一、核心设计思路
1. 系统架构
2. 核心特性
- 多维度分析：价格对比、销量排行、库存状态、价格波动趋势；
- 自动化：定时批量采集数据，自动生成分析报告；
- 可视化：用
matplotlib/plotly生成直观的图表（柱状图 / 折线图）； - 易扩展：支持新增分析维度（如好评率、店铺评分）。
3. 技术栈
- 数据采集：
requests（京东 API 调用）、threading（批量采集提速）； - 数据存储：
pandas（数据处理）、csv（轻量存储）、pymysql（可选数据库存储）； - 可视化：
matplotlib（基础图表）、plotly（交互式图表，可选）； - 定时任务：
schedule（自动更新数据）。
二、前置准备
1. 安装依赖库
bash
运行
# 核心依赖：API请求 + 数据处理 + 可视化 + 定时任务 pip install requests pandas matplotlib schedule pymysql python-dotenv
2. 基础配置
- 京东 API 密钥：在聚合数据官网领取「京东商品详情 API」的
AppKey（免费版足够测试，生产可升级）； - 竞品配置：整理需要分析的竞品列表（按品类分组，如「保温杯」「充电宝」）。
三、完整系统代码
1. 配置文件（.env）
ini
# .env 文件（敏感配置分离） JD_API_KEY=你的聚合数据AppKey # 可选：MySQL配置（不用数据库则留空） DB_HOST=localhost DB_USER=root DB_PASSWORD=你的数据库密码 DB_NAME=jd_competitor_analysis
2. 核心代码（jd_competitor_analysis.py）
python
运行
import requests import pandas as pd import matplotlib.pyplot as plt import schedule import time import os import json from datetime import datetime from dotenv import load_dotenv import threading import pymysql from pymysql.err import OperationalError # 加载配置 load_dotenv() plt.rcParams["font.sans-serif"] = ["SimHei"] # 解决中文显示问题 plt.rcParams["axes.unicode_minus"] = False # ===================== 全局配置 ===================== # API配置 JD_API_KEY = os.getenv("JD_API_KEY") JD_API_URL = "https://v.juhe.cn/jd/item/detail" # 竞品配置（核心：按品类分组，配置商品ID、名称、品牌） COMPETITOR_CONFIG = { "保温杯": [ {"sku_id": "100060195820", "name": "京东京造轻量保温杯", "brand": "京东京造"}, {"sku_id": "100080907904", "name": "膳魔师不锈钢保温杯", "brand": "膳魔师"}, {"sku_id": "100095380451", "name": "象印便携保温杯", "brand": "象印"} ], "充电宝": [ {"sku_id": "100012345678", "name": "小米充电宝20000mAh", "brand": "小米"}, {"sku_id": "100012345679", "name": "罗马仕充电宝10000mAh", "brand": "罗马仕"}, {"sku_id": "100012345680", "name": "品胜充电宝15000mAh", "brand": "品胜"} ] } # 数据存储路径 DATA_DIR = "jd_competitor_data" CSV_FILE = os.path.join(DATA_DIR, "competitor_data.csv") REPORT_DIR = "jd_competitor_reports" # 定时更新频率（单位：小时） UPDATE_INTERVAL = 1 # ==================================================== # 初始化目录 os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(REPORT_DIR, exist_ok=True) class JDCompetitorAnalyzer: def __init__(self): self.api_key = JD_API_KEY self.competitor_config = COMPETITOR_CONFIG self.data = pd.DataFrame() # 存储所有竞品数据 def get_item_detail(self, sku_id): """调用京东API获取单个商品详情""" params = { "key": self.api_key, "sku_id": sku_id } headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" } try: response = requests.get(JD_API_URL, params=params, headers=headers, timeout=10) if response.status_code == 200: result = response.json() if result["error_code"] == 0: item = result["result"] # 提取核心字段（标准化） return { "品类": "", # 后续填充 "商品ID": sku_id, "商品名称": item.get("title", ""), "品牌": "", # 后续填充 "价格": float(item.get("price", 0.0)), "销量": item.get("sales", "0"), "库存状态": item.get("stock_state", "未知"), "店铺名称": item.get("shop_name", ""), "采集时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } else: print(f"API调用失败（{sku_id}）：{result['reason']}") else: print(f"HTTP请求失败（{sku_id}）：状态码{response.status_code}") except Exception as e: print(f"获取商品数据异常（{sku_id}）：{str(e)}") return None def batch_collect_data(self): """批量采集所有竞品数据""" print(f"\n[{datetime.now()}] 开始批量采集竞品数据...") all_data = [] # 多线程采集（提升效率） lock = threading.Lock() threads = [] def collect_category(category, items): """采集单个品类的商品数据""" category_data = [] for item in items: detail = self.get_item_detail(item["sku_id"]) if detail: detail["品类"] = category detail["品牌"] = item["brand"] category_data.append(detail) print(f"✅ 采集完成：{category} - {item['name']}") with lock: all_data.extend(category_data) # 启动线程 for category, items in self.competitor_config.items(): t = threading.Thread(target=collect_category, args=(category, items)) threads.append(t) t.start() # 等待所有线程完成 for t in threads: t.join() # 转换为DataFrame并存储 if all_data: self.data = pd.DataFrame(all_data) # 追加写入CSV（保留历史数据） if os.path.exists(CSV_FILE): self.data.to_csv(CSV_FILE, mode="a", header=False, index=False, encoding="utf-8-sig") else: self.data.to_csv(CSV_FILE, index=False, encoding="utf-8-sig") print(f"✅ 批量采集完成，共采集{len(all_data)}条数据，已保存到{CSV_FILE}") else: print("❌ 未采集到任何数据") def load_historical_data(self): """加载历史采集数据""" if os.path.exists(CSV_FILE): self.data = pd.read_csv(CSV_FILE, encoding="utf-8-sig") print(f"✅ 加载历史数据完成，共{len(self.data)}条") else: print("⚠️ 无历史数据，先执行批量采集") def analyze_price(self, category=None): """价格分析：对比同品类商品价格，生成柱状图""" self.load_historical_data() if self.data.empty: return # 筛选品类（默认分析所有） analyze_data = self.data if category is None else self.data[self.data["品类"] == category] if analyze_data.empty: print(f"❌ 无{category}品类的分析数据") return # 取最新采集的价格（按采集时间排序） analyze_data["采集时间"] = pd.to_datetime(analyze_data["采集时间"]) latest_data = analyze_data.sort_values("采集时间").groupby(["品类", "商品ID"]).last().reset_index() # 生成价格对比图 plt.figure(figsize=(12, 6)) for category in latest_data["品类"].unique(): cat_data = latest_data[latest_data["品类"] == category] plt.bar( [f"{row['品牌']}\n{row['商品名称'][:8]}..." for _, row in cat_data.iterrows()], cat_data["价格"], label=category ) plt.title("京东竞品价格对比分析", fontsize=14) plt.xlabel("商品（品牌+名称）", fontsize=12) plt.ylabel("价格（元）", fontsize=12) plt.xticks(rotation=45, ha="right") plt.legend() plt.tight_layout() # 保存图表 price_report_path = os.path.join(REPORT_DIR, f"价格对比分析_{datetime.now().strftime('%Y%m%d%H%M')}.png") plt.savefig(price_report_path, dpi=300) print(f"✅ 价格分析完成，图表已保存到{price_report_path}") def analyze_sales(self, category=None): """销量分析：同品类销量排行，生成横向柱状图""" self.load_historical_data() if self.data.empty: return analyze_data = self.data if category is None else self.data[self.data["品类"] == category] if analyze_data.empty: print(f"❌ 无{category}品类的分析数据") return # 处理销量数据（统一格式：提取数字） def parse_sales(sales_str): if pd.isna(sales_str) or sales_str == "暂无数据": return 0 sales_str = str(sales_str).replace("+", "").replace("万", "0000") try: return int(float(sales_str.replace(" ", ""))) except: return 0 analyze_data["销量数值"] = analyze_data["销量"].apply(parse_sales) latest_data = analyze_data.sort_values("采集时间").groupby(["品类", "商品ID"]).last().reset_index() # 生成销量排行图 plt.figure(figsize=(12, 6)) for category in latest_data["品类"].unique(): cat_data = latest_data[latest_data["品类"] == category].sort_values("销量数值", ascending=True) plt.barh( [f"{row['品牌']}\n{row['商品名称'][:8]}..." for _, row in cat_data.iterrows()], cat_data["销量数值"], label=category ) plt.title("京东竞品销量排行分析", fontsize=14) plt.xlabel("销量（数值）", fontsize=12) plt.ylabel("商品（品牌+名称）", fontsize=12) plt.legend() plt.tight_layout() # 保存图表 sales_report_path = os.path.join(REPORT_DIR, f"销量排行分析_{datetime.now().strftime('%Y%m%d%H%M')}.png") plt.savefig(sales_report_path, dpi=300) print(f"✅ 销量分析完成，图表已保存到{sales_report_path}") def generate_summary_report(self): """生成竞品分析汇总报告（文本+图表）""" print(f"\n[{datetime.now()}] 开始生成竞品分析汇总报告...") self.load_historical_data() if self.data.empty: return # 生成价格和销量分析图表 self.analyze_price() self.analyze_sales() # 生成文本汇总报告 summary = [] summary.append(f"# 京东竞品分析汇总报告（{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}）") summary.append(f"## 数据概况") summary.append(f"- 采集品类数：{len(self.data['品类'].unique())}") summary.append(f"- 商品总数：{len(self.data['商品ID'].unique())}") summary.append(f"- 历史数据总量：{len(self.data)}条") # 按品类汇总 summary.append("\n## 各品类分析") for category in self.data["品类"].unique(): cat_data = self.data[self.data["品类"] == category] latest_cat = cat_data.sort_values("采集时间").groupby("商品ID").last().reset_index() summary.append(f"\n### {category}") # 价格最低商品 min_price_item = latest_cat.loc[latest_cat["价格"].idxmin()] summary.append(f"- 价格最低：{min_price_item['品牌']} {min_price_item['商品名称'][:20]}...（{min_price_item['价格']}元）") # 销量最高商品 latest_cat["销量数值"] = latest_cat["销量"].apply(lambda x: parse_sales(x) if pd.notna(x) else 0) max_sales_item = latest_cat.loc[latest_cat["销量数值"].idxmax()] summary.append(f"- 销量最高：{max_sales_item['品牌']} {max_sales_item['商品名称'][:20]}...（{max_sales_item['销量']}）") # 库存状态统计 stock_stats = latest_cat["库存状态"].value_counts().to_dict() summary.append(f"- 库存状态：{stock_stats}") # 保存文本报告 report_path = os.path.join(REPORT_DIR, f"竞品分析汇总报告_{datetime.now().strftime('%Y%m%d%H%M')}.md") with open(report_path, "w", encoding="utf-8") as f: f.write("\n".join(summary)) print(f"✅ 汇总报告生成完成，已保存到{report_path}") def run_scheduled_task(self): """启动定时任务：自动采集+分析""" print(f"\n[{datetime.now()}] 启动竞品分析定时任务，每{UPDATE_INTERVAL}小时更新一次...") # 立即执行一次 self.batch_collect_data() self.generate_summary_report() # 配置定时任务 schedule.every(UPDATE_INTERVAL).hours.do(self.batch_collect_data) schedule.every(UPDATE_INTERVAL).hours.do(self.generate_summary_report) # 持续运行 while True: schedule.run_pending() time.sleep(60) # 辅助函数：解析销量数值 def parse_sales(sales_str): if pd.isna(sales_str) or sales_str == "暂无数据": return 0 sales_str = str(sales_str).replace("+", "").replace("万", "0000") try: return int(float(sales_str.replace(" ", ""))) except: return 0 # 主函数 if __name__ == "__main__": analyzer = JDCompetitorAnalyzer() # 可选操作： # 1. 单次批量采集+生成报告 # analyzer.batch_collect_data() # analyzer.generate_summary_report() # 2. 启动定时任务（自动采集+分析） analyzer.run_scheduled_task()
四、代码核心说明
1. 关键功能模块
表格
| 方法 | 核心作用 | 亮点 |
get_item_detail |
单商品数据采集 | 异常捕获 + 字段标准化，保证数据一致性 |
batch_collect_data |
批量采集 | 多线程提速，按品类分组采集，追加保存历史数据 |
analyze_price |
价格对比分析 | 生成柱状图，支持单品类 / 全品类分析 |
analyze_sales |
销量排行分析 | 处理非标准销量数据（如「10 万 +」），生成横向柱状图 |
generate_summary_report |
汇总报告生成 | 文本报告（MD 格式）+ 可视化图表，自动汇总关键指标 |
run_scheduled_task |
定时任务 | 自动采集 + 分析，支持自定义更新频率 |
2. 核心配置修改
COMPETITOR_CONFIG：替换为你的竞品列表，按「品类→商品」分组，填写sku_id（京东商品 ID）、name（商品名称）、brand（品牌）；UPDATE_INTERVAL：修改定时更新频率（建议≥1 小时，避免 API 调用超限）；JD_API_KEY：在.env文件中填写聚合数据的 AppKey。
3. 避坑要点
- API 调用限制：聚合数据免费版有调用次数限制，批量采集时控制线程数和频率；
- 销量数据处理：京东 API 返回的销量可能是「10 万 +」「暂无数据」等非标准格式，代码中已做解析；
- 中文显示：设置
plt.rcParams解决 matplotlib 中文乱码问题； - 数据存储：CSV 文件用
utf-8-sig编码，避免 Excel 打开乱码。
五、运行与使用
1. 单次运行（采集 + 生成报告）
注释掉定时任务，执行单次采集和报告生成：
python
运行
if __name__ == "__main__": analyzer = JDCompetitorAnalyzer() analyzer.batch_collect_data() analyzer.generate_summary_report()
运行命令：
bash
运行
python jd_competitor_analysis.py
2. 定时运行（自动更新）
启用定时任务，程序会持续运行，按设定频率自动采集数据并生成报告：
python
运行
if __name__ == "__main__": analyzer = JDCompetitorAnalyzer() analyzer.run_scheduled_task()
3. 输出结果示例
- 数据文件：
jd_competitor_data/competitor_data.csv（保留所有历史采集数据）； - 可视化图表：
jd_competitor_reports/价格对比分析_202602101530.png、销量排行分析_202602101530.png； - 汇总报告：
jd_competitor_reports/竞品分析汇总报告_202602101530.md（Markdown 格式，可直接打开查看）。
六、进阶优化（可选）
- 数据库存储：替换 CSV 为 MySQL，支持更高效的历史数据查询和分析；
- 价格趋势分析：基于历史数据生成价格波动折线图，分析竞品调价规律；
- 多维度扩展：新增「好评率」「店铺评分」「配送方式」等分析维度；
- 告警功能：当竞品价格低于设定阈值时，触发邮件 / 钉钉告警；
- 交互式可视化：用
plotly替换matplotlib，生成可交互的网页版图表； - 数据清洗优化：处理重复数据、缺失值，提升分析准确性。
总结
- 核心逻辑：通过多线程批量采集京东竞品数据，基于 pandas 做数据处理，matplotlib 生成可视化图表，最终输出结构化的分析报告；
- 关键配置：修改
COMPETITOR_CONFIG添加竞品，配置 API 密钥，调整更新频率； - 实用性：支持定时自动更新，保留历史数据，可生成价格 / 销量等多维度分析报告，满足竞品分析核心需求；
- 扩展性：代码模块化设计，可轻松新增分析维度、告警功能、数据库存储等。
这套系统完全适配中小商家 / 个人的竞品分析需求，代码结构清晰，无需复杂的开发经验即可上手，且能根据实际需求灵活扩展功能。
