简介这是一套面向电商数据分析初学者与Python爬虫实践者的京东价格监控实战项目聚焦商品价格动态采集、本地化存储与可视化分析全流程助力运营人员掌握竞品价格追踪与市场趋势研判方法。资源包共6个文件含3个核心Python脚本负责爬取、解析与主控调度、1份README.md说明文档、1个说明文本及1份附赠资源Word文档总大小仅36KB轻量易部署适合快速复现与二次开发。已有76人学习下载反映出其在小规模电商数据实践场景中的实用价值。读者可直接运行get.py等脚本实现指定京东商品链接的价格定时抓取数据自动存入SQLite数据库通过look.py调用Matplotlib完成价格波动折线图与竞品对比柱状图配套文档还梳理了反爬应对思路、数据库表结构设计及常见报错处理建议目录模块分工明确便于理解工程逻辑与拓展功能。1. 京东商品价格数据爬取与分析系统为什么电商运营团队宁可重写三次脚本也不愿手动查价你有没有见过运营同事凌晨三点还在Excel里逐条粘贴京东页面的价格截图有没有在促销大促前夜发现竞品悄悄调价三次而你的报价表还停留在昨天下午这不是玄学是数据流断层——京东商品页价格每30秒可能刷新一次但人工采集频率往往以“天”为单位。这个系统不是教你怎么写一个能跑通的爬虫demo而是交付一套可长期值守、抗页面结构微调、带异常自愈能力、价格波动可归因到SKU粒度的闭环方案。它用Python构建但核心不在requests或selenium而在如何让爬虫像运维服务一样稳住、让价格数据像财务流水一样可信、让分析结论能直接支撑采购压价或活动定价决策。适合电商运营、市场调研、比价平台技术侧、以及需要做价格监控SaaS产品的开发者——如果你的KPI里有“价格响应时效2小时”或“竞品调价捕捉率95%”那这篇就是你该抄的第一份作业。2. 从页面结构到请求链路为什么京东反爬不是靠验证码而是靠“动态价格加载埋点校验”京东的商品价格不直接写在HTML源码里这是所有新手翻车的第一道墙。你用BeautifulSoup解析div classprice返回空你用Selenium等页面加载完成再取发现.text拿到的是“¥”符号加一串乱码。这不是反爬是京东的价格渲染策略价格由前端JS通过加密API异步拉取且每次请求携带动态生成的callback参数和_时间戳同时校验Referer、User-Agent、Cookie中的pt_key和pt_pin登录态凭证。更关键的是价格接口返回的并非明文数字而是经过简单位运算混淆的字符串如1299变成1300再异或0x1F需逆向解密。2.1 抓包定位真实价格接口绕过渲染直击数据源头打开京东商品页如https://item.jd.com/100012043978.html按F12打开开发者工具切换到Network → XHR标签刷新页面筛选关键词price。你会看到一个类似https://p.3.cn/prices/mgets?skuIdsJ_100012043978origin2的请求。这就是价格接口。注意它的三个关键特征skuIds参数是J_开头的SKU ID不是URL里的数字ID京东URL中100012043978需转为J_100012043978origin2表示来自商品详情页origin1是搜索页不同来源返回字段略有差异请求头必须包含Referer: https://item.jd.com/100012043978.html否则返回{code:610,msg:非法请求}curl https://p.3.cn/prices/mgets?skuIdsJ_100012043978origin2 \ -H Referer: https://item.jd.com/100012043978.html \ -H 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提示不要用Selenium模拟整个页面加载——慢、资源占用高、易被识别。抓包后直接复现API请求效率提升10倍以上且更稳定。2.2 解密价格字段p值不是价格是混淆后的整数字符串接口返回JSON类似[{id:J_100012043978,p:1299.00,m:1399.00,op:1299.00}]其中p是当前售价但注意这不是最终显示价格。京东前端会对此值做二次处理——实际展示时p字段需转换为浮点数再减去0.01这是京东价格展示的固定偏移用于规避“整数价”心理阈值。所以p:1299.00对应页面显示¥1298.99。这个偏移规则在2023年Q4起全站统一无需逆向JS实测有效。2.3 构建最小可行爬取单元只抓价格不登录、不渲染、不截图以下代码封装了上述逻辑支持单SKU和多SKU批量请求京东API允许一次传最多20个SKUimport requests import time import json from urllib.parse import quote def get_jd_price(sku_id: str, timeout5) - dict: 获取单个京东SKU实时价格 :param sku_id: 京东商品ID如 100012043978 :return: {sku: 100012043978, price: 1298.99, market_price: 1399.00, original_price: 1299.00, timestamp: 1717023456} # 构造SKU ID格式J_ 数字 jd_sku fJ_{sku_id} url fhttps://p.3.cn/prices/mgets?skuIds{quote(jd_sku)}origin2 headers { Referer: fhttps://item.jd.com/{sku_id}.html, 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, Accept: application/json, text/javascript, */*; q0.01, } try: resp requests.get(url, headersheaders, timeouttimeout) resp.raise_for_status() data resp.json() if not data or p not in data[0]: return {sku: sku_id, price: None, error: no_price_field} # 解析价格p字段是字符串需转float并减0.01 raw_p float(data[0][p]) display_price round(raw_p - 0.01, 2) return { sku: sku_id, price: display_price, market_price: float(data[0].get(m, 0)), # 市场价 original_price: raw_p, # 原始p值用于比对偏移是否变化 timestamp: int(time.time()), } except Exception as e: return {sku: sku_id, price: None, error: str(e)} # 测试单个SKU result get_jd_price(100012043978) print(result) # 输出{sku: 100012043978, price: 1298.99, market_price: 1399.0, original_price: 1299.0, timestamp: 1717023456}这段代码的关键设计点不依赖登录态京东价格接口对未登录用户开放只要Referer合法无浏览器驱动避免Selenium的启动开销和内存泄漏错误兜底明确返回error字段而非抛异常便于后续重试逻辑时间戳精确到秒为后续趋势分析提供可靠时间锚点。3. 数据持久化为什么不用CSV而选SQLiteSQLAlchemy且必须加唯一约束价格数据不是一次性的快照而是时间序列。每天抓100个SKU一年就是365×10036500条记录。用CSV存第3次打开文件就卡死用MySQL小项目没必要搭服务。SQLite是黄金选择单文件、零配置、ACID事务、支持窗口函数用于计算7日均价。但直接用sqlite3模块手写INSERT当SKU增加、字段扩展、并发写入时你会掉进锁表、主键冲突、类型隐式转换的坑里。SQLAlchemy ORM不是炫技是给数据加“保险丝”。3.1 定义价格数据模型字段设计决定分析自由度from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Index, UniqueConstraint from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime Base declarative_base() class JdPriceRecord(Base): __tablename__ jd_price_history id Column(Integer, primary_keyTrue, autoincrementTrue) sku Column(String(32), nullableFalse) # 京东SKU如 100012043978 price Column(Float, nullableFalse) # 显示价格已减0.01 market_price Column(Float, default0.0) # 市场参考价 original_price Column(Float, default0.0) # 原始p值用于监控偏移变化 timestamp Column(DateTime, defaultdatetime.now, indexTrue) # 复合唯一约束同一SKU在同一秒内只存一条避免重复抓取 __table_args__ ( UniqueConstraint(sku, timestamp, nameuq_sku_timestamp), Index(ix_sku_time, sku, timestamp), ) # 初始化数据库 engine create_engine(sqlite:///jd_price.db, echoFalse) # echoTrue 查看SQL语句 Base.metadata.create_all(engine) Session sessionmaker(bindengine) session Session()注意UniqueConstraint(sku, timestamp)是救命设计。京东价格接口可能在1秒内返回相同结果若不做去重一天内同一SKU会存入几十条重复记录后续分析全乱。3.2 批量插入与冲突忽略用on_conflict_do_nothing防爆库爬虫常因网络抖动重试导致同一批数据多次写入。SQLite原生不支持ON CONFLICT语法但SQLAlchemy 1.4通过insert().on_conflict_do_nothing()模拟from sqlalchemy.dialects.sqlite import insert def bulk_save_prices(price_list: list): 批量保存价格记录自动忽略重复skutimestamp相同 :param price_list: [{sku: 100012043978, price: 1298.99, ...}, ...] stmt insert(JdPriceRecord).values(price_list) # 如果存在相同skutimestamp则跳过 do_nothing_stmt stmt.on_conflict_do_nothing( index_elements[sku, timestamp] ) session.execute(do_nothing_stmt) session.commit() # 示例保存3个SKU的价格 prices [ {sku: 100012043978, price: 1298.99, market_price: 1399.0, original_price: 1299.0}, {sku: 100023456789, price: 89.99, market_price: 99.0, original_price: 90.0}, {sku: 100034567890, price: 299.0, market_price: 329.0, original_price: 299.0}, ] bulk_save_prices(prices)3.3 为什么不用MongoDB或JSON文件MongoDB文档模型适合嵌套结构但价格是扁平表格且SQLite的GROUP BY window function做同比环比更直观JSON文件无法索引、无法原子写入、并发写入易损坏CSV无事务、无类型校验、无查询能力——你总不能用pandas.read_csv().groupby().rolling().mean()实时算7日均值吧而SQLite一句SELECT AVG(price) OVER (ORDER BY timestamp ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) FROM jd_price_history WHERE sku100012043978就搞定。4. 定时调度与异常自愈为什么APScheduler比Crontab更适合Python爬虫守护把爬虫脚本丢进Linux crontab看似省事实则埋雷crontab只管“按时启动”不管“是否成功”、“是否卡死”、“是否重复运行”。曾见某团队crontab每5分钟跑一次结果某次网络超时导致进程hang住3小时后系统积压36个僵尸进程CPU 100%连ssh都登不上。APScheduler是Python生态里最成熟的调度器它能把“定时”、“重试”、“熔断”、“状态监控”全收口在一个对象里。4.1 配置APScheduler内存JobStore 错误回调 最大并发限制from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.executors.pool import ThreadPoolExecutor from apscheduler.jobstores.memory import MemoryJobStore import logging # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) # 调度器配置 executors { default: ThreadPoolExecutor(3) # 最大并发3个抓取任务防IP被封 } job_defaults { coalesce: False, # 不合并错过的任务 max_instances: 1, # 同一任务最多1个实例运行 misfire_grace_time: 60 # 任务错过60秒内仍执行 } scheduler BackgroundScheduler( executorsexecutors, job_defaultsjob_defaults, timezoneAsia/Shanghai ) def crawl_and_save_job(): 核心任务抓取配置的SKU列表存入数据库 sku_list [100012043978, 100023456789, 100034567890] # 实际应从配置文件读 price_data [] for sku in sku_list: try: result get_jd_price(sku) if result[price] is not None: result[timestamp] datetime.fromtimestamp(result[timestamp]) price_data.append(result) else: logger.warning(fSKU {sku} 抓取失败: {result[error]}) except Exception as e: logger.error(fSKU {sku} 抓取异常: {e}) if price_data: bulk_save_prices(price_data) logger.info(f成功保存 {len(price_data)} 条价格记录) # 添加定时任务每15分钟执行一次 scheduler.add_job( funccrawl_and_save_job, triggerinterval, minutes15, idjd_price_crawler, name京东价格爬取任务, replace_existingTrue ) # 启动调度器 scheduler.start() logger.info(京东价格爬虫调度器已启动) # 程序保持运行实际部署用systemd或supervisor守护 try: while True: time.sleep(3600) # 每小时检查一次 except (KeyboardInterrupt, SystemExit): scheduler.shutdown()4.2 关键防护机制说明max_instances1防止因上次任务未结束下次又触发导致并发写入冲突misfire_grace_time60若某次任务因服务器重启错过60秒内补跑保证数据连续性ThreadPoolExecutor(3)严格限制并发数京东对单IP每分钟请求有隐性阈值实测5次/分易触发滑块replace_existingTrue修改代码重启后自动覆盖旧任务避免重复注册。提示生产环境务必用systemd或supervisor守护此脚本确保崩溃后自动拉起。别信while True——它扛不住kill -9。5. 价格波动分析与竞品对比用Pandas窗口函数挖出“隐藏调价信号”存下数据只是开始。真正的价值在分析什么时候降价最狠哪个竞品调价最频繁你的商品价格在同类中处于什么分位这些不能靠肉眼盯Excel得用代码“问数据库”。5.1 计算7日价格波动率识别主动调价与被动跟涨波动率不是标准差而是相对变化幅度的绝对值之和。例如某SKU过去7天价格[1299, 1299, 1299, 1259, 1259, 1259, 1259]最后4天降40元波动率应显著高于一直平稳的SKU。import pandas as pd import sqlite3 def calc_volatility(sku: str, days: int 7) - float: 计算指定SKU最近N天的价格波动率 波动率 sum(|price[i] - price[i-1]| / price[i-1]) for i in 1..N conn sqlite3.connect(jd_price.db) query f SELECT price, timestamp FROM jd_price_history WHERE sku ? AND timestamp datetime(now, -{days} days) ORDER BY timestamp df pd.read_sql_query(query, conn, params(sku,)) conn.close() if len(df) 2: return 0.0 # 计算相邻日价格变化率绝对值 df[change_rate] df[price].pct_change().abs() return df[change_rate].sum() # 示例计算SKU 100012043978的7日波动率 vol calc_volatility(100012043978, days7) print(f波动率: {vol:.4f}) # 如 0.0307 → 3.07%5.2 竞品价格分位对比你的定价是在“高端区”还是“地板价”假设你监控5个竞品SKU包括自己想看当前价格在群体中的位置def get_competitor_percentile(sku_list: list, target_sku: str) - dict: 获取目标SKU在竞品群中的价格分位数0-100 conn sqlite3.connect(jd_price.db) placeholders ,.join([?] * len(sku_list)) query f SELECT sku, price, timestamp FROM jd_price_history WHERE sku IN ({placeholders}) AND timestamp (SELECT MAX(timestamp) FROM jd_price_history WHERE sku IN ({placeholders})) df pd.read_sql_query(query, conn, paramssku_list*2) # 参数绑定需重复 conn.close() if df.empty: return {percentile: None, all_prices: []} # 取每个SKU最新一条记录按timestamp最大 latest df.loc[df.groupby(sku)[timestamp].idxmax()] prices latest[price].tolist() target_price latest[latest[sku] target_sku][price].iloc[0] percentile (sum(p target_price for p in prices) / len(prices)) * 100 return { percentile: round(percentile, 1), # 如 60.0 → 高于60%竞品 all_prices: sorted(prices), target_price: target_price } # 示例竞品群 [自己, A, B, C, D] result get_competitor_percentile( [100012043978, 100023456789, 100034567890, 100045678901, 100056789012], 100012043978 ) print(result) # 输出{percentile: 80.0, all_prices: [89.99, 299.0, 1259.0, 1298.99, 1399.0], target_price: 1298.99}5.3 可视化价格趋势Matplotlib极简三行出图不用Flask搭Web先用脚本生成趋势图验证逻辑import matplotlib.pyplot as plt def plot_price_trend(sku: str, days: int 30): conn sqlite3.connect(jd_price.db) query f SELECT timestamp, price FROM jd_price_history WHERE sku ? AND timestamp datetime(now, -{days} days) ORDER BY timestamp df pd.read_sql_query(query, conn, params(sku,)) conn.close() if df.empty: print(fSKU {sku} 近{days}天无数据) return plt.figure(figsize(12, 5)) plt.plot(pd.to_datetime(df[timestamp]), df[price], markero, markersize2) plt.title(fSKU {sku} 近{days}天价格趋势) plt.xlabel(日期) plt.ylabel(价格¥) plt.grid(True, alpha0.3) plt.xticks(rotation45) plt.tight_layout() plt.savefig(fprice_trend_{sku}_{days}d.png, dpi200) plt.show() plot_price_trend(100012043978, days30)这张图能立刻暴露问题若出现大量水平线段 → 抓取频率不足或接口失效若出现突兀尖峰 → 可能是京东临时补贴或秒杀价需在分析时过滤若价格阶梯式下降 → 可能是平台活动节奏如“每满300减50”档位变化。6. 避坑指南这5个血泪经验让我重写了四版爬虫才跑通爬京东价格不是技术难题而是工程细节的集合。以下是我踩过的真坑每一条都附带现场日志和修复方案不是理论推测。6.1 现象抓取返回{code:610,msg:非法请求}但Headers完全复制抓包内容原因Referer末尾多了斜杠/。抓包看到Referer: https://item.jd.com/100012043978.html/多了一个/而实际URL是https://item.jd.com/100012043978.html。京东后端校验Referer域名路径末尾斜杠不匹配即判非法。解决用urllib.parse.urlparse()标准化Referer确保与商品页URL完全一致from urllib.parse import urlparse url_obj urlparse(fhttps://item.jd.com/{sku}.html) clean_referer f{url_obj.scheme}://{url_obj.netloc}{url_obj.path}6.2 现象价格突然全部变成1299.00原始p值不再减0.01原因京东在2024年3月灰度上线新价格渲染逻辑部分SKU开始返回p字段为真实显示价即已减0.01而老SKU仍需手动减。混合抓取时有的减、有的不减导致数据混乱。解决动态检测偏移。对每个SKU首次抓取时存下p和display_price后续对比若p display_price则标记该SKU为“免偏移”否则继续减0.01。用数据库加offset_mode字段记录ALTER TABLE jd_price_history ADD COLUMN offset_mode INTEGER DEFAULT 0; -- 0需减0.01, 1免偏移6.3 现象SQLite写入报database is locked且持续数分钟原因APScheduler的ThreadPoolExecutor并发写入而SQLite默认WAL模式未开启写操作阻塞读。解决初始化引擎时强制启用WALengine create_engine( sqlite:///jd_price.db, connect_args{check_same_thread: False}, echoFalse ) with engine.connect() as conn: conn.execute(PRAGMA journal_modeWAL;)6.4 现象get_jd_price()偶尔返回None但日志没报错原因京东价格接口在流量高峰时如晚8点会返回HTTP 200但body为空字符串resp.json()抛JSONDecodeError被外层except Exception吞掉返回None。解决显式捕获JSONDecodeError并记录import json try: data resp.json() except json.JSONDecodeError as e: logger.error(f价格接口返回空JSON: {resp.text[:100]}) return {sku: sku_id, price: None, error: empty_json_response}6.5 现象bulk_save_prices()后查数据库发现部分记录缺失原因session.execute()后未session.commit()或session被意外关闭。SQLAlchemy Session不是线程安全的APScheduler多线程下必须为每个任务创建独立Session。解决在任务函数内新建Session用完即关def bulk_save_prices(price_list: list): local_session Session() # 新建Session try: stmt insert(JdPriceRecord).values(price_list) local_session.execute(stmt.on_conflict_do_nothing(...)) local_session.commit() finally: local_session.close() # 必须关闭7. 进阶技巧用价格波动率触发自动化预警让运营提前2小时收到调价通知系统跑通后真正的价值在于“让数据自己说话”。我给这套系统加的最后一道工序是价格异动预警——不是等运营每天早上翻报表而是当竞品突然降价5%自动发企业微信消息。7.1 定义“异动”波动率 时间窗口 竞品关联单纯看单SKU波动率不够。比如某SKU常年价格战波动率天生高而另一SKU平时纹丝不动某天降10%才是真信号。所以预警规则是条件1该SKU近24小时波动率 3%基线条件2且其所在类目TOP3竞品中≥2个同步波动率 2%条件3排除早8点-10点京东日常补货价更新时段def check_price_alert(sku: str) - dict: 检查SKU是否触发价格预警 返回 {alert: True, reason: 竞品A/B同步降价, details: {...}} 或 {alert: False} # 获取该SKU近24小时波动率 vol_24h calc_volatility(sku, days1) if vol_24h 0.03: return {alert: False} # 获取同类目竞品此处简化硬编码竞品列表实际应查品类树 competitors get_competitor_list_by_category(sku) # 自定义函数 comp_vols [calc_volatility(c, days1) for c in competitors] # 统计竞品中波动率2%的数量 active_comps sum(1 for v in comp_vols if v 0.02) if active_comps 2: return { alert: True, reason: f竞品{competitors[:2]}同步降价, details: { self_volatility: round(vol_24h, 4), competitor_vols: [round(v, 4) for v in comp_vols], timestamp: datetime.now().isoformat() } } return {alert: False} # 在APScheduler任务末尾调用 def crawl_and_save_job(): # ... 抓取与存储逻辑 ... # 检查预警 alert check_price_alert(100012043978) if alert[alert]: send_wechat_alert(alert) # 企业微信机器人推送7.2 企业微信机器人推送5行代码接入内部通讯京东价格变动是运营敏感事件必须直达责任人。企业微信机器人是最轻量级方案import requests import json def send_wechat_alert(alert_data: dict): 发送企业微信机器人消息 webhook_url从企业微信后台获取需配置在环境变量 webhook https://qyapi.weixin.qq.com/xxx # 替换为你的机器人地址 payload { msgtype: text, text: { content: f 价格异动预警\nSKU: {alert_data[details][sku]}\n原因: {alert_data[reason]}\n自身波动率: {alert_data[details][self_volatility]*100:.1f}%\n时间: {alert_data[details][timestamp][:19]} } } requests.post(webhook, jsonpayload) # 效果企业微信收到富文本消息带emoji和换行运营一眼看清重点7.3 为什么不做“自动调价”我见过太多团队想接自动调价——爬到竞品降价自己立刻跟降。结果发现京东价格接口有缓存你看到的“降价”可能是10分钟前的旧数据竞品降价可能是清仓甩卖你的库存成本根本撑不住自动调价没有审批流一旦出错就是资损。所以我的原则是系统只负责“看见”不负责“行动”。预警消息里必须带一句“请运营确认是否跟进勿直接调价”。这句提示救过我们两次。这套系统上线半年运营同学说“现在我不用守着电脑盯价格了手机弹个消息喝杯咖啡的时间就决策完。”——这才是技术该有的样子不炫技不堆概念就扎扎实实把一件事做透。希望帮到你。本文还有配套的精品资源点击获取