你有没有遇到过这样的情况：同一件商品，在不同平台、不同时间、甚至不同账号下显示的价格完全不同？这就是典型的"价格歧视"现象。作为消费者，我们往往在信息不对称中处于弱势地位，而商家则通过复杂的算法动态调整价格，最大化利润。

在参加B站AI创造公开赛后，我决定用AI技术来解决这个痛点。本文将分享如何从零开始构建一个能够打破价格歧视的智能购物助手。这个助手不仅能实时监控多个电商平台的价格变化，还能分析历史价格趋势，识别真正的优惠时机，帮你避开商家的定价陷阱。

与市面上大多数购物助手不同，我们的方案重点不是简单的比价，而是通过AI智能体技术理解商家的定价策略，预测价格走势，并在最佳时机发出购买建议。整个系统基于Python和主流AI框架构建，代码完全开源，你可以根据自己的需求进行定制。

1. 价格歧视背后的技术逻辑与AI破局思路

价格歧视并不是什么新鲜概念，但在大数据和AI时代，商家的定价策略变得更加精细和隐蔽。常见的价格歧视策略包括：

• 用户画像歧视：根据你的浏览历史、购买能力、地理位置等信息差异化定价
• 时间动态定价：在需求高峰期提高价格，在低谷期降低价格
• 平台差异定价：同一商品在不同电商平台设置不同价格
• 新老用户歧视：给新用户更多优惠，老用户反而价格更高

传统的手动比价方式效率低下，而且很难捕捉到瞬时的价格变化。AI智能购物助手的核心价值在于：

1. 自动化监控：7×24小时不间断监控目标商品的价格变化
2. 智能分析：利用机器学习算法识别价格模式和历史趋势
3. 策略预测：基于市场供需、节假日等因素预测价格走势
4. 主动提醒：在价格达到理想区间时及时通知用户

2. 技术架构与核心组件设计

我们的智能购物助手采用模块化设计，主要包括以下核心组件：

2.1 数据采集层

负责从各大电商平台抓取商品信息，需要考虑反爬虫策略和请求频率控制。

2.2 数据处理层

对采集的原始数据进行清洗、去重、标准化处理。

2.3 AI分析引擎

核心的智能分析模块，包括价格趋势预测、异常检测等算法。

2.4 用户交互层

提供Web界面和消息通知功能。

# 系统架构核心类图示意 class ShoppingAssistant: def __init__(self): self.data_collector = DataCollector() self.price_analyzer = PriceAnalyzer() self.notification_manager = NotificationManager() def monitor_product(self, product_url, target_price): """监控指定商品，达到目标价格时通知""" pass def analyze_trend(self, product_id): """分析商品价格趋势""" pass

3. 环境准备与依赖配置

在开始编码前，需要准备以下开发环境：

3.1 Python环境要求

# 创建虚拟环境 python -m venv shopping_assistant source shopping_assistant/bin/activate # Linux/Mac # shopping_assistant\Scripts\activate # Windows # 安装核心依赖 pip install requests beautifulsoup4 selenium pip install pandas numpy scikit-learn pip install matplotlib seaborn plotly pip install flask django celery pip install transformers torch tensorflow

3.2 配置文件设置

创建config.yaml文件管理应用配置：

# config.yaml database: host: localhost port: 5432 name: price_monitor user: your_username password: your_password api_keys: openai: your_openai_key telegram: your_telegram_bot_token crawler: request_delay: 2 # 请求间隔秒数 retry_times: 3 timeout: 30 notification: email: smtp_server: smtp.gmail.com port: 587 username: your_email@gmail.com password: your_app_password telegram: bot_token: your_bot_token chat_id: your_chat_id

4. 数据采集模块实现

数据采集是整个系统的基础，需要处理各种反爬虫机制。

4.1 基础爬虫类实现

import requests from bs4 import BeautifulSoup import time import random from urllib.parse import urljoin import logging class BaseCrawler: def __init__(self, delay=2, timeout=30): self.delay = delay self.timeout = timeout 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 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', } self.session.headers.update(self.headers) def get_with_retry(self, url, max_retries=3): """带重试机制的GET请求""" for attempt in range(max_retries): try: response = self.session.get(url, timeout=self.timeout) response.raise_for_status() time.sleep(self.delay + random.uniform(0, 1)) return response except requests.RequestException as e: logging.warning(f"请求失败 {url}, 尝试 {attempt + 1}/{max_retries}: {e}") if attempt < max_retries - 1: time.sleep(2 ** attempt) # 指数退避 else: raise e

