github star gitee star atomgit star PyPI Downloads AI 编程 AI 交流群

大家好,我是正在实战各种AI项目的程序员晚枫。

欢迎来到第四个实战项目!这次我们要解决电商人的永恒难题:竞品价格监控

竞争对手今天降价了吗?他们的促销策略是什么?我们的定价有没有竞争力?手动查太费劲,用Python自动化监控才是正解。


项目背景

需求场景

你是某品牌的电商运营负责人:

  1. 竞品经常偷偷改价,发现时已经丢单了
  2. 大促期间价格波动快,人工盯不过来
  3. 想分析竞品的价格策略和促销规律

目标产出

搭建价格监控系统,自动抓取、分析竞品数据并预警。


模拟数据准备

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# 生成模拟竞品价格数据(实际项目中用爬虫获取)
np.random.seed(42)

def generate_competitor_data(days=90):
"""生成模拟竞品价格数据"""
dates = pd.date_range(end=datetime.now(), periods=days, freq='D')

competitors = ['竞对A', '竞对B', '竞对C', '竞对D']
products = [
{'name': '无线耳机Pro', 'our_price': 299, 'category': '数码'},
{'name': '智能手表X', 'our_price': 599, 'category': '数码'},
{'name': '蓝牙音箱', 'our_price': 199, 'category': '数码'},
{'name': '充电宝20000mAh', 'our_price': 129, 'category': '数码'},
]

data = []
for date in dates:
for product in products:
for competitor in competitors:
# 基础价格(围绕我们价格的±20%波动)
base_factor = np.random.uniform(0.85, 1.15)
base_price = product['our_price'] * base_factor

# 周末促销(概率30%,降价10-20%)
if date.weekday() >= 5 and np.random.random() < 0.3:
promo_discount = np.random.uniform(0.8, 0.9)
price = int(base_price * promo_discount)
is_promo = True
else:
price = int(base_price)
is_promo = False

# 偶尔的价格波动
if np.random.random() < 0.1:
price += np.random.randint(-20, 21)
price = max(price, 50) # 最低限价

data.append({
'日期': date,
'产品': product['name'],
'品类': product['category'],
'竞对': competitor,
'价格': price,
'是否促销': is_promo,
'我方价格': product['our_price']
})

return pd.DataFrame(data)

# 生成数据
df = generate_competitor_data()
df.to_csv('competitor_prices.csv', index=False, encoding='utf-8-sig')
print(f"生成了 {len(df)} 条价格记录")
print(f"涉及 {df['产品'].nunique()} 个产品,{df['竞对'].nunique()} 个竞对")

价格监控分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class PriceMonitor:
def __init__(self, data_path):
self.df = pd.read_csv(data_path, encoding='utf-8-sig')
self.df['日期'] = pd.to_datetime(self.df['日期'])
print(f"加载了 {len(self.df)} 条价格记录")

def calculate_metrics(self):
"""计算价格指标"""
# 计算与竞对的价格差异
self.df['价差'] = self.df['价格'] - self.df['我方价格']
self.df['价差率'] = (self.df['价格'] - self.df['我方价格']) / self.df['我方价格'] * 100

# 按产品和竞对汇总
summary = self.df.groupby(['产品', '竞对']).agg({
'价格': ['mean', 'min', 'max', 'std'],
'价差': 'mean',
'价差率': 'mean',
'是否促销': 'sum'
}).round(2)

return summary

def competitive_analysis(self):
"""竞争力分析"""
print("\n【竞争力分析报告】")
print("="*60)

for product in self.df['产品'].unique():
product_df = self.df[self.df['产品'] == product]
our_price = product_df['我方价格'].iloc[0]

print(f"\n📱 {product} (我方定价: ¥{our_price})")
print("-" * 40)

comp_stats = product_df.groupby('竞对').agg({
'价格': 'mean',
'价差率': 'mean',
'是否促销': 'sum'
}).round(2)

# 判断竞争态势
avg_comp_price = comp_stats['价格'].mean()
if our_price < avg_comp_price * 0.95:
status = "🟢 价格优势"
elif our_price > avg_comp_price * 1.05:
status = "🔴 价格劣势"
else:
status = "🟡 价格持平"

