京东商品价格波动怎么监控？Python+接口方案完整源码

盯竞品价格这件事，手动刷新撑不过半小时：你不知道它什么时候降价、降了多少、是不是历史新低，等你刷到的时候，库存可能已经没了。

这篇文章给一套能直接跑的方案：接口定时取价 → SQLite 存快照 → 规则判断波动 → 群机器人推送告警，外加一条离线价格曲线。全部只用 Python 标准库，pip install 一步省掉。

一、方案总览

┌────────────┐   轮询(30min)   ┌──────────────┐   规则判断   ┌────────────┐
│ 京东价格接口 │ ──────────────→ │ SQLite 快照库 │ ──────────→ │ 告警推送     │
│ item_price │                 │ snapshots    │             │ 企微/钉钉/飞书│
└────────────┘                 └──────────────┘             └────────────┘
        ↑                             │
   item_get 补详情                本地统计 + HTML 价格曲线
   item_history_price 查历史低价

核心思路只有四步：定时取价 → 存快照 → 和上一条/历史对比 → 触发规则就推送。难点全在细节里：取价用哪个接口、-1 怎么处理、怎么避免同一件事反复骚扰群。

三个接口怎么分工

item_price 是监控场景的最优解：单次调用只回三个字段，同样的套餐额度能盯更多商品；商品上下架、改价它都第一时间反映。

公共参数三个接口一致：key、secret（必须拼在 URL query 上）、cache（监控必须传 no，原因见后文坑 3）、result_type=json。item_get 额外支持 domain_type=jd（自营）/ gongye（工业）/ yiyao（医药）/ hk（全球购）。

二、精简版源码：60 行跑通闭环

先给一个最小可用版：轮询 + 快照 + 三条告警规则 + 群机器人推送。复制改两个 SKU 就能跑。

# -*- coding: utf-8 -*-
"""京东价格监控精简版：item_price 轮询 + SQLite 快照 + 阈值告警"""
import json
import os
import sqlite3
import time
import urllib.parse
import urllib.request
from datetime import datetime

API = "https://api-gw.onebound.cn/jd/item_price/"
KEY, SECRET = os.getenv("ONEBOUND_KEY", ""), os.getenv("ONEBOUND_SECRET", "")
WEBHOOK = os.getenv("PRICE_WEBHOOK", "")          # 企业微信/钉钉群机器人地址，可留空

WATCH = {
                                             # 监控清单：sku -> (名称, 目标价)
    "100012345678": ("竞品A·蓝牙耳机", 199),
    "100045678901": ("竞品B·机械键盘", None),
}
DROP_PCT = 5                                       # 单次波动告警阈值 %
INTERVAL = 1800                                    # 轮询间隔秒

conn = sqlite3.connect("price.db")
conn.execute("CREATE TABLE IF NOT EXISTS snap(sku TEXT, ts TEXT, price REAL, UNIQUE(sku, ts))")


def fetch(sku):
    """取当前价。接口约定：price = -1 表示商品下架或不存在。"""
    url = f"{API}?{urllib.parse.urlencode({'key': KEY, 'secret': SECRET, 'num_iid': sku, 'cache': 'no'})}"
    with urllib.request.urlopen(url, timeout=15) as r:
        item = json.loads(r.read().decode("utf-8")).get("item") or {
   }
    return float(item.get("price") or 0)


def notify(text):
    print(text)
    if WEBHOOK:                                    # 企业微信/钉钉机器人都是这个格式
        req = urllib.request.Request(WEBHOOK, data=json.dumps(
            {
   "msgtype": "text", "text": {
   "content": text}}).encode(),
            headers={
   "Content-Type": "application/json"})
        urllib.request.urlopen(req, timeout=8).read()


def check(sku, name, target):
    price = fetch(sku)
    now = datetime.now().strftime("%Y-%m-%d %H:%M")
    prev = conn.execute("SELECT price FROM snap WHERE sku=? ORDER BY ts DESC LIMIT 1",
                        (sku,)).fetchone()
    conn.execute("INSERT OR REPLACE INTO snap VALUES(?,?,?)", (sku, now, price))
    conn.commit()
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    if price == -1:
        notify(f"[下架] {name}({sku}) 接口返回 -1，商品下架或不存在")
        return
    if prev:
        pct = (price - prev[0]) / prev[0] * 100
        if abs(pct) >= DROP_PCT:
            word = "降价" if pct < 0 else "涨价"
            notify(f"[{word} {abs(pct):.1f}%] {name}({sku}) ¥{prev[0]:.2f} → ¥{price:.2f}")
    if target and price <= target:
        notify(f"[到目标价] {name}({sku}) 现价 ¥{price:.2f} ≤ 目标 ¥{target}")


if __name__ == "__main__":
    while True:
        for sku, (name, target) in WATCH.items():
            try:
                check(sku, name, target)
            except Exception as e:
                print(f"[{sku}] 取价失败：{e}")
            time.sleep(1)                          # 商品间隔，避免瞬间并发超限
        time.sleep(INTERVAL)

