app.py

import gradio as gr
import pandas as pd
import numpy as np
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from sklearn.cluster import DBSCAN
from scipy import stats
from typing import Dict, List, Tuple, Any, Optional
import warnings
warnings.filterwarnings('ignore')

class OutlierDetective:
    def __init__(self):
        self.df = None
        self.outlier_results = {}
        self.numeric_columns = []
        
    def load_data(self, file_path: str) -> pd.DataFrame:
        """Load data from various file formats"""
        try:
            if file_path.endswith('.csv'):
                df = pd.read_csv(file_path, encoding='utf-8')
            elif file_path.endswith(('.xlsx', '.xls')):
                df = pd.read_excel(file_path)
            elif file_path.endswith('.json'):
                df = pd.read_json(file_path)
            elif file_path.endswith('.parquet'):
                df = pd.read_parquet(file_path)
            else:
                df = pd.read_csv(file_path)
            
            self.df = df
            # Identify numeric columns
            self.numeric_columns = df.select_dtypes(include=[np.number]).columns.tolist()
            return df
        except Exception as e:
            raise Exception(f"Error loading file: {str(e)}")
    
    def detect_iqr_outliers(self, column: str) -> Dict[str, Any]:
        """Detect outliers using Interquartile Range (IQR) method"""
        if column not in self.numeric_columns:
            return {}
        
        series = self.df[column].dropna()
        Q1 = series.quantile(0.25)
        Q3 = series.quantile(0.75)
        IQR = Q3 - Q1
        
        lower_bound = Q1 - 1.5 * IQR
        upper_bound = Q3 + 1.5 * IQR
        
        outlier_mask = (series < lower_bound) | (series > upper_bound)
        outlier_indices = series[outlier_mask].index.tolist()
        outlier_values = series[outlier_mask].tolist()
        
        return {
            'method': 'IQR',
            'lower_bound': lower_bound,
            'upper_bound': upper_bound,
            'outlier_indices': outlier_indices,
            'outlier_values': outlier_values,
            'outlier_count': len(outlier_indices),
            'outlier_percentage': (len(outlier_indices) / len(series)) * 100,
            'explanation': f"Values below {lower_bound:.2f} or above {upper_bound:.2f} are considered outliers"
        }
    
    def detect_zscore_outliers(self, column: str, threshold: float = 3) -> Dict[str, Any]:
        """Detect outliers using Z-score method"""
        if column not in self.numeric_columns:
            return {}
        
        series = self.df[column].dropna()
        z_scores = np.abs(stats.zscore(series))
        
        outlier_mask = z_scores > threshold
        outlier_indices = series[outlier_mask].index.tolist()
        outlier_values = series[outlier_mask].tolist()
        outlier_zscores = z_scores[outlier_mask].tolist()
        
        return {
            'method': 'Z-Score',
            'threshold': threshold,
            'outlier_indices': outlier_indices,
            'outlier_values': outlier_values,
            'outlier_zscores': outlier_zscores,
            'outlier_count': len(outlier_indices),
            'outlier_percentage': (len(outlier_indices) / len(series)) * 100,
            'explanation': f"Values with |z-score| > {threshold} are considered outliers"
        }
    
    def detect_modified_zscore_outliers(self, column: str, threshold: float = 3.5) -> Dict[str, Any]:
        """Detect outliers using Modified Z-score (MAD) method"""
        if column not in self.numeric_columns:
            return {}
        
        series = self.df[column].dropna()
        median = series.median()
        mad = stats.median_abs_deviation(series)
        
        if mad == 0:
            return {
                'method': 'Modified Z-Score',
                'outlier_count': 0,
                'outlier_percentage': 0,
                'explanation': "MAD is zero - no outliers detected using this method"
            }
        
        modified_z_scores = 0.6745 * (series - median) / mad
        
        outlier_mask = np.abs(modified_z_scores) > threshold
        outlier_indices = series[outlier_mask].index.tolist()
        outlier_values = series[outlier_mask].tolist()
        outlier_scores = modified_z_scores[outlier_mask].tolist()
        
        return {
            'method': 'Modified Z-Score',
            'threshold': threshold,
            'median': median,
            'mad': mad,
            'outlier_indices': outlier_indices,
            'outlier_values': outlier_values,
            'outlier_scores': outlier_scores,
            'outlier_count': len(outlier_indices),
            'outlier_percentage': (len(outlier_indices) / len(series)) * 100,
            'explanation': f"Values with |modified z-score| > {threshold} are considered outliers (robust to extreme values)"
        }
    
    def detect_isolation_forest_outliers(self, columns: List[str], contamination: float = 0.1) -> Dict[str, Any]:
        """Detect multivariate outliers using Isolation Forest"""
        if not columns or len(columns) < 1:
            return {}
        
        # Filter to only numeric columns that exist
        valid_columns = [col for col in columns if col in self.numeric_columns]
        if not valid_columns:
            return {}
        
        # Prepare data
        data = self.df[valid_columns].dropna()
        if len(data) < 10:  # Need minimum data points
            return {}
        
        # Standardize the data
        scaler = StandardScaler()
        scaled_data = scaler.fit_transform(data)
        
        # Fit Isolation Forest
        iso_forest = IsolationForest(contamination=contamination, random_state=42)
        outlier_labels = iso_forest.fit_predict(scaled_data)
        
        # Get outlier indices and scores
        outlier_mask = outlier_labels == -1
        outlier_indices = data[outlier_mask].index.tolist()
        outlier_scores = iso_forest.score_samples(scaled_data)
        outlier_score_values = outlier_scores[outlier_mask].tolist()
        
        return {
            'method': 'Isolation Forest',
            'contamination': contamination,
            'columns_used': valid_columns,
            'outlier_indices': outlier_indices,
            'outlier_scores': outlier_score_values,
            'outlier_count': len(outlier_indices),
            'outlier_percentage': (len(outlier_indices) / len(data)) * 100,
            'explanation': f"Multivariate outlier detection using {len(valid_columns)} features with {contamination*100}% expected contamination"
        }
    
    def detect_dbscan_outliers(self, columns: List[str], eps: float = 0.5, min_samples: int = 5) -> Dict[str, Any]:
        """Detect outliers using DBSCAN clustering"""
        if not columns or len(columns) < 1:
            return {}
        
        # Filter to only numeric columns that exist
        valid_columns = [col for col in columns if col in self.numeric_columns]
        if not valid_columns:
            return {}
        
        # Prepare data
        data = self.df[valid_columns].dropna()
        if len(data) < min_samples * 2:  # Need minimum data points
            return {}
        
        # Standardize the data
        scaler = StandardScaler()
        scaled_data = scaler.fit_transform(data)
        
        # Apply DBSCAN
        dbscan = DBSCAN(eps=eps, min_samples=min_samples)
        cluster_labels = dbscan.fit_predict(scaled_data)
        
        # Points labeled as -1 are outliers
        outlier_mask = cluster_labels == -1
        outlier_indices = data[outlier_mask].index.tolist()
        
        # Count clusters
        n_clusters = len(set(cluster_labels)) - (1 if -1 in cluster_labels else 0)
        
        return {
            'method': 'DBSCAN',
            'eps': eps,
            'min_samples': min_samples,
            'columns_used': valid_columns,
            'n_clusters': n_clusters,
            'outlier_indices': outlier_indices,
            'outlier_count': len(outlier_indices),
            'outlier_percentage': (len(outlier_indices) / len(data)) * 100,
            'explanation': f"Density-based outlier detection found {n_clusters} clusters using {len(valid_columns)} features"
        }
    
    def analyze_outliers(self, selected_columns: List[str] = None, methods: List[str] = None) -> Dict[str, Any]:
        """Comprehensive outlier analysis"""
        if self.df is None:
            return {}
        
        if selected_columns is None:
            selected_columns = self.numeric_columns
        else:
            # Filter to only numeric columns
            selected_columns = [col for col in selected_columns if col in self.numeric_columns]
        
        if not selected_columns:
            return {}
        
        if methods is None:
            methods = ['IQR', 'Z-Score', 'Modified Z-Score', 'Isolation Forest']
        
        results = {}
        
        # Single-column methods
        for column in selected_columns:
            results[column] = {}
            
            if 'IQR' in methods:
                results[column]['IQR'] = self.detect_iqr_outliers(column)
            
            if 'Z-Score' in methods:
                results[column]['Z-Score'] = self.detect_zscore_outliers(column)
            
            if 'Modified Z-Score' in methods:
                results[column]['Modified Z-Score'] = self.detect_modified_zscore_outliers(column)
        
        # Multi-column methods
        if len(selected_columns) > 1:
            if 'Isolation Forest' in methods:
                results['Multivariate'] = {}
                results['Multivariate']['Isolation Forest'] = self.detect_isolation_forest_outliers(selected_columns)
            
            if 'DBSCAN' in methods:
                if 'Multivariate' not in results:
                    results['Multivariate'] = {}
                results['Multivariate']['DBSCAN'] = self.detect_dbscan_outliers(selected_columns)
        
        self.outlier_results = results
        return results
    
    def generate_outlier_report(self) -> str:
        """Generate comprehensive outlier analysis report"""
        if not self.outlier_results:
            return "No outlier analysis results available. Please run the analysis first."
        
        report = "#Outlier Detection Report\n\n"
        
        total_outliers_by_method = {}
        all_outlier_indices = set()
        
        for column, methods in self.outlier_results.items():
            if column == 'Multivariate':
                continue
            
            for method, result in methods.items():
                if isinstance(result, dict) and 'outlier_count' in result:
                    total_outliers_by_method.setdefault(method, 0)
                    total_outliers_by_method[method] += result['outlier_count']
                    if 'outlier_indices' in result:
                        all_outlier_indices.update(result['outlier_indices'])
        
        if 'Multivariate' in self.outlier_results:
            for method, result in self.outlier_results['Multivariate'].items():
                if isinstance(result, dict) and 'outlier_count' in result:
                    total_outliers_by_method[method] = result['outlier_count']
                    if 'outlier_indices' in result:
                        all_outlier_indices.update(result['outlier_indices'])
        
        report += "## Summary\n"
        report += f"- **Total rows analyzed:** {len(self.df):,}\n"
        report += f"- **Unique outlier rows found:** {len(all_outlier_indices)}\n"
        report += f"- **Percentage of outlier rows:** {(len(all_outlier_indices)/len(self.df)*100):.2f}%\n\n"
        
        report += "### Outliers by Method:\n"
        for method, count in total_outliers_by_method.items():
            report += f"- **{method}:** {count} outliers\n"
        
        report += "\n## Detailed Results\n\n"
        
        for column, methods in self.outlier_results.items():
            if column == 'Multivariate':
                continue
            
            report += f"### Column: `{column}`\n\n"
            for method, result in methods.items():
                if not isinstance(result, dict) or ('outlier_count' in result and result['outlier_count'] == 0):
                    report += f"**{method}:** No outliers detected\n"
                    continue
                
                report += f"**{method}:**\n"
                report += f"- Outliers found: {result['outlier_count']} ({result['outlier_percentage']:.2f}%)\n"
                report += f"- Explanation: {result['explanation']}\n"
                
                if 'outlier_values' in result and result['outlier_values']:
                    sample_values = result['outlier_values'][:5]
                    formatted_values = ', '.join([f'{v:.3f}' if isinstance(v, (int, float)) else str(v) for v in sample_values])
                    report += f"- Example outliers: {formatted_values}"
                    if len(result['outlier_values']) > 5:
                        report += f" (and {len(result['outlier_values']) - 5} more)"
                    report += "\n"
                report += "\n"
        
        if 'Multivariate' in self.outlier_results:
            report += "### Multivariate Analysis\n\n"
            for method, result in self.outlier_results['Multivariate'].items():
                if not isinstance(result, dict) or 'outlier_count' not in result:
                    continue
                report += f"**{method}:**\n"
                report += f"- Outliers found: {result['outlier_count']} ({result['outlier_percentage']:.2f}%)\n"
                report += f"- Explanation: {result['explanation']}\n\n"
        
        return report

if __name__ == "__main__":
    def run_outlier_detection(file):
        detector = OutlierDetective()
        df = detector.load_data(file.name)
        detector.analyze_outliers()
        return detector.generate_outlier_report()

    iface = gr.Interface(fn=run_outlier_detection, 
                         inputs=gr.File(label="Upload a dataset"), 
                         outputs="text", 
                         title="Outlier Detection App")
    iface.launch()