Delete Analysis_code/7.ensemble

Analysis_code/7.ensemble/analysis_of_shap.ipynb

@@ -1,3 +0,0 @@
-version https://git-lfs.github.com/spec/v1
-oid sha256:d4427789841c51710ee09faff31ec283a2cc099a31ed44010a65cfba7f578a98
-size 2440086

Analysis_code/7.ensemble/ensemble.ipynb

@@ -1,3 +0,0 @@
-version https://git-lfs.github.com/spec/v1
-oid sha256:b948db39e385865eccedfded6410f2d4719ad87d5a95d857b62ba34f1995065b
-size 20952

Analysis_code/7.ensemble/shap_utils.py

@@ -1,1260 +0,0 @@
-from joblib import load
-import pandas as pd
-import numpy as np
-import matplotlib.pyplot as plt
-import os
-import sys
-from contextlib import contextmanager
-from io import StringIO
-from typing import Optional, Tuple, List, Dict, Any
-from dataclasses import dataclass
-from abc import ABC, abstractmethod
-from sklearn.metrics import confusion_matrix
-from sklearn.preprocessing import LabelEncoder
-from scipy.stats import wasserstein_distance
-import shap
-import torch
-import warnings
-
-# 모든 경고 무시
-warnings.filterwarnings('ignore')
-warnings.filterwarnings('ignore', category=UserWarning, message='.*unrecognized nn.Module.*')
-warnings.filterwarnings('ignore', category=UserWarning, message='.*LayerNorm.*')
-warnings.filterwarnings('ignore', category=UserWarning, message='.*unrecognized.*')
-warnings.filterwarnings('ignore', module='shap.*')
-
-# 모델 클래스 import를 위한 경로 설정
-try:
-    current_file_dir = os.path.dirname(os.path.abspath(__file__))
-    models_path = os.path.abspath(os.path.join(current_file_dir, '../models'))
-except NameError:
-    cwd = os.getcwd()
-    if 'Analysis_code' in cwd:
-        analysis_code_dir = cwd[:cwd.index('Analysis_code') + len('Analysis_code')]
-        models_path = os.path.join(analysis_code_dir, 'models')
-    else:
-        models_path = '/workspace/visibility_prediction/Analysis_code/models'
-
-if models_path not in sys.path:
-    sys.path.insert(0, models_path)
-
-from ft_transformer import FTTransformer
-from resnet_like import ResNetLike
-from deepgbm import DeepGBM
-
-
-# ==========================================
-# 1. Config 클래스
-# ==========================================
-
-class SHAPConfig:
-    """SHAP 분석 설정을 관리하는 클래스."""
-    
-    # 상수 정의
-    FOLD_YEARS = [[2018, 2019], [2018, 2019], [2019, 2020]]
-    TEST_YEAR = 2021
-    DROP_COLUMNS = ['multi_class', 'year']
-    FEATURE_COLUMNS = [
-        'temp_C', 'precip_mm', 'wind_speed', 'wind_dir', 'hm',
-        'vap_pressure', 'dewpoint_C', 'loc_pressure', 'sea_pressure',
-        'solarRad', 'snow_cm', 'cloudcover', 'lm_cloudcover', 'low_cloudbase',
-        'groundtemp', 'O3', 'NO2', 'PM10', 'PM25', 'year',
-        'month', 'hour', 'ground_temp - temp_C', 'hour_sin', 'hour_cos',
-        'month_sin', 'month_cos', 'multi_class'
-    ]
-    
-    # 기본 설정값
-    DEFAULT_N_FOLDS = 3
-    DEFAULT_N_BACKGROUND_SAMPLES = 200
-    DEFAULT_N_TEST_SAMPLES = 500
-    DEFAULT_FIGSIZE = (20, 8)
-    DEFAULT_TOP_N = 3
-    DEFAULT_TARGET_CLASS = 0
-    DEFAULT_DEVICE = 'cpu'
-    DEFAULT_PLOT_MARGIN = 0.05
-    
-    def __init__(
-        self,
-        n_folds: int = DEFAULT_N_FOLDS,
-        n_background_samples: int = DEFAULT_N_BACKGROUND_SAMPLES,
-        n_test_samples: int = DEFAULT_N_TEST_SAMPLES,
-        figsize: Tuple[int, int] = DEFAULT_FIGSIZE,
-        top_n: int = DEFAULT_TOP_N,
-        target_class: int = DEFAULT_TARGET_CLASS,
-        device: str = DEFAULT_DEVICE
-    ):
-        self.n_folds = n_folds
-        self.n_background_samples = n_background_samples
-        self.n_test_samples = n_test_samples
-        self.figsize = figsize
-        self.top_n = top_n
-        self.target_class = target_class
-        self.device = device
-        self._validate()
-    
-    def _validate(self):
-        """설정값 검증."""
-        if self.n_folds <= 0:
-            raise ValueError("n_folds must be positive")
-        if self.n_background_samples <= 0:
-            raise ValueError("n_background_samples must be positive")
-        if self.n_test_samples <= 0:
-            raise ValueError("n_test_samples must be positive")
-        if self.top_n <= 0:
-            raise ValueError("top_n must be positive")
-
-
-# ==========================================
-# 2. 결과 데이터 클래스
-# ==========================================
-
-@dataclass
-class ImportanceData:
-    """Feature importance 데이터를 담는 클래스."""
-    df: pd.DataFrame
-    top_n_features: List[str]
-    top_n_share: float
-
-
-@dataclass
-class SHAPResult:
-    """SHAP 분석 결과를 담는 클래스."""
-    shap_train: Optional[np.ndarray]
-    X_train: Optional[pd.DataFrame]
-    shap_test: Optional[np.ndarray]
-    X_test: Optional[pd.DataFrame]
-    feature_names: List[str]
-    importance_train: Optional[ImportanceData]
-    importance_test: Optional[ImportanceData]
-    importance_df_combined: Optional[pd.DataFrame]
-    metadata: Dict[str, Any]
-    
-    def to_dict(self) -> Dict[str, Any]:
-        """기존 dict 형태로 변환 (하위 호환성)."""
-        return {
-            'shap_train': self.shap_train,
-            'X_train': self.X_train,
-            'shap_test': self.shap_test,
-            'X_test': self.X_test,
-            'feature_names': self.feature_names,
-            'top_n_features_train': self.importance_train.top_n_features if self.importance_train else None,
-            'top_n_share_train': self.importance_train.top_n_share if self.importance_train else None,
-            'importance_df_train': self.importance_train.df if self.importance_train else None,
-            'top_n_features_test': self.importance_test.top_n_features if self.importance_test else None,
-            'top_n_share_test': self.importance_test.top_n_share if self.importance_test else None,
-            'importance_df_test': self.importance_test.df if self.importance_test else None,
-            'importance_df_combined': self.importance_df_combined,
-            **self.metadata
-        }
-
-
-# ==========================================
-# 3. 헬퍼 클래스들
-# ==========================================
-
-class PreprocessingUtils:
-    """전처리 유틸리티 클래스."""
-    
-    @staticmethod
-    def add_derived_features(df: pd.DataFrame) -> pd.DataFrame:
-        """파생 변수 추가."""
-        df = df.copy()
-        df['hour_sin'] = np.sin(2 * np.pi * df['hour'] / 24)
-        df['hour_cos'] = np.cos(2 * np.pi * df['hour'] / 24)
-        df['month_sin'] = np.sin(2 * np.pi * df['month'] / 12)
-        df['month_cos'] = np.cos(2 * np.pi * df['month'] / 12)
-        df['ground_temp - temp_C'] = df['groundtemp'] - df['temp_C']
-        return df
-    
-    @staticmethod
-    def preprocessing(df: pd.DataFrame, config: SHAPConfig) -> pd.DataFrame:
-        """데이터 전처리."""
-        df = df[df.columns].copy()
-        df['year'] = df['year'].astype('int')
-        df['month'] = df['month'].astype('int')
-        df['hour'] = df['hour'].astype('int')
-        df = PreprocessingUtils.add_derived_features(df).copy()
-        df['multi_class'] = df['multi_class'].astype('int')
-        df.loc[df['wind_dir'] == '정온', 'wind_dir'] = "0"
-        df['wind_dir'] = df['wind_dir'].astype('int')
-        df = df[config.FEATURE_COLUMNS].copy()
-        return df
-
-
-class ShapValueExtractor:
-    """SHAP 값 추출 및 변환 유틸리티 클래스."""
-    
-    @staticmethod
-    def _to_numpy(value: Any) -> np.ndarray:
-        """다양한 타입을 numpy array로 변환."""
-        if isinstance(value, torch.Tensor):
-            return value.cpu().numpy()
-        elif isinstance(value, list):
-            return np.array(value)
-        elif isinstance(value, np.ndarray):
-            return value
-        else:
-            return np.array(value)
-    
-    @staticmethod
-    def _combine_num_cat_shap(shap_num: Any, shap_cat: Any) -> np.ndarray:
-        """Numerical과 categorical SHAP 값을 결합."""
-        shap_num = ShapValueExtractor._to_numpy(shap_num)
-        shap_cat = ShapValueExtractor._to_numpy(shap_cat)
-        return np.concatenate([shap_num, shap_cat], axis=-1)
-    
-    @staticmethod
-    def _extract_from_list_of_tuples(shap_list: List[Tuple]) -> np.ndarray:
-        """튜플 리스트에서 SHAP 값을 추출."""
-        combined = []
-        for shap_item in shap_list:
-            if isinstance(shap_item, tuple):
-                combined.append(ShapValueExtractor._combine_num_cat_shap(shap_item[0], shap_item[1]))
-            else:
-                combined.append(ShapValueExtractor._to_numpy(shap_item))
-        return np.array(combined)
-    
-    @staticmethod
-    def _extract_from_list(shap_list: List[Any]) -> np.ndarray:
-        """리스트에서 SHAP 값을 추출."""
-        if len(shap_list) == 0:
-            return np.array([])
-        
-        if isinstance(shap_list[0], tuple):
-            return ShapValueExtractor._extract_from_list_of_tuples(shap_list)
-        
-        if len(shap_list) == 2:
-            return ShapValueExtractor._combine_num_cat_shap(shap_list[0], shap_list[1])
-        
-        try:
-            return np.array([ShapValueExtractor._to_numpy(item) for item in shap_list])
-        except (ValueError, TypeError):
-            converted = [ShapValueExtractor._to_numpy(item) for item in shap_list]
-            if len(converted) == 2:
-                return np.concatenate(converted, axis=-1)
-            else:
-                return np.array(converted)
-    
-    @staticmethod
-    def extract(shap_values: Any, target_class: int = 0) -> np.ndarray:
-        """SHAP 값을 추출하여 numpy array로 변환."""
-        # 튜플인 경우
-        if isinstance(shap_values, tuple):
-            return ShapValueExtractor._combine_num_cat_shap(shap_values[0], shap_values[1])
-        
-        # 리스트인 경우
-        if isinstance(shap_values, list):
-            target_shap = shap_values[target_class]
-            
-            if isinstance(target_shap, tuple):
-                return ShapValueExtractor._combine_num_cat_shap(target_shap[0], target_shap[1])
-            
-            if isinstance(target_shap, list):
-                return ShapValueExtractor._extract_from_list(target_shap)
-            
-            return ShapValueExtractor._to_numpy(target_shap)
-        
-        # numpy array인 경우
-        if isinstance(shap_values, np.ndarray):
-            if len(shap_values.shape) == 3:
-                return shap_values[:, target_class, :]
-            return shap_values
-        
-        # 기타 타입
-        return ShapValueExtractor._to_numpy(shap_values)
-
-
-@contextmanager
-def suppress_shap_output():
-    """SHAP DeepExplainer의 출력을 억제하는 컨텍스트 매니저."""
-    old_stdout = sys.stdout
-    sys.stdout = StringIO()
-    try:
-        with warnings.catch_warnings():
-            warnings.simplefilter('ignore')
-            yield
-    finally:
-        sys.stdout = old_stdout
-
-
-class ModelWrapper(torch.nn.Module):
-    """PyTorch 모델을 SHAP DeepExplainer에 맞게 래핑하는 클래스."""
-    
-    def __init__(self, base_model, numerical_cols):
-        super().__init__()
-        self.base_model = base_model
-        self.num_features = len(numerical_cols)
-    
-    def forward(self, *inputs):
-        """SHAP이 리스트로 전달하는 경우 처리"""
-        if len(inputs) == 1 and isinstance(inputs[0], (list, tuple)):
-            x_num, x_cat = inputs[0]
-        elif len(inputs) == 2:
-            x_num, x_cat = inputs
-        else:
-            x = inputs[0]
-            x_num = x[:, :self.num_features]
-            x_cat = x[:, self.num_features:].long()
-            return self.base_model(x_num, x_cat)
-        
-        if x_cat.dtype != torch.long:
-            x_cat = x_cat.long()
-        
-        return self.base_model(x_num, x_cat)
-
-
-# ==========================================
-# 4. DataLoader 클래스
-# ==========================================
-
-class DataLoader:
-    """데이터 로딩 및 전처리를 담당하는 클래스."""
-    
-    def __init__(self, config: SHAPConfig):
-        self.config = config
-    
-    def load_raw_data(
-        self,
-        region: str,
-        data_sample: str,
-        fold_idx: int
-    ) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
-        """원본 데이터 로드 (전처리만 적용)."""
-        if data_sample == 'pure':
-            dat_path = f"../../data/data_for_modeling/{region}_train.csv"
-            train_data = PreprocessingUtils.preprocessing(
-                pd.read_csv(dat_path, index_col=0), self.config
-            )
-            train_data = train_data.loc[
-                ~train_data['year'].isin([self.config.TEST_YEAR - (fold_idx + 1)]), :
-            ]
-        else:
-            train_data = pd.read_csv(
-                f"../../data/data_oversampled/{data_sample}/{data_sample}_{fold_idx+1}_{region}.csv"
-            )
-            train_data = PreprocessingUtils.preprocessing(train_data, self.config)
-        
-        val_data = pd.read_csv(f"../../data/data_for_modeling/{region}_train.csv")
-        val_data = PreprocessingUtils.preprocessing(val_data, self.config)
-        val_data = val_data.loc[
-            val_data['year'].isin([self.config.TEST_YEAR - (fold_idx + 1)]), :
-        ]
-        
-        test_data = pd.read_csv(f"../../data/data_for_modeling/{region}_test.csv")
-        test_data = PreprocessingUtils.preprocessing(test_data, self.config)
-        test_data = test_data.loc[test_data['year'].isin([self.config.TEST_YEAR]), :]
-        
-        # 컬럼 정렬
-        common_columns = train_data.columns.to_list()
-        train_data = train_data[common_columns]
-        val_data = val_data[common_columns]
-        test_data = test_data[common_columns]
-        
-        return train_data, val_data, test_data
-    
-    def prepare_for_tree_model(
-        self,
-        region: str,
-        data_sample: str,
-        fold_idx: int
-    ) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
-        """Tree 모델용 데이터 준비."""
-        train_data = pd.read_csv(
-            f"../../data/data_oversampled/{data_sample}/{data_sample}_{fold_idx+1}_{region}.csv"
-        )
-        val_data = pd.read_csv(f"../../data/data_for_modeling/{region}_train.csv")
-        test_data = pd.read_csv(f"../../data/data_for_modeling/{region}_test.csv")
-        
-        train_data = PreprocessingUtils.preprocessing(train_data, self.config).copy()
-        val_data = PreprocessingUtils.preprocessing(val_data, self.config).copy()
-        test_data = PreprocessingUtils.preprocessing(test_data, self.config).copy()
-        
-        train_data = train_data.loc[
-            train_data['year'].isin(self.config.FOLD_YEARS[fold_idx]), :
-        ]
-        val_data = val_data.loc[
-            ~val_data['year'].isin(self.config.FOLD_YEARS[fold_idx]), :
-        ]
-        test_data = test_data.loc[test_data['year'].isin([self.config.TEST_YEAR]), :]
-        
-        return train_data, val_data, test_data
-    
-    def prepare_for_dl_model(
-        self,
-        region: str,
-        data_sample: str,
-        fold_idx: int,
-        scaler: Any
-    ) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.Index, pd.Index,
-               torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
-        """Deep Learning 모델용 데이터 준비 (scaler 적용, 텐서 변환)."""
-        train_data, val_data, test_data = self.load_raw_data(region, data_sample, fold_idx)
-        
-        X_train = train_data.drop(columns=self.config.DROP_COLUMNS)
-        X_val = val_data.drop(columns=self.config.DROP_COLUMNS)
-        X_test = test_data.drop(columns=self.config.DROP_COLUMNS)
-        
-        # 범주형 & 연속형 변수 분리
-        categorical_cols = X_train.select_dtypes(include=['object', 'category', 'int64']).columns
-        numerical_cols = X_train.select_dtypes(include=['float64']).columns
-        
-        # Label Encoding
-        label_encoders = {}
-        for col in categorical_cols:
-            le = LabelEncoder()
-            le.fit(X_train[col])
-            label_encoders[col] = le
-        
-        for col in categorical_cols:
-            X_train[col] = label_encoders[col].transform(X_train[col])
-            X_val[col] = label_encoders[col].transform(X_val[col])
-            X_test[col] = label_encoders[col].transform(X_test[col])
-        
-        # Scaler 변환
-        X_train[numerical_cols] = scaler.transform(X_train[numerical_cols])
-        X_val[numerical_cols] = scaler.transform(X_val[numerical_cols])
-        X_test[numerical_cols] = scaler.transform(X_test[numerical_cols])
-        
-        # 텐서 변환
-        X_train_num = torch.tensor(X_train[numerical_cols].values, dtype=torch.float32)
-        X_train_cat = torch.tensor(X_train[categorical_cols].values, dtype=torch.long)
-        X_val_num = torch.tensor(X_val[numerical_cols].values, dtype=torch.float32)
-        X_val_cat = torch.tensor(X_val[categorical_cols].values, dtype=torch.long)
-        X_test_num = torch.tensor(X_test[numerical_cols].values, dtype=torch.float32)
-        X_test_cat = torch.tensor(X_test[categorical_cols].values, dtype=torch.long)
-        
-        return (X_train, X_val, X_test, categorical_cols, numerical_cols,
-                X_train_num, X_train_cat, X_val_num, X_val_cat, X_test_num, X_test_cat)
-
-
-# ==========================================
-# 5. SHAPAnalyzer 베이스 클래스
-# ==========================================
-
-class SHAPAnalyzer(ABC):
-    """SHAP 분석을 위한 추상 베이스 클래스."""
-    
-    def __init__(self, config: SHAPConfig, data_loader: DataLoader):
-        self.config = config
-        self.data_loader = data_loader
-    
-    @abstractmethod
-    def create_explainer(self, model: Any, background_data: Any) -> Any:
-        """SHAP Explainer 생성."""
-        pass
-    
-    @abstractmethod
-    def calculate_shap_values(self, explainer: Any, sample_data: Any) -> Any:
-        """SHAP 값 계산."""
-        pass
-    
-    def _calculate_importance(
-        self,
-        shap_values: np.ndarray,
-        feature_names: List[str],
-        period_name: str = ''
-    ) -> ImportanceData:
-        """누적 설명력(Cumulative Share) 계산."""
-        global_importances = np.abs(shap_values).mean(axis=0)
-        
-        importance_df = pd.DataFrame({
-            'feature': feature_names,
-            'importance': global_importances
-        }).sort_values('importance', ascending=False)
-        
-        total_importance = importance_df['importance'].sum()
-        top_n_df = importance_df.head(self.config.top_n)
-        top_n_importance = top_n_df['importance'].sum()
-        top_n_share = (top_n_importance / total_importance) * 100
-        top_n_features = top_n_df['feature'].tolist()
-        
-        if period_name:
-            print(f"{period_name} - Top {self.config.top_n} Dominant Features: {top_n_features}")
-            print(f"{period_name} - Cumulative Importance Share (Top {self.config.top_n}): {top_n_share:.2f}%")
-        else:
-            print(f"Top {self.config.top_n} Dominant Features: {top_n_features}")
-            print(f"Cumulative Importance Share (Top {self.config.top_n}): {top_n_share:.2f}%")
-        print(f"{'-'*40}")
-        
-        return ImportanceData(
-            df=importance_df,
-            top_n_features=top_n_features,
-            top_n_share=top_n_share
-        )
-    
-    def _create_combined_importance_df(
-        self,
-        importance_train: Optional[ImportanceData],
-        importance_test: Optional[ImportanceData],
-        feature_names: List[str]
-    ) -> Optional[pd.DataFrame]:
-        """Train과 Test 중요도 DataFrame 통합."""
-        if importance_train is None and importance_test is None:
-            return None
-        
-        if importance_train is not None:
-            train_df = importance_train.df[['feature', 'importance']].copy()
-            train_df.columns = ['feature', 'importance_train']
-            total_importance_train = train_df['importance_train'].sum()
-            train_df['share_train'] = (train_df['importance_train'] / total_importance_train) * 100
-        else:
-            train_df = pd.DataFrame({'feature': feature_names})
-            train_df['importance_train'] = np.nan
-            train_df['share_train'] = np.nan
-        
-        if importance_test is not None:
-            test_df = importance_test.df[['feature', 'importance']].copy()
-            test_df.columns = ['feature', 'importance_test']
-            total_importance_test = test_df['importance_test'].sum()
-            test_df['share_test'] = (test_df['importance_test'] / total_importance_test) * 100
-        else:
-            test_df = pd.DataFrame({'feature': feature_names})
-            test_df['importance_test'] = np.nan
-            test_df['share_test'] = np.nan
-        
-        importance_df_combined = pd.merge(train_df, test_df, on='feature', how='outer')
-        
-        if importance_train is not None:
-            importance_df_combined = importance_df_combined.sort_values(
-                'importance_train', ascending=False
-            ).reset_index(drop=True)
-        elif importance_test is not None:
-            importance_df_combined = importance_df_combined.sort_values(
-                'importance_test', ascending=False
-            ).reset_index(drop=True)
-        else:
-            importance_df_combined = importance_df_combined.sort_values('feature').reset_index(drop=True)
-        
-        return importance_df_combined
-
-
-# ==========================================
-# 6. TreeSHAPAnalyzer 구현
-# ==========================================
-
-class TreeSHAPAnalyzer(SHAPAnalyzer):
-    """Tree 기반 모델(XGBoost, LightGBM) SHAP 분석 클래스."""
-    
-    def create_explainer(self, model: Any, background_data: Any = None) -> shap.TreeExplainer:
-        """TreeExplainer 생성."""
-        with warnings.catch_warnings():
-            warnings.filterwarnings('ignore', category=UserWarning, message='.*deprecated binary model format.*')
-            return shap.TreeExplainer(model.get_booster())
-    
-    def calculate_shap_values(self, explainer: shap.TreeExplainer, sample_data: pd.DataFrame) -> np.ndarray:
-        """Tree SHAP 값 계산."""
-        shap_values = explainer.shap_values(sample_data)
-        if isinstance(shap_values, list):
-            return shap_values[self.config.target_class]
-        return shap_values
-    
-    def analyze(
-        self,
-        model_name: str,
-        region: str,
-        data_sample: str,
-        show_plot: bool = True,
-        show_train_plot: bool = True,
-        show_test_plot: bool = True,
-        suppress_warnings: bool = True
-    ) -> SHAPResult:
-        """Tree 모델 SHAP 분석 수행."""
-        # 모델 로드
-        model_path = f"../save_model/{model_name}_optima/{model_name}_{data_sample}_{region}.pkl"
-        models = load(model_path)
-        
-        if len(models) != self.config.n_folds:
-            raise ValueError(f"로드된 모델 개수({len(models)})가 n_folds({self.config.n_folds})와 일치하지 않습니다.")
-        
-        # 각 Fold에 대해 SHAP 값 계산
-        oof_shap_values = []
-        oof_X_data = []
-        test_shap_values = []
-        
-        for fold_idx in range(self.config.n_folds):
-            _, val_data, test_data = self.data_loader.prepare_for_tree_model(
-                region, data_sample, fold_idx
-            )
-            
-            X_val = val_data.drop(columns=self.config.DROP_COLUMNS)
-            if fold_idx == 0:
-                X_test = test_data.drop(columns=self.config.DROP_COLUMNS)
-            
-            explainer = self.create_explainer(models[fold_idx])
-            shap_val = self.calculate_shap_values(explainer, X_val)
-            oof_shap_values.append(shap_val)
-            oof_X_data.append(X_val)
-            
-            shap_test = self.calculate_shap_values(explainer, X_test)
-            test_shap_values.append(shap_test)
-        
-        # 데이터 통합
-        all_shap_train = np.concatenate(oof_shap_values, axis=0)
-        all_X_train = pd.concat(oof_X_data, axis=0)
-        all_shap_test = np.mean(test_shap_values, axis=0)
-        feature_names = list(all_X_train.columns)
-        
-        # 중요도 계산
-        importance_train = self._calculate_importance(
-            all_shap_train, feature_names, period_name='Train Period'
-        )
-        importance_test = self._calculate_importance(
-            all_shap_test, feature_names, period_name='Test Period'
-        )
-        print()
-        
-        importance_df_combined = self._create_combined_importance_df(
-            importance_train, importance_test, feature_names
-        )
-        
-        # 시각화
-        if show_plot:
-            visualizer = SHAPVisualizer(self.config)
-            visualizer.plot_summary(
-                all_shap_train, all_X_train, all_shap_test, X_test,
-                feature_names, importance_train, importance_test,
-                show_train_plot=show_train_plot, show_test_plot=show_test_plot,
-                model_name=model_name, data_sample=data_sample, region=region
-            )
-        
-        return SHAPResult(
-            shap_train=all_shap_train,
-            X_train=all_X_train,
-            shap_test=all_shap_test,
-            X_test=X_test,
-            feature_names=feature_names,
-            importance_train=importance_train,
-            importance_test=importance_test,
-            importance_df_combined=importance_df_combined,
-            metadata={'model_name': model_name, 'region': region, 'data_sample': data_sample}
-        )
-
-
-# ==========================================
-# 7. DeepSHAPAnalyzer 구현
-# ==========================================
-
-class DeepSHAPAnalyzer(SHAPAnalyzer):
-    """Deep Learning 모델 SHAP 분석 클래스."""
-    
-    def __init__(self, config: SHAPConfig, data_loader: DataLoader, device: str = 'cpu'):
-        super().__init__(config, data_loader)
-        self.device = device
-    
-    def create_explainer(
-        self,
-        model: torch.nn.Module,
-        background_data: Tuple[torch.Tensor, torch.Tensor, pd.Index],
-        model_name: str
-    ) -> shap.DeepExplainer:
-        """DeepExplainer 생성."""
-        X_bg_num, X_bg_cat, numerical_cols = background_data
-        wrapped_model = ModelWrapper(model, numerical_cols)
-        wrapped_model.to(self.device)
-        wrapped_model.eval()
-        
-        with suppress_shap_output():
-            if model_name in ('ft_transformer', 'deepgbm'):
-                X_bg_cat_float = X_bg_cat.float()
-                explainer = shap.DeepExplainer(wrapped_model, [X_bg_num, X_bg_cat_float])
-            elif model_name == 'resnet_like':
-                X_bg_combined = torch.cat([X_bg_num, X_bg_cat.float()], dim=1)
-                explainer = shap.DeepExplainer(wrapped_model, X_bg_combined)
-            else:
-                raise ValueError(f"Unknown model_name: {model_name}")
-        
-        return explainer
-    
-    def calculate_shap_values(
-        self,
-        explainer: shap.DeepExplainer,
-        sample_data: Tuple[torch.Tensor, torch.Tensor],
-        model_name: str
-    ) -> Any:
-        """Deep SHAP 값 계산."""
-        X_sample_num, X_sample_cat = sample_data
-        
-        with suppress_shap_output():
-            if model_name in ('ft_transformer', 'deepgbm'):
-                X_sample_cat_float = X_sample_cat.float()
-                shap_values = explainer.shap_values([X_sample_num, X_sample_cat_float])
-            elif model_name == 'resnet_like':
-                X_sample_combined = torch.cat([X_sample_num, X_sample_cat.float()], dim=1)
-                shap_values = explainer.shap_values(X_sample_combined)
-            else:
-                raise ValueError(f"Unknown model_name: {model_name}")
-        
-        return shap_values
-    
-    def analyze(
-        self,
-        model_name: str,
-        region: str,
-        data_sample: str,
-        show_plot: bool = True,
-        show_train_plot: bool = True,
-        show_test_plot: bool = True
-    ) -> Optional[SHAPResult]:
-        """Deep Learning 모델 SHAP 분석 수행."""
-        print(f"[{model_name.upper()}] {region} - {data_sample} SHAP Analysis Start...")
-        
-        # 모델 및 scaler 로드
-        try:
-            model_path = f"../save_model/{model_name}_optima/{model_name}_{data_sample}_{region}.pkl"
-            models = load(model_path)
-            
-            if data_sample == 'pure':
-                scaler_filename = f'{model_name}_pure_{region}_scaler.pkl'
-            else:
-                scaler_filename = f'{model_name}_{data_sample}_{region}_scaler.pkl'
-            scaler_path = f'../save_model/{model_name}_optima/scaler/{scaler_filename}'
-            scalers = load(scaler_path)
-        except Exception as e:
-            print(f"모델 또는 scaler 로드 실패: {e}")
-            return None
-        
-        if len(models) != self.config.n_folds or len(scalers) != self.config.n_folds:
-            raise ValueError(
-                f"모델 개수({len(models)}) 또는 scaler 개수({len(scalers)})가 "
-                f"n_folds({self.config.n_folds})와 일치하지 않습니다."
-            )
-        
-        # 각 fold의 SHAP 값 저장
-        all_shap_train = []
-        all_shap_test = []
-        all_train_data = []
-        all_test_data = []
-        feature_names = None
-        
-        for fold_idx in range(self.config.n_folds):
-            print(f"  Processing Fold {fold_idx + 1}...")
-            
-            try:
-                (X_train, X_val, X_test, categorical_cols, numerical_cols,
-                 X_train_num, X_train_cat, X_val_num, X_val_cat, X_test_num, X_test_cat) = \
-                    self.data_loader.prepare_for_dl_model(region, data_sample, fold_idx, scalers[fold_idx])
-                
-                if feature_names is None:
-                    feature_names = list(numerical_cols) + list(categorical_cols)
-                
-                model = models[fold_idx]
-                model.to(self.device)
-                model.eval()
-                
-                # 배경 데이터 샘플링
-                n_bg = min(self.config.n_background_samples, len(X_val_num))
-                bg_indices = np.random.choice(len(X_val_num), n_bg, replace=False)
-                X_bg_num = X_val_num[bg_indices].to(self.device)
-                X_bg_cat = X_val_cat[bg_indices].to(self.device)
-                
-                # Validation 데이터 샘플링
-                n_train = min(self.config.n_test_samples, len(X_val_num))
-                available_indices = np.setdiff1d(np.arange(len(X_val_num)), bg_indices)
-                if len(available_indices) < n_train:
-                    train_indices = available_indices
-                else:
-                    train_indices = np.random.choice(available_indices, n_train, replace=False)
-                X_train_sample_num = X_val_num[train_indices].to(self.device)
-                X_train_sample_cat = X_val_cat[train_indices].to(self.device)
-                
-                # 테스트 데이터 샘플링
-                n_test = min(self.config.n_test_samples, len(X_test_num))
-                test_indices = np.random.choice(len(X_test_num), n_test, replace=False)
-                X_test_sample_num = X_test_num[test_indices].to(self.device)
-                X_test_sample_cat = X_test_cat[test_indices].to(self.device)
-                
-                # SHAP Explainer 생성 및 계산
-                try:
-                    explainer = self.create_explainer(
-                        model, (X_bg_num, X_bg_cat, numerical_cols), model_name
-                    )
-                    
-                    shap_values_train = self.calculate_shap_values(
-                        explainer, (X_train_sample_num, X_train_sample_cat), model_name
-                    )
-                    shap_val_train = ShapValueExtractor.extract(shap_values_train, self.config.target_class)
-                    all_shap_train.append(shap_val_train)
-                    X_train_original = X_val.iloc[train_indices].copy()
-                    all_train_data.append(X_train_original)
-                    
-                    shap_values_test = self.calculate_shap_values(
-                        explainer, (X_test_sample_num, X_test_sample_cat), model_name
-                    )
-                    shap_val_test = ShapValueExtractor.extract(shap_values_test, self.config.target_class)
-                    all_shap_test.append(shap_val_test)
-                    X_test_original = X_test.iloc[test_indices].copy()
-                    all_test_data.append(X_test_original)
-                    
-                except Exception as e:
-                    print(f"  Warning: Fold {fold_idx + 1} SHAP 계산 실패: {e}")
-                    continue
-            except Exception as e:
-                print(f"  Warning: Fold {fold_idx + 1} 처리 실패: {e}")
-                continue
-        
-        if len(all_shap_train) == 0 and len(all_shap_test) == 0:
-            print("  모든 fold에서 SHAP 계산 실패")
-            return None
-        
-        # 데이터 통합
-        all_shap_train_combined = np.concatenate(all_shap_train, axis=0) if len(all_shap_train) > 0 else None
-        all_train_data_combined = pd.concat(all_train_data, axis=0).reset_index(drop=True) if len(all_train_data) > 0 else None
-        
-        if len(all_shap_test) > 0:
-            all_shap_test_combined = np.mean(all_shap_test, axis=0)
-            all_test_data_combined = all_test_data[0] if len(all_test_data) > 0 else None
-        else:
-            all_shap_test_combined = None
-            all_test_data_combined = None
-        
-        # 중요도 계산
-        if all_shap_train_combined is not None:
-            importance_train = self._calculate_importance(
-                all_shap_train_combined, feature_names, period_name='Train Period'
-            )
-        else:
-            importance_train = None
-        
-        if all_shap_test_combined is not None:
-            importance_test = self._calculate_importance(
-                all_shap_test_combined, feature_names, period_name='Test Period'
-            )
-            print()
-        else:
-            importance_test = None
-        
-        importance_df_combined = self._create_combined_importance_df(
-            importance_train, importance_test, feature_names
-        )
-        
-        # 시각화
-        if show_plot:
-            visualizer = SHAPVisualizer(self.config)
-            visualizer.plot_summary(
-                all_shap_train_combined, all_train_data_combined,
-                all_shap_test_combined, all_test_data_combined,
-                feature_names, importance_train, importance_test,
-                show_train_plot=show_train_plot, show_test_plot=show_test_plot,
-                model_name=model_name, data_sample=data_sample, region=region
-            )
-        
-        print(f"  [{model_name.upper()}] SHAP Analysis Complete!\n")
-        
-        return SHAPResult(
-            shap_train=all_shap_train_combined,
-            X_train=all_train_data_combined[feature_names] if all_train_data_combined is not None else None,
-            shap_test=all_shap_test_combined,
-            X_test=all_test_data_combined[feature_names] if all_test_data_combined is not None else None,
-            feature_names=feature_names,
-            importance_train=importance_train,
-            importance_test=importance_test,
-            importance_df_combined=importance_df_combined,
-            metadata={'model_name': model_name, 'region': region, 'data_sample': data_sample}
-        )
-
-
-# ==========================================
-# 8. SHAPVisualizer 클래스
-# ==========================================
-
-class SHAPVisualizer:
-    """SHAP 시각화를 담당하는 클래스."""
-    
-    def __init__(self, config: SHAPConfig):
-        self.config = config
-    
-    def _calculate_xlim(
-        self,
-        shap_train: Optional[np.ndarray],
-        shap_test: Optional[np.ndarray]
-    ) -> Tuple[Optional[float], Optional[float]]:
-        """x축 범위 계산."""
-        if shap_train is not None and shap_test is not None:
-            x_min = min(shap_train.min(), shap_test.min())
-            x_max = max(shap_train.max(), shap_test.max())
-        elif shap_train is not None:
-            x_min, x_max = shap_train.min(), shap_train.max()
-        elif shap_test is not None:
-            x_min, x_max = shap_test.min(), shap_test.max()
-        else:
-            return None, None
-        
-        margin = (x_max - x_min) * self.config.DEFAULT_PLOT_MARGIN
-        x_min -= margin
-        x_max += margin
-        return x_min, x_max
-    
-    def plot_summary(
-        self,
-        shap_train: Optional[np.ndarray],
-        X_train: Optional[pd.DataFrame],
-        shap_test: Optional[np.ndarray],
-        X_test: Optional[pd.DataFrame],
-        feature_names: List[str],
-        importance_train: Optional[ImportanceData],
-        importance_test: Optional[ImportanceData],
-        show_train_plot: bool = True,
-        show_test_plot: bool = True,
-        model_name: Optional[str] = None,
-        data_sample: Optional[str] = None,
-        region: Optional[str] = None
-    ) -> None:
-        """SHAP summary plot 생성."""
-        num_plots = int(show_train_plot and shap_train is not None) + int(show_test_plot and shap_test is not None)
-        if num_plots == 0:
-            return
-        
-        x_min, x_max = self._calculate_xlim(shap_train, shap_test)
-        
-        # Title prefix 생성
-        title_parts = []
-        if model_name is not None:
-            title_parts.append(model_name.upper())
-        if data_sample is not None:
-            title_parts.append(data_sample)
-        if region is not None:
-            title_parts.append(region.upper())
-        
-        title_prefix = " - ".join(title_parts) + "\n" if title_parts else ""
-        
-        if num_plots == 2:
-            fig, axes = plt.subplots(1, 2, figsize=self.config.figsize)
-            
-            if show_train_plot and shap_train is not None:
-                plt.sca(axes[0])
-                shap.summary_plot(
-                    shap_train,
-                    X_train[feature_names] if isinstance(X_train, pd.DataFrame) else X_train,
-                    show=False, plot_size=None
-                )
-                if x_min is not None and x_max is not None:
-                    plt.xlim(x_min, x_max)
-                share_text = (
-                    f"\nTop {self.config.top_n} Share: {importance_train.top_n_share:.1f}%"
-                    if importance_train else ""
-                )
-                plt.title(f"{title_prefix}Train Period (2018-20): Class {self.config.target_class}{share_text}")
-            
-            if show_test_plot and shap_test is not None:
-                plt.sca(axes[1])
-                shap.summary_plot(
-                    shap_test,
-                    X_test[feature_names] if isinstance(X_test, pd.DataFrame) else X_test,
-                    show=False, plot_size=None
-                )
-                if x_min is not None and x_max is not None:
-                    plt.xlim(x_min, x_max)
-                share_text = (
-                    f"\nTop {self.config.top_n} Share: {importance_test.top_n_share:.1f}%"
-                    if importance_test else ""
-                )
-                plt.title(f"{title_prefix}Test Period (2021): Class {self.config.target_class}{share_text}")
-        else:
-            plt.figure(figsize=(self.config.figsize[0]//2, self.config.figsize[1]))
-            if show_train_plot and shap_train is not None:
-                data = X_train[feature_names] if isinstance(X_train, pd.DataFrame) else X_train
-                shap_v = shap_train
-                share_text = (
-                    f"\nTop {self.config.top_n} Share: {importance_train.top_n_share:.1f}%"
-                    if importance_train else ""
-                )
-                title = f"{title_prefix}Train Period (2018-20): Class {self.config.target_class}{share_text}"
-            elif show_test_plot and shap_test is not None:
-                data = X_test[feature_names] if isinstance(X_test, pd.DataFrame) else X_test
-                shap_v = shap_test
-                share_text = (
-                    f"\nTop {self.config.top_n} Share: {importance_test.top_n_share:.1f}%"
-                    if importance_test else ""
-                )
-                title = f"{title_prefix}Test Period (2021): Class {self.config.target_class}{share_text}"
-            else:
-                return
-            
-            shap.summary_plot(shap_v, data, show=False)
-            if x_min is not None and x_max is not None:
-                plt.xlim(x_min, x_max)
-            plt.title(title)
-        
-        plt.tight_layout()
-        plt.show()
-    
-    def plot_elbow(self, importance_df_combined: pd.DataFrame) -> None:
-        """Elbow plot 생성."""
-        elbow = importance_df_combined.loc[:, ['feature', 'importance_train', 'share_train']]
-        elbow = elbow.sort_values('importance_train', ascending=False)
-        plt.figure(figsize=(8, 5))
-        plt.plot(range(1, len(elbow) + 1), elbow['share_train'].values, marker='o')
-        plt.xlabel('Feature Rank')
-        plt.ylabel('Share Train (%)')
-        plt.title('Elbow Plot for Feature Importance (share_train)')
-        plt.xticks(range(1, len(elbow) + 1))
-        plt.grid(True, linestyle='--', alpha=0.5)
-        plt.show()
-
-
-# ==========================================
-# 9. WassersteinCalculator 클래스
-# ==========================================
-
-class WassersteinCalculator:
-    """Wasserstein Distance 계산을 담당하는 클래스."""
-    
-    def __init__(self, config: SHAPConfig, data_loader: DataLoader):
-        self.config = config
-        self.data_loader = data_loader
-    
-    def _compute_wd_for_feature(
-        self,
-        train_dist: np.ndarray,
-        test_dist: np.ndarray
-    ) -> Dict[str, float]:
-        """단일 feature에 대한 WD 계산."""
-        wd_val = wasserstein_distance(train_dist, test_dist)
-        return {
-            'WD_Score': wd_val,
-            'Train_Mean': np.mean(train_dist),
-            'Test_Mean': np.mean(test_dist),
-            'Train_Std': np.std(train_dist),
-            'Test_Std': np.std(test_dist),
-            'Train_Count': len(train_dist),
-            'Test_Count': len(test_dist)
-        }
-    
-    def calculate(
-        self,
-        region: str,
-        data_sample: str,
-        top_features: List[str],
-        model_type: str = 'tree',
-        fold_idx: int = 0
-    ) -> pd.DataFrame:
-        """Wasserstein Distance 계산."""
-        # 데이터 로드
-        if model_type == 'tree':
-            train_df, val_df, test_df = self.data_loader.prepare_for_tree_model(
-                region, data_sample, fold_idx
-            )
-        elif model_type == 'dl':
-            train_df, val_df, test_df = self.data_loader.load_raw_data(
-                region, data_sample, fold_idx
-            )
-        else:
-            raise ValueError(f"Unknown model_type: {model_type}. Must be 'tree' or 'dl'")
-        
-        train_period = pd.concat([train_df, val_df], axis=0)
-        test_period = test_df
-        
-        results = []
-        
-        print(f"[{region.upper()}] Wasserstein Distance Analysis")
-        print(f"Model Type: {model_type.upper()}, Data Sample: {data_sample}")
-        print(f"{'-'*60}")
-        
-        for feature in top_features:
-            if feature in train_period.columns:
-                dist_train = train_period[feature].dropna().values
-                dist_test = test_period[feature].dropna().values
-                
-                if len(dist_train) == 0 or len(dist_test) == 0:
-                    print(f"Warning: {feature} 변수에 유효한 데이터가 없습니다.")
-                    continue
-                
-                wd_metrics = self._compute_wd_for_feature(dist_train, dist_test)
-                results.append({
-                    'Region': region,
-                    'Feature': feature,
-                    **wd_metrics
-                })
-                print(f"Feature: {feature:20} | WD: {wd_metrics['WD_Score']:.4f} | "
-                      f"Train Mean: {wd_metrics['Train_Mean']:.4f} | Test Mean: {wd_metrics['Test_Mean']:.4f}")
-            else:
-                print(f"Warning: {feature} 변수를 찾을 수 없습니다.")
-        
-        print()
-        return pd.DataFrame(results)
-    
-    def calculate_from_shap_result(
-        self,
-        shap_result: Dict[str, Any],
-        top_n: int = 3,
-        use_train_features: bool = True,
-        model_type: str = 'tree',
-        fold_idx: int = 0
-    ) -> pd.DataFrame:
-        """SHAP 분석 결과에서 상위 변수를 자동 추출하여 WD 계산."""
-        # 상위 변수 추출
-        if use_train_features:
-            if 'top_n_features_train' in shap_result and shap_result['top_n_features_train'] is not None:
-                top_features = shap_result['top_n_features_train'][:top_n]
-            elif 'importance_df_train' in shap_result and shap_result['importance_df_train'] is not None:
-                top_features = shap_result['importance_df_train'].head(top_n)['feature'].tolist()
-            else:
-                raise ValueError("SHAP 결과에서 Train 상위 변수를 찾을 수 없습니다.")
-        else:
-            if 'top_n_features_test' in shap_result and shap_result['top_n_features_test'] is not None:
-                top_features = shap_result['top_n_features_test'][:top_n]
-            elif 'importance_df_test' in shap_result and shap_result['importance_df_test'] is not None:
-                top_features = shap_result['importance_df_test'].head(top_n)['feature'].tolist()
-            else:
-                raise ValueError("SHAP 결과에서 Test 상위 변수를 찾을 수 없습니다.")
-        
-        region = shap_result.get('region', 'unknown')
-        data_sample = shap_result.get('data_sample', 'unknown')
-        
-        return self.calculate(region, data_sample, top_features, model_type, fold_idx)
-
-
-# ==========================================
-# 10. 하위 호환성 레이어
-# ==========================================
-
-def analyze_shap_values_across_folds(
-    model_name: str,
-    region: str,
-    data_sample: str,
-    target_class: int = 0,
-    n_folds: int = 3,
-    figsize: Tuple[int, int] = (20, 10),
-    show_plot: bool = True,
-    show_train_plot: bool = True,
-    show_test_plot: bool = True,
-    suppress_warnings: bool = True,
-    top_n: int = 3
-) -> Dict[str, Any]:
-    """Tree 모델 SHAP 분석 (하위 호환성 래퍼 함수).
-    
-    Deprecated: Use TreeSHAPAnalyzer instead.
-    """
-    config = SHAPConfig(
-        n_folds=n_folds,
-        figsize=figsize,
-        top_n=top_n,
-        target_class=target_class
-    )
-    data_loader = DataLoader(config)
-    analyzer = TreeSHAPAnalyzer(config, data_loader)
-    result = analyzer.analyze(
-        model_name, region, data_sample,
-        show_plot=show_plot,
-        show_train_plot=show_train_plot,
-        show_test_plot=show_test_plot
-    )
-    return result.to_dict()
-
-
-def analyze_dl_model_shap(
-    model_name: str,
-    region: str,
-    data_sample: str,
-    target_class: int = 0,
-    n_folds: int = 3,
-    n_background_samples: int = 200,
-    n_test_samples: int = 500,
-    device: str = 'cpu',
-    figsize: Tuple[int, int] = (20, 8),
-    show_plot: bool = True,
-    show_train_plot: bool = True,
-    show_test_plot: bool = True,
-    top_n: int = 3
-) -> Optional[Dict[str, Any]]:
-    """Deep Learning 모델 SHAP 분석 (하위 호환성 래퍼 함수).
-    
-    Deprecated: Use DeepSHAPAnalyzer instead.
-    """
-    config = SHAPConfig(
-        n_folds=n_folds,
-        n_background_samples=n_background_samples,
-        n_test_samples=n_test_samples,
-        figsize=figsize,
-        top_n=top_n,
-        target_class=target_class,
-        device=device
-    )
-    data_loader = DataLoader(config)
-    analyzer = DeepSHAPAnalyzer(config, data_loader, device=device)
-    result = analyzer.analyze(
-        model_name, region, data_sample,
-        show_plot=show_plot,
-        show_train_plot=show_train_plot,
-        show_test_plot=show_test_plot
-    )
-    return result.to_dict() if result else None
-
-
-def calculate_wasserstein_distance(
-    region: str,
-    data_sample: str,
-    top_features: List[str],
-    model_type: str = 'tree',
-    fold_idx: int = 0
-) -> pd.DataFrame:
-    """Wasserstein Distance 계산 (하위 호환성 래퍼 함수).
-    
-    Deprecated: Use WassersteinCalculator instead.
-    """
-    config = SHAPConfig()
-    data_loader = DataLoader(config)
-    calculator = WassersteinCalculator(config, data_loader)
-    return calculator.calculate(region, data_sample, top_features, model_type, fold_idx)
-
-
-def calculate_wd_from_shap_result(
-    shap_result: Dict[str, Any],
-    top_n: int = 3,
-    use_train_features: bool = True,
-    model_type: str = 'tree',
-    fold_idx: int = 0
-) -> pd.DataFrame:
-    """SHAP 결과에서 WD 계산 (하위 호환성 래퍼 함수).
-    
-    Deprecated: Use WassersteinCalculator instead.
-    """
-    config = SHAPConfig()
-    data_loader = DataLoader(config)
-    calculator = WassersteinCalculator(config, data_loader)
-    return calculator.calculate_from_shap_result(
-        shap_result, top_n, use_train_features, model_type, fold_idx
-    )
-
-
-def elbow_plot(result: Dict[str, Any]) -> None:
-    """Elbow plot 생성 (하위 호환성 래퍼 함수)."""
-    config = SHAPConfig()
-    visualizer = SHAPVisualizer(config)
-    if 'importance_df_combined' in result:
-        visualizer.plot_elbow(result['importance_df_combined'])
-
-
-# ==========================================
-# 유틸리티 함수 (기존 호환성 유지)
-# ==========================================
-
-def calculate_csi(y_true: np.ndarray, y_pred: np.ndarray) -> float:
-    """CSI(Critical Success Index) 점수를 계산합니다."""
-    cm = confusion_matrix(y_true, y_pred)
-    H = cm[0, 0] + cm[1, 1]
-    F = cm[1, 0] + cm[2, 0] + cm[0, 1] + cm[2, 1]
-    M = cm[0, 2] + cm[1, 2]
-    csi = H / (H + F + M + 1e-10)
-    return csi
-
-
-def csi_metric(y_true: np.ndarray, pred_prob: np.ndarray) -> Tuple[str, float, bool]:
-    """LightGBM용 CSI 메트릭 함수."""
-    y_pred_binary = np.argmax(pred_prob, axis=1)
-    score = calculate_csi(y_true, y_pred_binary)
-    return 'CSI', score, True
-
-
-def eval_metric_csi(y_true: np.ndarray, pred_prob: np.ndarray) -> float:
-    """XGBoost용 CSI 메트릭 함��."""
-    pred = np.argmax(pred_prob, axis=1)
-    csi = calculate_csi(y_true, pred)
-    return -1 * csi
-
-
-# 기존 상수들 (하위 호환성)
-FOLD_YEARS = SHAPConfig.FOLD_YEARS
-TEST_YEAR = SHAPConfig.TEST_YEAR
-DROP_COLUMNS = SHAPConfig.DROP_COLUMNS
-FEATURE_COLUMNS = SHAPConfig.FEATURE_COLUMNS

Analysis_code/7.ensemble/utils.py

@@ -1,512 +0,0 @@
-# 표준 라이브러리
-import os
-import sys
-import warnings
-from typing import List, Tuple, Any
-
-# 서드파티 라이브러리
-import numpy as np
-import pandas as pd
-import torch
-import torch.nn.functional as F
-from joblib import load
-from sklearn.metrics import confusion_matrix
-from sklearn.preprocessing import LabelEncoder
-
-# 경고 무시
-warnings.filterwarnings('ignore')
-
-# ==========================================
-# 상수 정의
-# ==========================================
-TEST_YEAR = 2021
-DROP_COLUMNS = ['multi_class', 'year']
-FEATURE_COLUMNS = [
-    'temp_C', 'precip_mm', 'wind_speed', 'wind_dir', 'hm',
-    'vap_pressure', 'dewpoint_C', 'loc_pressure', 'sea_pressure',
-    'solarRad', 'snow_cm', 'cloudcover', 'lm_cloudcover', 'low_cloudbase',
-    'groundtemp', 'O3', 'NO2', 'PM10', 'PM25', 'year',
-    'month', 'hour', 'ground_temp - temp_C', 'hour_sin', 'hour_cos',
-    'month_sin', 'month_cos', 'multi_class'
-]
-
-# 모델 타입 상수
-TREE_MODELS = ['xgb', 'lgb']
-DL_MODELS = ['ft_transformer', 'resnet_like', 'deepgbm']
-
-# ==========================================
-# 경로 설정
-# ==========================================
-def _setup_model_path() -> str:
-    """모델 클래스 import를 위한 경로를 설정하고 반환합니다.
-    
-    Returns:
-        models 디렉토리의 절대 경로
-    """
-    try:
-        current_file_dir = os.path.dirname(os.path.abspath(__file__))
-        models_path = os.path.abspath(os.path.join(current_file_dir, '../models'))
-    except NameError:
-        # 노트북 환경에서 __file__이 없는 경우
-        cwd = os.getcwd()
-        if 'Analysis_code' in cwd:
-            analysis_code_dir = cwd[:cwd.index('Analysis_code') + len('Analysis_code')]
-            models_path = os.path.join(analysis_code_dir, 'models')
-        else:
-            models_path = '/workspace/visibility_prediction/Analysis_code/models'
-    
-    if models_path not in sys.path:
-        sys.path.insert(0, models_path)
-    
-    return models_path
-
-# 모델 경로 설정 및 import
-_setup_model_path()
-from ft_transformer import FTTransformer
-from resnet_like import ResNetLike
-from deepgbm import DeepGBM
-
-# ==========================================
-# 메트릭 함수 (pickle 호환성)
-# ==========================================
-def calculate_csi(y_true: np.ndarray, y_pred: np.ndarray) -> float:
-    """CSI(Critical Success Index) 점수를 계산합니다.
-    
-    Args:
-        y_true: 실제 레이블
-        y_pred: 예측 레이블
-        
-    Returns:
-        CSI 점수 (0~1 사이의 값)
-    """
-    cm = confusion_matrix(y_true, y_pred)
-    
-    # 혼동 행렬에서 H(Hit), F(False alarm), M(Miss) 추출
-    H = cm[0, 0] + cm[1, 1]
-    F = cm[1, 0] + cm[2, 0] + cm[0, 1] + cm[2, 1]
-    M = cm[0, 2] + cm[1, 2]
-    
-    # CSI 계산
-    csi = H / (H + F + M + 1e-10)
-    return csi
-
-def eval_metric_csi(y_true: np.ndarray, pred_prob: np.ndarray) -> float:
-    """XGBoost용 CSI 메트릭 함수.
-    
-    Args:
-        y_true: 실제 레이블
-        pred_prob: 예측 확률
-        
-    Returns:
-        CSI 점수의 음수값
-    """
-    pred = np.argmax(pred_prob, axis=1)
-    csi = calculate_csi(y_true, pred)
-    return -1 * csi
-
-def csi_metric(y_true: np.ndarray, pred_prob: np.ndarray) -> Tuple[str, float, bool]:
-    """LightGBM용 CSI 메트릭 함수.
-    
-    Args:
-        y_true: 실제 레이블
-        pred_prob: 예측 확률 (shape: [n_samples, n_classes])
-        
-    Returns:
-        ('CSI', score, higher_better) 튜플
-    """
-    y_pred_binary = np.argmax(pred_prob, axis=1)
-    score = calculate_csi(y_true, y_pred_binary)
-    return 'CSI', score, True
-
-# ==========================================
-# Pickle 호환성 함수 등록
-# ==========================================
-def _register_pickle_functions(model_name: str) -> None:
-    """모델 로드를 위해 __main__ 모듈에 필요한 함수를 등록합니다.
-    
-    모델이 저장될 때 __main__ 모듈에서 정의된 함수를 참조했을 수 있으므로,
-    pickle이 모델을 로드할 때 함수를 찾을 수 있도록 등록합니다.
-    
-    Args:
-        model_name: 모델 이름 ('xgb' 또는 'lgb')
-    """
-    if '__main__' not in sys.modules:
-        return
-    
-    main_module = sys.modules['__main__']
-    main_module.calculate_csi = calculate_csi
-    
-    if model_name == 'xgb':
-        main_module.eval_metric_csi = eval_metric_csi
-    elif model_name == 'lgb':
-        main_module.csi_metric = csi_metric
-
-# ==========================================
-# 데이터 전처리 함수
-# ==========================================
-def add_derived_features(df: pd.DataFrame) -> pd.DataFrame:
-    """제거했던 파생 변수들을 복구합니다.
-    
-    Args:
-        df: 원본 데이터프레임
-        
-    Returns:
-        파생 변수가 추가된 데이터프레임
-    """
-    df = df.copy()
-    df['hour_sin'] = np.sin(2 * np.pi * df['hour'] / 24)
-    df['hour_cos'] = np.cos(2 * np.pi * df['hour'] / 24)
-    df['month_sin'] = np.sin(2 * np.pi * df['month'] / 12)
-    df['month_cos'] = np.cos(2 * np.pi * df['month'] / 12)
-    df['ground_temp - temp_C'] = df['groundtemp'] - df['temp_C']
-    return df
-
-def preprocessing(df: pd.DataFrame) -> pd.DataFrame:
-    """데이터 전처리 함수.
-    
-    Args:
-        df: 원본 데이터프레임
-        
-    Returns:
-        전처리된 데이터프레임
-    """
-    df = df[df.columns].copy()
-    df['year'] = df['year'].astype('int')
-    df['month'] = df['month'].astype('int')
-    df['hour'] = df['hour'].astype('int')
-    df = add_derived_features(df).copy()
-    df['multi_class'] = df['multi_class'].astype('int')
-    df.loc[df['wind_dir'] == '정온', 'wind_dir'] = "0"
-    df['wind_dir'] = df['wind_dir'].astype('int')
-    df = df[FEATURE_COLUMNS].copy()
-    return df
-
-# ==========================================
-# 모델 로드 함수
-# ==========================================
-def load_model(model_name: str, region: str, data_sample: str) -> List[Any]:
-    """모델을 로드하는 함수.
-    
-    Args:
-        model_name: 모델 이름 ('xgb', 'lgb', 'ft_transformer', 'resnet_like', 'deepgbm')
-        region: 지역명 (예: 'seoul', 'busan')
-        data_sample: 데이터 샘플 타입 (예: 'pure', 'smote', 'ctgan10000')
-        
-    Returns:
-        로드된 모델 리스트 (각 fold별 모델)
-        
-    Raises:
-        FileNotFoundError: 모델 파일을 찾을 수 없는 경우
-        ValueError: 모델 이름이 유효하지 않은 경우
-    """
-    if model_name not in TREE_MODELS + DL_MODELS:
-        raise ValueError(
-            f"Unknown model_name: {model_name}. "
-            f"Must be one of {TREE_MODELS + DL_MODELS}"
-        )
-    
-    # Tree-based 모델의 경우 pickle 호환성 함수 등록
-    if model_name in TREE_MODELS:
-        _register_pickle_functions(model_name)
-    
-    model_path = f"../save_model/{model_name}_optima/{model_name}_{data_sample}_{region}.pkl"
-    
-    if not os.path.exists(model_path):
-        raise FileNotFoundError(
-            f"모델 파일을 찾을 수 없습니다: {model_path}"
-        )
-    
-    try:
-        models = load(model_path)
-    except Exception as e:
-        raise FileNotFoundError(f"모델 로드 실패 ({model_path}): {e}")
-    
-    return models
-
-def load_scaler(model_name: str, region: str, data_sample: str) -> List[Any]:
-    """각 fold의 scaler를 로드하는 함수.
-    
-    Args:
-        model_name: 모델 이름
-        region: 지역명
-        data_sample: 데이터 샘플 타입
-        
-    Returns:
-        로드된 scaler 리스트 (각 fold별 scaler)
-        
-    Raises:
-        FileNotFoundError: Scaler 파일을 찾을 수 없는 경우
-    """
-    if data_sample == 'pure':
-        scaler_filename = f'{model_name}_pure_{region}_scaler.pkl'
-    else:
-        scaler_filename = f'{model_name}_{data_sample}_{region}_scaler.pkl'
-    
-    scaler_path = f'../save_model/{model_name}_optima/scaler/{scaler_filename}'
-    
-    if not os.path.exists(scaler_path):
-        raise FileNotFoundError(
-            f"Scaler 파일을 찾을 수 없습니다: {scaler_path}"
-        )
-    
-    try:
-        scalers = load(scaler_path)
-    except Exception as e:
-        raise FileNotFoundError(f"Scaler 로드 실패 ({scaler_path}): {e}")
-    
-    return scalers
-
-# ==========================================
-# Tree-based 모델용 헬퍼 함수
-# ==========================================
-def _prepare_test_data_tree(region: str) -> Tuple[pd.DataFrame, np.ndarray]:
-    """Tree-based 모델용 test 데이터를 준비합니다.
-    
-    Args:
-        region: 지역명
-        
-    Returns:
-        tuple: (X_test, y_test)
-            - X_test: 전처리된 test 데이터 (설명변수만)
-            - y_test: test 데이터의 target 값 (multi_class)
-    """
-    test_data = pd.read_csv(f"../../data/data_for_modeling/{region}_test.csv")
-    test_data = preprocessing(test_data)
-    test_data = test_data.loc[test_data['year'].isin([TEST_YEAR]), :]
-    X_test = test_data.drop(columns=DROP_COLUMNS)
-    y_test = test_data['multi_class'].values
-    return X_test, y_test
-
-def _predict_tree_model(model: Any, X_test: pd.DataFrame) -> np.ndarray:
-    """Tree-based 모델로 예측을 수행합니다.
-    
-    Args:
-        model: 학습된 모델 (XGBoost 또는 LightGBM)
-        X_test: Test 데이터
-        
-    Returns:
-        예측 확률 (n_samples, n_classes)
-    """
-    return model.predict_proba(X_test)
-
-# ==========================================
-# Deep Learning 모델용 헬퍼 함수
-# ==========================================
-def _prepare_test_data_dl(
-    region: str,
-    data_sample: str,
-    fold_idx: int,
-    scaler: Any
-) -> Tuple[torch.Tensor, torch.Tensor, pd.Index, pd.Index, np.ndarray]:
-    """Deep learning 모델용 test 데이터를 준비합니다.
-    
-    Args:
-        region: 지역명
-        data_sample: 데이터 샘플 타입
-        fold_idx: Fold 인덱스 (0, 1, 2)
-        scaler: 해당 fold의 scaler (QuantileTransformer)
-        
-    Returns:
-        tuple: (X_test_num, X_test_cat, categorical_cols, numerical_cols, y_test)
-            - X_test_num: Numerical features tensor
-            - X_test_cat: Categorical features tensor
-            - categorical_cols: 범주형 변수 컬럼명
-            - numerical_cols: 연속형 변수 컬럼명
-            - y_test: test 데이터의 target 값 (multi_class)
-    """
-    # Train 데이터 로드 (Label Encoder 학습용)
-    if data_sample == 'pure':
-        train_path = f"../../data/data_for_modeling/{region}_train.csv"
-        train_data = preprocessing(pd.read_csv(train_path, index_col=0))
-        train_data = train_data.loc[
-            ~train_data['year'].isin([TEST_YEAR - (fold_idx + 1)]), :
-        ]
-    else:
-        train_data = pd.read_csv(
-            f"../../data/data_oversampled/{data_sample}/"
-            f"{data_sample}_{fold_idx+1}_{region}.csv"
-        )
-        train_data = preprocessing(train_data)
-    
-    # Test 데이터 로드
-    test_data = pd.read_csv(f"../../data/data_for_modeling/{region}_test.csv")
-    test_data = preprocessing(test_data)
-    test_data = test_data.loc[test_data['year'].isin([TEST_YEAR]), :]
-    
-    # 컬럼 정렬 (일관성 유지)
-    common_columns = train_data.columns.to_list()
-    train_data = train_data[common_columns]
-    test_data = test_data[common_columns]
-    
-    # 설명변수 분리
-    X_train = train_data.drop(columns=DROP_COLUMNS)
-    X_test = test_data.drop(columns=DROP_COLUMNS)
-    
-    # 범주형 & 연속형 변수 분리
-    categorical_cols = X_train.select_dtypes(
-        include=['object', 'category', 'int64']
-    ).columns
-    numerical_cols = X_train.select_dtypes(include=['float64']).columns
-    
-    # 범주형 변수 Label Encoding (train 데이터 기준으로 학습)
-    label_encoders = {}
-    for col in categorical_cols:
-        le = LabelEncoder()
-        le.fit(X_train[col])
-        label_encoders[col] = le
-    
-    # 변환 적용
-    for col in categorical_cols:
-        X_test[col] = label_encoders[col].transform(X_test[col])
-    
-    # 연속형 변수 Scaler 변환
-    X_test[numerical_cols] = scaler.transform(X_test[numerical_cols])
-    
-    # 텐서 변환
-    X_test_num = torch.tensor(
-        X_test[numerical_cols].values, dtype=torch.float32
-    )
-    X_test_cat = torch.tensor(
-        X_test[categorical_cols].values, dtype=torch.long
-    )
-    
-    # Target 값 추출
-    y_test = test_data['multi_class'].values
-    
-    return X_test_num, X_test_cat, categorical_cols, numerical_cols, y_test
-
-def _predict_dl_model(
-    model: torch.nn.Module,
-    X_test_num: torch.Tensor,
-    X_test_cat: torch.Tensor,
-    model_name: str,
-    device: str,
-    batch_size: int = 1024
-) -> np.ndarray:
-    """Deep learning 모델로 예측을 수행합니다.
-    
-    Args:
-        model: PyTorch 모델
-        X_test_num: Numerical features tensor
-        X_test_cat: Categorical features tensor
-        model_name: 모델 이름 ('ft_transformer', 'resnet_like', 'deepgbm')
-        device: 연산 장치 ('cpu' 또는 'cuda')
-        batch_size: 배치 크기 (메모리 효율을 위해, 기본값: 1024)
-    
-    Returns:
-        예측 확률 (n_samples, n_classes)
-        
-    Raises:
-        ValueError: 알 수 없는 모델 이름인 경우
-    """
-    model.eval()
-    model.to(device)
-    
-    n_samples = X_test_num.shape[0]
-    all_probs = []
-    
-    with torch.no_grad():
-        for i in range(0, n_samples, batch_size):
-            end_idx = min(i + batch_size, n_samples)
-            batch_num = X_test_num[i:end_idx].to(device)
-            batch_cat = X_test_cat[i:end_idx].to(device)
-            
-            # 모델별 입력 형식 처리
-            # 모든 딥러닝 모델이 (x_num, x_cat) 두 개의 텐서를 받음
-            # resnet_like는 내부에서 torch.cat으로 결합함
-            if model_name in DL_MODELS:
-                output = model(batch_num, batch_cat)
-            else:
-                raise ValueError(
-                    f"Unknown model_name: {model_name}. "
-                    f"Must be one of {DL_MODELS}"
-                )
-            
-            # Softmax 적용하여 확률 변환
-            probs = F.softmax(output, dim=1)
-            all_probs.append(probs.cpu().numpy())
-    
-    return np.vstack(all_probs)
-
-# ==========================================
-# 메인 함수
-# ==========================================
-def predict_test_proba(
-    model_name: str,
-    region: str,
-    data_sample: str,
-    device: str = 'cpu',
-    n_folds: int = 3
-) -> Tuple[np.ndarray, np.ndarray]:
-    """Test 데이터에 대한 예측 확률과 target 값을 반환하는 함수.
-    
-    Args:
-        model_name: 모델 이름 ('xgb', 'lgb', 'ft_transformer', 'resnet_like', 'deepgbm')
-        region: 지역명 (예: 'seoul', 'busan')
-        data_sample: 데이터 샘플 타입 (예: 'pure', 'smote', 'ctgan10000')
-        device: 연산 장치 ('cpu' 또는 'cuda', 기본값: 'cpu')
-        n_folds: Fold 개수 (기본값: 3)
-    
-    Returns:
-        tuple: (probs, y_test)
-            - probs: shape (n_folds, n_samples, n_classes) - 각 fold별 예측 확률
-            - y_test: shape (n_samples,) - test 데이터의 target 값 (multi_class)
-        
-    Raises:
-        ValueError: 모델 이름이 유효하지 않거나 fold 개수가 일치하지 않는 경우
-        FileNotFoundError: 모델 또는 scaler 파일을 찾을 수 없는 경우
-    """
-    # 모델 타입 확인
-    if model_name not in TREE_MODELS + DL_MODELS:
-        raise ValueError(
-            f"Unknown model_name: {model_name}. "
-            f"Must be one of {TREE_MODELS + DL_MODELS}"
-        )
-    
-    # 모델 로드
-    models = load_model(model_name, region, data_sample)
-    
-    if len(models) != n_folds:
-        raise ValueError(
-            f"로드된 모델 개수({len(models)})가 "
-            f"n_folds({n_folds})와 일치하지 않습니다."
-        )
-    
-    # Tree-based 모델 처리
-    if model_name in TREE_MODELS:
-        X_test, y_test = _prepare_test_data_tree(region)
-        all_probs = []
-        
-        for fold_idx in range(n_folds):
-            prob = _predict_tree_model(models[fold_idx], X_test)
-            all_probs.append(prob)
-        
-        return np.array(all_probs), y_test
-    
-    # Deep learning 모델 처리
-    else:
-        # Scaler 로드
-        scalers = load_scaler(model_name, region, data_sample)
-        
-        if len(scalers) != n_folds:
-            raise ValueError(
-                f"로드된 scaler 개수({len(scalers)})가 "
-                f"n_folds({n_folds})와 일치하지 않습니다."
-            )
-        
-        all_probs = []
-        y_test = None
-        
-        for fold_idx in range(n_folds):
-            # Test 데이터 준비
-            X_test_num, X_test_cat, _, _, y_test = _prepare_test_data_dl(
-                region, data_sample, fold_idx, scalers[fold_idx]
-            )
-            
-            # 예측
-            prob = _predict_dl_model(
-                models[fold_idx], X_test_num, X_test_cat, model_name, device
-            )
-            all_probs.append(prob)
-        
-        return np.array(all_probs), y_test