在电商团队的日常运营中，竞品价格监控往往依赖人工刷新页面、复制数据、Excel 对比，不仅耗时（日均 8-10 小时），还容易错过调价窗口期。本文基于某 3C 电商团队的实战经验，分享如何用 Python 脚本实现全自动化竞品价格监控，包含多平台采集、智能分析、自动告警全流程，最终将人工成本降至 1 小时 / 天，调价响应速度从 4 小时缩短至 1.5 小时。
一、核心需求与系统架构
- 业务痛点
需监控 10 个核心竞品（覆盖京东、淘宝、拼多多 3 大平台），每个平台 5-8 个 SKU，人工逐页查询需 2 小时 / 次。
价格波动频繁（如大促期间每 30 分钟调价 1 次），人工记录易遗漏，导致错失调价时机。
数据分散在 Excel 中，无法快速生成趋势图，调价决策依赖经验而非数据。 - 系统架构设计
graph LR
A[定时任务调度] --> B[多平台采集模块]
B --> C[数据清洗层]
C --> D[SQLite存储]
D --> E[价格分析引擎]
E --> F[阈值告警]
F --> G[企业微信通知]
D --> H[可视化看板]
核心目标：实现 “定时采集→自动存储→异常告警→趋势分析” 闭环，减少 80% 人工操作。
二、全流程实现：从采集到告警 - 多平台价格采集模块（抗反爬 + API 兼容）
针对不同平台特性，采用 “官方 API + 智能爬虫” 混合采集方案：
（1）京东 / 淘宝 API 采集（稳定优先）
利用平台开放 API 获取结构化数据，避免反爬风险：
python
运行
import requests
import hashlib
import time
class JDPriceCrawler:
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 + "".join([f"{k}{v}" for k, v in sorted_params]) + 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)
price = response.json()["jingdong_ware_price_get_response"]["price"]["p"]
return {"platform": "jd", "sku_id": sku_id, "price": float(price), "crawl_time": time.time()}
except Exception as e:
print(f"京东采集失败：{e}")
return None
（2）拼多多 / 抖音爬虫（无 API 平台）
用requests+BeautifulSoup爬取非开放平台数据，配合反爬策略：
python
运行
import requests
from bs4 import BeautifulSoup
import random
import time
class PDDPriceCrawler:
def init(self):
self.headers = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36...",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36..."
]
self.proxies = [ # 代理IP池（可从第三方购买）
{"http": "http://123.45.67.89:8080"},
{"http": "http://98.76.54.32:8080"}
]
def get_price(self, goods_id):
"""爬取拼多多商品价格"""
url = f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}"
try:
# 随机请求头和代理，降低反爬概率
response = requests.get(
url,
headers={"User-Agent": random.choice(self.headers)},
proxies=random.choice(self.proxies),
timeout=10
)
soup = BeautifulSoup(response.text, "html.parser")
# 解析价格（需根据实际页面结构调整选择器）
price = soup.select_one(".p-price").text.strip("¥")
return {"platform": "pdd", "sku_id": goods_id, "price": float(price), "crawl_time": time.time()}
except Exception as e:
print(f"拼多多采集失败：{e}")
return None
二、数据存储与定时调度
- 轻量存储方案（SQLite）
无需复杂数据库，用 SQLite 存储价格历史，适合中小团队：
python
运行
import sqlite3
from datetime import datetime
class PriceStorage:
def init(self, db_name="price_monitor.db"):
self.conn = sqlite3.connect(db_name)
self._create_table()
def _create_table(self):
"""创建价格记录表"""
self.conn.execute('''
CREATE TABLE IF NOT EXISTS price_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT,
sku_id TEXT,
price REAL,
crawl_time DATETIME,
UNIQUE(platform, sku_id, crawl_time) # 去重
)
''')
self.conn.commit()
def save_price(self, price_data):
"""存储价格数据"""
crawl_time = datetime.fromtimestamp(price_data["crawl_time"]).strftime("%Y-%m-%d %H:%M:%S")
self.conn.execute('''
INSERT OR IGNORE INTO price_history (platform, sku_id, price, crawl_time)
VALUES (?, ?, ?, ?)
''', (price_data["platform"], price_data["sku_id"], price_data["price"], crawl_time))
self.conn.commit()
- 定时任务调度（自动采集）
用schedule库设置定时采集，避免人工触发：
python
运行
import schedule
import time
from concurrent.futures import ThreadPoolExecutor
class PriceMonitor:
def init(self, crawlers, storage, interval=60): # 每60分钟采集1次
self.crawlers = crawlers # 爬虫实例列表：[JD, PDD, ...]
self.storage = storage
self.interval = interval
# 监控的SKU列表（可从配置文件读取）
self.monitor_list = {
"jd": ["100012345678", "100012345679"],
"pdd": ["1234567890123"]
}
def collect_single_sku(self, platform, sku_id):
"""采集单个SKU价格"""
crawler = next(c for c in self.crawlers if c.platform == platform)
price_data = crawler.get_price(sku_id)
if price_data:
self.storage.save_price(price_data)
print(f"已存储 {platform} {sku_id} 价格：{price_data['price']}")
def batch_collect(self):
"""批量采集所有监控商品"""
print(f"开始采集：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# 多线程并发采集，提升效率
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"采集完成：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
def start(self):
"""启动监控"""
self.batch_collect() # 立即执行1次
schedule.every(self.interval).minutes.do(self.batch_collect) # 定时执行
while True:
schedule.run_pending()
time.sleep(60)
启动监控系统
if name == "main":
# 初始化组件
jd_crawler = JDPriceCrawler(app_key="你的appkey", app_secret="你的appsecret")
pdd_crawler = PDDPriceCrawler()
storage = PriceStorage()
monitor = PriceMonitor(crawlers=[jd_crawler, pdd_crawler], storage=storage, interval=60)
monitor.start()
三、智能分析与自动告警
价格波动分析（识别调价时机）
计算价格变化率，设置阈值触发告警：
python
运行
class PriceAnalyzer:
def init(self, storage, threshold=0.05): # 5%波动阈值self.storage = storage self.threshold = thresholddef get_price_change(self, platform, sku_id):
"""计算最近2次价格变化率""" # 获取最近2条记录 cursor = self.storage.conn.execute(''' SELECT price, crawl_time FROM price_history WHERE platform = ? AND sku_id = ? ORDER BY crawl_time DESC LIMIT 2 ''', (platform, sku_id)) records = cursor.fetchall() if len(records) < 2: return None # 数据不足 prev_price, prev_time = records[1][0], records[1][1] current_price, current_time = records[0][0], records[0][1] change_rate = (current_price - prev_price) / prev_price return { "platform": platform, "sku_id": sku_id, "prev_price": prev_price, "current_price": current_price, "change_rate": change_rate, "time_range": f"{prev_time} → {current_time}" }- 企业微信告警（实时通知）
价格波动超阈值时，自动发送通知到团队群：
python
运行
import requests
class AlertSender:
def init(self, webhook):
self.webhook = webhook # 企业微信机器人webhook
def send_alert(self, change_info):
"""发送价格变动告警"""
change_type = "上涨" if change_info["change_rate"] > 0 else "下降"
msg = f"""
【竞品价格异动告警】
平台：{change_info['platform']}
商品ID：{change_info['sku_id']}
原价：{change_info['prev_price']}元
现价：{change_info['current_price']}元
变动幅度：{abs(change_info['change_rate'])*100:.2f}%（{change_type}）
时间：{change_info['time_range']}
"""
# 调用企业微信机器人API
requests.post(self.webhook, json={"msgtype": "text", "text": {"content": msg}})
分析+告警联动
analyzer = PriceAnalyzer(storage)
alert_sender = AlertSender(webhook="你的企业微信机器人webhook")
每次采集后执行分析
def analyze_after_collect():
for platform, sku_list in monitor.monitor_list.items():
for sku_id in sku_list:
change_info = analyzer.get_price_change(platform, sku_id)
if change_info and abs(change_info["change_rate"]) >= analyzer.threshold:
alert_sender.send_alert(change_info)
在PriceMonitor的batch_collect方法末尾添加
monitor.batch_collect = lambda: (monitor._batch_collect(), analyze_after_collect())
四、可视化看板（辅助决策）
用pandas+matplotlib生成价格趋势图，直观展示竞品动态：
python
运行
import pandas as pd
import matplotlib.pyplot as plt
def generate_trend_chart(storage, platform, sku_id):
"""生成价格趋势图"""
df = pd.read_sql_query(f'''
SELECT crawl_time, price FROM price_history
WHERE platform = '{platform}' AND sku_id = '{sku_id}'
ORDER BY crawl_time
''', storage.conn)
df["crawl_time"] = pd.to_datetime(df["crawl_time"])
plt.figure(figsize=(10, 5))
plt.plot(df["crawl_time"], df["price"], marker='o')
plt.title(f"{platform} {sku_id} 价格趋势")
plt.xlabel("时间")
plt.ylabel("价格（元）")
plt.grid(True)
plt.savefig(f"{platform}_{sku_id}_trend.png")
plt.close()
五、实施效果与优化建议
- 核心收益
效率提升：日均人工操作从 8 小时降至 1 小时（仅需处理告警和决策）。
响应加速：竞品调价后平均 1 小时内触发告警，响应速度提升 3 倍（原需 4 小时人工发现）。
数据驱动：基于 7 天价格趋势制定调价策略，避免盲目跟价，毛利率提升 2.3%。 - 优化方向
反爬增强：接入动态代理池（如 BrightData），应对平台反爬升级。
智能调价：对接内部 ERP 系统，当竞品降价超 8% 时自动生成调价单，无需人工介入。
多维度分析：增加销量、评论数等维度，判断竞品调价是否为 “清库存” 而非长期策略。
通过这套 Python 脚本，团队可彻底摆脱重复的人工监控工作，将精力集中在策略制定上。核心是通过 “自动化采集→智能分析→实时告警” 的闭环，让数据主动 “说话”，实现从 “被动跟价” 到 “主动决策” 的转变
