import pandas as pd import numpy as np from datetime import datetime from typing import List, Dict import re # Valid ICD-10 code patterns (simplified) ICD10_PATTERNS = [ r'^[A-Z][0-9]{2}(\.[0-9]{1,4})?$', # Standard ICD-10 r'^[A-Z][0-9]{2}$', # Without decimal ] class ValidationEngine: def __init__(self): self.required_columns = [ 'patient_id', 'age', 'gender', 'diagnosis_code', 'treatment_cost', 'claim_date', 'hospital_id', 'provider_id', 'risk_score', 'duplicate_flag' ] self.date_formats = [ '%Y-%m-%d', '%m/%d/%Y', '%d/%m/%Y', '%Y/%m/%d' ] def validate(self, df: pd.DataFrame) -> Dict: """Run all validation rules""" errors = [] warnings = [] # Check 1: Missing columns missing_cols = [col for col in self.required_columns if col not in df.columns] if missing_cols: errors.append({ "type": "missing_columns", "severity": "high", "message": f"Missing required columns: {', '.join(missing_cols)}", "columns": missing_cols }) # Check 2: Missing values for col in df.columns: missing_count = df[col].isna().sum() if missing_count > 0: missing_indices = df[df[col].isna()].index.tolist() errors.append({ "type": "missing_values", "severity": "medium", "column": col, "count": int(missing_count), "row_indices": missing_indices, "message": f"Column '{col}' has {missing_count} missing values" }) # Check 3: Date format validation if 'claim_date' in df.columns: date_errors = self._validate_dates(df['claim_date']) errors.extend(date_errors) # Check 4: Out-of-range values range_errors = self._validate_ranges(df) errors.extend(range_errors) # Check 5: Duplicate detection duplicate_errors = self._validate_duplicates(df) errors.extend(duplicate_errors) # Check 6: Invalid ICD codes if 'diagnosis_code' in df.columns: icd_errors = self._validate_icd_codes(df['diagnosis_code']) errors.extend(icd_errors) # Check 7: Statistical outliers outlier_errors = self._validate_outliers(df) errors.extend(outlier_errors) # Generate summary summary = { "total_records": len(df), "total_errors": len(errors), "errors_by_type": {}, "errors_by_severity": {"high": 0, "medium": 0, "low": 0} } for error in errors: error_type = error.get("type", "unknown") summary["errors_by_type"][error_type] = summary["errors_by_type"].get(error_type, 0) + 1 severity = error.get("severity", "low") summary["errors_by_severity"][severity] = summary["errors_by_severity"].get(severity, 0) + 1 return { "valid": len(errors) == 0, "summary": summary, "errors": errors, "warnings": warnings } def _validate_dates(self, date_series: pd.Series) -> List[Dict]: """Validate date formats""" errors = [] for idx, date_val in enumerate(date_series): if pd.isna(date_val): continue date_str = str(date_val) valid = False for fmt in self.date_formats: try: datetime.strptime(date_str, fmt) valid = True break except: continue if not valid: errors.append({ "type": "date_format_mismatch", "severity": "medium", "row_index": int(idx), "value": date_str, "message": f"Invalid date format at row {idx}: {date_str}" }) return errors def _validate_ranges(self, df: pd.DataFrame) -> List[Dict]: """Validate value ranges""" errors = [] # Age validation if 'age' in df.columns: age_series = pd.to_numeric(df['age'], errors='coerce') invalid_ages = (age_series < 0) | (age_series > 150) | age_series.isna() for idx in df[invalid_ages].index: errors.append({ "type": "out_of_range", "severity": "high", "column": "age", "row_index": int(idx), "value": df.loc[idx, 'age'], "message": f"Age out of range (0-150) at row {idx}" }) # Treatment cost validation if 'treatment_cost' in df.columns: cost_series = pd.to_numeric(df['treatment_cost'], errors='coerce') invalid_costs = (cost_series < 0) | (cost_series > 1000000) | cost_series.isna() for idx in df[invalid_costs].index: errors.append({ "type": "out_of_range", "severity": "medium", "column": "treatment_cost", "row_index": int(idx), "value": df.loc[idx, 'treatment_cost'], "message": f"Treatment cost out of range at row {idx}" }) # Risk score validation if 'risk_score' in df.columns: risk_series = pd.to_numeric(df['risk_score'], errors='coerce') invalid_risks = (risk_series < 0) | (risk_series > 10) | risk_series.isna() for idx in df[invalid_risks].index: errors.append({ "type": "out_of_range", "severity": "medium", "column": "risk_score", "row_index": int(idx), "value": df.loc[idx, 'risk_score'], "message": f"Risk score out of range (0-10) at row {idx}" }) # Gender validation if 'gender' in df.columns: valid_genders = ['M', 'F', 'Male', 'Female', 'm', 'f', 'male', 'female'] invalid_genders = ~df['gender'].isin(valid_genders) for idx in df[invalid_genders].index: errors.append({ "type": "out_of_range", "severity": "low", "column": "gender", "row_index": int(idx), "value": df.loc[idx, 'gender'], "message": f"Invalid gender value at row {idx}" }) return errors def _validate_duplicates(self, df: pd.DataFrame) -> List[Dict]: """Detect duplicate records""" errors = [] if 'patient_id' in df.columns: duplicates = df[df.duplicated(subset=['patient_id'], keep=False)] if len(duplicates) > 0: duplicate_indices = duplicates.index.tolist() errors.append({ "type": "duplicate_detection", "severity": "high", "row_indices": [int(idx) for idx in duplicate_indices], "count": len(duplicate_indices), "message": f"Found {len(duplicate_indices)} duplicate patient IDs" }) # Check for exact duplicate rows exact_duplicates = df[df.duplicated(keep=False)] if len(exact_duplicates) > 0: duplicate_indices = exact_duplicates.index.tolist() errors.append({ "type": "duplicate_detection", "severity": "high", "row_indices": [int(idx) for idx in duplicate_indices], "count": len(duplicate_indices), "message": f"Found {len(duplicate_indices)} exact duplicate rows" }) return errors def _validate_icd_codes(self, code_series: pd.Series) -> List[Dict]: """Validate ICD-10 diagnosis codes""" errors = [] for idx, code in enumerate(code_series): if pd.isna(code): continue code_str = str(code).strip().upper() valid = False for pattern in ICD10_PATTERNS: if re.match(pattern, code_str): valid = True break if not valid: errors.append({ "type": "invalid_icd_code", "severity": "medium", "row_index": int(idx), "value": code_str, "message": f"Invalid ICD-10 code at row {idx}: {code_str}" }) return errors def _validate_outliers(self, df: pd.DataFrame) -> List[Dict]: """Detect statistical outliers using IQR method""" errors = [] numeric_cols = ['age', 'treatment_cost', 'risk_score'] for col in numeric_cols: if col not in df.columns: continue series = pd.to_numeric(df[col], errors='coerce') 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 outliers = (series < lower_bound) | (series > upper_bound) outlier_indices = df[outliers].index.tolist() if len(outlier_indices) > 0: errors.append({ "type": "statistical_outlier", "severity": "low", "column": col, "row_indices": [int(idx) for idx in outlier_indices], "count": len(outlier_indices), "message": f"Found {len(outlier_indices)} statistical outliers in '{col}'" }) return errors