print(f"整体态势: {status}")
print(f"竞对均价: ¥{avg_comp_price:.0f}")
print("\n各竞对详情:")
for comp, row in comp_stats.iterrows():
diff_symbol = "↑" if row['价差率'] > 0 else "↓"
print(f" {comp}: ¥{row['价格']:.0f} ({diff_symbol}{abs(row['价差率']):.1f}%) "
f"促销{int(row['是否促销'])}次")

def detect_price_changes(self, days=7):
"""检测近期价格变动"""
recent_date = self.df['日期'].max() - timedelta(days=days)
recent_df = self.df[self.df['日期'] >= recent_date]

print(f"\n【近{days}天价格变动预警】")
print("="*60)

changes = []
for product in recent_df['产品'].unique():
for competitor in recent_df['竞对'].unique():
comp_df = recent_df[(recent_df['产品'] == product) &
(recent_df['竞对'] == competitor)]
if len(comp_df) >= 2:
first_price = comp_df.iloc[0]['价格']
last_price = comp_df.iloc[-1]['价格']
change = last_price - first_price
change_pct = change / first_price * 100

if abs(change_pct) > 5: # 变动超过5%才报告
changes.append({
'产品': product,
'竞对': competitor,
'原价格': first_price,
'现价格': last_price,
'变动': change,
'变动率': change_pct
})

if changes:
changes_df = pd.DataFrame(changes)
changes_df = changes_df.sort_values('变动率', ascending=False)

for _, row in changes_df.head(10).iterrows():
direction = "📈 涨价" if row['变动'] > 0 else "📉 降价"
print(f"{direction} | {row['产品']} | {row['竞对']} | "
f"¥{row['原价格']} → ¥{row['现价格']} ({row['变动率']:+.1f}%)")
else:
print("✅ 近期无重大价格变动")

return changes

def promotion_pattern(self):
"""促销规律分析"""
print("\n【促销规律分析】")
print("="*60)

# 按星期统计促销频率
self.df['星期'] = self.df['日期'].dt.day_name()
promo_by_day = self.df.groupby(['竞对', '星期'])['是否促销'].mean() * 100

print("\n各竞对促销偏好(按星期):")
for competitor in self.df['竞对'].unique():
print(f"\n {competitor}:")
comp_promo = promo_by_day[competitor].sort_values(ascending=False)
for day, pct in comp_promo.head(3).items():
bar = "█" * int(pct / 5)
print(f" {day}: {bar} {pct:.1f}%")

# 促销力度分析
promo_df = self.df[self.df['是否促销'] == True]
if len(promo_df) > 0:
promo_power = promo_df.groupby('竞对').apply(
lambda x: ((x['我方价格'] - x['价格']) / x['我方价格'] * 100).mean()
).round(1)

print("\n促销力度排行(平均折扣深度):")
for comp, discount in promo_power.sort_values(ascending=False).items():
print(f" {comp}: {discount:.1f}% off")

可视化分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
import matplotlib.pyplot as plt
import seaborn as sns

class PriceVisualizer:
def __init__(self, monitor):
self.monitor = monitor
plt.style.use('seaborn-v0_8')

def plot_price_trends(self):
"""价格趋势对比图"""
fig, axes = plt.subplots(2, 2, figsize=(16, 10))
axes = axes.flatten()

products = self.monitor.df['产品'].unique()

for idx, product in enumerate(products[:4]):
ax = axes[idx]
product_df = self.monitor.df[self.monitor.df['产品'] == product]
our_price = product_df['我方价格'].iloc[0]

# 绘制各竞对的价格线
for competitor in product_df['竞对'].unique():
comp_df = product_df[product_df['竞对'] == competitor]
ax.plot(comp_df['日期'], comp_df['价格'],
label=competitor, alpha=0.7, linewidth=1.5)

# 绘制我方价格线
ax.axhline(y=our_price, color='red', linestyle='--',
linewidth=2, label='Our Price')

ax.set_title(f'{product}', fontsize=12, fontweight='bold')
ax.set_ylabel('Price (CNY)')
ax.legend(fontsize=8)
ax.grid(True, alpha=0.3)

plt.suptitle('Competitor Price Trends by Product', fontsize=16, fontweight='bold')
plt.tight_layout()
plt.savefig('price_trends.png', dpi=300, bbox_inches='tight')
plt.show()

def plot_price_distribution(self):
"""价格分布箱线图"""
plt.figure(figsize=(12, 6))

