简介这套京东商品价格爬取与分析系统是一份完整的Python项目代码面向电商运营人员、市场研究者和爬虫技术学习者。系统通过定时爬虫抓取京东指定商品链接的价格信息并将其存储至本地数据库再以数据可视化与统计方法分析价格波动趋势、开展竞品对比辅助定价策略与市场研判。压缩包共6个文件包含3个Python脚本分别负责信息爬取、数据展示与主流程控制、说明txt、README及配套Word文档整体体积仅36KB结构紧凑清晰便于直接阅读和二次开发。目前已有76人学习下载适合有一定Python基础并希望掌握价格监控全流程的读者可参考其数据库表设计、定时任务写法及可视化思路快速迁移到自有项目中扩展分析维度。1. 京东商品价格爬取与分析系统为什么要把盯价格这件小事做成自动化做电商运营或者市场研究的人大多经历过这种场景早上打开京东看一遍竞品价格下午再刷一次促销来了赶紧截图通知同事。一天两三个小时搭进去最后复盘时发现——昨天竞品到底几点调的价、涨了又降了几轮完全说不清楚。京东商品价格数据爬取与分析系统就是把这件事变成一套自动化的流程用 Python 爬虫定时抓取京东指定商品链接的价格信息存进本地数据库再借数据可视化与统计方法分析价格波动趋势以及与竞品的对比给电商运营和市场研究提供决策依据。它适合两类人一类是被手动比价耗掉大量时间的运营另一类是想要一个完整数据链路练手的 Python 学习者。整个系统的技术栈并不复杂requests、MySQL、pandas、ECharts 这几件常用工具就够。2. 用 requests 定时抓京东价格从商品链接到价格快照的采集层实现采集层是整条链路的入口抓回来的数据脏不脏、全不全后面所有统计分析都建立在这层基础上。抓京东价格网上很多教程一上来就是 selenium 或者 pyppeteer 模拟浏览器但监控指定商品链接的价格用 requests 就够了。网络爬虫原理不难理解构造 HTTP 请求解析响应里的 JSON 字段把结果落库剩下的交给定时调度。2.1 先拆解采集目标SKU、价格接口与反爬边界京东商品链接的标准格式是https://item.jd.com/100012043978.html中间那一串数字就是 SKU也就是 wareId是整个监控周期里所有表的主键来源。开始写爬虫之前我一般先把要监控的链接整理成一张清单skuId、竞品分组、品牌、类目。因为后面做竞品对比时光有价格没有上下文统计出来的数字没有运营意义。采集目标字段上要抓这几类当前价 p、原价 op、移动端价 m、库存状态。当前价就是页面上展示的最终成交价原价用于算折扣力度移动端价在部分品类上会明显低于 PC 端。库存状态不能省无货商品的价格经常会变成 0 或者空字符串如果不单独记录分析阶段很难解释「突然冒出来的 0 元价格」是怎么来的。为什么不直接对页面 HTML 做 xpath 解析因为商品页的价格区域是 JS 异步渲染出来的requests 拿到的静态源码里价格节点经常是空的。花半天调 xpath 表达式不如直接命中价格接口。这是京东价格爬虫最常见的认知差很多人觉得解析 HTML 才算爬虫实际上读 JSON 接口又快又稳也更容易做异常处理。常见的做法是请求京东的价格接口p.3.cn/prices/mgets按 SKU 集合批量返回价格 JSON不需要登录、不需要解析复杂页面。反爬边界方面单 IP 高频轮询会被限制。我的习惯是把请求频率压在每两秒一条并把 User-Agent、Referer 伪装成正常浏览器。频率克制是稳定性问题不是玄学定时任务半夜挂掉、价格抓到一半断掉十次里有八次是前面频率太激进。商品量超过几百个之后单机串行抓一轮要很久那时候再考虑分布式爬虫、代理池这些扩展单机监控场景用不上。2.2 最小可复现的爬虫脚本requests 加价格 JSON 接口先写一个从商品链接里解析 SKU 的小工具。京东链接有时带#none、#crumbs-wrap这类锚点尾巴正则要把核心部分摘出来。import re def parse_sku_from_url(url: str) - str: 从京东商品链接里解析 SKU 数字串。 支持 https://item.jd.com/100012043978.html 和带锚点的形式。 match re.search(r/(\d{6,})\.html, url) if not match: raise ValueError(f无法从链接中解析 SKU: {url}) return match.group(1)这里用\d{6,}是因为京东 SKU 至少是 6 位数字能过滤掉路径里的其他干扰项。测试时直接用parse_sku_from_url(https://item.jd.com/100012043978.html#none)就能看到提取结果。接下来是核心的取价函数。价格接口返回的是一组 JSON 数组里面包含当前价、原价、移动端价字段个别场景下还会带 skuId 字段。为了兼容不同返回格式取值时做了一层兜底。import time import requests HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36, Referer: https://item.jd.com/, Accept: application/json } def fetch_price(sku_id: str) - dict: 请求京东价格接口返回当前价、原价、移动端价。 url https://p.3.cn/prices/mgets params { skuIds: fJ_{sku_id}, pduid: str(int(time.time() * 1000)) # 时间戳参数避开缓存结果 } resp requests.get(url, paramsparams, headersHEADERS, timeout10) resp.raise_for_status() data resp.json() if not data: raise ValueError(f价格接口返回空数组: sku{sku_id}) return data[0]参数说明skuIds必须带J_前缀这是京东价格接口的固定格式pduid放一个毫秒级时间戳目的是让每次请求的 URL 都不完全一样减少命中缓存导致拿到陈旧价格的概率timeout10不能省爬虫挂掉最常见的原因就是请求卡死没有超时。接口返回的p是当前价op是原价m是移动端价字段都是字符串入库时转成 DECIMAL。2.3 定时调度与失败重试APScheduler 调度加连续失败熔断采集层只剩最后一块定时。很多人用while True time.sleep(3600)硬写进程一重启调度就丢了半夜任务挂了没人知道。常见做法是用 APScheduler 的 CronTrigger把抓取时间写成 cron 表达式进程重启后调度自然恢复。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger import logging import time def job_crawl_all(): 定时任务入口遍历监控清单抓价格并落库。 sku_list load_monitor_sku_list() # 从数据库读取监控商品 SKU 列表 failures 0 for sku in sku_list: try: price_data fetch_price(sku) save_snapshot(sku, price_data) # 存到数据库见第 3 章 failures 0 # 成功后重置连续失败计数 except Exception as exc: failures 1 logging.error(sku%s 抓取失败: %s, sku, exc) if failures 5: logging.critical(连续失败5次中断本轮抓取等待下次调度) break time.sleep(2) # 两个 SKU 之间保持节奏避免触发限频 if __name__ __main__: scheduler BlockingScheduler(timezoneAsia/Shanghai) # 每天 8:15、12:15、18:15 各抓一轮 trigger CronTrigger(hour8,12,18, minute15) scheduler.add_job(job_crawl_all, trigger, idjd_price_crawl) scheduler.start()为什么选 CronTrigger 而不是 interval 间隔触发价格变动集中在白天一天三到四次足够还原波动轨迹也给自己留了手动核对的窗口。timezone必须显式指定否则服务器时区不对任务会在奇怪的时间点跑。连续失败熔断是个小技巧一旦连续 5 次失败就中断本轮避免坏链路在日志里刷几千行错误也避免 IP 被限制得更狠。save_snapshot和load_monitor_sku_list的具体实现放到下一章数据库部分一起写。提示定时任务凌晨跑的时候最好顺手检查一下调度时间是不是被服务器时区影响了。APScheduler 的时区配置是后半夜翻车的高发区。3. 存进 MySQL三张表搞定价格快照与幂等写入数据库是整个系统的记忆。爬虫抓回来的数据如果不落库分析无从谈起。京东价格监控这类系统对数据库的操作其实只有 insert 和 select但恰恰是 insert 这一环最容易出问题重复抓取、同一时刻多条快照、价格字段精度丢失。先把表结构设计对后面踩坑会少一半。3.1 表结构设计商品表、价格快照表、采集日志表三张表各司其职商品表存静态信息价格快照表存每次抓取的价格采集日志表留底用于排查问题。建表语句如下。CREATE TABLE product ( id INT AUTO_INCREMENT PRIMARY KEY, sku_id VARCHAR(20) NOT NULL COMMENT 京东SKU, title VARCHAR(255) COMMENT 商品标题, brand VARCHAR(64) COMMENT 品牌用于竞品分组, category VARCHAR(64) COMMENT 类目, competitor_group VARCHAR(32) COMMENT 竞品分组标签, url VARCHAR(255), UNIQUE KEY uk_sku (sku_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE price_snapshot ( id BIGINT AUTO_INCREMENT PRIMARY KEY, sku_id VARCHAR(20) NOT NULL, price DECIMAL(10,2) NOT NULL COMMENT 当前价, op_price DECIMAL(10,2) COMMENT 原价, plus_price DECIMAL(10,2) COMMENT 移动端/Plus价, stock_status TINYINT DEFAULT 1 COMMENT 1在售 0无货, crawl_time DATETIME NOT NULL, UNIQUE KEY uk_sku_time (sku_id, crawl_time), KEY idx_time (crawl_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE crawl_log ( id BIGINT AUTO_INCREMENT PRIMARY KEY, sku_id VARCHAR(20) NOT NULL, status VARCHAR(16) COMMENT success/failed, error_msg VARCHAR(512), created_at DATETIME NOT NULL ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;三个设计点要说明一下。第一价格字段用DECIMAL(10,2)绝不能用 FLOAT——float 在累计波动计算时会出现 0.1 0.2 不等于 0.3 的精度问题价格数据对精度敏感必须用定点数。第二price_snapshot的唯一键(sku_id, crawl_time)是幂等写入的基础同一秒内重复抓同一条 SKU数据库会直接拒绝重复记录。第三crawl_log表单独留 error_msg 字段排查问题时没有日志就只能靠猜这是血泪经验。3.2 幂等写入唯一键加 insert 冲突更新定时任务偶尔会重跑手动补数据、调度器重启、上一轮任务卡住被重复触发。没有幂等保护重跑一次就多一堆重复快照后面统计的环比会全部错位。幂等写入的常见做法是配合唯一键用 MySQL 的INSERT ... ON DUPLICATE KEY UPDATE。import pymysql from datetime import datetime def save_snapshot(sku_id: str, price_data: dict): 保存价格快照。若同 sku 同时间已存在则原地更新价格。 依赖表上的 uk_sku_time 唯一键。 sql INSERT INTO price_snapshot (sku_id, price, op_price, plus_price, crawl_time) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE priceVALUES(price) conn get_conn() # 连接来自连接池见 3.3 try: with conn.cursor() as cur: cur.execute(sql, ( sku_id, price_data.get(p), price_data.get(op), price_data.get(m), datetime.now().strftime(%Y-%m-%d %H:%M:%S) )) conn.commit() except Exception: conn.rollback() raise finally: conn.close()逻辑说明第一次插入正常写入如果同一秒内同 SKU 已经有一条快照唯一键冲突后执行UPDATE把 price 刷成最新值。这样不管任务被触发几次同一秒只保留一条记录。如果想改成每天一条只需把唯一键换成(sku_id, DATE(crawl_time))语义就从「秒级去重」变成「日级去重」。注意price_data.get(p)可能返回字符串空值入库前最好做一次类型转换空值统一转 None避免把空字符串写进 DECIMAL 字段导致报错。3.3 连接池与事务边界PooledDB 的配置参数定时任务每轮要抓几十上百个 SKU如果每条都新建连接MySQL 的握手开销会吃掉不少时间还会把数据库连接数打满。用 DBUtils 的 PooledDB 做连接池是 Python 生态里常见的做法。from dbutils.pooled_db import PooledDB import pymysql POOL PooledDB( creatorpymysql, maxconnections10, mincached2, maxcached5, blockingTrue, ping1, host127.0.0.1, port3306, userjd_app, passwordyour_password, databasejd_price, charsetutf8mb4 ) def get_conn(): 从连接池取一个连接用完记得 close 归还。 return POOL.connection()几个参数值得调mincached2让进程启动时就预热两个连接定时任务第一轮不会因为建连慢而超时maxcached5控制空闲连接上限避免任务结束后连接长时间占着不释放blockingTrue很关键连接池满了之后请求会排队等待而不是直接抛异常ping1每次复用连接前检查一下连接是否还活着MySQL 的 wait_timeout 默认 8 小时定时任务间隔超过这个时间连接池里拿到的就是死连接。事务边界上单条insert不需要显式开事务commit一次即可。批量写多个 SKU 时用executemany拼一条 SQL全部成功后再commit失败整体rollback。我见过有人把每条 insert 都包一层事务性能慢一倍不说日志里全是死锁超时。4. 波动分析与竞品对比用统计方法和 ECharts 把价格趋势讲清楚数据落库只是第一步。运营和市场研究的同事真正要的是「这商品最近价格稳不稳」「竞品什么时候调过价」「我们该不该跟进」。这部分用 pandas 做统计再用 ECharts 出图是性价比最高的组合。4.1 波动率怎么算才不骗自己环比、移动平均与异常剔除从库里读出一段价格序列之后直接拿原始快照算波动率很容易被噪声带偏。一天抓三次上午 100 元、中午促销 80 元、下午恢复 100 元如果按快照粒度算环比这张表就没法看了。常见做法是先按天聚合再算统计指标。import pandas as pd def load_daily_prices(sku_id: str, start_date: str) - pd.DataFrame: 从 price_snapshot 读取某 SKU 的日度价格序列。 df pd.read_sql( SELECT sku_id, crawl_time, price FROM price_snapshot WHERE sku_id %s AND crawl_time %s ORDER BY crawl_time , conget_conn(), params(sku_id, start_date) ) df[crawl_time] pd.to_datetime(df[crawl_time]) df[price] pd.to_numeric(df[price], errorscoerce) # 按天聚合取当天最低价促销口径更敏感 daily df.groupby(df[crawl_time].dt.date)[price].min() return daily daily_ret daily.pct_change() # 环比今天的价格相对昨天变动的百分比 rolling_mean daily.rolling(7).mean() # 7 日移动平均平滑促销造成的单点波动 volatility daily_ret.std() # 波动率环比序列的标准差参数口径上min()适合运营盯促销的场景能捕捉「今天出现了历史低价」如果想看平稳价格改用mean()。pct_change()直接给出相邻两天的变化率一眼就能定位哪天调了价。rolling(7).mean()把 7 天内来回波动的噪声抹平判断趋势用它比原始曲线可靠得多。波动率std()用来给商品分档波动率高的商品值得每天盯波动率低的可以改成每周看一次这是给运营排优先级的好抓手。异常剔除不能省。无货商品的价格可能被抓成 0偶尔接口还会返回瞬时错价。我在分析前会先做一道过滤价格小于等于 0 的直接丢价格低于原价三折以上的也丢——正常促销很少打到三折以下这种值大概率是脏数据。4.2 竞品对比的对齐方式同 SKU、同类目、同价格带怎么选竞品对比最容易犯的错是「拿苹果手机和红米手机比价格」。对比前必须先明确对齐口径否则统计出来的落差没有运营意义。我平时用三种口径对比口径数据来源字段适合场景同 SKU 不同店铺competitor_group京东自营与第三方经销商差价监控同类目category品牌均价、最低价排名同价格带op_price 分档价格敏感区间的竞争态势同 SKU 对比是严格意义上的竞品同一个货自营和第三方卖不同价。建表时在 product 表里给这些链接打上同一个competitor_group标签比如「iPhone15 自营/经销商」分析时按组聚合。同类目对比看的是品牌间价格带分布常用于新品定价。同价格带对比对运营最有参考价值把 0-100、100-500、500-1000 分成几档看自己商品所在档位里同行最低价是多少、促销频率多高。分组对比的核心 SQL 很简单GROUP BY competitor_group, DATE(crawl_time)取每天最低价。算出来之后哪家调价更频繁、最低价凹陷出现在哪天都比单个商品的趋势更有决策价值。4.3 ECharts 出图趋势折线、预警线与竞品多系列对比统计分析的结果最终要画出来。Flask 加 ECharts 是常见做法后端把 pandas 算好的序列转成 JSON 吐给前端前端用 ECharts 渲染。后端接口可以长这样。from flask import Flask, jsonify app Flask(__name__) app.route(/api/price_trend/sku_id) def price_trend(sku_id): 返回某 SKU 的日价格、7日均线以及同竞品组其他商品的价格序列。 trend build_trend_data(sku_id) # 复用 4.1 的 pandas 逻辑 return jsonify(trend)前端 ECharts 的核心配置如下。const chart echarts.init(document.getElementById(trend)); chart.setOption({ xAxis: { type: category, data: trend.dates }, yAxis: { type: value, scale: true }, tooltip: { trigger: axis }, series: [ { name: 日均最低价, type: line, data: trend.prices, smooth: true }, { name: 7日均线, type: line, data: trend.ma7, lineStyle: { type: dashed } }, { name: 竞品A, type: line, data: trend.competitorA, smooth: true } ], visualMap: { dimension: 1, pieces: [{ lte: trend.warnLine, color: #d9534f }] // 低于预警线的点标红 } });这里有两个实战技巧。一是markLine可以画一条预警线比如「低于 900 元就通知运营跟进」价格跌破线时人眼能立刻看到。二是visualMap的pieces配置可以把低于阈值的点标红比人肉盯数字高效得多。竞品对比图用多系列折线每个竞品一条线重合区间就是价格竞争胶着区如果竞品数量超过五个折线图会变成毛线团那时候改用箱线图按周展示价格分布更清晰。5. 避坑指南京东价格爬虫的常见翻车现场与排查路径这套系统跑起来之后真正消耗精力的不是写代码而是数据跑歪了你得能定位。下面五条是我在实际维护中被反复教育过的坑按「现象、原因、解决」写清楚希望能帮你少走几趟。5.1 价格接口突然返回空数组或 302是限频还是被风控了现象爬虫跑了两周一直正常某天开始fetch_price返回空数组或者 requests 跟随重定向后拿到了一个登录页的 HTML。原因短时间请求频率太高触发了对该 IP 的限制也可能是并发任务没控制好多个进程同时抓同一个价格接口。解决先看抓取日志里的时间戳如果失败集中在同一分钟基本就是频率问题。把time.sleep(2)放宽到time.sleep(3)并把抓取任务改成单进程串行。如果已经抓了几千次换个出口 IP 再跑同时把pduid的随机性加大。爬虫被限制不是玄学频率曲线平缓就能活很久。5.2 同一条商品链接一天抓到两个价格是改价还是抓到了促销价现象上午抓的 100 元下午变成 80 元第二天又回到 100 元环比表上出现一条 -20%。原因这大概率不是手滑改价而是商品在秒杀时段切换了促销状态。价格接口里的 p 字段是「当前生效价」秒杀开始是秒杀价秒杀结束恢复原价同一 SKU 一天内出现两个价格完全正常。解决分析时先按天聚合用min()取当天最低价作为促销敏感口径并把op原价一起存下来。真要判断是否长期调价看 7 日均线而不是单日曲线。如果秒杀频繁的商品经常误导分析给商品表加一个is_flash_sale标记字段。5.3 无货商品价格是 0 或空值直接进库会把统计搞崩现象某商品断货后价格字段抓到 0.00环比计算出现 -100%移动平均线被砸出一个大坑竞品对比的最低价比真实情况低一大截。原因京东对无货商品的接口返回并不统一有时 p 是 0有时 p 是空字符串甚至整个数组为空。爬虫把异常值当成正常价格写了进去。解决入库存前判断stock_status只有有货时才把价格写进 price 字段分析阶段再做一次过滤price 0的记录直接剔除。两个环节都做数据库里留原始状态分析时按需清洗。还有一种做法无货时把 price 写成 NULL这样 SQL 聚合函数MIN()、AVG()会自动跳过它比写 0 干净得多。5.4 凌晨定时任务集体失败日志里全是连接超时现象前一天数据正常第二天发现凌晨那轮任务零写入crawl_log 里全是TimeoutError或者Cant connect to MySQL server。原因两个叠加问题一是服务器在凌晨重启导致进程没有拉起二是 MySQL 的 wait_timeout 把连接池里的空闲连接断掉了进程复用死连接时直接超时。解决用 systemd 或 supervisor 管理爬虫进程设置Restartalways确保进程死了能自动拉起连接池配置里把ping1打开复用连接前先探活。再加一个对账手段每天早上跑一条 SQL数一下昨天的快照数量如果明显少于预期的「SKU 数乘以抓取次数」立刻告警。丢数据的代价远比晚半小时发现大。5.5 数据量涨起来后查询变慢索引和归档怎么处理现象跑了三四个月后price_snapshot 表动辄几十万行按 SKU 查趋势时ORDER BY crawl_time越来越慢ECharts 接口响应从 200ms 涨到 3 秒。原因表里索引设计不合理或者数据量增长超过了 MySQL 单表舒适区。解决先确认查询条件是否命中索引。按 SKU 查时间序列应该用(sku_id, crawl_time)联合索引我已经在 3.1 的唯一键里加上了这条路径不必再建。历史数据按季度归档超过半年的快照迁移到price_snapshot_archive表正常查询只扫最近半年。索引不是越多越好联合索引的字段顺序要跟查询条件对齐这是数据库同步都得注意的通用问题。6. 数据质量验证跑完一个周期后怎么确认系统没在骗你整个系统跑了两三周之后最难的不是爬虫挂掉而是数据看起来正常、实际上已经偷偷坏了。我吃过一次亏某个 SKU 的接口字段突然变了p 字段一直返回空字符串爬虫没抛异常0 元价格在数据库里躺了一周。等做波动分析时才发现几条 -100% 的环比把均值拉得没法看。后来我把数据质量校验写成了固定流程每次抓取结束后顺手跑一遍而不是等分析阶段被数据打脸。第一道校验是数量对账直接查当天快照数够不够预期-- 每天应抓 N 个 SKU * 3 轮核对当天快照总数是否匹配 SELECT DATE(crawl_time) AS dt, COUNT(DISTINCT sku_id) AS sku_cnt, COUNT(*) AS snapshot_cnt FROM price_snapshot WHERE crawl_time CURDATE() GROUP BY DATE(crawl_time);如果sku_cnt明显少于监控表里的 SKU 数量说明有商品整轮没抓到先去查 crawl_log 定位失败原因。如果snapshot_cnt远大于sku_cnt乘以 3多半是调度重复触发了检查唯一键是否生效。第二道校验是价格合理性用 pandas 跑一遍极差检查# 抽查同一天内同一 SKU 的价格极差超过 20% 中位数的多半是混入了异常值 bad daily_price.groupby(sku_id)[price].apply( lambda s: (s.max() - s.min()) 0.2 * s.median() ) bad_skus bad[bad].index.tolist() if bad_skus: print(疑似异常SKU:, bad_skus)这道检查专门抓「接口字段突变」和「促销价误读」这类爬虫不报错但数据错误的情况。第三道校验最笨也最有效每周人工打开两三个商品链接跟库里最近一条快照对一下差别超过 5% 就去查接口返回是否变了。这个习惯我保留到现在每次手动核对都能发现一些 SQL 和代码测不出来的细节比如某品类突然改了口径、某商品 SKU 被下架重新上架。如果你打算把这套系统长期跑下去我的建议是宁可把抓取频率压低一点也要把数据质量校验做扎实。价格数据最贵的地方不是抓而是「你知道哪一条是可信的」。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/17 9:29:21 · 阅读 1.2k 从爬虫到可视化：Python构建京东商品价格监控与分析系统
简介这套京东商品价格爬取与分析系统是一份完整的Python项目代码面向电商运营人员、市场研究者和爬虫技术学习者。系统通过定时爬虫抓取京东指定商品链接的价格信息并将其存储至本地数据库再以数据可视化与统计方法分析价格波动趋势、开展竞品对比辅助定价策略与市场研判。压缩包共6个文件包含3个Python脚本分别负责信息爬取、数据展示与主流程控制、说明txt、README及配套Word文档整体体积仅36KB结构紧凑清晰便于直接阅读和二次开发。目前已有76人学习下载适合有一定Python基础并希望掌握价格监控全流程的读者可参考其数据库表设计、定时任务写法及可视化思路快速迁移到自有项目中扩展分析维度。1. 京东商品价格爬取与分析系统为什么要把盯价格这件小事做成自动化做电商运营或者市场研究的人大多经历过这种场景早上打开京东看一遍竞品价格下午再刷一次促销来了赶紧截图通知同事。一天两三个小时搭进去最后复盘时发现——昨天竞品到底几点调的价、涨了又降了几轮完全说不清楚。京东商品价格数据爬取与分析系统就是把这件事变成一套自动化的流程用 Python 爬虫定时抓取京东指定商品链接的价格信息存进本地数据库再借数据可视化与统计方法分析价格波动趋势以及与竞品的对比给电商运营和市场研究提供决策依据。它适合两类人一类是被手动比价耗掉大量时间的运营另一类是想要一个完整数据链路练手的 Python 学习者。整个系统的技术栈并不复杂requests、MySQL、pandas、ECharts 这几件常用工具就够。2. 用 requests 定时抓京东价格从商品链接到价格快照的采集层实现采集层是整条链路的入口抓回来的数据脏不脏、全不全后面所有统计分析都建立在这层基础上。抓京东价格网上很多教程一上来就是 selenium 或者 pyppeteer 模拟浏览器但监控指定商品链接的价格用 requests 就够了。网络爬虫原理不难理解构造 HTTP 请求解析响应里的 JSON 字段把结果落库剩下的交给定时调度。2.1 先拆解采集目标SKU、价格接口与反爬边界京东商品链接的标准格式是https://item.jd.com/100012043978.html中间那一串数字就是 SKU也就是 wareId是整个监控周期里所有表的主键来源。开始写爬虫之前我一般先把要监控的链接整理成一张清单skuId、竞品分组、品牌、类目。因为后面做竞品对比时光有价格没有上下文统计出来的数字没有运营意义。采集目标字段上要抓这几类当前价 p、原价 op、移动端价 m、库存状态。当前价就是页面上展示的最终成交价原价用于算折扣力度移动端价在部分品类上会明显低于 PC 端。库存状态不能省无货商品的价格经常会变成 0 或者空字符串如果不单独记录分析阶段很难解释「突然冒出来的 0 元价格」是怎么来的。为什么不直接对页面 HTML 做 xpath 解析因为商品页的价格区域是 JS 异步渲染出来的requests 拿到的静态源码里价格节点经常是空的。花半天调 xpath 表达式不如直接命中价格接口。这是京东价格爬虫最常见的认知差很多人觉得解析 HTML 才算爬虫实际上读 JSON 接口又快又稳也更容易做异常处理。常见的做法是请求京东的价格接口p.3.cn/prices/mgets按 SKU 集合批量返回价格 JSON不需要登录、不需要解析复杂页面。反爬边界方面单 IP 高频轮询会被限制。我的习惯是把请求频率压在每两秒一条并把 User-Agent、Referer 伪装成正常浏览器。频率克制是稳定性问题不是玄学定时任务半夜挂掉、价格抓到一半断掉十次里有八次是前面频率太激进。商品量超过几百个之后单机串行抓一轮要很久那时候再考虑分布式爬虫、代理池这些扩展单机监控场景用不上。2.2 最小可复现的爬虫脚本requests 加价格 JSON 接口先写一个从商品链接里解析 SKU 的小工具。京东链接有时带#none、#crumbs-wrap这类锚点尾巴正则要把核心部分摘出来。import re def parse_sku_from_url(url: str) - str: 从京东商品链接里解析 SKU 数字串。 支持 https://item.jd.com/100012043978.html 和带锚点的形式。 match re.search(r/(\d{6,})\.html, url) if not match: raise ValueError(f无法从链接中解析 SKU: {url}) return match.group(1)这里用\d{6,}是因为京东 SKU 至少是 6 位数字能过滤掉路径里的其他干扰项。测试时直接用parse_sku_from_url(https://item.jd.com/100012043978.html#none)就能看到提取结果。接下来是核心的取价函数。价格接口返回的是一组 JSON 数组里面包含当前价、原价、移动端价字段个别场景下还会带 skuId 字段。为了兼容不同返回格式取值时做了一层兜底。import time import requests HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36, Referer: https://item.jd.com/, Accept: application/json } def fetch_price(sku_id: str) - dict: 请求京东价格接口返回当前价、原价、移动端价。 url https://p.3.cn/prices/mgets params { skuIds: fJ_{sku_id}, pduid: str(int(time.time() * 1000)) # 时间戳参数避开缓存结果 } resp requests.get(url, paramsparams, headersHEADERS, timeout10) resp.raise_for_status() data resp.json() if not data: raise ValueError(f价格接口返回空数组: sku{sku_id}) return data[0]参数说明skuIds必须带J_前缀这是京东价格接口的固定格式pduid放一个毫秒级时间戳目的是让每次请求的 URL 都不完全一样减少命中缓存导致拿到陈旧价格的概率timeout10不能省爬虫挂掉最常见的原因就是请求卡死没有超时。接口返回的p是当前价op是原价m是移动端价字段都是字符串入库时转成 DECIMAL。2.3 定时调度与失败重试APScheduler 调度加连续失败熔断采集层只剩最后一块定时。很多人用while True time.sleep(3600)硬写进程一重启调度就丢了半夜任务挂了没人知道。常见做法是用 APScheduler 的 CronTrigger把抓取时间写成 cron 表达式进程重启后调度自然恢复。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger import logging import time def job_crawl_all(): 定时任务入口遍历监控清单抓价格并落库。 sku_list load_monitor_sku_list() # 从数据库读取监控商品 SKU 列表 failures 0 for sku in sku_list: try: price_data fetch_price(sku) save_snapshot(sku, price_data) # 存到数据库见第 3 章 failures 0 # 成功后重置连续失败计数 except Exception as exc: failures 1 logging.error(sku%s 抓取失败: %s, sku, exc) if failures 5: logging.critical(连续失败5次中断本轮抓取等待下次调度) break time.sleep(2) # 两个 SKU 之间保持节奏避免触发限频 if __name__ __main__: scheduler BlockingScheduler(timezoneAsia/Shanghai) # 每天 8:15、12:15、18:15 各抓一轮 trigger CronTrigger(hour8,12,18, minute15) scheduler.add_job(job_crawl_all, trigger, idjd_price_crawl) scheduler.start()为什么选 CronTrigger 而不是 interval 间隔触发价格变动集中在白天一天三到四次足够还原波动轨迹也给自己留了手动核对的窗口。timezone必须显式指定否则服务器时区不对任务会在奇怪的时间点跑。连续失败熔断是个小技巧一旦连续 5 次失败就中断本轮避免坏链路在日志里刷几千行错误也避免 IP 被限制得更狠。save_snapshot和load_monitor_sku_list的具体实现放到下一章数据库部分一起写。提示定时任务凌晨跑的时候最好顺手检查一下调度时间是不是被服务器时区影响了。APScheduler 的时区配置是后半夜翻车的高发区。3. 存进 MySQL三张表搞定价格快照与幂等写入数据库是整个系统的记忆。爬虫抓回来的数据如果不落库分析无从谈起。京东价格监控这类系统对数据库的操作其实只有 insert 和 select但恰恰是 insert 这一环最容易出问题重复抓取、同一时刻多条快照、价格字段精度丢失。先把表结构设计对后面踩坑会少一半。3.1 表结构设计商品表、价格快照表、采集日志表三张表各司其职商品表存静态信息价格快照表存每次抓取的价格采集日志表留底用于排查问题。建表语句如下。CREATE TABLE product ( id INT AUTO_INCREMENT PRIMARY KEY, sku_id VARCHAR(20) NOT NULL COMMENT 京东SKU, title VARCHAR(255) COMMENT 商品标题, brand VARCHAR(64) COMMENT 品牌用于竞品分组, category VARCHAR(64) COMMENT 类目, competitor_group VARCHAR(32) COMMENT 竞品分组标签, url VARCHAR(255), UNIQUE KEY uk_sku (sku_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE price_snapshot ( id BIGINT AUTO_INCREMENT PRIMARY KEY, sku_id VARCHAR(20) NOT NULL, price DECIMAL(10,2) NOT NULL COMMENT 当前价, op_price DECIMAL(10,2) COMMENT 原价, plus_price DECIMAL(10,2) COMMENT 移动端/Plus价, stock_status TINYINT DEFAULT 1 COMMENT 1在售 0无货, crawl_time DATETIME NOT NULL, UNIQUE KEY uk_sku_time (sku_id, crawl_time), KEY idx_time (crawl_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE crawl_log ( id BIGINT AUTO_INCREMENT PRIMARY KEY, sku_id VARCHAR(20) NOT NULL, status VARCHAR(16) COMMENT success/failed, error_msg VARCHAR(512), created_at DATETIME NOT NULL ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;三个设计点要说明一下。第一价格字段用DECIMAL(10,2)绝不能用 FLOAT——float 在累计波动计算时会出现 0.1 0.2 不等于 0.3 的精度问题价格数据对精度敏感必须用定点数。第二price_snapshot的唯一键(sku_id, crawl_time)是幂等写入的基础同一秒内重复抓同一条 SKU数据库会直接拒绝重复记录。第三crawl_log表单独留 error_msg 字段排查问题时没有日志就只能靠猜这是血泪经验。3.2 幂等写入唯一键加 insert 冲突更新定时任务偶尔会重跑手动补数据、调度器重启、上一轮任务卡住被重复触发。没有幂等保护重跑一次就多一堆重复快照后面统计的环比会全部错位。幂等写入的常见做法是配合唯一键用 MySQL 的INSERT ... ON DUPLICATE KEY UPDATE。import pymysql from datetime import datetime def save_snapshot(sku_id: str, price_data: dict): 保存价格快照。若同 sku 同时间已存在则原地更新价格。 依赖表上的 uk_sku_time 唯一键。 sql INSERT INTO price_snapshot (sku_id, price, op_price, plus_price, crawl_time) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE priceVALUES(price) conn get_conn() # 连接来自连接池见 3.3 try: with conn.cursor() as cur: cur.execute(sql, ( sku_id, price_data.get(p), price_data.get(op), price_data.get(m), datetime.now().strftime(%Y-%m-%d %H:%M:%S) )) conn.commit() except Exception: conn.rollback() raise finally: conn.close()逻辑说明第一次插入正常写入如果同一秒内同 SKU 已经有一条快照唯一键冲突后执行UPDATE把 price 刷成最新值。这样不管任务被触发几次同一秒只保留一条记录。如果想改成每天一条只需把唯一键换成(sku_id, DATE(crawl_time))语义就从「秒级去重」变成「日级去重」。注意price_data.get(p)可能返回字符串空值入库前最好做一次类型转换空值统一转 None避免把空字符串写进 DECIMAL 字段导致报错。3.3 连接池与事务边界PooledDB 的配置参数定时任务每轮要抓几十上百个 SKU如果每条都新建连接MySQL 的握手开销会吃掉不少时间还会把数据库连接数打满。用 DBUtils 的 PooledDB 做连接池是 Python 生态里常见的做法。from dbutils.pooled_db import PooledDB import pymysql POOL PooledDB( creatorpymysql, maxconnections10, mincached2, maxcached5, blockingTrue, ping1, host127.0.0.1, port3306, userjd_app, passwordyour_password, databasejd_price, charsetutf8mb4 ) def get_conn(): 从连接池取一个连接用完记得 close 归还。 return POOL.connection()几个参数值得调mincached2让进程启动时就预热两个连接定时任务第一轮不会因为建连慢而超时maxcached5控制空闲连接上限避免任务结束后连接长时间占着不释放blockingTrue很关键连接池满了之后请求会排队等待而不是直接抛异常ping1每次复用连接前检查一下连接是否还活着MySQL 的 wait_timeout 默认 8 小时定时任务间隔超过这个时间连接池里拿到的就是死连接。事务边界上单条insert不需要显式开事务commit一次即可。批量写多个 SKU 时用executemany拼一条 SQL全部成功后再commit失败整体rollback。我见过有人把每条 insert 都包一层事务性能慢一倍不说日志里全是死锁超时。4. 波动分析与竞品对比用统计方法和 ECharts 把价格趋势讲清楚数据落库只是第一步。运营和市场研究的同事真正要的是「这商品最近价格稳不稳」「竞品什么时候调过价」「我们该不该跟进」。这部分用 pandas 做统计再用 ECharts 出图是性价比最高的组合。4.1 波动率怎么算才不骗自己环比、移动平均与异常剔除从库里读出一段价格序列之后直接拿原始快照算波动率很容易被噪声带偏。一天抓三次上午 100 元、中午促销 80 元、下午恢复 100 元如果按快照粒度算环比这张表就没法看了。常见做法是先按天聚合再算统计指标。import pandas as pd def load_daily_prices(sku_id: str, start_date: str) - pd.DataFrame: 从 price_snapshot 读取某 SKU 的日度价格序列。 df pd.read_sql( SELECT sku_id, crawl_time, price FROM price_snapshot WHERE sku_id %s AND crawl_time %s ORDER BY crawl_time , conget_conn(), params(sku_id, start_date) ) df[crawl_time] pd.to_datetime(df[crawl_time]) df[price] pd.to_numeric(df[price], errorscoerce) # 按天聚合取当天最低价促销口径更敏感 daily df.groupby(df[crawl_time].dt.date)[price].min() return daily daily_ret daily.pct_change() # 环比今天的价格相对昨天变动的百分比 rolling_mean daily.rolling(7).mean() # 7 日移动平均平滑促销造成的单点波动 volatility daily_ret.std() # 波动率环比序列的标准差参数口径上min()适合运营盯促销的场景能捕捉「今天出现了历史低价」如果想看平稳价格改用mean()。pct_change()直接给出相邻两天的变化率一眼就能定位哪天调了价。rolling(7).mean()把 7 天内来回波动的噪声抹平判断趋势用它比原始曲线可靠得多。波动率std()用来给商品分档波动率高的商品值得每天盯波动率低的可以改成每周看一次这是给运营排优先级的好抓手。异常剔除不能省。无货商品的价格可能被抓成 0偶尔接口还会返回瞬时错价。我在分析前会先做一道过滤价格小于等于 0 的直接丢价格低于原价三折以上的也丢——正常促销很少打到三折以下这种值大概率是脏数据。4.2 竞品对比的对齐方式同 SKU、同类目、同价格带怎么选竞品对比最容易犯的错是「拿苹果手机和红米手机比价格」。对比前必须先明确对齐口径否则统计出来的落差没有运营意义。我平时用三种口径对比口径数据来源字段适合场景同 SKU 不同店铺competitor_group京东自营与第三方经销商差价监控同类目category品牌均价、最低价排名同价格带op_price 分档价格敏感区间的竞争态势同 SKU 对比是严格意义上的竞品同一个货自营和第三方卖不同价。建表时在 product 表里给这些链接打上同一个competitor_group标签比如「iPhone15 自营/经销商」分析时按组聚合。同类目对比看的是品牌间价格带分布常用于新品定价。同价格带对比对运营最有参考价值把 0-100、100-500、500-1000 分成几档看自己商品所在档位里同行最低价是多少、促销频率多高。分组对比的核心 SQL 很简单GROUP BY competitor_group, DATE(crawl_time)取每天最低价。算出来之后哪家调价更频繁、最低价凹陷出现在哪天都比单个商品的趋势更有决策价值。4.3 ECharts 出图趋势折线、预警线与竞品多系列对比统计分析的结果最终要画出来。Flask 加 ECharts 是常见做法后端把 pandas 算好的序列转成 JSON 吐给前端前端用 ECharts 渲染。后端接口可以长这样。from flask import Flask, jsonify app Flask(__name__) app.route(/api/price_trend/sku_id) def price_trend(sku_id): 返回某 SKU 的日价格、7日均线以及同竞品组其他商品的价格序列。 trend build_trend_data(sku_id) # 复用 4.1 的 pandas 逻辑 return jsonify(trend)前端 ECharts 的核心配置如下。const chart echarts.init(document.getElementById(trend)); chart.setOption({ xAxis: { type: category, data: trend.dates }, yAxis: { type: value, scale: true }, tooltip: { trigger: axis }, series: [ { name: 日均最低价, type: line, data: trend.prices, smooth: true }, { name: 7日均线, type: line, data: trend.ma7, lineStyle: { type: dashed } }, { name: 竞品A, type: line, data: trend.competitorA, smooth: true } ], visualMap: { dimension: 1, pieces: [{ lte: trend.warnLine, color: #d9534f }] // 低于预警线的点标红 } });这里有两个实战技巧。一是markLine可以画一条预警线比如「低于 900 元就通知运营跟进」价格跌破线时人眼能立刻看到。二是visualMap的pieces配置可以把低于阈值的点标红比人肉盯数字高效得多。竞品对比图用多系列折线每个竞品一条线重合区间就是价格竞争胶着区如果竞品数量超过五个折线图会变成毛线团那时候改用箱线图按周展示价格分布更清晰。5. 避坑指南京东价格爬虫的常见翻车现场与排查路径这套系统跑起来之后真正消耗精力的不是写代码而是数据跑歪了你得能定位。下面五条是我在实际维护中被反复教育过的坑按「现象、原因、解决」写清楚希望能帮你少走几趟。5.1 价格接口突然返回空数组或 302是限频还是被风控了现象爬虫跑了两周一直正常某天开始fetch_price返回空数组或者 requests 跟随重定向后拿到了一个登录页的 HTML。原因短时间请求频率太高触发了对该 IP 的限制也可能是并发任务没控制好多个进程同时抓同一个价格接口。解决先看抓取日志里的时间戳如果失败集中在同一分钟基本就是频率问题。把time.sleep(2)放宽到time.sleep(3)并把抓取任务改成单进程串行。如果已经抓了几千次换个出口 IP 再跑同时把pduid的随机性加大。爬虫被限制不是玄学频率曲线平缓就能活很久。5.2 同一条商品链接一天抓到两个价格是改价还是抓到了促销价现象上午抓的 100 元下午变成 80 元第二天又回到 100 元环比表上出现一条 -20%。原因这大概率不是手滑改价而是商品在秒杀时段切换了促销状态。价格接口里的 p 字段是「当前生效价」秒杀开始是秒杀价秒杀结束恢复原价同一 SKU 一天内出现两个价格完全正常。解决分析时先按天聚合用min()取当天最低价作为促销敏感口径并把op原价一起存下来。真要判断是否长期调价看 7 日均线而不是单日曲线。如果秒杀频繁的商品经常误导分析给商品表加一个is_flash_sale标记字段。5.3 无货商品价格是 0 或空值直接进库会把统计搞崩现象某商品断货后价格字段抓到 0.00环比计算出现 -100%移动平均线被砸出一个大坑竞品对比的最低价比真实情况低一大截。原因京东对无货商品的接口返回并不统一有时 p 是 0有时 p 是空字符串甚至整个数组为空。爬虫把异常值当成正常价格写了进去。解决入库存前判断stock_status只有有货时才把价格写进 price 字段分析阶段再做一次过滤price 0的记录直接剔除。两个环节都做数据库里留原始状态分析时按需清洗。还有一种做法无货时把 price 写成 NULL这样 SQL 聚合函数MIN()、AVG()会自动跳过它比写 0 干净得多。5.4 凌晨定时任务集体失败日志里全是连接超时现象前一天数据正常第二天发现凌晨那轮任务零写入crawl_log 里全是TimeoutError或者Cant connect to MySQL server。原因两个叠加问题一是服务器在凌晨重启导致进程没有拉起二是 MySQL 的 wait_timeout 把连接池里的空闲连接断掉了进程复用死连接时直接超时。解决用 systemd 或 supervisor 管理爬虫进程设置Restartalways确保进程死了能自动拉起连接池配置里把ping1打开复用连接前先探活。再加一个对账手段每天早上跑一条 SQL数一下昨天的快照数量如果明显少于预期的「SKU 数乘以抓取次数」立刻告警。丢数据的代价远比晚半小时发现大。5.5 数据量涨起来后查询变慢索引和归档怎么处理现象跑了三四个月后price_snapshot 表动辄几十万行按 SKU 查趋势时ORDER BY crawl_time越来越慢ECharts 接口响应从 200ms 涨到 3 秒。原因表里索引设计不合理或者数据量增长超过了 MySQL 单表舒适区。解决先确认查询条件是否命中索引。按 SKU 查时间序列应该用(sku_id, crawl_time)联合索引我已经在 3.1 的唯一键里加上了这条路径不必再建。历史数据按季度归档超过半年的快照迁移到price_snapshot_archive表正常查询只扫最近半年。索引不是越多越好联合索引的字段顺序要跟查询条件对齐这是数据库同步都得注意的通用问题。6. 数据质量验证跑完一个周期后怎么确认系统没在骗你整个系统跑了两三周之后最难的不是爬虫挂掉而是数据看起来正常、实际上已经偷偷坏了。我吃过一次亏某个 SKU 的接口字段突然变了p 字段一直返回空字符串爬虫没抛异常0 元价格在数据库里躺了一周。等做波动分析时才发现几条 -100% 的环比把均值拉得没法看。后来我把数据质量校验写成了固定流程每次抓取结束后顺手跑一遍而不是等分析阶段被数据打脸。第一道校验是数量对账直接查当天快照数够不够预期-- 每天应抓 N 个 SKU * 3 轮核对当天快照总数是否匹配 SELECT DATE(crawl_time) AS dt, COUNT(DISTINCT sku_id) AS sku_cnt, COUNT(*) AS snapshot_cnt FROM price_snapshot WHERE crawl_time CURDATE() GROUP BY DATE(crawl_time);如果sku_cnt明显少于监控表里的 SKU 数量说明有商品整轮没抓到先去查 crawl_log 定位失败原因。如果snapshot_cnt远大于sku_cnt乘以 3多半是调度重复触发了检查唯一键是否生效。第二道校验是价格合理性用 pandas 跑一遍极差检查# 抽查同一天内同一 SKU 的价格极差超过 20% 中位数的多半是混入了异常值 bad daily_price.groupby(sku_id)[price].apply( lambda s: (s.max() - s.min()) 0.2 * s.median() ) bad_skus bad[bad].index.tolist() if bad_skus: print(疑似异常SKU:, bad_skus)这道检查专门抓「接口字段突变」和「促销价误读」这类爬虫不报错但数据错误的情况。第三道校验最笨也最有效每周人工打开两三个商品链接跟库里最近一条快照对一下差别超过 5% 就去查接口返回是否变了。这个习惯我保留到现在每次手动核对都能发现一些 SQL 和代码测不出来的细节比如某品类突然改了口径、某商品 SKU 被下架重新上架。如果你打算把这套系统长期跑下去我的建议是宁可把抓取频率压低一点也要把数据质量校验做扎实。价格数据最贵的地方不是抓而是「你知道哪一条是可信的」。希望帮到你。本文还有配套的精品资源点击获取