데이터프레임의 메모리 사용량을 최적화하는 함수

import pandas as pd
import numpy as np
from typing import Optional

def memory_optimizer(
    df: pd.DataFrame,
    enable_category: bool = True,
    enable_downcast: bool = True,
    safe_mode: bool = True,
    verbose: bool = True
) -> pd.DataFrame:
    """
    데이터 정확성을 보장하는 메모리 최적화 함수
    
    Parameters:
        df: 입력 DataFrame
        enable_category: 문자열 범주형 변환 활성화 (기본 True)
        enable_downcast: 숫자형 다운캐스트 활성화 (기본 True)
        safe_mode: 모든 변환 시 원본 백업 복사 (기본 True)
        verbose: 진행 상황 출력 (기본 True)
        
    Returns:
        최적화된 DataFrame (실패 시 원본 반환)
    """
    
    # 1. 원본 백업 (safe_mode 활성화 시)
    original_df = df.copy(deep=True) if safe_mode else None
    initial_memory = df.memory_usage(deep=True).sum()
    
    try:
        # 2. 숫자형 다운캐스트 (정확성 검증 포함)
        if enable_downcast:
            for col in df.select_dtypes(include=['number']).columns:
                try:
                    # 현재 타입 및 범위 확인
                    col_min = df[col].min()
                    col_max = df[col].max()
                    
                    # 정수형 다운캐스트
                    if pd.api.types.is_integer_dtype(df[col]):
                        if col_min > np.iinfo(np.int8).min and col_max < np.iinfo(np.int8).max:
                            df[col] = pd.to_numeric(df[col], downcast='integer')
                        elif col_min > np.iinfo(np.int16).min and col_max < np.iinfo(np.int16).max:
                            df[col] = pd.to_numeric(df[col], downcast='integer')
                        # ... 다른 범위 체크 생략
                    
                    # 실수형 다운캐스트
                    elif pd.api.types.is_float_dtype(df[col]):
                        df[col] = pd.to_numeric(df[col], downcast='float')
                        
                    # 변환 후 검증
                    if not np.allclose(df[col], original_df[col], equal_nan=True):
                        raise ValueError(f"{col} 값 변경 발생")
                        
                except Exception as e:
                    if safe_mode:
                        df[col] = original_df[col]
                    if verbose:
                        print(f"⚠️ 숫자형 변환 실패 ({col}): {str(e)}")

        # 3. 범주형 변환 (NaN 안전 처리)
        if enable_category:
            for col in df.select_dtypes(include=['object', 'string']).columns:
                try:
                    # 변환 조건: 고유값 비율 < 50% + NaN 비율 < 20%
                    unique_ratio = df[col].nunique() / len(df[col])
                    nan_ratio = df[col].isna().mean()
                    
                    if unique_ratio < 0.5 and nan_ratio < 0.2:
                        # 백업 생성 (safe_mode 시)
                        col_backup = df[col].copy() if safe_mode else None
                        
                        # 범주형 변환 시도
                        df[col] = df[col].astype('category')
                        
                        # 변환 검증
                        if not df[col].equals(col_backup.astype('category')) and safe_mode:
                            raise ValueError("범주 변환 불일치")
                            
                except Exception as e:
                    if safe_mode and col_backup is not None:
                        df[col] = col_backup
                    if verbose:
                        print(f"⚠️ 범주형 변환 실패 ({col}): {str(e)}")

        # 4. 최종 검증
        if safe_mode:
            for col in df.columns:
                if not df[col].equals(original_df[col].astype(df[col].dtype)):
                    raise ValueError(f"최종 검증 실패: {col}")

        # 5. 결과 리포트
        if verbose:
            optimized_memory = df.memory_usage(deep=True).sum()
            reduction = (initial_memory - optimized_memory) / initial_memory * 100
            print(f"✅ 메모리 사용량: {initial_memory:,} → {optimized_memory:,} ({reduction:.1f}% 감소)")
            print("🔍 타입 변화:")
            print(pd.concat([
                original_df.dtypes.rename('before'),
                df.dtypes.rename('after')
            ], axis=1))
            
        return df

    except Exception as e:
        if verbose:
            print(f"❌ 최적화 실패: {str(e)}")
        return original_df if safe_mode else df