目錄
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.weekdaydata['month'] = data['成交時間'].dt.monthdata['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 = Nonedef initialize_model():"""初始化模型和編碼器"""global model, scaler, label_encoder_category, label_encoder_storetry: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_categorylabel_encoder_store = le_storereturn Trueexcept 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_storetry:# 如果模型未初始化，先初始化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 modelif 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_predresidual_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()
