data_pipeline.py

import pandas as pd
import numpy as np
import glob
from sklearn.ensemble import IsolationForest

def load_folder_csvs(folder_path):
    files = glob.glob(folder_path + "/*.csv")
    dfs = [pd.read_csv(f) for f in files]
    return pd.concat(dfs, ignore_index=True)

enrol_raw = load_folder_csvs("api_data_aadhar_enrolment")
demo_raw  = load_folder_csvs("api_data_aadhar_demographic")
bio_raw   = load_folder_csvs("api_data_aadhar_biometric")


def clean_common(df):
    df.columns = df.columns.str.strip().str.lower()
    
    df["state"] = df["state"].astype(str).str.strip().str.lower()
    df["district"] = df["district"].astype(str).str.strip().str.lower()
    df["pincode"] = df["pincode"].astype(str).str.zfill(6)
    
    df["date"] = pd.to_datetime(df["date"], errors="coerce")
    df = df.dropna(subset=["date", "state", "district"])
    
    return df

enrol = clean_common(enrol_raw).copy()
demo  = clean_common(demo_raw).copy()
bio   = clean_common(bio_raw).copy()

enrol["total_enrol"] = (
    enrol["age_0_5"] +
    enrol["age_5_17"] +
    enrol["age_18_greater"]
)

demo["total_demo"] = (
    demo["demo_age_5_17"] +
    demo["demo_age_17_"]
)

bio["total_bio"] = (
    bio["bio_age_5_17"] +
    bio["bio_age_17_"]
)

group_cols = ["date", "state", "district"]

enrol_g = enrol.groupby(group_cols, as_index=False)["total_enrol"].sum()
demo_g  = demo.groupby(group_cols, as_index=False)["total_demo"].sum()
bio_g   = bio.groupby(group_cols, as_index=False)["total_bio"].sum()

df = (
    enrol_g
    .merge(demo_g, on=group_cols, how="left")
    .merge(bio_g, on=group_cols, how="left")
)

df = df.fillna(0)

df["update_churn"] = (df["total_demo"] + df["total_bio"]) / (df["total_enrol"] + 1)
df["bio_ratio"] = df["total_bio"] / (df["total_demo"] + df["total_bio"] + 1)

# State-normalized baselines
df["enrol_z"] = df.groupby("state")["total_enrol"].transform(
    lambda x: (x - x.mean()) / (x.std() + 1)
)

df["churn_z"] = df.groupby("state")["update_churn"].transform(
    lambda x: (x - x.mean()) / (x.std() + 1)
)

features = ["total_enrol", "update_churn", "bio_ratio"]

iso = IsolationForest(
    n_estimators=200,
    contamination=0.05,
    random_state=42
)

df["anomaly_flag"] = iso.fit_predict(df[features])
df["anomaly_flag"] = df["anomaly_flag"].map({1: 0, -1: 1})

def label_issue(row):
    if row["total_enrol"] < 5 and row["update_churn"] < 0.2:
        return "service_ghost_zone"
    if row["bio_ratio"] > 0.6:
        return "biometric_stress"
    if row["update_churn"] > 3:
        return "process_anomaly"
    return "normal"

df["issue_type"] = df.apply(label_issue, axis=1)

df["health_score"] = (
    100
    - 30 * df["anomaly_flag"]
    - 20 * df["update_churn"].clip(0, 3)
    - 20 * df["bio_ratio"]
)

df["health_score"] = df["health_score"].clip(0, 100)

print(df.head())
df.to_csv("uidai_final_output.csv", index=False)
print("Data saved to uidai_final_output.csv")