在电商运营中，商品价格是影响转化率和竞争力的核心因素。手动跟踪竞品价格效率低下，而基于 API 构建的自动化价格监控系统能实时捕捉价格波动，为定价策略提供数据支撑。本文将从零开始，带您实现一套覆盖京东、淘宝双平台的商品价格监控系统，包含数据采集、存储、分析、告警全流程。
一、系统架构设计
价格监控系统需实现 “实时采集 - 时序存储 - 波动分析 - 智能告警” 的闭环，整体架构如下：

核心模块说明：

• 数据采集层：调用京东 / 淘宝 API，定时获取目标商品价格；
• 数据清洗层：处理空值、格式转换、去重，确保数据一致性；
• 时序存储层：存储价格历史数据（重点保留时间戳 + 价格字段）；
• 分析引擎：计算价格波动率、同比 / 环比，识别异常波动；
• 告警系统：基于阈值（如降价 5%）触发通知（邮件 / 钉钉）；
• 可视化看板：展示价格趋势、竞品对比图表。

二、前置准备：API 接入与环境配置

1. 平台 API 权限申请

平台	所需凭证	核心接口	申请要点
京东	AppKey + AppSecret	jingdong.ware.price.get	个人开发者可申请 “商品价格查询” 权限
淘宝	AppKey + AppSecret + Token	taobao.item.price.get	需通过淘宝联盟备案，获取session

2. 开发环境配置

• 编程语言：Python 3.8+（推荐）

• 核心库：

  • requests/aiohttp：API 请求（异步推荐aiohttp）；
  • pymongo：MongoDB 数据库交互；
  • schedule：定时任务调度；
  • matplotlib：价格趋势可视化；
  • dingtalkchatbot：钉钉告警。

• 环境安装：

  bash

  pip install requests pymongo schedule matplotlib dingtalkchatbot
  

三、核心功能实现：从 API 调用到数据存储

1. 多平台 API 价格采集模块

京东 API 价格采集（含签名生成）

python

运行

import requests
import hashlib
import time

class JDPriceCollector:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = "https://api.jd.com/routerjson"
    
    def generate_sign(self, params):
        """生成京东API签名"""
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        sign_str = self.app_secret
        for k, v in sorted_params:
            sign_str += f"{k}{v}"
        sign_str += self.app_secret
        return hashlib.md5(sign_str.encode()).hexdigest().upper()
    
    def get_price(self, sku_id):
        """获取单个商品价格"""
        params = {
            "app_key": self.app_key,
            "method": "jingdong.ware.price.get",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "skuId": sku_id,
            "sign_method": "md5"
        }
        # 生成签名
        params["sign"] = self.generate_sign(params)
        
        try:
            response = requests.post(self.base_url, data=params, timeout=10)
            result = response.json()
            # 解析价格（京东API返回结构需按实际调整）
            price = result.get("jingdong_ware_price_get_response", {}) \
                        .get("price", {}) \
                        .get("p", 0)  # "p"为实际售价字段
            return {
                "platform": "jd",
                "sku_id": sku_id,
                "price": float(price),
                "crawl_time": time.time()
            }
        except Exception as e:
            print(f"京东API调用失败：{e}")
            return None

淘宝 API 价格采集（简化版）

python

运行

class TaobaoPriceCollector:
    def __init__(self, app_key, app_secret, session):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = session  # 用户授权session
        self.base_url = "https://eco.taobao.com/router/rest"
    
    def generate_sign(self, params):
        """淘宝签名生成（逻辑类似京东，需注意编码）"""
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        sign_str = self.app_secret
        for k, v in sorted_params:
            sign_str += f"{k}{v}"
        sign_str += self.app_secret
        return hashlib.md5(sign_str.encode()).hexdigest().upper()
    
    def get_price(self, num_iid):
        """获取淘宝商品价格"""
        params = {
            "app_key": self.app_key,
            "method": "taobao.item.price.get",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "num_iid": num_iid,
            "session": self.session,
            "sign_method": "md5"
        }
        params["sign"] = self.generate_sign(params)
        
        try:
            response = requests.post(self.base_url, data=params, timeout=10)
            result = response.json()
            price = result.get("item_price_get_response", {}) \
                        .get("price", {}) \
                        .get("price", 0)  # 淘宝售价字段
            return {
                "platform": "taobao",
                "sku_id": num_iid,
                "price": float(price),
                "crawl_time": time.time()
            }
        except Exception as e:
            print(f"淘宝API调用失败：{e}")
            return None