# 准备数据
plot_data = []
labels = []

for product in self.monitor.df['产品'].unique():
product_df = self.monitor.df[self.monitor.df['产品'] == product]
for competitor in product_df['竞对'].unique():
comp_prices = product_df[product_df['竞对'] == competitor]['价格'].values
plot_data.append(comp_prices)
labels.append(f"{product}\n{competitor}")

bp = plt.boxplot(plot_data, labels=labels, patch_artist=True)

# 美化
colors = plt.cm.Set3(np.linspace(0, 1, len(bp['boxes'])))
for patch, color in zip(bp['boxes'], colors):
patch.set_facecolor(color)

plt.xticks(rotation=45, ha='right')
plt.ylabel('Price (CNY)')
plt.title('Price Distribution by Product & Competitor', fontsize=14, fontweight='bold')
plt.grid(True, alpha=0.3, axis='y')
plt.tight_layout()
plt.savefig('price_distribution.png', dpi=300, bbox_inches='tight')
plt.show()

def plot_competitive_matrix(self):
"""竞争力矩阵热力图"""
# 计算每个产品的平均价差率
matrix = self.monitor.df.groupby(['产品', '竞对'])['价差率'].mean().unstack()

plt.figure(figsize=(10, 6))
sns.heatmap(matrix, annot=True, fmt='.1f', cmap='RdYlGn_r',
center=0, cbar_kws={'label': 'Price Difference %'})
plt.title('Competitive Price Matrix (% vs Our Price)',
fontsize=14, fontweight='bold')
plt.xlabel('Competitor')
plt.ylabel('Product')
plt.tight_layout()
plt.savefig('competitive_matrix.png', dpi=300, bbox_inches='tight')
plt.show()

def plot_promotion_calendar(self):
"""促销日历图"""
# 按日期统计促销次数
daily_promo = self.monitor.df.groupby('日期')['是否促销'].sum()

plt.figure(figsize=(14, 4))
plt.bar(daily_promo.index, daily_promo.values,
color='coral', alpha=0.7, width=0.8)
plt.title('Daily Promotion Count (All Competitors)',
fontsize=14, fontweight='bold')
plt.xlabel('Date')
plt.ylabel('Number of Promotions')
plt.grid(True, alpha=0.3, axis='y')
plt.tight_layout()
plt.savefig('promotion_calendar.png', dpi=300, bbox_inches='tight')
plt.show()

自动预警系统

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
class PriceAlertSystem:
def __init__(self, monitor):
self.monitor = monitor
self.alerts = []

def check_alerts(self):
"""检查各类预警条件"""
print("\n【价格预警系统】")
print("="*60)

# 1. 低价预警(竞对低于我方15%以上)
low_price = self.monitor.df[
self.monitor.df['价差率'] < -15
][['日期', '产品', '竞对', '价格', '价差率']].copy()

if len(low_price) > 0:
print(f"\n🔴 低价预警: 发现 {len(low_price)} 条竞对低价记录")
print("最近5条:")
for _, row in low_price.tail(5).iterrows():
print(f" {row['日期'].strftime('%m-%d')} | {row['产品']} | "
f"{row['竞对']} | ¥{row['价格']} ({row['价差率']:.1f}%)")

# 2. 促销预警(竞对频繁促销)
promo_freq = self.monitor.df.groupby('竞对')['是否促销'].sum()
high_promo = promo_freq[promo_freq > promo_freq.mean() + promo_freq.std()]

if len(high_promo) > 0:
print(f"\n🟠 促销预警: 以下竞对促销频率异常")
for comp, count in high_promo.items():
print(f" {comp}: {count} 次促销")

# 3. 价格异动预警(单日变动>10%)
latest_date = self.monitor.df['日期'].max()
latest_df = self.monitor.df[self.monitor.df['日期'] == latest_date]
prev_df = self.monitor.df[self.monitor.df['日期'] == latest_date - timedelta(days=1)]

if len(prev_df) > 0:
for _, curr in latest_df.iterrows():
prev = prev_df[(prev_df['产品'] == curr['产品']) &
(prev_df['竞对'] == curr['竞对'])]
if len(prev) > 0:
change = (curr['价格'] - prev['价格'].iloc[0]) / prev['价格'].iloc[0] * 100
if abs(change) > 10:
direction = "大涨" if change > 0 else "大跌"
print(f"\n⚠️ 异动预警: {curr['产品']} - {curr['竞对']} 价格{direction} {abs(change):.1f}%")

