简介这是一套面向电商运营人员、市场分析师及Python初学者的价格监测与分析实践项目聚焦京东平台商品价格动态追踪与竞品对比分析。系统基于Python爬虫实现定时抓取指定商品链接的实时价格数据并持久化存储至本地SQLite数据库进一步通过Matplotlib/Seaborn可视化与基础统计方法如均值、波动率、时间序列趋势拟合挖掘价格变化规律与竞品价差特征为促销策略制定与市场定位提供数据支撑。压缩包共6个文件3个核心Python脚本爬取get.py、查看look.py、主控main.py1份README.md说明文档1个说明txt1份附赠docx扩展资料总计36KB结构精简、模块职责清晰便于快速部署与二次开发。目前已有76人学习下载配套代码完整、注释充分包含反爬应对思路、数据库建表逻辑及可视化图表生成示例是入门级电商数据采集与分析的实用参考方案。1. 项目缘起一个电商运营的日常痛点与自动化解法做电商运营或者市场研究的朋友对下面这个场景肯定不陌生每天上班第一件事就是打开电脑手动刷新几十个甚至上百个竞品或者自家商品的京东页面把价格、促销信息、库存状态一个个抄到Excel里然后开始做对比、看趋势。日复一日枯燥不说还容易出错更关键的是你永远不知道在你睡觉的时候竞争对手是不是偷偷调了价、上了新券。这种靠人力“盯盘”的方式效率低、反应慢在快节奏的电商竞争中非常被动。我当初接手一个家电品类的运营项目时就深陷这个泥潭。我们需要监控5个核心SKU和超过20个竞品SKU的价格手动记录根本跟不上节奏。于是一个念头自然就冒出来了能不能写个程序让它自动去帮我盯价格把数据存下来还能自动分析给我看这就是“京东商品价格数据爬取与分析系统”最初的想法。它不是什么高深莫测的AI项目而是一个用Python技术栈解决具体业务痛点的实用工具。核心目标就三个自动抓取、持久存储、直观分析。今天我就把这个从零搭建的过程、踩过的坑以及最终成型的系统思路完整地分享出来。这个系统非常适合有一定Python基础想将技术应用于实际业务的开发者或者是渴望用技术手段提升工作效率的电商从业者。接下来我会从为什么选择这些技术组件开始一步步拆解如何实现一个稳定、可扩展的京东价格监控系统。2. 核心架构设计为什么是这些技术栈在动手写代码之前先花点时间聊聊技术选型。一个好的架构是项目成功的一半也能避免后期很多推倒重来的麻烦。我们这个系统的核心流程可以抽象为定时触发 - 网络请求 - 数据解析 - 数据存储 - 数据分析与呈现。围绕这个流程我选择了以下组件并解释一下为什么是它们。2.1 爬虫层Requests BeautifulSoup vs. Selenium爬取京东页面数据首先面临的选择是使用轻量级的静态页面解析库如RequestsBeautifulSoup/lxml还是使用能模拟浏览器行为的Selenium。Requests BeautifulSoup (我的选择)京东的商品详情页大部分关键价格信息如商品售价、促销价、Plus会员价是直接嵌入在HTML源码中的通过简单的网络请求即可获取。这种方式速度快、资源消耗低非常适合高频次、定时执行的爬虫任务。Requests库负责发送HTTP请求获取网页原始代码BeautifulSoup则负责从复杂的HTML中精准地“捞出”我们需要的数据。注意京东对爬虫有一定反制措施直接使用Requests可能会遇到请求失败或返回错误页面。关键在于模拟一个真实浏览器的请求头User-Agent并管理好请求频率如添加随机延时避免被识别为恶意爬虫。后续会详细讲如何设置。Selenium如果目标数据是通过JavaScript动态加载的比如某些需要滚动才能加载的评论、部分活动价那么就需要Selenium这种能驱动真实浏览器的工具。但它速度慢、占用资源多不适合大规模、高并发的定时抓取。对于核心价格数据京东的页面渲染策略目前仍以静态为主因此优先考虑轻量级方案。结论对于价格监控这个核心需求RequestsBeautifulSoup的组合是效率最高的选择。我们将用它作为数据抓取的基石。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数据存下来不是目的分析出价值才是。这里我推荐PandasMatplotlib的组合。Pandas是数据分析的“瑞士军刀”。它可以将数据库里读取的数据轻松转换为DataFrame结构进行清洗、过滤、分组、聚合比如计算日均价、最高最低价、时间序列分析等操作无比顺畅。Matplotlib是Python绘图的基础库功能强大且灵活。虽然其默认样式比较学术化但通过简单配置就能画出美观的趋势图、对比柱状图。对于更交互式的图表可以考虑Plotly但Matplotlib的稳定性和可控性在生成报告图片时更有优势。最终技术栈确定Requests抓取 BeautifulSoup解析 SQLite存储 APScheduler调度 PandasMatplotlib分析可视化。这是一个在功能、复杂度、学习成本之间取得很好平衡的方案。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 lxml3.2 解析京东价格页面的核心技巧京东的商品页面结构会随时间变化但核心价格信息的HTML元素相对稳定。我们的任务是找到这些元素的“坐标”CSS选择器或XPath。手动定位元素打开Chrome浏览器访问一个京东商品页例如某个手机右键点击商品价格选择“检查”。开发者工具会高亮显示对应的HTML代码。你需要观察并找到包含价格的标签。通常价格会在类似span classprice J-p-100000000000这样的标签里。注意class名中的数字如100000000000是商品SKU ID是动态的不能直接用我们需要寻找更稳定的父级容器。寻找稳定路径向上查看父级div经常会发现一个class为summary-price J-summary-price的容器。这个容器的结构相对稳定。我们可以先定位到这个容器再在其内部寻找价格标签。编写解析函数下面是一个示例函数它尝试从多个可能的路径提取价格提高鲁棒性。import requests from bs4 import BeautifulSoup import re import time import random def fetch_jd_price(product_url, sku_idNone): 抓取京东商品页面价格 :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;q0.9, Referer: https://www.jd.com/ } try: # 添加随机延时模拟人类操作避免请求过快 time.sleep(random.uniform(1, 3)) response requests.get(product_url, headersheaders, timeout10) 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(stripTrue) # 查找促销价如秒杀价 prom_tag price_wrap.find(span, class_re.compile(price J-p-)) if prom_tag: price_info[promotion_price] prom_tag.get_text(stripTrue) # 策略2如果策略1没找到尝试更通用的查找可能页面结构已变 if not price_info.get(price): # 尝试查找所有包含“¥”符号的span并取第一个风险较高作为备选 all_price_spans soup.find_all(span, stringre.compile(¥)) if all_price_spans: price_info[price] all_price_spans[0].get_text(stripTrue) # 获取商品标题用于记录和验证 title_tag soup.find(div, class_sku-name) if title_tag: price_info[product_name] title_tag.get_text(stripTrue) 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关键点解析请求头HeadersUser-Agent是必须的否则京东服务器可能返回非人类访问的页面。Referer设置为京东首页让请求看起来更自然。延时Sleeptime.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, headersheaders, proxiesproxies, timeout10)对于免费代理稳定性和速度是挑战需要自己写代码测试代理是否有效。对于严肃的商业项目建议使用可靠的付费代理服务。对于个人或小规模监控控制好请求频率如每小时一次每次监控少于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 [ ritem\.jd\.com/(\d)\.html, ritem\.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 None4.3 实现数据存储类我们将数据库操作封装成一个类这样代码更清晰也便于复用。import sqlite3 from datetime import datetime class PriceDB: def __init__(self, db_pathjd_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)实操心得索引的重要性在sku_id和fetch_time上创建复合索引能极大提升“查询某个商品历史价格”这类操作的速度。当数据量达到几十万条时有无索引的查询性能差异是数量级的。数据清洗_parse_price函数至关重要。网页上抓取的价格是字符串可能包含“¥”、“$”、“,”等字符必须清洗成浮点数才能进行数值计算和比较。错误处理与事务数据库操作要用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(levellogging.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_pathjd_price_data.db): self.db PriceDB(db_path) self.product_list [] # 初始化为空可以从文件或数据库加载 def load_product_list(self, file_pathproducts.txt): 从文本文件加载监控商品列表每行一个URL try: with open(file_path, r, encodingutf-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(minutes30) scheduler.add_job(monitor.monitor_all_products, trigger, idjd_price_job) logger.info(价格监控调度器已启动按 CtrlC 退出。) try: scheduler.start() except (KeyboardInterrupt, SystemExit): logger.info(收到停止信号调度器正在关闭...) scheduler.shutdown() logger.info(调度器已关闭。)5.2 调度器配置详解与高级用法BlockingScheduler这是一种阻塞式调度器。当scheduler.start()被调用后程序会停在这里直到收到中断信号如CtrlC。这适合将监控脚本作为独立的守护进程运行。IntervalTrigger最简单的间隔触发器。除了minutes还可以设置seconds、hours、days。例如IntervalTrigger(hours2)表示每两小时执行一次。任务持久化可选默认情况下APScheduler的任务信息存在内存里。如果程序重启所有的定时任务设置都会丢失。对于需要7x24小时运行的服务可以配置作业存储Job Store到数据库如SQLAlchemy支持的各种数据库这样即使程序重启任务也能恢复。但对于我们这个单机监控脚本内存存储通常够用。并发控制默认情况下如果上一个任务还没执行完下一个任务时间又到了APScheduler会等待当前任务完成。你可以通过max_instances参数控制同一个任务的最大并发实例数。踩坑记录时区问题APScheduler默认使用UTC时间。如果你在中国并且希望任务在本地时间运行需要在创建调度器时指定时区scheduler BlockingScheduler(timezoneAsia/Shanghai)。日志管理一定要配置日志并输出到文件如FileHandler。当程序在后台运行时日志文件是你排查问题的唯一依据。日志级别设为INFO可以记录任务开始、结束、成功、失败等信息。优雅退出使用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_pathjd_price_data.db, sku_idNone, days_back30): 从数据库加载指定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(daysdays_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, paramsparams) 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], units) # 设置时间为索引可选对于时间序列分析很方便 df.set_index(fetch_time_dt, inplaceTrue) return df # 示例加载某个商品最近7天的数据 df_single load_price_data_from_db(sku_id100000000000, days_back7) 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 fSKU: {sku_id} # 绘制商品售价 plt.plot(df.index, df[price], markero, linewidth2, label商品售价, colorblue) # 如果有促销价也绘制出来 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], markers, label促销价格, colorred, zorder5) # 可以用虚线连接促销价点更直观 plt.plot(promo_df.index, promo_df[promotion_price], linestyle--, colorred, alpha0.5) plt.title(f{product_name} - 近期价格趋势 ({df.index.min().date()} 至 {df.index.max().date()}), fontsize16) plt.xlabel(抓取时间, fontsize12) plt.ylabel(价格 (元), fontsize12) plt.grid(True, linestyle--, alpha0.7) plt.legend() plt.xticks(rotation45) # 旋转X轴标签避免重叠 plt.tight_layout() # 自动调整子图参数使之填充整个图像区域 # 保存图片 filename fprice_trend_{sku_id}_{datetime.now().strftime(%Y%m%d_%H%M%S)}.png plt.savefig(filename, dpi300) 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_back7): 绘制多个商品竞品的价格对比折线图 plt.figure(figsize(16, 9)) all_data {} for sku_id in sku_id_list: df load_price_data_from_db(db_path, sku_idsku_id, days_backdays_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., linewidth1.5, labellabel) all_data[sku_id] df else: print(fSKU {sku_id} 无数据已跳过。) if not all_data: print(所有SKU均无数据无法绘图。) return plt.title(f竞品价格对比 ({days_back}天数据), fontsize18) plt.xlabel(时间, fontsize14) plt.ylabel(价格 (元), fontsize14) plt.grid(True, linestyle--, alpha0.5) plt.legend(locupper left, bbox_to_anchor(1, 1)) # 将图例放在图表外侧避免遮挡 plt.xticks(rotation45) plt.tight_layout() filename fcompetitor_comparison_{datetime.now().strftime(%Y%m%d_%H%M%S)}.png plt.savefig(filename, dpi300) print(f竞品对比图已保存为: {filename}) plt.show() # 使用示例对比三个竞品 competitor_skus [100000000001, 100000000002, 100000000003] plot_multi_product_comparison(jd_price_data.db, competitor_skus, days_back14)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_retries3): for i in range(max_retries): try: result fetch_jd_price(url) if result: # 如果成功抓取到数据直接返回 return result else: # 如果fetch_jd_price内部返回None如解析失败也视为失败进行重试 logger.warning(f第{i1}次抓取失败解析无结果准备重试... URL: {url}) time.sleep(2 ** i) # 指数退避延时 except Exception as e: logger.error(f第{i1}次抓取发生异常: {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: 竞品B7.3 部署为系统服务要让这个脚本在服务器上7x24小时运行最好的方式是将其部署为系统服务。Linux (Systemd)创建一个.service文件如jd-price-monitor.service定义启动命令、工作目录、用户、重启策略等。然后使用systemctl enable和systemctl start来管理它。[Unit] DescriptionJD Price Monitor Service Afternetwork.target [Service] Typesimple Useryour_username WorkingDirectory/path/to/your/script ExecStart/path/to/your/venv/bin/python /path/to/your/script/main.py Restarton-failure RestartSec10s [Install] WantedBymulti-user.targetWindows (任务计划程序)可以创建一个基本的任务计划设置触发器如“每天”、“重复任务间隔”操作是启动Python解释器运行你的脚本。更稳定的方式可能是使用NSSM(the Non-Sucking Service Manager) 将Python脚本包装成Windows服务。7.4 数据安全与隐私数据库备份定期备份你的SQLite数据库文件。可以写一个简单的脚本每天将数据库文件复制到另一个位置或云存储。敏感信息切勿将包含真实商品监控列表、代理IP等信息的配置文件提交到公开的代码仓库如GitHub。使用.gitignore文件忽略它们。8. 扩展思路这个系统还能做什么一个基础的监控系统搭建完成后可以根据业务需求进行丰富和扩展库存监控除了价格还可以解析页面的库存状态“有货”、“无货”、“仅剩X件”这对于把握补货时机和竞品库存情况很有价值。促销信息提取抓取商品页面的促销标签如“满减”、“赠品”、“百亿补贴”并进行结构化存储和分析了解竞品的营销节奏。价格预警在系统中集成邮件或钉钉/企业微信机器人通知。当监控的商品价格低于某个阈值底价报警或者发生剧烈波动时自动发送警报让你第一时间做出反应。竞品对标报告自动化将数据分析部分固化每周或每月自动运行一次生成包含趋势图、对比图和统计摘要的PDF或HTML报告并自动发送给相关同事。对接BI工具将SQLite数据库中的数据定期同步到更专业的商业智能BI工具中如Metabase、Superset利用其强大的仪表盘和协作功能进行更深入的分析。这个“京东商品价格数据爬取与分析系统”就像你不知疲倦的数字化助理它接管了最枯燥的数据收集工作让你能腾出精力专注于基于数据的策略思考和决策。从一行行代码到最终产生业务价值这个过程本身就是技术赋能业务的最佳体现。本文还有配套的精品资源点击获取