为了打破价格歧视，我用AI做了一个智能购物助手【B站AI创造公开赛】

在电商购物时，你是否经常遇到同一件商品在不同平台、不同时间价格差异巨大的情况？这就是典型的价格歧视现象。作为开发者，我决定用AI技术来解决这个问题，打造一个能够自动比价、分析价格趋势的智能购物助手。本文将完整分享从需求分析到技术实现的全部过程，包含可运行的代码示例和实战经验。

1. 项目背景与核心概念

1.1 什么是价格歧视

价格歧视是指商家针对不同消费者群体制定不同价格策略的行为。在电商领域，这种策略表现为：

• 平台差异：同一商品在淘宝、京东、拼多多等平台价格不同
• 时间差异：同一商品在不同时间段（如促销季、工作日）价格波动
• 用户差异：基于用户画像、浏览历史等因素个性化定价

1.2 智能购物助手的价值

传统比价工具往往只能提供简单的价格对比，而基于AI的智能购物助手能够：

• 实时监控多个电商平台的价格变化
• 分析历史价格趋势，预测最佳购买时机
• 识别虚假促销和价格陷阱
• 提供个性化的购物建议

1.3 技术选型思路

本项目采用Python作为主要开发语言，结合以下技术栈：

• 数据采集：Requests + Selenium用于网页数据抓取
• 数据处理：Pandas + NumPy进行数据清洗和分析
• AI模型：Scikit-learn用于价格预测模型
• 可视化：Matplotlib + Pyecharts展示价格趋势
• 部署：Flask构建Web服务接口

2. 环境准备与版本说明

2.1 开发环境配置

# requirements.txt # Python版本要求：3.8+ requests==2.31.0 selenium==4.15.0 pandas==2.1.3 numpy==1.25.2 scikit-learn==1.3.0 matplotlib==3.7.2 flask==2.3.3 beautifulsoup4==4.12.2 schedule==1.2.0

2.2 浏览器驱动配置

# 下载Chrome驱动，版本需与本地Chrome浏览器匹配 # 下载地址：https://chromedriver.chromium.org/ import os from selenium import webdriver def setup_driver(): # 配置Chrome选项 options = webdriver.ChromeOptions() options.add_argument('--headless') # 无头模式 options.add_argument('--no-sandbox') options.add_argument('--disable-dev-shm-usage') # 设置驱动路径 driver_path = '/path/to/chromedriver' # 根据实际路径修改 driver = webdriver.Chrome(executable_path=driver_path, options=options) return driver

2.3 项目目录结构

smart-shopping-assistant/ ├── src/ │ ├── crawlers/ # 爬虫模块 │ ├── models/ # AI模型 │ ├── analysis/ # 数据分析 │ ├── utils/ # 工具函数 │ └── app.py # Flask应用 ├── data/ │ ├── raw/ # 原始数据 │ ├── processed/ # 处理后的数据 │ └── models/ # 训练好的模型 ├── config/ │ └── settings.py # 配置文件 └── tests/ # 测试文件

3. 核心功能模块设计

3.1 多平台数据采集器

# src/crawlers/base_crawler.py import requests from selenium import webdriver from bs4 import BeautifulSoup import time import json from abc import ABC, abstractmethod class BaseCrawler(ABC): def __init__(self): self.session = requests.Session() self.set_headers() def set_headers(self): """设置请求头，模拟真实浏览器访问""" self.headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', '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', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', } @abstractmethod def search_product(self, keyword): """搜索商品抽象方法""" pass @abstractmethod def get_product_details(self, product_url): """获取商品详情抽象方法""" pass # src/crawlers/taobao_crawler.py class TaobaoCrawler(BaseCrawler): def __init__(self): super().__init__() self.base_url = "https://s.taobao.com" def search_product(self, keyword): """淘宝商品搜索""" search_url = f"{self.base_url}/search" params = { 'q': keyword, 's': '0' # 从第0条开始 } try: response = self.session.get(search_url, params=params, headers=self.headers) soup = BeautifulSoup(response.text, 'html.parser') products = [] items = soup.select('.item.J_MouserOnverReq') for item in items[:10]: # 取前10个商品 product = { 'title': item.select_one('.title a').text.strip(), 'price': item.select_one('.price strong').text, 'sales': item.select_one('.deal-cnt').text, 'shop': item.select_one('.shopname').text, 'url': item.select_one('.title a')['href'] } products.append(product) return products except Exception as e: print(f"淘宝搜索出错: {e}") return [] # src/crawlers/jd_crawler.py class JDCrawler(BaseCrawler): def __init__(self): super().__init__() self.base_url = "https://search.jd.com/Search" def search_product(self, keyword): """京东商品搜索""" params = { 'keyword': keyword, 'enc': 'utf-8' } try: response = self.session.get(self.base_url, params=params, headers=self.headers) soup = BeautifulSoup(response.text, 'html.parser') products = [] items = soup.select('.gl-item') for item in items[:10]: product = { 'title': item.select_one('.p-name em').text, 'price': item.select_one('.p-price i').text, 'shop': item.select_one('.p-shopnum a').text if item.select_one('.p-shopnum a') else '自营', 'url': item.select_one('.p-img a')['href'] } products.append(product) return products except Exception as e: print(f"京东搜索出错: {e}") return []

