你有没有遇到过这样的情况同一件商品在不同平台、不同时间、甚至不同账号下显示的价格完全不同这就是典型的价格歧视现象。作为消费者我们往往在信息不对称中处于弱势地位而商家则通过复杂的算法动态调整价格最大化利润。在参加B站AI创造公开赛后我决定用AI技术来解决这个痛点。本文将分享如何从零开始构建一个能够打破价格歧视的智能购物助手。这个助手不仅能实时监控多个电商平台的价格变化还能分析历史价格趋势识别真正的优惠时机帮你避开商家的定价陷阱。与市面上大多数购物助手不同我们的方案重点不是简单的比价而是通过AI智能体技术理解商家的定价策略预测价格走势并在最佳时机发出购买建议。整个系统基于Python和主流AI框架构建代码完全开源你可以根据自己的需求进行定制。1. 价格歧视背后的技术逻辑与AI破局思路价格歧视并不是什么新鲜概念但在大数据和AI时代商家的定价策略变得更加精细和隐蔽。常见的价格歧视策略包括用户画像歧视根据你的浏览历史、购买能力、地理位置等信息差异化定价时间动态定价在需求高峰期提高价格在低谷期降低价格平台差异定价同一商品在不同电商平台设置不同价格新老用户歧视给新用户更多优惠老用户反而价格更高传统的手动比价方式效率低下而且很难捕捉到瞬时的价格变化。AI智能购物助手的核心价值在于自动化监控7×24小时不间断监控目标商品的价格变化智能分析利用机器学习算法识别价格模式和历史趋势策略预测基于市场供需、节假日等因素预测价格走势主动提醒在价格达到理想区间时及时通知用户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): 分析商品价格趋势 pass3. 环境准备与依赖配置在开始编码前需要准备以下开发环境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 tensorflow3.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_emailgmail.com password: your_app_password telegram: bot_token: your_bot_token chat_id: your_chat_id4. 数据采集模块实现数据采集是整个系统的基础需要处理各种反爬虫机制。4.1 基础爬虫类实现import requests from bs4 import BeautifulSoup import time import random from urllib.parse import urljoin import logging class BaseCrawler: def __init__(self, delay2, timeout30): 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/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, } self.session.headers.update(self.headers) def get_with_retry(self, url, max_retries3): 带重试机制的GET请求 for attempt in range(max_retries): try: response self.session.get(url, timeoutself.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 e4.2 电商平台特定爬虫实现以京东为例实现商品信息抓取class JDCrawler(BaseCrawler): def __init__(self): super().__init__() self.base_url https://search.jd.com/Search def search_products(self, keyword, page1): 搜索商品 params { keyword: keyword, page: page, s: (page - 1) * 30 1 } response self.get_with_retry(self.base_url, paramsparams) 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(stripTrue), 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(stripTrue) 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, days30): 获取商品价格历史需要调用价格接口 # 这里简化实现实际需要调用京东价格接口或自行存储历史数据 pass5. 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(window7).mean() df[price_std_7] df[price].rolling(window7).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_estimators100, random_state42) 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(contamination0.1, random_state42) 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_pathconfig.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, encodingutf-8) as f: self.config yaml.safe_load(f) def setup_database(self): 初始化数据库 self.conn sqlite3.connect(shopping_assistant.db, check_same_threadFalse) 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_priceNone): 监控单个商品 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, days30): 获取商品价格历史 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_exceptionsTrue) # 使用示例 async def main(): assistant SmartShoppingAssistant() # 添加监控商品 jd_url https://item.jd.com/100000000001.html # 示例商品 await assistant.monitor_product(jd_url, target_price1999) # 启动持续监控 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(targetstart_assistant, daemonTrue) monitor_thread.start() app.run(debugTrue, port5000)7.2 前端界面HTML!DOCTYPE html html head title智能购物助手/title script srchttps://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 classcontainer h1智能购物助手 - 打破价格歧视/h1 div classproduct-form h3添加监控商品/h3 form idaddProductForm input typeurl idproductUrl placeholder输入商品链接 required stylewidth: 300px; padding: 8px; input typenumber idtargetPrice placeholder目标价格可选 step0.01 stylewidth: 150px; padding: 8px; button typesubmit开始监控/button /form /div div idalertsContainer/div div classchart-container canvas idpriceChart/canvas /div div idproductsList h3监控中的商品/h3 div idproductsContainer/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 /html8. 部署与生产环境注意事项将系统部署到生产环境时需要考虑以下关键点8.1 服务器配置建议# docker-compose.yml 示例 version: 3.8 services: web: build: . ports: - 5000:5000 environment: - DATABASE_URLpostgresql://user:passdb:5432/shopping_assistant depends_on: - db - redis db: image: postgres:13 environment: - POSTGRES_DBshopping_assistant - POSTGRES_USERuser - POSTGRES_PASSWORDpass 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 反爬虫策略应对问题现象原因分析解决方案请求频繁被拦截IP被识别为爬虫使用代理IP池降低请求频率返回验证码页面触发反爬机制使用OCR识别验证码或人工打码数据加载通过JavaScript动态渲染内容使用Selenium或Playwright9.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): # 拼多多特定实现 pass10.2 高级AI功能情感分析分析商品评论情感倾向竞争分析监控竞品价格策略需求预测基于市场趋势预测商品需求个性化推荐基于用户偏好推荐商品10.3 移动端适配开发React Native或Flutter移动应用提供更便捷的使用体验。这个智能购物助手项目展示了如何将AI技术应用于解决实际生活问题。通过系统化的架构设计和模块化开发我们构建了一个可扩展、易维护的解决方案。最重要的是这个项目完全开源你可以基于自己的需求进行定制和扩展。在实际使用过程中建议先从监控少量重要商品开始逐步优化算法参数。同时要遵守各平台的使用条款合理控制请求频率避免对目标网站造成过大压力。