Python构建京东价格监控系统：从爬虫到数据分析的完整实践

简介：这是一套面向电商运营人员、市场分析师及Python初学者的价格监测与分析实践项目，聚焦京东平台商品价格动态追踪与竞品对比分析。系统基于Python爬虫实现定时抓取指定商品链接的实时价格数据，并持久化存储至本地SQLite数据库；进一步通过Matplotlib/Seaborn可视化与基础统计方法（如均值、波动率、时间序列趋势拟合）挖掘价格变化规律与竞品价差特征，为促销策略制定与市场定位提供数据支撑。压缩包共6个文件（3个核心Python脚本：爬取get.py、查看look.py、主控main.py；1份README.md说明文档；1个说明txt；1份附赠docx扩展资料），总计36KB，结构精简、模块职责清晰，便于快速部署与二次开发。目前已有76人学习下载，配套代码完整、注释充分，包含反爬应对思路、数据库建表逻辑及可视化图表生成示例，是入门级电商数据采集与分析的实用参考方案。

1. 项目缘起：一个电商运营的日常痛点与自动化解法

做电商运营或者市场研究的朋友，对下面这个场景肯定不陌生：每天上班第一件事，就是打开电脑，手动刷新几十个甚至上百个竞品或者自家商品的京东页面，把价格、促销信息、库存状态一个个抄到Excel里，然后开始做对比、看趋势。日复一日，枯燥不说，还容易出错，更关键的是，你永远不知道在你睡觉的时候，竞争对手是不是偷偷调了价、上了新券。这种靠人力“盯盘”的方式，效率低、反应慢，在快节奏的电商竞争中非常被动。

我当初接手一个家电品类的运营项目时，就深陷这个泥潭。我们需要监控5个核心SKU和超过20个竞品SKU的价格，手动记录根本跟不上节奏。于是，一个念头自然就冒出来了：能不能写个程序，让它自动去帮我盯价格，把数据存下来，还能自动分析给我看？这就是“京东商品价格数据爬取与分析系统”最初的想法。它不是什么高深莫测的AI项目，而是一个用Python技术栈解决具体业务痛点的实用工具。核心目标就三个：自动抓取、持久存储、直观分析。今天，我就把这个从零搭建的过程、踩过的坑以及最终成型的系统思路，完整地分享出来。

这个系统非常适合有一定Python基础，想将技术应用于实际业务的开发者，或者是渴望用技术手段提升工作效率的电商从业者。接下来，我会从为什么选择这些技术组件开始，一步步拆解如何实现一个稳定、可扩展的京东价格监控系统。

2. 核心架构设计：为什么是这些技术栈？

在动手写代码之前，先花点时间聊聊技术选型。一个好的架构是项目成功的一半，也能避免后期很多推倒重来的麻烦。我们这个系统的核心流程可以抽象为：定时触发 -> 网络请求 -> 数据解析 -> 数据存储 -> 数据分析与呈现。围绕这个流程，我选择了以下组件，并解释一下为什么是它们。

2.1 爬虫层：Requests + BeautifulSoup vs. Selenium

爬取京东页面数据，首先面临的选择是使用轻量级的静态页面解析库（如Requests+BeautifulSoup/lxml），还是使用能模拟浏览器行为的Selenium。

• Requests + BeautifulSoup (我的选择)：京东的商品详情页，大部分关键价格信息（如商品售价、促销价、Plus会员价）是直接嵌入在HTML源码中的，通过简单的网络请求即可获取。这种方式速度快、资源消耗低，非常适合高频次、定时执行的爬虫任务。Requests库负责发送HTTP请求获取网页原始代码，BeautifulSoup则负责从复杂的HTML中精准地“捞出”我们需要的数据。

  注意：京东对爬虫有一定反制措施，直接使用Requests可能会遇到请求失败或返回错误页面。关键在于模拟一个真实浏览器的请求头（User-Agent），并管理好请求频率（如添加随机延时），避免被识别为恶意爬虫。后续会详细讲如何设置。

• Selenium：如果目标数据是通过JavaScript动态加载的（比如某些需要滚动才能加载的评论、部分活动价），那么就需要Selenium这种能驱动真实浏览器的工具。但它速度慢、占用资源多，不适合大规模、高并发的定时抓取。对于核心价格数据，京东的页面渲染策略目前仍以静态为主，因此优先考虑轻量级方案。

