File size: 3,655 Bytes
0cfd2fc | 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 | """Data validation and handling module."""
import pandas as pd
import numpy as np
from typing import Tuple, List, Dict, Optional
from config.settings import MISSING_VALUE_THRESHOLD, NUMERIC_FEATURES_DTYPE, CATEGORICAL_FEATURES_DTYPE
class DataHandler:
"""Handles data loading, validation, and basic exploration."""
def __init__(self, df: pd.DataFrame):
"""Initialize with a DataFrame."""
self.df = df
self.original_df = df.copy()
self.numeric_cols = []
self.categorical_cols = []
self._identify_column_types()
def _identify_column_types(self) -> None:
"""Identify numeric and categorical columns."""
self.numeric_cols = self.df.select_dtypes(include=NUMERIC_FEATURES_DTYPE).columns.tolist()
self.categorical_cols = self.df.select_dtypes(include=CATEGORICAL_FEATURES_DTYPE).columns.tolist()
def get_data_summary(self) -> Dict:
"""Return summary statistics of the dataset."""
return {
"shape": self.df.shape,
"dtypes": self.df.dtypes.to_dict(),
"missing_values": self.df.isnull().sum().to_dict(),
"missing_percentage": (self.df.isnull().sum() / len(self.df) * 100).round(2).to_dict(),
"numeric_columns": self.numeric_cols,
"categorical_columns": self.categorical_cols,
}
def validate_data(self) -> Tuple[bool, str]:
"""Validate dataset integrity."""
if self.df.empty:
return False, "Dataset is empty."
if len(self.df) < 10:
return False, "Dataset must have at least 10 rows."
if len(self.df.columns) < 2:
return False, "Dataset must have at least 2 columns."
return True, "Data validation passed."
def handle_missing_values(self, strategy: str = "drop_columns") -> None:
"""Handle missing values in the dataset.
Args:
strategy: "drop_columns" (default) or "drop_rows"
"""
if strategy == "drop_columns":
# Drop columns with >50% missing values
missing_pct = (self.df.isnull().sum() / len(self.df)) * 100
cols_to_drop = missing_pct[missing_pct > MISSING_VALUE_THRESHOLD * 100].index.tolist()
self.df.drop(columns=cols_to_drop, inplace=True)
self._identify_column_types()
# Drop rows with any remaining missing values
self.df.dropna(inplace=True)
elif strategy == "drop_rows":
self.df.dropna(inplace=True)
def get_columns(self) -> List[str]:
"""Return list of all columns."""
return self.df.columns.tolist()
def get_column_info(self, column: str) -> Dict:
"""Get detailed info about a specific column."""
col_data = self.df[column]
info = {
"name": column,
"dtype": str(col_data.dtype),
"missing_count": col_data.isnull().sum(),
"missing_percentage": round(col_data.isnull().sum() / len(self.df) * 100, 2),
}
if column in self.numeric_cols:
info.update({
"mean": col_data.mean(),
"median": col_data.median(),
"std": col_data.std(),
"min": col_data.min(),
"max": col_data.max(),
"unique_count": col_data.nunique(),
})
else:
info.update({
"unique_count": col_data.nunique(),
"top_values": col_data.value_counts().head(5).to_dict(),
})
return info
|