import numpy as np import pandas as pd import joblib import directory import os import scipy.stats as stats model_s1 = joblib.load(os.path.join(directory.ROOT_DIR,'model/stage1_model.pkl')) scaler_s1 = joblib.load(os.path.join(directory.ROOT_DIR,'model/stage1_scaler.pkl')) selector_s1 = joblib.load(os.path.join(directory.ROOT_DIR,'model/stage1_selector.pkl')) model_s2 = joblib.load(os.path.join(directory.ROOT_DIR,'model/stage2_model.pkl')) scaler_s2 = joblib.load(os.path.join(directory.ROOT_DIR,'model/stage2_scaler.pkl')) selector_s2 = joblib.load(os.path.join(directory.ROOT_DIR,'model/stage2_selector.pkl')) feat = joblib.load(os.path.join(directory.ROOT_DIR,'model/feature_list.pkl')) MIN_HC_PROB = 0.20 CI_PERCENTILE = 95 def rf_mean_ci(rf_model, X, ci_percentile=95): alpha = (100 - ci_percentile) / 100 z = stats.norm.ppf(1 - alpha / 2) # tree-level probabilities tree_proba = np.array([ tree.predict_proba(X) for tree in rf_model.estimators_ ]) # (n_trees, n_samples, n_classes) mean = tree_proba.mean(axis=0) std = tree_proba.std(axis=0, ddof=1) se = std / np.sqrt(tree_proba.shape[0]) lo = np.clip(mean - z * se, 0, 1) hi = np.clip(mean + z * se, 0, 1) return mean, lo, hi def predict(df_new): # ------------------------ # 전처리 (기존 유지) # ------------------------ X = df_new[feat].to_numpy(dtype=np.float32) # ======================== # Stage 1: HC vs Patient # ======================== X_s1 = scaler_s1.transform(selector_s1.transform(X)) mean_s1, lo_s1, hi_s1 = rf_mean_ci( model_s1, X_s1, CI_PERCENTILE ) hc_mean = mean_s1[0, 0] hc_lo = lo_s1[0, 0] hc_hi = hi_s1[0, 0] pat_mean = mean_s1[0, 1] pat_lo = lo_s1[0, 1] pat_hi = hi_s1[0, 1] # ======================== # Stage 2: subtype (mean only) # ======================== X_s2 = scaler_s2.transform(selector_s2.transform(X)) proba_s2 = model_s2.predict_proba(X_s2)[0] mdd_cond = proba_s2[0] anx_cond = proba_s2[1] psy_cond = proba_s2[2] # ======================== # 결합 확률 + CI 전파 # ======================== mdd_mean = pat_mean * mdd_cond mdd_lo = pat_lo * mdd_cond mdd_hi = pat_hi * mdd_cond anx_mean = pat_mean * anx_cond anx_lo = pat_lo * anx_cond anx_hi = pat_hi * anx_cond psy_mean = pat_mean * psy_cond psy_lo = pat_lo * psy_cond psy_hi = pat_hi * psy_cond # ======================== # MIN_HC_PROB 정책 보정 # ======================== if hc_mean < MIN_HC_PROB: remaining = 1.0 - MIN_HC_PROB patient_total = mdd_mean + anx_mean + psy_mean if patient_total > 0: scale = remaining / patient_total mdd_mean *= scale mdd_lo *= scale mdd_hi *= scale anx_mean *= scale anx_lo *= scale anx_hi *= scale psy_mean *= scale psy_lo *= scale psy_hi *= scale # HC는 mean 기준으로 고정 hc_scale = MIN_HC_PROB / hc_mean if hc_mean > 0 else 1.0 hc_mean = MIN_HC_PROB hc_lo = min(hc_lo * hc_scale, MIN_HC_PROB) hc_hi = min(hc_hi * hc_scale, 1.0) # ======================== # Return (요청한 구조) # ======================== return { 'HC': {'mean': hc_mean, 'lo': hc_lo, 'hi': hc_hi}, 'MDD': {'mean': mdd_mean, 'lo': mdd_lo, 'hi': mdd_hi}, 'ANX': {'mean': anx_mean, 'lo': anx_lo, 'hi': anx_hi}, 'PSY': {'mean': psy_mean, 'lo': psy_lo, 'hi': psy_hi}, } # df_new = pd.read_excel('new_data.xlsx') # probs = predict(df_new) # print(f"HC: {probs['HC']*100:.1f}%, MDD: {probs['MDD']*100:.1f}%, ANX: {probs['ANX']*100:.1f}%, PSY: {probs['PSY']*100:.1f}%") # 선민쌤이 건들고싶을면 건드는 곳. '''MIN_HC_PROB = 0.10 # 모델 최저점. hc를 몇을 최저로 둘거냐. CI_PERCENTILE = 95 # 얘도 마찬가지로 작아질수록 좁아짐, 높아질수록 넓어짐 CI_MULTIPLIER = 3.0 # 값의 폭을 조절하는건데, 작아질수록 좁아지고 높아질수록 넓어짐 def predict_with_tree_ci(rf_model, X, ci_percentile=95): alpha = (100 - ci_percentile) / 100 z = stats.norm.ppf(1 - alpha / 2) n_classes = len(rf_model.classes_) all_proba = [] for tree in rf_model.estimators_: proba = tree.predict_proba(X) if proba.shape[1] < n_classes: full_proba = np.zeros((X.shape[0], n_classes)) for i, c in enumerate(tree.classes_): class_idx = np.where(rf_model.classes_ == c)[0][0] full_proba[:, class_idx] = proba[:, i] proba = full_proba else: ordered_proba = np.zeros((X.shape[0], n_classes)) for i, c in enumerate(tree.classes_): class_idx = np.where(rf_model.classes_ == c)[0][0] ordered_proba[:, class_idx] = proba[:, i] proba = ordered_proba all_proba.append(proba) all_proba = np.array(all_proba) n_trees = all_proba.shape[0] mean = all_proba.mean(axis=0) std = all_proba.std(axis=0, ddof=1) se = std / np.sqrt(n_trees) se = se * CI_MULTIPLIER ci_lo = np.clip(mean - z * se, 0, 1) ci_hi = np.clip(mean + z * se, 0, 1) return mean, ci_lo, ci_hi def predict(df_new): if 'sex' in df_new.columns: df_new = df_new.copy() df_new['sex'] = df_new['sex'].astype(str).str.strip().str.upper() df_new['sex'] = df_new['sex'].replace({'M': 1, '남': 1, 'F': 0, '여': 0}) df_new['sex'] = pd.to_numeric(df_new['sex'], errors='coerce') X = df_new[feat].to_numpy(dtype=np.float32) # Stage 1 X_s1_scaled = scaler_s1.transform(X) X_s1_selected = selector_s1.transform(X_s1_scaled) mean_s1, ci_lo_s1, ci_hi_s1 = predict_with_tree_ci(model_s1, X_s1_selected, CI_PERCENTILE) hc_mean = mean_s1[0, 0] hc_lo = ci_lo_s1[0, 0] hc_hi = ci_hi_s1[0, 0] pat_mean = mean_s1[0, 1] pat_lo = ci_lo_s1[0, 1] pat_hi = ci_hi_s1[0, 1] # Stage 2 X_s2_selected = selector_s2.transform(X) X_s2_scaled = scaler_s2.transform(X_s2_selected) proba_s2 = model_s2.predict_proba(X_s2_scaled)[0] mdd_cond = proba_s2[0] anx_cond = proba_s2[1] psy_cond = proba_s2[2] mdd_mean = pat_mean * mdd_cond mdd_lo = pat_lo * mdd_cond mdd_hi = pat_hi * mdd_cond anx_mean = pat_mean * anx_cond anx_lo = pat_lo * anx_cond anx_hi = pat_hi * anx_cond psy_mean = pat_mean * psy_cond psy_lo = pat_lo * psy_cond psy_hi = pat_hi * psy_cond if hc_mean < MIN_HC_PROB: remaining = 1.0 - MIN_HC_PROB patient_total = mdd_mean + anx_mean + psy_mean if patient_total > 0: scale = remaining / patient_total mdd_mean *= scale mdd_lo *= scale mdd_hi *= scale anx_mean *= scale anx_lo *= scale anx_hi *= scale psy_mean *= scale psy_lo *= scale psy_hi *= scale hc_scale = MIN_HC_PROB / hc_mean if hc_mean > 0 else 1 hc_mean = MIN_HC_PROB hc_lo = min(hc_lo * hc_scale, MIN_HC_PROB) hc_hi = min(hc_hi * hc_scale, 1.0) return { 'HC': {'mean': hc_mean, 'lo': hc_lo, 'hi': hc_hi}, 'MDD': {'mean': mdd_mean, 'lo': mdd_lo, 'hi': mdd_hi}, 'ANX': {'mean': anx_mean, 'lo': anx_lo, 'hi': anx_hi}, 'PSY': {'mean': psy_mean, 'lo': psy_lo, 'hi': psy_hi}, } def format_result(probs): lines = [] for label in ['HC', 'MDD', 'ANX', 'PSY']: p = probs[label] pm = ((p['hi'] - p['lo']) / 2) * 100 lines.append( f"{label}: {p['mean'] * 100:5.1f}% ±{pm:4.1f}" )''' #return "\n".join(lines) #if __name__ == "__main__": # 엑셀파일 넣는곳 # TEST_PATH = r"C:\Users\User\Desktop\Goomin\Bwave\4class\stage1_not_used_extract\stage1_not_used_seed2026_all.xlsx" # df_test = pd.read_excel(TEST_PATH) # for i in range(len(df_test)): # df_row = df_test.iloc[[i]] # probs = predict(df_row) # print(f"\n[{i + 1}/{len(df_test)}]") # 여기는 이름 넣으시는 공간 # print(format_result(probs))