京东商品价格爬取与分析系统：API直采+SQLite存储+APScheduler调度

简介：这是一套面向电商数据分析初学者与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 class="price">，返回空；你用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?skuIds=J_100012043978&origin=2的请求。这就是价格接口。注意它的三个关键特征：

• skuIds参数是J_开头的SKU ID，不是URL里的数字ID（京东URL中100012043978需转为J_100012043978）
• origin=2表示来自商品详情页，origin=1是搜索页，不同来源返回字段略有差异
• 请求头必须包含Referer: https://item.jd.com/100012043978.html，否则返回{"code":610,"msg":"非法请求"}

curl 'https://p.3.cn/prices/mgets?skuIds=J_100012043978&origin=2' \ -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个SKU）：

import requests import time import json from urllib.parse import quote def get_jd_price(sku_id: str, timeout=5) -> 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 = f"J_{sku_id}" url = f"https://p.3.cn/prices/mgets?skuIds={quote(jd_sku)}&origin=2" headers = { "Referer": f"https://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, */*; q=0.01", } try: resp = requests.get(url, headers=headers, timeout=timeout) 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而选SQLite+SQLAlchemy，且必须加唯一约束

价格数据不是一次性的快照，而是时间序列。每天抓100个SKU，一年就是365×100=36500条记录。用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_key=True, autoincrement=True) sku = Column(String(32), nullable=False) # 京东SKU，如 "100012043978" price = Column(Float, nullable=False) # 显示价格（已减0.01） market_price = Column(Float, default=0.0) # 市场参考价 original_price = Column(Float, default=0.0) # 原始p值（用于监控偏移变化） timestamp = Column(DateTime, default=datetime.now, index=True) # 复合唯一约束：同一SKU在同一秒内只存一条，避免重复抓取 __table_args__ = ( UniqueConstraint('sku', 'timestamp', name='uq_sku_timestamp'), Index('ix_sku_time', 'sku', 'timestamp'), ) # 初始化数据库 engine = create_engine("sqlite:///jd_price.db", echo=False) # echo=True 查看SQL语句 Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) 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): """ 批量保存价格记录，自动忽略重复（sku+timestamp相同） :param price_list: [{"sku": "100012043978", "price": 1298.99, ...}, ...] """ stmt = insert(JdPriceRecord).values(price_list) # 如果存在相同sku+timestamp，则跳过 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 sku='100012043978'就搞定。

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(level=logging.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( executors=executors, job_defaults=job_defaults, timezone='Asia/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(f"SKU {sku} 抓取失败: {result['error']}") except Exception as e: logger.error(f"SKU {sku} 抓取异常: {e}") if price_data: bulk_save_prices(price_data) logger.info(f"成功保存 {len(price_data)} 条价格记录") # 添加定时任务：每15分钟执行一次 scheduler.add_job( func=crawl_and_save_job, trigger='interval', minutes=15, id='jd_price_crawler', name='京东价格爬取任务', replace_existing=True ) # 启动调度器 scheduler.start() logger.info("京东价格爬虫调度器已启动") # 程序保持运行（实际部署用systemd或supervisor守护） try: while True: time.sleep(3600) # 每小时检查一次 except (KeyboardInterrupt, SystemExit): scheduler.shutdown()

4.2 关键防护机制说明

• max_instances=1：防止因上次任务未结束，下次又触发，导致并发写入冲突；
• misfire_grace_time=60：若某次任务因服务器重启错过，60秒内补跑，保证数据连续性；
• ThreadPoolExecutor(3)：严格限制并发数，京东对单IP每分钟请求有隐性阈值（实测＞5次/分易触发滑块）；
• replace_existing=True：修改代码重启后，自动覆盖旧任务，避免重复注册。

提示：生产环境务必用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", days=7) 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, params=sku_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(f"SKU {sku} 近{days}天无数据") return plt.figure(figsize=(12, 5)) plt.plot(pd.to_datetime(df["timestamp"]), df["price"], marker="o", markersize=2) plt.title(f"SKU {sku} 近{days}天价格趋势") plt.xlabel("日期") plt.ylabel("价格（¥）") plt.grid(True, alpha=0.3) plt.xticks(rotation=45) plt.tight_layout() plt.savefig(f"price_trend_{sku}_{days}d.png", dpi=200) plt.show() plot_price_trend("100012043978", days=30)

这张图能立刻暴露问题：

• 若出现大量水平线段 → 抓取频率不足或接口失效；
• 若出现突兀尖峰 → 可能是京东临时补贴或秒杀价，需在分析时过滤；
• 若价格阶梯式下降 → 可能是平台活动节奏（如“每满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(f"https://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模式未开启，写操作阻塞读。
解决：初始化引擎时强制启用WAL：

engine = create_engine( "sqlite:///jd_price.db", connect_args={"check_same_thread": False}, echo=False ) with engine.connect() as conn: conn.execute("PRAGMA journal_mode=WAL;")

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, days=1) if vol_24h < 0.03: return {"alert": False} # 获取同类目竞品（此处简化：硬编码竞品列表，实际应查品类树） competitors = get_competitor_list_by_category(sku) # 自定义函数 comp_vols = [calc_volatility(c, days=1) 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, json=payload) # 效果：企业微信收到富文本消息，带emoji和换行，运营一眼看清重点

7.3 为什么不做“自动调价”？

我见过太多团队想接自动调价——爬到竞品降价，自己立刻跟降。结果发现：

• 京东价格接口有缓存，你看到的“降价”可能是10分钟前的旧数据；
• 竞品降价可能是清仓甩卖，你的库存成本根本撑不住；
• 自动调价没有审批流，一旦出错就是资损。

所以我的原则是：系统只负责“看见”，不负责“行动”。预警消息里必须带一句：“请运营确认是否跟进，勿直接调价”。这句提示，救过我们两次。

这套系统上线半年，运营同学说：“现在我不用守着电脑盯价格了，手机弹个消息，喝杯咖啡的时间就决策完。”——这才是技术该有的样子：不炫技，不堆概念，就扎扎实实把一件事做透。希望帮到你。

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