import gradio as gr import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler from sklearn.tree import DecisionTreeClassifier from sklearn.svm import SVC import xgboost as xgb import statsmodels.api as sm from sklearn.metrics import confusion_matrix, classification_report, roc_auc_score, accuracy_score, precision_score, recall_score, f1_score import warnings import json # --- 初始設定與資料載入 --- warnings.filterwarnings("ignore", category=UserWarning) warnings.filterwarnings("ignore", category=FutureWarning) # plt.rcParams['font.family'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False def processDisplayDataframe(df): # 假設 df 已經存在 num_cols = df.select_dtypes(include=np.number).columns # 先抓出數值欄位名稱 df[num_cols] = df[num_cols].map(lambda x: f"{x:.4f}") return df def load_data(): """ 載入並對資料進行固定的預處理。 此函式只在應用程式啟動時執行一次。 """ df = pd.read_csv('sampled_data.csv') df_processed = df.copy() df_processed = df_processed.drop('id', axis=1) df_processed['Gender'] = df_processed['Gender'].apply(lambda x: 1 if x == 'Male' else 0) age_mapping = {'< 1 Year': 0, '1-2 Year': 1, '> 2 Years': 2} df_processed['Vehicle_Age'] = df_processed['Vehicle_Age'].map(age_mapping) df_processed['Vehicle_Damage'] = df_processed['Vehicle_Damage'].apply(lambda x: 1 if x == 'Yes' else 0) return df, df_processed df_original, df_processed = load_data() ALL_FEATURES = [col for col in df_processed.columns if col != 'Response'] NUMERICAL_FEATURES = [f for f in df_original.select_dtypes(include=np.number).columns.tolist() if f in ALL_FEATURES] # --- EDA 相關函式 --- def update_eda_section(selected_features): if not selected_features: return pd.DataFrame(), pd.DataFrame(), gr.update(choices=[], value=None), None stats = df_processed[selected_features].describe().T.reset_index().rename(columns={'index': 'Feature'}) corrs = df_processed[selected_features + ['Response']].corr(numeric_only=True)['Response'].drop('Response').to_frame().reset_index() corrs.columns = ['Feature', 'Correlation with Response'] first_feature_plot = generate_feature_plot(selected_features[0]) plot_selector_update = gr.update(choices=selected_features, value=selected_features[0]) stats = processDisplayDataframe(stats) corrs = processDisplayDataframe(corrs) return stats, corrs, plot_selector_update, first_feature_plot def generate_feature_plot(feature): if not feature: return None fig, ax = plt.subplots() if feature in NUMERICAL_FEATURES: sns.histplot(data=df_processed, x=feature, hue='Response', kde=True, ax=ax, palette='viridis', multiple="stack") ax.set_title(f'Histogram of "{feature}" (colored by Response)') else: sns.countplot(data=df_processed, x=feature, hue='Response', ax=ax, palette='viridis') ax.set_title(f'Count Plot of "{feature}" (colored by Response)') plt.tight_layout() return fig # --- 核心訓練與評估函式 --- def train_and_evaluate(history_log, model_name, features, dt_criterion, dt_max_depth, xgb_n_estimators, xgb_max_depth, xgb_learning_rate, svm_c, svm_kernel): """ 當使用者點擊 "執行模型訓練" 按鈕時觸發。 整合了資料準備、模型訓練、評估、結果視覺化以及紀錄日誌的完整流程。 """ if not features: # 如果沒有選擇特徵，只回傳錯誤訊息和空的日誌 return "錯誤：請至少選擇一個特徵！", None, None, None, pd.DataFrame(history_log, columns=LOG_COLUMNS), history_log # --- 1. 資料準備 --- X = df_processed[features] y = df_processed['Response'] # 2. 先切分資料，再進行標準化，避免資料外洩 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42, stratify=y) # 複製 X_train 和 X_test 以避免 SettingWithCopyWarning X_train_scaled = X_train.copy() X_test_scaled = X_test.copy() # 3. 準備 Scaler numerical_cols_in_x = [f for f in NUMERICAL_FEATURES if f in X_train.columns] if numerical_cols_in_x: scaler = StandardScaler() # 4. 只在 X_train 上 fit_transform X_train_scaled[numerical_cols_in_x] = scaler.fit_transform(X_train[numerical_cols_in_x]) # 5. 在 X_test 上 "只" transform X_test_scaled[numerical_cols_in_x] = scaler.transform(X_test[numerical_cols_in_x]) # --- 2. 模型選擇與訓練 --- params = {} if model_name == '羅吉斯回歸': # params = {'C': lr_c, 'solver': lr_solver} params = {} # statsmodels 不使用這些參數 X_train_sm = sm.add_constant(X_train); X_test_sm = sm.add_constant(X_test) logit_model = sm.Logit(y_train, X_train_sm) result = logit_model.fit(disp=0) y_pred_proba = result.predict(X_test_sm); y_pred = (y_pred_proba > 0.5).astype(int) importances, title = result.tvalues.drop('const', errors='ignore'), 'Feature t-values' else: if model_name == '決策樹': params = {'criterion': dt_criterion, 'max_depth': dt_max_depth} model = DecisionTreeClassifier(**params, random_state=42, class_weight='balanced') elif model_name == 'XGBoost': params = {'n_estimators': int(xgb_n_estimators), 'max_depth': int(xgb_max_depth), 'learning_rate': xgb_learning_rate} scale_pos_weight = y_train.value_counts()[0] / y_train.value_counts()[1] model = xgb.XGBClassifier(**params, scale_pos_weight=scale_pos_weight, use_label_encoder=False, eval_metric='logloss', random_state=42) elif model_name == 'SVM': params = {'C': svm_c, 'kernel': svm_kernel} model = SVC(**params, probability=True, random_state=42, class_weight='balanced') model.fit(X_train, y_train) y_pred = model.predict(X_test); y_pred_proba = model.predict_proba(X_test)[:, 1] if model_name == 'SVM' and svm_kernel == 'linear': importances, title = model.coef_[0], 'Feature Coefficients' elif model_name in ['決策樹', 'XGBoost']: importances, title = model.feature_importances_, 'Feature Importance' else: importances, title = None, 'Feature Importance' # --- 3. 評估與繪圖 --- accuracy_value = accuracy_score(y_test, y_pred) precision_value = precision_score(y_test, y_pred) recall_value = recall_score(y_test, y_pred) f1_score_value = f1_score(y_test, y_pred) roc_auc_value = roc_auc_score(y_test, y_pred_proba) accuracy_text = f"準確率 分數: {accuracy_value:.4f}" precision_text = f"精確率 分數: {precision_value:.4f}" recall_text = f"召回率 分數: {recall_value:.4f}" f1_score_text = f"F1 分數: {f1_score_value:.4f}" roc_auc_text = f"ROC-AUC 分數: {roc_auc_value:.4f}" report_dict = classification_report(y_test, y_pred, target_names=['not purchase insurance (0)', 'purchase insurance (1)'], output_dict=True) classfy_report = pd.DataFrame({ 'not purchase insurance (0)':report_dict['not purchase insurance (0)'], 'purchase insurance (1)':report_dict['purchase insurance (1)'], }, columns=[ 'not purchase insurance (0)', 'purchase insurance (1)']).T classfy_report.insert(0, "index", classfy_report.index) classfy_report = processDisplayDataframe(classfy_report) avg_report = pd.DataFrame([ report_dict["macro avg"], report_dict["weighted avg"], ], index=["macro avg", "weighted avg"]) avg_report.insert(0, "index", avg_report.index) avg_report = processDisplayDataframe(avg_report) # 2. 轉成 DataFrame（每個類別一列） # df_report = pd.DataFrame(report_dict).T # T = transpose，讓 index 變成類別名稱 # df_report.insert(0, "index", df_report.index) # # print(df_report) # df_report = processDisplayDataframe(df_report) cm = confusion_matrix(y_test, y_pred) fig_cm, ax_cm = plt.subplots(); sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', ax=ax_cm, xticklabels=['Predicted 0', 'Predicted 1'], yticklabels=['Actual 0', 'Actual 1']); ax_cm.set_title('Confusion Matrix'); ax_cm.set_xlabel('Predicted Label'); ax_cm.set_ylabel('Actual Label'); plt.tight_layout() fig_imp, ax_imp = plt.subplots() if importances is not None: feature_imp = pd.Series(importances, index=features).sort_values(ascending=False) sns.barplot(x=feature_imp, y=feature_imp.index, ax=ax_imp); ax_imp.set_title(title) else: ax_imp.text(0.5, 0.5, 'This model/kernel cannot directly display feature importance', ha='center', va='center'); ax_imp.set_title(title) plt.tight_layout() # --- 4. 紀錄日誌 --- new_log_entry = [ pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S'), model_name, ', '.join(features), json.dumps(params), f"{accuracy_value:.4f}", f"{precision_value:.4f}", f"{recall_value:.4f}", f"{f1_score_value:.4f}", f"{roc_auc_value:.4f}", ] # 將新紀錄加到歷史紀錄的開頭 updated_log = [new_log_entry] + history_log log_df = pd.DataFrame(updated_log, columns=LOG_COLUMNS) return classfy_report, avg_report, accuracy_text, precision_text, recall_text, f1_score_text, roc_auc_text, fig_cm, fig_imp, log_df, updated_log # --- Gradio 介面設計 --- LOG_COLUMNS = ["時間", "模型", "特徵", "參數", "準確率", "精確率", "召回率", "F1 分數", "ROC-AUC 分數"] with gr.Blocks(theme=gr.themes.Soft()) as demo: # 用於儲存日誌的隱藏狀態元件 log_state = gr.State([]) gr.Markdown("# 投保預測模型建置專案") gr.Markdown("在左側選擇特徵並點擊按鈕進行探索，或調整參數後點擊按鈕以訓練模型。") with gr.Row(): with gr.Column(scale=1): gr.Markdown("## 1. 特徵選擇與探索") feature_selector = gr.CheckboxGroup(ALL_FEATURES, label="選擇特徵", value=['Previously_Insured', 'Vehicle_Damage', 'Policy_Sales_Channel', 'Vehicle_Age', 'Age']) with gr.Row(): select_all_btn = gr.Button("全部選取"); deselect_all_btn = gr.Button("全部清除") with gr.Accordion("特徵探索 (EDA)", open=True): eda_run_btn = gr.Button("執行資料探索", variant="secondary") eda_stats = gr.DataFrame(label="敘述性統計") eda_corr = gr.DataFrame(label="與目標 'Response' 的相關係數") eda_plot_selector = gr.Dropdown(label="選擇要視覺化的特徵") eda_plot = gr.Plot(label="視覺化") gr.Markdown("## 2. 模型選擇與超參數調整") model_selector = gr.Dropdown(['羅吉斯回歸', '決策樹', 'XGBoost', 'SVM'], label="選擇模型", value='決策樹') with gr.Group(visible=False) as lr_box: gr.Markdown("#### 羅吉斯回歸") with gr.Group(visible=True) as dt_box: gr.Markdown("#### 決策樹"); dt_criterion = gr.Radio(['gini', 'entropy'], value='gini', label="評估標準"); dt_max_depth = gr.Slider(3, 30, value=8, step=1, label="最大深度") with gr.Group(visible=False) as xgb_box: gr.Markdown("#### XGBoost"); xgb_n_estimators = gr.Slider(50, 500, value=100, step=10, label="樹的數量"); xgb_max_depth = gr.Slider(3, 15, value=5, step=1, label="最大深度"); xgb_learning_rate = gr.Slider(0.01, 0.3, value=0.1, step=0.01, label="學習率") with gr.Group(visible=False) as svm_box: gr.Markdown("#### SVM"); svm_c = gr.Slider(0.01, 10.0, value=1.0, step=0.01, label="C (懲罰參數)"); svm_kernel = gr.Radio(['linear', 'rbf', 'poly'], value='linear', label="核心") run_btn = gr.Button("🚀 執行模型訓練", variant="primary") with gr.Column(scale=2): gr.Markdown("## 3. 模型評估結果") model_output_accuracy = gr.Textbox(label="準確率 分數") model_output_precision = gr.Textbox(label="精確率 分數") model_output_recall = gr.Textbox(label="召回率 分數") model_output_f1_score = gr.Textbox(label="F1 分數") model_output_auc = gr.Textbox(label="AUC 分數") model_output_report = gr.DataFrame(label="分類報告") model_output_report_avg = gr.DataFrame(label="平均報告") model_plot_cm = gr.Plot(label="混淆矩陣") model_plot_importance = gr.Plot(label="特徵重要性/係數") with gr.Accordion("操作紀錄 (History Log)", open=False): log_df_display = gr.DataFrame(headers=LOG_COLUMNS, datatype=["str", "str", "str", "str", "str"]) # --- 事件處理 --- eda_run_btn.click(update_eda_section, inputs=feature_selector, outputs=[eda_stats, eda_corr, eda_plot_selector, eda_plot]) eda_plot_selector.change(generate_feature_plot, inputs=eda_plot_selector, outputs=eda_plot) def show_hyperparameters(model_name): return {lr_box: gr.update(visible=model_name == '羅吉斯回歸'), dt_box: gr.update(visible=model_name == '決策樹'), xgb_box: gr.update(visible=model_name == 'XGBoost'), svm_box: gr.update(visible=model_name == 'SVM')} model_selector.change(show_hyperparameters, inputs=model_selector, outputs=[lr_box, dt_box, xgb_box, svm_box]) def select_all_features(): return gr.update(value=ALL_FEATURES) def deselect_all_features(): return gr.update(value=[]) select_all_btn.click(select_all_features, None, feature_selector) deselect_all_btn.click(deselect_all_features, None, feature_selector) run_btn.click( train_and_evaluate, inputs=[log_state, model_selector, feature_selector, dt_criterion, dt_max_depth, xgb_n_estimators, xgb_max_depth, xgb_learning_rate, svm_c, svm_kernel], outputs=[model_output_report, model_output_report_avg, model_output_accuracy, model_output_precision, model_output_recall, model_output_f1_score, model_output_auc, model_plot_cm, model_plot_importance, log_df_display, log_state] ) if __name__ == "__main__": demo.launch()