# 4. 综合建议
print("\n💡 应对建议:")

# 找出最激进的竞对
avg_discount = self.monitor.df.groupby('竞对')['价差率'].mean()
most_aggressive = avg_discount.idxmin()

print(f" 1. 重点关注: {most_aggressive} 整体价格低 {abs(avg_discount.min()):.1f}%")
print(f" 2. 建议跟进促销的产品: {low_price['产品'].value_counts().index[0] if len(low_price) > 0 else '暂无'}")
print(f" 3. 可考虑提价的产品: 竞争力强的产品")

return self.alerts

def generate_report(self, output_file='price_monitor_report.html'):
"""生成HTML报告"""
html_content = f"""
<!DOCTYPE html>
<html>
<head>
<title>竞品价格监控报告</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 40px; }}
h1 {{ color: #333; }}
h2 {{ color: #666; border-bottom: 2px solid #ddd; padding-bottom: 10px; }}
table {{ border-collapse: collapse; width: 100%; margin: 20px 0; }}
th, td {{ border: 1px solid #ddd; padding: 12px; text-align: left; }}
th {{ background-color: #4CAF50; color: white; }}
tr:nth-child(even) {{ background-color: #f2f2f2; }}
.alert {{ color: red; font-weight: bold; }}
.good {{ color: green; }}
</style>
</head>
<body>
<h1>竞品价格监控报告</h1>
<p>生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}</p>

<h2>监控概览</h2>
<ul>
<li>监控产品数: {self.monitor.df['产品'].nunique()}</li>
<li>监控竞对数: {self.monitor.df['竞对'].nunique()}</li>
<li>数据时间范围: {self.monitor.df['日期'].min().strftime('%Y-%m-%d')} ~ {self.monitor.df['日期'].max().strftime('%Y-%m-%d')}</li>
</ul>

<h2>价格对比表</h2>
{self.monitor.calculate_metrics().to_html()}
</body>
</html>
"""

with open(output_file, 'w', encoding='utf-8') as f:
f.write(html_content)

print(f"\n ✓ HTML报告已生成: {output_file}")

主程序

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def main():
"""主程序"""
print("="*60)
print("竞品价格监控与分析系统")
print("="*60)

# 1. 加载数据
print("\n[1/5] 加载价格数据...")
monitor = PriceMonitor('competitor_prices.csv')

# 2. 竞争力分析
print("\n[2/5] 竞争力分析...")
monitor.calculate_metrics()
monitor.competitive_analysis()

# 3. 价格变动检测
print("\n[3/5] 检测价格变动...")
monitor.detect_price_changes(days=7)

# 4. 促销规律
print("\n[4/5] 促销规律分析...")
monitor.promotion_pattern()

# 5. 可视化
print("\n[5/5] 生成可视化图表...")
visualizer = PriceVisualizer(monitor)
visualizer.plot_price_trends()
visualizer.plot_price_distribution()
visualizer.plot_competitive_matrix()
visualizer.plot_promotion_calendar()

# 6. 预警系统
print("\n[6/6] 运行预警系统...")
alert_system = PriceAlertSystem(monitor)
alert_system.check_alerts()
alert_system.generate_report()

print("\n" + "="*60)
print("分析完成!")
print("="*60)

if __name__ == "__main__":
main()

项目总结

学到的技能

  • ✅ 竞品数据采集与处理
  • ✅ 价格竞争力分析
  • ✅ 促销规律挖掘
  • ✅ 自动预警系统设计

实战价值

场景应用
日常监控定时抓取竞品价格,发现异常立即通知
大促备战分析历史促销规律,制定应对策略
定价决策基于竞对价格分布,确定最优定价区间
市场洞察识别竞对的运营节奏和战略意图

扩展方向

  • 接入真实电商平台API或爬虫
  • 机器学习预测竞对价格走势
  • 自动化调价系统对接
  • 多维度竞品分析(评价、销量等)

进阶:价格弹性分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 计算价格弹性系数
import pandas as pd
import numpy as np
from scipy import stats