结论：对于价格监控这个核心需求，Requests+BeautifulSoup的组合是效率最高的选择。我们将用它作为数据抓取的基石。

2.2 数据存储层：SQLite vs. MySQL

抓取到的数据需要持久化。这里有两个主流选择：轻量级文件数据库SQLite和传统关系型数据库MySQL。

• SQLite：数据库就是一个文件，无需安装配置数据库服务器，Python原生支持。对于单机运行、数据量不是特别巨大（日增数万条以内）的场景，简单、便携、零运维是它的最大优点。我们的监控系统通常部署在一台固定的电脑或服务器上，SQLite完全够用。
• MySQL：更适合多机访问、高并发写入、数据量极大（日增百万级）或需要复杂数据库集群管理的场景。它需要独立的数据库服务，配置相对复杂。

结论：考虑到本系统的典型应用场景是个人或小团队使用，追求部署简便，我选择了SQLite作为存储方案。它让我们可以专注于业务逻辑，而不是数据库运维。

2.3 任务调度层：APScheduler

我们需要系统能定时（比如每30分钟、每小时）自动执行一次爬取任务。虽然可以用操作系统的定时任务（如Linux的cron或Windows的任务计划程序）来调用Python脚本，但这不利于脚本内部管理复杂的任务逻辑和错误处理。

APScheduler是一个强大的Python库，它允许你在Python程序内部定义和调度任务。你可以轻松地设置“每隔X分钟执行一次函数A”，并且能很好地处理任务执行时的异常，还能实现任务的暂停、恢复等管理功能。它让我们的系统从一个被动执行的脚本，变成了一个可以自主管理生命周期的常驻服务。

2.4 数据分析与可视化层：Pandas + Matplotlib/Plotly

数据存下来不是目的，分析出价值才是。这里我推荐Pandas+Matplotlib的组合。

• Pandas：是数据分析的“瑞士军刀”。它可以将数据库里读取的数据轻松转换为DataFrame结构，进行清洗、过滤、分组、聚合（比如计算日均价、最高最低价）、时间序列分析等操作，无比顺畅。
• Matplotlib：是Python绘图的基础库，功能强大且灵活。虽然其默认样式比较学术化，但通过简单配置就能画出美观的趋势图、对比柱状图。对于更交互式的图表，可以考虑Plotly，但Matplotlib的稳定性和可控性在生成报告图片时更有优势。

最终技术栈确定：Requests（抓取） +BeautifulSoup（解析） +SQLite（存储） +APScheduler（调度） +Pandas&Matplotlib（分析可视化）。这是一个在功能、复杂度、学习成本之间取得很好平衡的方案。

3. 实战第一步：构建稳健的京东商品数据爬虫

有了架构蓝图，我们开始动手实现最核心也最易出问题的部分——爬虫。这一节，我会详细到每一个HTTP请求头和每一个CSS选择器。

3.1 环境搭建与依赖安装

首先，确保你的Python环境（建议3.7以上）已经准备好。然后，通过pip安装我们所需的库。我强烈建议使用虚拟环境（venv）来管理项目依赖，避免污染全局环境。

# 创建并激活虚拟环境 (以Linux/macOS为例) python3 -m venv jd_price_env source jd_price_env/bin/activate # 安装核心依赖 pip install requests beautifulsoup4 apscheduler pandas matplotlib # 可选：如果需要更强大的HTML解析，可以安装lxml，它比Python内置的html.parser更快更容错 pip install lxml

3.2 解析京东价格页面的核心技巧

京东的商品页面结构会随时间变化，但核心价格信息的HTML元素相对稳定。我们的任务是找到这些元素的“坐标”（CSS选择器或XPath）。

1. 手动定位元素：打开Chrome浏览器，访问一个京东商品页（例如某个手机），右键点击商品价格，选择“检查”。开发者工具会高亮显示对应的HTML代码。你需要观察并找到包含价格的标签。通常，价格会在类似<span class="price J-p-100000000000">这样的标签里。注意：class名中的数字（如100000000000）是商品SKU ID，是动态的，不能直接用！我们需要寻找更稳定的父级容器。
2. 寻找稳定路径：向上查看父级<div>，经常会发现一个class为summary-price J-summary-price的容器。这个容器的结构相对稳定。我们可以先定位到这个容器，再在其内部寻找价格标签。
3. 编写解析函数：下面是一个示例函数，它尝试从多个可能的路径提取价格，提高鲁棒性。

