Spaces:
Sleeping
Sleeping
File size: 5,996 Bytes
bd41331 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | import pandas as pd
import numpy as np
import joblib
import logging
from typing import Dict, Union, List, Tuple, Optional
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from src.config.parameters import DATA_PARAMS, ARTIFACT_PATHS
logger = logging.getLogger(__name__)
class DataPreprocessor:
"""Data preprocessing class for fraud detection"""
def __init__(self):
self.scaler = StandardScaler()
self.legit_amount_mean = None
self.amount_bin_edges = None
self.top_corr_features = None
self.feature_columns = None
def load_data(self, file_path: str) -> pd.DataFrame:
"""Load dataset from file path"""
try:
logger.info(f"Loading data from {file_path}")
df = pd.read_csv(file_path)
logger.info(f"Data loaded successfully. Shape: {df.shape}")
return df
except Exception as e:
logger.error(f"Error loading data: {str(e)}")
raise
def basic_preprocessing(self, df: pd.DataFrame) -> pd.DataFrame:
"""Perform basic preprocessing steps"""
logger.info("Starting basic preprocessing")
missing_values = df.isnull().sum().sum()
logger.info(f"Total missing values: {missing_values}")
class_distribution = df['Class'].value_counts()
logger.info(f"Class distribution:\n{class_distribution}")
return df
def split_data(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
"""Split data into train and test sets"""
logger.info("Splitting data into train and test sets")
X = df.drop('Class', axis=1)
y = df['Class']
X_train, X_test, y_train, y_test = train_test_split(
X, y,
test_size=DATA_PARAMS['test_size'],
stratify=y if DATA_PARAMS['stratify'] else None,
random_state=DATA_PARAMS['random_state']
)
logger.info(f"Train set: {X_train.shape[0]} samples, Fraud cases: {y_train.sum()}")
logger.info(f"Test set: {X_test.shape[0]} samples, Fraud cases: {y_test.sum()}")
return X_train, X_test, y_train, y_test
def save_preprocessing_artifacts(self, df: pd.DataFrame, X_train: pd.DataFrame):
"""Save preprocessing artifacts for later use"""
logger.info("Saving preprocessing artifacts")
joblib.dump(self.scaler, ARTIFACT_PATHS['scaler'])
joblib.dump(self.legit_amount_mean, ARTIFACT_PATHS['legit_amount_mean'])
joblib.dump(self.amount_bin_edges, ARTIFACT_PATHS['amount_bin_edges'])
joblib.dump(self.top_corr_features, ARTIFACT_PATHS['top_corr_features'])
joblib.dump(X_train.columns.tolist(), ARTIFACT_PATHS['feature_columns'])
logger.info("Preprocessing artifacts saved successfully")
def load_preprocessing_artifacts(self):
"""Load preprocessing artifacts"""
try:
self.scaler = joblib.load(ARTIFACT_PATHS['scaler'])
self.legit_amount_mean = joblib.load(ARTIFACT_PATHS['legit_amount_mean'])
self.amount_bin_edges = joblib.load(ARTIFACT_PATHS['amount_bin_edges'])
self.top_corr_features = joblib.load(ARTIFACT_PATHS['top_corr_features'])
self.feature_columns = joblib.load(ARTIFACT_PATHS['feature_columns'])
logger.info("Preprocessing artifacts loaded successfully")
except Exception as e:
logger.error(f"Error loading preprocessing artifacts: {str(e)}")
raise
def preprocess_new_data(self, raw_input: Union[Dict, pd.DataFrame]) -> pd.DataFrame:
"""Preprocess new data for prediction"""
if isinstance(raw_input, dict):
normalized = {
k: (v if isinstance(v, (list, np.ndarray, pd.Series)) else [v])
for k, v in raw_input.items()
}
df_new = pd.DataFrame(normalized)
else:
df_new = raw_input.copy()
required_columns = ['Time', 'Amount'] + [f'V{i}' for i in range(1, 29)]
for col in required_columns:
if col not in df_new.columns:
raise ValueError(f"Missing required feature: {col}")
from src.utils.feature_engineering import FeatureEngineer
fe = FeatureEngineer()
fe.scaler = self.scaler
fe.legit_amount_mean = self.legit_amount_mean
fe.amount_bin_edges = self.amount_bin_edges
fe.top_corr_features = self.top_corr_features
df_processed = fe.engineer_features(df_new, is_training=False)
for col in self.feature_columns:
if col not in df_processed.columns:
df_processed[col] = 0
df_processed = df_processed.loc[:, ~df_processed.columns.duplicated()]
df_processed = df_processed[self.feature_columns]
if list(df_processed.columns) != list(self.feature_columns):
raise ValueError(
f"Feature mismatch!\nExpected: {self.feature_columns}\nGot: {df_processed.columns.tolist()}"
)
return df_processed
def validate_input_data(data: Union[Dict, pd.DataFrame]) -> bool:
"""Validate input data format and required fields"""
required_fields = ['Time', 'Amount'] + [f'V{i}' for i in range(1, 29)]
if isinstance(data, dict):
missing_fields = [field for field in required_fields if field not in data]
else:
missing_fields = [field for field in required_fields if field not in data.columns]
if missing_fields:
logger.error(f"Missing required fields: {missing_fields}")
return False
return True |