实测触发效果（用模拟价格序列验证）：

[降价 6.8%] 竞品A·蓝牙耳机(100012345678) ¥220.00 → ¥205.00
[降价 7.8%] 竞品A·蓝牙耳机(100012345678) ¥205.00 → ¥189.00
[到目标价] 竞品A·蓝牙耳机(100012345678) 现价 ¥189.00 ≤ 目标 ¥199
[下架] 竞品A·蓝牙耳机(100012345678) 接口返回 -1，商品下架或不存在

注意两个写法细节：

1. 先查上一条，再落库。顺序反了的话，同一分钟内重跑会把自己当成"上一条"，涨跌幅永远算不出来。
2. price = -1 是合法状态，不是错误。接口文档明确：当值为 -1 时商品 ID 视为下架或不存在。把它当 0 处理，你的报表里会出现"暴跌 100%"。

三、工程版：多商品 + 告警冷却 + 价格曲线

精简版跑几天你就会遇到两个新问题：商品一多告警刷屏；想看趋势只能查库。文末附的 jd_price_monitor.py 在精简版之上补齐了这些：

取数层  item_price(轻量轮询) / item_get --full(详情)   指数退避重试
判断层  四条规则：降价≥5% / 涨价≥5% / 到目标价 / 历史新低 / 下架
        同规则 6 小时冷却，降价已报则不再重复报历史新低
存储层  SQLite 三张表：watchlist / snapshots / alert_log
报表层  report 命令（最低/最高/涨跌/采样数）+ chart 命令生成离线 HTML 曲线

用 seed 造 30 天数据后，chart 的输出长这样（演示数据）：

曲线自动标注区间最高/最低点和目标价虚线，发群里比截图数据库直观得多。

四、六个真实会踩的坑

1. price 字段"会对不上"是文档明说的

item_get 的返回说明里有一句原注：当前商品 id 价格【价格会对不上】。多 SKU 商品的主价格只是默认 SKU 的价格。监控到单品粒度，用 item_price 传对应 SKU 的 ID，或者解析 skus.sku[]。

2. -1、0、空值要分三种情况处理

-1 = 下架/不存在；0 或空 = 字段异常，视为本次取价失败，不要落库——否则历史最低价会被污染，"历史新低"规则从此失灵。

3. 监控必须 cache=no

接口默认走缓存（速度快），但监控要的是变化，缓存期内价格根本没更新。日常查询用缓存省配额，监控轮询一律关。

4. 判断成功别只看 HTTP 200

接口出错也返回 200，必须判 error_code == "0000"。高频错误码：

5. 告警不做冷却，群会被你炸掉

降价 5% 和历史新低经常是同一件事，两条都发就是刷屏。工程版做了两层去重：同规则 6 小时冷却（alert_log 表），降价已触发就不再报历史新低。大促期间可以调小冷却，平时宁可少打扰。

6. 轮询节奏要跟着价格波动规律走

京东改价集中在 0 点、早 10 点、大促节点。均匀 30 分钟一轮，90% 的请求都浪费在没变化的深夜。更划算的做法是分级轮询：

平时（0:00-8:00）    60 分钟一轮
白天（8:00-22:00）   30 分钟一轮
大促/预热期          5 分钟一轮（单独起一个高优先级清单）

额度是否扛得住看响应里的 api_info 字段（形如 today:28 max:10000），每天先读一次做预算，快打满就自动降频。

五、部署：一台最小机器就够

数据量不大时，一台最低配云主机 + SQLite + crontab 就能稳定跑，不需要 MySQL 和消息队列：

# crontab：每 30 分钟抓一轮，日志按天切
*/30 * * * * cd /opt/jd-monitor && /usr/bin/python3 jd_price_monitor.py once >> logs/run.log 2>&1
# 每天 8 点推送报表链接
0 8 * * * cd /opt/jd-monitor && python3 jd_price_monitor.py report --days 1 && python3 jd_price_monitor.py chart --days 7

Windows 下用任务计划程序建两个基本相同的任务即可。想升级再说：Redis 缓存、MySQL、按类目批量监控，都是在 fetch() 和存储层上加东西，主流程不动。

合规提醒：走数据接口获取的是公开商品信息，注意控制频率、遵守服务商与平台的使用条款；监控数据用于自身选品和比价决策，不要转售原始数据或爬取用户个人信息。

结语

价格监控的技术含量不在"取到价格"，而在三件事：把 -1、空值、缓存这些边角处理干净；告警规则宁缺毋滥；数据攒起来之后能变成趋势图。

先跑通精简版确认数据对得上，再换工程版把冷却、报表、曲线补齐，一条稳定的盯价管道就成型了。之后要做比价站、降价订阅、价格弹性分析，都是在这条管道上加业务而已。