import requests from bs4 import BeautifulSoup import re import time import random def fetch_jd_price(product_url, sku_id=None): """ 抓取京东商品页面价格 :param product_url: 商品链接 :param sku_id: 商品SKU ID，可从url中解析，用于更精确的CSS选择器 :return: 返回一个字典，包含各种价格信息，抓取失败返回None """ headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Referer': 'https://www.jd.com/' } try: # 添加随机延时，模拟人类操作，避免请求过快 time.sleep(random.uniform(1, 3)) response = requests.get(product_url, headers=headers, timeout=10) response.raise_for_status() # 检查HTTP请求是否成功 response.encoding = 'utf-8' soup = BeautifulSoup(response.text, 'lxml') # 使用lxml解析器，需要先安装 price_info = {} # 策略1：尝试从常见的价格容器中提取 price_wrap = soup.find('div', class_='summary-price') if price_wrap: # 查找商品售价（通常是一个大的数字） price_tag = price_wrap.find('span', class_=re.compile('p-price')) if price_tag: price_info['price'] = price_tag.get_text(strip=True) # 查找促销价（如秒杀价） prom_tag = price_wrap.find('span', class_=re.compile('price J-p-')) if prom_tag: price_info['promotion_price'] = prom_tag.get_text(strip=True) # 策略2：如果策略1没找到，尝试更通用的查找（可能页面结构已变） if not price_info.get('price'): # 尝试查找所有包含“¥”符号的span，并取第一个（风险较高，作为备选） all_price_spans = soup.find_all('span', string=re.compile('¥')) if all_price_spans: price_info['price'] = all_price_spans[0].get_text(strip=True) # 获取商品标题，用于记录和验证 title_tag = soup.find('div', class_='sku-name') if title_tag: price_info['product_name'] = title_tag.get_text(strip=True) else: price_info['product_name'] = '未知商品' # 获取当前时间戳 price_info['fetch_time'] = int(time.time()) price_info['url'] = product_url return price_info if price_info.get('price') else None except requests.exceptions.RequestException as e: print(f"网络请求失败: {e}, URL: {product_url}") return None except Exception as e: print(f"解析页面时发生未知错误: {e}, URL: {product_url}") return None

关键点解析：

• 请求头（Headers）：User-Agent是必须的，否则京东服务器可能返回非人类访问的页面。Referer设置为京东首页，让请求看起来更自然。
• 延时（Sleep）：time.sleep(random.uniform(1, 3))是礼貌爬虫的必备。随机的1-3秒延时能极大降低被封IP的风险。
• 错误处理：使用try...except包裹核心代码，捕获网络超时、连接错误、解析异常等，确保单个商品抓取失败不会导致整个程序崩溃。
• 多重解析策略：页面结构可能微调。代码中提供了两种查找价格的策略，先尝试精准定位（策略1），失败后再用更宽泛的查找（策略2），提高了代码的适应性。
• 正则表达式辅助：使用re.compile('p-price')来匹配类名中包含p-price的标签，这比精确匹配整个类名更灵活，因为京东的类名后面经常跟着动态哈希值。

3.3 应对反爬机制：IP代理与请求策略

如果监控的商品很多，频率较高，即使加了延时，单个IP也可能被限制。这时需要考虑代理IP池。你可以使用一些付费或免费的代理IP服务，然后在requests请求中通过proxies参数设置。

proxies = { 'http': 'http://your-proxy-ip:port', 'https': 'https://your-proxy-ip:port', } response = requests.get(url, headers=headers, proxies=proxies, timeout=10)

对于免费代理，稳定性和速度是挑战，需要自己写代码测试代理是否有效。对于严肃的商业项目，建议使用可靠的付费代理服务。对于个人或小规模监控，控制好请求频率（如每小时一次，每次监控少于50个商品），使用真实User-Agent和随机延时，通常可以稳定运行很长时间。

4. 数据持久化：设计数据库与实现存储逻辑

爬取到的数据需要有条理地存起来。我们使用SQLite，因为它无需安装服务器。

4.1 数据库表设计

我们需要一张核心表来存储每一次抓取的价格快照。表结构设计如下：

