""" 📊 DATA QUALITY ANALYZER v1.0 ============================== Comprehensive data quality analysis and scoring: - Missing value analysis - Duplicate detection - Outlier identification - Class imbalance detection - Data type validation - Cardinality checks """ import numpy as np import pandas as pd import logging from typing import Dict, List, Any, Optional from dataclasses import dataclass, field logger = logging.getLogger(__name__) @dataclass class QualityIssue: """Represents a single data quality issue""" severity: str # 'low', 'medium', 'high', 'critical' category: str # 'missing', 'duplicate', 'outlier', 'imbalance', 'type' column: Optional[str] description: str recommendation: str impact_score: float # 0-100 @dataclass class QualityReport: """Complete data quality report""" overall_score: float = 0.0 grade: str = 'F' issues: List[QualityIssue] = field(default_factory=list) metrics: Dict[str, Any] = field(default_factory=dict) recommendations: List[str] = field(default_factory=list) column_scores: Dict[str, float] = field(default_factory=dict) class DataQualityAnalyzer: """ Production-grade data quality analysis. Provides: - Comprehensive quality scoring (0-100) - Issue detection and categorization - Actionable recommendations - Column-level quality metrics """ def __init__(self): self.report: Optional[QualityReport] = None def analyze(self, df: pd.DataFrame, target_col: Optional[str] = None) -> QualityReport: """ Run complete data quality analysis. Args: df: DataFrame to analyze target_col: Optional target column name for imbalance checks Returns: QualityReport with scores and recommendations """ logger.info(f"📊 Starting data quality analysis ({len(df)} rows, {len(df.columns)} cols)...") self.report = QualityReport() self.report.metrics['n_rows'] = len(df) self.report.metrics['n_cols'] = len(df.columns) # Run all checks self._check_missing_values(df) self._check_duplicates(df) self._check_outliers(df) self._check_cardinality(df) self._check_data_types(df) if target_col and target_col in df.columns: self._check_class_imbalance(df, target_col) # Calculate overall score self._calculate_overall_score() # Generate summary recommendations self._generate_recommendations() logger.info(f" ✅ Quality Score: {self.report.overall_score:.1f}/100 (Grade: {self.report.grade})") return self.report def _check_missing_values(self, df: pd.DataFrame): """Analyze missing values in each column""" logger.info(" 🔍 Checking missing values...") missing_pct = df.isnull().sum() / len(df) * 100 self.report.metrics['missing_summary'] = missing_pct.to_dict() for col, pct in missing_pct.items(): self.report.column_scores[col] = self.report.column_scores.get(col, 100) if pct > 50: self.report.issues.append(QualityIssue( severity='critical', category='missing', column=col, description=f"Column '{col}' has {pct:.1f}% missing values", recommendation=f"Consider dropping '{col}' or using advanced imputation", impact_score=30 )) self.report.column_scores[col] -= 50 elif pct > 30: self.report.issues.append(QualityIssue( severity='high', category='missing', column=col, description=f"Column '{col}' has {pct:.1f}% missing values", recommendation=f"Apply imputation strategy for '{col}'", impact_score=20 )) self.report.column_scores[col] -= 30 elif pct > 10: self.report.issues.append(QualityIssue( severity='medium', category='missing', column=col, description=f"Column '{col}' has {pct:.1f}% missing values", recommendation=f"Review missing pattern in '{col}'", impact_score=10 )) self.report.column_scores[col] -= 15 def _check_duplicates(self, df: pd.DataFrame): """Check for duplicate rows""" logger.info(" 🔍 Checking duplicates...") dup_count = df.duplicated().sum() dup_pct = dup_count / len(df) * 100 self.report.metrics['duplicate_count'] = int(dup_count) self.report.metrics['duplicate_pct'] = round(dup_pct, 2) if dup_pct > 20: self.report.issues.append(QualityIssue( severity='high', category='duplicate', column=None, description=f"Dataset has {dup_pct:.1f}% duplicate rows ({dup_count} rows)", recommendation="Review and remove duplicate rows to avoid data leakage", impact_score=20 )) elif dup_pct > 5: self.report.issues.append(QualityIssue( severity='medium', category='duplicate', column=None, description=f"Dataset has {dup_pct:.1f}% duplicate rows", recommendation="Consider deduplication before training", impact_score=10 )) def _check_outliers(self, df: pd.DataFrame): """Detect outliers in numeric columns using IQR method""" logger.info(" 🔍 Checking outliers...") numeric_cols = df.select_dtypes(include=[np.number]).columns outlier_summary = {} for col in numeric_cols: series = df[col].dropna() if len(series) < 10: continue q1, q3 = series.quantile([0.25, 0.75]) iqr = q3 - q1 if iqr == 0: continue lower_bound = q1 - 1.5 * iqr upper_bound = q3 + 1.5 * iqr outliers = ((series < lower_bound) | (series > upper_bound)).sum() outlier_pct = outliers / len(series) * 100 outlier_summary[col] = { 'count': int(outliers), 'pct': round(outlier_pct, 2), 'lower_bound': round(lower_bound, 2), 'upper_bound': round(upper_bound, 2) } self.report.column_scores[col] = self.report.column_scores.get(col, 100) if outlier_pct > 10: self.report.issues.append(QualityIssue( severity='medium', category='outlier', column=col, description=f"Column '{col}' has {outlier_pct:.1f}% outliers ({outliers} values)", recommendation=f"Consider outlier treatment for '{col}' (capping, transformation, or removal)", impact_score=10 )) self.report.column_scores[col] -= 10 self.report.metrics['outlier_summary'] = outlier_summary def _check_cardinality(self, df: pd.DataFrame): """Check cardinality of categorical columns""" logger.info(" 🔍 Checking cardinality...") cardinality_summary = {} for col in df.columns: n_unique = df[col].nunique() unique_ratio = n_unique / len(df) cardinality_summary[col] = { 'n_unique': int(n_unique), 'unique_ratio': round(unique_ratio, 4) } # High cardinality categorical if df[col].dtype == 'object' and n_unique > 100: self.report.issues.append(QualityIssue( severity='medium', category='cardinality', column=col, description=f"Column '{col}' has high cardinality ({n_unique} unique values)", recommendation=f"Consider target encoding or frequency encoding for '{col}'", impact_score=5 )) # Constant column if n_unique == 1: self.report.issues.append(QualityIssue( severity='high', category='cardinality', column=col, description=f"Column '{col}' is constant (only 1 unique value)", recommendation=f"Drop constant column '{col}' - provides no information", impact_score=15 )) # Near-unique column (likely ID) if unique_ratio > 0.95 and df[col].dtype in ['int64', 'object']: self.report.issues.append(QualityIssue( severity='low', category='cardinality', column=col, description=f"Column '{col}' appears to be an ID column ({unique_ratio:.1%} unique)", recommendation=f"Consider excluding '{col}' from features", impact_score=5 )) self.report.metrics['cardinality_summary'] = cardinality_summary def _check_data_types(self, df: pd.DataFrame): """Validate data types and detect mismatches""" logger.info(" 🔍 Checking data types...") type_summary = {} for col in df.columns: dtype = str(df[col].dtype) type_summary[col] = dtype # Check for numeric stored as string if df[col].dtype == 'object': try: numeric_count = pd.to_numeric(df[col], errors='coerce').notna().sum() if numeric_count / len(df) > 0.9: self.report.issues.append(QualityIssue( severity='low', category='type', column=col, description=f"Column '{col}' appears numeric but stored as string", recommendation=f"Convert '{col}' to numeric type for better performance", impact_score=3 )) except: pass self.report.metrics['data_types'] = type_summary def _check_class_imbalance(self, df: pd.DataFrame, target_col: str): """Check for class imbalance in target variable""" logger.info(f" 🔍 Checking class imbalance in '{target_col}'...") if df[target_col].dtype in ['float64', 'float32']: # Regression target - skip imbalance check return class_counts = df[target_col].value_counts() n_classes = len(class_counts) if n_classes < 2: return imbalance_ratio = class_counts.max() / class_counts.min() self.report.metrics['class_distribution'] = class_counts.to_dict() self.report.metrics['imbalance_ratio'] = round(imbalance_ratio, 2) self.report.metrics['n_classes'] = n_classes if imbalance_ratio > 10: self.report.issues.append(QualityIssue( severity='critical', category='imbalance', column=target_col, description=f"Severe class imbalance (ratio: {imbalance_ratio:.1f}:1)", recommendation="Apply SMOTE, class weighting, or undersampling", impact_score=25 )) elif imbalance_ratio > 5: self.report.issues.append(QualityIssue( severity='high', category='imbalance', column=target_col, description=f"Significant class imbalance (ratio: {imbalance_ratio:.1f}:1)", recommendation="Consider class balancing techniques", impact_score=15 )) elif imbalance_ratio > 3: self.report.issues.append(QualityIssue( severity='medium', category='imbalance', column=target_col, description=f"Moderate class imbalance (ratio: {imbalance_ratio:.1f}:1)", recommendation="Monitor model performance across classes", impact_score=8 )) def _calculate_overall_score(self): """Calculate overall data quality score""" base_score = 100 # Deduct points for each issue based on severity severity_weights = { 'critical': 15, 'high': 10, 'medium': 5, 'low': 2 } for issue in self.report.issues: base_score -= severity_weights.get(issue.severity, 5) # Ensure score is between 0-100 self.report.overall_score = max(0, min(100, base_score)) # Assign grade if self.report.overall_score >= 90: self.report.grade = 'A' elif self.report.overall_score >= 80: self.report.grade = 'B' elif self.report.overall_score >= 70: self.report.grade = 'C' elif self.report.overall_score >= 60: self.report.grade = 'D' else: self.report.grade = 'F' def _generate_recommendations(self): """Generate prioritized recommendations""" # Group issues by severity critical_issues = [i for i in self.report.issues if i.severity == 'critical'] high_issues = [i for i in self.report.issues if i.severity == 'high'] if critical_issues: self.report.recommendations.append( f"🚨 CRITICAL: Address {len(critical_issues)} critical issues before training" ) if high_issues: self.report.recommendations.append( f"⚠️ HIGH PRIORITY: Fix {len(high_issues)} high-severity issues for better accuracy" ) # Add specific recommendations for issue in self.report.issues[:5]: # Top 5 issues self.report.recommendations.append(issue.recommendation) def get_summary_dict(self) -> Dict[str, Any]: """Get report as dictionary for API response""" if not self.report: return {} return { 'overall_score': self.report.overall_score, 'grade': self.report.grade, 'n_issues': len(self.report.issues), 'issues': [ { 'severity': i.severity, 'category': i.category, 'column': i.column, 'description': i.description, 'recommendation': i.recommendation } for i in self.report.issues ], 'metrics': self.report.metrics, 'recommendations': self.report.recommendations, 'column_scores': self.report.column_scores } def analyze_data_quality(df: pd.DataFrame, target_col: str = None) -> Dict[str, Any]: """ Convenience function to run data quality analysis. Args: df: DataFrame to analyze target_col: Optional target column Returns: Dictionary with quality report """ analyzer = DataQualityAnalyzer() analyzer.analyze(df, target_col) return analyzer.get_summary_dict()