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

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

import pandas as pd
import numpy as np
from typing import Optional
from concurrent.futures import ThreadPoolExecutor

def memory_optimizer(
    df: pd.DataFrame,
    enable_category: bool = True,
    enable_downcast: bool = True,
    safe_mode: bool = True,
    verbose: bool = True,
    n_workers: int = 4
) -> pd.DataFrame:
    """
    고속화된 메모리 최적화 함수
    
    Parameters:
        df: 입력 DataFrame
        enable_category: 문자열 범주형 변환 활성화 (기본 True)
        enable_downcast: 숫자형 다운캐스트 활성화 (기본 True)
        safe_mode: 모든 변환 시 원본 백업 복사 (기본 True)
        verbose: 진행 상황 출력 (기본 True)
        n_workers: 병렬 처리에 사용할 스레드 수 (기본 4)
        
    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:
        # 숫자형 다운캐스트 최적화 함수
        def _downcast_numeric(col_data):
            try:
                if pd.api.types.is_integer_dtype(col_data):
                    return pd.to_numeric(col_data, downcast='integer')
                elif pd.api.types.is_float_dtype(col_data):
                    return pd.to_numeric(col_data, downcast='float')
                return col_data
            except:
                return col_data

        # 범주형 변환 최적화 함수
        def _convert_to_category(col_data):
            try:
                unique_ratio = col_data.nunique() / len(col_data)
                nan_ratio = col_data.isna().mean()
                if unique_ratio < 0.5 and nan_ratio < 0.2:
                    return col_data.astype('category')
                return col_data
            except:
                return col_data

        # 2. 병렬 처리 적용
        with ThreadPoolExecutor(max_workers=n_workers) as executor:
            if enable_downcast:
                numeric_cols = df.select_dtypes(include=['number']).columns
                if len(numeric_cols) > 0:
                    results = list(executor.map(_downcast_numeric, [df[col] for col in numeric_cols]))
                    for col, result in zip(numeric_cols, results):
                        df[col] = result

            if enable_category:
                object_cols = df.select_dtypes(include=['object', 'string']).columns
                if len(object_cols) > 0:
                    results = list(executor.map(_convert_to_category, [df[col] for col in object_cols]))
                    for col, result in zip(object_cols, results):
                        df[col] = result

        # 3. 최종 검증 (벡터화된 검증)
        if safe_mode and original_df is not None:
            mismatch = False
            for col in df.columns:
                if not df[col].equals(original_df[col].astype(df[col].dtype)):
                    mismatch = True
                    if verbose:
                        print(f"⚠️ 검증 실패 컬럼: {col}")
                    if safe_mode:
                        df[col] = original_df[col]
            if mismatch:
                raise ValueError("일부 컬럼에서 검증 실패")

        # 4. 결과 리포트
        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}% 감소)")
            if safe_mode and original_df is not None:
                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