• 表名：product_price_history
• 字段：
  • id: INTEGER, 主键，自增长。
  • product_url: TEXT, 商品链接，用于标识商品。
  • product_name: TEXT, 商品名称（每次抓取都存，因为商品名可能变化）。
  • sku_id: TEXT, 商品SKU ID（从URL中解析，是更稳定的唯一标识）。
  • price: REAL, 商品价格（浮点数）。
  • promotion_price: REAL, 促销价格（可为空）。
  • fetch_time: INTEGER, 抓取时间戳（Unix时间戳）。
  • created_at: TEXT, 记录创建时间（格式化的时间，如 ‘2023-10-27 14:30:00‘）。

为什么需要sku_id和product_url两个标识？因为URL可能会因为营销活动带上不同的参数，但其核心的SKU ID是不变的。从URL中提取SKU ID能让我们更准确地追踪同一个商品。

4.2 从URL中提取SKU ID

京东商品URL的SKU ID通常出现在链接中，例如https://item.jd.com/100000000000.html，那么SKU ID就是100000000000。我们需要一个函数来提取它。

import re def extract_sku_id_from_url(url): """ 从京东商品URL中提取SKU ID 支持 item.jd.com/100000000000.html 和 item.m.jd.com/product/100000000000.html 等格式 """ patterns = [ r'item\.jd\.com/(\d+)\.html', r'item\.m\.jd\.com/product/(\d+)\.html', r'/(\d+)\.html' ] for pattern in patterns: match = re.search(pattern, url) if match: return match.group(1) return None

4.3 实现数据存储类

我们将数据库操作封装成一个类，这样代码更清晰，也便于复用。

