""" Data processing module for the Business Intelligence Dashboard. Handles data loading, cleaning, validation, and filtering operations. """ import pandas as pd import numpy as np from typing import Dict, List, Tuple, Optional, Any from utils import get_column_types, detect_datetime_columns, validate_dataframe def load_data(file_path: str) -> Tuple[Optional[pd.DataFrame], str]: """ Load data from CSV or Excel file. Args: file_path: Path to the data file Returns: Tuple of (DataFrame, error_message). DataFrame is None if loading failed. """ try: if file_path.endswith('.csv'): df = pd.read_csv(file_path) elif file_path.endswith(('.xlsx', '.xls')): df = pd.read_excel(file_path) else: return None, "Unsupported file format. Please upload CSV or Excel files." # Validate the loaded DataFrame is_valid, error_msg = validate_dataframe(df) if not is_valid: return None, f"Invalid data: {error_msg}" # Clean numeric columns with commas (e.g., "1,234" -> 1234) df = clean_numeric_columns(df) # Auto-detect and convert datetime columns datetime_cols = detect_datetime_columns(df) for col in datetime_cols: try: df[col] = pd.to_datetime(df[col], errors='coerce') except Exception: pass return df, "" except Exception as e: return None, f"Error loading file: {str(e)}" def clean_numeric_columns(df: pd.DataFrame) -> pd.DataFrame: """ Clean columns that contain numeric values with commas or other formatting. Converts strings like '1,234', '50,000' to proper numeric types. Args: df: Input DataFrame Returns: DataFrame with cleaned numeric columns """ df_cleaned = df.copy() for col in df_cleaned.columns: # Skip if already numeric if pd.api.types.is_numeric_dtype(df_cleaned[col]): continue # Check if column contains string values that look like numbers if df_cleaned[col].dtype == 'object': # Get a sample of non-null values sample = df_cleaned[col].dropna().head(100) if len(sample) > 0: # Check if values contain commas and look numeric sample_str = sample.astype(str) # Count how many values look like comma-separated numbers numeric_looking = sample_str.str.match(r'^-?[\d,]+\.?\d*$').sum() # If more than 50% look like numbers with commas, try to convert if numeric_looking > len(sample) * 0.5: try: # Remove commas and convert to numeric df_cleaned[col] = df_cleaned[col].astype(str).str.replace(',', '').replace('--', '0') df_cleaned[col] = pd.to_numeric(df_cleaned[col], errors='coerce') except Exception: # If conversion fails, keep original pass return df_cleaned def get_data_summary(df: pd.DataFrame) -> Dict[str, Any]: """ Generate comprehensive summary statistics for the DataFrame. Args: df: Input DataFrame Returns: Dictionary containing summary statistics """ col_types = get_column_types(df) summary = { 'shape': df.shape, 'columns': df.columns.tolist(), 'dtypes': df.dtypes.to_dict(), 'column_types': col_types, 'missing_values': df.isnull().sum().to_dict(), 'duplicate_rows': df.duplicated().sum(), 'memory_usage': df.memory_usage(deep=True).sum() / 1024 / 1024, # MB } # Numerical statistics if col_types['numerical']: numerical_stats = df[col_types['numerical']].describe().to_dict() summary['numerical_stats'] = numerical_stats # Categorical statistics if col_types['categorical']: categorical_stats = {} for col in col_types['categorical']: categorical_stats[col] = { 'unique_count': df[col].nunique(), 'top_values': df[col].value_counts().head(10).to_dict(), 'mode': df[col].mode().iloc[0] if len(df[col].mode()) > 0 else None } summary['categorical_stats'] = categorical_stats return summary def get_correlation_matrix(df: pd.DataFrame) -> Optional[pd.DataFrame]: """ Calculate correlation matrix for numerical columns. Args: df: Input DataFrame Returns: Correlation matrix DataFrame or None if no numerical columns """ col_types = get_column_types(df) if not col_types['numerical']: return None return df[col_types['numerical']].corr() def apply_filters(df: pd.DataFrame, filters: Dict[str, Any]) -> pd.DataFrame: """ Apply filters to the DataFrame based on user selections. Args: df: Input DataFrame filters: Dictionary of filter specifications Format: { 'column_name': { 'type': 'numerical' | 'categorical' | 'datetime', 'min': value, # for numerical 'max': value, # for numerical 'values': [list], # for categorical 'start_date': value, # for datetime 'end_date': value # for datetime } } Returns: Filtered DataFrame """ filtered_df = df.copy() for column, filter_spec in filters.items(): if column not in filtered_df.columns: continue filter_type = filter_spec.get('type') if filter_type == 'numerical': min_val = filter_spec.get('min') max_val = filter_spec.get('max') if min_val is not None: filtered_df = filtered_df[filtered_df[column] >= min_val] if max_val is not None: filtered_df = filtered_df[filtered_df[column] <= max_val] elif filter_type == 'categorical': values = filter_spec.get('values', []) if values: filtered_df = filtered_df[filtered_df[column].isin(values)] elif filter_type == 'datetime': start_date = filter_spec.get('start_date') end_date = filter_spec.get('end_date') if start_date is not None: filtered_df = filtered_df[filtered_df[column] >= pd.to_datetime(start_date)] if end_date is not None: filtered_df = filtered_df[filtered_df[column] <= pd.to_datetime(end_date)] return filtered_df def clean_data(df: pd.DataFrame, drop_duplicates: bool = False, fill_numerical: Optional[str] = None, fill_categorical: Optional[str] = None) -> pd.DataFrame: """ Clean the DataFrame by handling missing values and duplicates. Args: df: Input DataFrame drop_duplicates: Whether to drop duplicate rows fill_numerical: Strategy for filling numerical NaNs ('mean', 'median', 'zero', or None) fill_categorical: Strategy for filling categorical NaNs ('mode', 'unknown', or None) Returns: Cleaned DataFrame """ cleaned_df = df.copy() col_types = get_column_types(cleaned_df) # Handle duplicates if drop_duplicates: cleaned_df = cleaned_df.drop_duplicates() # Handle missing values in numerical columns if fill_numerical and col_types['numerical']: for col in col_types['numerical']: if fill_numerical == 'mean': cleaned_df[col].fillna(cleaned_df[col].mean(), inplace=True) elif fill_numerical == 'median': cleaned_df[col].fillna(cleaned_df[col].median(), inplace=True) elif fill_numerical == 'zero': cleaned_df[col].fillna(0, inplace=True) # Handle missing values in categorical columns if fill_categorical and col_types['categorical']: for col in col_types['categorical']: if fill_categorical == 'mode': mode_val = cleaned_df[col].mode() if len(mode_val) > 0: cleaned_df[col].fillna(mode_val.iloc[0], inplace=True) elif fill_categorical == 'unknown': cleaned_df[col].fillna('Unknown', inplace=True) return cleaned_df def aggregate_data(df: pd.DataFrame, group_by: str, value_column: str, agg_method: str = 'sum') -> pd.DataFrame: """ Aggregate data by grouping and applying an aggregation method. Args: df: Input DataFrame group_by: Column to group by value_column: Column to aggregate agg_method: Aggregation method ('sum', 'mean', 'count', 'median', 'min', 'max') Returns: Aggregated DataFrame """ if group_by not in df.columns or value_column not in df.columns: return pd.DataFrame() agg_methods = { 'sum': 'sum', 'mean': 'mean', 'count': 'count', 'median': 'median', 'min': 'min', 'max': 'max' } method = agg_methods.get(agg_method, 'sum') try: aggregated = df.groupby(group_by)[value_column].agg(method).reset_index() aggregated.columns = [group_by, f'{value_column}_{method}'] return aggregated except Exception: return pd.DataFrame() def get_data_preview(df: pd.DataFrame, n_rows: int = 10, head: bool = True) -> pd.DataFrame: """ Get a preview of the DataFrame. Args: df: Input DataFrame n_rows: Number of rows to return head: If True, return first n rows; if False, return last n rows Returns: Preview DataFrame """ if head: return df.head(n_rows) else: return df.tail(n_rows)