2. 数据存储设计：时序数据库选型与实现

价格数据是典型的时序数据（按时间戳有序生成），需支持高写入、时序查询（如 “过去 7 天价格”）。推荐使用MongoDB（灵活）或InfluxDB（专为时序优化）。

MongoDB 存储实现

python

运行

from pymongo import MongoClient
from datetime import datetime

class PriceStorage:
    def __init__(self, db_name="price_monitor"):
        self.client = MongoClient("mongodb://localhost:27017/")
        self.db = self.client[db_name]
        # 创建价格历史集合（按平台+SKU索引，加速查询）
        self.price_collection = self.db["price_history"]
        self.price_collection.create_index([("platform", 1), ("sku_id", 1), ("crawl_time", -1)])
    
    def save_price(self, price_data):
        """存储单条价格数据"""
        # 补充格式化时间（便于可视化）
        price_data["crawl_datetime"] = datetime.fromtimestamp(price_data["crawl_time"])
        self.price_collection.insert_one(price_data)
        print(f"已存储 {price_data['platform']} {price_data['sku_id']} 价格：{price_data['price']}")
    
    def get_price_history(self, platform, sku_id, days=7):
        """获取最近N天价格历史"""
        start_time = time.time() - days * 24 * 3600
        return list(self.price_collection.find({
            "platform": platform,
            "sku_id": sku_id,
            "crawl_time": {"$gte": start_time}
        }).sort("crawl_time", 1))  # 按时间升序

三、核心逻辑：价格监控与波动分析

1. 定时采集任务

使用schedule库实现定时采集，支持多平台、多商品并行监控：

python

运行

import schedule
import time
from concurrent.futures import ThreadPoolExecutor

class PriceMonitor:
    def __init__(self, jd_collector, taobao_collector, storage, interval=30):
        self.jd_collector = jd_collector
        self.taobao_collector = taobao_collector
        self.storage = storage
        self.interval = interval  # 采集间隔（分钟）
        self.monitor_list = {  # 监控商品列表：{平台: [SKU列表]}
            "jd": ["100012345678", "100012345679"],  # 京东SKU
            "taobao": ["598765432101"]  # 淘宝商品ID
        }
    
    def collect_single_sku(self, platform, sku_id):
        """采集单个商品价格"""
        try:
            if platform == "jd":
                price_data = self.jd_collector.get_price(sku_id)
            elif platform == "taobao":
                price_data = self.taobao_collector.get_price(sku_id)
            else:
                return
            
            if price_data:
                self.storage.save_price(price_data)
                # 采集后立即分析价格波动
                self.analyze_price_fluctuation(platform, sku_id)
        except Exception as e:
            print(f"采集 {platform} {sku_id} 失败：{e}")
    
    def batch_collect(self):
        """批量采集所有监控商品"""
        print(f"开始批量采集（{time.ctime()}）")
        with ThreadPoolExecutor(max_workers=5) as executor:  # 并发采集
            for platform, sku_list in self.monitor_list.items():
                for sku_id in sku_list:
                    executor.submit(self.collect_single_sku, platform, sku_id)
        print(f"批量采集完成（{time.ctime()}）")
    
    def start_monitoring(self):
        """启动监控任务"""
        # 立即执行一次
        self.batch_collect()
        # 定时执行
        schedule.every(self.interval).minutes.do(self.batch_collect)
        while True:
            schedule.run_pending()
            time.sleep(60)

2. 价格波动分析：识别异常与趋势

分析模块需计算价格变化率、历史最低 / 最高价，并通过阈值判断是否触发告警： python

运行