import sqlite3 from datetime import datetime class PriceDB: def __init__(self, db_path='jd_price_data.db'): self.db_path = db_path self._init_db() def _init_db(self): """初始化数据库，创建表""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS product_price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_url TEXT NOT NULL, product_name TEXT, sku_id TEXT, price REAL, promotion_price REAL, fetch_time INTEGER, created_at TEXT DEFAULT (datetime('now', 'localtime')) ) ''') # 创建索引以提高查询速度，特别是按商品和时间的查询 cursor.execute('CREATE INDEX IF NOT EXISTS idx_sku_time ON product_price_history (sku_id, fetch_time)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_fetch_time ON product_price_history (fetch_time)') conn.commit() conn.close() def insert_price_record(self, price_info_dict): """插入一条价格记录""" # 从URL提取SKU ID sku_id = extract_sku_id_from_url(price_info_dict.get('url', '')) price_info_dict['sku_id'] = sku_id conn = sqlite3.connect(self.db_path) cursor = conn.cursor() try: cursor.execute(''' INSERT INTO product_price_history (product_url, product_name, sku_id, price, promotion_price, fetch_time) VALUES (?, ?, ?, ?, ?, ?) ''', ( price_info_dict.get('url'), price_info_dict.get('product_name'), sku_id, self._parse_price(price_info_dict.get('price')), # 价格需要清洗转换 self._parse_price(price_info_dict.get('promotion_price')), price_info_dict.get('fetch_time') )) conn.commit() record_id = cursor.lastrowid print(f"记录插入成功，ID: {record_id}, 商品: {price_info_dict.get('product_name')}") return record_id except sqlite3.Error as e: print(f"插入数据库失败: {e}") conn.rollback() return None finally: conn.close() def _parse_price(self, price_str): """将价格字符串（如'¥2999.00'）转换为浮点数""" if not price_str: return None # 移除货币符号、逗号等非数字字符（除了小数点） try: # 匹配数字和小数点 import re num_str = re.search(r'[\d\.]+', price_str.replace(',', '')) if num_str: return float(num_str.group()) else: return None except ValueError: return None # 使用示例 if __name__ == '__main__': db = PriceDB() sample_data = { 'url': 'https://item.jd.com/100000000000.html', 'product_name': '示例商品', 'price': '¥2999.00', 'promotion_price': '¥2799.00', 'fetch_time': int(time.time()) } db.insert_price_record(sample_data)

实操心得：

1. 索引的重要性：在sku_id和fetch_time上创建复合索引，能极大提升“查询某个商品历史价格”这类操作的速度。当数据量达到几十万条时，有无索引的查询性能差异是数量级的。
2. 数据清洗：_parse_price函数至关重要。网页上抓取的价格是字符串，可能包含“¥”、“$”、“,”等字符，必须清洗成浮点数才能进行数值计算和比较。
3. 错误处理与事务：数据库操作要用try...except包裹，并在出错时rollback。使用finally确保数据库连接被关闭，避免资源泄露。

5. 让系统自动运行：集成APScheduler实现定时任务

现在，我们有了抓取单品的函数和存储数据的类。接下来，我们需要一个“大脑”来定时协调这些工作。这就是APScheduler的用武之地。

5.1 创建核心任务函数

这个函数将串联爬取和存储的流程，并处理一个商品列表。

from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger import logging # 配置日志，方便查看任务运行情况 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.FileHandler('price_monitor.log'), logging.StreamHandler()]) logger = logging.getLogger(__name__) class PriceMonitor: def __init__(self, db_path='jd_price_data.db'): self.db = PriceDB(db_path) self.product_list = [] # 初始化为空，可以从文件或数据库加载 def load_product_list(self, file_path='products.txt'): """从文本文件加载监控商品列表，每行一个URL""" try: with open(file_path, 'r', encoding='utf-8') as f: self.product_list = [line.strip() for line in f if line.strip()] logger.info(f"成功加载 {len(self.product_list)} 个商品链接。") except FileNotFoundError: logger.warning(f"商品列表文件 {file_path} 不存在，将使用空列表。") self.product_list = [] def monitor_single_product(self, url): """监控单个商品：抓取 -> 存储""" logger.info(f"开始抓取商品: {url}") price_info = fetch_jd_price(url) if price_info: record_id = self.db.insert_price_record(price_info) if record_id: logger.info(f"商品抓取并存储成功: {price_info.get('product_name')}，价格: {price_info.get('price')}") else: logger.error(f"商品抓取成功但存储失败: {url}") else: logger.error(f"商品抓取失败: {url}") # 抓取一个商品后，建议短暂随机休眠，分散请求 time.sleep(random.uniform(0.5, 2)) def monitor_all_products(self): """监控所有商品列表中的商品""" logger.info("====== 开始本轮价格监控任务 ======") if not self.product_list: logger.warning("商品列表为空，请检查 products.txt 文件。") return for url in self.product_list: self.monitor_single_product(url) logger.info("====== 本轮价格监控任务结束 ======\n") # 主程序入口 if __name__ == '__main__': monitor = PriceMonitor() monitor.load_product_list() # 加载商品列表 # 创建调度器 scheduler = BlockingScheduler() # 添加定时任务：每30分钟执行一次 monitor_all_products 函数 # 使用IntervalTrigger，间隔30分钟 trigger = IntervalTrigger(minutes=30) scheduler.add_job(monitor.monitor_all_products, trigger, id='jd_price_job') logger.info("价格监控调度器已启动，按 Ctrl+C 退出。") try: scheduler.start() except (KeyboardInterrupt, SystemExit): logger.info("收到停止信号，调度器正在关闭...") scheduler.shutdown() logger.info("调度器已关闭。")

5.2 调度器配置详解与高级用法

• BlockingScheduler：这是一种阻塞式调度器。当scheduler.start()被调用后，程序会停在这里，直到收到中断信号（如Ctrl+C）。这适合将监控脚本作为独立的守护进程运行。
• IntervalTrigger：最简单的间隔触发器。除了minutes，还可以设置seconds、hours、days。例如IntervalTrigger(hours=2)表示每两小时执行一次。
• 任务持久化（可选）：默认情况下，APScheduler的任务信息存在内存里。如果程序重启，所有的定时任务设置都会丢失。对于需要7x24小时运行的服务，可以配置作业存储（Job Store）到数据库（如SQLAlchemy支持的各种数据库），这样即使程序重启，任务也能恢复。但对于我们这个单机监控脚本，内存存储通常够用。
• 并发控制：默认情况下，如果上一个任务还没执行完，下一个任务时间又到了，APScheduler会等待当前任务完成。你可以通过max_instances参数控制同一个任务的最大并发实例数。

踩坑记录：

1. 时区问题：APScheduler默认使用UTC时间。如果你在中国，并且希望任务在本地时间运行，需要在创建调度器时指定时区：scheduler = BlockingScheduler(timezone='Asia/Shanghai')。
2. 日志管理：一定要配置日志，并输出到文件（如FileHandler）。当程序在后台运行时，日志文件是你排查问题的唯一依据。日志级别设为INFO可以记录任务开始、结束、成功、失败等信息。
3. 优雅退出：使用try...except捕获KeyboardInterrupt和SystemExit异常，并在其中调用scheduler.shutdown()，可以让调度器在退出前完成当前正在执行的任务，避免数据丢失或状态不一致。

6. 从数据到洞察：使用Pandas与Matplotlib进行可视化分析

数据已经源源不断地存进了数据库，现在是让数据说话的时候了。我们将使用Pandas进行数据分析和清洗，然后用Matplotlib生成图表。

6.1 连接数据库并加载数据

首先，我们需要从SQLite数据库中读取历史价格数据。

import pandas as pd import matplotlib.pyplot as plt from datetime import datetime, timedelta def load_price_data_from_db(db_path='jd_price_data.db', sku_id=None, days_back=30): """ 从数据库加载指定SKU或所有商品在最近N天的价格数据 :param db_path: 数据库路径 :param sku_id: 商品SKU ID，为None则加载所有商品 :param days_back: 回溯多少天的数据 :return: pandas DataFrame """ conn = sqlite3.connect(db_path) # 计算时间点 cutoff_time = int((datetime.now() - timedelta(days=days_back)).timestamp()) if sku_id: query = ''' SELECT product_name, sku_id, price, promotion_price, fetch_time, created_at FROM product_price_history WHERE sku_id = ? AND fetch_time >= ? ORDER BY fetch_time ASC ''' params = (sku_id, cutoff_time) else: query = ''' SELECT product_name, sku_id, price, promotion_price, fetch_time, created_at FROM product_price_history WHERE fetch_time >= ? ORDER BY sku_id, fetch_time ASC ''' params = (cutoff_time,) df = pd.read_sql_query(query, conn, params=params) conn.close() if df.empty: print(f"未找到SKU为 {sku_id} 的商品在最近 {days_back} 天的数据。") return df # 将时间戳转换为datetime类型，便于Pandas进行时间序列分析 df['fetch_time_dt'] = pd.to_datetime(df['fetch_time'], unit='s') # 设置时间为索引（可选，对于时间序列分析很方便） df.set_index('fetch_time_dt', inplace=True) return df # 示例：加载某个商品最近7天的数据 df_single = load_price_data_from_db(sku_id='100000000000', days_back=7) print(df_single.head())

6.2 单商品价格趋势分析

这是最基础的分析：一个商品的价格随时间如何波动？

def plot_single_product_trend(df, sku_id): """ 绘制单个商品的价格趋势图 """ if df.empty: print("数据为空，无法绘图。") return plt.figure(figsize=(14, 7)) product_name = df['product_name'].iloc[0] if not df['product_name'].empty else f'SKU: {sku_id}' # 绘制商品售价 plt.plot(df.index, df['price'], marker='o', linewidth=2, label='商品售价', color='blue') # 如果有促销价，也绘制出来 if 'promotion_price' in df.columns and not df['promotion_price'].isna().all(): # 只绘制非空的促销价数据点 promo_df = df.dropna(subset=['promotion_price']) plt.scatter(promo_df.index, promo_df['promotion_price'], marker='s', label='促销价格', color='red', zorder=5) # 可以用虚线连接促销价点，更直观 plt.plot(promo_df.index, promo_df['promotion_price'], linestyle='--', color='red', alpha=0.5) plt.title(f'{product_name} - 近期价格趋势 ({df.index.min().date()} 至 {df.index.max().date()})', fontsize=16) plt.xlabel('抓取时间', fontsize=12) plt.ylabel('价格 (元)', fontsize=12) plt.grid(True, linestyle='--', alpha=0.7) plt.legend() plt.xticks(rotation=45) # 旋转X轴标签，避免重叠 plt.tight_layout() # 自动调整子图参数，使之填充整个图像区域 # 保存图片 filename = f'price_trend_{sku_id}_{datetime.now().strftime("%Y%m%d_%H%M%S")}.png' plt.savefig(filename, dpi=300) print(f"趋势图已保存为: {filename}") plt.show() # 使用示例 if not df_single.empty: plot_single_product_trend(df_single, '100000000000')

6.3 多商品（竞品）价格对比分析

监控的核心价值之一就是对比。我们可以将多个竞品放在同一张图上进行对比。

def plot_multi_product_comparison(db_path, sku_id_list, days_back=7): """ 绘制多个商品（竞品）的价格对比折线图 """ plt.figure(figsize=(16, 9)) all_data = {} for sku_id in sku_id_list: df = load_price_data_from_db(db_path, sku_id=sku_id, days_back=days_back) if not df.empty: # 以商品名作为图例标签，如果商品名太长可以截取 label = df['product_name'].iloc[0][:30] + '...' if len(df['product_name'].iloc[0]) > 30 else df['product_name'].iloc[0] # 使用促销价（如果有）或售价进行绘图，这里统一用售价 plt.plot(df.index, df['price'], marker='.', linewidth=1.5, label=label) all_data[sku_id] = df else: print(f"SKU {sku_id} 无数据，已跳过。") if not all_data: print("所有SKU均无数据，无法绘图。") return plt.title(f'竞品价格对比 ({days_back}天数据)', fontsize=18) plt.xlabel('时间', fontsize=14) plt.ylabel('价格 (元)', fontsize=14) plt.grid(True, linestyle='--', alpha=0.5) plt.legend(loc='upper left', bbox_to_anchor=(1, 1)) # 将图例放在图表外侧，避免遮挡 plt.xticks(rotation=45) plt.tight_layout() filename = f'competitor_comparison_{datetime.now().strftime("%Y%m%d_%H%M%S")}.png' plt.savefig(filename, dpi=300) print(f"竞品对比图已保存为: {filename}") plt.show() # 使用示例：对比三个竞品 competitor_skus = ['100000000001', '100000000002', '100000000003'] plot_multi_product_comparison('jd_price_data.db', competitor_skus, days_back=14)

6.4 生成统计摘要报告

除了图表，我们还需要一些关键的数字指标。

def generate_price_summary(df, sku_id): """ 生成价格数据统计摘要 """ if df.empty: return None summary = { 'SKU': sku_id, '商品名称': df['product_name'].iloc[0], '分析时段': f"{df.index.min().strftime('%Y-%m-%d %H:%M')} 至 {df.index.max().strftime('%Y-%m-%d %H:%M')}", '数据点数': len(df), '平均价格': df['price'].mean(), '价格中位数': df['price'].median(), '最高价格': df['price'].max(), '最低价格': df['price'].min(), '价格标准差': df['price'].std(), # 波动性 '当前价格': df['price'].iloc[-1] if len(df) > 0 else None, '是否有促销': '是' if ('promotion_price' in df.columns and not df['promotion_price'].isna().any()) else '否' } # 计算价格变化（最新价对比最初价） if len(df) > 1: price_change = df['price'].iloc[-1] - df['price'].iloc[0] price_change_pct = (price_change / df['price'].iloc[0]) * 100 summary['价格变化(绝对值)'] = round(price_change, 2) summary['价格变化(百分比)'] = f"{round(price_change_pct, 2)}%" return summary # 使用示例并打印报告 summary = generate_price_summary(df_single, '100000000000') if summary: print("=== 价格分析报告 ===") for key, value in summary.items(): print(f"{key}: {value}")

可视化进阶技巧：

• 子图（Subplots）：如果你需要同时观察趋势、分布（直方图）和每日均价（箱线图），可以使用plt.subplots创建多个子图在一张画布上。
• 样式美化：Matplotlib的默认样式比较朴素。你可以使用plt.style.use('seaborn-v0_8-darkgrid')来切换为更美观的seaborn样式（需要安装seaborn库），或者手动设置颜色、字体等。
• 动态仪表盘：对于更复杂的实时监控，可以考虑使用Plotly或Dash库构建交互式Web仪表盘，但这超出了本文基础系统的范围。对于日常运营报告，静态图片和文本摘要已经足够。

7. 系统优化与生产环境部署思考

一个能跑起来的脚本和一个健壮的生产系统之间，还有不少距离。以下是几个关键的优化和部署考量点。

7.1 错误处理与系统健壮性

我们的脚本可能会遇到各种意外：网络暂时中断、京东页面改版、数据库磁盘满、甚至被暂时封IP。系统必须能优雅地处理这些错误，并继续运行或至少记录下清晰的错误信息。

• 重试机制：对于网络请求失败，可以加入简单的重试逻辑。但要注意，对于“页面不存在”（404）或“访问被拒绝”（403）这类错误，重试是没用的。

def fetch_jd_price_with_retry(url, max_retries=3): for i in range(max_retries): try: result = fetch_jd_price(url) if result: # 如果成功抓取到数据，直接返回 return result else: # 如果fetch_jd_price内部返回None（如解析失败），也视为失败，进行重试 logger.warning(f"第{i+1}次抓取失败（解析无结果），准备重试... URL: {url}") time.sleep(2 ** i) # 指数退避延时 except Exception as e: logger.error(f"第{i+1}次抓取发生异常: {e}，准备重试... URL: {url}") time.sleep(2 ** i) # 指数退避延时 logger.error(f"抓取失败，已达最大重试次数 {max_retries}。 URL: {url}") return None

• 心跳与监控：可以增加一个简单的“心跳”任务，定期（比如每天一次）向一个日志文件或发送一封邮件报告系统状态（如“今日成功抓取X条记录，失败Y条”）。这能让你在不主动查看日志的情况下，也知道系统是否在正常工作。
• 日志分级与轮转：使用Python的logging模块，设置不同的级别（DEBUG, INFO, WARNING, ERROR）。对于长期运行的程序，需要配置日志轮转（RotatingFileHandler），防止日志文件无限增大占满磁盘。

7.2 配置化管理

将商品列表、数据库路径、抓取频率、代理IP等配置信息从代码中分离出来，使用配置文件（如config.ini或config.yaml）或环境变量来管理。这样在修改配置时无需改动代码，也便于在不同环境（开发、生产）中部署。

# config.yaml 示例 monitor: interval_minutes: 30 request_timeout: 10 user_agent: "Mozilla/5.0 ..." database: path: "./data/jd_price.db" products: - url: "https://item.jd.com/100000000000.html" alias: "主力商品A" - url: "https://item.jd.com/100000000001.html" alias: "竞品B"

7.3 部署为系统服务

要让这个脚本在服务器上7x24小时运行，最好的方式是将其部署为系统服务。

• Linux (Systemd)：创建一个.service文件（如jd-price-monitor.service），定义启动命令、工作目录、用户、重启策略等。然后使用systemctl enable和systemctl start来管理它。

  [Unit] Description=JD Price Monitor Service After=network.target [Service] Type=simple User=your_username WorkingDirectory=/path/to/your/script ExecStart=/path/to/your/venv/bin/python /path/to/your/script/main.py Restart=on-failure RestartSec=10s [Install] WantedBy=multi-user.target

• Windows (任务计划程序)：可以创建一个基本的任务计划，设置触发器（如“每天”、“重复任务间隔”），操作是启动Python解释器运行你的脚本。更稳定的方式可能是使用NSSM(the Non-Sucking Service Manager) 将Python脚本包装成Windows服务。

7.4 数据安全与隐私

• 数据库备份：定期备份你的SQLite数据库文件。可以写一个简单的脚本，每天将数据库文件复制到另一个位置或云存储。
• 敏感信息：切勿将包含真实商品监控列表、代理IP等信息的配置文件提交到公开的代码仓库（如GitHub）。使用.gitignore文件忽略它们。

8. 扩展思路：这个系统还能做什么？

一个基础的监控系统搭建完成后，可以根据业务需求进行丰富和扩展：

1. 库存监控：除了价格，还可以解析页面的库存状态（“有货”、“无货”、“仅剩X件”），这对于把握补货时机和竞品库存情况很有价值。
2. 促销信息提取：抓取商品页面的促销标签，如“满减”、“赠品”、“百亿补贴”，并进行结构化存储和分析，了解竞品的营销节奏。
3. 价格预警：在系统中集成邮件或钉钉/企业微信机器人通知。当监控的商品价格低于某个阈值（底价报警），或者发生剧烈波动时，自动发送警报，让你第一时间做出反应。
4. 竞品对标报告自动化：将数据分析部分固化，每周或每月自动运行一次，生成包含趋势图、对比图和统计摘要的PDF或HTML报告，并自动发送给相关同事。
5. 对接BI工具：将SQLite数据库中的数据定期同步到更专业的商业智能（BI）工具中，如Metabase、Superset，利用其强大的仪表盘和协作功能进行更深入的分析。

这个“京东商品价格数据爬取与分析系统”就像你不知疲倦的数字化助理，它接管了最枯燥的数据收集工作，让你能腾出精力，专注于基于数据的策略思考和决策。从一行行代码到最终产生业务价值，这个过程本身，就是技术赋能业务的最佳体现。

本文还有配套的精品资源，点击获取