| """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": |
| |
| 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() |
| |
| |
| 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 |
|
|
