🏥 Healthcare Data Integration & ETL Pipeline

# Import Required Libraries
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import hashlib
import re
from faker import Faker
import warnings
warnings.filterwarnings('ignore')

# AWS SDK
import boto3
from botocore.exceptions import ClientError

# PySpark imports (for Glue)
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql.window import Window

# Snowflake connector
import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas

# Data quality
from great_expectations.dataset import PandasDataset

# Visualization
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.express as px

# Set styles
sns.set_style('whitegrid')
plt.rcParams['figure.figsize'] = (14, 6)

# Initialize Faker for synthetic data generation
fake = Faker()
Faker.seed(42)
np.random.seed(42)

print("✅ All libraries imported successfully!")
print(f"📅 Pipeline Run Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

# HIPAA Compliance Functions
def hash_phi(value):
    """
    One-way hash for PHI fields (SSN, etc.) using SHA-256.
    In production, use AWS KMS or similar encryption service.
    """
    if pd.isna(value):
        return None
    return hashlib.sha256(str(value).encode()).hexdigest()

def mask_phone(phone):
    """
    Mask phone number: (555) 123-4567 -> (XXX) XXX-4567
    """
    if pd.isna(phone):
        return None
    cleaned = re.sub(r'\D', '', phone)
    if len(cleaned) >= 4:
        return 'XXX-XXX-' + cleaned[-4:]
    return 'XXX-XXX-XXXX'

def mask_email(email):
    """
    Mask email: john.doe@example.com -> j***e@example.com
    """
    if pd.isna(email):
        return None
    parts = email.split('@')
    if len(parts) == 2:
        username = parts[0]
        if len(username) > 2:
            return f"{username[0]}{'*' * (len(username)-2)}{username[-1]}@{parts[1]}"
    return 'masked@domain.com'

def create_deidentified_id(patient_id, salt='HEALTHCARE_DW_2025'):
    """
    Create consistent de-identified patient ID for analytics.
    """
    combined = f"{patient_id}_{salt}"
    hashed = hashlib.sha256(combined.encode()).hexdigest()[:16]
    return f"DEID_{hashed.upper()}"

print("✅ HIPAA compliance functions defined")
print("🔒 PHI encryption methods ready:")
print("   • SSN: SHA-256 one-way hash")
print("   • Phone: Masked to last 4 digits")
print("   • Email: Masked username")
print("   • Patient ID: De-identified with salt")

# Data Quality Assessment Framework
def assess_data_quality(df, table_name, critical_columns):
    """
    Comprehensive data quality assessment with detailed reporting.
    Returns quality score and list of issues found.
    """
    print(f"\n🔍 Data Quality Assessment: {table_name}")
    print("=" * 80)
    
    quality_report = {
        'table_name': table_name,
        'total_records': len(df),
        'total_columns': len(df.columns),
        'assessment_date': datetime.now(),
        'issues': []
    }
    
    # 1. Completeness Check
    print("\n📊 Completeness Analysis:")
    missing_summary = df.isnull().sum()
    missing_pct = (missing_summary / len(df) * 100).round(2)
    
    for col in critical_columns:
        if col in df.columns:
            missing_count = missing_summary[col]
            if missing_count > 0:
                issue = f"Critical column '{col}' has {missing_count} ({missing_pct[col]}%) missing values"
                quality_report['issues'].append(issue)
                print(f"   ⚠️  {issue}")
            else:
                print(f"   ✅ {col}: 100% complete")
    
    # 2. Duplicate Check
    print("\n🔄 Duplicate Analysis:")
    if 'patient_id' in df.columns:
        duplicates = df['patient_id'].duplicated().sum()
        if duplicates > 0:
            issue = f"Found {duplicates} duplicate patient_id records"
            quality_report['issues'].append(issue)
            print(f"   ⚠️  {issue}")
        else:
            print(f"   ✅ No duplicate patient_id found")
    
    # 3. Calculate Overall Quality Score
    completeness_score = (1 - df.isnull().sum().sum() / (len(df) * len(df.columns))) * 100
    duplicate_score = (1 - (duplicates / len(df) if 'patient_id' in df.columns else 0)) * 100
    overall_score = (completeness_score + duplicate_score) / 2
    
    quality_report['completeness_score'] = round(completeness_score, 2)
    quality_report['overall_quality_score'] = round(overall_score, 2)
    
    print(f"\n🎯 Quality Score: {overall_score:.1f}%")
    print(f"   • Completeness: {completeness_score:.1f}%")
    print(f"   • Uniqueness: {duplicate_score:.1f}%")
    
    if len(quality_report['issues']) == 0:
        print("\n✅ No critical issues found!")
    else:
        print(f"\n⚠️  {len(quality_report['issues'])} issue(s) identified")
    
    return quality_report

print("✅ Data quality framework initialized")
print("📋 Quality checks include:")
print("   • Completeness validation")
print("   • Duplicate detection")
print("   • Statistical profiling")
print("   • Referential integrity")

# AWS Glue Deduplication Job (PySpark)
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.window import Window

# Initialize Spark session (in Glue, this is pre-configured)
spark = SparkSession.builder \
    .appName("Healthcare-Deduplication") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

def deduplicate_patients(df_patients):
    """
    Multi-stage patient deduplication using deterministic and fuzzy matching.
    Implements match scoring, clustering, and survivorship rules.
    """
    print("\n🔄 Starting patient deduplication process...")
    
    # Stage 1: Exact match deduplication on MRN
    print("\n📊 Stage 1: Exact MRN matching")
    df_dedupe = df_patients.withColumn(
        "row_num",
        row_number().over(
            Window.partitionBy("medical_record_number")
                  .orderBy(desc("last_visit_date"))
        )
    ).filter(col("row_num") == 1).drop("row_num")
    
    duplicates_removed = df_patients.count() - df_dedupe.count()
    print(f"   • Removed {duplicates_removed:,} duplicate MRN records")
    
    # Stage 2: Fuzzy matching on name + DOB
    print("\n📊 Stage 2: Fuzzy name/DOB matching")
    df_dedupe = df_dedupe.withColumn(
        "name_key",
        concat(
            lower(regexp_replace(col("first_name"), "[^a-zA-Z]", "")),
            lower(regexp_replace(col("last_name"), "[^a-zA-Z]", ""))
        )
    ).withColumn(
        "match_key",
        concat(col("name_key"), col("date_of_birth"))
    )
    
    # Identify potential duplicates using match key
    df_match = df_dedupe.withColumn(
        "match_count",
        count("*").over(Window.partitionBy("match_key"))
    ).filter(col("match_count") > 1)
    
    print(f"   • Found {df_match.count():,} potential fuzzy matches")
    
    # Stage 3: Apply survivorship rules
    print("\n📊 Stage 3: Applying survivorship rules")
    df_final = df_dedupe.withColumn(
        "survivorship_rank",
        row_number().over(
            Window.partitionBy("match_key")
                  .orderBy(
                      when(col("source_system") == "SALESFORCE", 1)
                      .when(col("source_system") == "EPIC_EHR", 2)
                      .otherwise(3),
                      desc("last_visit_date")
                  )
        )
    ).filter(col("survivorship_rank") == 1).drop("survivorship_rank", "name_key", "match_key")
    
    final_duplicates = df_dedupe.count() - df_final.count()
    print(f"   • Removed {final_duplicates:,} records after survivorship")
    
    print(f"\n✅ Deduplication complete!")
    print(f"   • Original records: {df_patients.count():,}")
    print(f"   • Final records: {df_final.count():,}")
    print(f"   • Total duplicates removed: {df_patients.count() - df_final.count():,}")
    print(f"   • Deduplication rate: {((df_patients.count() - df_final.count()) / df_patients.count() * 100):.2f}%")
    
    return df_final

print("✅ Deduplication pipeline defined")
print("🔧 Using PySpark for scalable processing")
print("📊 Stages: Exact match → Fuzzy match → Survivorship")

# Generate Synthetic Patient Demographics
print("🏥 Generating synthetic patient data...\n")

# Insurance payers distribution (realistic US healthcare market share)
insurance_payers = ['Medicare', 'Medicaid', 'Blue Cross Blue Shield', 'UnitedHealthcare', 
                   'Aetna', 'Cigna', 'Humana', 'Kaiser Permanente', 'Self-Pay']
insurance_weights = [0.20, 0.15, 0.18, 0.15, 0.10, 0.08, 0.06, 0.05, 0.03]

# Generate patient records
num_patients = 50000
patients_data = []

for i in range(num_patients):
    patient_id = f"PT{str(i+1).zfill(8)}"
    mrn = f"MRN{str(np.random.randint(100000, 999999))}"
    
    # Demographics
    gender = np.random.choice(['M', 'F'], p=[0.49, 0.51])
    first_name = fake.first_name_male() if gender == 'M' else fake.first_name_female()
    last_name = fake.last_name()
    
    # Age distribution: realistic healthcare population
    age = int(np.random.choice([
        np.random.randint(0, 18),    # Pediatric: 15%
        np.random.randint(18, 45),   # Young adult: 30%
        np.random.randint(45, 65),   # Middle age: 30%
        np.random.randint(65, 95)    # Senior: 25%
    ], p=[0.15, 0.30, 0.30, 0.25]))
    
    dob = fake.date_of_birth(minimum_age=age, maximum_age=age)
    
    # Contact information
    ssn = fake.ssn()
    phone = fake.phone_number()
    email = f"{first_name.lower()}.{last_name.lower()}@{fake.free_email_domain()}"
    
    # Address
    address = fake.street_address()
    city = fake.city()
    state = fake.state_abbr()
    zip_code = fake.zipcode()
    
    # Insurance
    insurance = np.random.choice(insurance_payers, p=insurance_weights)
    insurance_id = f"{insurance[:3].upper()}{np.random.randint(1000000, 9999999)}"
    
    # Last visit date (within past 2 years)
    days_ago = np.random.randint(1, 730)
    last_visit = datetime.now() - timedelta(days=days_ago)
    
    # Source system (to test deduplication)
    source = np.random.choice(['EPIC_EHR', 'CERNER_EHR', 'SALESFORCE'], p=[0.60, 0.30, 0.10])
    
    patients_data.append({
        'patient_id': patient_id,
        'medical_record_number': mrn,
        'first_name': first_name,
        'last_name': last_name,
        'date_of_birth': dob,
        'age': age,
        'gender': gender,
        'ssn': ssn,
        'phone': phone,
        'email': email,
        'address': address,
        'city': city,
        'state': state,
        'zip_code': zip_code,
        'insurance_provider': insurance,
        'insurance_id': insurance_id,
        'last_visit_date': last_visit,
        'source_system': source,
        'created_date': datetime.now()
    })

df_patients = pd.DataFrame(patients_data)

print(f"✅ Generated {len(df_patients):,} patient records")
print(f"\n📊 Sample Patient Data:")
print(df_patients[['patient_id', 'first_name', 'last_name', 'age', 'gender', 'insurance_provider']].head(3))
print(f"\n📈 Age Distribution:")
print(f"   • Pediatric (0-17): {len(df_patients[df_patients['age'] < 18]):,} ({len(df_patients[df_patients['age'] < 18])/len(df_patients)*100:.1f}%)")
print(f"   • Young Adult (18-44): {len(df_patients[(df_patients['age'] >= 18) & (df_patients['age'] < 45)]):,} ({len(df_patients[(df_patients['age'] >= 18) & (df_patients['age'] < 45)])/len(df_patients)*100:.1f}%)")
print(f"   • Middle Age (45-64): {len(df_patients[(df_patients['age'] >= 45) & (df_patients['age'] < 65)]):,} ({len(df_patients[(df_patients['age'] >= 45) & (df_patients['age'] < 65)])/len(df_patients)*100:.1f}%)")
print(f"   • Senior (65+): {len(df_patients[df_patients['age'] >= 65]):,} ({len(df_patients[df_patients['age'] >= 65])/len(df_patients)*100:.1f}%)")

# Generate Clinical Encounters (Hospital Visits)
print("🏥 Generating clinical encounters...\n")

# Common ICD-10 diagnosis codes
icd10_codes = {
    'E11.9': 'Type 2 diabetes without complications',
    'I10': 'Essential hypertension',
    'J44.9': 'COPD, unspecified',
    'F41.9': 'Anxiety disorder, unspecified',
    'M79.3': 'Fibromyalgia',
    'J06.9': 'Upper respiratory infection',
    'N39.0': 'Urinary tract infection',
    'K21.9': 'Gastro-esophageal reflux disease',
    'M25.511': 'Pain in right shoulder',
    'R51': 'Headache',
    'I25.10': 'Coronary artery disease',
    'E78.5': 'Hyperlipidemia',
    'F33.1': 'Major depressive disorder, recurrent',
    'J45.909': 'Asthma, unspecified',
    'I50.9': 'Heart failure, unspecified'
}

# Common CPT procedure codes
cpt_codes = {
    '99213': 'Office visit, established patient, moderate complexity',
    '99214': 'Office visit, established patient, high complexity',
    '99204': 'Office visit, new patient, high complexity',
    '80053': 'Comprehensive metabolic panel',
    '85025': 'Complete blood count with differential',
    '93000': 'Electrocardiogram, routine ECG',
    '36415': 'Blood draw (venipuncture)',
    '99285': 'Emergency department visit, high severity',
    '99223': 'Hospital inpatient care, high complexity',
    '71020': 'Chest X-ray, 2 views'
}

# Encounter types
encounter_types = ['Outpatient', 'Emergency', 'Inpatient', 'Telemedicine', 'Surgery']
encounter_weights = [0.55, 0.20, 0.15, 0.07, 0.03]

# Generate 3.5 encounters per patient on average
num_encounters = 175000
encounters_data = []

for i in range(num_encounters):
    # Select random patient
    patient = df_patients.sample(1).iloc[0]
    
    encounter_id = f"ENC{str(i+1).zfill(9)}"
    encounter_type = np.random.choice(encounter_types, p=encounter_weights)
    
    # Encounter date within past 2 years
    days_ago = np.random.randint(1, 730)
    encounter_date = datetime.now() - timedelta(days=days_ago)
    
    # Diagnosis codes (1-3 per encounter)
    num_diagnoses = np.random.choice([1, 2, 3], p=[0.60, 0.30, 0.10])
    diagnosis_codes = np.random.choice(list(icd10_codes.keys()), size=num_diagnoses, replace=False)
    primary_diagnosis = diagnosis_codes[0]
    
    # Procedure codes
    num_procedures = np.random.choice([1, 2, 3, 4], p=[0.50, 0.30, 0.15, 0.05])
    procedure_codes = np.random.choice(list(cpt_codes.keys()), size=num_procedures, replace=False)
    
    # Length of stay (for inpatient)
    if encounter_type == 'Inpatient':
        los = int(np.random.gamma(shape=2, scale=2)) + 1  # Realistic LOS distribution
    elif encounter_type == 'Surgery':
        los = int(np.random.gamma(shape=1.5, scale=1.5)) + 1
    else:
        los = 0
    
    # Discharge disposition
    if encounter_type in ['Inpatient', 'Surgery', 'Emergency']:
        discharge = np.random.choice(['Home', 'Home with Home Health', 'SNF', 'Rehab', 'Deceased'], 
                                    p=[0.70, 0.15, 0.08, 0.05, 0.02])
    else:
        discharge = 'Home'
    
    # Attending provider
    provider = f"Dr. {fake.last_name()}"
    facility = np.random.choice(['Main Hospital', 'North Campus', 'South Campus', 'East Clinic', 'West Clinic'])
    
    encounters_data.append({
        'encounter_id': encounter_id,
        'patient_id': patient['patient_id'],
        'encounter_type': encounter_type,
        'encounter_date': encounter_date,
        'facility': facility,
        'provider_name': provider,
        'primary_diagnosis_code': primary_diagnosis,
        'primary_diagnosis_desc': icd10_codes[primary_diagnosis],
        'all_diagnosis_codes': '|'.join(diagnosis_codes),
        'procedure_codes': '|'.join(procedure_codes),
        'length_of_stay': los,
        'discharge_disposition': discharge,
        'created_date': datetime.now()
    })

df_encounters = pd.DataFrame(encounters_data)

print(f"✅ Generated {len(df_encounters):,} clinical encounters")
print(f"\n📊 Sample Encounter Data:")
print(df_encounters[['encounter_id', 'patient_id', 'encounter_type', 'primary_diagnosis_desc', 'length_of_stay']].head(3))
print(f"\n📈 Encounter Type Distribution:")
for enc_type in encounter_types:
    count = len(df_encounters[df_encounters['encounter_type'] == enc_type])
    print(f"   • {enc_type}: {count:,} ({count/len(df_encounters)*100:.1f}%)")
print(f"\n🏥 Average Length of Stay (Inpatient): {df_encounters[df_encounters['encounter_type']=='Inpatient']['length_of_stay'].mean():.1f} days")

# Generate Billing Claims
print("💰 Generating billing claims data...\n")

# Generate claims for each encounter
claims_data = []

for idx, encounter in df_encounters.iterrows():
    claim_id = f"CLM{str(idx+1).zfill(9)}"
    
    # Get patient insurance
    patient = df_patients[df_patients['patient_id'] == encounter['patient_id']].iloc[0]
    
    # Calculate claim amount based on encounter type
    if encounter['encounter_type'] == 'Surgery':
        base_amount = np.random.uniform(15000, 85000)
    elif encounter['encounter_type'] == 'Inpatient':
        base_amount = np.random.uniform(8000, 45000) * encounter['length_of_stay']
    elif encounter['encounter_type'] == 'Emergency':
        base_amount = np.random.uniform(1200, 8500)
    elif encounter['encounter_type'] == 'Telemedicine':
        base_amount = np.random.uniform(75, 250)
    else:  # Outpatient
        base_amount = np.random.uniform(150, 1500)
    
    claim_amount = round(base_amount, 2)
    
    # Insurance payment (typically 70-90% of claim)
    insurance_paid = round(claim_amount * np.random.uniform(0.70, 0.90), 2)
    
    # Patient responsibility (copay + deductible)
    patient_paid = round(claim_amount - insurance_paid, 2)
    
    # Claim status
    claim_status = np.random.choice(['Paid', 'Pending', 'Denied', 'Appealed'], 
                                   p=[0.82, 0.10, 0.06, 0.02])
    
    # Denied claims have $0 insurance payment
    if claim_status == 'Denied':
        insurance_paid = 0.00
        denial_reason = np.random.choice([
            'Prior authorization required',
            'Medical necessity not met',
            'Non-covered service',
            'Duplicate claim',
            'Coding error'
        ])
    else:
        denial_reason = None
    
    # Filing date (1-30 days after encounter)
    filing_delay = np.random.randint(1, 31)
    filing_date = encounter['encounter_date'] + timedelta(days=filing_delay)
    
    claims_data.append({
        'claim_id': claim_id,
        'encounter_id': encounter['encounter_id'],
        'patient_id': encounter['patient_id'],
        'insurance_provider': patient['insurance_provider'],
        'claim_amount': claim_amount,
        'insurance_paid': insurance_paid,
        'patient_responsibility': patient_paid,
        'claim_status': claim_status,
        'denial_reason': denial_reason,
        'filing_date': filing_date,
        'created_date': datetime.now()
    })

df_claims = pd.DataFrame(claims_data)

print(f"✅ Generated {len(df_claims):,} billing claims")
print(f"\n📊 Sample Claims Data:")
print(df_claims[['claim_id', 'patient_id', 'claim_amount', 'insurance_paid', 'claim_status']].head(3))

# Calculate financial metrics
total_billed = df_claims['claim_amount'].sum()
total_paid = df_claims['insurance_paid'].sum()
collection_rate = (total_paid / total_billed * 100)

print(f"\n💰 Financial Metrics:")
print(f"   • Total Billed: ${total_billed:,.2f}")
print(f"   • Total Collected: ${total_paid:,.2f}")
print(f"   • Collection Rate: {collection_rate:.1f}%")
print(f"   • Average Claim Amount: ${df_claims['claim_amount'].mean():,.2f}")

print(f"\n📈 Claim Status Distribution:")
for status in ['Paid', 'Pending', 'Denied', 'Appealed']:
    count = len(df_claims[df_claims['claim_status'] == status])
    print(f"   • {status}: {count:,} ({count/len(df_claims)*100:.1f}%)")

denied_claims = df_claims[df_claims['claim_status'] == 'Denied']
denial_revenue_loss = denied_claims['claim_amount'].sum()
print(f"\n⚠️  Denied Claims Revenue Loss: ${denial_revenue_loss:,.2f}")

# Run Data Quality Assessments
print("🔍 Running comprehensive data quality assessment...\n")

# Assess each dataset
critical_patient_cols = ['patient_id', 'medical_record_number', 'first_name', 'last_name', 'date_of_birth']
critical_encounter_cols = ['encounter_id', 'patient_id', 'encounter_date', 'primary_diagnosis_code']
critical_claims_cols = ['claim_id', 'encounter_id', 'patient_id', 'claim_amount']

patient_quality = assess_data_quality(df_patients, 'PATIENTS', critical_patient_cols)
encounter_quality = assess_data_quality(df_encounters, 'ENCOUNTERS', critical_encounter_cols)
claims_quality = assess_data_quality(df_claims, 'CLAIMS', critical_claims_cols)

# Summary
print("\n" + "="*80)
print("📋 DATA QUALITY SUMMARY")
print("="*80)
print(f"\n{'Table':<15} {'Records':>12} {'Columns':>10} {'Quality Score':>15}")
print("-" * 80)
print(f"{'PATIENTS':<15} {patient_quality['total_records']:>12,} {patient_quality['total_columns']:>10} {patient_quality['overall_quality_score']:>14.1f}%")
print(f"{'ENCOUNTERS':<15} {encounter_quality['total_records']:>12,} {encounter_quality['total_columns']:>10} {encounter_quality['overall_quality_score']:>14.1f}%")
print(f"{'CLAIMS':<15} {claims_quality['total_records']:>12,} {claims_quality['total_columns']:>10} {claims_quality['overall_quality_score']:>14.1f}%")
print("-" * 80)

# Calculate average quality score
avg_quality = np.mean([
    patient_quality['overall_quality_score'],
    encounter_quality['overall_quality_score'],
    claims_quality['overall_quality_score']
])
print(f"\n🎯 Overall Data Warehouse Quality Score: {avg_quality:.1f}%")

if avg_quality >= 95:
    print("✅ Quality threshold met - Ready for production deployment")
else:
    print("⚠️  Quality below threshold - Review issues before deployment")