在电商运营中，选品和价格监控是核心竞争力的来源。手动翻页查价格、看库存不仅效率低，还容易错过爆款时机。本文将从架构设计、API 选型、核心代码实现到生产级部署，完整讲解如何通过 API 接口搭建一套自动化选品与价格监控系统。
一、系统架构设计
一套完整的选品价格监控系统通常分为五层：
表格
层级 职责 关键技术
数据采集层 对接各电商平台 API，定时抓取商品数据 REST API、异步请求、签名认证
数据清洗层 处理空值、格式转换、去重、异常值标记 数据校验、正则匹配
时序存储层 存储价格历史（时间戳 + 价格） MongoDB / InfluxDB / PostgreSQL
分析引擎层 计算价格波动率、同比环比、识别异常 统计分析、阈值判断
告警与可视化 触发通知、展示趋势图、生成选品报表 Webhook、ECharts、Excel 导出
二、API 接入方案选型
2.1 主流平台 API 对比
表格
平台 API 类型 核心接口 认证方式 适用场景
京东 官方 JOS / 第三方 Open Claw item_get、item_search AppKey + 签名 / API Key 自营监控、选品库搭建
淘宝/天猫 淘宝开放平台 taobao.item.price.get AppKey + Token 淘宝联盟选品
亚马逊 Selling Partner API (SP-API) getPricing、getCatalogItem LWA + AWS 签名 跨境竞品分析
拼多多 官方 API / 第三方服务 商品详情接口 签名认证 百亿补贴监控
苏宁易购 开放平台 item_search MD5 签名 全渠道比价
2.2 接入前准备
以京东为例，接入流程如下：
注册开发者账号：在开放平台申请 AppKey 和 AppSecret
申请 API 权限：按需申请最小权限（如仅申请"商品价格查询"）
获取 Access Token：通过 OAuth 2.0 或长期 Refresh Token 获取调用凭证
配置白名单：设置服务器 IP 白名单，增强安全性
三、核心功能实现
3.1 商品数据采集模块
单商品详情获取（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 签名"""
# 1. 参数按 key 排序
sorted_params = sorted(params.items())
# 2. 拼接 key=value
sign_str = ''.join([f"{k}{v}" for k, v in sorted_params])
# 3. 首尾拼接 app_secret
sign_str = f"{self.app_secret}{sign_str}{self.app_secret}"
# 4. MD5 加密转大写
return hashlib.md5(sign_str.encode()).hexdigest().upper()
def get_item_price(self, sku_id):
"""获取单个 SKU 价格"""
params = {
"method": "jingdong.ware.price.get",
"app_key": self.app_key,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"v": "2.0",
"skuId": sku_id
}
params["sign"] = self._generate_sign(params)
try:
resp = requests.get(self.base_url, params=params, timeout=10)
if resp.status_code == 200:
return resp.json()
except Exception as e:
print(f"请求异常: {e}")
return None
使用示例
collector = JDPriceCollector("your_app_key", "your_app_secret")
price_data = collector.get_item_price("100012345678")
关键词搜索选品（批量获取）
def keyword_search(keyword, page=1, page_size=20, api_key="your_key"):
"""按关键词搜索商品，筛选潜力爆款"""
url = "https://api-gw.xxx.cn/jd/item_search"
params = {
"key": api_key,
"q": keyword,
"page": page,
"page_size": page_size,
"sort": "salesCount_desc" # 按销量降序
}
resp = requests.get(url, params=params, timeout=10)
if resp.status_code == 200:
result = resp.json()
items = result.get("items", [])
# 选品过滤逻辑
filtered = []
for item in items:
price = float(item.get("price", 0))
sales = int(item.get("sales", 0))
stock = item.get("stock_status")
# 筛选规则：价格50-200元，销量>30，库存充足
if 50 <= price <= 200 and sales > 30 and stock == "有货":
filtered.append({
"title": item.get("title"),
"price": price,
"sales": sales,
"shop": item.get("shop_name"),
"url": item.get("item_url")
})
return filtered
return []
3.2 价格监控与告警模块
多商品批量监控
import time
import pandas as pd
from datetime import datetime
class PriceMonitor:
def init(self, collector, storage, alert_webhook=None):
self.collector = collector
self.storage = storage
self.alert_webhook = alert_webhook
self.monitor_list = [] # 监控商品列表
def add_monitor(self, sku_id, target_price, platform="jd"):
"""添加监控商品"""
self.monitor_list.append({
"sku_id": sku_id,
"target_price": target_price,
"platform": platform
})
def check_price(self, item):
"""检查单商品价格"""
sku_id = item["sku_id"]
target = item["target_price"]
# 获取实时价格
data = self.collector.get_item_price(sku_id)
if not data:
return None
current_price = float(data.get("price", 9999))
title = data.get("title", "未知商品")
# 存储历史价格
record = {
"platform": item["platform"],
"sku_id": sku_id,
"title": title,
"price": current_price,
"crawl_time": time.time(),
"crawl_datetime": datetime.now()
}
self.storage.save_price(record)
# 价格告警逻辑
if current_price <= target:
self._send_alert(title, current_price, target)
return {"alert": True, "price": current_price, "target": target}
return {"alert": False, "price": current_price}
def _send_alert(self, title, current, target):
"""发送告警通知（支持钉钉/企业微信/邮件）"""
msg = f"🚨 价格达标提醒\n商品：{title}\n当前价：¥{current}\n目标价：¥{target}"
if self.alert_webhook:
requests.post(self.alert_webhook, json={"text": msg})
print(msg)
def start_monitoring(self, interval=600):
"""启动定时监控循环"""
print(f"启动监控，共 {len(self.monitor_list)} 个商品，间隔 {interval//60} 分钟")
while True:
for item in self.monitor_list:
self.check_price(item)
time.sleep(2) # 避免请求过快
time.sleep(interval)
3.3 数据存储设计
价格数据是典型的时序数据，推荐使用 MongoDB 或 InfluxDB：
Python
复制
from pymongo import MongoClient
class PriceStorage:
def init(self, db_name="price_monitor"):
self.client = MongoClient("mongodb://localhost:27017/")
self.db = self.client[db_name]
self.collection = self.db["price_history"]
# 创建复合索引加速查询
self.collection.create_index([
("platform", 1),
("sku_id", 1),
("crawl_time", -1)
])
def save_price(self, price_data):
"""存储单条价格记录"""
self.collection.insert_one(price_data)
def get_price_history(self, platform, sku_id, days=7):
"""获取最近 N 天价格历史"""
start_time = time.time() - days * 24 * 3600
return list(self.collection.find({
"platform": platform,
"sku_id": sku_id,
"crawl_time": {"$gte": start_time}
}).sort("crawl_time", 1))
四、智能选品策略引擎
4.1 多维度筛选模型
基于采集的数据，可构建自动化选品评分体系：
def product_scoring(item):
"""商品综合评分算法"""
score = 0
# 价格维度（50-200元为最佳转化区间）
price = float(item.get("price", 0))
if 50 <= price <= 200:
score += 30
elif 200 < price <= 500:
score += 20
# 销量维度（日销>30为活跃款）
sales = int(item.get("sales", 0))
if sales > 1000:
score += 25
elif sales > 100:
score += 20
elif sales > 30:
score += 15
# 评价维度
rating = float(item.get("rating", 0))
if rating >= 4.8:
score += 20
elif rating >= 4.5:
score += 15
# 库存维度
stock = int(item.get("stock", 0))
if stock > 500:
score += 15
elif stock > 100:
score += 10
# 店铺维度（自营加分）
if item.get("is_self") == 1:
score += 10
return score
批量选品示例
def batch_select_products(keyword, top_n=20):
items = keyword_search(keyword)
scored = [(item, product_scoring(item)) for item in items]
scored.sort(key=lambda x: x[1], reverse=True)
return scored[:top_n]
4.2 竞品对标分析
通过自动关联同款商品，实现跨平台比价：
Python
def cross_platform_compare(sku_mapping):
"""跨平台价格对比"""
results = []
for product in sku_mapping:
prices = {}
for platform, sku in product["skus"].items():
data = get_price(platform, sku)
prices[platform] = {
"price": data["price"],
"stock": data["stock"],
"promotion": data.get("coupons", [])
}
# 计算价差
min_price = min(p["price"] for p in prices.values())
max_price = max(p["price"] for p in prices.values())
results.append({
"name": product["name"],
"prices": prices,
"spread": max_price - min_price,
"spread_rate": (max_price - min_price) / min_price * 100
})
return results
五、生产级部署与优化
5.1 频率控制与限流处理
各平台 API 均有调用限制，需实现动态等待机制：
表格
平台 默认限流 应对策略
京东 视权限等级 分散采集时间，错开 1-2 分钟
亚马逊 SP-API 每小时 4000 次 本地 Redis 缓存，重复 ASIN ≥24 小时查询
淘宝 10 次/分钟（基础版） 令牌桶算法平滑请求
Python
复制
from ratelimit import limits, sleep_and_retry
@sleep_and_retry
@limits(calls=10, period=60) # 每分钟最多10次
def api_call_with_rate_limit(url, params):
return requests.get(url, params=params)
5.2 高可用架构设计
Python
复制
代理 IP 池（应对封禁）
PROXY_POOL = [
"http://user:pass@1.1.1.1:8080",
"http://user:pass@2.2.2.2:8081"
]
def get_with_proxy(url, params, max_retry=3):
for i in range(max_retry):
proxy = {"http": random.choice(PROXY_POOL)}
try:
resp = requests.get(url, params=params, proxies=proxy, timeout=10)
if resp.status_code == 200:
return resp.json()
except Exception as e:
print(f"第{i+1}次请求失败: {e}")
time.sleep(2 ** i) # 指数退避
return None
5.3 告警风暴防护
避免同一商品频繁触发告警：
class AlertManager:
def init(self, cooldown=3600):
self.cooldown = cooldown # 冷静期1小时
self.last_alert = {} # 记录上次告警时间
def should_alert(self, sku_id):
now = time.time()
if sku_id in self.last_alert:
if now - self.last_alert[sku_id] < self.cooldown:
return False
self.last_alert[sku_id] = now
return True
六、实战避坑指南
表格
问题场景 解决方案
签名错误 日志记录完整参数拼接过程，用官方工具验证本地签名
数据缺失/重复 添加重试机制（最多3次），存储前按 SKU+时间戳去重
价格异常（如0元） 数据清洗时标记异常值，设置合理价格区间过滤
IP 被封 启用住宅代理池，配合请求头随机化、随机延迟 2-5 秒
页面改版导致解析失败 优先使用官方 API 而非网页爬虫，API 返回结构化 JSON 不受页面改版影响
七、总结与扩展方向
通过 API 接口实现电商选品与价格监控，核心价值在于：
实时性：分钟级更新，捕捉转瞬即逝的价格波动
可扩展性：支持多平台、多商品，随业务需求灵活扩展
决策支撑：基于数据而非经验制定定价策略，提升竞争力
后续可重点优化方向：
竞品自动发现：通过商品标题分词 + 图像识别，自动匹配同款竞品
价格预测：基于 LSTM 时序模型预测未来价格走势，辅助决策
动态定价联动：监控到竞品降价后，自动调整自身促销策略
可视化看板：用 Flask + ECharts 构建实时价格趋势大屏
