项目准备（flask+pyhon+MachineLearning）-3

目录

1.商品信息

2. 商品销售预测

2.1 机器学习

2.2 预测功能

3. 模型评估

---

1.商品信息 @app.route('/products') def products(): """商品分析页面""" data = load_data() # 计算当前期间和上期间 current_period = data[data['成交时间'] >= data['成交时间'].max() - timedelta(days=30)] previous_period = data[(data['成交时间'] < data['成交时间'].max() - timedelta(days=30)) & (data['成交时间'] >= data['成交时间'].max() - timedelta(days=60))] # 计算商品指标 current_sales = current_period.groupby('商品ID').apply( lambda x: (x['销量'] * x['单价']).sum() ) previous_sales = previous_period.groupby('商品ID').apply( lambda x: (x['销量'] * x['单价']).sum() ) product_metrics = pd.DataFrame({ 'current_sales': current_sales, 'previous_sales': previous_sales }).fillna(0) product_metrics['growth_rate'] = ( (product_metrics['current_sales'] - product_metrics['previous_sales']) / product_metrics['previous_sales'] ).fillna(0) max_competitor_sales = product_metrics['current_sales'].max() product_metrics['market_share'] = ( product_metrics['current_sales'] / max_competitor_sales ) # BCG矩阵分类 growth_rate_threshold = product_metrics['growth_rate'].median() market_share_threshold = product_metrics['market_share'].median() def classify_product(row): if row['growth_rate'] >= growth_rate_threshold: return '明星商品' if row['market_share'] >= market_share_threshold else '问题商品' else: return '现金牛' if row['market_share'] >= market_share_threshold else '瘦狗' product_metrics['category'] = product_metrics.apply(classify_product, axis=1) # 统计分类结果 category_stats = product_metrics.groupby('category').agg({ 'current_sales': ['count', 'sum'] }) category_stats.columns = ['product_count', 'sales_amount'] category_stats['sales_percentage'] = ( category_stats['sales_amount'] / category_stats['sales_amount'].sum() ) return render_template('products.html', category_statistics=category_stats.to_dict('index'), growth_rate_threshold=float(growth_rate_threshold), market_share_threshold=float(market_share_threshold)) 2. 商品销售预测 2.1 机器学习 def prepare_features(data): """准备特征数据""" # 删除包含NaN的行 data = data.dropna(subset=['销量', '单价', '类别ID', '门店编号']) # 时间特征 data['weekday'] = data['成交时间'].dt.weekday data['month'] = data['成交时间'].dt.month data['hour'] = data['成交时间'].dt.hour # 类别编码 le_category = LabelEncoder() le_store = LabelEncoder() # 拟合编码器 le_category.fit(data['类别ID'].astype(str)) # 将类别ID转换为字符串 le_store.fit(data['门店编号'].astype(str)) # 转换数据 data['类别编码'] = le_category.transform(data['类别ID'].astype(str)) data['门店编码'] = le_store.transform(data['门店编号'].astype(str)) # 特征选择 features = ['类别编码', '门店编码', '单价', 'weekday', 'month', 'hour'] target = '销量' # 确保所有特征都是数值类型 X = data[features].astype(float) y = data[target].astype(float) return X, y, le_category, le_store # 创建全局变量来存储模型和编码器 model = None scaler = None label_encoder_category = None label_encoder_store = None def initialize_model(): """初始化模型和编码器""" global model, scaler, label_encoder_category, label_encoder_store try: data = load_data() X, y, le_category, le_store = prepare_features(data) # 训练模型 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) # 标准化特征 scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) # 训练决策树模型 model = DecisionTreeRegressor(random_state=42, max_depth=10) model.fit(X_train_scaled, y_train) # 保存编码器 label_encoder_category = le_category label_encoder_store = le_store return True except Exception as e: print(f"模型初始化错误: {str(e)}") return False 2.2 预测功能 @app.route('/predict', methods=['POST']) def predict(): """处理预测请求""" global model, scaler, label_encoder_category, label_encoder_store try: # 如果模型未初始化，先初始化 if model is None or scaler is None: if not initialize_model(): return jsonify({'error': '模型初始化失败'}), 500 # 获取表单数据 category = request.form['category'] store = request.form['store'] price = float(request.form['price']) weekday = int(request.form['weekday']) month = int(request.form['month']) try: # 转换类别编码和门店编码 category_encoded = label_encoder_category.transform([str(category)])[0] store_encoded = label_encoder_store.transform([str(store)])[0] except ValueError as e: return jsonify({'error': f'无效的输入数据: {str(e)}'}), 400 # 准备预测数据 pred_data = pd.DataFrame([[ category_encoded, store_encoded, price, weekday, month, 12 # 使用默认时间 ]], columns=['类别编码', '门店编码', '单价', 'weekday', 'month', 'hour']) # 标准化预测数据 pred_data_scaled = scaler.transform(pred_data) # 预测 prediction = model.predict(pred_data_scaled)[0] # 确保预测结果为正整数 prediction = max(0, round(prediction)) # 获取门店信息 store_info = STORE_INFO.get(store, {}) store_name = store_info.get('name', f'门店{store}') # 加载历史数据进行分析 data = load_data() # 计算该类别的历史平均销量 category_avg = data[data['类别ID'].astype(str) == str(category)]['销量'].mean() # 计算该门店的历史平均销量 store_avg = data[data['门店编号'].astype(str) == str(store)]['销量'].mean() # 计算价格区间的平均销量 price_range = 0.1 # 价格范围±10% price_lower = price * (1 - price_range) price_upper = price * (1 + price_range) price_avg = data[(data['单价'] >= price_lower) & (data['单价'] <= price_upper)]['销量'].mean() # 计算同时段（星期几和月份）的历史平均销量 time_avg = data[(data['成交时间'].dt.weekday == weekday) & (data['成交时间'].dt.month == month)]['销量'].mean() # 生成分析结果 analysis = { 'category_comparison': round((prediction / category_avg * 100) if category_avg > 0 else 100), 'store_comparison': round((prediction / store_avg * 100) if store_avg > 0 else 100), 'price_comparison': round((prediction / price_avg * 100) if price_avg > 0 else 100), 'time_comparison': round((prediction / time_avg * 100) if time_avg > 0 else 100), 'category_avg': round(category_avg if not pd.isna(category_avg) else 0), 'store_avg': round(store_avg if not pd.isna(store_avg) else 0), 'price_avg': round(price_avg if not pd.isna(price_avg) else 0), 'time_avg': round(time_avg if not pd.isna(time_avg) else 0) } return jsonify({ 'prediction': int(prediction), 'category': category, 'category_name': CATEGORY_NAMES.get(category, f'类别{category}'), 'store': store, 'store_name': store_name, 'price': price, 'weekday': weekday, 'month': month, 'analysis': analysis }) except Exception as e: print(f"预测错误: {str(e)}") return jsonify({'error': str(e)}), 400 @app.route('/prediction') def prediction_page(): """销售预测页面""" data = load_data() categories = sorted(data['类别ID'].astype(str).unique().tolist()) stores = sorted(data['门店编号'].astype(str).unique().tolist()) # 创建类别选项列表，包含ID和名称 category_options = [ {'id': cat_id, 'name': CATEGORY_NAMES.get(cat_id, f'类别{cat_id}')} for cat_id in categories ] # 创建门店选项列表 store_options = [ {'id': store_id, 'name': STORE_INFO.get(store_id, {}).get('name', f'门店{store_id}')} for store_id in stores ] # 初始化模型（如果需要） global model if model is None: initialize_model() return render_template('prediction.html', categories=category_options, stores=store_options) 3. 模型评估 @app.route('/model_evaluation') def model_evaluation(): """模型评估页面""" data = load_data() # 准备特征 X, y, le_category, le_store = prepare_features(data) # 训练模型并获取评估结果 _, _, metrics, feature_importance, scatter_data, residual_data, feature_names, importance_scores = train_models(X, y) return render_template('model_evaluation.html', metrics=metrics, feature_importance=feature_importance, scatter_data=scatter_data, residual_data=residual_data, feature_names=feature_names, importance_scores=importance_scores)

