基于AI的智能购物助手开发：Python实现价格监控与预测

为了打破价格歧视我用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 requests2.31.0 selenium4.15.0 pandas2.1.3 numpy1.25.2 scikit-learn1.3.0 matplotlib3.7.2 flask2.3.3 beautifulsoup44.12.2 schedule1.2.02.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_pathdriver_path, optionsoptions) return driver2.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/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.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, paramsparams, headersself.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, paramsparams, headersself.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_pathdata/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, days30): 获取商品价格历史 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 df3.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[fprice_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_size0.2, random_state42 ) self.model RandomForestRegressor(n_estimators100, random_state42) 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_ahead7): 预测未来价格 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(daysi) for i in range(1, days_ahead1)] 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_priceNone): 添加监控商品 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, days30): 分析价格趋势 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_hours6): 启动定时监控 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(debugTrue, host0.0.0.0, port5000)4.3 前端界面示例!-- templates/index.html -- !DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title智能购物助手/title script srchttps://cdn.jsdelivr.net/npm/echarts5.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 classcontainer h1智能购物助手/h1 div classsearch-box input typetext idproductInput placeholder输入商品名称 button onclicksearchProduct()搜索比价/button button onclickanalyzePrice()分析趋势/button /div div idresults/div div idchart classchart-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 /html5. 部署与优化方案5.1 生产环境部署# docker-compose.yml version: 3.8 services: web: build: . ports: - 5000:5000 volumes: - ./data:/app/data environment: - FLASK_ENVproduction - DATABASE_URLsqlite:///data/prices.db restart: unless-stopped scheduler: build: . command: python scheduler.py volumes: - ./data:/app/data environment: - FLASK_ENVproduction 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(hostlocalhost, port6379, db0) 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_minutes30): 设置缓存数据 self.redis_client.setex( key, timedelta(minutesexpire_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 fsearch:{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 results6. 常见问题与解决方案6.1 反爬虫策略应对问题现象原因分析解决方案请求被拒绝IP被识别为爬虫使用代理IP轮换设置合理的请求间隔验证码拦截触发反爬机制集成验证码识别服务或手动处理数据加载异常动态页面渲染使用Selenium模拟真实浏览器行为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 anomalies6.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[fprice_lag_{lag}] df[price].shift(lag) df df.dropna() feature_columns (self.features self.additional_features [fprice_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_size0.2, random_state42 ) self.model GradientBoostingRegressor(n_estimators200, random_state42) 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 功能扩展建议多用户支持实现用户注册登录个性化监控列表移动端应用开发React Native或Flutter移动应用浏览器插件开发Chrome插件实时显示价格信息社交媒体集成对接微信、钉钉等消息通知更多电商平台扩展拼多多、抖音电商等平台支持7.2 技术优化方向分布式爬虫使用Scrapy-Redis实现分布式数据采集实时数据处理引入Kafka或RabbitMQ处理实时价格数据机器学习平台集成MLflow进行模型管理和实验跟踪微服务架构将系统拆分为多个微服务提高可维护性7.3 商业化考虑数据API服务为其他开发者提供价格数据API白标解决方案为企业客户提供定制化比价解决方案广告与推广在合适位置展示相关商品推荐高级功能订阅提供更高级的分析功能作为付费服务这个智能购物助手项目展示了如何将AI技术应用于解决实际生活中的价格歧视问题。通过完整的代码实现和系统设计读者可以在此基础上继续扩展功能打造更强大的购物辅助工具。