Spaces:
Configuration error
Configuration error
| 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 | |