电商竞品全自动监控 + 数据分析落地教程：从 0 到 1 搭建你的竞品情报雷达
一、为什么你需要一套竞品监控系统？
在电商行业，信息差就是利润差。对手降价了、上了新品、积累了差评、调整了广告策略——这些变化每天都在发生，但 90% 的团队仍然靠人工刷新页面、复制粘贴到 Excel 来跟踪竞品。某 3C 电商团队的实测数据显示：人工监控 10 个核心竞品（覆盖京东、淘宝、拼多多 3 大平台），每次耗时 2 小时，日均 8-10 小时，且大促期间价格波动频繁（每 30 分钟调价 1 次），人工记录极易遗漏，导致错失调价窗口期。
自动化竞品监控系统的核心价值：
- 效率跃迁：日均人工操作从 8 小时降至 1 小时，调价响应速度从 4 小时缩短至 1.5 小时
- 信息闭环：实现"定时采集 → 自动存储 → 智能分析 → 实时告警 → 策略决策"全链路
- 数据驱动：基于 7 天价格趋势制定调价策略，避免盲目跟价，毛利率可提升 2.3%
本文将带你从零搭建一套多平台、全自动化、可落地的竞品监控与数据分析系统，全程提供可运行的 Python 代码。
二、系统整体架构
plain
┌──────────────────────────────────────────────────────────────────┐
│ 电商竞品全自动监控系统 │
├──────────────┬──────────────┬──────────────┬─────────────────────┤
│ 数据采集层 │ 数据处理层 │ 智能分析层 │ 应用展示层 │
├──────────────┼──────────────┼──────────────┼─────────────────────┤
│ • 官方 API │ • 数据清洗 │ • 价格波动 │ • 实时监控看板 │
│ • 智能爬虫 │ • 去重过滤 │ • 销量异常 │ • 钉钉/企微告警 │
│ • 反爬策略 │ • 结构化存储 │ • 趋势预测 │ • 竞品对比报告 │
│ • 多平台适配 │ • 增量同步 │ • 上新追踪 │ • 自动调价建议 │
└──────────────┴──────────────┴──────────────┴─────────────────────┘技术栈选型：
表格
| 层级 | 技术选型 | 说明 |
|---|---|---|
| 数据采集 | Python + Requests / aiohttp / Playwright | API + 爬虫混合采集 |
| 数据存储 | DuckDB / SQLite + Redis | 轻量列式存储 + 缓存去重 |
| 数据处理 | Pandas + DuckDB SQL | 清洗、聚合、窗口分析 |
| 智能分析 | 统计模型 + DeepSeek / GPT | 价格预测、异常检测 |
| 定时调度 | schedule / APScheduler | 分钟级/小时级自动执行 |
| 可视化 | Matplotlib / ECharts / Grafana | 趋势图、雷达图、仪表盘 |
| 告警推送 | 钉钉 / 企业微信 Webhook | 实时价格异动通知 |
三、数据采集层：多平台混合采集策略
3.1 采集策略矩阵
不同平台的数据开放程度不同，需采用"官方 API 优先 + 智能爬虫兜底"的混合策略：
表格
| 平台 | 采集方式 | 可获取字段 | 稳定性 |
|---|---|---|---|
| 京东 | 开放平台 API | 价格、库存、标题、SKU、评价 | ⭐⭐⭐⭐⭐ |
| 淘宝/天猫 | 开放平台 API | 价格、销量、标题、主图、评价 | ⭐⭐⭐⭐⭐ |
| 拼多多 | 爬虫（Vue 加密接口） | 价格、销量、标题、SKU | ⭐⭐⭐ |
| 抖音电商 | 抖店开放平台 | 价格、销量、直播数据 | ⭐⭐⭐⭐ |
| 亚马逊 | 爬虫 / 第三方数据服务 | 价格、排名、评论、库存 | ⭐⭐⭐ |
⚠️ 合规提示：优先使用官方开放平台 API，爬虫仅抓取公开可见数据，遵守平台robots.txt及用户协议，控制请求频率，避免对平台服务器造成压力。
3.2 京东 API 采集模块（稳定优先）
京东开放平台提供结构化数据接口，签名规则按官方规范实现：
Python
import requestsimport hashlibimport timeimport jsonclass 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 签名（按官方规则：参数排序 + MD5）"""
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)
data = response.json()["jingdong_ware_price_get_response"]["price"]
return {
"platform": "jd",
"sku_id": sku_id,
"price": float(data["p"]),
"market_price": float(data.get("m", 0)),
"crawl_time": time.time()
}
except Exception as e:
print(f"京东采集失败：{e}")
return None3.3 异步并发爬虫模块（多平台适配）
对于无开放 API 的平台，使用
aiohttp 实现高并发异步采集，配合代理池和随机 User-Agent 降低被封风险：Python
import aiohttpimport asynciofrom bs4 import BeautifulSoupimport randomfrom datetime import datetimeclass MultiPlatformCrawler:
def __init__(self):
self.proxies = [
{"http": "http://123.45.67.89:8080"},
{"http": "http://98.76.54.32:8080"},
]
self.user_agents = [
"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.headers = {
"User-Agent": random.choice(self.user_agents),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}
async def fetch_page(self, url, session):
"""异步获取页面内容"""
try:
proxy = random.choice(self.proxies)
async with session.get(
url,
headers=self.headers,
proxy=proxy.get("http"),
timeout=aiohttp.ClientTimeout(total=15)
) as response:
return await response.text()
except Exception as e:
print(f"请求失败：{url}, 错误：{e}")
return None
async def parse_product(self, html, platform):
"""解析商品信息（根据实际页面结构调整选择器）"""
if not html:
return None
soup = BeautifulSoup(html, "html.parser")
try:
if platform == "tmall":
title = soup.select_one(".tb-detail-hd h1").text.strip()
price = float(soup.select_one(".tm-price").text.strip())
sales_text = soup.select_one(".tm-ind-sellCount").text.strip()
elif platform == "pdd":
title = soup.select_one(".goods-title").text.strip()
price = float(soup.select_one(".p-price").text.strip("¥"))
sales_text = soup.select_one(".p-sales").text.strip()
else:
return None
# 统一销量格式
sales = self._parse_sales(sales_text)
return {
"platform": platform,
"title": title,
"price": price,
"sales": sales,
"crawl_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
except Exception as e:
print(f"解析失败：{e}")
return None
def _parse_sales(self, sales_text):
"""统一解析销量（处理"万""千"等单位）"""
sales_text = sales_text.replace("销量", "").replace("+", "").strip()
if "万" in sales_text:
return int(float(sales_text.replace("万", "")) * 10000)
elif "千" in sales_text:
return int(float(sales_text.replace("千", "")) * 1000)
else:
return int(sales_text) if sales_text.isdigit() else 0
async def batch_crawl(self, url_list):
"""批量并发采集"""
connector = aiohttp.TCPConnector(limit=5) # 限制并发避免被封
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [self.fetch_page(url, session) for url in url_list]
htmls = await asyncio.gather(*tasks, return_exceptions=True)
results = []
for html, url in zip(htmls, url_list):
if isinstance(html, str):
platform = "tmall" if "tmall" in url else "pdd"
result = await self.parse_product(html, platform)
if result:
results.append(result)
return results3.4 反爬策略清单
拼多多等平台前端使用深度定制的 Vue 框架，接口参数层层加密，需采取多重反爬措施：
表格
| 策略 | 实现方式 |
|---|---|
| 代理 IP 池 | 接入 BrightData / 芝麻代理等付费代理服务，每次请求轮换 IP |
| User-Agent 池 | 维护 50+ 真实浏览器 UA，随机切换 |
| 请求频率控制 | 单 IP 每秒 ≤2 次，夜间（2:00-6:00）适当放宽 |
| Cookie 轮换 | 维护多个账号 Cookie 池，模拟真实用户会话 |
| 行为模拟 | 使用 Playwright/Selenium 模拟鼠标移动、页面滚动 |
| 验证码识别 | 接入 2Captcha / 极验等打码平台自动过验证 |
四、数据处理层：清洗、去重与结构化存储
4.1 为什么选 DuckDB？
传统方案用 Python + Pandas 处理，但竞品监控数据量大、更新频繁，DuckDB 是更优选择：
表格
| 维度 | Python + Pandas | DuckDB |
|---|---|---|
| 内存占用 | 全部加载到内存 | 列式压缩，按需读取 |
| SQL 能力 | 需写循环和条件逻辑 | 直接写 SQL 聚合、JOIN、窗口函数 |
| 脏数据处理 | 容易整批崩溃 | RETURN_NULL_ON_ERROR 优雅容错 |
| 部署成本 | 需 Python 环境 | 单文件二进制，零依赖 |
| 100 万行处理 | ~15 秒 | ~0.3 秒 |
4.2 数据清洗与存储
Python
import duckdbfrom datetime import datetimeclass CompetitorDataPipeline:
def __init__(self, db_name="competitor_monitor.db"):
self.con = duckdb.connect(db_name)
self._init_tables()
def _init_tables(self):
"""初始化数据表结构"""
self.con.execute("""
CREATE TABLE IF NOT EXISTS raw_data (
id INTEGER PRIMARY KEY,
platform VARCHAR(20),
sku_id VARCHAR(50),
title VARCHAR(500),
price DOUBLE,
market_price DOUBLE,
sales BIGINT,
stock INTEGER,
review_count INTEGER,
rating DOUBLE,
crawl_time TIMESTAMP,
url VARCHAR(500)
)
""")
self.con.execute("""
CREATE TABLE IF NOT EXISTS price_changes (
id INTEGER PRIMARY KEY,
sku_id VARCHAR(50),
platform VARCHAR(20),
old_price DOUBLE,
new_price DOUBLE,
change_rate DOUBLE,
change_type VARCHAR(10),
detect_time TIMESTAMP
)
""")
self.con.commit()
def insert_or_update(self, records):
"""插入新数据，并自动检测价格变化"""
for record in records:
# 查找该商品最新记录
existing = self.con.execute("""
SELECT price, sales FROM raw_data
WHERE sku_id = ? AND platform = ?
ORDER BY crawl_time DESC LIMIT 1
""", (record["sku_id"], record["platform"])).fetchone()
if existing:
old_price, old_sales = existing
new_price = record.get("price", 0)
# 检测价格变化
if old_price > 0 and abs(new_price - old_price) / old_price > 0.01:
change_rate = (new_price - old_price) / old_price
self.con.execute("""
INSERT INTO price_changes
(sku_id, platform, old_price, new_price, change_rate, change_type, detect_time)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
record["sku_id"], record["platform"], old_price, new_price,
change_rate, "上涨" if change_rate > 0 else "下降", datetime.now()
))
# 插入原始数据
self.con.execute("""
INSERT INTO raw_data
(platform, sku_id, title, price, market_price, sales, crawl_time, url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
record["platform"], record["sku_id"], record.get("title", ""),
record.get("price", 0), record.get("market_price", 0),
record.get("sales", 0), record.get("crawl_time", datetime.now()),
record.get("url", "")
))
self.con.commit()
def get_latest_snapshot(self, sku_id=None, platform=None):
"""获取最新数据快照"""
query = """
SELECT * FROM raw_data
WHERE (sku_id, platform, crawl_time) IN (
SELECT sku_id, platform, MAX(crawl_time)
FROM raw_data
GROUP BY sku_id, platform
)
"""
if sku_id:
query += f" AND sku_id = '{sku_id}'"
if platform:
query += f" AND platform = '{platform}'"
return self.con.execute(query).fetchdf()4.3 Redis 实时去重
使用 Redis 缓存已采集商品的关键字段，避免重复写入和无效更新：
Python
import redisimport jsonimport hashlibclass SmartDeduplicator:
def __init__(self, host="localhost", port=6379):
self.r = redis.Redis(host=host, port=port, db=0, decode_responses=True)
self.price_threshold = 0.03 # 价格变化≥3%才视为新数据
self.sales_threshold = 0.20 # 销量变化≥20%才视为新数据
def _generate_key(self, item):
"""生成商品唯一标识"""
unique_str = f"{item['platform']}:{item['sku_id']}"
return f"competitor:{hashlib.md5(unique_str.encode()).hexdigest()}"
def is_significant_change(self, item):
"""判断是否为重要变化（避免频繁写入无变化数据）"""
key = self._generate_key(item)
existing = self.r.get(key)
if not existing:
return True # 新商品
old = json.loads(existing)
# 价格变化检测
if old.get("price", 0) > 0:
price_change = abs(item["price"] - old["price"]) / old["price"]
if price_change >= self.price_threshold:
return True
# 销量变化检测
if old.get("sales", 0) > 0:
sales_change = abs(item["sales"] - old["sales"]) / max(old["sales"], 1)
if sales_change >= self.sales_threshold:
return True
return False
def update(self, item, ttl=7*86400):
"""更新缓存，保留7天"""
key = self._generate_key(item)
self.r.set(key, json.dumps(item), ex=tpl)五、智能分析层：从数据到洞察
5.1 价格波动异常检测
使用 SQL 窗口函数计算 Z-Score，识别异常调价行为：
Python
def detect_price_anomalies(self, days=7):
"""检测价格异常波动（Z-Score > 2 为异常）"""
return self.con.execute(f"""
WITH daily_stats AS (
SELECT
sku_id,
platform,
DATE(crawl_time) as date,
AVG(price) as avg_price,
STDDEV_SAMP(price) OVER (PARTITION BY sku_id ORDER BY DATE(crawl_time) ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) as rolling_std
FROM raw_data
WHERE crawl_time >= CURRENT_DATE - INTERVAL '{days}' DAY
GROUP BY sku_id, platform, DATE(crawl_time)
),
baseline AS (
SELECT
sku_id,
platform,
AVG(avg_price) as baseline_price,
STDDEV_SAMP(avg_price) as baseline_std
FROM daily_stats
GROUP BY sku_id, platform
)
SELECT
d.sku_id,
d.platform,
d.date,
d.avg_price,
b.baseline_price,
ROUND((d.avg_price - b.baseline_price) / NULLIF(b.baseline_std, 0), 2) as z_score
FROM daily_stats d
JOIN baseline b ON d.sku_id = b.sku_id AND d.platform = b.platform
WHERE ABS((d.avg_price - b.baseline_price) / NULLIF(b.baseline_std, 0)) > 2
ORDER BY ABS(z_score) DESC
""").fetchdf()5.2 竞品定价策略分析
Python
def analyze_pricing_strategy(self, competitor_sku):
"""分析竞品定价策略"""
df = self.con.execute(f"""
SELECT
DATE(crawl_time) as date,
price,
market_price,
sales
FROM raw_data
WHERE sku_id = '{competitor_sku}'
ORDER BY crawl_time
""").fetchdf()
if len(df) < 3:
return {"error": "数据不足"}
# 计算折扣率趋势
df["discount_rate"] = (df["market_price"] - df["price"]) / df["market_price"]
# 识别促销模式
promo_periods = df[df["discount_rate"] > 0.3] # 折扣率>30%视为大促
analysis = {
"avg_price": round(df["price"].mean(), 2),
"price_volatility": round(df["price"].std(), 2),
"avg_discount": round(df["discount_rate"].mean() * 100, 1),
"promo_frequency": f"每{len(df)//max(len(promo_periods), 1)}天一次",
"price_trend": "上涨" if df["price"].iloc[-1] > df["price"].iloc[0] else "下降",
"suggested_monitor_freq": "每2小时" if df["price"].std() > 5 else "每天"
}
return analysis5.3 DeepSeek 大模型价格预测（进阶）
接入大模型对历史价格数据进行趋势分析和未来预测：
Python
import openaifrom tenacity import retry, stop_after_attempt, wait_exponentialclass PricePredictor:
def __init__(self, api_key):
openai.api_base = "https://api.deepseek.com/v1"
openai.api_key = api_key
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def predict_trend(self, price_history):
"""使用 DeepSeek 分析价格趋势并预测"""
history_str = "\n".join(
f"{row['date']}: ¥{row['price']} (销量{row['sales']})"
for _, row in price_history.tail(30).iterrows()
)
prompt = f"""你是一位资深电商价格分析师，请分析以下商品的价格趋势：
历史数据（最近30天）：{history_str}请分析：
1. 价格变化的周期性和规律
2. 价格与销量的相关性
3. 可能的促销活动时间点
预测未来7天价格变化，并用JSON格式回复：
{{
"current_price": 当前价格,
"predicted_prices": {{"1d": x, "3d": x, "7d": x}},
"confidence": 置信度(0-100),
"trend": "上涨/下降/震荡",
"recommendations": ["建议1", "建议2"]
}}"""
response = await openai.ChatCompletion.acreate(
model="deepseek-chat",
messages=[
{"role": "system", "content": "你是一位资深的电商价格分析师。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=800
)
return response.choices[0].message.content六、实时预警与通知系统
6.1 风险分级体系
表格
| 级别 | 触发条件 | 响应时效 | 处置动作 |
|---|---|---|---|
| P0 紧急 | 核心竞品降价 ≥15% 或库存清零 | 即时 | 钉钉@所有人 + 自动创建调价工单 |
| P1 重要 | 竞品降价 5%-15% 或上新 | 30分钟内 | 企业微信推送 + 生成对比报告 |
| P2 一般 | 价格微调 <5% 或评价变化 | 2小时内 | 日报汇总 |
6.2 钉钉/企微告警推送
Python
import requestsclass AlertSender:
def __init__(self, webhook_url):
self.webhook = webhook_url
def send_price_alert(self, change_info, level="P1"):
"""发送价格变动告警"""
change_type = "上涨" if change_info["change_rate"] > 0 else "下降"
emoji = "🔴" if level == "P0" else ("🟠" if level == "P1" else "🟡")
msg = {
"msgtype": "markdown",
"markdown": {
"title": f"{emoji} 竞品价格{level}预警",
"text": f"""### {emoji} 竞品价格异动告警
**平台：** {change_info['platform']} **商品：** {change_info.get('title', change_info['sku_id'])} **原价：** ¥{change_info['old_price']} **现价：** ¥{change_info['new_price']} **变动幅度：** {abs(change_info['change_rate'])*100:.1f}%（{change_type}）
**检测时间：** {change_info['detect_time']} > 建议：{'立即跟进调价' if level == 'P0' else '关注后续走势'}"""
}
}
requests.post(self.webhook, json=msg, timeout=10)七、可视化看板与自动报告
7.1 价格趋势图生成
Python
import matplotlib.pyplot as pltimport pandas as pddef generate_trend_chart(pipeline, sku_id, platform, days=30):
"""生成竞品价格趋势图"""
df = pipeline.con.execute(f"""
SELECT DATE(crawl_time) as date,
AVG(price) as price,
AVG(sales) as sales
FROM raw_data
WHERE sku_id = '{sku_id}' AND platform = '{platform}'
AND crawl_time >= CURRENT_DATE - INTERVAL '{days}' DAY
GROUP BY DATE(crawl_time)
ORDER BY date
""").fetchdf()
if len(df) == 0:
return None
fig, ax1 = plt.subplots(figsize=(12, 6))
# 价格曲线
color1 = '#2E6BB4'
ax1.set_xlabel('日期', fontsize=12)
ax1.set_ylabel('价格（元）', color=color1, fontsize=12)
ax1.plot(df['date'], df['price'], color=color1, marker='o', linewidth=2, label='价格')
ax1.tick_params(axis='y', labelcolor=color1)
ax1.grid(True, alpha=0.3)
# 销量柱状图
ax2 = ax1.twinx()
color2 = '#E87A3A'
ax2.set_ylabel('销量', color=color2, fontsize=12)
ax2.bar(df['date'], df['sales'], color=color2, alpha=0.3, label='销量')
ax2.tick_params(axis='y', labelcolor=color2)
plt.title(f'{platform} - {sku_id} 价格与销量趋势（近{days}天）', fontsize=14, fontweight='bold')
fig.tight_layout()
filename = f"charts/{platform}_{sku_id}_trend_{datetime.now().strftime('%Y%m%d')}.png"
plt.savefig(filename, dpi=150, bbox_inches='tight')
plt.close()
return filename7.2 竞品对比雷达图
Python
import numpy as npdef generate_competitor_radar(pipeline, sku_list):
"""生成多竞品对比雷达图"""
categories = ['价格竞争力', '销量规模', '评价质量', '折扣力度', '上架时长']
# 获取各竞品最新数据
data = []
for sku in sku_list:
row = pipeline.con.execute(f"""
SELECT
price,
sales,
rating,
(market_price - price) / market_price as discount,
DATEDIFF('day', MIN(crawl_time), CURRENT_DATE) as days_online
FROM raw_data
WHERE sku_id = '{sku}'
GROUP BY sku_id
""").fetchone()
if row:
# 标准化到 0-10 分
data.append([
10 - row[0] / 100, # 价格越低分越高（简化示例）
min(row[1] / 1000, 10),
row[2] * 2,
row[3] * 10,
min(row[4] / 30, 10)
])
# 绘制雷达图
angles = np.linspace(0, 2 * np.pi, len(categories), endpoint=False).tolist()
angles += angles[:1]
fig, ax = plt.subplots(figsize=(8, 8), subplot_kw=dict(projection='polar'))
for i, d in enumerate(data):
values = d + d[:1]
ax.plot(angles, values, 'o-', linewidth=2, label=sku_list[i])
ax.fill(angles, values, alpha=0.15)
ax.set_xticks(angles[:-1])
ax.set_xticklabels(categories)
ax.set_ylim(0, 10)
ax.legend(loc='upper right', bbox_to_anchor=(1.3, 1.1))
plt.title('竞品多维对比分析', fontsize=16, fontweight='bold', pad=20)
plt.savefig("charts/competitor_radar.png", dpi=150, bbox_inches='tight')
plt.close()八、完整系统集成与一键部署
8.1 主控调度器
Python
import scheduleimport timefrom concurrent.futures import ThreadPoolExecutorclass CompetitorMonitorSystem:
def __init__(self, config):
self.config = config
self.pipeline = CompetitorDataPipeline(config["db_name"])
self.deduplicator = SmartDeduplicator()
self.alert = AlertSender(config["webhook_url"])
self.crawlers = {
"jd": JDPriceCrawler(config["jd_app_key"], config["jd_app_secret"]),
"multi": MultiPlatformCrawler()
}
self.monitor_list = config["monitor_skus"] # 监控列表
def collect_all(self):
"""执行全量采集"""
print(f"【{datetime.now()}】开始执行竞品数据采集...")
all_records = []
# 1. API 采集（京东等）
for sku in self.monitor_list.get("jd", []):
record = self.crawlers["jd"].get_price(sku)
if record:
all_records.append(record)
# 2. 爬虫采集（其他平台）
urls = self.monitor_list.get("urls", [])
if urls:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
records = loop.run_until_complete(
self.crawlers["multi"].batch_crawl(urls)
)
all_records.extend(records)
# 3. 去重与存储
significant = [r for r in all_records if self.deduplicator.is_significant_change(r)]
if significant:
self.pipeline.insert_or_update(significant)
print(f"发现 {len(significant)} 条重要变化，已入库")
# 4. 告警检测
for record in significant:
# 检查是否为降价
changes = self.pipeline.con.execute(f"""
SELECT * FROM price_changes
WHERE sku_id = '{record['sku_id']}'
ORDER BY detect_time DESC LIMIT 1
""").fetchone()
if changes:
change_rate = changes[4]
level = "P0" if abs(change_rate) > 0.15 else ("P1" if abs(change_rate) > 0.05 else "P2")
self.alert.send_price_alert({
"sku_id": changes[0],
"platform": changes[1],
"old_price": changes[2],
"new_price": changes[3],
"change_rate": change_rate,
"detect_time": changes[6]
}, level)
else:
print("未发现重要变化")
def generate_daily_report(self):
"""生成每日竞品报告"""
# 价格变化汇总
changes = self.pipeline.con.execute("""
SELECT * FROM price_changes
WHERE detect_time >= CURRENT_DATE
ORDER BY change_rate DESC
""").fetchdf()
# 生成趋势图
for _, row in changes.iterrows():
generate_trend_chart(self.pipeline, row["sku_id"], row["platform"])
print(f"【日报】今日共检测到 {len(changes)} 次价格变动，报告已生成")
def start(self, interval_minutes=60):
"""启动定时监控"""
print(f"🚀 竞品监控系统已启动，每 {interval_minutes} 分钟采集一次")
self.collect_all() # 立即执行一次
schedule.every(interval_minutes).minutes.do(self.collect_all)
schedule.every().day.at("09:00").do(self.generate_daily_report)
while True:
schedule.run_pending()
time.sleep(60)# ===================== 配置文件示例 =====================CONFIG = {
"db_name": "competitor_monitor.db",
"jd_app_key": "your_jd_app_key",
"jd_app_secret": "your_jd_app_secret",
"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=xxx",
"monitor_skus": {
"jd": ["100012345678", "100012345679"],
"urls": [
"https://detail.tmall.com/item.htm?id=xxx",
"https://mobile.yangkeduo.com/goods.html?goods_id=xxx"
]
}}if __name__ == "__main__":
system = CompetitorMonitorSystem(CONFIG)
system.start(interval_minutes=60)九、落地效果与优化方向
9.1 核心收益
某 3C 电商团队落地该系统后的实测效果：
表格
| 指标 | 落地前 | 落地后 | 提升 |
|---|---|---|---|
| 日均人工耗时 | 8-10 小时 | 1 小时 | 节省 87.5% |
| 调价响应速度 | 4 小时 | 1.5 小时 | 提升 2.7 倍 |
| 数据准确率 | 85% | 99% | 提升 14% |
| 毛利率影响 | 被动跟价 | 数据驱动决策 | +2.3% |
9.2 进阶优化方向
表格
| 优化点 | 方案 |
|---|---|
| 智能调价 | 对接内部 ERP，当竞品降价超 8% 时自动生成调价单 |
| 销量反查 | 通过订单号样本分析竞品真实销量时间分布 |
| 广告监控 | 追踪竞品关键词排名和广告落地页变化，反推投放策略 |
| 多模态分析 | 结合商品主图变化识别品牌升级信号 |
| 知识图谱 | 构建"品牌-品类-价格带"知识图谱，挖掘关联竞争关系 |
十、总结
本文从数据采集 → 清洗存储 → 智能分析 → 实时告警 → 可视化报告五个维度，完整拆解了电商竞品全自动监控系统的搭建方案。核心要点：
- 混合采集：官方 API 保证稳定性，智能爬虫覆盖无 API 平台，代理池 + 频率控制应对反爬
- 轻量存储：DuckDB 替代 Pandas 处理百万级数据，Redis 实现智能去重
- 多维分析：SQL 窗口函数做异常检测，大模型做趋势预测，雷达图做多维对比
- 分级告警：P0/P1/P2 三级响应，确保核心竞品异动分钟级触达运营团队
- 一键部署：单 Python 脚本 + cron 即可运行，零额外基础设施依赖
部署该系统后，运营团队可将精力从"机械性复制粘贴"转移到"策略性分析决策"，真正实现从被动跟价到主动布局的竞争情报升级。
如遇任何疑问或有进一步的需求，请随时与我私信或者评论联系。