class PriceAnalyzer:
    def __init__(self, storage, threshold=0.05):  # 价格波动阈值（5%）
        self.storage = storage
        self.threshold = threshold  # 降价5%以上触发告警
    
    def get_price_change_rate(self, platform, sku_id):
        """计算最近一次价格与上一次的变化率"""
        # 获取最近两条价格记录
        history = self.storage.get_price_history(platform, sku_id, days=1)  # 1天内数据
        if len(history) < 2:
            return 0.0  # 数据不足，不计算
        
        # 按时间排序（旧→新）
        history_sorted = sorted(history, key=lambda x: x["crawl_time"])
        prev_price = history_sorted[-2]["price"]
        current_price = history_sorted[-1]["price"]
        
        # 计算变化率（（当前-上次）/上次）
        if prev_price == 0:
            return 0.0
        return (current_price - prev_price) / prev_price
    
    def analyze_price_fluctuation(self, platform, sku_id):
        """分析价格波动，触发告警"""
        change_rate = self.get_price_change_rate(platform, sku_id)
        if abs(change_rate) >= self.threshold:
            # 获取当前价格
            current_price = self.storage.get_price_history(platform, sku_id, days=1)[-1]["price"]
            # 构建告警信息
            trend = "降价" if change_rate < 0 else "涨价"
            msg = (f"【价格异动告警】\n"
                   f"平台：{platform}\n"
                   f"商品ID：{sku_id}\n"
                   f"当前价格：{current_price}元\n"
                   f"波动幅度：{change_rate*100:.2f}%\n"
                   f"时间：{time.ctime()}")
            # 发送告警
            self.send_alert(msg)
            print(msg)
    
    def send_alert(self, msg):
        """发送告警（支持钉钉/邮件）"""
        # 钉钉告警示例（需提前创建机器人）
        from dingtalkchatbot.chatbot import DingtalkChatbot
        webhook = "https://oapi.dingtalk.com/robot/send?access_token=你的token"
        bot = DingtalkChatbot(webhook)
        bot.send_text(msg=msg, is_at_all=False)  # 不@所有人

四、系统部署与扩展

1. 部署步骤

1. 配置 API 凭证：将京东 / 淘宝的app_key、app_secret等写入环境变量（避免硬编码）；

2. 初始化数据库：启动 MongoDB，创建price_monitor数据库；

3. 启动监控：

   python

   运行

   if __name__ == "__main__":
       # 初始化组件
       jd_collector = JDPriceCollector(app_key="你的京东app_key", app_secret="你的京东app_secret")
       taobao_collector = TaobaoPriceCollector(
           app_key="你的淘宝app_key", 
           app_secret="你的淘宝app_secret",
           session="你的淘宝session"
       )
       storage = PriceStorage()
       analyzer = PriceAnalyzer(storage)
       
       # 启动监控
       monitor = PriceMonitor(jd_collector, taobao_collector, storage, interval=30)
       monitor.start_monitoring()
   

2. 扩展方向

• 多平台支持：接入拼多多、亚马逊 API，实现跨平台价格对比；
• 可视化看板：用 Flask+ECharts 构建实时价格趋势图；
• 智能预测：基于 LSTM 模型预测未来价格走势；
• 竞品自动关联：通过商品标题分词，自动匹配同款竞品。

五、实战避坑指南

问题场景	解决方案
API 调用频繁被限流	1. 分散采集时间（如不同商品错开 1 分钟） 2. 申请更高权限 3. 使用代理 IP 池
价格数据缺失 / 重复	1. 采集时添加重试机制 2. 存储前去重（按 SKU + 时间戳） 3. 标记异常值（如价格 = 0）
告警风暴（频繁触发）	1. 增加 “冷静期”（1 小时内同一商品不重复告警） 2. 分层阈值（如 5%→10%→20% 阶梯告警）
签名错误排查困难	1. 日志记录完整签名参数拼接过程 2. 用官方签名工具验证本地生成的签名

总结

通过本文的实战方案，您可以构建一套稳定的商品价格监控系统，实现从 API 数据采集到智能告警的全自动化。核心价值在于：

1. 实时性：分钟级更新，捕捉转瞬即逝的价格波动；
2. 可扩展性：支持多平台、多商品，随业务需求灵活扩展；
3. 决策支撑：基于数据而非经验制定定价策略，提升竞争力。

后续可重点优化 “竞品自动发现” 和 “价格预测” 功能，让系统从 “被动监控” 升级为 “主动决策辅助”