项目4：竞品价格监控与分析，我用Python实时掌握市场动态

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或爬虫
  • 机器学习预测竞对价格走势
  • 自动化调价系统对接
  • 多维度竞品分析（评价、销量等）

下节预告

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

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

💬 加入学习交流群

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

👉 点击加入交流群

群里不定期分享：

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

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

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

👉 点击了解详情

课程导航

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

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

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

进群

嗨，扫码 进群

微信

扫一扫，分享到微信

微信分享二维码
本站总访问量次

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置：

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

编程博主，全网粉丝30w+，开源项目：python-office，我的微信：python-office