4. 训练模型

def train_models(X, y): """训练模型""" # 数据分割 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) # 标准化特征 scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) X_test_scaled = scaler.transform(X_test) # 训练决策树模型 dt_model = DecisionTreeRegressor(random_state=42, max_depth=10) dt_model.fit(X_train_scaled, y_train) # 预测 y_pred = dt_model.predict(X_test_scaled) # 计算模型指标 metrics = { 'r2_score': r2_score(y_test, y_pred), 'mse': mean_squared_error(y_test, y_pred), 'mae': mean_absolute_error(y_test, y_pred), 'rmse': np.sqrt(mean_squared_error(y_test, y_pred)) } # 特征重要性 feature_importance = [] for name, importance in zip(X.columns, dt_model.feature_importances_): correlation = np.corrcoef(X[name], y)[0, 1] feature_importance.append({ 'name': name, 'importance': importance, 'correlation': correlation }) # 准备图表数据 scatter_data = [[float(actual), float(pred)] for actual, pred in zip(y_test, y_pred)] residuals = y_test - y_pred residual_data = [[float(pred), float(residual)] for pred, residual in zip(y_pred, residuals)] return dt_model, scaler, metrics, feature_importance, scatter_data, residual_data, X.columns.tolist(), dt_model.feature_importances_.tolist()