3.2 价格数据存储与管理

# src/utils/database.py import sqlite3 import pandas as pd from datetime import datetime class PriceDatabase: def __init__(self, db_path='data/prices.db'): self.db_path = db_path self.init_database() def init_database(self): """初始化数据库表结构""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_name TEXT NOT NULL, platform TEXT NOT NULL, price REAL NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, url TEXT, shop_name TEXT ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS price_alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_name TEXT NOT NULL, target_price REAL NOT NULL, current_price REAL, status TEXT DEFAULT 'active', created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() conn.close() def insert_price_record(self, product_data): """插入价格记录""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO price_history (product_name, platform, price, url, shop_name) VALUES (?, ?, ?, ?, ?) ''', ( product_data['name'], product_data['platform'], product_data['price'], product_data['url'], product_data['shop'] )) conn.commit() conn.close() def get_price_history(self, product_name, days=30): """获取商品价格历史""" conn = sqlite3.connect(self.db_path) query = ''' SELECT platform, price, timestamp FROM price_history WHERE product_name = ? AND timestamp >= datetime('now', ?) ORDER BY timestamp DESC ''' df = pd.read_sql_query(query, conn, params=(product_name, f'-{days} days')) conn.close() return df

3.3 AI价格预测模型

# src/models/price_predictor.py import pandas as pd import numpy as np from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import train_test_split from sklearn.metrics import mean_absolute_error, mean_squared_error import joblib from datetime import datetime, timedelta class PricePredictor: def __init__(self): self.model = None self.features = ['day_of_week', 'month', 'is_weekend', 'days_to_holiday'] def prepare_features(self, df): """准备特征数据""" df['timestamp'] = pd.to_datetime(df['timestamp']) df['day_of_week'] = df['timestamp'].dt.dayofweek df['month'] = df['timestamp'].dt.month df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int) # 简单的节假日处理（实际项目中需要更复杂的逻辑） holidays = ['01-01', '05-01', '10-01'] # 元旦、劳动节、国庆节 df['days_to_holiday'] = df['timestamp'].apply( lambda x: self.calculate_days_to_holiday(x, holidays) ) return df def calculate_days_to_holiday(self, date, holidays): """计算距离最近节日的天数""" date_str = date.strftime('%m-%d') min_days = 365 for holiday in holidays: holiday_date = datetime.strptime(f"{date.year}-{holiday}", "%Y-%m-%d") days_diff = abs((date - holiday_date).days) min_days = min(min_days, days_diff) return min_days def train_model(self, price_data): """训练价格预测模型""" df = self.prepare_features(price_data) # 创建滞后特征 for lag in [1, 3, 7]: df[f'price_lag_{lag}'] = df['price'].shift(lag) df = df.dropna() X = df[self.features + ['price_lag_1', 'price_lag_3', 'price_lag_7']] y = df['price'] X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 ) self.model = RandomForestRegressor(n_estimators=100, random_state=42) self.model.fit(X_train, y_train) # 评估模型 y_pred = self.model.predict(X_test) mae = mean_absolute_error(y_test, y_pred) print(f"模型MAE: {mae:.2f}") # 保存模型 joblib.dump(self.model, 'data/models/price_predictor.pkl') def predict_price(self, historical_data, days_ahead=7): """预测未来价格""" if self.model is None: self.model = joblib.load('data/models/price_predictor.pkl') last_date = historical_data['timestamp'].max() future_dates = [last_date + timedelta(days=i) for i in range(1, days_ahead+1)] predictions = [] for date in future_dates: features = self.create_future_features(date, historical_data) prediction = self.model.predict([features])[0] predictions.append({'date': date, 'predicted_price': prediction}) return pd.DataFrame(predictions) def create_future_features(self, date, historical_data): """为未来日期创建特征""" day_of_week = date.weekday() month = date.month is_weekend = 1 if day_of_week in [5, 6] else 0 # 节假日计算 holidays = ['01-01', '05-01', '10-01'] days_to_holiday = self.calculate_days_to_holiday(date, holidays) # 使用最近的价格作为滞后特征 recent_prices = historical_data['price'].tail(7).values price_lag_1 = recent_prices[-1] if len(recent_prices) >= 1 else historical_data['price'].mean() price_lag_3 = recent_prices[-3] if len(recent_prices) >= 3 else historical_data['price'].mean() price_lag_7 = recent_prices[-7] if len(recent_prices) >= 7 else historical_data['price'].mean() return [day_of_week, month, is_weekend, days_to_holiday, price_lag_1, price_lag_3, price_lag_7]