# 价格和销量数据
df = pd.DataFrame({
'price': [100, 110, 95, 120, 90, 105, 115, 85],
'quantity': [500, 420, 580, 350, 620, 460, 380, 700]
})

# 对数回归计算弹性
log_price = np.log(df['price'])
log_quantity = np.log(df['quantity'])
slope, intercept, r_value, p_value, std_err = stats.linregress(log_price, log_quantity)

print(f"价格弹性系数: {slope:.2f}")
# |弹性| > 1 → 弹性商品(降价能增收)
# |弹性| < 1 → 刚性商品(可以提价)
print(f"R²: {r_value**2:.3f}")

价格监控告警

1
2
3
4
5
6
7
8
9
10
11
12
13
# 设置价格异常告警
def check_price_alert(current_prices, baseline_prices, threshold=0.1):
alerts = []
for product in current_prices:
curr = current_prices[product]
base = baseline_prices[product]
change = (curr - base) / base

if abs(change) > threshold:
direction = '降价' if change < 0 else '涨价'
alerts.append(f"⚠️ {product} {direction}{abs(change)*100:.1f}% (现价: ¥{curr})")

return alerts

避坑指南

❌ 坑1:促销价 vs 日常价混淆

1
2
3
4
5
# 需要区分促销和正常调价
# 方法:看价格是否在短时间内回落
df['is_promo'] = df['price'] < df.groupby('product')['price'].transform(
lambda x: x.rolling(30, min_periods=1).quantile(0.25)
)

竞品分析框架

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
# 竞品价格分析完整流程
import pandas as pd
import numpy as np

np.random.seed(42)

# 模拟竞品价格数据(3个竞品,50个商品,30天)
products = [f'商品{i:03d}' for i in range(1, 51)]
competitors = ['我方', '竞品A', '竞品B']

price_data = []
for product in products:
base_price = np.random.uniform(50, 500)
for comp in competitors:
for day in range(30):
price = base_price * (1 + np.random.uniform(-0.1, 0.1))
if comp != '我方':
price *= np.random.uniform(0.85, 1.15)
price_data.append({
'product': product,
'competitor': comp,
'day': day,
'price': round(price, 2)
})

df = pd.DataFrame(price_data)

# 1. 价格对比分析
price_comparison = df.pivot_table(values='price', index='product', columns='competitor', aggfunc='mean')
price_comparison['price_diff_a'] = ((price_comparison['我方'] - price_comparison['竞品A']) / price_comparison['竞品A'] * 100).round(1)
price_comparison['price_diff_b'] = ((price_comparison['我方'] - price_comparison['竞品B']) / price_comparison['竞品B'] * 100).round(1)

print("=== 价格差异分析 ===")
print(f"我方价格高于竞品A: {(price_comparison['price_diff_a'] > 0).sum()}个商品")
print(f"我方价格低于竞品A: {(price_comparison['price_diff_a'] < 0).sum()}个商品")
print(f"平均价格差异(vs竞品A): {price_comparison['price_diff_a'].mean():.1f}%")

# 2. 价格波动分析
price_volatility = df.groupby('competitor')['price'].std()
print(f"\n价格波动(标准差): {price_volatility.round(2)}")

# 3. 找出竞品大幅降价的商品
for comp in ['竞品A', '竞品B']:
comp_data = df[df['competitor'] == comp]
price_change = comp_data.groupby('product')['price'].agg(['first', 'last'])
price_change['change_pct'] = ((price_change['last'] - price_change['first']) / price_change['first'] * 100)
big_drops = price_change[price_change['change_pct'] < -5]
if len(big_drops) > 0:
print(f"\n⚠️ {comp}大幅降价商品:")
print(big_drops[['change_pct']].round(1))

价格策略建议

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 基于分析结果的价格策略
def pricing_strategy(product, our_price, comp_prices, cost):
'''基于竞品价格的定价策略'''
min_comp = min(comp_prices)
max_comp = max(comp_prices)
avg_comp = np.mean(comp_prices)

margin = (our_price - cost) / our_price # 毛利率

if our_price > max_comp * 1.1:
return f"⚠️ 定价过高(比最高竞品贵10%+),建议降至{max_comp*1.05:.0f}"
elif our_price < min_comp * 0.9 and margin < 0.15:
return f"⚠️ 定价过低且毛利不足,建议提价至{min_comp*0.95:.0f}"
elif margin < 0.1:
return f"⚠️ 毛利过低({margin*100:.1f}%),需评估是否调整"
else:
return f"✅ 定价合理(毛利{margin*100:.1f}%,价格区间内)"