4.2 电商平台特定爬虫实现

以京东为例，实现商品信息抓取：

class JDCrawler(BaseCrawler): def __init__(self): super().__init__() self.base_url = "https://search.jd.com/Search" def search_products(self, keyword, page=1): """搜索商品""" params = { 'keyword': keyword, 'page': page, 's': (page - 1) * 30 + 1 } response = self.get_with_retry(self.base_url, params=params) return self.parse_search_results(response.text) def parse_search_results(self, html): """解析搜索结果页面""" soup = BeautifulSoup(html, 'html.parser') products = [] items = soup.select('.gl-item') for item in items: try: product = { 'sku_id': item.get('data-sku'), 'title': item.select_one('.p-name em').get_text(strip=True), 'price': float(item.select_one('.p-price i').get_text()), 'url': urljoin('https://item.jd.com/', item.select_one('.p-img a')['href']), 'shop': item.select_one('.p-shop a').get_text(strip=True) if item.select_one('.p-shop a') else '自营', 'image': item.select_one('.p-img img')['src'] or item.select_one('.p-img img')['data-lazy-img'] } products.append(product) except Exception as e: logging.error(f"解析商品信息失败: {e}") continue return products def get_price_history(self, sku_id, days=30): """获取商品价格历史（需要调用价格接口）""" # 这里简化实现，实际需要调用京东价格接口或自行存储历史数据 pass

5. AI价格分析与预测引擎

这是系统的智能核心，使用机器学习算法分析价格模式。

5.1 价格趋势分析类

import pandas as pd import numpy as np from sklearn.ensemble import RandomForestRegressor from sklearn.preprocessing import StandardScaler import joblib from datetime import datetime, timedelta class PriceAnalyzer: def __init__(self): self.model = None self.scaler = StandardScaler() def prepare_features(self, price_data): """准备机器学习特征""" df = pd.DataFrame(price_data) df['date'] = pd.to_datetime(df['timestamp']) df = df.set_index('date') # 生成时间特征 df['day_of_week'] = df.index.dayofweek df['day_of_month'] = df.index.day df['month'] = df.index.month df['is_weekend'] = (df['day_of_week'] >= 5).astype(int) # 生成统计特征 df['price_ma_7'] = df['price'].rolling(window=7).mean() df['price_std_7'] = df['price'].rolling(window=7).std() df['price_change'] = df['price'].pct_change() # 节假日特征（简化实现） df['is_holiday'] = self._identify_holidays(df.index) return df.dropna() def _identify_holidays(self, dates): """识别节假日（简化版）""" # 实际应用中应该使用更完善的节假日库 holidays = [] for date in dates: # 简单的节假日判断逻辑 if date.month == 1 and date.day == 1: # 元旦 holidays.append(1) elif date.month == 10 and 1 <= date.day <= 7: # 国庆 holidays.append(1) else: holidays.append(0) return holidays def train_model(self, historical_data): """训练价格预测模型""" df = self.prepare_features(historical_data) # 准备训练数据 features = ['day_of_week', 'day_of_month', 'month', 'is_weekend', 'price_ma_7', 'price_std_7', 'is_holiday'] X = df[features] y = df['price'] # 标准化特征 X_scaled = self.scaler.fit_transform(X) # 训练模型 self.model = RandomForestRegressor(n_estimators=100, random_state=42) self.model.fit(X_scaled, y) return self.model def predict_price(self, future_dates): """预测未来价格""" if not self.model: raise ValueError("模型未训练，请先调用train_model方法") predictions = [] for date in future_dates: features = self._create_features_for_date(date) features_scaled = self.scaler.transform([features]) pred_price = self.model.predict(features_scaled)[0] predictions.append({'date': date, 'predicted_price': pred_price}) return predictions def _create_features_for_date(self, date): """为指定日期创建特征""" # 这里需要根据历史数据计算移动平均等特征 # 简化实现 return [ date.dayofweek, date.day, date.month, 1 if date.dayofweek >= 5 else 0, 0, 0, 0 # 简化处理，实际需要真实计算 ]