4. 完整系统集成与实战

4.1 核心业务逻辑实现

# src/app/core.py import schedule import time from threading import Thread from datetime import datetime from crawlers.taobao_crawler import TaobaoCrawler from crawlers.jd_crawler import JDCrawler from utils.database import PriceDatabase from models.price_predictor import PricePredictor class SmartShoppingAssistant: def __init__(self): self.taobao_crawler = TaobaoCrawler() self.jd_crawler = JDCrawler() self.database = PriceDatabase() self.predictor = PricePredictor() self.monitoring_products = [] def add_product_to_monitor(self, product_name, target_price=None): """添加监控商品""" product_info = { 'name': product_name, 'target_price': target_price, 'added_time': datetime.now() } self.monitoring_products.append(product_info) print(f"已添加监控商品: {product_name}") def search_and_compare(self, product_name): """搜索并比较价格""" print(f"正在搜索商品: {product_name}") # 多平台搜索 taobao_results = self.taobao_crawler.search_product(product_name) jd_results = self.jd_crawler.search_product(product_name) all_results = [] # 处理淘宝结果 for result in taobao_results: price = float(result['price']) record = { 'name': product_name, 'platform': '淘宝', 'price': price, 'url': result['url'], 'shop': result['shop'], 'title': result['title'] } all_results.append(record) self.database.insert_price_record(record) # 处理京东结果 for result in jd_results: price = float(result['price']) record = { 'name': product_name, 'platform': '京东', 'price': price, 'url': result['url'], 'shop': result['shop'], 'title': result['title'] } all_results.append(record) self.database.insert_price_record(record) return all_results def analyze_price_trend(self, product_name, days=30): """分析价格趋势""" history_data = self.database.get_price_history(product_name, days) if len(history_data) == 0: return None # 基本统计 min_price = history_data['price'].min() max_price = history_data['price'].max() avg_price = history_data['price'].mean() # 价格预测 predictions = self.predictor.predict_price(history_data) analysis_result = { 'product_name': product_name, 'period_days': days, 'min_price': min_price, 'max_price': max_price, 'avg_price': avg_price, 'current_price': history_data['price'].iloc[0], 'price_trend': '上涨' if history_data['price'].iloc[0] > avg_price else '下降', 'predictions': predictions.to_dict('records') } return analysis_result def start_monitoring(self, interval_hours=6): """启动定时监控""" def monitoring_job(): for product in self.monitoring_products: self.search_and_compare(product['name']) analysis = self.analyze_price_trend(product['name']) self.check_price_alert(product, analysis) # 设置定时任务 schedule.every(interval_hours).hours.do(monitoring_job) print(f"价格监控已启动，每{interval_hours}小时执行一次") # 运行调度器 while True: schedule.run_pending() time.sleep(1) def check_price_alert(self, product, analysis): """检查价格提醒""" if product['target_price'] and analysis['current_price'] <= product['target_price']: self.send_alert(product, analysis) def send_alert(self, product, analysis): """发送价格提醒""" message = f""" 🎉 价格提醒！🎉 商品：{product['name']} 当前价格：{analysis['current_price']}元 目标价格：{product['target_price']}元 达到目标，建议购买！ """ print(message) # 实际项目中可以集成邮件、短信、微信通知等 # 使用示例 if __name__ == "__main__": assistant = SmartShoppingAssistant() # 添加监控商品 assistant.add_product_to_monitor("iPhone 15", 5000) assistant.add_product_to_monitor("华为Mate 60", 6000) # 单次搜索测试 results = assistant.search_and_compare("iPhone 15") for result in results: print(f"平台: {result['platform']}, 价格: {result['price']}, 店铺: {result['shop']}") # 启动监控（在实际应用中可以在后台运行） # assistant.start_monitoring()