竞品分析报告模板

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def competitor_report(our_data, comp_data):
'''生成竞品分析报告'''
import pandas as pd

# 合并数据
merged = our_data.merge(comp_data, on='product_id', suffixes=('_ours', '_comp'))

# 价格对比
merged['price_diff'] = merged['price_ours'] - merged['price_comp']
merged['price_diff_pct'] = (merged['price_diff'] / merged['price_comp'] * 100).round(1)

report = "=== 竞品价格分析报告 ===

"

# 总体对比
report += f"我方均价: ¥{merged['price_ours'].mean():.0f}
"
report += f"竞品均价: ¥{merged['price_comp'].mean():.0f}
"
report += f"价差: {merged['price_diff_pct'].mean():.1f}%

"

# 价格优势商品
cheaper = merged[merged['price_diff_pct'] < -5]
report += f"我方价格优势商品(低5%+): {len(cheaper)}
"

# 价格劣势商品
expensive = merged[merged['price_diff_pct'] > 5]
report += f"我方价格劣势商品(高5%+): {len(expensive)}

"

# 建议
report += "=== 建议 ===
"
if len(expensive) > 0:
report += f"- {len(expensive)}个商品需要考虑降价或提升价值
"
if len(cheaper) > 0:
report += f"- {len(cheaper)}个商品有提价空间
"

return report

print(competitor_report(our_df, comp_df))

竞品监控完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
import pandas as pd
import numpy as np
from scipy import stats

np.random.seed(42)

# 模拟竞品价格数据
products = [f'商品{i:03d}' for i in range(1, 31)]
data = []
for prod in products:
base = np.random.uniform(50, 500)
for comp in ['我方', '竞品A', '竞品B', '竞品C']:
for day in range(30):
price = base * (1 + np.random.uniform(-0.08, 0.08))
data.append({'product': prod, 'competitor': comp, 'day': day, 'price': round(price, 2)})

df = pd.DataFrame(data)

# ===== 价格对比 =====
avg_price = df.pivot_table(values='price', index='product', columns='competitor', aggfunc='mean')
print('=== 各方均价对比 ===')
print(avg_price.describe().round(0))

# ===== 价格差异分析 =====
for comp in ['竞品A', '竞品B', '竞品C']:
diff = ((avg_price['我方'] - avg_price[comp]) / avg_price[comp] * 100).round(1)
print(f'\n我方 vs {comp}:')
print(f' 我方更贵: {(diff > 0).sum()}个商品')
print(f' 我方更便宜: {(diff < 0).sum()}个商品')
print(f' 平均价差: {diff.mean():.1f}%')

# ===== 价格波动分析 =====
volatility = df.groupby('competitor')['price'].std().round(2)
print(f'\n价格波动性(标准差):')
print(volatility)

# ===== 降价预警 =====
for comp in ['竞品A', '竞品B', '竞品C']:
comp_df = df[df['competitor'] == comp]
price_change = comp_df.groupby('product')['price'].agg(['first', 'last'])
price_change['change'] = ((price_change['last'] - price_change['first']) / price_change['first'] * 100).round(1)
big_drops = price_change[price_change['change'] < -5]
if len(big_drops) > 0:
print(f'\n⚠️ {comp}大幅降价: {len(big_drops)}个商品')
print(big_drops[['change']].sort_values('change'))

竞品监控系统架构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import pandas as pd
import numpy as np
from scipy import stats

np.random.seed(42)

# ===== 1. 竞品价格数据采集 =====
# 实际工作中可以用爬虫或API获取,这里模拟数据
products = [f'SKU{i:04d}' for i in range(1, 101)]
competitors = ['我方', '竞品A', '竞品B', '竞品C']

price_records = []
for prod in products:
base_price = np.random.uniform(30, 500)
category = np.random.choice(['电子', '服装', '食品', '家居'], p=[0.3, 0.3, 0.2, 0.2])
for comp in competitors:
markup = np.random.uniform(-0.15, 0.15)
for day in range(60):
daily_price = base_price * (1 + markup) * (1 + np.random.uniform(-0.05, 0.05))
# 模拟促销
if np.random.random() < 0.05:
daily_price *= 0.8 # 8折促销
price_records.append({
'product_id': prod,
'category': category,
'competitor': comp,
'day': day,
'price': round(daily_price, 2),
'is_promo': np.random.random() < 0.05
})

