import pandas as pd import numpy as np import re from datetime import datetime, date from collections import Counter, defaultdict from typing import Dict, List, Tuple, Any, Optional import phonenumbers from phonenumbers import NumberParseException import logging from dataclasses import dataclass from utils.logger import setup_logger logger = setup_logger(__name__) @dataclass class ValidationIssue: """Represents a data quality issue found during validation""" category: str severity: str # 'HIGH', 'MEDIUM', 'LOW' description: str affected_records: List[int] examples: List[Dict[str, Any]] count: int recommendation: str class AdvancedDataValidator: """ Advanced data quality validator focusing on logical inconsistencies and business rule violations """ def __init__(self, company_founding_year: int = 1990): """ Initialize the advanced validator Args: company_founding_year: Year the company was founded (for join date validation) """ self.company_founding_year = company_founding_year self.issues: List[ValidationIssue] = [] self.current_year = datetime.now().year def validate_dataset(self, df: pd.DataFrame) -> Dict[str, Any]: """ Perform comprehensive advanced validation on the dataset Args: df: DataFrame to validate Returns: Dictionary containing validation results and detailed report """ logger.info(f"Starting advanced validation on dataset with {len(df)} records") # Clear previous issues self.issues = [] # Perform all validation checks self._detect_duplicate_identities(df) self._analyze_data_patterns(df) self._validate_business_logic(df) self._check_contextual_integrity(df) # Generate comprehensive report report = self._generate_validation_report(df) logger.info(f"Advanced validation completed. Found {len(self.issues)} issue categories") return report def _detect_duplicate_identities(self, df: pd.DataFrame): """Detect duplicate email addresses and phone numbers""" logger.info("Detecting duplicate identities...") # Email duplicate detection if 'email' in df.columns: self._check_duplicate_emails(df) # Phone number duplicate detection phone_columns = [col for col in df.columns if 'phone' in col.lower()] for col in phone_columns: self._check_duplicate_phones(df, col) # Contact information overlap self._check_contact_overlap(df) def _check_duplicate_emails(self, df: pd.DataFrame): """Check for duplicate email addresses""" if 'email' not in df.columns: return # Normalize emails (lowercase, strip whitespace) normalized_emails = df['email'].str.lower().str.strip() # Find duplicates email_counts = normalized_emails.value_counts() duplicates = email_counts[email_counts > 1] if len(duplicates) > 0: affected_records = [] examples = [] for email, count in duplicates.items(): duplicate_indices = df[normalized_emails == email].index.tolist() affected_records.extend(duplicate_indices) # Add example example_records = df.loc[duplicate_indices].to_dict('records') examples.append({ 'email': email, 'count': count, 'records': example_records[:3] # Show first 3 examples }) self.issues.append(ValidationIssue( category="Duplicate Email Addresses", severity="HIGH", description=f"Found {len(duplicates)} email addresses used by multiple records", affected_records=affected_records, examples=examples[:5], # Show first 5 examples count=len(affected_records), recommendation="Review duplicate emails for data entry errors or legitimate shared accounts. Consider implementing unique email constraints." )) def _check_duplicate_phones(self, df: pd.DataFrame, phone_column: str): """Check for duplicate phone numbers with normalization""" if phone_column not in df.columns: return # Normalize phone numbers normalized_phones = [] parse_errors = [] for idx, phone in df[phone_column].items(): if pd.isna(phone): normalized_phones.append(None) continue try: # Try to parse and format phone number parsed = phonenumbers.parse(str(phone), "US") # Assume US if no country code if phonenumbers.is_valid_number(parsed): normalized = phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164) normalized_phones.append(normalized) else: # Fallback: remove all non-digits digits_only = re.sub(r'\D', '', str(phone)) normalized_phones.append(digits_only if digits_only else None) except NumberParseException: # Fallback: remove all non-digits digits_only = re.sub(r'\D', '', str(phone)) normalized_phones.append(digits_only if digits_only else None) parse_errors.append(idx) # Create series with normalized phones normalized_series = pd.Series(normalized_phones, index=df.index) # Find duplicates (excluding None values) valid_phones = normalized_series.dropna() phone_counts = valid_phones.value_counts() duplicates = phone_counts[phone_counts > 1] if len(duplicates) > 0: affected_records = [] examples = [] for phone, count in duplicates.items(): duplicate_indices = df[normalized_series == phone].index.tolist() affected_records.extend(duplicate_indices) # Add example example_records = df.loc[duplicate_indices].to_dict('records') examples.append({ 'normalized_phone': phone, 'count': count, 'records': example_records[:3] }) self.issues.append(ValidationIssue( category=f"Duplicate Phone Numbers ({phone_column})", severity="HIGH", description=f"Found {len(duplicates)} phone numbers used by multiple records", affected_records=affected_records, examples=examples[:5], count=len(affected_records), recommendation="Review duplicate phone numbers for data entry errors or shared contact information. Consider business rules for shared phones." )) def _check_contact_overlap(self, df: pd.DataFrame): """Check for records sharing multiple contact methods""" if 'email' not in df.columns: return phone_columns = [col for col in df.columns if 'phone' in col.lower()] if not phone_columns: return # Group by email and check for multiple records with same phone overlaps = [] for email in df['email'].unique(): if pd.isna(email): continue email_records = df[df['email'] == email] if len(email_records) > 1: # Check if they also share phone numbers for phone_col in phone_columns: phones = email_records[phone_col].dropna().unique() if len(phones) == 1 and len(email_records) > 1: overlaps.append({ 'email': email, 'phone': phones[0], 'phone_column': phone_col, 'record_count': len(email_records), 'indices': email_records.index.tolist() }) if overlaps: affected_records = [] for overlap in overlaps: affected_records.extend(overlap['indices']) self.issues.append(ValidationIssue( category="Complete Contact Information Overlap", severity="MEDIUM", description=f"Found {len(overlaps)} cases where multiple records share both email and phone", affected_records=affected_records, examples=overlaps[:5], count=len(affected_records), recommendation="These may be legitimate duplicates or data entry errors. Review for consolidation opportunities." )) def _analyze_data_patterns(self, df: pd.DataFrame): """Analyze data patterns for anomalies""" logger.info("Analyzing data patterns for anomalies...") # Age distribution analysis if 'age' in df.columns: self._analyze_age_patterns(df) # Date clustering analysis date_columns = [col for col in df.columns if 'date' in col.lower() or col.lower() in ['dob', 'birth_date', 'birthdate']] for col in date_columns: self._analyze_date_clustering(df, col) def _analyze_age_patterns(self, df: pd.DataFrame): """Analyze age distribution for suspicious patterns""" if 'age' not in df.columns: return ages = df['age'].dropna() if len(ages) == 0: return # Check for excessive rounding (ages ending in 0 or 5) rounded_ages = ages[ages % 5 == 0] rounded_percentage = len(rounded_ages) / len(ages) * 100 if rounded_percentage > 40: # More than 40% rounded ages is suspicious examples = df[df['age'] % 5 == 0].head(10).to_dict('records') self.issues.append(ValidationIssue( category="Suspicious Age Rounding", severity="MEDIUM", description=f"{rounded_percentage:.1f}% of ages are rounded to multiples of 5, suggesting artificial data", affected_records=df[df['age'] % 5 == 0].index.tolist(), examples=examples, count=len(rounded_ages), recommendation="Review age data collection process. Consider if ages were estimated rather than calculated from birthdates." )) # Check for unrealistic age clustering age_counts = ages.value_counts() max_count = age_counts.max() total_unique_ages = len(age_counts) if max_count > len(ages) * 0.1 and total_unique_ages > 10: # More than 10% have same age most_common_age = age_counts.index[0] examples = df[df['age'] == most_common_age].head(10).to_dict('records') self.issues.append(ValidationIssue( category="Excessive Age Clustering", severity="MEDIUM", description=f"Age {most_common_age} appears {max_count} times ({max_count/len(ages)*100:.1f}% of records)", affected_records=df[df['age'] == most_common_age].index.tolist(), examples=examples, count=max_count, recommendation="Investigate why this specific age is so common. May indicate default value usage or data entry errors." )) def _analyze_date_clustering(self, df: pd.DataFrame, date_column: str): """Analyze date fields for suspicious clustering patterns""" if date_column not in df.columns: return # Convert to datetime if not already try: dates = pd.to_datetime(df[date_column], errors='coerce').dropna() except: return if len(dates) == 0: return # Check for January 1st clustering (common default date) jan_1_dates = dates[dates.dt.strftime('%m-%d') == '01-01'] jan_1_percentage = len(jan_1_dates) / len(dates) * 100 if jan_1_percentage > 20: # More than 20% on January 1st examples = df[pd.to_datetime(df[date_column], errors='coerce').dt.strftime('%m-%d') == '01-01'].head(10).to_dict('records') self.issues.append(ValidationIssue( category=f"Suspicious Date Clustering - January 1st ({date_column})", severity="MEDIUM", description=f"{jan_1_percentage:.1f}% of {date_column} values are January 1st, suggesting default values", affected_records=df[pd.to_datetime(df[date_column], errors='coerce').dt.strftime('%m-%d') == '01-01'].index.tolist(), examples=examples, count=len(jan_1_dates), recommendation="Review data collection process. January 1st clustering often indicates missing or unknown dates filled with defaults." )) # Check for excessive clustering on specific dates date_counts = dates.dt.date.value_counts() max_count = date_counts.max() if max_count > len(dates) * 0.05 and len(date_counts) > 20: # More than 5% on same date most_common_date = date_counts.index[0] examples = df[pd.to_datetime(df[date_column], errors='coerce').dt.date == most_common_date].head(10).to_dict('records') self.issues.append(ValidationIssue( category=f"Excessive Date Clustering ({date_column})", severity="MEDIUM", description=f"Date {most_common_date} appears {max_count} times ({max_count/len(dates)*100:.1f}% of records)", affected_records=df[pd.to_datetime(df[date_column], errors='coerce').dt.date == most_common_date].index.tolist(), examples=examples, count=max_count, recommendation="Investigate clustering around this specific date. May indicate bulk data imports or system-generated dates." )) def _validate_business_logic(self, df: pd.DataFrame): """Validate business logic rules""" logger.info("Validating business logic rules...") # Join date validation join_date_columns = [col for col in df.columns if 'join' in col.lower() or 'hire' in col.lower() or 'start' in col.lower()] for col in join_date_columns: self._validate_join_dates(df, col) # Age vs birthdate consistency if 'age' in df.columns and any('birth' in col.lower() or 'dob' in col.lower() for col in df.columns): self._validate_age_birthdate_consistency(df) # Temporal relationship validation self._validate_temporal_relationships(df) def _validate_join_dates(self, df: pd.DataFrame, join_column: str): """Validate join dates for business logic""" if join_column not in df.columns: return try: join_dates = pd.to_datetime(df[join_column], errors='coerce') except: return current_date = datetime.now() company_founding = datetime(self.company_founding_year, 1, 1) # Future join dates future_joins = df[join_dates > current_date] if len(future_joins) > 0: examples = future_joins.head(10).to_dict('records') self.issues.append(ValidationIssue( category=f"Future Join Dates ({join_column})", severity="HIGH", description=f"Found {len(future_joins)} records with join dates in the future", affected_records=future_joins.index.tolist(), examples=examples, count=len(future_joins), recommendation="Review and correct future join dates. These are likely data entry errors." )) # Join dates before company founding early_joins = df[join_dates < company_founding] if len(early_joins) > 0: examples = early_joins.head(10).to_dict('records') self.issues.append(ValidationIssue( category=f"Join Dates Before Company Founding ({join_column})", severity="HIGH", description=f"Found {len(early_joins)} records with join dates before company founding ({self.company_founding_year})", affected_records=early_joins.index.tolist(), examples=examples, count=len(early_joins), recommendation="Verify company founding year or correct impossible join dates." )) def _validate_age_birthdate_consistency(self, df: pd.DataFrame): """Validate consistency between age and birthdate""" age_col = 'age' birth_cols = [col for col in df.columns if 'birth' in col.lower() or 'dob' in col.lower()] if not birth_cols or age_col not in df.columns: return birth_col = birth_cols[0] # Use first birthdate column found try: birth_dates = pd.to_datetime(df[birth_col], errors='coerce') ages = df[age_col] except: return # Calculate age from birthdate current_date = datetime.now() calculated_ages = ((current_date - birth_dates).dt.days / 365.25).round().astype('Int64') # Find inconsistencies (allow 1 year tolerance for birthday timing) inconsistent = df[abs(ages - calculated_ages) > 1] if len(inconsistent) > 0: examples = [] for idx, row in inconsistent.head(10).iterrows(): examples.append({ 'record_index': idx, 'stated_age': row[age_col], 'calculated_age': calculated_ages.loc[idx], 'birthdate': row[birth_col], 'difference': abs(row[age_col] - calculated_ages.loc[idx]) }) self.issues.append(ValidationIssue( category="Age-Birthdate Inconsistency", severity="MEDIUM", description=f"Found {len(inconsistent)} records where stated age doesn't match calculated age from birthdate", affected_records=inconsistent.index.tolist(), examples=examples, count=len(inconsistent), recommendation="Review age calculation logic or data entry processes. Consider using birthdate as single source of truth for age." )) def _validate_temporal_relationships(self, df: pd.DataFrame): """Validate temporal relationships between date fields""" # Find birth and join date columns birth_cols = [col for col in df.columns if 'birth' in col.lower() or 'dob' in col.lower()] join_cols = [col for col in df.columns if 'join' in col.lower() or 'hire' in col.lower() or 'start' in col.lower()] if not birth_cols or not join_cols: return birth_col = birth_cols[0] join_col = join_cols[0] try: birth_dates = pd.to_datetime(df[birth_col], errors='coerce') join_dates = pd.to_datetime(df[join_col], errors='coerce') except: return # Check for join dates before birth dates (impossible) impossible = df[(join_dates < birth_dates) & birth_dates.notna() & join_dates.notna()] if len(impossible) > 0: examples = impossible.head(10).to_dict('records') self.issues.append(ValidationIssue( category="Impossible Date Relationships", severity="HIGH", description=f"Found {len(impossible)} records where join date is before birth date", affected_records=impossible.index.tolist(), examples=examples, count=len(impossible), recommendation="Critical data error: Join dates cannot be before birth dates. Review and correct these records immediately." )) # Check for unreasonably young employees (joined before age 16) young_joins = df[((join_dates - birth_dates).dt.days / 365.25 < 16) & birth_dates.notna() & join_dates.notna()] if len(young_joins) > 0: examples = [] for idx, row in young_joins.head(10).iterrows(): age_at_join = (join_dates.loc[idx] - birth_dates.loc[idx]).days / 365.25 examples.append({ 'record_index': idx, 'birth_date': birth_dates.loc[idx], 'join_date': join_dates.loc[idx], 'age_at_join': round(age_at_join, 1) }) self.issues.append(ValidationIssue( category="Unreasonably Young Employees", severity="MEDIUM", description=f"Found {len(young_joins)} records where employees joined before age 16", affected_records=young_joins.index.tolist(), examples=examples, count=len(young_joins), recommendation="Review hiring policies and data accuracy. Consider if these are data entry errors or special cases (internships, etc.)." )) def _check_contextual_integrity(self, df: pd.DataFrame): """Check contextual data integrity""" logger.info("Checking contextual data integrity...") # Bulk import pattern detection join_cols = [col for col in df.columns if 'join' in col.lower() or 'hire' in col.lower() or 'start' in col.lower()] for col in join_cols: self._detect_bulk_import_patterns(df, col) # Age range validation if 'age' in df.columns: self._validate_age_ranges(df) # Employment duration analysis self._analyze_employment_duration(df) def _detect_bulk_import_patterns(self, df: pd.DataFrame, date_column: str): """Detect patterns suggesting bulk data imports""" if date_column not in df.columns: return try: dates = pd.to_datetime(df[date_column], errors='coerce').dropna() except: return if len(dates) == 0: return # Group by month-year and look for unusual spikes monthly_counts = dates.dt.to_period('M').value_counts().sort_index() if len(monthly_counts) > 1: mean_monthly = monthly_counts.mean() std_monthly = monthly_counts.std() # Find months with unusually high activity (more than 2 standard deviations above mean) threshold = mean_monthly + (2 * std_monthly) spike_months = monthly_counts[monthly_counts > threshold] if len(spike_months) > 0: examples = [] for period, count in spike_months.head(5).items(): month_data = df[pd.to_datetime(df[date_column], errors='coerce').dt.to_period('M') == period] examples.append({ 'period': str(period), 'count': count, 'percentage_of_total': round(count / len(dates) * 100, 1), 'sample_records': month_data.head(3).to_dict('records') }) affected_records = [] for period in spike_months.index: period_records = df[pd.to_datetime(df[date_column], errors='coerce').dt.to_period('M') == period] affected_records.extend(period_records.index.tolist()) self.issues.append(ValidationIssue( category=f"Bulk Import Pattern Detection ({date_column})", severity="LOW", description=f"Detected {len(spike_months)} months with unusually high {date_column} activity, suggesting bulk imports", affected_records=affected_records, examples=examples, count=sum(spike_months), recommendation="Review data import processes. Bulk imports may indicate data migration or system changes that should be documented." )) def _validate_age_ranges(self, df: pd.DataFrame): """Validate that ages fall within reasonable working ranges""" if 'age' not in df.columns: return ages = df['age'].dropna() # Check for unreasonably young employees (under 16) too_young = df[df['age'] < 16] if len(too_young) > 0: examples = too_young.head(10).to_dict('records') self.issues.append(ValidationIssue( category="Unreasonably Young Ages", severity="HIGH", description=f"Found {len(too_young)} records with ages under 16", affected_records=too_young.index.tolist(), examples=examples, count=len(too_young), recommendation="Review minimum age policies and data accuracy. Ages under 16 are unusual for employment records." )) # Check for unreasonably old employees (over 80) too_old = df[df['age'] > 80] if len(too_old) > 0: examples = too_old.head(10).to_dict('records') self.issues.append(ValidationIssue( category="Unreasonably Old Ages", severity="MEDIUM", description=f"Found {len(too_old)} records with ages over 80", affected_records=too_old.index.tolist(), examples=examples, count=len(too_old), recommendation="Review retirement policies and data accuracy. Ages over 80 may indicate data entry errors or special employment arrangements." )) def _analyze_employment_duration(self, df: pd.DataFrame): """Analyze employment duration patterns""" join_cols = [col for col in df.columns if 'join' in col.lower() or 'hire' in col.lower() or 'start' in col.lower()] if not join_cols: return join_col = join_cols[0] try: join_dates = pd.to_datetime(df[join_col], errors='coerce') current_date = datetime.now() # Calculate employment duration in years employment_duration = ((current_date - join_dates).dt.days / 365.25) # Check for unusually long employment (over 50 years) very_long = df[employment_duration > 50] if len(very_long) > 0: examples = [] for idx, row in very_long.head(10).iterrows(): duration = employment_duration.loc[idx] examples.append({ 'record_index': idx, 'join_date': join_dates.loc[idx], 'duration_years': round(duration, 1) }) self.issues.append(ValidationIssue( category="Unusually Long Employment Duration", severity="MEDIUM", description=f"Found {len(very_long)} records with employment duration over 50 years", affected_records=very_long.index.tolist(), examples=examples, count=len(very_long), recommendation="Review join dates for accuracy. Employment over 50 years may indicate data entry errors or legacy system issues." )) except Exception as e: logger.warning(f"Could not analyze employment duration: {e}") def _generate_validation_report(self, df: pd.DataFrame) -> Dict[str, Any]: """Generate comprehensive validation report""" logger.info("Generating validation report...") # Categorize issues by severity high_severity = [issue for issue in self.issues if issue.severity == "HIGH"] medium_severity = [issue for issue in self.issues if issue.severity == "MEDIUM"] low_severity = [issue for issue in self.issues if issue.severity == "LOW"] # Calculate overall data quality score total_records = len(df) total_affected = len(set().union(*[issue.affected_records for issue in self.issues])) quality_score = max(0, 100 - (total_affected / total_records * 100)) if total_records > 0 else 100 # Generate summary statistics summary = { 'total_records': total_records, 'total_issues_found': len(self.issues), 'high_severity_issues': len(high_severity), 'medium_severity_issues': len(medium_severity), 'low_severity_issues': len(low_severity), 'total_affected_records': total_affected, 'data_quality_score': round(quality_score, 1), 'validation_timestamp': datetime.now().isoformat() } # Prepare detailed issues detailed_issues = [] for issue in self.issues: detailed_issues.append({ 'category': issue.category, 'severity': issue.severity, 'description': issue.description, 'count': issue.count, 'affected_percentage': round(issue.count / total_records * 100, 2), 'examples': issue.examples, 'recommendation': issue.recommendation }) # Generate recommendations by priority recommendations = [] # High priority recommendations if high_severity: recommendations.append({ 'priority': 'HIGH', 'title': 'Critical Data Issues Requiring Immediate Attention', 'items': [issue.recommendation for issue in high_severity] }) # Medium priority recommendations if medium_severity: recommendations.append({ 'priority': 'MEDIUM', 'title': 'Data Quality Improvements', 'items': [issue.recommendation for issue in medium_severity] }) # Low priority recommendations if low_severity: recommendations.append({ 'priority': 'LOW', 'title': 'Process Improvements and Monitoring', 'items': [issue.recommendation for issue in low_severity] }) return { 'summary': summary, 'detailed_issues': detailed_issues, 'recommendations': recommendations, 'validation_passed': len(high_severity) == 0, 'issues_by_severity': { 'high': [issue.__dict__ for issue in high_severity], 'medium': [issue.__dict__ for issue in medium_severity], 'low': [issue.__dict__ for issue in low_severity] } }