4.2 Web服务接口开发

# src/app.py from flask import Flask, request, jsonify, render_template from core import SmartShoppingAssistant import json app = Flask(__name__) assistant = SmartShoppingAssistant() @app.route('/') def index(): """首页""" return render_template('index.html') @app.route('/api/search', methods=['POST']) def search_product(): """商品搜索API""" data = request.json product_name = data.get('product_name') if not product_name: return jsonify({'error': '商品名称不能为空'}), 400 try: results = assistant.search_and_compare(product_name) return jsonify({'success': True, 'data': results}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/analyze', methods=['POST']) def analyze_price(): """价格分析API""" data = request.json product_name = data.get('product_name') days = data.get('days', 30) try: analysis = assistant.analyze_price_trend(product_name, days) if analysis: return jsonify({'success': True, 'data': analysis}) else: return jsonify({'error': '没有找到该商品的价格数据'}), 404 except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/monitor', methods=['POST']) def add_monitor(): """添加监控API""" data = request.json product_name = data.get('product_name') target_price = data.get('target_price') try: assistant.add_product_to_monitor(product_name, target_price) return jsonify({'success': True, 'message': '监控已添加'}) except Exception as e: return jsonify({'error': str(e)}), 500 if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000)

4.3 前端界面示例

<!-- templates/index.html --> <!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>智能购物助手</title> <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script> <style> .container { max-width: 1200px; margin: 0 auto; padding: 20px; } .search-box { margin-bottom: 30px; } .result-item { border: 1px solid #ddd; padding: 15px; margin: 10px 0; } .chart-container { height: 400px; margin: 20px 0; } </style> </head> <body> <div class="container"> <h1>智能购物助手</h1> <div class="search-box"> <input type="text" id="productInput" placeholder="输入商品名称"> <button onclick="searchProduct()">搜索比价</button> <button onclick="analyzePrice()">分析趋势</button> </div> <div id="results"></div> <div id="chart" class="chart-container"></div> </div> <script> async function searchProduct() { const productName = document.getElementById('productInput').value; const response = await fetch('/api/search', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({product_name: productName}) }); const data = await response.json(); displayResults(data.data); } function displayResults(results) { const container = document.getElementById('results'); container.innerHTML = ''; results.forEach(item => { const div = document.createElement('div'); div.className = 'result-item'; div.innerHTML = ` <h3>${item.title}</h3> <p>平台: ${item.platform} | 价格: ¥${item.price} | 店铺: ${item.shop}</p> `; container.appendChild(div); }); } async function analyzePrice() { const productName = document.getElementById('productInput').value; const response = await fetch('/api/analyze', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({product_name: productName}) }); const data = await response.json(); displayChart(data.data); } function displayChart(analysis) { const chart = echarts.init(document.getElementById('chart')); const option = { title: { text: `${analysis.product_name} 价格分析` }, tooltip: { trigger: 'axis' }, xAxis: { type: 'category', data: analysis.predictions.map(p => p.date) }, yAxis: { type: 'value' }, series: [{ data: analysis.predictions.map(p => p.predicted_price), type: 'line', smooth: true }] }; chart.setOption(option); } </script> </body> </html>

5. 部署与优化方案

5.1 生产环境部署

# docker-compose.yml version: '3.8' services: web: build: . ports: - "5000:5000" volumes: - ./data:/app/data environment: - FLASK_ENV=production - DATABASE_URL=sqlite:///data/prices.db restart: unless-stopped scheduler: build: . command: python scheduler.py volumes: - ./data:/app/data environment: - FLASK_ENV=production restart: unless-stopped # Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["python", "src/app.py"]

5.2 性能优化策略

# src/utils/cache.py import redis import json from datetime import timedelta class CacheManager: def __init__(self): self.redis_client = redis.Redis(host='localhost', port=6379, db=0) def get_cached_data(self, key): """获取缓存数据""" cached = self.redis_client.get(key) if cached: return json.loads(cached) return None def set_cached_data(self, key, data, expire_minutes=30): """设置缓存数据""" self.redis_client.setex( key, timedelta(minutes=expire_minutes), json.dumps(data) ) def clear_cache(self, pattern): """清除缓存""" keys = self.redis_client.keys(pattern) if keys: self.redis_client.delete(*keys) # 优化后的搜索函数 def optimized_search(product_name): cache_key = f"search:{product_name}" cache_manager = CacheManager() # 尝试从缓存获取 cached_results = cache_manager.get_cached_data(cache_key) if cached_results: return cached_results # 缓存未命中，执行实际搜索 results = actual_search_function(product_name) # 缓存结果 cache_manager.set_cached_data(cache_key, results) return results

6. 常见问题与解决方案

6.1 反爬虫策略应对

6.2 数据准确性保障

# src/utils/validator.py import re class DataValidator: @staticmethod def validate_price(price_str): """验证价格格式""" if not price_str: return False # 匹配数字和小数点 pattern = r'^\d+(\.\d{1,2})?$' return bool(re.match(pattern, str(price_str))) @staticmethod def clean_price(price_str): """清洗价格数据""" # 移除货币符号和空格 cleaned = re.sub(r'[^\d.]', '', str(price_str)) try: return float(cleaned) except ValueError: return None @staticmethod def detect_anomaly(prices): """检测价格异常值""" if len(prices) < 3: return [] import numpy as np prices_array = np.array(prices) Q1 = np.percentile(prices_array, 25) Q3 = np.percentile(prices_array, 75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR anomalies = [] for i, price in enumerate(prices): if price < lower_bound or price > upper_bound: anomalies.append(i) return anomalies

6.3 模型预测准确性提升

# src/models/advanced_predictor.py from sklearn.ensemble import GradientBoostingRegressor from sklearn.preprocessing import StandardScaler import numpy as np class AdvancedPricePredictor(PricePredictor): def __init__(self): super().__init__() self.scaler = StandardScaler() self.additional_features = ['seasonality', 'promotion_effect'] def enhance_features(self, df): """增强特征工程""" df = super().prepare_features(df) # 添加季节性特征 df['seasonality'] = df['month'].apply(self.calculate_seasonality) # 促销效应（简化版） df['promotion_effect'] = df['price'].rolling(7).std().fillna(0) return df def calculate_seasonality(self, month): """计算季节性因子""" # 简化的季节性调整，实际项目需要更复杂的逻辑 season_factors = { 1: 1.1, 2: 1.0, 3: 0.9, 4: 0.95, 5: 1.0, 6: 1.05, 7: 1.1, 8: 1.15, 9: 1.05, 10: 1.0, 11: 0.95, 12: 1.1 } return season_factors.get(month, 1.0) def train_advanced_model(self, price_data): """训练增强版模型""" df = self.enhance_features(price_data) # 创建更丰富的滞后特征 for lag in [1, 2, 3, 7, 14, 30]: df[f'price_lag_{lag}'] = df['price'].shift(lag) df = df.dropna() feature_columns = (self.features + self.additional_features + [f'price_lag_{lag}' for lag in [1, 2, 3, 7, 14, 30]]) X = df[feature_columns] y = df['price'] # 特征标准化 X_scaled = self.scaler.fit_transform(X) X_train, X_test, y_train, y_test = train_test_split( X_scaled, y, test_size=0.2, random_state=42 ) self.model = GradientBoostingRegressor(n_estimators=200, random_state=42) self.model.fit(X_train, y_train) # 模型评估 y_pred = self.model.predict(X_test) mae = mean_absolute_error(y_test, y_pred) print(f"增强模型MAE: {mae:.2f}")

7. 项目扩展与优化方向

7.1 功能扩展建议

1. 多用户支持：实现用户注册登录，个性化监控列表
2. 移动端应用：开发React Native或Flutter移动应用
3. 浏览器插件：开发Chrome插件，实时显示价格信息
4. 社交媒体集成：对接微信、钉钉等消息通知
5. 更多电商平台：扩展拼多多、抖音电商等平台支持

7.2 技术优化方向

1. 分布式爬虫：使用Scrapy-Redis实现分布式数据采集
2. 实时数据处理：引入Kafka或RabbitMQ处理实时价格数据
3. 机器学习平台：集成MLflow进行模型管理和实验跟踪
4. 微服务架构：将系统拆分为多个微服务，提高可维护性

7.3 商业化考虑

1. 数据API服务：为其他开发者提供价格数据API
2. 白标解决方案：为企业客户提供定制化比价解决方案
3. 广告与推广：在合适位置展示相关商品推荐
4. 高级功能订阅：提供更高级的分析功能作为付费服务

这个智能购物助手项目展示了如何将AI技术应用于解决实际生活中的价格歧视问题。通过完整的代码实现和系统设计，读者可以在此基础上继续扩展功能，打造更强大的购物辅助工具。