df = pd.DataFrame(price_records)

# ===== 2. 价格监控指标 =====
# 均价对比
avg_prices = df.pivot_table(values='price', index='product_id', columns='competitor', aggfunc='mean')

# 价格差异
for comp in ['竞品A', '竞品B', '竞品C']:
avg_prices[f'vs_{comp}_diff%'] = ((avg_prices['我方'] - avg_prices[comp]) / avg_prices[comp] * 100).round(1)

# 价格波动
price_volatility = df.groupby('competitor')['price'].std().round(2)
print('价格波动性(标准差):')
print(price_volatility)

# 促销频率
promo_freq = df[df['is_promo']].groupby('competitor').size() / df.groupby('competitor').size() * 100
print(f'\n促销频率:')
print(promo_freq.round(1))

# ===== 3. 异常价格检测 =====
def detect_price_anomaly(df, product_id, threshold=2):
# Z-score方法检测异常价格
prod_data = df[df['product_id'] == product_id]
for comp in prod_data['competitor'].unique():
comp_prices = prod_data[prod_data['competitor'] == comp]['price']
z_scores = np.abs(stats.zscore(comp_prices))
anomalies = comp_prices[z_scores > threshold]
if len(anomalies) > 0:
print(f'{product_id} - {comp}: {len(anomalies)}个异常价格')

# ===== 4. 价格策略建议 =====
def pricing_recommendation(row):
# 基于竞品价格和自身定位给出建议
our_price = row['我方']
min_comp = min(row[['竞品A', '竞品B', '竞品C']].min(), our_price * 1.3)
max_comp = max(row[['竞品A', '竞品B', '竞品C']].max(), our_price * 0.7)

if our_price > max_comp * 1.1:
return '降价建议'
elif our_price < min_comp * 0.85:
return '可提价'
else:
return '价格合理'

avg_prices['recommendation'] = avg_prices.apply(pricing_recommendation, axis=1)
print('\n=== 价格策略建议 ===')
print(avg_prices['recommendation'].value_counts())

价格弹性系数计算

1
2
3
4
5
6
7
8
9
10
11
# 计算价格弹性(需求量变化率 / 价格变化率)
elasticity = df.groupby('product_id').apply(
lambda x: pd.Series({
'price_elasticity': np.corrcoef(x['price'], x.groupby('day')['price'].count())[0,1] if len(x) > 2 else np.nan
})
).reset_index()

# 弹性>1: 弹性商品(降价增收)
# 弹性<1: 刚性商品(可提价)
print('弹性商品(降价可增收):', (elasticity['price_elasticity'] > 1).sum())
print('刚性商品(可提价):', (elasticity['price_elasticity'] < 1).sum())

下节预告

下一项目是数据看板搭建,学习如何用Dash/Streamlit制作交互式数据看板,让数据分析成果一目了然。

👉 继续阅读:项目5-交互式数据看板搭建


💬 加入学习交流群

扫码加入Python学习交流群,和数千名同学一起进步:

👉 点击加入交流群

群里不定期分享:

  • 数据分析实战案例
  • Python学习资料
  • 求职面试经验
  • 行业最新动态

推荐:AI Python数据分析实战营

🎁 限时福利:送《利用Python进行数据分析》实体书

👉 点击了解详情


课程导航

上一篇: 项目3-库存分析与销量预测

下一篇: 项目5-交互式数据看板搭建


PS:知己知彼,百战不殆。监控竞品是运营的必修课,用Python自动化后,你可以把精力放在更重要的策略制定上。



📚 推荐教材

主教材《Excel+Python 飞速搞定数据分析与处理(图灵出品)》

💬 联系我

平台账号/链接
微信扫码加好友
微博@程序员晚枫
知乎@程序员晚枫
抖音@程序员晚枫
小红书@程序员晚枫
B 站Python 自动化办公社区

主营业务:AI 编程培训、企业内训、技术咨询

🎓 AI 编程实战课程

想系统学习 AI 编程?程序员晚枫的 AI 编程实战课 帮你从零上手!