5.2 价格异常检测

from sklearn.ensemble import IsolationForest from sklearn.svm import OneClassSVM import numpy as np class PriceAnomalyDetector: def __init__(self): self.detector = IsolationForest(contamination=0.1, random_state=42) def detect_anomalies(self, price_series): """检测价格异常点""" prices = np.array(price_series).reshape(-1, 1) # 训练异常检测模型 anomalies = self.detector.fit_predict(prices) # 返回异常点索引 anomaly_indices = np.where(anomalies == -1)[0] return anomaly_indices def is_good_deal(self, current_price, price_history): """判断当前价格是否是好deal""" historical_prices = [p['price'] for p in price_history] if len(historical_prices) < 10: return False # 数据不足，无法判断 mean_price = np.mean(historical_prices) std_price = np.std(historical_prices) # 如果当前价格低于历史平均价的1个标准差，认为是好deal return current_price < (mean_price - std_price)

6. 完整系统集成与实战演示

现在我们将各个模块整合成一个完整的购物助手系统。

6.1 主程序实现

import asyncio import aioschedule as schedule from datetime import datetime import json import sqlite3 class SmartShoppingAssistant: def __init__(self, config_path='config.yaml'): self.load_config(config_path) self.setup_database() self.crawlers = { 'jd': JDCrawler(), # 可以添加其他平台的爬虫 } self.analyzer = PriceAnalyzer() self.anomaly_detector = PriceAnomalyDetector() def load_config(self, config_path): """加载配置文件""" import yaml with open(config_path, 'r', encoding='utf-8') as f: self.config = yaml.safe_load(f) def setup_database(self): """初始化数据库""" self.conn = sqlite3.connect('shopping_assistant.db', check_same_thread=False) self.create_tables() def create_tables(self): """创建数据表""" cursor = self.conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, sku_id TEXT NOT NULL, title TEXT NOT NULL, url TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(platform, sku_id) ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id INTEGER, price REAL NOT NULL, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (product_id) REFERENCES products (id) ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS price_alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id INTEGER, target_price REAL NOT NULL, is_active BOOLEAN DEFAULT TRUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (product_id) REFERENCES products (id) ) ''') self.conn.commit() async def monitor_product(self, product_url, target_price=None): """监控单个商品""" platform = self.identify_platform(product_url) crawler = self.crawlers.get(platform) if not crawler: raise ValueError(f"不支持的平台: {platform}") # 获取当前价格 current_price = await crawler.get_current_price(product_url) # 保存价格记录 product_id = self.save_product_info(platform, product_url) self.save_price_record(product_id, current_price) # 分析价格趋势 history = self.get_price_history(product_id) if len(history) > 10: # 有足够历史数据时进行分析 is_good_deal = self.anomaly_detector.is_good_deal(current_price, history) if is_good_deal or (target_price and current_price <= target_price): await self.send_alert(product_id, current_price, target_price) return current_price def identify_platform(self, url): """识别电商平台""" if 'jd.com' in url: return 'jd' elif 'taobao.com' in url or 'tmall.com' in url: return 'taobao' else: return 'unknown' def save_product_info(self, platform, url): """保存商品信息到数据库""" cursor = self.conn.cursor() # 这里简化实现，实际需要从URL解析商品信息 cursor.execute(''' INSERT OR IGNORE INTO products (platform, sku_id, title, url) VALUES (?, ?, ?, ?) ''', (platform, 'temp_sku', '临时商品', url)) self.conn.commit() return cursor.lastrowid def save_price_record(self, product_id, price): """保存价格记录""" cursor = self.conn.cursor() cursor.execute(''' INSERT INTO price_history (product_id, price) VALUES (?, ?) ''', (product_id, price)) self.conn.commit() def get_price_history(self, product_id, days=30): """获取商品价格历史""" cursor = self.conn.cursor() cursor.execute(''' SELECT price, timestamp FROM price_history WHERE product_id = ? AND timestamp >= datetime('now', ?) ORDER BY timestamp ''', (product_id, f'-{days} days')) return [{'price': row[0], 'timestamp': row[1]} for row in cursor.fetchall()] async def send_alert(self, product_id, current_price, target_price): """发送价格提醒""" message = f"🎉 价格提醒！商品当前价格: {current_price}元" if target_price: message += f"，低于目标价格: {target_price}元" # 这里可以实现邮件、Telegram等通知方式 print(f"发送提醒: {message}") # 示例：Telegram通知实现 # await self.send_telegram_message(message) async def start_monitoring(self): """启动监控服务""" print("智能购物助手开始运行...") # 添加定时任务 schedule.every(1).hours.do(self.run_monitoring_job) while True: await schedule.run_pending() await asyncio.sleep(1) async def run_monitoring_job(self): """执行监控任务""" print(f"{datetime.now()}: 执行价格监控...") # 获取所有活跃的监控任务 cursor = self.conn.cursor() cursor.execute(''' SELECT p.url, pa.target_price FROM price_alerts pa JOIN products p ON pa.product_id = p.id WHERE pa.is_active = TRUE ''') tasks = [] for row in cursor.fetchall(): task = self.monitor_product(row[0], row[1]) tasks.append(task) # 并发执行所有监控任务 await asyncio.gather(*tasks, return_exceptions=True) # 使用示例 async def main(): assistant = SmartShoppingAssistant() # 添加监控商品 jd_url = "https://item.jd.com/100000000001.html" # 示例商品 await assistant.monitor_product(jd_url, target_price=1999) # 启动持续监控 await assistant.start_monitoring() if __name__ == "__main__": asyncio.run(main())

7. 前端界面与用户交互

为了让普通用户也能使用，我们开发一个简单的Web界面。

7.1 Flask Web应用

from flask import Flask, render_template, request, jsonify import threading import asyncio app = Flask(__name__) assistant = None def run_async(coro): """在单独线程中运行异步函数""" loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete(coro) @app.route('/') def index(): return render_template('index.html') @app.route('/api/add_product', methods=['POST']) def add_product(): """添加商品监控API""" data = request.json product_url = data.get('url') target_price = data.get('target_price') try: # 在新线程中运行异步函数 result = run_async(assistant.monitor_product(product_url, target_price)) return jsonify({'success': True, 'current_price': result}) except Exception as e: return jsonify({'success': False, 'error': str(e)}) @app.route('/api/price_history/<product_id>') def get_price_history(product_id): """获取价格历史API""" history = assistant.get_price_history(int(product_id)) return jsonify(history) @app.route('/api/analysis/<product_id>') def get_analysis(product_id): """获取价格分析API""" history = assistant.get_price_history(int(product_id)) if len(history) > 10: analysis = assistant.analyzer.analyze_trend(history) return jsonify(analysis) else: return jsonify({'error': '数据不足'}) def start_assistant(): """启动购物助手后台服务""" global assistant assistant = SmartShoppingAssistant() run_async(assistant.start_monitoring()) if __name__ == '__main__': # 在后台线程中启动监控服务 monitor_thread = threading.Thread(target=start_assistant, daemon=True) monitor_thread.start() app.run(debug=True, port=5000)

7.2 前端界面HTML

<!DOCTYPE html> <html> <head> <title>智能购物助手</title> <script src="https://cdn.jsdelivr.net/npm/chart.js"></script> <style> .container { max-width: 1200px; margin: 0 auto; padding: 20px; } .product-form { background: #f5f5f5; padding: 20px; border-radius: 8px; margin-bottom: 20px; } .chart-container { height: 400px; margin: 20px 0; } .alert { padding: 10px; margin: 10px 0; border-radius: 4px; } .alert-success { background: #d4edda; color: #155724; } </style> </head> <body> <div class="container"> <h1>智能购物助手 - 打破价格歧视</h1> <div class="product-form"> <h3>添加监控商品</h3> <form id="addProductForm"> <input type="url" id="productUrl" placeholder="输入商品链接" required style="width: 300px; padding: 8px;"> <input type="number" id="targetPrice" placeholder="目标价格（可选）" step="0.01" style="width: 150px; padding: 8px;"> <button type="submit">开始监控</button> </form> </div> <div id="alertsContainer"></div> <div class="chart-container"> <canvas id="priceChart"></canvas> </div> <div id="productsList"> <h3>监控中的商品</h3> <div id="productsContainer"></div> </div> </div> <script> // 前端JavaScript代码实现交互功能 document.getElementById('addProductForm').addEventListener('submit', async function(e) { e.preventDefault(); const url = document.getElementById('productUrl').value; const targetPrice = document.getElementById('targetPrice').value; try { const response = await fetch('/api/add_product', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ url, target_price: targetPrice }) }); const result = await response.json(); if (result.success) { showAlert('商品添加成功！当前价格：' + result.current_price, 'success'); loadProducts(); } else { showAlert('添加失败：' + result.error, 'error'); } } catch (error) { showAlert('网络错误：' + error.message, 'error'); } }); function showAlert(message, type) { const alertDiv = document.createElement('div'); alertDiv.className = `alert alert-${type}`; alertDiv.textContent = message; document.getElementById('alertsContainer').appendChild(alertDiv); setTimeout(() => alertDiv.remove(), 5000); } async function loadProducts() { // 加载商品列表和价格图表 // 实现省略... } </script> </body> </html>

8. 部署与生产环境注意事项

将系统部署到生产环境时需要考虑以下关键点：

8.1 服务器配置建议

# docker-compose.yml 示例 version: '3.8' services: web: build: . ports: - "5000:5000" environment: - DATABASE_URL=postgresql://user:pass@db:5432/shopping_assistant depends_on: - db - redis db: image: postgres:13 environment: - POSTGRES_DB=shopping_assistant - POSTGRES_USER=user - POSTGRES_PASSWORD=pass volumes: - db_data:/var/lib/postgresql/data redis: image: redis:6-alpine volumes: - redis_data:/data volumes: db_data: redis_data:

8.2 性能优化配置

# 性能优化建议 OPTIMIZATION_CONFIG = { 'crawler': { 'concurrent_requests': 10, # 并发请求数 'delay_between_requests': 1.0, # 请求间隔 'timeout': 30, 'retry_times': 3 }, 'database': { 'pool_size': 20, 'max_overflow': 30, 'pool_pre_ping': True }, 'cache': { 'redis_ttl': 3600, # 缓存1小时 'price_history_ttl': 86400 # 价格历史缓存24小时 } }

9. 常见问题与解决方案

在实际使用过程中可能会遇到以下问题：

9.1 反爬虫策略应对

9.2 数据准确性保障

class DataValidator: """数据验证器，确保采集数据的准确性""" def validate_price(self, price): """验证价格数据合理性""" if not isinstance(price, (int, float)): return False if price <= 0 or price > 1000000: # 假设商品价格范围 return False return True def detect_price_outliers(self, price_series): """检测价格异常值""" prices = np.array(price_series) Q1 = np.percentile(prices, 25) Q3 = np.percentile(prices, 75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR return (prices < lower_bound) | (prices > upper_bound)

9.3 系统稳定性维护

• 监控告警：设置系统健康检查，异常时自动告警
• 数据备份：定期备份价格历史和用户配置
• 错误恢复：实现断点续爬和任务重试机制
• 日志记录：详细的运行日志便于问题排查

10. 扩展功能与未来展望

基础版本完成后，可以考虑以下扩展功能：

10.1 多平台支持扩展

# 扩展更多电商平台爬虫 class TaoBaoCrawler(BaseCrawler): def get_product_info(self, url): # 淘宝特定实现 pass class P DDCrawler(BaseCrawler): def get_product_info(self, url): # 拼多多特定实现 pass

10.2 高级AI功能

• 情感分析：分析商品评论情感倾向
• 竞争分析：监控竞品价格策略
• 需求预测：基于市场趋势预测商品需求
• 个性化推荐：基于用户偏好推荐商品

10.3 移动端适配

开发React Native或Flutter移动应用，提供更便捷的使用体验。

这个智能购物助手项目展示了如何将AI技术应用于解决实际生活问题。通过系统化的架构设计和模块化开发，我们构建了一个可扩展、易维护的解决方案。最重要的是，这个项目完全开源，你可以基于自己的需求进行定制和扩展。

在实际使用过程中，建议先从监控少量重要商品开始，逐步优化算法参数。同时要遵守各平台的使用条款，合理控制请求频率，避免对目标网站造成过大压力。