preprocessor.py

import pandas as pd
import numpy as np
import joblib
import logging
from pathlib import Path

logger = logging.getLogger(__name__)


class FraudPreprocessor:
    def __init__(self):
        self.scaler = None
        self.feature_columns = []
        self.is_fitted = False

    def transform(self, df: pd.DataFrame) -> pd.DataFrame:
        df = self._engineer_features(df.copy())
        df['Amount_scaled'] = self.scaler.transform(df[['Amount']])
        df.drop(['Time', 'Amount'], axis=1, inplace=True)
        return df

    def _engineer_features(self, df: pd.DataFrame) -> pd.DataFrame:
        df['Hour'] = (df['Time'] // 3600) % 24
        df['Amount_log'] = np.log1p(df['Amount'])
        df['Is_round_amount'] = (df['Amount'] % 1 == 0).astype(int)
        return df

    @staticmethod
    def load_from_components(models_dir: Path) -> 'FraudPreprocessor':
        """Load preprocessor from individual component files."""
        p = FraudPreprocessor()
        p.scaler = joblib.load(models_dir / 'scaler.pkl')
        p.feature_columns = joblib.load(models_dir / 'feature_columns.pkl')
        p.is_fitted = True
        logger.info("Preprocessor loaded from components.")
        return p